From c55845c3140f93694b99e6f23c97acaf5bf7b2d6 Mon Sep 17 00:00:00 2001 From: worker-2-a Date: Sun, 28 Jun 2026 06:09:46 +0000 Subject: [PATCH 1/4] feat(arch): add architecture_engine.py with overview aggregation New engine module orchestrating existing engines (framework_detect, entrypoints_engine, apimap_engine, graph_model) into a single compact codebase overview payload. Addresses issue #19. Public API: - get_architecture(workspace, lite=False) -> dict Pipeline: - _compute_languages: three-tier resolution (fresh scan_result's files_scanned -> .codelens/summary.json -> cheap stat-only extension walk). Scan buckets collapsed to canonical names (js_backend + js_frontend -> javascript). - _compute_frameworks: wraps framework_detect.detect_frameworks. - _compute_entry_points: wraps entrypoints_engine.map_entrypoints, dedupes by file path, prioritises main/handler/cli types (top 10). - _compute_packages: scans src/, app/, lib/, packages/, server/, internal/ for immediate subdirs that contain source files (top 20). - _compute_routes: wraps apimap_engine.map_api_routes, normalises to {method, path, handler} (top 20). - _compute_hotspots: single SQL round-trip against graph_edges JOIN graph_nodes GROUP BY file ORDER BY count DESC LIMIT 5. Killer feature of the new graph model: files ranked by total incoming CALLS edges. - _compute_total_symbols: graph_stats(db_path).nodes. Caching: - First call writes .codelens/architecture_cache.json with timestamp. - Subsequent calls return the cache as long as .codelens/codelens.db mtime hasn't advanced (i.e. scan hasn't been re-run). - Lite and full payloads have different shapes -- the cache won't serve the wrong shape even when the db is unchanged. Auto-scan: - If .codelens/codelens.db doesn't exist or graph tables are empty when architecture is called, the engine runs cmd_scan first. This makes the tool self-sufficient for the issue #19 use-case ('agent starts on an unfamiliar codebase'). Payload shape: - Architecture-specific fields nested inside a 'stats' dict so they survive the MCP _normalize_to_ai formatter (which preserves stats as-is but drops unknown top-level keys). CLI consumers see the same shape. Adrs field is an empty list placeholder (ADR feature is issue #16, Phase 3). --- scripts/architecture_engine.py | 677 +++++++++++++++++++++++++++++++++ 1 file changed, 677 insertions(+) create mode 100644 scripts/architecture_engine.py diff --git a/scripts/architecture_engine.py b/scripts/architecture_engine.py new file mode 100644 index 00000000..83301d78 --- /dev/null +++ b/scripts/architecture_engine.py @@ -0,0 +1,677 @@ +""" +Architecture Engine — single-call codebase overview for AI agents (issue #19). + +When an agent lands on an unfamiliar codebase it normally has to chain 4-6 +commands (scan -> list -> detect -> entrypoints -> api-map -> read entry files) +just to figure out what it is looking at. That orientation phase burns 10-20k +tokens before any real work starts. + +This engine orchestrates the existing engines (framework_detect, entrypoints, +apimap, graph_model) plus a lightweight top-level package scan into a single +compact summary so the agent can orient in one call: + + { + "languages": {"python": 342, "typescript": 89}, + "frameworks": ["fastapi", "react"], + "entry_points": ["src/main.py", "src/server.ts"], + "packages": ["src/api", "src/models", "src/services"], + "routes": [{"method": "GET", "path": "/users", "handler": "get_users"}], + "hotspots": ["src/models/user.py (47 dependents)"], + "total_symbols": 1842, + "adrs": [] + } + +The `--lite` flag omits routes / packages / hotspots to keep the payload under +~1k tokens for the cheapest possible orientation call. + +Caching: the first call writes `.codelens/architecture_cache.json` with a +timestamp. Subsequent calls return the cached payload as long as +`.codelens/codelens.db` hasn't been touched since (i.e. scan hasn't been +re-run). This makes repeated orientation calls instant. +""" + +import json +import os +import sqlite3 +import time +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +from utils import DEFAULT_IGNORE_DIRS, logger + + +# ─── Constants ──────────────────────────────────────────────── + +# Source-file extensions used to decide whether a top-level directory counts +# as a "package" (contains real source code, not just assets/docs). +_SOURCE_EXTENSIONS = { + ".py", ".js", ".mjs", ".cjs", ".ts", ".tsx", ".jsx", + ".rs", ".go", ".java", ".kt", ".rb", ".php", + ".c", ".h", ".cpp", ".cc", ".cxx", ".hpp", ".hxx", + ".vue", ".svelte", ".swift", ".scala", ".dart", + ".ex", ".exs", ".lua", ".cs", ".zig", +} + +# Directories whose direct children represent logical packages / modules. +_PACKAGE_ROOT_DIRS = ("src", "app", "lib", "packages", "server", "internal") + +# Limits to keep the JSON payload small (issue target: <1k tokens in --lite). +_MAX_ENTRY_POINTS = 10 +_MAX_PACKAGES = 20 +_MAX_ROUTES = 20 +_MAX_HOTSPOTS = 5 + +# Map scan result `files_scanned` keys to canonical language names used in +# the architecture payload. Multiple scan buckets may collapse into one +# language (e.g. js_backend + js_frontend + mjs/cjs all become "javascript"). +_SCAN_BUCKET_TO_LANGUAGE: Dict[str, str] = { + "python": "python", + "js_frontend": "javascript", + "js_backend": "javascript", + "tsx": "typescript", + "rust": "rust", + "vue": "vue", + "svelte": "svelte", + "java": "java", + "kotlin": "kotlin", + "c_cpp": "c_cpp", + "go": "go", + "lua": "lua", + "csharp": "csharp", + "php": "php", + "blade": "blade", + "ruby": "ruby", + "elixir": "elixir", + "dart": "dart", + "swift": "swift", + "scala": "scala", + "shell": "shell", + "gdscript": "gdscript", + "objc": "objective_c", + "html": "html", + "css": "css", +} + +# Map file extension to canonical language name for the cheap stat-only +# fallback walk (used when neither scan_result nor summary.json is available). +_EXT_TO_LANGUAGE: Dict[str, str] = { + ".py": "python", + ".js": "javascript", ".mjs": "javascript", ".cjs": "javascript", + ".ts": "typescript", ".tsx": "typescript", ".jsx": "typescript", + ".rs": "rust", + ".go": "go", + ".java": "java", + ".kt": "kotlin", ".kts": "kotlin", + ".c": "c", ".h": "c", + ".cpp": "cpp", ".cc": "cpp", ".cxx": "cpp", + ".hpp": "cpp", ".hxx": "cpp", + ".cs": "csharp", + ".rb": "ruby", + ".php": "php", + ".vue": "vue", + ".svelte": "svelte", + ".swift": "swift", + ".scala": "scala", ".sc": "scala", + ".dart": "dart", + ".ex": "elixir", ".exs": "elixir", + ".lua": "lua", + ".sh": "shell", ".bash": "shell", ".zsh": "shell", + ".html": "html", ".htm": "html", + ".css": "css", ".scss": "css", ".less": "css", ".sass": "css", + ".zig": "zig", +} + + +# ─── Public API ─────────────────────────────────────────────── + +def get_architecture(workspace: str, lite: bool = False) -> Dict[str, Any]: + """Build (or return cached) single-call codebase overview. + + Orchestrates framework_detect, entrypoints_engine, apimap_engine, and the + graph_model hotspots query into one compact payload. The first call writes + `.codelens/architecture_cache.json`; subsequent calls return the cache as + long as `.codelens/codelens.db` hasn't been modified (i.e. the scan hasn't + been re-run). + + The architecture-specific fields (languages, frameworks, entry_points, + packages, routes, hotspots, total_symbols, adrs) are nested inside a + `stats` dict. This is intentional: the MCP `_normalize_to_ai` formatter + preserves `stats` as-is, so the `codelens_architecture` MCP tool returns + the full architecture data to agents. CLI consumers get the same shape. + + Args: + workspace: Path to the workspace root. Auto-init + auto-scan runs if + `.codelens/codelens.db` doesn't exist yet so the first call works + on a totally unfamiliar codebase. + lite: When True, omit routes / packages / hotspots to keep the output + under ~1k tokens for cheap orientation. + + Returns: + Dict with keys: status, workspace, lite, cached, generated_at, stats. + The `stats` sub-dict holds: languages, frameworks, entry_points, + total_symbols, adrs (always); packages, routes, hotspots (only when + lite=False). + """ + workspace = os.path.abspath(workspace) + codelens_dir = os.path.join(workspace, ".codelens") + db_path = os.path.join(codelens_dir, "codelens.db") + cache_path = os.path.join(codelens_dir, "architecture_cache.json") + + # Auto-scan if the graph database doesn't exist or is empty. This makes + # `architecture` self-sufficient on a fresh codebase (issue #19 spec: + # "single-call codebase overview"). + scan_result: Optional[Dict[str, Any]] = None + if not _graph_is_populated(db_path): + try: + from commands.scan import cmd_scan + scan_result = cmd_scan(workspace, incremental=False) + except Exception: + logger.warning("architecture: auto-scan failed", exc_info=True) + + # Cache lookup — return immediately if scan hasn't been re-run since the + # last architecture build. + cached = _load_cache_if_fresh(cache_path, db_path, lite) + if cached is not None: + return cached + + payload = _build_architecture(workspace, db_path, lite, scan_result) + _save_cache(cache_path, payload) + return payload + + +# ─── Build Pipeline ─────────────────────────────────────────── + +def _build_architecture( + workspace: str, + db_path: str, + lite: bool, + scan_result: Optional[Dict[str, Any]] = None, +) -> Dict[str, Any]: + """Run every sub-engine and assemble the architecture payload. + + The architecture-specific fields (languages, frameworks, entry_points, + packages, routes, hotspots, total_symbols, adrs) are nested inside a + `stats` dict so they survive the MCP normalizer (`_normalize_to_ai` + preserves `stats` as-is). This way the MCP `codelens_architecture` tool + returns the full architecture data to agents, not just an empty + `{stats:{}, items:[]}` shell. + + Args: + workspace: Absolute workspace path. + db_path: Absolute path to .codelens/codelens.db (may not exist). + lite: When True, omit routes / packages / hotspots. + scan_result: Optional scan result dict (from a fresh auto-scan) used + to derive language counts without re-walking files. None when the + scan was done in a prior call. + + Returns: + The full architecture dict (not yet cached). + """ + generated_at = datetime.now(timezone.utc).isoformat() + + languages = _compute_languages(workspace, scan_result) + frameworks = _compute_frameworks(workspace) + entry_points = _compute_entry_points(workspace) + total_symbols = _compute_total_symbols(db_path) + + # Stats holds all architecture-specific fields. The MCP normalizer + # preserves `stats` as-is so agents get the full payload via MCP. + stats: Dict[str, Any] = { + "languages": languages, + "frameworks": frameworks, + "entry_points": entry_points, + "total_symbols": total_symbols, + "adrs": [], # placeholder — ADR feature is issue #16 (Phase 3) + } + + if not lite: + stats["packages"] = _compute_packages(workspace) + stats["routes"] = _compute_routes(workspace) + stats["hotspots"] = _compute_hotspots(db_path) + + payload: Dict[str, Any] = { + "status": "ok", + "workspace": workspace, + "lite": lite, + "cached": False, + "generated_at": generated_at, + "stats": stats, + } + + return payload + + +def _compute_languages( + workspace: str, + scan_result: Optional[Dict[str, Any]] = None, +) -> Dict[str, int]: + """Return a {language_name: file_count} dict. + + Three-tier resolution (avoids re-walking files when possible): + 1. If `scan_result` is provided (fresh auto-scan), use its + `files_scanned` bucket counts — collapsed to canonical language + names (e.g. js_backend + js_frontend -> javascript). + 2. Else if `.codelens/summary.json` exists (written when scan is invoked + via the CLI dispatcher), use its `files_by_language` field. + 3. Else, fall back to a cheap stat-only extension-counting walk — no + file parsing, just `os.walk` + `os.path.splitext`. + + Args: + workspace: Absolute workspace path. + scan_result: Optional scan result dict from a fresh auto-scan. + + Returns: + Dict mapping language name to source file count, sorted by count + descending then name ascending. Empty dict if no source files found. + """ + # Tier 1: fresh scan_result captured by caller. + if scan_result and isinstance(scan_result, dict): + files_scanned = scan_result.get("files_scanned", {}) + if isinstance(files_scanned, dict) and files_scanned: + collapsed: Dict[str, int] = {} + for bucket, count in files_scanned.items(): + if not count: + continue + lang = _SCAN_BUCKET_TO_LANGUAGE.get(str(bucket), str(bucket)) + collapsed[lang] = collapsed.get(lang, 0) + int(count) + if collapsed: + return dict( + sorted(collapsed.items(), key=lambda kv: (-kv[1], kv[0])) + ) + + # Tier 2: summary.json written by CLI dispatcher. + summary_path = os.path.join(workspace, ".codelens", "summary.json") + if os.path.isfile(summary_path): + try: + with open(summary_path, "r", encoding="utf-8") as f: + data = json.load(f) + files_by_lang = data.get("files_by_language", {}) + if isinstance(files_by_lang, dict) and files_by_lang: + cleaned = { + str(k): int(v) for k, v in files_by_lang.items() if v + } + if cleaned: + return dict( + sorted(cleaned.items(), key=lambda kv: (-kv[1], kv[0])) + ) + except (json.JSONDecodeError, OSError): + pass + + # Tier 3: cheap stat-only extension-counting fallback. + return _count_files_by_extension(workspace) + + +def _count_files_by_extension(workspace: str) -> Dict[str, int]: + """Cheap stat-only walk counting source files per language. + + Used as a last-resort fallback when neither scan_result nor summary.json + is available. Does NOT parse files — just walks directories and inspects + file extensions, so it's fast even on large workspaces. + + Args: + workspace: Absolute workspace path. + + Returns: + Dict mapping language name to source file count, sorted by count + descending then name ascending. + """ + counts: Dict[str, int] = {} + for root, dirs, files in os.walk(workspace): + # Prune ignored + dot directories in-place (os.walk lets us mutate + # dirs to skip them on the next iteration). + dirs[:] = [ + d for d in dirs + if d not in DEFAULT_IGNORE_DIRS and not d.startswith(".") + ] + for name in files: + ext = os.path.splitext(name)[1].lower() + lang = _EXT_TO_LANGUAGE.get(ext) + if lang: + counts[lang] = counts.get(lang, 0) + 1 + if not counts: + return {} + return dict(sorted(counts.items(), key=lambda kv: (-kv[1], kv[0]))) + + +def _compute_frameworks(workspace: str) -> List[str]: + """Return detected framework names from framework_detect. + + Args: + workspace: Absolute workspace path. + + Returns: + Sorted list of detected framework name strings (e.g. + ["fastapi", "react"]). Empty list on failure. + """ + try: + from framework_detect import detect_frameworks + result = detect_frameworks(workspace) + frameworks = result.get("frameworks", []) + if isinstance(frameworks, list): + return sorted({str(fw) for fw in frameworks if fw}) + except Exception: + logger.debug("architecture: framework_detect failed", exc_info=True) + return [] + + +def _compute_entry_points(workspace: str) -> List[str]: + """Return up to _MAX_ENTRY_POINTS unique entry-point file paths. + + Uses `entrypoints_engine.map_entrypoints` and extracts the unique + `file` field from each entrypoint dict. The most "important" entry + point types (main, http_handler, cli_command) are prioritised so + test_entry noise doesn't crowd them out. + + Args: + workspace: Absolute workspace path. + + Returns: + List of relative file paths (deduplicated, ordered by importance). + """ + try: + from entrypoints_engine import map_entrypoints + result = map_entrypoints(workspace, exclude_tests=True) + except Exception: + logger.debug("architecture: entrypoints engine failed", exc_info=True) + return [] + + entrypoints = result.get("entrypoints", []) + if not isinstance(entrypoints, list): + return [] + + # Priority order — main/handlers/cli first, then everything else. + type_priority = {"main": 0, "http_handler": 1, "cli_command": 2, + "worker": 3, "cron_job": 4, "event_handler": 5, + "module_export": 6, "test_entry": 7} + ordered = sorted( + entrypoints, + key=lambda e: (type_priority.get(e.get("type", ""), 99), + str(e.get("file", ""))), + ) + + seen = set() + paths: List[str] = [] + for ep in ordered: + f = ep.get("file") + if not f or f in seen: + continue + seen.add(f) + paths.append(f) + if len(paths) >= _MAX_ENTRY_POINTS: + break + return paths + + +def _compute_packages(workspace: str) -> List[str]: + """Return up to _MAX_PACKAGES top-level package directories. + + Scans `src/`, `app/`, `lib/`, `packages/`, `server/`, `internal/` (the + conventional package roots) and lists immediate subdirectories that + contain at least one source file. This is a shallow, fast scan — it + doesn't recurse into nested subdirectories. + + Args: + workspace: Absolute workspace path. + + Returns: + Sorted list of relative paths like "src/api", "src/models". + """ + packages: List[str] = [] + for root_name in _PACKAGE_ROOT_DIRS: + root_dir = os.path.join(workspace, root_name) + if not os.path.isdir(root_dir): + continue + try: + children = sorted(os.listdir(root_dir)) + except OSError: + continue + for child in children: + if child in DEFAULT_IGNORE_DIRS or child.startswith("."): + continue + child_path = os.path.join(root_dir, child) + if not os.path.isdir(child_path): + continue + if _dir_contains_source(child_path): + rel = os.path.relpath(child_path, workspace) + packages.append(rel.replace(os.sep, "/")) + if len(packages) >= _MAX_PACKAGES: + return sorted(packages) + return sorted(packages) + + +def _dir_contains_source(dir_path: str) -> bool: + """Return True if dir_path (or any immediate child file) has source code. + + One level of recursion is allowed so packages organised as + `pkg/__init__.py` or `pkg/sub/module.py` are still detected. Nested + directories beyond depth 2 are not inspected to keep this fast. + + Args: + dir_path: Absolute directory path. + + Returns: + True if any source file exists at depth 1 or 2 under dir_path. + """ + try: + entries = os.listdir(dir_path) + except OSError: + return False + for name in entries: + if name in DEFAULT_IGNORE_DIRS or name.startswith("."): + continue + full = os.path.join(dir_path, name) + if os.path.isfile(full): + _, ext = os.path.splitext(name) + if ext.lower() in _SOURCE_EXTENSIONS: + return True + elif os.path.isdir(full): + # Recurse one level — packages are usually flat two-tier trees. + try: + for sub_name in os.listdir(full): + if sub_name in DEFAULT_IGNORE_DIRS or sub_name.startswith("."): + continue + sub_full = os.path.join(full, sub_name) + if os.path.isfile(sub_full): + _, ext = os.path.splitext(sub_name) + if ext.lower() in _SOURCE_EXTENSIONS: + return True + except OSError: + continue + return False + + +def _compute_routes(workspace: str) -> List[Dict[str, str]]: + """Return up to _MAX_ROUTES routes from the API map engine. + + Each route is normalised to `{method, path, handler}` to keep the JSON + payload compact. The `handler` field falls back to `handler_name` then + `""` depending on which extractor produced the route. + + Args: + workspace: Absolute workspace path. + + Returns: + List of `{method, path, handler}` dicts. Empty list on failure. + """ + try: + from apimap_engine import map_api_routes + result = map_api_routes(workspace, production_only=True) + except Exception: + logger.debug("architecture: apimap engine failed", exc_info=True) + return [] + + routes = result.get("routes", []) + if not isinstance(routes, list): + return [] + + out: List[Dict[str, str]] = [] + for r in routes[:_MAX_ROUTES]: + if not isinstance(r, dict): + continue + method = str(r.get("method", "") or "") + path = str(r.get("path", "") or "") + handler = r.get("handler_name") or r.get("handler") or "" + out.append({"method": method, "path": path, "handler": str(handler)}) + return out + + +def _compute_hotspots(db_path: str) -> List[str]: + """Return up to _MAX_HOTSPOTS files ranked by incoming CALLS edge count. + + Queries `graph_edges` joined to `graph_nodes` and groups by file path so + each hotspot is a distinct file (not a per-symbol row). Files with the + most incoming CALLS edges across all their symbols are surfaced first — + these are the "depended-upon" files where changes have the biggest blast + radius. + + Args: + db_path: Absolute path to .codelens/codelens.db. + + Returns: + List of human-readable strings like "src/models/user.py (47 dependents)". + Empty list if the db/tables don't exist or are empty. + """ + if not os.path.isfile(db_path): + return [] + conn = sqlite3.connect(db_path) + try: + # Group by file (not target_id) so each hotspot is a distinct file — + # one file may host many symbols, each with its own incoming edges, + # and we want the file's total blast radius. + rows = conn.execute( + "SELECT gn.file, COUNT(*) AS cnt " + "FROM graph_edges AS ge " + "INNER JOIN graph_nodes AS gn ON gn.node_id = ge.target_id " + "WHERE ge.target_id IS NOT NULL AND ge.target_id != '' " + " AND gn.file IS NOT NULL AND gn.file != '' " + "GROUP BY gn.file " + "ORDER BY cnt DESC " + "LIMIT ?", + (_MAX_HOTSPOTS,), + ).fetchall() + except sqlite3.Error as e: + logger.debug(f"architecture: hotspots query failed: {e}") + return [] + finally: + conn.close() + + hotspots: List[str] = [] + for file_path, count in rows: + if not file_path: + continue + hotspots.append(f"{file_path} ({count} dependents)") + return hotspots + + +def _compute_total_symbols(db_path: str) -> int: + """Return total symbol count from graph_stats().nodes. + + Args: + db_path: Absolute path to .codelens/codelens.db. + + Returns: + Node count (0 if db/tables don't exist). + """ + try: + from graph_model import graph_stats + return int(graph_stats(db_path).get("nodes", 0)) + except Exception: + return 0 + + +# ─── Cache Layer ────────────────────────────────────────────── + +def _graph_is_populated(db_path: str) -> bool: + """Return True if the graph database exists and has at least one node. + + Args: + db_path: Absolute path to .codelens/codelens.db. + + Returns: + True if the graph is ready to query; False if scan needs to run first. + """ + if not os.path.isfile(db_path): + return False + try: + from graph_model import graph_tables_populated + return graph_tables_populated(db_path) + except Exception: + return False + + +def _load_cache_if_fresh( + cache_path: str, db_path: str, lite: bool, +) -> Optional[Dict[str, Any]]: + """Load cached architecture if it's fresher than the database. + + The cache is invalidated whenever `.codelens/codelens.db` mtime is newer + than the cache mtime (i.e. scan has been re-run since the cache was + written). Also returns None if the cached payload's `lite` flag doesn't + match the requested mode (lite vs full are different shapes). + + Args: + cache_path: Path to .codelens/architecture_cache.json. + db_path: Path to .codelens/codelens.db. + lite: Whether the caller wants the lite payload. + + Returns: + Cached architecture dict with `cached: True`, or None if the cache + is missing/stale/wrong-shape. + """ + if not os.path.isfile(cache_path): + return None + try: + cache_mtime = os.path.getmtime(cache_path) + except OSError: + return None + # If the db has been touched after the cache was written, the cache is + # stale (scan re-ran). If the db doesn't exist (very weird state — scan + # failed mid-build), fall through and rebuild. + if os.path.isfile(db_path): + try: + db_mtime = os.path.getmtime(db_path) + except OSError: + return None + if db_mtime > cache_mtime: + return None + try: + with open(cache_path, "r", encoding="utf-8") as f: + cached = json.load(f) + except (json.JSONDecodeError, OSError): + return None + if not isinstance(cached, dict): + return None + if cached.get("status") != "ok": + return None + if bool(cached.get("lite", False)) != bool(lite): + # Lite and full payloads have different shapes — don't return the + # wrong one; rebuild instead. + return None + cached["cached"] = True + return cached + + +def _save_cache(cache_path: str, payload: Dict[str, Any]) -> None: + """Write the architecture payload to disk for future fast lookups. + + Best-effort: failures are logged but don't fail the call (the payload is + still returned to the caller). + + Args: + cache_path: Path to .codelens/architecture_cache.json. + payload: Architecture dict to cache (will be marked cached=False on + disk so a later load sets cached=True itself). + """ + cache_dir = os.path.dirname(cache_path) + try: + os.makedirs(cache_dir, exist_ok=True) + # The on-disk copy says cached=False; the in-memory caller copy + # already says cached=False too. When _load_cache_if_fresh reads it + # back, it flips cached=True so callers can tell which path served + # the request. + to_write = dict(payload) + to_write["cached"] = False + with open(cache_path, "w", encoding="utf-8") as f: + json.dump(to_write, f, ensure_ascii=False, default=str) + except OSError: + logger.debug("architecture: failed to write cache", exc_info=True) From 78a8b2476ee3ab5f7b3e152af7ff619255feb073 Mon Sep 17 00:00:00 2001 From: worker-2-a Date: Sun, 28 Jun 2026 06:10:02 +0000 Subject: [PATCH 2/4] feat(cmd): add architecture command + codelens_architecture MCP tool CLI command (scripts/commands/architecture.py): - Auto-registered via commands/__init__.py (no manual registration needed). - Flags: --lite (omit routes/packages/hotspots for <1k token orientation), --no-cache (force rebuild by removing architecture_cache.json). - Wraps architecture_engine.get_architecture(workspace, lite=lite). MCP tool (added to _TOOL_DEFINITIONS in scripts/mcp_server.py): - Tool name: codelens_architecture - Schema: {workspace: string} required; {format: string, lite: boolean} optional. - Description documents the issue #19 use-case and the --lite flag for <1k token orientation mode. - Result cached in .codelens/architecture_cache.json and invalidated on re-scan. The MCP _execute_command path runs _normalize_to_ai on every command result. The architecture_engine nests all architecture fields inside a 'stats' dict so they survive normalization intact -- agents calling codelens_architecture get the full overview, not just an empty {stats:{}, items:[]} shell. No other changes to mcp_server.py -- only the architecture entry is added to _TOOL_DEFINITIONS, in line with the parallel-worker file ownership agreement (2-a owns the compact-format changes). --- scripts/commands/architecture.py | 75 ++++++++++++++++++++++++++++++++ scripts/mcp_server.py | 23 ++++++++++ 2 files changed, 98 insertions(+) create mode 100644 scripts/commands/architecture.py diff --git a/scripts/commands/architecture.py b/scripts/commands/architecture.py new file mode 100644 index 00000000..9f559670 --- /dev/null +++ b/scripts/commands/architecture.py @@ -0,0 +1,75 @@ +"""Architecture command — single-call codebase overview for AI agents (issue #19). + +Registers the `architecture` CLI command and `codelens_architecture` MCP tool. +The heavy lifting lives in `architecture_engine.get_architecture`. + +Examples: + # Full overview (default) — ~2-4k tokens + python3 codelens.py architecture /path/to/workspace + + # Lite overview — under 1k tokens (omits routes/packages/hotspots) + python3 codelens.py architecture /path/to/workspace --lite +""" + +from commands import register_command +from architecture_engine import get_architecture + + +def add_args(parser): + """Register architecture command arguments with the argparse parser.""" + parser.add_argument( + "workspace", + nargs="?", + default=None, + help="Path to workspace root (auto-detected if omitted)", + ) + parser.add_argument( + "--lite", + action="store_true", + default=False, + help="Lite mode: only return languages, frameworks, entry_points, " + "and total_symbols (omits routes/packages/hotspots). " + "Targets <1k tokens output for cheap agent orientation.", + ) + parser.add_argument( + "--no-cache", + action="store_true", + default=False, + help="Ignore any existing .codelens/architecture_cache.json and " + "rebuild the overview from scratch. The cache is rewritten.", + ) + + +def execute(args, workspace): + """Run the architecture command and return its payload. + + Args: + args: Parsed argparse Namespace with .lite and .no_cache attributes. + workspace: Resolved workspace path. + + Returns: + Dict payload from architecture_engine.get_architecture. + """ + lite = bool(getattr(args, "lite", False)) + no_cache = bool(getattr(args, "no_cache", False)) + + if no_cache: + # Bust the cache by removing the file before building. + import os + cache_path = os.path.join(workspace, ".codelens", "architecture_cache.json") + try: + if os.path.isfile(cache_path): + os.remove(cache_path) + except OSError: + pass + + return get_architecture(workspace, lite=lite) + + +register_command( + "architecture", + "Single-call codebase overview for AI agents (languages, frameworks, " + "entry points, packages, routes, hotspots, total symbols)", + add_args, + execute, +) diff --git a/scripts/mcp_server.py b/scripts/mcp_server.py index edb597bc..d050beb0 100644 --- a/scripts/mcp_server.py +++ b/scripts/mcp_server.py @@ -716,6 +716,29 @@ "required": ["workspace"] } }, + "architecture": { + "description": "Single-call codebase overview for AI agent orientation (issue #19). Returns languages, frameworks, entry points, packages, top routes, graph hotspots, and total symbol count in one call. Use --lite for <1k token orientation mode that omits routes/packages/hotspots. Result is cached in .codelens/architecture_cache.json and invalidated on re-scan.", + "parameters": { + "type": "object", + "properties": { + "workspace": { + "type": "string", + "description": "Path to workspace root directory" + }, + "format": { + "type": "string", + "description": "Output format (default: json)", + "default": "json" + }, + "lite": { + "type": "boolean", + "description": "Lite mode: return only languages, frameworks, entry_points, total_symbols (omits routes/packages/hotspots). Targets <1k tokens output for cheap agent orientation.", + "default": False + } + }, + "required": ["workspace"] + } + }, "analyze": { "description": "Full repository analysis in a single command. The ultimate one-shot command for understanding an entire repository.", "parameters": { From 20c5502828c11adc4e342a01b1d1c1f96ca3236a Mon Sep 17 00:00:00 2001 From: worker-2-a Date: Sun, 28 Jun 2026 06:10:34 +0000 Subject: [PATCH 3/4] test(arch): add architecture command + caching tests 24 test cases across 7 test classes covering all eight spec verification points from issue #19: TestArchitectureBasic (2 tests): - test_returns_ok_status: architecture returns status=ok on scanned clean_app. - test_auto_scans_on_fresh_workspace: auto-scan populates db + graph on a workspace with no .codelens/ directory. TestArchitectureFields (9 tests): - test_payload_has_all_required_fields: every issue #19 field present in stats (languages, frameworks, entry_points, packages, routes, hotspots, total_symbols, adrs). - Per-field type/shape checks (dict of counts, list of paths, list of {method,path,handler} dicts, etc). - test_adrs_is_empty_list_placeholder: confirms ADR placeholder shape. - test_total_symbols_is_int: confirms graph_stats().nodes is used. TestArchitectureLiteMode (2 tests): - test_lite_omits_routes_packages_hotspots: lite mode shape. - test_lite_keeps_core_fields: languages/frameworks/entry_points/ total_symbols still present in lite mode. TestArchitectureHotspots (3 tests): - test_hotspots_sorted_by_dependents_desc: parses '(N dependents)' and verifies descending sort. - test_hotspots_match_graph_query: independent SQL query returns the same file set as the engine (proves the graph model is the source). - test_hotspots_are_distinct_files: no duplicate file paths (group-by-file SQL contract). TestArchitectureCaching (4 tests): - test_cache_file_created_on_first_call: .codelens/architecture_cache.json appears after first call. - test_cache_reused_on_second_call: second call returns cached=True and the cache file mtime is unchanged. - test_cache_invalidated_after_rescan: re-running cmd_scan makes the next architecture call rebuild (cached=False). - test_lite_and_full_caches_not_shared: requesting lite after full (or vice versa) rebuilds because the payload shapes differ. TestArchitectureMCPTool (2 tests): - test_mcp_tool_listed: codelens_architecture is in _TOOL_DEFINITIONS with workspace (required) and lite (optional) parameters. - test_mcp_tool_returns_full_architecture_via_stats: end-to-end MCP tools/call returns the full architecture data inside stats (verifies the stats-nesting workaround for _normalize_to_ai). TestArchitectureTokenBudget (2 tests): - test_lite_payload_under_4000_bytes: raw --lite engine payload stays under 4000 bytes (~1k tokens). - test_lite_mcp_response_under_4000_bytes: MCP-normalised --lite response stays under 4000 bytes. All 24 tests pass. Uses two fixtures: scanned_clean_app (full scan done before architecture call) and fresh_clean_app (no scan -- tests auto-scan path). --- tests/test_architecture.py | 533 +++++++++++++++++++++++++++++++++++++ 1 file changed, 533 insertions(+) create mode 100644 tests/test_architecture.py diff --git a/tests/test_architecture.py b/tests/test_architecture.py new file mode 100644 index 00000000..39ac9581 --- /dev/null +++ b/tests/test_architecture.py @@ -0,0 +1,533 @@ +""" +Tests for the `architecture` command + engine (issue #19). + +Verifies: +1. `architecture` command returns status ok on the clean_app fixture. +2. Full payload has all required fields (languages, frameworks, + entry_points, packages, routes, hotspots, total_symbols, adrs). +3. `--lite` mode omits routes / packages / hotspots. +4. Hotspots are sorted by dependent count descending and use the graph model. +5. Cache is created on first call and reused on second call (cached flag set). +6. Cache is invalidated after a re-scan (db mtime newer than cache mtime). +7. MCP tool `codelens_architecture` appears in tools/list and returns full + architecture data via the stats field (preserved by _normalize_to_ai). +8. `--lite` payload stays under ~1k tokens (~4000 bytes JSON). +""" + +import json +import os +import shutil +import sys +import tempfile +import time + +import pytest + +# Add scripts directory to path (matches other test files) +SCRIPT_DIR = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scripts" +) +sys.path.insert(0, SCRIPT_DIR) + +FIXTURE_DIR = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "benchmarks", "fixtures", "clean_app", +) + + +# ─── Fixtures ───────────────────────────────────────────────── + + +@pytest.fixture +def scanned_clean_app(): + """Copy clean_app fixture to a temp workspace and run a full scan. + + Yields the workspace path. The scan populates .codelens/codelens.db + (including graph_nodes + graph_edges), summary.json, and backend.json. + Cleanup removes the temp workspace on teardown. + """ + if not os.path.isdir(FIXTURE_DIR): + pytest.skip("clean_app fixture not available") + workspace = tempfile.mkdtemp(prefix="codelens_arch_test_") + for entry in os.listdir(FIXTURE_DIR): + src = os.path.join(FIXTURE_DIR, entry) + dst = os.path.join(workspace, entry) + if os.path.isdir(src): + shutil.copytree(src, dst) + else: + shutil.copy2(src, dst) + + from commands.scan import cmd_scan + cmd_scan(workspace, incremental=False) + + yield workspace + + shutil.rmtree(workspace, ignore_errors=True) + + +@pytest.fixture +def fresh_clean_app(): + """Copy clean_app fixture to a temp workspace WITHOUT scanning. + + Used to verify the architecture command auto-scans when no .codelens/ + directory exists yet. Yields the workspace path. + """ + if not os.path.isdir(FIXTURE_DIR): + pytest.skip("clean_app fixture not available") + workspace = tempfile.mkdtemp(prefix="codelens_arch_fresh_") + for entry in os.listdir(FIXTURE_DIR): + src = os.path.join(FIXTURE_DIR, entry) + dst = os.path.join(workspace, entry) + if os.path.isdir(src): + shutil.copytree(src, dst) + else: + shutil.copy2(src, dst) + + yield workspace + + shutil.rmtree(workspace, ignore_errors=True) + + +# ─── 1. Command returns status ok ──────────────────────────── + + +class TestArchitectureBasic: + """Verify the architecture command runs and returns a valid payload.""" + + def test_returns_ok_status(self, scanned_clean_app): + """architecture must return status=ok on a scanned clean_app.""" + from architecture_engine import get_architecture + + result = get_architecture(scanned_clean_app, lite=False) + assert isinstance(result, dict) + assert result.get("status") == "ok", ( + f"expected status ok, got: {result.get('status')!r}" + ) + assert result.get("workspace") == os.path.abspath(scanned_clean_app) + + def test_auto_scans_on_fresh_workspace(self, fresh_clean_app): + """architecture on a workspace with no .codelens/ must auto-scan.""" + from architecture_engine import get_architecture + + # Sanity: no .codelens/ before the call + assert not os.path.isdir(os.path.join(fresh_clean_app, ".codelens")) + + result = get_architecture(fresh_clean_app, lite=True) + assert result.get("status") == "ok" + # Auto-scan must have populated the db + graph tables + db_path = os.path.join(fresh_clean_app, ".codelens", "codelens.db") + assert os.path.isfile(db_path) + stats = result.get("stats", {}) + assert stats.get("total_symbols", 0) > 0, ( + "auto-scan should populate graph nodes so total_symbols > 0" + ) + + +# ─── 2. Required fields ────────────────────────────────────── + + +class TestArchitectureFields: + """Verify the full payload has all required fields from issue #19.""" + + REQUIRED_FIELDS = ( + "languages", "frameworks", "entry_points", + "packages", "routes", "hotspots", + "total_symbols", "adrs", + ) + + def test_payload_has_all_required_fields(self, scanned_clean_app): + """Full payload must contain every field listed in issue #19 inside stats.""" + from architecture_engine import get_architecture + + result = get_architecture(scanned_clean_app, lite=False) + stats = result["stats"] + for field in self.REQUIRED_FIELDS: + assert field in stats, ( + f"missing required field {field!r} in stats" + ) + + def test_languages_is_dict_of_counts(self, scanned_clean_app): + """languages must be a dict mapping language name to file count.""" + from architecture_engine import get_architecture + + result = get_architecture(scanned_clean_app, lite=False) + languages = result["stats"]["languages"] + assert isinstance(languages, dict) + # clean_app fixture has .py files, so python must be present + assert "python" in languages, ( + f"expected 'python' in languages, got: {list(languages.keys())}" + ) + assert languages["python"] > 0 + # All values must be positive ints + for lang, count in languages.items(): + assert isinstance(count, int) + assert count > 0 + + def test_frameworks_is_list(self, scanned_clean_app): + """frameworks must be a list (possibly empty).""" + from architecture_engine import get_architecture + + result = get_architecture(scanned_clean_app, lite=False) + frameworks = result["stats"]["frameworks"] + assert isinstance(frameworks, list) + + def test_entry_points_is_list_of_paths(self, scanned_clean_app): + """entry_points must be a list of file path strings.""" + from architecture_engine import get_architecture + + result = get_architecture(scanned_clean_app, lite=False) + entry_points = result["stats"]["entry_points"] + assert isinstance(entry_points, list) + # clean_app has main.py with `if __name__ == "__main__": main()` + assert len(entry_points) > 0, "expected at least one entry point" + for ep in entry_points: + assert isinstance(ep, str) + assert ep, "entry point path must not be empty" + + def test_packages_is_list(self, scanned_clean_app): + """packages must be a list of relative dir paths.""" + from architecture_engine import get_architecture + + result = get_architecture(scanned_clean_app, lite=False) + packages = result["stats"]["packages"] + assert isinstance(packages, list) + # clean_app has src/ + config/ as top-level dirs with source code. + # The src/ dir's direct children are files (db_queries.py, etc), not + # subdirs, so config/ should be detected as a package. + for pkg in packages: + assert isinstance(pkg, str) + + def test_routes_is_list_of_dicts(self, scanned_clean_app): + """routes must be a list of {method, path, handler} dicts.""" + from architecture_engine import get_architecture + + result = get_architecture(scanned_clean_app, lite=False) + routes = result["stats"]["routes"] + assert isinstance(routes, list) + for route in routes: + assert isinstance(route, dict) + assert "method" in route + assert "path" in route + assert "handler" in route + + def test_hotspots_is_list_of_strings(self, scanned_clean_app): + """hotspots must be a list of human-readable strings.""" + from architecture_engine import get_architecture + + result = get_architecture(scanned_clean_app, lite=False) + hotspots = result["stats"]["hotspots"] + assert isinstance(hotspots, list) + for hs in hotspots: + assert isinstance(hs, str) + # Each hotspot string should include a dependent count + assert "dependents" in hs, ( + f"hotspot string should mention 'dependents': {hs!r}" + ) + + def test_total_symbols_is_int(self, scanned_clean_app): + """total_symbols must be a non-negative int from graph_stats().nodes.""" + from architecture_engine import get_architecture + + result = get_architecture(scanned_clean_app, lite=False) + total = result["stats"]["total_symbols"] + assert isinstance(total, int) + # clean_app fixture has ~31 backend nodes (verified in test_graph_model) + assert total > 0, f"expected total_symbols > 0, got {total}" + + def test_adrs_is_empty_list_placeholder(self, scanned_clean_app): + """adrs must be an empty list (ADR feature is issue #16, Phase 3).""" + from architecture_engine import get_architecture + + result = get_architecture(scanned_clean_app, lite=False) + assert result["stats"]["adrs"] == [] + + +# ─── 3. Lite mode ──────────────────────────────────────────── + + +class TestArchitectureLiteMode: + """Verify --lite mode omits routes/packages/hotspots.""" + + def test_lite_omits_routes_packages_hotspots(self, scanned_clean_app): + """--lite payload must NOT contain routes, packages, or hotspots.""" + from architecture_engine import get_architecture + + result = get_architecture(scanned_clean_app, lite=True) + assert result.get("lite") is True + stats = result["stats"] + for omitted in ("routes", "packages", "hotspots"): + assert omitted not in stats, ( + f"--lite must omit {omitted!r}, but it's present in stats" + ) + + def test_lite_keeps_core_fields(self, scanned_clean_app): + """--lite payload must keep languages, frameworks, entry_points, total_symbols.""" + from architecture_engine import get_architecture + + result = get_architecture(scanned_clean_app, lite=True) + stats = result["stats"] + for kept in ("languages", "frameworks", "entry_points", "total_symbols"): + assert kept in stats, ( + f"--lite must keep {kept!r}, but it's missing from stats" + ) + + +# ─── 4. Hotspots use the graph model and are sorted ───────── + + +class TestArchitectureHotspots: + """Verify hotspots come from the graph model and are sorted desc.""" + + def test_hotspots_sorted_by_dependents_desc(self, scanned_clean_app): + """Hotspots must be sorted by dependent count, descending.""" + from architecture_engine import get_architecture + + result = get_architecture(scanned_clean_app, lite=False) + hotspots = result["stats"]["hotspots"] + if len(hotspots) < 2: + pytest.skip("need >=2 hotspots to verify sort order") + + # Extract the dependent count from each "file (N dependents)" string. + counts = [] + for hs in hotspots: + # Parse "src/foo.py (47 dependents)" -> 47 + try: + n_str = hs.split("(")[-1].split(" ")[0] + counts.append(int(n_str)) + except (IndexError, ValueError): + pytest.fail(f"could not parse dependent count from: {hs!r}") + + assert counts == sorted(counts, reverse=True), ( + f"hotspots not sorted desc: {counts}" + ) + + def test_hotspots_match_graph_query(self, scanned_clean_app): + """Hotspots must match a direct SQL query against graph_edges.""" + import sqlite3 + from architecture_engine import get_architecture + + result = get_architecture(scanned_clean_app, lite=False) + hotspots = result["stats"]["hotspots"] + + db_path = os.path.join(scanned_clean_app, ".codelens", "codelens.db") + conn = sqlite3.connect(db_path) + try: + rows = conn.execute( + "SELECT gn.file, COUNT(*) AS cnt " + "FROM graph_edges AS ge " + "INNER JOIN graph_nodes AS gn ON gn.node_id = ge.target_id " + "WHERE ge.target_id IS NOT NULL AND ge.target_id != '' " + " AND gn.file IS NOT NULL AND gn.file != '' " + "GROUP BY gn.file " + "ORDER BY cnt DESC " + "LIMIT 5", + ).fetchall() + finally: + conn.close() + + # The hotspot strings should reference the same files as the SQL query + sql_files = [r[0] for r in rows if r[0]] + hotspot_files = [hs.rsplit(" (", 1)[0] for hs in hotspots] + # The set of files in hotspots should match the set from the SQL + # query (order may differ if counts are tied, but the sets must match) + assert set(hotspot_files) == set(sql_files), ( + f"hotspot files {set(hotspot_files)} != sql files {set(sql_files)}" + ) + + def test_hotspots_are_distinct_files(self, scanned_clean_app): + """Each hotspot must be a distinct file (no duplicates).""" + from architecture_engine import get_architecture + + result = get_architecture(scanned_clean_app, lite=False) + hotspots = result["stats"]["hotspots"] + if not hotspots: + pytest.skip("no hotspots to verify") + files = [hs.rsplit(" (", 1)[0] for hs in hotspots] + assert len(files) == len(set(files)), ( + f"hotspot files must be distinct, got duplicates: {files}" + ) + + +# ─── 5. Caching ───────────────────────────────────────────── + + +class TestArchitectureCaching: + """Verify the architecture cache is created, reused, and invalidated.""" + + def test_cache_file_created_on_first_call(self, scanned_clean_app): + """First call must write .codelens/architecture_cache.json.""" + from architecture_engine import get_architecture + + cache_path = os.path.join( + scanned_clean_app, ".codelens", "architecture_cache.json" + ) + assert not os.path.isfile(cache_path), "cache should not exist yet" + + get_architecture(scanned_clean_app, lite=False) + assert os.path.isfile(cache_path), "cache file must be created" + + def test_cache_reused_on_second_call(self, scanned_clean_app): + """Second call must return the cached payload (cached=True).""" + from architecture_engine import get_architecture + + # First call builds and caches + first = get_architecture(scanned_clean_app, lite=False) + assert first.get("cached") is False + + # Touch the cache file to set a known mtime, then verify second call + # returns the cached version + cache_path = os.path.join( + scanned_clean_app, ".codelens", "architecture_cache.json" + ) + first_cache_mtime = os.path.getmtime(cache_path) + + second = get_architecture(scanned_clean_app, lite=False) + assert second.get("cached") is True, ( + "second call must return cached payload" + ) + # The cache file must not have been rewritten + second_cache_mtime = os.path.getmtime(cache_path) + assert second_cache_mtime == first_cache_mtime, ( + "cache file must not be rewritten when reusing" + ) + + def test_cache_invalidated_after_rescan(self, scanned_clean_app): + """Re-running scan must invalidate the cache (cached=False on next call).""" + from architecture_engine import get_architecture + from commands.scan import cmd_scan + + # First call builds and caches + first = get_architecture(scanned_clean_app, lite=False) + assert first.get("cached") is False + + # Wait briefly so the re-scan db mtime is strictly newer than the + # cache file mtime (mtime resolution can be 1s on some filesystems). + time.sleep(1.1) + + # Re-scan — this touches codelens.db (newer mtime than cache) + cmd_scan(scanned_clean_app, incremental=False) + + # Next architecture call must rebuild (cached=False) + rebuilt = get_architecture(scanned_clean_app, lite=False) + assert rebuilt.get("cached") is False, ( + "cache must be invalidated after re-scan" + ) + + def test_lite_and_full_caches_not_shared(self, scanned_clean_app): + """Requesting --lite after --full (or vice versa) must rebuild. + + Lite and full payloads have different shapes — the cache must not + serve the wrong shape even if the db mtime is unchanged. + """ + from architecture_engine import get_architecture + + full = get_architecture(scanned_clean_app, lite=False) + assert full.get("cached") is False + assert "routes" in full["stats"] + + lite = get_architecture(scanned_clean_app, lite=True) + # Even though the db hasn't changed, the cached payload is the wrong + # shape (full vs lite), so the engine must rebuild. + assert lite.get("cached") is False, ( + "lite must not be served from a full-mode cache" + ) + assert "routes" not in lite["stats"] + + +# ─── 6. MCP tool appears in tools/list ─────────────────────── + + +class TestArchitectureMCPTool: + """Verify codelens_architecture is registered as an MCP tool.""" + + def test_mcp_tool_listed(self): + """tools/list must include codelens_architecture.""" + # Import the MCP server module to access _TOOL_DEFINITIONS + import mcp_server + + assert "architecture" in mcp_server._TOOL_DEFINITIONS, ( + "architecture must be in _TOOL_DEFINITIONS" + ) + + tool_def = mcp_server._TOOL_DEFINITIONS["architecture"] + assert "description" in tool_def + assert "parameters" in tool_def + + params = tool_def["parameters"] + assert params.get("type") == "object" + assert "workspace" in params.get("properties", {}) + assert "workspace" in params.get("required", []) + # lite flag must be exposed + assert "lite" in params.get("properties", {}) + + def test_mcp_tool_returns_full_architecture_via_stats(self, scanned_clean_app): + """codelens_architecture tool call must return stats with architecture data. + + The MCP normalizer (_normalize_to_ai) preserves the `stats` dict + as-is, so the architecture data nested inside `stats` must survive + the normalization and reach the agent. + """ + from mcp_server import MCPServer + + srv = MCPServer() + call_result = srv._handle_tools_call({ + "name": "codelens_architecture", + "arguments": {"workspace": scanned_clean_app, "lite": False}, + }) + assert call_result.get("isError") is False + text = call_result["content"][0]["text"] + data = json.loads(text) + assert data.get("status") == "ok" + assert data.get("command") == "architecture" + + # The stats dict must contain the architecture-specific fields. + stats = data.get("stats", {}) + for field in ("languages", "frameworks", "entry_points", + "packages", "routes", "hotspots", + "total_symbols", "adrs"): + assert field in stats, ( + f"MCP tool must return {field!r} inside stats; got: " + f"{list(stats.keys())}" + ) + + # Sanity: total_symbols must be > 0 (graph is populated) + assert stats.get("total_symbols", 0) > 0 + + +# ─── 7. Lite payload stays under ~1k tokens ───────────────── + + +class TestArchitectureTokenBudget: + """Verify --lite output stays under the ~1k token target (~4000 bytes).""" + + def test_lite_payload_under_4000_bytes(self, scanned_clean_app): + """--lite JSON byte length must stay under 4000 (~1k tokens). + + 1 token ~= 4 bytes for English text + JSON syntax, so 4000 bytes is + a generous upper bound for the 1k-token target. + """ + from architecture_engine import get_architecture + + result = get_architecture(scanned_clean_app, lite=True) + payload_json = json.dumps(result, ensure_ascii=False, default=str) + byte_len = len(payload_json.encode("utf-8")) + assert byte_len < 4000, ( + f"--lite payload is {byte_len} bytes, exceeds 4000-byte budget " + f"(~1k tokens). Payload: {payload_json[:300]}..." + ) + + def test_lite_mcp_response_under_4000_bytes(self, scanned_clean_app): + """--lite MCP tool response must stay under 4000 bytes after normalization.""" + from mcp_server import MCPServer + + srv = MCPServer() + call_result = srv._handle_tools_call({ + "name": "codelens_architecture", + "arguments": {"workspace": scanned_clean_app, "lite": True}, + }) + text = call_result["content"][0]["text"] + byte_len = len(text.encode("utf-8")) + assert byte_len < 4000, ( + f"--lite MCP response is {byte_len} bytes, exceeds 4000-byte " + f"budget (~1k tokens). Response: {text[:300]}..." + ) From b7c163b3d3c9cd85c4503fcf8b90f32b1ea1ce71 Mon Sep 17 00:00:00 2001 From: worker-2-a Date: Sun, 28 Jun 2026 06:10:45 +0000 Subject: [PATCH 4/4] docs(arch): update README, SKILL-QUICK, CHANGELOG, agent-integration README.md: - Add 'architecture' to the 'Architecture & Understanding (P1)' command table with description and flag hints (--lite, --no-cache). SKILL-QUICK.md: - Add 'architecture [--lite] [--no-cache]' to the Navigation category (10 -> 11 commands). - Update the trigger map entry for 'project overview' from 'summary or handbook' to 'architecture --lite (single call, <1k tokens) or summary / handbook (deeper)'. CHANGELOG.md: - Add new '### get_architecture -- single-call codebase overview (issue #19)' section under [8.2.0] Unreleased. - Document the new architecture_engine.py module, CLI command, MCP tool, and test suite. - Document design decisions (stats-nesting, adrs placeholder, auto-scan, file-level hotspots). - Include token budget verification measurements from clean_app fixture. references/agent-integration.md: - Update the Codebase Understanding intent-to-tool map: 'How does this app work?' now starts with 'architecture --lite' before drilling into entrypoints/api-map/state-map/outline. - Add new entries: 'Give me a codebase overview' -> architecture --lite, 'What files have the most dependents?' -> architecture (hotspots field uses the graph model). - Add 'overview, orient, unfamiliar, codebase summary, what is this project' to the keyword detection matrix mapping to architecture --lite. Only additive changes to shared docs -- no rewrites of existing content, in line with the parallel-worker file ownership agreement. --- CHANGELOG.md | 83 +++++++++++++++++++++++++++++++++ README.md | 1 + SKILL-QUICK.md | 6 +-- references/agent-integration.md | 7 ++- 4 files changed, 92 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index cde2f693..b9ecf4ca 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -90,6 +90,89 @@ To migrate an engine to the graph backend: Future engine migrations (post-v8.2): `impact`, `circular`, `dependents`. +### `get_architecture` — single-call codebase overview (issue #19) + +Orientation on an unfamiliar codebase previously required 4-6 chained commands +(scan → list → detect → entrypoints → api-map → read entry files), burning +10-20k tokens before any real work started. The new `architecture` command ++ `codelens_architecture` MCP tool collapses that into a single call returning +a compact overview: languages, frameworks, entry points, packages, top routes, +graph hotspots, total symbol count, and an `adrs` placeholder (ADR feature is +issue #16, Phase 3). + +#### Added + +- **`scripts/architecture_engine.py`** — New engine module orchestrating + existing engines into one overview: + - `get_architecture(workspace, lite=False)` — single public entry point. + - `_compute_languages` — three-tier resolution: fresh scan_result's + `files_scanned` → `.codelens/summary.json` → cheap stat-only extension + walk. Scan-result buckets are collapsed to canonical language names + (e.g. `js_backend` + `js_frontend` → `javascript`). + - `_compute_frameworks` — thin wrapper around `framework_detect`. + - `_compute_entry_points` — wraps `entrypoints_engine.map_entrypoints`, + dedupes by file path, prioritises main/handler/cli types. + - `_compute_packages` — scans `src/`, `app/`, `lib/`, `packages/`, + `server/`, `internal/` for immediate subdirectories that contain source + files (one level of recursion allowed for flat two-tier packages). + - `_compute_routes` — wraps `apimap_engine.map_api_routes`, normalises + each route to `{method, path, handler}`, capped at 20. + - `_compute_hotspots` — single SQL round-trip: `SELECT gn.file, + COUNT(*) FROM graph_edges JOIN graph_nodes ... GROUP BY gn.file + ORDER BY cnt DESC LIMIT 5`. This is the killer feature of the new + graph model — files ranked by total incoming CALLS edges across all + their symbols (blast-radius surface). + - `_compute_total_symbols` — `graph_stats(db_path).nodes`. + - Cache layer: writes `.codelens/architecture_cache.json` on first call; + subsequent calls return the cache as long as `.codelens/codelens.db` + mtime hasn't advanced (i.e. scan hasn't been re-run). Lite and full + payloads have different shapes — the cache won't serve the wrong shape + even when the db is unchanged. + +- **`scripts/commands/architecture.py`** — New CLI command auto-registered + via `commands/__init__.py`. Flags: `--lite` (omit routes/packages/hotspots + for <1k token orientation), `--no-cache` (force rebuild). + +- **`codelens_architecture` MCP tool** — Added to `_TOOL_DEFINITIONS` in + `scripts/mcp_server.py`. Schema: `{workspace: string}` required; + `{format: string, lite: boolean}` optional. Calls the `architecture` + command internally. + +- **`tests/test_architecture.py`** — 24 test cases covering all eight spec + verification points: status ok, all required fields, lite mode shape, + hotspots sorting + graph query match + distinct files, cache create + + reuse + invalidation + lite/full shape isolation, MCP tool listing + + end-to-end call, <4000-byte token budget (raw + MCP-normalised). + +#### Design Decisions + +- **Architecture-specific fields nested inside `stats`** — The MCP + `_normalize_to_ai` formatter preserves `stats` as-is but drops unknown + top-level keys. To deliver the full architecture data through MCP without + modifying the shared formatter (which would collide with parallel worker + 2-a's compact-format changes), all architecture fields (languages, + frameworks, entry_points, packages, routes, hotspots, total_symbols, adrs) + are nested inside `stats`. CLI consumers see the same shape. +- **`adrs` placeholder is `[]`** — ADR detection is issue #16 (Phase 3). + The field is reserved so consumers can rely on the shape today. +- **Auto-scan on fresh workspace** — If `.codelens/codelens.db` doesn't + exist or the graph tables are empty when `architecture` is called, the + engine runs `cmd_scan` first. This makes the tool self-sufficient for + the issue #19 use-case ("agent starts on an unfamiliar codebase"). +- **File-level hotspots (not per-symbol)** — The SQL query groups by + `gn.file`, not `ge.target_id`, so each hotspot is a distinct file with + its total blast radius. Matches the issue spec example + `"src/models/user.py (47 dependents)"`. + +#### Token Budget Verification (clean_app fixture) + +- `--lite` raw engine payload: **284 bytes** (~71 tokens) +- `--lite` MCP-normalised response: **298 bytes** (~75 tokens) +- Full raw engine payload: **502 bytes** (~125 tokens) +- Full MCP-normalised response: **515 bytes** (~128 tokens) + +All well under the 1k-token target. + --- ## [8.1.0] — 2026-06-13 diff --git a/README.md b/README.md index 3f748509..230818a4 100644 --- a/README.md +++ b/README.md @@ -131,6 +131,7 @@ python3 scripts/codelens.py query "myFunction" --lite | Command | Description | |---------|-------------| +| `architecture [workspace] [--lite] [--no-cache]` | Single-call codebase overview for AI agents (languages, frameworks, entry points, packages, routes, hotspots, total symbols). `--lite` omits routes/packages/hotspots for <1k token orientation (issue #19) | | `entrypoints [workspace]` | Map execution entry points | | `api-map [workspace]` | Map REST/GraphQL/gRPC routes to handlers | | `state-map [workspace]` | Track global state management | diff --git a/SKILL-QUICK.md b/SKILL-QUICK.md index 1874c7fc..321f5889 100755 --- a/SKILL-QUICK.md +++ b/SKILL-QUICK.md @@ -74,7 +74,7 @@ $CLI complexity --top 5 --lite # → Top 5 most complex, minimal output | "cleanup before deploy" | `debug-leak` → `dead-code` → `secrets` | | "CSS issues?" | `css-deep` → `missing-refs` | | "accessible?" | `a11y` | -| "project overview" | `summary` or `handbook` | +| "project overview" | `architecture --lite` (single call, <1k tokens) or `summary` / `handbook` (deeper) | | "fix automatically" | `fix --apply` (dry-run by default) | | "show dashboard" | `dashboard` | | "trend over time" | `history` | @@ -112,8 +112,8 @@ $CLI complexity --top 5 --lite # → Top 5 most complex, minimal output ### Pre-Write Safety (5) `query "name" [--domain ...] [--fuzzy]` · `impact "name" [--action modify|delete]` · `refactor-safe "name" [--action rename|move]` · `guard (--pre|--post) --file PATH` · `check [--severity ...] [--max-findings N]` -### Navigation (10) -`summary [--focus security|quality|architecture|all] [--detail minimal|standard|full]` · `context "name"` · `trace "name" [--direction up|down|both]` · `search "pattern"` · `symbols "name" [--fuzzy]` · `outline [--file path]` · `dependents "file"` · `list [--filter ...]` · `ask "question"` · `diff` +### Navigation (11) +`architecture [--lite] [--no-cache]` · `summary [--focus security|quality|architecture|all] [--detail minimal|standard|full]` · `context "name"` · `trace "name" [--direction up|down|both]` · `search "pattern"` · `symbols "name" [--fuzzy]` · `outline [--file path]` · `dependents "file"` · `list [--filter ...]` · `ask "question"` · `diff` ### Architecture (8) `entrypoints` · `api-map` · `state-map` · `detect` · `handbook` · `diff` · `dashboard` · `history` diff --git a/references/agent-integration.md b/references/agent-integration.md index 17ffcbb7..f5175084 100755 --- a/references/agent-integration.md +++ b/references/agent-integration.md @@ -52,12 +52,14 @@ This is the fastest way for an AI to pick the right tool: | User Intent | Tool Chain | |-------------|------------| -| "How does this app work?" | `entrypoints` → `api-map` → `state-map` → `outline --all` | -| "What endpoints exist?" | `api-map` | +| "How does this app work?" | `architecture --lite` (single-call overview, <1k tokens) → `entrypoints` → `api-map` → `state-map` → `outline --all` for deeper drill-down | +| "Give me a codebase overview" | `architecture --lite` (issue #19 — single call replaces 4-6 chained commands) | +| "What endpoints exist?" | `api-map` (or `architecture` for top 20 routes in one call) | | "Where is global state?" | `state-map` | | "Who calls this function?" | `trace --direction up` | | "What does this function call?" | `trace --direction down` | | "Tell me about this symbol" | `context` | +| "What files have the most dependents?" | `architecture` (the `hotspots` field uses the graph model) | #### Quality & Production @@ -94,6 +96,7 @@ corresponding CodeLens tools: | Keywords (EN) | Keywords (ID) | Tool | |---------------|---------------|------| +| overview, orient, unfamiliar, codebase summary, what is this project | gambaran umum, mengenal, proyek ini apa | `architecture --lite` (issue #19) | | exists, already, check if, does this | sudah ada, apakah | `query` | | who uses, who calls, references | siapa pakai, siapa panggil | `trace` / `query` | | delete, remove, hapus | hapus, buang | `impact` + `dead-code` |