# 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: custom postfix `.dsl` data + markdown + Mermaid + prefix tree text 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 a custom postfix `.dsl` (machine) + markdown + Mermaid (visual) + prefix tree text (human). The postfix `.dsl` is a custom DSL tailored to the audit's record shapes — tagged records (each "word" is a constructor with a known arity), length-prefixed lists, whitespace-tokenized, with `"..."` quoting only when needed. The prefix tree renderer is a separate view of the same data, generated by a recursive walker. 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), `dataclasses` (stdlib), `re` (stdlib for the DSL tokenizer). 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 (custom postfix .dsl / markdown / Mermaid / tree)" } 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_dsl_produced = false expensive_ops_dsl_produced = false state_mutations_dsl_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 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 (custom postfix .dsl / tree / markdown / Mermaid) **Files:** `src/code_path_audit.py`, `tests/test_code_path_audit.py`. This phase is one commit. Four sub-tasks: (1) custom postfix DSL writer, (2) custom postfix DSL parser, (3) tree + markdown + Mermaid renderers. ### Task 3.1: Custom postfix DSL writer - [ ] **Step 3.1.1: Add the `to_dsl` and `dump_dsl` functions to `src/code_path_audit.py`** Append: ```python DSL_ATOM_QUOTE_CHARS: frozenset[str] = frozenset({'"', "'"}) def _needs_quoting(atom: str) -> bool: """Bare atoms are quoted only if they contain whitespace or special chars.""" if not atom: return True return any(c in atom for c in (" ", "\t", "\n", "\\", '"', "'", "()[]{}")) def _format_atom(value: object) -> str: if value is None: return "nil" if isinstance(value, bool): return "true" if value else "false" if isinstance(value, (int, float)): return str(value) s = str(value) if _needs_quoting(s): return f'"{s}"' return s def to_dsl(profile: ActionProfile) -> str: """Serialize an ActionProfile to custom postfix DSL (RPN) text.""" lines: list[str] = ["\ ActionProfile (postfix DSL)", f"\ generated 2026-06-07 by src.code_path_audit", ""] # action act = profile.action lines.append(" \ action( name description entry_points )") lines.append(f" {_format_atom(act.name)}") lines.append(f" {_format_atom(act.description)}") lines.append(f" {len(act.entry_points)}") for ep in act.entry_points: lines.append(f" {_format_atom(ep)}") lines.append(f" {len(act.entry_points)} list") lines.append(" action") lines.append("") # expensive_ops lines.append(f" \ expensive_ops: {len(profile.expensive_ops)} items") lines.append(f" {len(profile.expensive_ops)}") for op in profile.expensive_ops: lines.append(f" {_format_atom(op.callee)} {_format_atom(op.cost_class)} {_format_atom(op.data_size_estimate)} {op.line} {op.weight} exp-op") lines.append(f" {len(profile.expensive_ops)} list") lines.append("") # state_mutations lines.append(f" \ state_mutations: {len(profile.state_mutations)} items") lines.append(f" {len(profile.state_mutations)}") for m in profile.state_mutations: lines.append(f" {_format_atom(m.target)} {_format_atom(m.kind)} {m.line} mut") lines.append(f" {len(profile.state_mutations)} list") lines.append("") # redundancy lines.append(f" \ redundancy: {len(profile.redundancy)} items") lines.append(f" {len(profile.redundancy)}") for op, count in profile.redundancy: lines.append(f" {_format_atom(op)} {count} pair") lines.append(f" {len(profile.redundancy)} list") lines.append("") # total_load_estimate lines.append(" \ total_load_estimate") lines.append(f" {profile.total_load_estimate} int") lines.append("") # unresolved_calls lines.append(f" \ unresolved_calls: {len(profile.unresolved_calls)} items") lines.append(f" {len(profile.unresolved_calls)}") for c in profile.unresolved_calls: lines.append(f" {_format_atom(c)}") lines.append(f" {len(profile.unresolved_calls)} list") lines.append("") return "\n".join(lines) def dump_dsl(profile: ActionProfile, path: str) -> None: Path(path).write_text(to_dsl(profile), encoding="utf-8") ``` - [ ] **Step 3.1.2: Add tests for the DSL writer** Append to `tests/test_code_path_audit.py`: ```python from src.code_path_audit import to_dsl, dump_dsl def test_to_dsl_contains_action_name() -> None: profile = trace_action(ACTIONS["ai_message_lifecycle"], max_depth=3) dsl = to_dsl(profile) assert "ai_message_lifecycle" in dsl assert "action" in dsl assert "exp-op" in dsl or "expensive_ops" in dsl def test_to_dsl_uses_postfix_order() -> None: """The DSL is postfix: tagged words come AFTER their args.""" profile = trace_action(ACTIONS["ai_message_lifecycle"], max_depth=3) dsl = to_dsl(profile) # action word should appear AFTER its 3 args action_idx = dsl.find("\n action\n") assert action_idx > 0 def test_dump_dsl_writes_file(tmp_path) -> None: profile = trace_action(ACTIONS["ai_message_lifecycle"], max_depth=3) target = tmp_path / "test.dsl" dump_dsl(profile, str(target)) content = target.read_text(encoding="utf-8") assert "ai_message_lifecycle" in content ``` ### Task 3.2: Custom postfix DSL parser - [ ] **Step 3.2.1: Add the `parse_dsl` function to `src/code_path_audit.py`** Append: ```python import re from typing import Any DSL_WORD_ARITY: dict[str, int] = { "action": 3, "fn": 3, "call": 1, "mut": 3, "exp-op": 5, "pair": 2, "int": 1, } def _tokenize_dsl(text: str) -> list[str]: """Whitespace-tokenize; preserve \"...\" quoted atoms; drop \\\\ line comments.""" tokens: list[str] = [] for line in text.splitlines(): line = re.sub(r"\\\\.*", "", line) # strip line comments if not line.strip(): continue i = 0 while i < len(line): c = line[i] if c.isspace(): i += 1 continue if c == '"': j = line.find('"', i + 1) if j == -1: j = len(line) tokens.append(line[i + 1 : j]) i = j + 1 else: j = i while j < len(line) and not line[j].isspace(): j += 1 tokens.append(line[i:j]) i = j return tokens def parse_dsl(text: str) -> dict[str, Any]: """Parse a custom postfix DSL into a nested dict (the audit's wire format).""" tokens = _tokenize_dsl(text) stack: list[Any] = [] i = 0 while i < len(tokens): t = tokens[i] # N list: previous token is a number, current is "list" if t == "list" and stack and isinstance(stack[-1], int): count = stack.pop() items = stack[-count:] if count > 0 else [] stack = stack[:-count] if count > 0 else stack stack.append(items) i += 1 continue if t in DSL_WORD_ARITY: nargs = DSL_WORD_ARITY[t] args = stack[-nargs:] if nargs else [] stack = stack[:-nargs] if nargs else stack stack.append({"_tag": t, "_args": args}) i += 1 continue if t == "nil": stack.append(None) elif t == "true": stack.append(True) elif t == "false": stack.append(False) elif t.lstrip("-").isdigit(): stack.append(int(t)) else: stack.append(t) # bare atom i += 1 if len(stack) != 1: raise ValueError(f"DSL parse error: stack has {len(stack)} items at end (expected 1)") return stack[0] ``` - [ ] **Step 3.2.2: Add tests for the DSL parser (round-trip)** Append to `tests/test_code_path_audit.py`: ```python from src.code_path_audit import to_dsl, parse_dsl, _tokenize_dsl def test_dsl_round_trip() -> None: profile = trace_action(ACTIONS["ai_message_lifecycle"], max_depth=3) dsl = to_dsl(profile) parsed = parse_dsl(dsl) assert isinstance(parsed, dict) assert parsed["_tag"] == "action-profile" or "action" in str(parsed) def test_dsl_tokenize_handles_quotes() -> None: tokens = _tokenize_dsl('"hello world" bare_atom 42 nil') assert tokens == ["hello world", "bare_atom", 42, None] or tokens == ["hello world", "bare_atom", "42", "nil"] def test_dsl_tokenize_strips_line_comments() -> None: tokens = _tokenize_dsl("\\ this is a comment\nbare_atom\n\\ another comment") assert "bare_atom" in tokens assert not any("comment" in t for t in tokens) def test_dsl_parser_length_prefixed_list() -> None: tokens = _tokenize_dsl("a b c 3 list") parsed = parse_dsl("a b c 3 list") assert parsed == ["a", "b", "c"] ``` ### Task 3.3: Tree + Markdown + Mermaid renderers - [ ] **Step 3.3.1: Add the `to_tree` function to `src/code_path_audit.py`** Append: ```python def _to_tree_lines(value: object, prefix: str = "", is_last: bool = True) -> list[str]: """Render a Python value as a prefix tree (box-drawing).""" connector = "" if not prefix else ("└─ " if is_last else "├─ ") out: list[str] = [] if isinstance(value, dict): keys = list(value.keys()) for i, k in enumerate(keys): last = (i == len(keys) - 1) out.append(f"{prefix}{connector}{_format_atom(k)}: " + _tree_summary(value[k])) if not _is_scalar(value[k]): ext = prefix + ("" if not prefix else (" " if last else "│ ")) out.extend(_to_tree_lines(value[k], ext, True)) elif isinstance(value, list): out.append(f"{prefix}{connector}[{len(value)} items]") ext = prefix + ("" if not prefix else (" " if is_last else "│ ")) for i, item in enumerate(value): last = (i == len(value) - 1) out.append(f"{ext}{'└─ ' if last else '├─ '}" + _tree_summary(item)) if not _is_scalar(item): out.extend(_to_tree_lines(item, ext + (" " if last else "│ "), True)) else: out.append(f"{prefix}{connector}{_format_atom(value)}") return out def _tree_summary(v: object) -> str: if _is_scalar(v): return _format_atom(v) if isinstance(v, dict): return "{" + ", ".join(f"{k}=..." for k in list(v.keys())[:3]) + "}" if isinstance(v, list): return f"[{len(v)} items]" return type(v).__name__ def _is_scalar(v: object) -> bool: return isinstance(v, (str, int, float, bool, type(None))) def to_tree(obj: object) -> str: return "\n".join(_to_tree_lines(obj)) ``` - [ ] **Step 3.3.2: 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.3.3: 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.4: Add tests for tree + markdown + mermaid** Append to `tests/test_code_path_audit.py`: ```python from src.code_path_audit import to_tree, to_markdown, to_mermaid def test_to_tree_contains_action_name() -> None: profile = trace_action(ACTIONS["ai_message_lifecycle"], max_depth=3) tree = to_tree({"action": {"name": profile.action.name, "description": profile.action.description}}) assert "ai_message_lifecycle" in tree assert "├─" in tree or "└─" in tree 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 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 custom postfix .dsl writer + parser + tree / markdown / Mermaid output to_dsl / dump_dsl: write ActionProfile as custom postfix DSL (RPN) with length-prefixed lists and tagged records. Whitespace- tokenized; bare atoms unquoted; \"...\" only when needed; \"\\\\\" for line comments; nil for null. parse_dsl / _tokenize_dsl: round-trip the .dsl back to nested dict. Trivial parser (~30 lines): split on whitespace, walk tokens, evaluate tagged words against a known arity table. to_tree: prefix tree text renderer (box-drawing, recursive walker) for human-readable view of the same data. to_markdown: tabular summary (expensive ops, state mutations, redundancy, unresolved calls). to_mermaid: from the existing CallGraph.render_mermaid. 33 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_dsl(profile, str(out_dir / "actions" / f"{args.action}.dsl")) (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}.{{dsl,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_dsl(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 custom postfix `.dsl` + MD + MMD to //actions/.{dsl,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. 36 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.{dsl,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_dsl, COST_CLASS_WEIGHTS cg = build_call_graph() out = Path('docs/reports/code_path_audit/2026-06-07') (out).mkdir(parents=True, exist_ok=True) # to_dsl takes an ActionProfile. For the indexes, wrap each in a synthetic container. # The implementer adds a _to_dsl_dict() helper in src/code_path_audit.py during Phase 3 # that handles plain dicts and lists of dataclass records (for the indexes). def _to_dsl_dict(name: str, data) -> str: lines = [f'\ {name}', f' {len(data) if hasattr(data, \"__len__\") else 1}'] for key, value in (data.items() if isinstance(data, dict) else [(name, data)]): lines.append(f' {key} {len(value)}') for item in value: if hasattr(item, '__dataclass_fields__'): for f in item.__dataclass_fields__: lines.append(f' {getattr(item, f)!r}') lines.append(' {len(item.__dataclass_fields__)} list exp-op' if 'cost_class' in item.__dataclass_fields__ else ' {len(item.__dataclass_fields__)} list mut') lines.append(f' {len(value)} list') lines.append(' }') return chr(10).join(lines) (out / 'call_graph.dsl').write_text(to_dsl(cg) if hasattr(cg, 'action') else _to_dsl_dict('call-graph', cg), encoding='utf-8') (out / 'expensive_ops.dsl').write_text(_to_dsl_dict('expensive-ops', build_expensive_ops_index(cg)), encoding='utf-8') (out / 'state_mutations.dsl').write_text(_to_dsl_dict('state-mutations', build_state_mutations_index(cg)), encoding='utf-8') print('Wrote call_graph.dsl, expensive_ops.dsl, state_mutations.dsl') " ``` Expected: `Wrote call_graph.dsl, expensive_ops.dsl, state_mutations.dsl` (Note: the script above is illustrative. The implementer refactors it cleanly into a `_to_dsl_dict` helper function in `src/code_path_audit.py` during Phase 3; the script here just shows the intent.) - [ ] **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.dsl (full call graph of src/) - expensive_ops.dsl (per-function expensive-op index) - state_mutations.dsl (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: custom postfix `.dsl` data + markdown + Mermaid + prefix tree text 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). 36 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**, **36 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, pathlib, dataclasses, re). - **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`.