diff --git a/README.md b/README.md index 162b1d96..c39ca14f 100644 --- a/README.md +++ b/README.md @@ -2,12 +2,12 @@ > **Before an AI writes a new class/id/function, CodeLens must be checked. This is not optional.** -CodeLens is an AI-native code intelligence platform that gives AI agents **full visibility** into a codebase before they write any code. It prevents collision, overwrite of existing logic, security vulnerabilities, and dead code through 71 CLI commands, an MCP server with 69 tools (54 static + 15 dynamic), AST-based taint analysis, live CVE/OSV scanning, a plugin system with OWASP Top 10 + Compliance rule packs, a true graph data model (nodes + edges) for structural code queries, and token-efficient `--format compact` output for high-volume agent workflows (issue #17). +CodeLens is an AI-native code intelligence platform that gives AI agents **full visibility** into a codebase before they write any code. It prevents collision, overwrite of existing logic, security vulnerabilities, and dead code through 73 CLI commands, an MCP server with 71 tools (55 static + 16 dynamic), AST-based taint analysis, live CVE/OSV scanning, a plugin system with OWASP Top 10 + Compliance rule packs, a true graph data model (nodes + edges) for structural code queries, and token-efficient `--format compact` output for high-volume agent workflows (issue #17). ## Features -- **71 CLI Commands** — From basic scan/query to AST taint analysis, CVE scanning, plugin management, auto-fix, dashboards, CI/CD quality gates, and `graph-schema` for cheap graph-shape introspection -- **MCP Server (69 Tools)** — Native AI agent integration via Model Context Protocol (JSON-RPC over stdio), 54 statically-defined tools + 15 dynamically discovered, every tool accepts a `format` parameter (`json`/`markdown`/`ai`/`sarif`/`compact`) +- **73 CLI Commands** — From basic scan/query to AST taint analysis, CVE scanning, plugin management, auto-fix, dashboards, CI/CD quality gates, and `graph-schema` for cheap graph-shape introspection +- **MCP Server (71 Tools)** — Native AI agent integration via Model Context Protocol (JSON-RPC over stdio), 55 statically-defined tools + 16 dynamically discovered, every tool accepts a `format` parameter (`json`/`markdown`/`ai`/`sarif`/`compact`) - **Token-Efficient Compact Output (v8.2, issue #17)** — `--format compact` produces single-char-key JSON with abbreviated types, omitted null fields, and relative paths — ~50% smaller than `json` on real trace output. Combined with `--limit`/`--offset` pagination, 5 structural queries now cost <5k tokens (down from 30-80k) - **AST Taint Engine** — Tree-sitter based taint analysis with return-value propagation, scope hierarchy, and branch condition refinement - **Live CVE/OSV Scanning** — Real-time vulnerability data from OSV.dev API with SQLite cache, 9 ecosystems (PyPI, npm, crates.io, Go, Maven, NuGet, RubyGems, Pub, Hex) @@ -225,8 +225,8 @@ codelens/ │ ├── changelog.md # Older changelog (per-version highlights) │ └── agent-integration.md # AI agent integration guide ├── scripts/ -│ ├── codelens.py # CLI entry point (71 commands registered) -│ ├── mcp_server.py # MCP JSON-RPC server (69 tools) +│ ├── codelens.py # CLI entry point (73 commands registered) +│ ├── mcp_server.py # MCP JSON-RPC server (71 tools) │ ├── registry.py # Registry read/write/build │ ├── persistent_registry.py # SQLite persistent storage (WAL mode) │ ├── base_parser.py # Base tree-sitter parser diff --git a/SKILL-QUICK.md b/SKILL-QUICK.md index e73a1f98..fcb6a574 100755 --- a/SKILL-QUICK.md +++ b/SKILL-QUICK.md @@ -115,7 +115,7 @@ $CLI list --limit 5 --offset 10 --format compact # → paginated + co | "Cross-file taint" | `dataflow` | `taint` (taint is single-file, AST-deep) | | "Auto-fix issues" | `fix` | `check` (check just gates, doesn't fix) | -## All 71 Commands +## All 73 Commands ### Setup & Lifecycle (8+) `init` · `scan [--incremental] [--max-files N] [--full]` · `registry-validate` · `detect` · `watch [--debounce SECS] [--git-mode] [--interval SECS]` · `git-status` · `migrate` · `serve` · `lsp-status` (issue #33: `codelens --lsp-status` top-level flag is an alias of `codelens lsp-status` — both delegate to `hybrid_engine.get_lsp_status()` and return the identical payload) @@ -147,9 +147,9 @@ $CLI list --limit 5 --offset 10 --format compact # → paginated + co ### Tooling (1) `plugin ` -**Total: 71 commands** (auto-registered via `commands/__init__.py`; rerun `python3 scripts/sync_command_count.py --apply` after adding/removing a command) +**Total: 73 commands** (auto-registered via `commands/__init__.py`; rerun `python3 scripts/sync_command_count.py --apply` after adding/removing a command) -## MCP Server (69 Tools) +## MCP Server (71 Tools) Start the MCP server for AI agent integration: @@ -157,9 +157,9 @@ Start the MCP server for AI agent integration: python3 scripts/codelens.py serve ``` -Exposes 69 tools as `codelens_` (e.g., `codelens_query`, `codelens_taint`, `codelens_graph_schema`, `codelens_architecture`, `codelens_resolve_types`, `codelens_git_status`): +Exposes 71 tools as `codelens_` (e.g., `codelens_query`, `codelens_taint`, `codelens_graph_schema`, `codelens_architecture`, `codelens_resolve_types`, `codelens_git_status`): - 50 statically-defined tools (full JSON schemas in `mcp_server.py`) -- 15 dynamically-discovered tools (auto-discovered from `COMMAND_REGISTRY`; long-running `watch` and `serve` are excluded) +- 16 dynamically-discovered tools (auto-discovered from `COMMAND_REGISTRY`; long-running `watch` and `serve` are excluded) - Every tool accepts a `format` parameter (`json`/`markdown`/`ai`/`sarif`/`compact`). Use `format: "compact"` for token-efficient responses (~50% smaller than `json`). - `watch` and `serve` itself are excluded (long-running) diff --git a/SKILL.md b/SKILL.md index a22843c1..11b3359c 100755 --- a/SKILL.md +++ b/SKILL.md @@ -1,10 +1,10 @@ --- name: codelens description: > - CodeLens — AI-Native Code Intelligence. 71 commands for AI-powered code analysis, + CodeLens — AI-Native Code Intelligence. 73 commands for AI-powered code analysis, security auditing, quality scoring, AST-based taint analysis, live CVE scanning, and pre-write safety checks. Supports 28+ languages with tree-sitter + regex - fallback parsing. MCP server exposes 69 tools for AI agent integration. + fallback parsing. MCP server exposes 71 tools for AI agent integration. For quick command reference with validated output schemas, see SKILL-QUICK.md. For version history, see CHANGELOG.md. --- diff --git a/pyproject.toml b/pyproject.toml index 975ec607..1ae515c0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta" [project] name = "codelens" version = "8.2.0" -description = "Live Codebase Reference Intelligence — 71 commands for AI-powered code analysis, security auditing, and quality scoring" +description = "Live Codebase Reference Intelligence — 73 commands for AI-powered code analysis, security auditing, and quality scoring" readme = "README.md" license = {text = "MIT"} requires-python = ">=3.8" diff --git a/scripts/adr_engine.py b/scripts/adr_engine.py new file mode 100644 index 00000000..44b975d4 --- /dev/null +++ b/scripts/adr_engine.py @@ -0,0 +1,588 @@ +# @WHO: scripts/adr_engine.py +# @WHAT: Architecture Decision Records (ADR) SQLite-backed manager (issue #16) +# @PART: engine +# @ENTRY: manage_adr() +"""Architecture Decision Records (ADR) manager for CodeLens (issue #16). + +Provides persistent memory of *why* the codebase is structured the way it is, +so agents don't propose changes that violate intentional constraints. + +Storage layout +-------------- +- **SQLite DB:** ``/.codelens/adrs.db`` — single-table store with + ``id``, ``title``, ``context``, ``decision``, ``status``, ``superseded_by``, + ``created_at``, ``updated_at`` columns. + +Schema +------ +:: + + CREATE TABLE adrs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + title TEXT NOT NULL, + context TEXT NOT NULL DEFAULT '', + decision TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL DEFAULT 'proposed', + superseded_by INTEGER, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + FOREIGN KEY (superseded_by) REFERENCES adrs(id) + ); + +Statuses +-------- +- ``proposed`` — drafted but not yet accepted +- ``accepted`` — active decision the codebase should follow +- ``deprecated`` — superseded or no longer relevant; ``superseded_by`` should + point to the replacement ADR id (if any) +- ``rejected`` — proposed and explicitly rejected + +Actions +------- +- ``create`` — insert a new ADR, return the new record +- ``list`` — return all ADRs (optionally filtered by status) +- ``get`` — return a single ADR by id +- ``update`` — patch title/context/decision/status fields +- ``deprecate`` — set status=deprecated and link superseded_by +- ``delete`` — hard delete (rare; prefer deprecate) + +All actions return a structured dict suitable for JSON / MCP transport. +""" + +from __future__ import annotations + +import os +import sqlite3 +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + + +# ─── Constants ───────────────────────────────────────────────────────────── + +SCHEMA_VERSION = 1 +DB_FILENAME = "adrs.db" + +_VALID_STATUSES = {"proposed", "accepted", "deprecated", "rejected"} + +_CREATE_TABLE = """ +CREATE TABLE IF NOT EXISTS adrs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + title TEXT NOT NULL, + context TEXT NOT NULL DEFAULT '', + decision TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL DEFAULT 'proposed', + superseded_by INTEGER, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + FOREIGN KEY (superseded_by) REFERENCES adrs(id) +) +""" + +_CREATE_INDEX_STATUS = ( + "CREATE INDEX IF NOT EXISTS idx_adrs_status ON adrs(status)" +) + + +# ─── Path helpers ────────────────────────────────────────────────────────── + + +def adr_db_path(workspace: str) -> str: + """Return the absolute path to the ADR SQLite DB for a workspace.""" + return os.path.join(workspace, ".codelens", DB_FILENAME) + + +# ─── DB connection ───────────────────────────────────────────────────────── + + +def _connect(workspace: str) -> sqlite3.Connection: + """Open (and lazily initialize) the ADR DB for ``workspace``. + + Creates the ``.codelens/`` directory and the ``adrs`` table if missing. + Returns a connection with ``row_factory = sqlite3.Row`` so callers can + index results by column name. + """ + workspace = os.path.abspath(workspace) + db_path = adr_db_path(workspace) + os.makedirs(os.path.dirname(db_path), exist_ok=True) + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + conn.execute(_CREATE_TABLE) + conn.execute(_CREATE_INDEX_STATUS) + conn.commit() + return conn + + +def _now_iso() -> str: + """Return the current UTC timestamp in ISO 8601 format.""" + return datetime.now(timezone.utc).isoformat() + + +# ─── Validation ──────────────────────────────────────────────────────────── + + +def _validate_status(status: str) -> None: + """Raise ``ValueError`` if ``status`` is not a known ADR status.""" + if status not in _VALID_STATUSES: + raise ValueError( + f"Invalid ADR status: {status!r}. " + f"Must be one of: {sorted(_VALID_STATUSES)}" + ) + + +def _validate_title(title: str) -> None: + """Raise ``ValueError`` if ``title`` is empty or whitespace-only.""" + if not title or not title.strip(): + raise ValueError("ADR title cannot be empty") + + +def _validate_id(adr_id: Any) -> int: + """Coerce ``adr_id`` to int and validate it is positive.""" + try: + iid = int(adr_id) + except (TypeError, ValueError) as exc: + raise ValueError(f"ADR id must be an integer, got: {adr_id!r}") from exc + if iid < 1: + raise ValueError(f"ADR id must be >= 1, got: {iid}") + return iid + + +def _validate_superseded_by(superseded_by: Any) -> Optional[int]: + """Validate ``superseded_by``: ``None`` or a positive int.""" + if superseded_by is None or superseded_by == "": + return None + try: + iid = int(superseded_by) + except (TypeError, ValueError) as exc: + raise ValueError( + f"superseded_by must be an integer or null, got: {superseded_by!r}" + ) from exc + if iid < 1: + raise ValueError(f"superseded_by must be >= 1, got: {iid}") + return iid + + +# ─── Row → dict ──────────────────────────────────────────────────────────── + + +def _row_to_dict(row: sqlite3.Row) -> Dict[str, Any]: + """Convert a sqlite3.Row to a plain dict (JSON-serializable).""" + return { + "id": row["id"], + "title": row["title"], + "context": row["context"], + "decision": row["decision"], + "status": row["status"], + "superseded_by": row["superseded_by"], + "created_at": row["created_at"], + "updated_at": row["updated_at"], + } + + +# ─── Actions ─────────────────────────────────────────────────────────────── + + +def create_adr( + workspace: str, + title: str, + context: str = "", + decision: str = "", + status: str = "proposed", +) -> Dict[str, Any]: + """Create a new ADR record. + + Returns the freshly-inserted record (with its assigned ``id``). + """ + _validate_title(title) + _validate_status(status) + workspace = os.path.abspath(workspace) + + now = _now_iso() + conn = _connect(workspace) + try: + cur = conn.execute( + """INSERT INTO adrs (title, context, decision, status, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?)""", + (title.strip(), context or "", decision or "", status, now, now), + ) + new_id = cur.lastrowid + conn.commit() + row = conn.execute( + "SELECT * FROM adrs WHERE id = ?", (new_id,) + ).fetchone() + finally: + conn.close() + + record = _row_to_dict(row) + return { + "status": "ok", + "action": "created", + "adr": record, + } + + +def list_adrs( + workspace: str, + status_filter: Optional[str] = None, +) -> Dict[str, Any]: + """List all ADRs, optionally filtered by status. + + Returns a dict with ``total``, ``filtered`` count, and ``adrs`` list + sorted by id ascending. + """ + workspace = os.path.abspath(workspace) + if status_filter is not None: + _validate_status(status_filter) + + conn = _connect(workspace) + try: + if status_filter is not None: + rows = conn.execute( + "SELECT * FROM adrs WHERE status = ? ORDER BY id ASC", + (status_filter,), + ).fetchall() + else: + rows = conn.execute( + "SELECT * FROM adrs ORDER BY id ASC" + ).fetchall() + total = conn.execute("SELECT COUNT(*) AS c FROM adrs").fetchone()["c"] + finally: + conn.close() + + adrs = [_row_to_dict(r) for r in rows] + return { + "status": "ok", + "action": "list", + "total": total, + "filtered": len(adrs), + "filter": status_filter, + "adrs": adrs, + } + + +def get_adr(workspace: str, adr_id: int) -> Dict[str, Any]: + """Return a single ADR by id. + + Returns a ``not_found`` result (not an error) when the ADR doesn't exist, + so callers can distinguish "missing" from "broken". + """ + iid = _validate_id(adr_id) + workspace = os.path.abspath(workspace) + + conn = _connect(workspace) + try: + row = conn.execute( + "SELECT * FROM adrs WHERE id = ?", (iid,) + ).fetchone() + finally: + conn.close() + + if row is None: + return { + "status": "not_found", + "id": iid, + "message": f"ADR #{iid} not found in workspace {workspace!r}.", + } + + return { + "status": "ok", + "action": "get", + "adr": _row_to_dict(row), + } + + +def update_adr( + workspace: str, + adr_id: int, + title: Optional[str] = None, + context: Optional[str] = None, + decision: Optional[str] = None, + status: Optional[str] = None, +) -> Dict[str, Any]: + """Patch one or more fields of an ADR. + + Only fields explicitly passed (non-``None``) are updated. ``updated_at`` + is always refreshed. Returns the updated record. + """ + iid = _validate_id(adr_id) + workspace = os.path.abspath(workspace) + + if title is not None: + _validate_title(title) + if status is not None: + _validate_status(status) + + # Build SET clause dynamically. + updates: List[str] = [] + params: List[Any] = [] + if title is not None: + updates.append("title = ?") + params.append(title.strip()) + if context is not None: + updates.append("context = ?") + params.append(context) + if decision is not None: + updates.append("decision = ?") + params.append(decision) + if status is not None: + updates.append("status = ?") + params.append(status) + + if not updates: + # Nothing to update — return current record unchanged. + return get_adr(workspace, iid) + + updates.append("updated_at = ?") + params.append(_now_iso()) + params.append(iid) + + conn = _connect(workspace) + try: + cur = conn.execute( + f"UPDATE adrs SET {', '.join(updates)} WHERE id = ?", params + ) + if cur.rowcount == 0: + conn.close() + return { + "status": "not_found", + "id": iid, + "message": f"ADR #{iid} not found — no update performed.", + } + conn.commit() + row = conn.execute( + "SELECT * FROM adrs WHERE id = ?", (iid,) + ).fetchone() + finally: + conn.close() + + return { + "status": "ok", + "action": "updated", + "adr": _row_to_dict(row), + } + + +def deprecate_adr( + workspace: str, + adr_id: int, + superseded_by: Optional[int] = None, +) -> Dict[str, Any]: + """Mark an ADR as deprecated, optionally linking to its replacement. + + Sets ``status = 'deprecated'`` and (if ``superseded_by`` is provided) + ``superseded_by = ``. The replacement ADR must exist + and must not be the same as the deprecated one (self-reference forbidden). + """ + iid = _validate_id(adr_id) + sup_id = _validate_superseded_by(superseded_by) + workspace = os.path.abspath(workspace) + + if sup_id is not None and sup_id == iid: + raise ValueError( + f"ADR #{iid} cannot supersede itself — superseded_by must point " + "to a different ADR." + ) + + conn = _connect(workspace) + try: + # Verify the ADR being deprecated exists. + row = conn.execute( + "SELECT * FROM adrs WHERE id = ?", (iid,) + ).fetchone() + if row is None: + conn.close() + return { + "status": "not_found", + "id": iid, + "message": f"ADR #{iid} not found — cannot deprecate.", + } + + # If a replacement is named, verify it exists and is not deprecated. + if sup_id is not None: + sup_row = conn.execute( + "SELECT * FROM adrs WHERE id = ?", (sup_id,) + ).fetchone() + if sup_row is None: + conn.close() + return { + "status": "error", + "error": "superseded_by_not_found", + "message": ( + f"Replacement ADR #{sup_id} not found — cannot link " + f"ADR #{iid} to a non-existent record." + ), + "id": iid, + "superseded_by": sup_id, + } + + now = _now_iso() + if sup_id is not None: + conn.execute( + """UPDATE adrs + SET status = 'deprecated', + superseded_by = ?, + updated_at = ? + WHERE id = ?""", + (sup_id, now, iid), + ) + else: + conn.execute( + """UPDATE adrs + SET status = 'deprecated', + updated_at = ? + WHERE id = ?""", + (now, iid), + ) + conn.commit() + row = conn.execute( + "SELECT * FROM adrs WHERE id = ?", (iid,) + ).fetchone() + finally: + conn.close() + + return { + "status": "ok", + "action": "deprecated", + "adr": _row_to_dict(row), + } + + +def delete_adr(workspace: str, adr_id: int) -> Dict[str, Any]: + """Hard-delete an ADR record. + + Returns ``not_found`` if the id doesn't exist. Also clears any + ``superseded_by`` references pointing at the deleted id (sets them to + NULL) so referential integrity is preserved without blocking the delete. + """ + iid = _validate_id(adr_id) + workspace = os.path.abspath(workspace) + + conn = _connect(workspace) + try: + # Check existence first. + row = conn.execute( + "SELECT * FROM adrs WHERE id = ?", (iid,) + ).fetchone() + if row is None: + conn.close() + return { + "status": "not_found", + "id": iid, + "message": f"ADR #{iid} not found — nothing to delete.", + } + + # Clear dangling superseded_by references (defensive — we don't want + # to block deletes, but we also don't want to leave pointers to a + # deleted row). + conn.execute( + "UPDATE adrs SET superseded_by = NULL WHERE superseded_by = ?", + (iid,), + ) + conn.execute("DELETE FROM adrs WHERE id = ?", (iid,)) + conn.commit() + finally: + conn.close() + + return { + "status": "ok", + "action": "deleted", + "id": iid, + } + + +# ─── Top-level dispatcher ────────────────────────────────────────────────── + + +def manage_adr( + workspace: str, + action: str, + *, + id: Optional[int] = None, + title: Optional[str] = None, + context: Optional[str] = None, + decision: Optional[str] = None, + status: Optional[str] = None, + superseded_by: Optional[int] = None, + status_filter: Optional[str] = None, +) -> Dict[str, Any]: + """Dispatch an ADR action by name. + + This is the single entry point used by both the CLI command and the MCP + tool. Keeping the dispatch here (rather than in the command layer) means + the programmatic API matches the user-facing API 1:1. + + Actions: ``create``, ``list``, ``get``, ``update``, ``deprecate``, ``delete``. + """ + if action == "create": + if title is None: + return { + "status": "error", + "error": "missing_required_field", + "field": "title", + "message": "create action requires a 'title' argument.", + } + return create_adr( + workspace, + title=title, + context=context or "", + decision=decision or "", + status=status or "proposed", + ) + + if action == "list": + return list_adrs(workspace, status_filter=status_filter) + + if action == "get": + if id is None: + return { + "status": "error", + "error": "missing_required_field", + "field": "id", + "message": "get action requires an 'id' argument.", + } + return get_adr(workspace, id) + + if action == "update": + if id is None: + return { + "status": "error", + "error": "missing_required_field", + "field": "id", + "message": "update action requires an 'id' argument.", + } + return update_adr( + workspace, + id, + title=title, + context=context, + decision=decision, + status=status, + ) + + if action == "deprecate": + if id is None: + return { + "status": "error", + "error": "missing_required_field", + "field": "id", + "message": "deprecate action requires an 'id' argument.", + } + return deprecate_adr(workspace, id, superseded_by=superseded_by) + + if action == "delete": + if id is None: + return { + "status": "error", + "error": "missing_required_field", + "field": "id", + "message": "delete action requires an 'id' argument.", + } + return delete_adr(workspace, id) + + return { + "status": "error", + "error": "unknown_action", + "action": action, + "available_actions": [ + "create", "list", "get", "update", "deprecate", "delete" + ], + } diff --git a/scripts/commands/adr.py b/scripts/commands/adr.py new file mode 100644 index 00000000..82131579 --- /dev/null +++ b/scripts/commands/adr.py @@ -0,0 +1,194 @@ +# @WHO: scripts/commands/adr.py +# @WHAT: Architecture Decision Records CLI command (issue #16) +# @PART: commands +# @ENTRY: execute() +"""ADR command — Architecture Decision Records manager (issue #16). + +Provides persistent memory of *why* the codebase is structured the way it is, +so AI agents don't propose refactors that violate intentional constraints. +Backed by SQLite at ``.codelens/adrs.db``. + +Usage:: + + codelens adr create --title "Use SQLite over PostgreSQL" \\ + --context "Deployment simplicity for single-node setups" \\ + --decision "SQLite with WAL mode" --status accepted + + codelens adr list # list all ADRs + codelens adr list --status accepted # filter by status + codelens adr get --id 3 # fetch a single ADR + codelens adr update --id 3 --status deprecated + codelens adr deprecate --id 3 --superseded-by 7 + codelens adr delete --id 3 + +The storage layer and programmatic API live in :mod:`adr_engine` — this module +is a thin argparse wrapper that calls :func:`adr_engine.manage_adr`. +""" + +from __future__ import annotations + +from typing import Any, Dict + +from commands import register_command + + +def add_args(parser): + """Add ADR subcommand arguments to the parser.""" + sub = parser.add_subparsers(dest="adr_action", help="ADR action") + + # adr create + create = sub.add_parser( + "create", + help="Create a new Architecture Decision Record", + description=( + "Create a new ADR at .codelens/adrs.db. Required: --title. " + "Optional: --context, --decision, --status (default: proposed). " + "Returns the freshly-inserted record with its assigned id." + ), + ) + create.add_argument( + "--title", required=True, + help="Short title for the decision (e.g. 'Use SQLite over PostgreSQL')", + ) + create.add_argument( + "--context", default="", + help="Why is this decision needed? Background and constraints.", + ) + create.add_argument( + "--decision", default="", + help="The decision itself (what was chosen and why).", + ) + create.add_argument( + "--status", default="proposed", + choices=["proposed", "accepted", "deprecated", "rejected"], + help="Initial status (default: proposed)", + ) + + # adr list + list_p = sub.add_parser( + "list", + help="List all ADRs (optionally filtered by status)", + description=( + "List all ADRs in the workspace, sorted by id ascending. " + "Pass --status to filter (proposed/accepted/deprecated/rejected)." + ), + ) + list_p.add_argument( + "--status", default=None, + choices=["proposed", "accepted", "deprecated", "rejected"], + help="Filter by status (default: all statuses)", + ) + + # adr get + get_p = sub.add_parser( + "get", + help="Get a single ADR by id", + description="Fetch a single ADR record by its numeric id.", + ) + get_p.add_argument( + "--id", required=True, type=int, + help="ADR id (positive integer)", + ) + + # adr update + update = sub.add_parser( + "update", + help="Update one or more fields of an ADR", + description=( + "Patch an existing ADR. Only fields explicitly passed are " + "updated; updated_at is always refreshed. At least one of " + "--title, --context, --decision, --status must be provided." + ), + ) + update.add_argument("--id", required=True, type=int, help="ADR id") + update.add_argument("--title", default=None, help="New title") + update.add_argument("--context", default=None, help="New context") + update.add_argument("--decision", default=None, help="New decision") + update.add_argument( + "--status", default=None, + choices=["proposed", "accepted", "deprecated", "rejected"], + help="New status", + ) + + # adr deprecate + deprecate = sub.add_parser( + "deprecate", + help="Mark an ADR as deprecated (optionally link to a replacement)", + description=( + "Set status=deprecated. If --superseded-by is provided, the " + "replacement ADR must exist and must not be the same id. " + "Prefer this over `delete` — it preserves history." + ), + ) + deprecate.add_argument("--id", required=True, type=int, help="ADR id to deprecate") + deprecate.add_argument( + "--superseded-by", default=None, type=int, + help="Id of the ADR that supersedes this one (optional)", + ) + + # adr delete + delete = sub.add_parser( + "delete", + help="Hard-delete an ADR (prefer 'deprecate' to preserve history)", + description=( + "Permanently remove an ADR record. Also clears any " + "superseded_by references pointing at the deleted id. " + "Prefer 'deprecate' for normal workflow — use 'delete' only " + "for records created in error." + ), + ) + delete.add_argument("--id", required=True, type=int, help="ADR id to delete") + + +def execute(args, workspace): + """Dispatch the ADR subcommand.""" + action = getattr(args, "adr_action", None) + if not action: + return { + "status": "error", + "error": "no_action", + "message": "No ADR action specified.", + "usage": "codelens adr [args]", + "examples": [ + "codelens adr create --title 'Use SQLite' --decision 'WAL mode'", + "codelens adr list --status accepted", + "codelens adr get --id 3", + "codelens adr update --id 3 --status deprecated", + "codelens adr deprecate --id 3 --superseded-by 7", + "codelens adr delete --id 3", + ], + } + + # Lazy import so a broken adr_engine never breaks command discovery. + from adr_engine import manage_adr + + # Read arguments defensively — MCP calls pass through _ArgsNamespace which + # may not have every attribute set. + kwargs: Dict[str, Any] = {} + + if action == "create": + kwargs["title"] = getattr(args, "title", None) + kwargs["context"] = getattr(args, "context", None) + kwargs["decision"] = getattr(args, "decision", None) + kwargs["status"] = getattr(args, "status", None) or "proposed" + elif action == "list": + kwargs["status_filter"] = getattr(args, "status", None) + elif action in {"get", "update", "deprecate", "delete"}: + kwargs["id"] = getattr(args, "id", None) + if action == "update": + kwargs["title"] = getattr(args, "title", None) + kwargs["context"] = getattr(args, "context", None) + kwargs["decision"] = getattr(args, "decision", None) + kwargs["status"] = getattr(args, "status", None) + elif action == "deprecate": + kwargs["superseded_by"] = getattr(args, "superseded_by", None) + + return manage_adr(workspace, action, **kwargs) + + +register_command( + "adr", + "Architecture Decision Records manager (create/list/get/update/deprecate/delete)", + add_args, + execute, +) diff --git a/scripts/graph_model.py b/scripts/graph_model.py index 8b2e8cd5..0fb0b6cf 100644 --- a/scripts/graph_model.py +++ b/scripts/graph_model.py @@ -13,7 +13,7 @@ - New tables `graph_nodes` and `graph_edges` are additive (prefixed `graph_` to avoid colliding with any existing table name). - The flat registry tables and JSON files are untouched. -- All 71 existing CLI commands continue to work unchanged. +- All 73 existing CLI commands continue to work unchanged. Schema: graph_nodes( diff --git a/scripts/mcp_server.py b/scripts/mcp_server.py index ce3f2b27..9eebc822 100644 --- a/scripts/mcp_server.py +++ b/scripts/mcp_server.py @@ -1099,6 +1099,65 @@ "required": ["workspace"] } }, + "manage-adr": { + "description": ( + "Architecture Decision Records (ADR) manager — persistent memory " + "of *why* the codebase is structured the way it is, so agents " + "don't propose refactors that violate intentional constraints. " + "Backed by SQLite at .codelens/adrs.db. Actions: " + "create [context] [decision] [status], " + "list [status_filter], get <id>, " + "update <id> [title] [context] [decision] [status], " + "deprecate <id> [superseded_by], delete <id>. " + "Statuses: proposed (default), accepted, deprecated, rejected. " + "Prefer 'deprecate' over 'delete' to preserve history." + ), + "parameters": { + "type": "object", + "properties": { + "workspace": { + "type": "string", + "description": "Path to workspace root directory" + }, + "action": { + "type": "string", + "enum": ["create", "list", "get", "update", "deprecate", "delete"], + "description": "ADR action to perform" + }, + "id": { + "type": "integer", + "description": "ADR id (required for get/update/deprecate/delete). Positive integer." + }, + "title": { + "type": "string", + "description": "Short title (required for create, optional for update). E.g. 'Use SQLite over PostgreSQL'." + }, + "context": { + "type": "string", + "description": "Background and constraints driving the decision (optional for create/update)." + }, + "decision": { + "type": "string", + "description": "The decision itself — what was chosen and why (optional for create/update)." + }, + "status": { + "type": "string", + "enum": ["proposed", "accepted", "deprecated", "rejected"], + "description": "ADR status. Default for create is 'proposed'." + }, + "superseded_by": { + "type": "integer", + "description": "Id of the replacement ADR (optional, for deprecate action). Must exist and differ from id." + }, + "status_filter": { + "type": "string", + "enum": ["proposed", "accepted", "deprecated", "rejected"], + "description": "Filter list action by status (optional)." + } + }, + "required": ["workspace", "action"] + } + }, } diff --git a/skill.json b/skill.json index d236d4ae..507d902f 100755 --- a/skill.json +++ b/skill.json @@ -1,7 +1,7 @@ { "name": "codelens", "version": "8.2.0", - "description": "Live Codebase Reference Intelligence. 71 commands for AI-powered code analysis, security auditing, quality scoring, and pre-write safety checks. Supports 28+ languages with regex+AST hybrid parsing. Must activate before writing/editing/deleting any class, id, or function.", + "description": "Live Codebase Reference Intelligence. 73 commands for AI-powered code analysis, security auditing, quality scoring, and pre-write safety checks. Supports 28+ languages with regex+AST hybrid parsing. Must activate before writing/editing/deleting any class, id, or function.", "author": "codelens", "command_categories": { "setup": [ diff --git a/tests/test_adr.py b/tests/test_adr.py new file mode 100644 index 00000000..9c66ad1c --- /dev/null +++ b/tests/test_adr.py @@ -0,0 +1,551 @@ +"""Tests for the Architecture Decision Records (ADR) system (issue #16). + +Covers: +- ``adr_engine`` SQLite-backed CRUD: create / list / get / update / deprecate / delete +- Status validation (proposed/accepted/deprecated/rejected) +- Supersession linking (deprecate with --superseded-by) +- Referential integrity (self-supersession forbidden, missing replacement rejected) +- Soft-fail on missing ids (``not_found`` results, not exceptions) +- The CLI ``adr`` command auto-registers and dispatches subcommands +- The MCP ``manage-adr`` tool is statically defined in ``_TOOL_DEFINITIONS`` +- File header (@WHO/@WHAT/@PART/@ENTRY) is present +""" + +from __future__ import annotations + +import os +import sys +import tempfile + +import pytest + +# Make scripts/ importable. +SCRIPT_DIR = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scripts" +) +if SCRIPT_DIR not in sys.path: + sys.path.insert(0, SCRIPT_DIR) + +from adr_engine import ( # noqa: E402 + manage_adr, + create_adr, + list_adrs, + get_adr, + update_adr, + deprecate_adr, + delete_adr, + adr_db_path, + _VALID_STATUSES, +) +from commands import COMMAND_REGISTRY # noqa: E402 + + +# ─── Fixtures ────────────────────────────────────────────────────────────── + + +@pytest.fixture +def workspace(): + """Yield a temporary workspace directory.""" + d = tempfile.mkdtemp(prefix="codelens_adr_test_") + yield d + import shutil + shutil.rmtree(d, ignore_errors=True) + + +# ─── Path / constants ────────────────────────────────────────────────────── + + +class TestPathAndConstants: + """Sanity-checks for path helpers and module constants.""" + + def test_adr_db_path_under_codelens_dir(self, workspace): + p = adr_db_path(workspace) + assert p.endswith(os.path.join(".codelens", "adrs.db")) + assert os.path.isabs(p) + + def test_valid_statuses_set(self): + assert _VALID_STATUSES == {"proposed", "accepted", "deprecated", "rejected"} + + +# ─── create_adr ──────────────────────────────────────────────────────────── + + +class TestCreateAdr: + """``create_adr`` inserts a record and returns it with an assigned id.""" + + def test_create_returns_record_with_id(self, workspace): + r = create_adr(workspace, title="Use SQLite over PostgreSQL") + assert r["status"] == "ok" + assert r["action"] == "created" + adr = r["adr"] + assert adr["id"] >= 1 + assert adr["title"] == "Use SQLite over PostgreSQL" + assert adr["context"] == "" + assert adr["decision"] == "" + assert adr["status"] == "proposed" # default + assert adr["superseded_by"] is None + assert adr["created_at"] == adr["updated_at"] + + def test_create_with_all_fields(self, workspace): + r = create_adr( + workspace, + title="Use WAL mode", + context="Single-node deployment", + decision="Enable WAL for concurrent reads", + status="accepted", + ) + adr = r["adr"] + assert adr["context"] == "Single-node deployment" + assert adr["decision"] == "Enable WAL for concurrent reads" + assert adr["status"] == "accepted" + + def test_create_strips_title_whitespace(self, workspace): + r = create_adr(workspace, title=" Padded title ") + assert r["adr"]["title"] == "Padded title" + + def test_create_rejects_empty_title(self, workspace): + with pytest.raises(ValueError, match="title cannot be empty"): + create_adr(workspace, title="") + with pytest.raises(ValueError, match="title cannot be empty"): + create_adr(workspace, title=" ") + + def test_create_rejects_invalid_status(self, workspace): + with pytest.raises(ValueError, match="Invalid ADR status"): + create_adr(workspace, title="x", status="bogus") + + def test_create_assigns_sequential_ids(self, workspace): + r1 = create_adr(workspace, title="First") + r2 = create_adr(workspace, title="Second") + r3 = create_adr(workspace, title="Third") + assert r2["adr"]["id"] == r1["adr"]["id"] + 1 + assert r3["adr"]["id"] == r2["adr"]["id"] + 1 + + def test_create_persists_to_sqlite_file(self, workspace): + create_adr(workspace, title="Persisted") + assert os.path.isfile(adr_db_path(workspace)) + + +# ─── list_adrs ───────────────────────────────────────────────────────────── + + +class TestListAdrs: + """``list_adrs`` returns all ADRs, optionally filtered by status.""" + + def test_list_empty_workspace(self, workspace): + r = list_adrs(workspace) + assert r["status"] == "ok" + assert r["total"] == 0 + assert r["filtered"] == 0 + assert r["adrs"] == [] + assert r["filter"] is None + + def test_list_returns_all_sorted_by_id(self, workspace): + create_adr(workspace, title="C") + create_adr(workspace, title="A") + create_adr(workspace, title="B") + r = list_adrs(workspace) + assert r["total"] == 3 + assert r["filtered"] == 3 + titles = [a["title"] for a in r["adrs"]] + # Sorted by id (insertion order), not alphabetical + assert titles == ["C", "A", "B"] + + def test_list_filter_by_status(self, workspace): + create_adr(workspace, title="P1", status="proposed") + create_adr(workspace, title="A1", status="accepted") + create_adr(workspace, title="P2", status="proposed") + create_adr(workspace, title="D1", status="deprecated") + + r = list_adrs(workspace, status_filter="accepted") + assert r["filter"] == "accepted" + assert r["total"] == 4 # total still counts all + assert r["filtered"] == 1 + assert r["adrs"][0]["title"] == "A1" + + def test_list_filter_rejects_invalid_status(self, workspace): + with pytest.raises(ValueError, match="Invalid ADR status"): + list_adrs(workspace, status_filter="bogus") + + +# ─── get_adr ─────────────────────────────────────────────────────────────── + + +class TestGetAdr: + """``get_adr`` returns a single record or ``not_found``.""" + + def test_get_existing(self, workspace): + created = create_adr(workspace, title="Find me") + rid = created["adr"]["id"] + r = get_adr(workspace, rid) + assert r["status"] == "ok" + assert r["action"] == "get" + assert r["adr"]["id"] == rid + assert r["adr"]["title"] == "Find me" + + def test_get_missing_returns_not_found(self, workspace): + r = get_adr(workspace, 9999) + assert r["status"] == "not_found" + assert r["id"] == 9999 + assert "not found" in r["message"].lower() + + def test_get_rejects_non_integer_id(self, workspace): + with pytest.raises(ValueError, match="must be an integer"): + get_adr(workspace, "not-an-int") + + def test_get_rejects_non_positive_id(self, workspace): + with pytest.raises(ValueError, match="must be >= 1"): + get_adr(workspace, 0) + with pytest.raises(ValueError, match="must be >= 1"): + get_adr(workspace, -5) + + +# ─── update_adr ──────────────────────────────────────────────────────────── + + +class TestUpdateAdr: + """``update_adr`` patches fields and refreshes ``updated_at``.""" + + def test_update_single_field(self, workspace): + created = create_adr(workspace, title="Original") + rid = created["adr"]["id"] + original_updated = created["adr"]["updated_at"] + + r = update_adr(workspace, rid, decision="New decision") + assert r["status"] == "ok" + assert r["action"] == "updated" + assert r["adr"]["decision"] == "New decision" + assert r["adr"]["title"] == "Original" # unchanged + # updated_at must have moved forward (or stayed equal if same ms) + assert r["adr"]["updated_at"] >= original_updated + + def test_update_multiple_fields(self, workspace): + created = create_adr(workspace, title="Orig", status="proposed") + rid = created["adr"]["id"] + r = update_adr( + workspace, rid, + title="New title", + context="New context", + decision="New decision", + status="accepted", + ) + adr = r["adr"] + assert adr["title"] == "New title" + assert adr["context"] == "New context" + assert adr["decision"] == "New decision" + assert adr["status"] == "accepted" + + def test_update_with_no_fields_returns_current_record(self, workspace): + created = create_adr(workspace, title="Untouched") + rid = created["adr"]["id"] + r = update_adr(workspace, rid) + # Falls through to get_adr + assert r["status"] == "ok" + assert r["adr"]["title"] == "Untouched" + + def test_update_missing_returns_not_found(self, workspace): + r = update_adr(workspace, 9999, title="x") + assert r["status"] == "not_found" + + def test_update_rejects_empty_title(self, workspace): + created = create_adr(workspace, title="Valid") + rid = created["adr"]["id"] + with pytest.raises(ValueError, match="title cannot be empty"): + update_adr(workspace, rid, title="") + + def test_update_rejects_invalid_status(self, workspace): + created = create_adr(workspace, title="Valid") + rid = created["adr"]["id"] + with pytest.raises(ValueError, match="Invalid ADR status"): + update_adr(workspace, rid, status="bogus") + + +# ─── deprecate_adr ───────────────────────────────────────────────────────── + + +class TestDeprecateAdr: + """``deprecate_adr`` sets status=deprecated and links superseded_by.""" + + def test_deprecate_without_replacement(self, workspace): + created = create_adr(workspace, title="Old", status="accepted") + rid = created["adr"]["id"] + r = deprecate_adr(workspace, rid) + assert r["status"] == "ok" + assert r["action"] == "deprecated" + assert r["adr"]["status"] == "deprecated" + assert r["adr"]["superseded_by"] is None + + def test_deprecate_with_replacement_links_ids(self, workspace): + old = create_adr(workspace, title="Old", status="accepted") + new = create_adr(workspace, title="New", status="accepted") + r = deprecate_adr(workspace, old["adr"]["id"], superseded_by=new["adr"]["id"]) + assert r["adr"]["status"] == "deprecated" + assert r["adr"]["superseded_by"] == new["adr"]["id"] + + def test_deprecate_rejects_self_supersession(self, workspace): + created = create_adr(workspace, title="Self") + rid = created["adr"]["id"] + with pytest.raises(ValueError, match="cannot supersede itself"): + deprecate_adr(workspace, rid, superseded_by=rid) + + def test_deprecate_rejects_missing_replacement(self, workspace): + created = create_adr(workspace, title="Old") + rid = created["adr"]["id"] + r = deprecate_adr(workspace, rid, superseded_by=9999) + assert r["status"] == "error" + assert r["error"] == "superseded_by_not_found" + assert r["superseded_by"] == 9999 + + def test_deprecate_missing_adr_returns_not_found(self, workspace): + r = deprecate_adr(workspace, 9999) + assert r["status"] == "not_found" + + +# ─── delete_adr ──────────────────────────────────────────────────────────── + + +class TestDeleteAdr: + """``delete_adr`` removes a record and clears dangling references.""" + + def test_delete_existing(self, workspace): + created = create_adr(workspace, title="ToDelete") + rid = created["adr"]["id"] + r = delete_adr(workspace, rid) + assert r["status"] == "ok" + assert r["action"] == "deleted" + assert r["id"] == rid + # Verify it's really gone + assert get_adr(workspace, rid)["status"] == "not_found" + + def test_delete_missing_returns_not_found(self, workspace): + r = delete_adr(workspace, 9999) + assert r["status"] == "not_found" + + def test_delete_clears_dangling_superseded_by(self, workspace): + old = create_adr(workspace, title="Old", status="accepted") + new = create_adr(workspace, title="New", status="accepted") + deprecate_adr(workspace, old["adr"]["id"], superseded_by=new["adr"]["id"]) + + # Delete the replacement — old.superseded_by should become NULL + delete_adr(workspace, new["adr"]["id"]) + r = get_adr(workspace, old["adr"]["id"]) + assert r["adr"]["superseded_by"] is None + + +# ─── manage_adr dispatcher ───────────────────────────────────────────────── + + +class TestManageAdrDispatcher: + """``manage_adr`` is the single entry point used by CLI + MCP.""" + + def test_dispatch_create(self, workspace): + r = manage_adr(workspace, "create", title="Via dispatch") + assert r["status"] == "ok" + assert r["action"] == "created" + + def test_dispatch_list(self, workspace): + manage_adr(workspace, "create", title="A") + r = manage_adr(workspace, "list") + assert r["filtered"] == 1 + + def test_dispatch_get(self, workspace): + created = manage_adr(workspace, "create", title="X") + rid = created["adr"]["id"] + r = manage_adr(workspace, "get", id=rid) + assert r["status"] == "ok" + + def test_dispatch_update(self, workspace): + created = manage_adr(workspace, "create", title="X") + rid = created["adr"]["id"] + r = manage_adr(workspace, "update", id=rid, title="Y") + assert r["adr"]["title"] == "Y" + + def test_dispatch_deprecate(self, workspace): + created = manage_adr(workspace, "create", title="X") + rid = created["adr"]["id"] + r = manage_adr(workspace, "deprecate", id=rid) + assert r["adr"]["status"] == "deprecated" + + def test_dispatch_delete(self, workspace): + created = manage_adr(workspace, "create", title="X") + rid = created["adr"]["id"] + r = manage_adr(workspace, "delete", id=rid) + assert r["status"] == "ok" + + def test_dispatch_unknown_action_returns_error(self, workspace): + r = manage_adr(workspace, "bogus") + assert r["status"] == "error" + assert r["error"] == "unknown_action" + assert "create" in r["available_actions"] + + def test_dispatch_create_without_title_returns_structured_error(self, workspace): + r = manage_adr(workspace, "create") + assert r["status"] == "error" + assert r["error"] == "missing_required_field" + assert r["field"] == "title" + + def test_dispatch_get_without_id_returns_structured_error(self, workspace): + r = manage_adr(workspace, "get") + assert r["status"] == "error" + assert r["error"] == "missing_required_field" + assert r["field"] == "id" + + +# ─── CLI command registration ────────────────────────────────────────────── + + +class TestCliCommandRegistration: + """The ``adr`` command must auto-register from commands/adr.py.""" + + def test_adr_command_registered(self): + assert "adr" in COMMAND_REGISTRY + info = COMMAND_REGISTRY["adr"] + assert "help" in info + assert "add_args" in info + assert "execute" in info + assert callable(info["add_args"]) + assert callable(info["execute"]) + + def test_adr_command_help_text_mentions_actions(self): + help_text = COMMAND_REGISTRY["adr"]["help"] + for action in ("create", "list", "get", "update", "deprecate", "delete"): + assert action in help_text + + def test_execute_with_no_action_returns_usage_error(self, workspace): + from argparse import Namespace + info = COMMAND_REGISTRY["adr"] + # Simulate `codelens adr` with no subcommand + args = Namespace(adr_action=None) + r = info["execute"](args, workspace) + assert r["status"] == "error" + assert r["error"] == "no_action" + assert "usage" in r + assert "examples" in r + + +# ─── MCP tool registration ───────────────────────────────────────────────── + + +class TestMcpToolRegistration: + """The ``manage-adr`` MCP tool must be statically defined.""" + + def test_manage_adr_in_tool_definitions(self): + try: + from mcp_server import _TOOL_DEFINITIONS + except ImportError as exc: + pytest.skip(f"mcp_server not importable: {exc}") + assert "manage-adr" in _TOOL_DEFINITIONS + + def test_manage_adr_schema_has_required_fields(self): + try: + from mcp_server import _TOOL_DEFINITIONS + except ImportError as exc: + pytest.skip(f"mcp_server not importable: {exc}") + schema = _TOOL_DEFINITIONS["manage-adr"] + assert "description" in schema + assert "parameters" in schema + params = schema["parameters"] + assert params["type"] == "object" + assert "workspace" in params["properties"] + assert "action" in params["properties"] + assert "workspace" in params["required"] + assert "action" in params["required"] + + def test_manage_adr_action_enum_matches_engine(self): + try: + from mcp_server import _TOOL_DEFINITIONS + except ImportError as exc: + pytest.skip(f"mcp_server not importable: {exc}") + action_prop = _TOOL_DEFINITIONS["manage-adr"]["parameters"]["properties"]["action"] + assert set(action_prop["enum"]) == { + "create", "list", "get", "update", "deprecate", "delete" + } + + def test_manage_adr_status_enum_matches_engine(self): + try: + from mcp_server import _TOOL_DEFINITIONS + except ImportError as exc: + pytest.skip(f"mcp_server not importable: {exc}") + props = _TOOL_DEFINITIONS["manage-adr"]["parameters"]["properties"] + assert set(props["status"]["enum"]) == _VALID_STATUSES + assert set(props["status_filter"]["enum"]) == _VALID_STATUSES + + +# ─── File header convention ──────────────────────────────────────────────── + + +class TestFileHeaders: + """CONTRIBUTING.md mandates @WHO/@WHAT/@PART/@ENTRY headers on new files.""" + + def test_adr_engine_has_file_header(self): + path = os.path.join(SCRIPT_DIR, "adr_engine.py") + with open(path, "r", encoding="utf-8") as f: + head = f.read(500) + assert "# @WHO:" in head + assert "# @WHAT:" in head + assert "# @PART:" in head + assert "# @ENTRY:" in head + + def test_adr_command_has_file_header(self): + path = os.path.join(SCRIPT_DIR, "commands", "adr.py") + with open(path, "r", encoding="utf-8") as f: + head = f.read(500) + assert "# @WHO:" in head + assert "# @WHAT:" in head + assert "# @PART:" in head + assert "# @ENTRY:" in head + + +# ─── Cross-action workflow integration ───────────────────────────────────── + + +class TestEndToEndWorkflow: + """Simulate a real ADR lifecycle: create → accept → supersede → deprecate.""" + + def test_full_lifecycle(self, workspace): + # 1. Propose an ADR + r1 = manage_adr( + workspace, "create", + title="Use SQLite over PostgreSQL", + context="Deployment simplicity for single-node", + decision="TBD", + status="proposed", + ) + id1 = r1["adr"]["id"] + assert r1["adr"]["status"] == "proposed" + + # 2. Accept it after deliberation + r2 = manage_adr( + workspace, "update", + id=id1, + decision="SQLite with WAL mode, weekly VACUUM", + status="accepted", + ) + assert r2["adr"]["status"] == "accepted" + assert r2["adr"]["decision"] == "SQLite with WAL mode, weekly VACUUM" + + # 3. Time passes — a new ADR supersedes it + r3 = manage_adr( + workspace, "create", + title="Use PostgreSQL for multi-node", + context="Outgrew single-node SQLite", + decision="PostgreSQL 16 with streaming replication", + status="accepted", + ) + id3 = r3["adr"]["id"] + + # 4. Deprecate the old one, link to the new + r4 = manage_adr(workspace, "deprecate", id=id1, superseded_by=id3) + assert r4["adr"]["status"] == "deprecated" + assert r4["adr"]["superseded_by"] == id3 + + # 5. List — total 2, accepted 1, deprecated 1 + all_adrs = manage_adr(workspace, "list") + assert all_adrs["total"] == 2 + + accepted = manage_adr(workspace, "list", status_filter="accepted") + assert accepted["filtered"] == 1 + assert accepted["adrs"][0]["id"] == id3 + + deprecated = manage_adr(workspace, "list", status_filter="deprecated") + assert deprecated["filtered"] == 1 + assert deprecated["adrs"][0]["id"] == id1