diff --git a/backend/requirements-dev.txt b/backend/requirements-dev.txt index 529d6051..383d07e4 100644 --- a/backend/requirements-dev.txt +++ b/backend/requirements-dev.txt @@ -9,3 +9,7 @@ pytest==8.3.4 pytest-asyncio==0.25.2 + +# Used by linter/lint.py (the vulture dead-code check). watchfiles, also +# needed by lint.py, already comes in transitively via uvicorn[standard]. +vulture==2.16 diff --git a/linter/.gitignore b/linter/.gitignore new file mode 100644 index 00000000..7bfcd7a0 --- /dev/null +++ b/linter/.gitignore @@ -0,0 +1,2 @@ +__pycache__/ +*.pyc \ No newline at end of file diff --git a/linter/checks/.lintignore-max-folder-items b/linter/checks/.lintignore-max-folder-items new file mode 100644 index 00000000..e69de29b diff --git a/linter/checks/__init__.py b/linter/checks/__init__.py new file mode 100644 index 00000000..8baad9ce --- /dev/null +++ b/linter/checks/__init__.py @@ -0,0 +1,68 @@ +"""Check infrastructure: shared filter/match utilities.""" + +from __future__ import annotations + +import fnmatch +import os +from pathlib import Path + +LINTIGNORE_PREFIX = ".lintignore" + + +def _matches_any(text: str, patterns: list[str]) -> bool: + return any(fnmatch.fnmatch(text, p) for p in patterns) + + +def is_excluded(path: Path, root: Path, excludes: list[str]) -> bool: + rel = path.relative_to(root) + for part in rel.parts: + if _matches_any(part, excludes): + return True + return _matches_any(str(rel), excludes) + + +def is_excepted(rel_path: str, rule: str, exceptions: dict[str, list[str]]) -> bool: + return _matches_any(rel_path, exceptions.get(rule, [])) + + +def collect_lintignores(root: Path, excludes: list[str]) -> dict[Path, set[str]]: + """Scan *root* for ``.lintignore*`` sentinel files. + + Returns ``{directory: set_of_ignored_rules}``. + The special token ``"__all__"`` means every rule is ignored. + """ + ignores: dict[Path, set[str]] = {} + for dirpath_str, dirnames, filenames in os.walk(root): + dp = Path(dirpath_str) + if is_excluded(dp, root, excludes): + dirnames.clear() + continue + for fname in filenames: + if fname == LINTIGNORE_PREFIX: + ignores.setdefault(dp, set()).add("__all__") + elif fname.startswith(f"{LINTIGNORE_PREFIX}-"): + rule = fname[len(LINTIGNORE_PREFIX) + 1 :] + ignores.setdefault(dp, set()).add(rule) + return ignores + + +def is_lintignored( + path: Path, + root: Path, + rule: str, + ignores: dict[Path, set[str]], +) -> bool: + """Return True if *path* is covered by a ``.lintignore`` file for *rule*. + + Walks from *path* up to *root* checking each ancestor directory. + """ + current = path if path.is_dir() else path.parent + root = root.resolve() + while True: + rules = ignores.get(current) + if rules and ("__all__" in rules or rule in rules): + return True + if current == root: + break + current = current.parent + return False diff --git a/linter/checks/classes.py b/linter/checks/classes.py new file mode 100644 index 00000000..38e0bd04 --- /dev/null +++ b/linter/checks/classes.py @@ -0,0 +1,61 @@ +"""Class-level dead code detection with framework awareness. + +Pydantic BaseModel subclasses are auto-whitelisted: every annotated field is +part of the serialization schema and therefore intentionally "used". + +Non-framework classes are skipped for now (tier 2 — future cross-referencing). +""" + +from __future__ import annotations + +import ast +from pathlib import Path + +from . import is_excepted, is_excluded, is_lintignored + +FRAMEWORK_BASES = {"BaseModel"} + + +def _is_framework_model(cls: ast.ClassDef) -> bool: + return any( + (isinstance(b, ast.Name) and b.id in FRAMEWORK_BASES) + or (isinstance(b, ast.Attribute) and b.attr in FRAMEWORK_BASES) + for b in cls.bases + ) + + +def run_class_check( + root: Path, + exceptions: dict[str, list[str]], + excludes: list[str], + ignores: dict[Path, set[str]] | None = None, +) -> list[str]: + """Analyse classes in backend Python files and return errors.""" + errors: list[str] = [] + backend = root / "backend" + if not backend.is_dir(): + return errors + + for pyfile in sorted(backend.rglob("*.py")): + if is_excluded(pyfile, root, excludes): + continue + rel = str(pyfile.relative_to(root)) + if is_excepted(rel, "classes", exceptions): + continue + if ignores and is_lintignored(pyfile, root, "classes", ignores): + continue + try: + source = pyfile.read_text() + tree = ast.parse(source, filename=rel) + except (OSError, SyntaxError): + continue + + for node in ast.walk(tree): + if not isinstance(node, ast.ClassDef): + continue + if _is_framework_model(node): + continue + # Tier 2 placeholder: non-framework classes are skipped until + # cross-reference analysis is implemented. + + return errors diff --git a/linter/checks/endpoints.py b/linter/checks/endpoints.py new file mode 100644 index 00000000..6893b4c7 --- /dev/null +++ b/linter/checks/endpoints.py @@ -0,0 +1,210 @@ +"""Orphaned endpoint detection — cross-references backend routes with usage. + +Extracts all registered API routes from the backend (decorator and add_api_route +patterns) and checks whether each route's static path segments appear in the +frontend source or in other backend files (e.g. MCP servers that call endpoints +internally). Routes with no matching reference anywhere are flagged. + +Limitations (v1): + - Routes that end with a path parameter (e.g. /{id}) and have no trailing + static segment are skipped — they're too ambiguous to match. + - WebSocket routes in main.py are not checked. + - Backend-only endpoints (health checks, OAuth callbacks) should be excluded + via the exceptions list or endpoint-ignore-routes in config.json. +""" + +from __future__ import annotations + +import fnmatch +import re +from pathlib import Path + +from . import is_excepted, is_lintignored + +_DECORATOR_RE = re.compile( + r"@(\w+)\.router\.\w+\(\s*[\"']([^\"']+)[\"']" +) +_ADD_ROUTE_RE = re.compile( + r"(\w+)\.router\.add_api_route\(\s*[\"']([^\"']+)[\"']" +) +_SUBAPP_RE = re.compile( + r"(\w+)\s*=\s*SubApp\(\s*[\"']([^\"']+)[\"']" +) +_FUNC_DEF_RE = re.compile(r"\s*(?:async\s+)?def\s+(\w+)") +_ADD_ROUTE_FUNC_RE = re.compile(r"add_api_route\([^,]+,\s*(?:\w+\.)*(\w+)") + +_TEMPLATE_ASSIGN_RE = re.compile( + r"""(?:export\s+)?(?:const|let|var)\s+(\w+)\s*=\s*`([^`]*)`""" +) +_STRING_ASSIGN_RE = re.compile( + r"""(?:export\s+)?(?:const|let|var)\s+(\w+)\s*=\s*(['"])(.*?)\2""" +) +_TEMPLATE_REF_RE = re.compile(r"\$\{(\w+)\}") + + +def _static_tail(route_path: str) -> str: + """Return the trailing contiguous static segments of a route path. + + >>> _static_tail("/sessions/{id}/message") + '/message' + >>> _static_tail("/usage-summary") + '/usage-summary' + >>> _static_tail("/{id}") + '' + """ + parts = route_path.strip("/").split("/") + tail: list[str] = [] + for part in reversed(parts): + if part.startswith("{"): + break + tail.append(part) + tail.reverse() + return "/" + "/".join(tail) if tail else "" + + +def _resolve_frontend_vars(files: list[tuple[str, str]]) -> dict[str, str]: + """Collect const/let/var string assignments across files and resolve refs. + + Handles patterns like: + const API_BASE = "/api"; + const WORKSPACE_API = `${API_BASE}/outputs/workspace`; + """ + raw: dict[str, str] = {} + for _, text in files: + for m in _STRING_ASSIGN_RE.finditer(text): + raw.setdefault(m.group(1), m.group(3)) + for m in _TEMPLATE_ASSIGN_RE.finditer(text): + raw.setdefault(m.group(1), m.group(2)) + resolved = dict(raw) + for _ in range(5): + changed = False + for name, val in list(resolved.items()): + new_val = _TEMPLATE_REF_RE.sub( + lambda m: resolved.get(m.group(1), m.group(0)), val + ) + if new_val != val: + resolved[name] = new_val + changed = True + if not changed: + break + return resolved + + +def _expand_template_refs(text: str, resolved: dict[str, str]) -> str: + """Replace ``${VAR}`` references in *text* with resolved values.""" + return _TEMPLATE_REF_RE.sub( + lambda m: resolved.get(m.group(1), m.group(0)), text + ) + + +def _find_func_name(lines: list[str], decorator_idx: int) -> str: + for j in range(decorator_idx + 1, min(decorator_idx + 5, len(lines))): + m = _FUNC_DEF_RE.match(lines[j]) + if m: + return m.group(1) + return "" + + +def run_endpoint_check( + root: Path, + exceptions: dict[str, list[str]], + ignore_routes: list[str] | None = None, + ignores: dict[Path, set[str]] | None = None, +) -> list[str]: + """Find backend API endpoints with no matching frontend or backend reference.""" + backend_dir = root / "backend" + frontend_dir = root / "frontend" / "src" + if not backend_dir.exists() or not frontend_dir.exists(): + return [] + + _ignore_routes = ignore_routes or [] + var_to_name: dict[str, str] = {} + for py in backend_dir.rglob("*.py"): + if ".venv" in py.parts: + continue + for m in _SUBAPP_RE.finditer(py.read_text(errors="ignore")): + var_to_name[m.group(1)] = m.group(2) + + routes: list[tuple[str, str, str, int, str]] = [] + + for py in backend_dir.rglob("*.py"): + if ".venv" in py.parts: + continue + text = py.read_text(errors="ignore") + lines = text.splitlines() + rel = str(py.relative_to(root)) + + for i, line in enumerate(lines): + m = _DECORATOR_RE.search(line) + if m: + var, path = m.group(1), m.group(2) + name = var_to_name.get(var) + if name: + func = _find_func_name(lines, i) + routes.append((name, path, rel, i + 1, func)) + + m2 = _ADD_ROUTE_RE.search(line) + if m2: + var, path = m2.group(1), m2.group(2) + name = var_to_name.get(var) + if name: + fm = _ADD_ROUTE_FUNC_RE.search(line) + func = fm.group(1) if fm else "" + routes.append((name, path, rel, i + 1, func)) + + frontend_files: list[tuple[str, str]] = [] + for ext in ("*.ts", "*.tsx"): + for f in frontend_dir.rglob(ext): + frontend_files.append((str(f.relative_to(root)), f.read_text(errors="ignore"))) + + backend_files: list[tuple[str, str]] = [] + for py in backend_dir.rglob("*.py"): + if ".venv" in py.parts: + continue + backend_files.append((str(py.relative_to(root)), py.read_text(errors="ignore"))) + + resolved_vars = _resolve_frontend_vars(frontend_files) + + errors: list[str] = [] + for subapp_name, route_path, filepath, lineno, func_name in routes: + if is_excepted(filepath, "endpoints", exceptions): + continue + if ignores and is_lintignored(root / filepath, root, "endpoints", ignores): + continue + + full_path = f"{subapp_name}{route_path}" + + if any(fnmatch.fnmatch(full_path, p) for p in _ignore_routes): + continue + tail = _static_tail(route_path) + + if not tail: + continue + + found = False + for _fe_path, fe_text in frontend_files: + expanded = _expand_template_refs(fe_text, resolved_vars) + if full_path in expanded: + found = True + break + if subapp_name in expanded and tail in expanded: + found = True + break + + if not found: + for be_path, be_text in backend_files: + if be_path == filepath: + continue + if full_path in be_text: + found = True + break + + if not found: + label = func_name or route_path + errors.append( + f"{filepath}:{lineno}:1: warning: " + f"[endpoints] orphaned endpoint '{label}' " + f"(/api/{full_path}) — no frontend or backend reference found" + ) + + return sorted(errors) diff --git a/linter/checks/eslint.py b/linter/checks/eslint.py new file mode 100644 index 00000000..ebd730d9 --- /dev/null +++ b/linter/checks/eslint.py @@ -0,0 +1,50 @@ +"""ESLint runner for the TypeScript frontend.""" + +from __future__ import annotations + +import json +import subprocess +from pathlib import Path + +from . import is_lintignored + + +def run_eslint(root: Path, ignores: dict[Path, set[str]] | None = None) -> list[str]: + """Run ESLint on the TypeScript frontend and return errors.""" + frontend_dir = root / "frontend" + eslint_bin = frontend_dir / "node_modules" / ".bin" / "eslint" + if not eslint_bin.exists(): + return [] + + cmd = [str(eslint_bin), "src/", "--format", "json", "--no-warn-ignored"] + try: + result = subprocess.run( + cmd, capture_output=True, text=True, + cwd=str(frontend_dir), timeout=60, + ) + except (OSError, subprocess.TimeoutExpired): + return [] + + try: + data = json.loads(result.stdout) + except (json.JSONDecodeError, ValueError): + return [] + + errors: list[str] = [] + for entry in data: + try: + filepath = Path(entry["filePath"]) + rel = str(filepath.relative_to(root)) + except (ValueError, KeyError): + continue + if ignores and is_lintignored(filepath, root, "eslint", ignores): + continue + for msg in entry.get("messages", []): + sev = "error" if msg.get("severity", 0) >= 2 else "warning" + text = msg.get("message", "").replace("\n", " ").strip() + rule = msg.get("ruleId") or "unknown" + errors.append( + f"{rel}:{msg.get('line', 1)}:{msg.get('column', 1)}: " + f"{sev}: [eslint] {text} ({rule})" + ) + return errors diff --git a/linter/checks/knip.py b/linter/checks/knip.py new file mode 100644 index 00000000..08c34fec --- /dev/null +++ b/linter/checks/knip.py @@ -0,0 +1,66 @@ +"""Knip unused-code runner for the TypeScript frontend.""" + +from __future__ import annotations + +import json +import subprocess +from pathlib import Path + +from . import is_lintignored + +KIND_LABELS = { + "dependencies": "Unused dependency", + "devDependencies": "Unused devDependency", + "exports": "Unused export", + "types": "Unused exported type", + "unlisted": "Unlisted dependency", + "binaries": "Unused binary", + "files": "Unused file", + "duplicates": "Duplicate export", +} + + +def run_knip(root: Path, ignores: dict[Path, set[str]] | None = None) -> list[str]: + """Run Knip on the TypeScript frontend and return errors.""" + frontend_dir = root / "frontend" + knip_bin = frontend_dir / "node_modules" / ".bin" / "knip" + if not knip_bin.exists(): + return [] + + cmd = [str(knip_bin), "--reporter", "json"] + try: + result = subprocess.run( + cmd, capture_output=True, text=True, + cwd=str(frontend_dir), timeout=60, + ) + except (OSError, subprocess.TimeoutExpired): + return [] + + try: + data = json.loads(result.stdout) + except (json.JSONDecodeError, ValueError): + return [] + + errors: list[str] = [] + for entry in data.get("issues", []): + filepath = entry.get("file", "") + rel = f"frontend/{filepath}" + abs_path = root / rel + if ignores and is_lintignored(abs_path, root, "knip", ignores): + continue + for kind, label in KIND_LABELS.items(): + for item in entry.get(kind, []): + if isinstance(item, dict): + name = item.get("name", "") + line = item.get("line", 1) + col = item.get("col", 1) + elif isinstance(item, str): + name = item + line, col = 1, 1 + else: + continue + errors.append( + f"{rel}:{line}:{col}: error: " + f"[knip] {label} '{name}'" + ) + return errors diff --git a/linter/checks/structural.py b/linter/checks/structural.py new file mode 100644 index 00000000..68d41f84 --- /dev/null +++ b/linter/checks/structural.py @@ -0,0 +1,104 @@ +"""Structural checks: file length, folder size, and nested imports.""" + +from __future__ import annotations + +import ast +from pathlib import Path + +from . import _matches_any + +ANCHOR_FILES = ("__init__.py", "index.ts", "index.tsx", "index.js") + + +def _find_anchor_file(dirpath: Path, root: Path) -> str: + """Find a real file inside the folder to attach the diagnostic to. + + Prefers common entry-point files (__init__.py, index.ts, etc.) so the + error shows up inline when you open that file. Falls back to the first + file alphabetically, then the directory path itself. + """ + for name in ANCHOR_FILES: + candidate = dirpath / name + if candidate.exists(): + return str(candidate.relative_to(root)) + try: + first = sorted( + f for f in dirpath.iterdir() + if f.is_file() and not f.name.startswith(".") + ) + if first: + return str(first[0].relative_to(root)) + except OSError: + pass + return str(dirpath.relative_to(root)) + + +def check_file_lines( + filepath: Path, root: Path, max_lines: int, +) -> tuple[str, int] | None: + try: + count = len(filepath.read_text(errors="ignore").splitlines()) + except OSError: + return None + if count >= max_lines: + rel = filepath.relative_to(root) + msg = ( + f"{rel}:1:1: error: " + f"[max-file-lines] File has {count} lines (limit {max_lines})" + ) + return (msg, count) + return None + + +def check_folder_items( + dirpath: Path, root: Path, max_items: int, excludes: list[str], +) -> tuple[str, int] | None: + try: + items = [ + i for i in dirpath.iterdir() + if not i.name.startswith(".") and not _matches_any(i.name, excludes) + ] + except OSError: + return None + count = len(items) + if count >= max_items: + anchor = _find_anchor_file(dirpath, root) + rel = dirpath.relative_to(root) + msg = ( + f"{anchor}:1:1: error: " + f"[max-folder-items] Folder '{rel}' has {count} items (limit {max_items})" + ) + return (msg, count) + return None + + +def check_nested_imports(filepath: Path, root: Path) -> list[str]: + """Detect import statements inside function or method bodies.""" + if filepath.suffix != ".py": + return [] + try: + source = filepath.read_text(errors="ignore") + tree = ast.parse(source, filename=str(filepath)) + except (OSError, SyntaxError): + return [] + + errors: list[str] = [] + rel = str(filepath.relative_to(root)) + + def _visit(node: ast.AST, in_function: bool) -> None: + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + in_function = True + if in_function and isinstance(node, (ast.Import, ast.ImportFrom)): + if isinstance(node, ast.ImportFrom): + name = node.module or "" + else: + name = ", ".join(a.name for a in node.names) + errors.append( + f"{rel}:{node.lineno}:1: error: " + f"[no-nested-imports] Nested import '{name}'" + ) + for child in ast.iter_child_nodes(node): + _visit(child, in_function) + + _visit(tree, False) + return errors diff --git a/linter/checks/vulture.py b/linter/checks/vulture.py new file mode 100644 index 00000000..5d0acfce --- /dev/null +++ b/linter/checks/vulture.py @@ -0,0 +1,95 @@ +"""Vulture dead-code detection runner. + +Class-body findings (fields, methods inside a class) are filtered out here +and handled separately by checks/classes.py which understands Pydantic. +""" + +from __future__ import annotations + +import ast +import re +import shutil +import subprocess +from functools import lru_cache +from pathlib import Path + +from . import is_excepted, is_lintignored + +CONFIG_DIR = Path(__file__).resolve().parent.parent / "config" + + +@lru_cache(maxsize=64) +def _class_line_ranges(filepath: str) -> list[tuple[int, int]]: + """Return (start, end) line ranges for all class bodies in *filepath*.""" + try: + tree = ast.parse(Path(filepath).read_text()) + except (OSError, SyntaxError): + return [] + ranges: list[tuple[int, int]] = [] + for node in ast.walk(tree): + if isinstance(node, ast.ClassDef): + end = max(getattr(n, "lineno", node.lineno) for n in ast.walk(node)) + ranges.append((node.lineno, end)) + return ranges + + +def _is_inside_class(filepath: str, lineno: int) -> bool: + """True when *lineno* is strictly inside a class body. + + The class declaration line itself (``class Foo:``) is *not* considered + inside, so vulture's "unused class" findings still pass through. + """ + return any(start < lineno <= end for start, end in _class_line_ranges(filepath)) + + +def run_vulture( + root: Path, min_confidence: int, error_threshold: int, + exceptions: dict[str, list[str]], + ignores: dict[Path, set[str]] | None = None, +) -> list[str]: + """Run vulture on the Python backend and return errors.""" + vulture_bin = root / "backend" / ".venv" / "bin" / "vulture" + if not vulture_bin.exists(): + found = shutil.which("vulture") + if not found: + return [] + vulture_bin = Path(found) + + whitelist = CONFIG_DIR / "vulture_whitelist.py" + targets = ["backend"] + if (root / "debug.py").exists(): + targets.append("debug.py") + cmd = [str(vulture_bin), *targets] + if whitelist.exists(): + cmd.append(str(whitelist)) + cmd.extend([ + "--min-confidence", str(min_confidence), + "--exclude", ".venv,__pycache__,data,uv-bin", + "--ignore-decorators", "@*.router.*,@*.websocket,@app.*,@pytest.fixture,@pytest.fixture*", + "--ignore-names", "cls", + ]) + + try: + result = subprocess.run( + cmd, capture_output=True, text=True, cwd=str(root), timeout=30, + ) + except (OSError, subprocess.TimeoutExpired): + return [] + + errors: list[str] = [] + for line in result.stdout.strip().splitlines(): + m = re.match(r"^(.+):(\d+): (.+)$", line) + if not m: + continue + filepath, lineno, message = m.groups() + if is_excepted(filepath, "vulture", exceptions): + continue + if ignores and is_lintignored(root / filepath, root, "vulture", ignores): + continue + if _is_inside_class(str(root / filepath), int(lineno)): + continue + conf = re.search(r"\((\d+)% confidence\)", message) + confidence = int(conf.group(1)) if conf else 0 + severity = "error" if confidence >= error_threshold else "warning" + errors.append(f"{filepath}:{lineno}:1: {severity}: [vulture] {message}") + return errors diff --git a/linter/config/config.json b/linter/config/config.json new file mode 100644 index 00000000..c27b7219 --- /dev/null +++ b/linter/config/config.json @@ -0,0 +1,139 @@ +{ + "enabled": { + "max-file-lines": true, + "max-folder-items": true, + "no-nested-imports": false, + "vulture": true, + "eslint": false, + "knip": false, + "endpoints": false, + "classes": false + }, + "_notes": { + "no-nested-imports": "Off on purpose: this codebase uses function-level/lazy imports to break import cycles (400+ sites). Flagging them all is wrong for us.", + "eslint-knip": "Node tooling deferred to a later pass.", + "classes": "Placeholder check, not wired up. endpoints: orphaned-endpoint triage deferred." + }, + "rules": { + "max-file-lines": 300, + "max-folder-items": 7, + "vulture-min-confidence": 1, + "vulture-error-threshold": 1, + "no-nested-imports": true, + "endpoint-ignore-routes": ["*/callback", "*/callback/*"] + }, + "include_extensions": [".py", ".ts", ".tsx", ".js", ".jsx"], + "exclude": [ + "node_modules", + ".venv", + "dist", + "build", + "__pycache__", + ".git", + ".cursor", + ".vscode", + ".claude", + ".pytest_cache", + ".mypy_cache", + ".ruff_cache", + ".account-factory", + ".local-stash", + ".DS_Store", + "python-env", + "build-staging", + "webapp_template_cache", + "openswarm-cloud", + "uv-bin", + "data", + "public", + "openswarm_debug.egg-info" + ], + "exceptions": { + "max-file-lines": [ + "backend/apps/agents/agent_manager.py", + "backend/apps/agents/agents.py", + "backend/apps/agents/anthropic_proxy.py", + "backend/apps/agents/anthropic_to_openai.py", + "backend/apps/agents/browser_agent.py", + "backend/apps/agents/browser_agent_mcp_server.py", + "backend/apps/agents/browser_schema.py", + "backend/apps/agents/prompt_context.py", + "backend/apps/agents/providers/pricing.py", + "backend/apps/agents/providers/registry.py", + "backend/apps/dashboards/dashboards.py", + "backend/apps/discord_mcp_shim/server.py", + "backend/apps/mcp_registry/mcp_registry.py", + "backend/apps/nine_router/oauth.py", + "backend/apps/nine_router/process.py", + "backend/apps/nine_router/sync_custom.py", + "backend/apps/outputs/outputs.py", + "backend/apps/outputs/runtime.py", + "backend/apps/outputs/view_builder_templates.py", + "backend/apps/outputs/webapp_template/frontend/src/shared/styles/ThemeContext.tsx", + "backend/apps/service/client.py", + "backend/apps/service/service.py", + "backend/apps/settings/settings.py", + "backend/apps/skills/skills.py", + "backend/apps/subscription/router.py", + "backend/apps/tools_lib/tools_lib.py", + "backend/apps/web/web.py", + "backend/main.py", + "backend/tests/test_disconnect_resilience.py", + "backend/tests/test_outputs_runtime_cleanup.py", + "backend/tests/test_service.py", + "backend/tests/test_v2_invariants.py", + "backend/tests/test_v2_label_logic.py", + "electron/affiliateTracking.test.js", + "electron/main.js", + "frontend/src/app/Main.tsx", + "frontend/src/app/components/CommandPicker.tsx", + "frontend/src/app/components/DirectoryBrowser.tsx", + "frontend/src/app/components/DynamicIsland.tsx", + "frontend/src/app/components/GlobalSearchPalette.tsx", + "frontend/src/app/components/Layout/AppShell.tsx", + "frontend/src/app/components/Onboarding/OnboardingPanel.tsx", + "frontend/src/app/components/Onboarding/ac/AgenticCursor.tsx", + "frontend/src/app/components/Onboarding/ac/acRuntime.ts", + "frontend/src/app/components/PlanPicker.tsx", + "frontend/src/app/components/RichPromptEditor.tsx", + "frontend/src/app/components/SignInGate.tsx", + "frontend/src/app/components/useDomElementSelector.ts", + "frontend/src/app/pages/AgentChat/AgentChat.tsx", + "frontend/src/app/pages/AgentChat/ApprovalBar.tsx", + "frontend/src/app/pages/AgentChat/BrowserAgentInlineFeed.tsx", + "frontend/src/app/pages/AgentChat/ChatInput.tsx", + "frontend/src/app/pages/AgentChat/MessageBubble.tsx", + "frontend/src/app/pages/AgentChat/ToolCallBubble.tsx", + "frontend/src/app/pages/AgentChat/toolLabels.ts", + "frontend/src/app/pages/Analytics/PixelChart.tsx", + "frontend/src/app/pages/Commands/Commands.tsx", + "frontend/src/app/pages/Dashboard/AgentCard.tsx", + "frontend/src/app/pages/Dashboard/BrowserAgentOverlay.tsx", + "frontend/src/app/pages/Dashboard/BrowserCard.tsx", + "frontend/src/app/pages/Dashboard/Dashboard.tsx", + "frontend/src/app/pages/Dashboard/DashboardHeader.tsx", + "frontend/src/app/pages/Dashboard/DashboardToolbar.tsx", + "frontend/src/app/pages/Dashboard/DashboardViewCard.tsx", + "frontend/src/app/pages/Dashboard/NoteCard.tsx", + "frontend/src/app/pages/Dashboard/useCanvasControls.ts", + "frontend/src/app/pages/DashboardSelection/DashboardSelection.tsx", + "frontend/src/app/pages/Modes/Modes.tsx", + "frontend/src/app/pages/Settings/Settings.tsx", + "frontend/src/app/pages/Skills/SkillBuilderChat.tsx", + "frontend/src/app/pages/Skills/Skills.tsx", + "frontend/src/app/pages/Tools/Tools.tsx", + "frontend/src/app/pages/Views/ViewEditor.tsx", + "frontend/src/app/pages/Views/ViewPreview.tsx", + "frontend/src/app/pages/Views/useIframeElementSelector.ts", + "frontend/src/shared/browserCommandHandler.ts", + "frontend/src/shared/state/agentsSlice.ts", + "frontend/src/shared/state/dashboardLayoutSlice.ts", + "frontend/src/shared/ws/WebSocketManager.ts" + ], + "max-folder-items": [], + "no-nested-imports": [], + "vulture": ["backend/tests/*"], + "endpoints": [], + "classes": [] + } +} diff --git a/linter/config/pyrightconfig.json b/linter/config/pyrightconfig.json new file mode 100644 index 00000000..96fe0ba2 --- /dev/null +++ b/linter/config/pyrightconfig.json @@ -0,0 +1,14 @@ +{ + "include": ["../../backend"], + "exclude": [ + "../../backend/.venv", + "../../backend/uv-bin", + "../../backend/data", + "../../backend/tests" + ], + "typeCheckingMode": "strict", + "pythonVersion": "3.11", + "venvPath": "../../backend", + "venv": ".venv", + "reportMissingTypeStubs": false +} diff --git a/linter/lint.py b/linter/lint.py new file mode 100644 index 00000000..5186762b --- /dev/null +++ b/linter/lint.py @@ -0,0 +1,170 @@ +#!/usr/bin/env python3 +"""Unified linter: orchestrates structural checks, dead-code detection, and lint tools.""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +from pathlib import Path +from typing import Any + +from checks import is_excluded, is_excepted, is_lintignored, collect_lintignores +from checks.structural import check_file_lines, check_folder_items, check_nested_imports +from checks.vulture import run_vulture +from checks.eslint import run_eslint +from checks.knip import run_knip +from checks.endpoints import run_endpoint_check +from checks.classes import run_class_check +from watchfiles import watch, DefaultFilter + +SCRIPT_DIR = Path(__file__).resolve().parent +CONFIG_FILE = SCRIPT_DIR / "config" / "config.json" + + +def load_config() -> dict[str, Any]: + with open(CONFIG_FILE) as f: + return json.load(f) + + +def run_checks(root: Path) -> tuple[list[str], list[str], list[str], list[str], list[str], list[str]]: + config = load_config() + enabled: dict[str, bool] = config.get("enabled", {}) + rules: dict[str, int] = config["rules"] + excludes: list[str] = config["exclude"] + exceptions: dict[str, list[str]] = config["exceptions"] + extensions: list[str] = config["include_extensions"] + ignores = collect_lintignores(root, excludes) + + max_lines: int = rules["max-file-lines"] + max_items: int = rules["max-folder-items"] + check_imports: bool = rules.get("no-nested-imports", False) + structural_errors: list[str] = [] + + file_lines_on = enabled.get("max-file-lines", True) + folder_items_on = enabled.get("max-folder-items", True) + nested_imports_on = enabled.get("no-nested-imports", True) + + for dirpath_str, dirnames, filenames in os.walk(root): + dp = Path(dirpath_str) + + if is_excluded(dp, root, excludes): + dirnames.clear() + continue + + rel_dir = str(dp.relative_to(root)) + if ( + folder_items_on + and rel_dir != "." + and not is_excepted(rel_dir, "max-folder-items", exceptions) + and not is_lintignored(dp, root, "max-folder-items", ignores) + ): + result = check_folder_items(dp, root, max_items, excludes) + if result: + structural_errors.append(result[0]) + + for fname in filenames: + fp = dp / fname + if fp.suffix not in extensions: + continue + if is_excluded(fp, root, excludes): + continue + rel_file = str(fp.relative_to(root)) + if ( + file_lines_on + and not is_excepted(rel_file, "max-file-lines", exceptions) + and not is_lintignored(fp, root, "max-file-lines", ignores) + ): + result = check_file_lines(fp, root, max_lines) + if result: + structural_errors.append(result[0]) + if ( + nested_imports_on + and check_imports + and not is_excepted(rel_file, "no-nested-imports", exceptions) + and not is_lintignored(fp, root, "no-nested-imports", ignores) + ): + structural_errors.extend(check_nested_imports(fp, root)) + + vulture_errors: list[str] = [] + if enabled.get("vulture", True): + vulture_confidence = rules.get("vulture-min-confidence") + if vulture_confidence is not None: + vulture_error_threshold = rules.get("vulture-error-threshold", 100) + vulture_errors = run_vulture( + root, vulture_confidence, vulture_error_threshold, exceptions, ignores, + ) + + eslint_errors = run_eslint(root, ignores) if enabled.get("eslint", True) else [] + knip_errors = run_knip(root, ignores) if enabled.get("knip", True) else [] + endpoint_ignore_routes: list[str] = rules.get("endpoint-ignore-routes", []) + endpoint_errors = run_endpoint_check(root, exceptions, endpoint_ignore_routes, ignores) if enabled.get("endpoints", True) else [] + class_errors = run_class_check(root, exceptions, excludes, ignores) if enabled.get("classes", True) else [] + + return sorted(structural_errors), sorted(vulture_errors), sorted(eslint_errors), sorted(knip_errors), sorted(endpoint_errors), sorted(class_errors) + + +def _print_section(name: str, errors: list[str]) -> None: + print(f"{name}: checking...", flush=True) + for e in errors: + print(e, flush=True) + print(f"{name}: done. {len(errors)} error(s) found.", flush=True) + + +def print_results( + structural_errors: list[str], vulture_errors: list[str], + eslint_errors: list[str], knip_errors: list[str], + endpoint_errors: list[str], class_errors: list[str], +) -> None: + _print_section("structural", structural_errors) + _print_section("vulture", vulture_errors) + _print_section("eslint", eslint_errors) + _print_section("knip", knip_errors) + _print_section("endpoints", endpoint_errors) + _print_section("classes", class_errors) + + +def watch_loop(root: Path) -> None: + + config_dir = SCRIPT_DIR / "config" + + print_results(*run_checks(root)) + + class SourceFilter(DefaultFilter): + allowed_extensions = (".py", ".ts", ".tsx", ".js", ".jsx") + + def __call__(self, change: Any, path: str) -> bool: + if not super().__call__(change, path): + return False + if Path(path).suffix in self.allowed_extensions: + return True + p = Path(path) + if p.suffix == ".json" and (p.parent == SCRIPT_DIR or p.parent == config_dir): + return True + if p.name.startswith(".lintignore"): + return True + return Path(path).is_dir() + + for _changes in watch(root, watch_filter=SourceFilter()): + print_results(*run_checks(root)) + + +def main() -> None: + parser = argparse.ArgumentParser(description="Unified linter") + parser.add_argument("--watch", action="store_true", help="Watch for changes") + parser.add_argument("--root", type=str, default=".", help="Root directory") + args = parser.parse_args() + + root = Path(args.root).resolve() + + if args.watch: + watch_loop(root) + else: + results = run_checks(root) + print_results(*results) + sys.exit(1 if any(results) else 0) + + +if __name__ == "__main__": + main() diff --git a/linter/print_errors.sh b/linter/print_errors.sh new file mode 100755 index 00000000..ac6c8d89 --- /dev/null +++ b/linter/print_errors.sh @@ -0,0 +1,64 @@ +#!/bin/bash +# Print lint violations to stdout with colored formatting. +# Usage: bash linter/print_errors.sh [ROOT_DIR] + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +ROOT_DIR="${1:-$(dirname "$SCRIPT_DIR")}" + +YELLOW='\033[33m' +CYAN='\033[36m' +BOLD='\033[1m' +RESET='\033[0m' + +LINT_OUTPUT=$(python3 "$SCRIPT_DIR/lint.py" --root "$ROOT_DIR" 2>&1) +LINT_EXIT=$? + +if [ $LINT_EXIT -ne 0 ]; then + STRUCT_LINES=$(echo "$LINT_OUTPUT" | grep -v "^structural:" | grep -v "^vulture:" | grep -v "^eslint:" | grep -v "^knip:" | grep -v '\[vulture\]' | grep -v '\[eslint\]' | grep -v '\[knip\]') + VULTURE_LINES=$(echo "$LINT_OUTPUT" | grep '\[vulture\]') + ESLINT_LINES=$(echo "$LINT_OUTPUT" | grep '\[eslint\]') + KNIP_LINES=$(echo "$LINT_OUTPUT" | grep '\[knip\]') + + STRUCT_COUNT=$(echo "$STRUCT_LINES" | grep -cE ':\s+(error|warning):\s+') + VULTURE_COUNT=$(echo "$VULTURE_LINES" | grep -cE ':\s+(error|warning):\s+') + ESLINT_COUNT=$(echo "$ESLINT_LINES" | grep -cE ':\s+(error|warning):\s+') + KNIP_COUNT=$(echo "$KNIP_LINES" | grep -cE ':\s+(error|warning):\s+') + + if [ "$STRUCT_COUNT" -gt 0 ]; then + echo "" + echo -e "${YELLOW}${BOLD}[structural] Violations found:${RESET}" + echo "$STRUCT_LINES" | while IFS= read -r line; do + [ -n "$line" ] && echo -e "${YELLOW} $line${RESET}" + done + echo -e "${YELLOW}${BOLD} ${STRUCT_COUNT} violation(s) — fix or add exceptions in linter/config/config.json${RESET}" + fi + + if [ "$VULTURE_COUNT" -gt 0 ]; then + echo "" + echo -e "${CYAN}${BOLD}[vulture] Dead code found:${RESET}" + echo "$VULTURE_LINES" | while IFS= read -r line; do + [ -n "$line" ] && echo -e "${CYAN} $line${RESET}" + done + echo -e "${CYAN}${BOLD} ${VULTURE_COUNT} finding(s) — fix or add to linter/config/vulture_whitelist.py${RESET}" + fi + + if [ "$ESLINT_COUNT" -gt 0 ]; then + echo "" + echo -e "${YELLOW}${BOLD}[eslint] Lint errors found:${RESET}" + echo "$ESLINT_LINES" | while IFS= read -r line; do + [ -n "$line" ] && echo -e "${YELLOW} $line${RESET}" + done + echo -e "${YELLOW}${BOLD} ${ESLINT_COUNT} error(s) — fix or disable rules in frontend/eslint.config.mjs${RESET}" + fi + + if [ "$KNIP_COUNT" -gt 0 ]; then + echo "" + echo -e "${CYAN}${BOLD}[knip] Unused code/dependencies found:${RESET}" + echo "$KNIP_LINES" | while IFS= read -r line; do + [ -n "$line" ] && echo -e "${CYAN} $line${RESET}" + done + echo -e "${CYAN}${BOLD} ${KNIP_COUNT} finding(s) — remove unused code or update frontend/knip.json${RESET}" + fi + + echo "" +fi