manual_slop/conductor/tracks/code_path_audit_20260607/plan.md

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.

Timing (added 2026-06-08). This plan should not be executed until all 4 foundational tracks are shipped:

1. qwen_llama_grok_integration_20260606
2. data_oriented_error_handling_20260606
3. data_structure_strengthening_20260606
4. mcp_architecture_refactor_20260606

The 4 tracks will significantly reshape src/ai_client.py, src/mcp_client.py, src/app_controller.py, and src/type_aliases.py. Running this audit on the pre-refactor src/ would produce a report that's stale on day 1. The Tier 2 Tech Lead should verify the 4-tracks baseline (all marked [x] in conductor/tracks.md) before starting Phase 1.

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:

# 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

New-Item -ItemType File -Path src/code_path_audit.py -Force | Out-Null

• Step 0.3: Create empty tests/test_code_path_audit.py

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):

"""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

"""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:

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:

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:

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:

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

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

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." <commit_hash>

• 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:

@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:

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:

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), "<module>")
  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:

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:

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

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:

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:

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:

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:

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:

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:

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:

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 "<none>",
  max_depth=max_depth,
 )

• Step 3.3.4: Add tests for tree + markdown + mermaid

Append to tests/test_code_path_audit.py:

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 "<none>" in mmd

Task 3.4: Commit Phase 3

• Step 3.4.1: Stage and commit

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:

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:

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:

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:

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

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 <name> [--depth N]
[--output-dir DIR] [--date YYYY-MM-DD]. Writes custom postfix `.dsl` + MD + MMD
to <output-dir>/<date>/actions/<name>.{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

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

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):

# 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

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:

- [x] **Track: Code Path & Data Pipeline Audit** `[checkpoint: <last_commit_sha>]`
   *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 <last_commit_sha> with the SHA from the report commit in Phase 5.

• Step 6.2: Commit the tracks.md update

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.