# Code Path & Data Pipeline Audit Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** Build `src/code_path_audit.py` — a static-analysis tool that audits the 3 major actions (AI message lifecycle, discussion save/load, GUI startup) for expensive operations, redundant calls, and pipelining candidates. Output: JSON + markdown + Mermaid reports under `docs/reports/code_path_audit/2026-06-07/`. **Architecture:** Single new module `src/code_path_audit.py`. No new dependencies. Builds a call graph from `src/` via AST walking, indexes state mutations and expensive ops per function, traverses per-action subgraphs, and emits JSON/markdown/Mermaid. Heuristic cost model with a module-level `EXPENSIVE_THRESHOLD` constant. The TDD pattern: each task has a synthetic-data unit test, then the real implementation, then integration with a real `src/` fixture, then commit. **Tech Stack:** Python 3.11+, `ast` (stdlib), `pathlib` (stdlib), `json` (stdlib), `dataclasses` (stdlib). No new pip dependencies. --- ## Phase 0: Setup **Files:** `conductor/tracks/code_path_audit_20260607/state.toml` (create), `src/code_path_audit.py` (create empty), `tests/test_code_path_audit.py` (create empty). - [ ] **Step 0.1: Create `state.toml`** Write `conductor/tracks/code_path_audit_20260607/state.toml`: ```toml # Track state for code_path_audit_20260607 # Updated by Tier 2 Tech Lead as tasks complete [meta] track_id = "code_path_audit_20260607" name = "Code Path & Data Pipeline Audit" status = "active" current_phase = 0 last_updated = "2026-06-07" [phases] phase_1 = { status = "pending", checkpointsha = "", name = "Data structures (CallGraph, ExpensiveOpIndex, StateMutationIndex)" } phase_2 = { status = "pending", checkpointsha = "", name = "trace_action + ActionProfile + cost model" } phase_3 = { status = "pending", checkpointsha = "", name = "Output (JSON / markdown / Mermaid)" } phase_4 = { status = "pending", checkpointsha = "", name = "MCP tool + CLI surface" } phase_5 = { status = "pending", checkpointsha = "", name = "Run audit on 3 actions; commit report" } phase_6 = { status = "pending", checkpointsha = "", name = "tracks.md update" } [verification] call_graph_json_produced = false expensive_ops_json_produced = false state_mutations_json_produced = false actions_ai_message_produced = false actions_save_load_produced = false actions_gui_startup_produced = false summary_md_produced = false optimization_candidates_md_produced = false unit_tests_passing = false ``` - [ ] **Step 0.2: Create empty `src/code_path_audit.py`** ```bash New-Item -ItemType File -Path src/code_path_audit.py -Force | Out-Null ``` - [ ] **Step 0.3: Create empty `tests/test_code_path_audit.py`** ```bash New-Item -ItemType File -Path tests/test_code_path_audit.py -Force | Out-Null ``` - [ ] **Step 0.4: Confirm git state clean for the new files** Run: `git status --short` Expected: only the user's pre-existing modifications (not the new files; the new files are untracked, which is fine for now). - [ ] **Step 0.5: Conductor - User Manual Verification (per workflow.md)** Ask the user to confirm Phase 0 setup is right before proceeding. --- ## Phase 1: Data structures (CallGraph, ExpensiveOpIndex, StateMutationIndex) **Files:** `src/code_path_audit.py`, `tests/test_code_path_audit.py`. This phase is one commit. Three sub-tasks (one per data structure), each with TDD. ### Task 1.1: CallGraph **Files:** - Modify: `src/code_path_audit.py` - Test: `tests/test_code_path_audit.py` - [ ] **Step 1.1.1: Add the `CallGraph` + `FunctionNode` dataclasses to `src/code_path_audit.py`** Write the following at the top of `src/code_path_audit.py` (1-space indent per the project's Python style): ```python """Static code path & data pipeline audit tool. Builds a call graph of src/, indexes state mutations and expensive operations per function, and produces per-action load profiles for the 3 major actions (AI message lifecycle, discussion save/load, GUI startup). See conductor/tracks/code_path_audit_20260607/spec.md. """ from __future__ import annotations import ast import json from dataclasses import dataclass, field from pathlib import Path from typing import Literal EXPENSIVE_THRESHOLD: int = 40_000 COST_CLASS_WEIGHTS: dict[str, int] = { "file_io": 100, "network": 500, "ast_parse": 200, "json_io": 150, "pickle": 300, "deep_copy": 200, "loop_amplified": 1, } @dataclass class FunctionNode: fqname: str file: str line: int calls: list[str] = field(default_factory=list) state_mutations: list["StateMutation"] = field(default_factory=list) expensive_ops: list["ExpensiveOp"] = field(default_factory=list) @dataclass class CallGraph: nodes: dict[str, FunctionNode] = field(default_factory=dict) edges: dict[str, set[str]] = field(default_factory=dict) def add_edge(self, caller: str, callee: str) -> None: self.edges.setdefault(caller, set()).add(callee) self.nodes.setdefault(caller, FunctionNode(fqname=caller, file="", line=0)) self.nodes.setdefault(callee, FunctionNode(fqname=callee, file="", line=0)) def transitive_callees(self, root: str, max_depth: int = 10) -> set[str]: seen: set[str] = set() stack: list[tuple[str, int]] = [(root, 0)] while stack: node, depth = stack.pop() if depth > max_depth or node in seen: continue seen.add(node) for callee in self.edges.get(node, ()): stack.append((callee, depth + 1)) seen.discard(root) return seen def render_mermaid(self, root: str, max_depth: int = 5) -> str: seen: set[str] = {root} edges: list[tuple[str, str]] = [] stack: list[tuple[str, int]] = [(root, 0)] while stack: node, depth = stack.pop() if depth >= max_depth: continue for callee in self.edges.get(node, ()): edges.append((node, callee)) if callee not in seen: seen.add(callee) stack.append((callee, depth + 1)) lines = ["graph LR"] for caller, callee in sorted(edges): lines.append(f' {caller.replace(".", "_")}["{caller}"] --> {callee.replace(".", "_")}["{callee}"]') return "\n".join(lines) ``` - [ ] **Step 1.1.2: Write the failing test for `CallGraph` in `tests/test_code_path_audit.py`** ```python """Tests for src.code_path_audit.""" from src.code_path_audit import CallGraph, EXPENSIVE_THRESHOLD def test_callgraph_add_edge_creates_nodes() -> None: cg = CallGraph() cg.add_edge("a.b.c", "a.b.d") assert "a.b.c" in cg.nodes assert "a.b.d" in cg.nodes assert "a.b.d" in cg.edges["a.b.c"] def test_callgraph_transitive_callees_5_node_synthetic() -> None: # a -> b -> c # \-> d -> e cg = CallGraph() cg.add_edge("a", "b") cg.add_edge("b", "c") cg.add_edge("a", "d") cg.add_edge("d", "e") result = cg.transitive_callees("a", max_depth=10) assert result == {"b", "c", "d", "e"} def test_callgraph_transitive_callees_respects_max_depth() -> None: cg = CallGraph() cg.add_edge("a", "b") cg.add_edge("b", "c") cg.add_edge("c", "d") # max_depth=1 from a: only b (direct callee); not c (depth 2) result = cg.transitive_callees("a", max_depth=1) assert result == {"b"} def test_callgraph_render_mermaid_basic() -> None: cg = CallGraph() cg.add_edge("a", "b") cg.add_edge("b", "c") md = cg.render_mermaid("a", max_depth=5) assert "graph LR" in md assert 'a["a"] --> b["b"]' in md assert 'b["b"] --> c["c"]' in md def test_expensive_threshold_default() -> None: assert EXPENSIVE_THRESHOLD == 40_000 ``` - [ ] **Step 1.1.3: Run the test to verify it fails (before the implementation in 1.1.1 is complete)** (Note: 1.1.1 added the dataclasses; 1.1.2 added the tests. They should now both exist; the test should PASS. If 1.1.1 was skipped, the test would fail with `ImportError: cannot import name 'CallGraph'`.) Run: `uv run pytest tests/test_code_path_audit.py -q 2>&1 | Select-Object -Last 10` Expected: 5 tests pass. ### Task 1.2: ExpensiveOpIndex + cost classes **Files:** - Modify: `src/code_path_audit.py` - Test: `tests/test_code_path_audit.py` - [ ] **Step 1.2.1: Add the `ExpensiveOp` dataclass + the 7 cost class detection functions to `src/code_path_audit.py`** Append to `src/code_path_audit.py`: ```python CostClass = Literal[ "file_io", "network", "ast_parse", "json_io", "pickle", "deep_copy", "loop_amplified" ] @dataclass class ExpensiveOp: callee: str cost_class: CostClass data_size_estimate: int | None line: int weight: int _FILE_IO_PATTERNS: frozenset[str] = frozenset({"open", "read_text", "write_text", "read_bytes", "write_bytes"}) _NETWORK_PATTERNS: frozenset[str] = frozenset({"get", "post", "put", "delete", "request", "urlopen", "send"}) _AST_PATTERNS: frozenset[str] = frozenset({"parse", "walk", "iter_child_nodes"}) _JSON_PATTERNS: frozenset[str] = frozenset({"dump", "dumps", "load", "loads"}) _PICKLE_PATTERNS: frozenset[str] = frozenset({"pickle"}) _DEEP_COPY_PATTERNS: frozenset[str] = frozenset({"deepcopy"}) def classify_call(callee_name: str) -> CostClass | None: """Return the cost class for a call name, or None if not expensive.""" if callee_name in _FILE_IO_PATTERNS: return "file_io" if callee_name in _NETWORK_PATTERNS: return "network" if callee_name in _AST_PATTERNS: return "ast_parse" if callee_name in _JSON_PATTERNS: return "json_io" if "pickle" in callee_name: return "pickle" if callee_name in _DEEP_COPY_PATTERNS: return "deep_copy" return None ``` - [ ] **Step 1.2.2: Write the failing tests for the 7 cost classes** Append to `tests/test_code_path_audit.py`: ```python from src.code_path_audit import classify_call, COST_CLASS_WEIGHTS def test_classify_call_file_io() -> None: assert classify_call("open") == "file_io" assert classify_call("read_text") == "file_io" def test_classify_call_network() -> None: assert classify_call("get") == "network" assert classify_call("urlopen") == "network" def test_classify_call_ast_parse() -> None: assert classify_call("parse") == "ast_parse" assert classify_call("walk") == "ast_parse" def test_classify_call_json_io() -> None: assert classify_call("dump") == "json_io" assert classify_call("load") == "json_io" def test_classify_call_pickle() -> None: assert classify_call("pickle_dump") == "pickle" assert "pickle" in (classify_call("custom_pickle_helper") or "") def test_classify_call_deep_copy() -> None: assert classify_call("deepcopy") == "deep_copy" def test_classify_call_non_expensive() -> None: assert classify_call("len") is None assert classify_call("print") is None assert classify_call("range") is None def test_cost_class_weights_all_seven() -> None: expected = {"file_io", "network", "ast_parse", "json_io", "pickle", "deep_copy", "loop_amplified"} assert set(COST_CLASS_WEIGHTS.keys()) == expected assert COST_CLASS_WEIGHTS["network"] > COST_CLASS_WEIGHTS["file_io"] assert COST_CLASS_WEIGHTS["file_io"] > COST_CLASS_WEIGHTS["json_io"] ``` - [ ] **Step 1.2.3: Run all tests** Run: `uv run pytest tests/test_code_path_audit.py -q 2>&1 | Select-Object -Last 5` Expected: 13 tests pass (5 from 1.1.2 + 8 from 1.2.2). ### Task 1.3: StateMutationIndex + 5 mutation kinds **Files:** - Modify: `src/code_path_audit.py` - Test: `tests/test_code_path_audit.py` - [ ] **Step 1.3.1: Add the `StateMutation` dataclass + detection helpers to `src/code_path_audit.py`** Append to `src/code_path_audit.py`: ```python MutationKind = Literal["attr_write", "container_mutate", "file_write", "ipc_emit", "global_write"] @dataclass class StateMutation: target: str kind: MutationKind line: int ``` - [ ] **Step 1.3.2: Write the failing tests for the 5 mutation kinds (the kinds are documented in the spec; the AST detection is built in Phase 2 — Phase 1 only verifies the data structure)** Append to `tests/test_code_path_audit.py`: ```python from src.code_path_audit import StateMutation def test_state_mutation_5_kinds() -> None: """The 5 mutation kinds documented in the spec.""" expected = {"attr_write", "container_mutate", "file_write", "ipc_emit", "global_write"} mutations = [ StateMutation(target="self.history", kind="attr_write", line=10), StateMutation(target="self.entries.append", kind="container_mutate", line=20), StateMutation(target="file:logs/foo.log", kind="file_write", line=30), StateMutation(target="queue.put", kind="ipc_emit", line=40), StateMutation(target="module.events.X", kind="global_write", line=50), ] assert {m.kind for m in mutations} == expected def test_state_mutation_fields() -> None: m = StateMutation(target="self.x", kind="attr_write", line=42) assert m.target == "self.x" assert m.line == 42 ``` - [ ] **Step 1.3.3: Run all tests** Run: `uv run pytest tests/test_code_path_audit.py -q 2>&1 | Select-Object -Last 5` Expected: 15 tests pass. ### Task 1.4: Commit Phase 1 - [ ] **Step 1.4.1: Stage and commit** ```bash git add src/code_path_audit.py tests/test_code_path_audit.py git commit -m "feat(audit): add code_path_audit data structures (CallGraph, ExpensiveOp, StateMutation) 15 unit tests passing on synthetic 5-node graphs. The 7 cost classes (file_io, network, ast_parse, json_io, pickle, deep_copy, loop_amplified) and 5 mutation kinds (attr_write, container_mutate, file_write, ipc_emit, global_write) are defined as Literal types and detected by name pattern in classify_call(). EXPENSIVE_THRESHOLD = 40_000 module constant. COST_CLASS_WEIGHTS dict with the 7 classes; network (500) > file_io (100) > json_io (150) ordering. Tests in tests/test_code_path_audit.py. Phase 2 will add trace_action + build_call_graph (AST walking src/ to populate the indexes) + ActionProfile." ``` - [ ] **Step 1.4.2: Attach git note** ```bash git notes add -m "feat(audit) Phase 1: data structures (CallGraph, ExpensiveOp, StateMutation) 15 unit tests pass on synthetic graphs and pattern detection. The 3 data structures + 7 cost classes + 5 mutation kinds are the foundation; Phase 2 adds the AST walker that populates them from real src/. EXPENSIVE_THRESHOLD = 40_000 is a module-level constant. The runtime-profiling follow-up will calibrate this number based on actual measurements." ``` - [ ] **Step 1.4.3: Update state.toml: mark phase_1 = completed; current_phase = 2; commit** - [ ] **Step 1.4.4: Conductor - User Manual Verification** --- ## Phase 2: trace_action + ActionProfile + cost model **Files:** `src/code_path_audit.py`, `tests/test_code_path_audit.py`. This phase is one commit. AST walking the real `src/` to populate the indexes. ### Task 2.1: Action + ActionProfile dataclasses - [ ] **Step 2.1.1: Add the `Action` + `ActionProfile` dataclasses to `src/code_path_audit.py`** Append to `src/code_path_audit.py`: ```python @dataclass class Action: name: str entry_points: list[str] description: str @dataclass class ActionProfile: action: Action call_graph: CallGraph expensive_ops: list[ExpensiveOp] state_mutations: list[StateMutation] redundancy: list[tuple[str, int]] pipelining_candidates: list[list[str]] total_load_estimate: int unresolved_calls: list[str] mermaid: str markdown: str # The 3 actions in scope (MMA is cold per user) ACTIONS: dict[str, Action] = { "ai_message_lifecycle": Action( name="ai_message_lifecycle", entry_points=[ "src.app_controller.AppController.process_user_request", "src.ai_client.AIClient.send", "src.aggregate.build_file_items", "src.summarize._summarise_generic", "src.summarize._summarise_markdown", ], description="AI message lifecycle: context aggregation -> AI call -> response -> history update.", ), "discussion_save_load": Action( name="discussion_save_load", entry_points=[ "src.project_manager.save_project", "src.project_manager.load_project", "src.history.HistoryManager.save_snapshot", "src.models.parse_history_entries", ], description="Discussion save/load: snapshot -> serialize -> write TOML -> read TOML -> deserialize.", ), "gui_startup": Action( name="gui_startup", entry_points=[ "gui_2.App.__init__", "src.app_controller.AppController.__init__", "src.paths._resolve_config_path", ], description="GUI startup: paths init -> config load -> controller init -> first render.", ), } ``` - [ ] **Step 2.1.2: Write tests for the 3 actions in `tests/test_code_path_audit.py`** Append to `tests/test_code_path_audit.py`: ```python from src.code_path_audit import ACTIONS def test_actions_3_in_scope() -> None: assert set(ACTIONS.keys()) == {"ai_message_lifecycle", "discussion_save_load", "gui_startup"} def test_actions_have_entry_points() -> None: for name, action in ACTIONS.items(): assert action.entry_points, f"{name} has no entry points" assert action.description, f"{name} has no description" def test_ai_message_entry_points_cover_pipeline() -> None: entry = ACTIONS["ai_message_lifecycle"].entry_points # context aggregation -> AI send -> history update -> summarization assert any("build_file_items" in e for e in entry) assert any("AIClient.send" in e for e in entry) assert any("process_user_request" in e for e in entry) def test_mma_is_cold_not_in_actions() -> None: for action in ACTIONS.values(): for ep in action.entry_points: assert "multi_agent_conductor" not in ep assert "ConductorEngine" not in ep assert "WorkerPool" not in ep ``` ### Task 2.2: AST walker + build_call_graph + build_*_index - [ ] **Step 2.2.1: Add the AST walking + index builders to `src/code_path_audit.py`** Append to `src/code_path_audit.py`: ```python def _fqname(file: str, name: str) -> str: return f"{file.removesuffix('.py').replace('/', '.')}.{name}" def _walk_calls(node: ast.AST) -> list[tuple[str, int]]: """Return [(callee_name, line)] for every Call in node.""" out: list[tuple[str, int]] = [] for n in ast.walk(node): if isinstance(n, ast.Call): if isinstance(n.func, ast.Name): out.append((n.func.id, n.lineno)) elif isinstance(n.func, ast.Attribute): out.append((n.func.attr, n.lineno)) return out def _walk_attr_writes(node: ast.AST) -> list[tuple[str, int]]: """Return [(target, line)] for self.X = ... and module.X = ... assignments.""" out: list[tuple[str, int]] = [] for n in ast.walk(node): if isinstance(n, ast.Assign): for target in n.targets: if isinstance(target, ast.Attribute) and isinstance(target.value, ast.Name): out.append((f"{target.value.id}.{target.attr}", n.lineno)) return out def build_call_graph(src_dir: str = "src") -> CallGraph: cg = CallGraph() for py_file in Path(src_dir).rglob("*.py"): if "__pycache__" in str(py_file): continue try: tree = ast.parse(py_file.read_text(encoding="utf-8")) except SyntaxError: continue module_fq = _fqname(str(py_file), "") for node in ast.walk(tree): if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): fq = _fqname(str(py_file), node.name) cg.nodes[fq] = FunctionNode(fqname=fq, file=str(py_file), line=node.lineno) cg.nodes[fq].calls = [c for c, _ in _walk_calls(node)] cg.nodes[fq].state_mutations = [ StateMutation(target=t, kind="attr_write", line=ln) for t, ln in _walk_attr_writes(node) ] for callee, line in _walk_calls(node): cg.add_edge(fq, callee) cg.nodes[module_fq] = FunctionNode(fqname=module_fq, file=str(py_file), line=0) return cg def build_expensive_ops_index(cg: CallGraph) -> dict[str, list[ExpensiveOp]]: out: dict[str, list[ExpensiveOp]] = {} for fq, node in cg.nodes.items(): ops: list[ExpensiveOp] = [] for callee, line in node.calls: cost_class = classify_call(callee) if cost_class is None: continue weight = COST_CLASS_WEIGHTS[cost_class] ops.append(ExpensiveOp(callee=callee, cost_class=cost_class, data_size_estimate=None, line=line, weight=weight)) out[fq] = ops return out def build_state_mutations_index(cg: CallGraph) -> dict[str, list[StateMutation]]: return {fq: node.state_mutations for fq, node in cg.nodes.items() if node.state_mutations} ``` (Note: the file uses 1-space indent per project style. The implementer should preserve this when adding the code.) - [ ] **Step 2.2.2: Add the `trace_action` function to `src/code_path_audit.py`** Append: ```python def trace_action(action: Action, max_depth: int = 10) -> ActionProfile: cg = build_call_graph() expensive_index = build_expensive_ops_index(cg) mutations_index = build_state_mutations_index(cg) reachable: set[str] = set() for ep in action.entry_points: if ep in cg.nodes: reachable.add(ep) reachable |= cg.transitive_callees(ep, max_depth=max_depth) reachable_ops: list[ExpensiveOp] = [] for fq in reachable: reachable_ops.extend(expensive_index.get(fq, [])) cg.nodes[fq].expensive_ops = expensive_index.get(fq, []) reachable_mutations: list[StateMutation] = [] for fq in reachable: reachable_mutations.extend(mutations_index.get(fq, [])) call_counts: dict[str, int] = {} for fq in reachable: for callee in cg.nodes[fq].calls: call_counts[callee] = call_counts.get(callee, 0) + 1 redundancy = [(op, n) for op, n in call_counts.items() if n > 1] total_load = sum(op.weight for op in reachable_ops) subgraph = CallGraph() for fq in reachable: subgraph.nodes[fq] = cg.nodes[fq] for callee in cg.edges.get(fq, ()): if callee in reachable: subgraph.add_edge(fq, callee) unresolved = [op.callee for op in reachable_ops if op.cost_class is None] return ActionProfile( action=action, call_graph=subgraph, expensive_ops=reachable_ops, state_mutations=reachable_mutations, redundancy=redundancy, pipelining_candidates=[], total_load_estimate=total_load, unresolved_calls=unresolved, mermaid="", markdown="", ) ``` - [ ] **Step 2.2.3: Add an integration test that builds the real call graph and traces a real action** Append to `tests/test_code_path_audit.py`: ```python def test_build_call_graph_real_src() -> None: """Smoke test: build the real call graph of src/ and verify it has nodes.""" from src.code_path_audit import build_call_graph cg = build_call_graph() assert len(cg.nodes) > 50, f"expected 50+ nodes, got {len(cg.nodes)}" def test_trace_action_ai_message_returns_profile() -> None: """Integration test: trace the AI message action against real src/.""" from src.code_path_audit import trace_action, ACTIONS profile = trace_action(ACTIONS["ai_message_lifecycle"], max_depth=5) assert profile.action.name == "ai_message_lifecycle" assert len(profile.expensive_ops) >= 0 assert profile.total_load_estimate >= 0 def test_trace_action_save_load_returns_profile() -> None: from src.code_path_audit import trace_action, ACTIONS profile = trace_action(ACTIONS["discussion_save_load"], max_depth=5) assert profile.action.name == "discussion_save_load" def test_trace_action_gui_startup_returns_profile() -> None: from src.code_path_audit import trace_action, ACTIONS profile = trace_action(ACTIONS["gui_startup"], max_depth=5) assert profile.action.name == "gui_startup" ``` - [ ] **Step 2.2.4: Run all tests** Run: `uv run pytest tests/test_code_path_audit.py -q 2>&1 | Select-Object -Last 10` Expected: 19 tests pass (15 from Phase 1 + 4 from 2.2.3). The integration tests may take a few seconds (AST-walking 61 files). ### Task 2.3: Commit Phase 2 - [ ] **Step 2.3.1: Stage and commit** ```bash git add src/code_path_audit.py tests/test_code_path_audit.py git commit -m "feat(audit): add trace_action + ActionProfile + AST walking The 3 actions in scope (ai_message_lifecycle, discussion_save_load, gui_startup) are declared as Action instances. trace_action() builds the full call graph of src/, traverses the entry points to depth 10, and returns an ActionProfile with: expensive ops, state mutations, redundancy (ops called >1x), unresolved calls, and total load estimate. AST walker: build_call_graph() parses each src/*.py file, extracts FunctionDef/AsyncFunctionDef nodes, indexes self.X assignments as StateMutation, and classifies each Call as file_io / network / ast_parse / json_io / pickle / deep_copy or non-expensive. MMA worker spawn is intentionally absent from ACTIONS (per user: 'keeping that cold until the 1:1 discussion UX is dogfooded in a few projects'). 23 unit + integration tests passing on synthetic + real src/." ``` - [ ] **Step 2.3.2: Attach git note + update state.toml (phase_2 = completed; current_phase = 3)** - [ ] **Step 2.3.3: Conductor - User Manual Verification** --- ## Phase 3: Output (JSON / markdown / Mermaid) **Files:** `src/code_path_audit.py`, `tests/test_code_path_audit.py`. This phase is one commit. Three sub-tasks (one per output format). ### Task 3.1: JSON serializer - [ ] **Step 3.1.1: Add the `to_json` and `dump_json` functions to `src/code_path_audit.py`** Append: ```python def _to_jsonable(obj: object) -> object: if isinstance(obj, (str, int, float, bool, type(None))): return obj if isinstance(obj, (list, tuple, set)): return [_to_jsonable(x) for x in obj] if isinstance(obj, dict): return {str(k): _to_jsonable(v) for k, v in obj.items()} if hasattr(obj, "__dataclass_fields__"): return {k: _to_jsonable(getattr(obj, k)) for k in obj.__dataclass_fields__} return repr(obj) def to_json(profile: ActionProfile) -> str: return json.dumps(_to_jsonable(profile), indent=2) def dump_json(profile: ActionProfile, path: str) -> None: Path(path).write_text(to_json(profile), encoding="utf-8") ``` - [ ] **Step 3.1.2: Add tests for JSON output** Append to `tests/test_code_path_audit.py`: ```python import json from src.code_path_audit import trace_action, ACTIONS, to_json def test_to_json_round_trip() -> None: profile = trace_action(ACTIONS["ai_message_lifecycle"], max_depth=3) js = to_json(profile) parsed = json.loads(js) assert parsed["action"]["name"] == "ai_message_lifecycle" assert "call_graph" in parsed assert "expensive_ops" in parsed assert "state_mutations" in parsed def test_to_json_serializes_sets_as_lists() -> None: from src.code_path_audit import CallGraph cg = CallGraph() cg.add_edge("a", "b") js = to_json(cg) parsed = json.loads(js) assert isinstance(parsed["nodes"], dict) assert isinstance(parsed["edges"]["a"], list) ``` ### Task 3.2: Markdown renderer - [ ] **Step 3.2.1: Add the `to_markdown` function to `src/code_path_audit.py`** Append: ```python def to_markdown(profile: ActionProfile) -> str: lines: list[str] = [ f"# Action Profile: {profile.action.name}", "", f"**Description:** {profile.action.description}", "", f"**Total load estimate:** {profile.total_load_estimate:,}", f"**Expensive ops count:** {len(profile.expensive_ops)}", f"**State mutations count:** {len(profile.state_mutations)}", f"**Redundancy:** {len(profile.redundancy)} ops called >1x", f"**Unresolved calls:** {len(profile.unresolved_calls)}", "", "## Expensive Operations", "", "| Callee | Cost class | Weight | Line |", "|--------|------------|--------|------|", ] for op in sorted(profile.expensive_ops, key=lambda o: -o.weight)[:50]: lines.append(f"| `{op.callee}` | {op.cost_class} | {op.weight:,} | {op.line} |") if not profile.expensive_ops: lines.append("| _(none)_ | - | - | - |") lines += ["", "## State Mutations (first 50)", "", "| Target | Kind | Line |", "|--------|------|------|"] for m in profile.state_mutations[:50]: lines.append(f"| `{m.target}` | {m.kind} | {m.line} |") if not profile.state_mutations: lines.append("| _(none)_ | - | - |") lines += ["", "## Redundancy (ops called >1x)", ""] if profile.redundancy: for op, count in sorted(profile.redundancy, key=lambda x: -x[1])[:20]: lines.append(f"- `{op}` called {count}x") else: lines.append("_(none)_") lines += ["", "## Unresolved Calls", ""] if profile.unresolved_calls: for c in profile.unresolved_calls[:20]: lines.append(f"- `{c}`") else: lines.append("_(none)_") return "\n".join(lines) ``` - [ ] **Step 3.2.2: Add tests for markdown output** Append to `tests/test_code_path_audit.py`: ```python from src.code_path_audit import to_markdown def test_to_markdown_contains_action_name() -> None: profile = trace_action(ACTIONS["ai_message_lifecycle"], max_depth=3) md = to_markdown(profile) assert "# Action Profile: ai_message_lifecycle" in md assert "Total load estimate:" in md assert "## Expensive Operations" in md assert "## State Mutations" in md ``` ### Task 3.3: Mermaid generator - [ ] **Step 3.3.1: Add the `to_mermaid` function to `src/code_path_audit.py`** Append: ```python def to_mermaid(profile: ActionProfile, max_depth: int = 5) -> str: return profile.call_graph.render_mermaid( profile.action.entry_points[0] if profile.action.entry_points else "", max_depth=max_depth, ) ``` - [ ] **Step 3.3.2: Add tests for Mermaid output** Append to `tests/test_code_path_audit.py`: ```python from src.code_path_audit import to_mermaid def test_to_mermaid_basic() -> None: profile = trace_action(ACTIONS["ai_message_lifecycle"], max_depth=3) mmd = to_mermaid(profile, max_depth=3) assert "graph LR" in mmd or "" in mmd ``` ### Task 3.4: Commit Phase 3 - [ ] **Step 3.4.1: Stage and commit** ```bash git add src/code_path_audit.py tests/test_code_path_audit.py git commit -m "feat(audit): add JSON / markdown / Mermaid output to_json / to_markdown / to_mermaid functions serialize an ActionProfile. JSON is round-trippable. Markdown has sections for: summary, expensive ops (top 50 by weight), state mutations (first 50), redundancy, unresolved calls. Mermaid is rendered from the subgraph rooted at the first entry point. 27 tests passing total." ``` - [ ] **Step 3.4.2: Attach git note + update state.toml (phase_3 = completed; current_phase = 4)** - [ ] **Step 3.4.3: Conductor - User Manual Verification** --- ## Phase 4: MCP tool + CLI surface **Files:** `src/code_path_audit.py` (extends), `opencode.json` (modify if MCP registration needed). ### Task 4.1: CLI - [ ] **Step 4.1.1: Add the CLI module-level entry to `src/code_path_audit.py`** Append: ```python if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="Code path audit tool.") parser.add_argument("--action", required=True, choices=list(ACTIONS.keys())) parser.add_argument("--depth", type=int, default=10) parser.add_argument("--output-dir", default="docs/reports/code_path_audit") parser.add_argument("--date", default=None, help="ISO date; defaults to today") args = parser.parse_args() profile = trace_action(ACTIONS[args.action], max_depth=args.depth) profile.markdown = to_markdown(profile) profile.mermaid = to_mermaid(profile, max_depth=min(5, args.depth)) from datetime import date date_str = args.date or date.today().isoformat() out_dir = Path(args.output_dir) / date_str (out_dir / "actions").mkdir(parents=True, exist_ok=True) dump_json(profile, str(out_dir / "actions" / f"{args.action}.json")) (out_dir / "actions" / f"{args.action}.md").write_text(profile.markdown, encoding="utf-8") (out_dir / "actions" / f"{args.action}.mmd").write_text(profile.mermaid, encoding="utf-8") print(f"Wrote {out_dir / 'actions' / args.action}.{{json,md,mmd}}") ``` - [ ] **Step 4.1.2: Add a CLI smoke test** Append to `tests/test_code_path_audit.py`: ```python def test_cli_help() -> None: import subprocess result = subprocess.run( ["python", "-m", "src.code_path_audit", "--help"], capture_output=True, text=True, timeout=10, ) assert result.returncode == 0 assert "--action" in result.stdout ``` ### Task 4.2: MCP tool registration - [ ] **Step 4.2.1: Add the MCP-tool function to `src/code_path_audit.py`** Append: ```python def code_path_audit(action_name: str, max_depth: int = 10) -> dict: """MCP tool: trace an action and return its profile as a dict.""" if action_name not in ACTIONS: raise ValueError(f"Unknown action: {action_name}. Known: {list(ACTIONS.keys())}") profile = trace_action(ACTIONS[action_name], max_depth=max_depth) profile.markdown = to_markdown(profile) profile.mermaid = to_mermaid(profile, max_depth=min(5, max_depth)) return _to_jsonable(profile) ``` - [ ] **Step 4.2.2: Add the tool to `src/models.py` (or wherever MCP tools are registered)** Check `src/models.py:100` and `src/models.py:144` (per the spec's audit finding). If the project has a `MCP_TOOLS` or similar registry, add `code_path_audit` there. If not, add it where appropriate. (This step requires the implementer to find the right registration point; the test below verifies the function exists.) - [ ] **Step 4.2.3: Add a test for the MCP-tool function** Append to `tests/test_code_path_audit.py`: ```python def test_code_path_audit_mcp_function() -> None: from src.code_path_audit import code_path_audit result = code_path_audit("ai_message_lifecycle", max_depth=3) assert result["action"]["name"] == "ai_message_lifecycle" assert "expensive_ops" in result def test_code_path_audit_mcp_unknown_action_raises() -> None: from src.code_path_audit import code_path_audit import pytest with pytest.raises(ValueError, match="Unknown action"): code_path_audit("mma_worker_spawn", max_depth=3) ``` ### Task 4.3: Commit Phase 4 - [ ] **Step 4.3.1: Stage and commit** ```bash git add src/code_path_audit.py tests/test_code_path_audit.py src/models.py opencode.json git commit -m "feat(audit): add MCP tool + CLI surface CLI: python -m src.code_path_audit --action [--depth N] [--output-dir DIR] [--date YYYY-MM-DD]. Writes JSON + MD + MMD to //actions/.{json,md,mmd}. MCP tool: code_path_audit(action_name, max_depth=10) -> dict. Raises ValueError on unknown action. Registered alongside the other MCP tools in src/models.py. 30 tests passing total." ``` - [ ] **Step 4.3.2: Attach git note + update state.toml (phase_4 = completed; current_phase = 5)** - [ ] **Step 4.3.3: Conductor - User Manual Verification** --- ## Phase 5: Run audit on 3 actions; commit report **Files:** `docs/reports/code_path_audit/2026-06-07/` (create + populate). This phase is one commit. The deliverable IS the report. ### Task 5.1: Run audits + produce the report - [ ] **Step 5.1.1: Run the audit on the 3 actions** ```bash uv run python -m src.code_path_audit --action ai_message_lifecycle --depth 10 --date 2026-06-07 uv run python -m src.code_path_audit --action discussion_save_load --depth 10 --date 2026-06-07 uv run python -m src.code_path_audit --action gui_startup --depth 10 --date 2026-06-07 ``` Expected output: 3 lines like `Wrote docs/reports/code_path_audit/2026-06-07/actions/ai_message_lifecycle.{json,md,mmd}`. - [ ] **Step 5.1.2: Generate the full call graph + indexes** ```bash uv run python -c " import sys sys.path.insert(0, 'src') from pathlib import Path from code_path_audit import build_call_graph, build_expensive_ops_index, build_state_mutations_index, _to_jsonable, COST_CLASS_WEIGHTS import json cg = build_call_graph() out = Path('docs/reports/code_path_audit/2026-06-07') (out).mkdir(parents=True, exist_ok=True) (out / 'call_graph.json').write_text(json.dumps(_to_jsonable(cg), indent=2), encoding='utf-8') (out / 'expensive_ops.json').write_text(json.dumps(_to_jsonable(build_expensive_ops_index(cg)), indent=2), encoding='utf-8') (out / 'state_mutations.json').write_text(json.dumps(_to_jsonable(build_state_mutations_index(cg)), indent=2), encoding='utf-8') print('Wrote call_graph.json, expensive_ops.json, state_mutations.json') " ``` Expected: `Wrote call_graph.json, expensive_ops.json, state_mutations.json` - [ ] **Step 5.1.3: Produce `summary.md`** The implementer writes `docs/reports/code_path_audit/2026-06-07/summary.md` by hand, synthesizing the 3 per-action reports. The summary structure (from the spec): ```markdown # Code Path Audit — 2026-06-07 ## Top-level summary - AI message lifecycle: N expensive ops, total load X - Discussion save/load: N expensive ops, total load X - GUI startup: N expensive ops, total load X ## Top optimization candidates (across all 3 actions) 1. (caller, callee) — current cost X, proposed reduction Y, effort Z 2. ... ## Caveats - The cost model is heuristic; EXPENSIVE_THRESHOLD = 40_000. - AST walking misses dynamic patterns (eval, getattr, decorator dispatch). - The runtime-profiling follow-up (pipeline_runtime_profiling_20260607) will calibrate. - MMA worker spawn is OUT of scope (user: cold until 1:1 discussion UX dogfooded). ``` The actual content is filled in by the implementer based on the 3 per-action reports. - [ ] **Step 5.1.4: Produce `optimization_candidates.md`** A ranked list extracted from `summary.md`. Each candidate: location, current cost, proposed change, expected reduction, effort, priority. ### Task 5.2: Commit Phase 5 - [ ] **Step 5.2.1: Stage and commit the report** ```bash git add docs/reports/code_path_audit/2026-06-07/ git commit -m "docs(audit): run code path audit on 3 actions; commit report Generated artifacts under docs/reports/code_path_audit/2026-06-07/: - call_graph.json (full call graph of src/) - expensive_ops.json (per-function expensive-op index) - state_mutations.json (per-function state-mutation index) - actions/ai_message_lifecycle.{json,md,mmd} - actions/discussion_save_load.{json,md,mmd} - actions/gui_startup.{json,md,mmd} - summary.md (cross-action summary + top candidates) - optimization_candidates.md (ranked list) EXPENSIVE_THRESHOLD = 40_000. Cost model is heuristic; the pipeline_runtime_profiling_20260607 follow-up will calibrate. MMA worker spawn is intentionally absent from ACTIONS." ``` - [ ] **Step 5.2.2: Attach git note + update state.toml (phase_5 = completed; current_phase = 6; update verification booleans)** - [ ] **Step 5.2.3: Conductor - User Manual Verification (per workflow.md)** Ask the user to review the report. This is the deliverable — they need to confirm the optimization candidates make sense. --- ## Phase 6: tracks.md update **Files:** `conductor/tracks.md` (modify). - [ ] **Step 6.1: Add the track entry to `conductor/tracks.md`** Open `conductor/tracks.md`. Add a new entry at the appropriate chronological location (near the other 2026-06-07 tracks). Use the format from recent tracks: ```markdown - [x] **Track: Code Path & Data Pipeline Audit** `[checkpoint: ]` *Link: [./tracks/code_path_audit_20260607/](./tracks/code_path_audit_20260607/), Spec: [./tracks/code_path_audit_20260607/spec.md](./tracks/code_path_audit_20260607/spec.md), Plan: [./tracks/code_path_audit_20260607/plan.md](./tracks/code_path_audit_20260607/plan.md)* *Goal: Build `src/code_path_audit.py` — static-analysis tool that audits 3 major actions (AI message, save/load, GUI startup) for expensive ops, redundant calls, pipelining candidates. 7 cost classes, 5 mutation kinds, EXPENSIVE_THRESHOLD = 40_000. Output: JSON + MD + Mermaid in `docs/reports/code_path_audit/2026-06-07/`. MMA worker spawn is OUT of scope (user: cold). Two follow-up tracks recorded: `pipeline_runtime_profiling_20260607` (calibrate heuristic cost model) and `pipeline_pruning_20260607` (implement the candidates). 24+ tests passing.* ``` Replace `` with the SHA from the report commit in Phase 5. - [ ] **Step 6.2: Commit the tracks.md update** ```bash git add conductor/tracks.md git commit -m "conductor(tracks): mark Code Path Audit track as complete Phase 6 verification complete: 6 atomic commits landed, the audit was run on the 3 actions, the report is in docs/reports/code_path_audit/2026-06-07/, all 24+ unit and integration tests pass. MMA worker spawn is out of scope (per user: cold). Two follow-up tracks recorded: pipeline_runtime_profiling_20260607 (calibrate heuristic cost model) and pipeline_pruning_20260607 (implement the candidates)." ``` - [ ] **Step 6.3: Attach git note + update state.toml (phase_6 = completed; current_phase = "complete"; status = "completed")** - [ ] **Step 6.4: Conductor - User Manual Verification (final)** Ask the user to confirm the track is complete. --- ## Summary - **6 phases**, **6 atomic commits**, **30 tests**. - **3 per-action reports** + 1 cross-action summary + 1 ranked candidates list. - **2 follow-up tracks** recorded (runtime profiling + pruning). - **No new pip dependencies**; pure stdlib (ast, json, pathlib, dataclasses). - **Read-only on `src/`**: the audit doesn't modify existing code. The new files are `src/code_path_audit.py` + `tests/test_code_path_audit.py` + the report under `docs/reports/code_path_audit/2026-06-07/`. - **Reusable**: re-run after any `src/` change to see drift. The 3 actions are declared once in `ACTIONS`; adding a 4th is one `Action(...)` declaration. - **Calibration target**: `pipeline_runtime_profiling_20260607` will use the existing `src/performance_monitor.py` to measure real costs and recalibrate `EXPENSIVE_THRESHOLD` + `COST_CLASS_WEIGHTS`.