84 KiB
Plan: metadata_promotion_20260624 (EXHAUSTIVE TIER 3 EXECUTION CONTRACT)
Tier 1 exhaustive plan — 2026-06-25. This plan is the EXECUTABLE CONTRACT for Tier 2/Tier 3. Tier 2 reviews per phase; Tier 3 executes per task. No decisions remain for Tier 2/3 to make — every task has exact file:line refs, exact before/after code, exact test commands, and explicit rollback steps. If a Tier 3 encounters an unanticipated situation, they STOP and report to Tier 2 — do NOT improvise.
Scope summary: 12 phases (one per aggregate + collapsed-codepath audit + verification). 30 atomic tasks. ~36 atomic commits. Estimated 1 module-extensions to existing files + 12 new dataclasses added + 12 new test files (60+ tests) + 9 consumer files migrated (~213 access sites) + 1 styleguide clarification + 2 docs reports. NO day estimates.
Acceptance:
compute_effective_codepathsreturns< 1e+20(was 4.014e+22); 7 audit gates pass--strict; 10/11 batched test tiers PASS; new per-aggregate regression-guard tests pass.
0. Pre-flight (Tier 2 runs before Tier 3 starts)
These commands establish the baseline. Tier 2 captures the output and saves it as docs/reports/metadata_promotion_baseline_<date>.txt. Tier 3 refers to this baseline throughout.
# 0.1 Confirm the working tree is clean
git status --short
# Expect: no output (clean)
# 0.2 Confirm the spec+plan+metadata.json are committed
git log --oneline -1 -- conductor/tracks/metadata_promotion_20260624/
# Expect: 5 commits from the correction (spec + plan + metadata + styleguide + correction report)
# 0.3 Measure the baseline effective codepaths
uv run python -c "
import sys
sys.path.insert(0, 'scripts/code_path_audit')
sys.path.insert(0, 'src')
from code_path_audit import build_pcg
from code_path_audit_ssdl import count_branches_in_function
pcg = build_pcg('src').data
metadata_consumers = pcg.consumers.get('Metadata', [])
total = sum(2 ** count_branches_in_function(f, 'src') for f in metadata_consumers)
print(f'Baseline effective codepaths: {total:.3e}')
print(f'Metadata consumers: {len(metadata_consumers)}')
"
# Expect: 4.014e+22 ; 695 consumers
# 0.4 Confirm all 7 audit gates pass --strict (or note which are pre-existing failures)
uv run python scripts/audit_weak_types.py --strict
uv run python scripts/generate_type_registry.py --check
uv run python scripts/audit_main_thread_imports.py
uv run python scripts/audit_no_models_config_io.py
uv run python scripts/audit_code_path_audit_coverage.py --input-dir docs/reports/code_path_audit/latest --strict
uv run python scripts/audit_exception_handling.py --strict
uv run python scripts/audit_optional_in_3_files.py --strict
# Expect: all exit 0; note any failures as "pre-existing, not introduced by this track"
# 0.5 Confirm the baseline test suite is green (10/11 acceptable; RAG flake documented)
uv run python scripts/run_tests_batched.py
# Expect: 10/11 PASS
# 0.6 Capture baseline counts
git grep -nE "\.get\('[a-z_]+'," -- 'src/*.py' | wc -l
# Expect: 107
git grep -nE "\[[ ]*'[a-z_]+'[ ]*\]" -- 'src/*.py' | wc -l
# Expect: 106
If the baseline differs from these expected values, Tier 2 STOPS and reports. Do NOT proceed with a different baseline — the plan assumes these counts.
Phase 0: Add NEW per-aggregate dataclasses (no consumer migration yet)
Focus: Add the 11 NEW dataclasses (in their parent modules — per AGENTS.md "no new src/.py files" rule). No consumer migration in this phase. The existing dataclasses (Ticket, FileItem, ToolCall, ChatMessage, UsageStats, ContextPreset, MCPServerConfig) are REUSED UNCHANGED.
Acceptance: from src.type_aliases import CommsLogEntry, HistoryMessage, ToolDefinition, SessionInsights, DiscussionSettings, CustomSlice, MMAUsageStats, ProviderPayload, UIPanelConfig, PathInfo works; from src.rag_engine import RAGChunk works; constructors work with kwargs; to_dict() / from_dict() round-trip is lossless.
Task 0.1: Add 10 NEW dataclasses to src/type_aliases.py
WHERE: src/type_aliases.py (current 30 lines; will grow to ~200 lines)
HOW: manual-slop_edit_file with surgical old_string/new_string; OR write_file to overwrite (preferred for this scale of change)
Exact new_string for the file (replaces all current content):
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Callable, NamedTuple, TypeAlias
Metadata: TypeAlias = dict[str, Any]
@dataclass(frozen=True, slots=True)
class CommsLogEntry:
ts: str = ""
role: str = ""
kind: str = ""
direction: str = ""
model: str = "unknown"
source_tier: str = "main"
content: Any = None
error: str = ""
def to_dict(self) -> Metadata:
return {k: v for k, v in self.__dict__.items() if v not in (None, "", [], {}, 0, 0.0, False) or k in ("model",)}
@classmethod
def from_dict(cls, raw: Metadata) -> "CommsLogEntry":
valid = {f.name for f in field(cls) for _ in [None]} # noqa
return cls(**{k: v for k, v in raw.items() if k in {f.name for f in (__import__("dataclasses").fields(cls))}})
@dataclass(frozen=True, slots=True)
class HistoryMessage:
role: str = ""
content: Any = None
tool_calls: Any = None
tool_call_id: str = ""
name: str = ""
ts: str = ""
def to_dict(self) -> Metadata:
return {k: v for k, v in self.__dict__.items() if v not in (None, "", [], {}, 0, 0.0, False)}
@classmethod
def from_dict(cls, raw: Metadata) -> "HistoryMessage":
return cls(**{k: v for k, v in raw.items() if k in {f.name for f in (__import__("dataclasses").fields(cls))}})
@dataclass(frozen=True, slots=True)
class ToolDefinition:
name: str = ""
description: str = ""
parameters: Any = None
auto_start: bool = False
def to_dict(self) -> Metadata:
return {k: v for k, v in self.__dict__.items() if v not in (None, "", [], {}, 0, 0.0, False)}
@classmethod
def from_dict(cls, raw: Metadata) -> "ToolDefinition":
return cls(**{k: v for k, v in raw.items() if k in {f.name for f in (__import__("dataclasses").fields(cls))}})
@dataclass(frozen=True, slots=True)
class SessionInsights:
total_tokens: int = 0
call_count: int = 0
burn_rate: float = 0.0
session_cost: float = 0.0
completed_tickets: int = 0
efficiency: float = 0.0
def to_dict(self) -> Metadata:
return dict(self.__dict__)
@classmethod
def from_dict(cls, raw: Metadata) -> "SessionInsights":
return cls(**{k: v for k, v in raw.items() if k in {f.name for f in (__import__("dataclasses").fields(cls))}})
@dataclass(frozen=True, slots=True)
class DiscussionSettings:
temperature: float = 0.7
top_p: float = 1.0
max_output_tokens: int = 0
def to_dict(self) -> Metadata:
return dict(self.__dict__)
@classmethod
def from_dict(cls, raw: Metadata) -> "DiscussionSettings":
return cls(**{k: v for k, v in raw.items() if k in {f.name for f in (__import__("dataclasses").fields(cls))}})
@dataclass(frozen=True, slots=True)
class CustomSlice:
tag: str = ""
comment: str = ""
start_line: int = 0
end_line: int = 0
def to_dict(self) -> Metadata:
return dict(self.__dict__)
@classmethod
def from_dict(cls, raw: Metadata) -> "CustomSlice":
return cls(**{k: v for k, v in raw.items() if k in {f.name for f in (__import__("dataclasses").fields(cls))}})
@dataclass(frozen=True, slots=True)
class MMAUsageStats:
model: str = "unknown"
input: int = 0
output: int = 0
def to_dict(self) -> Metadata:
return {k: v for k, v in self.__dict__.items() if v not in (None, "", [], {}, 0, 0.0, False)}
@classmethod
def from_dict(cls, raw: Metadata) -> "MMAUsageStats":
return cls(**{k: v for k, v in raw.items() if k in {f.name for f in (__import__("dataclasses").fields(cls))}})
@dataclass(frozen=True, slots=True)
class ProviderPayload:
script: str = ""
args: Metadata = field(default_factory=dict)
output: str = ""
source_tier: str = "main"
def to_dict(self) -> Metadata:
return {k: v for k, v in self.__dict__.items() if v not in (None, "", [], {}, 0, 0.0, False)}
@classmethod
def from_dict(cls, raw: Metadata) -> "ProviderPayload":
return cls(**{k: v for k, v in raw.items() if k in {f.name for f in (__import__("dataclasses").fields(cls))}})
@dataclass(frozen=True, slots=True)
class UIPanelConfig:
separate_message_panel: bool = False
separate_response_panel: bool = False
separate_tool_calls_panel: bool = False
def to_dict(self) -> Metadata:
return dict(self.__dict__)
@classmethod
def from_dict(cls, raw: Metadata) -> "UIPanelConfig":
return cls(**{k: v for k, v in raw.items() if k in {f.name for f in (__import__("dataclasses").fields(cls))}})
@dataclass(frozen=True, slots=True)
class PathInfo:
logs_dir: str = ""
scripts_dir: str = ""
project_root: str = ""
def to_dict(self) -> Metadata:
return dict(self.__dict__)
@classmethod
def from_dict(cls, raw: Metadata) -> "PathInfo":
return cls(**{k: v for k, v in raw.items() if k in {f.name for f in (__import__("dataclasses").fields(cls))}})
FileItem: TypeAlias = "models.FileItem"
ToolCall: TypeAlias = "openai_schemas.ToolCall"
ChatMessage: TypeAlias = "openai_schemas.ChatMessage"
CommsLog: TypeAlias = list[CommsLogEntry]
History: TypeAlias = list[HistoryMessage]
FileItems: TypeAlias = list[FileItem]
CommsLogCallback: TypeAlias = Callable[[CommsLogEntry], None]
JsonPrimitive: TypeAlias = str | int | float | bool | None
JsonValue: TypeAlias = JsonPrimitive | list["JsonValue"] | dict[str, "JsonValue"]
class FileItemsDiff(NamedTuple):
refreshed: FileItems
changed: FileItems
Wait — the from_dict classmethods above use __import__("dataclasses").fields(cls) which is ugly. Use the cleaner pattern (already used at src/models.py:600 and src/openai_schemas.py:36-43):
Final new_string (the version Tier 3 actually writes):
from __future__ import annotations
from dataclasses import dataclass, field, fields
from typing import Any, Callable, NamedTuple, TypeAlias
Metadata: TypeAlias = dict[str, Any]
def _filter_known(raw: dict[str, Any], cls: type) -> dict[str, Any]:
valid = {f.name for f in fields(cls)}
return {k: v for k, v in raw.items() if k in valid}
@dataclass(frozen=True, slots=True)
class CommsLogEntry:
ts: str = ""
role: str = ""
kind: str = ""
direction: str = ""
model: str = "unknown"
source_tier: str = "main"
content: Any = None
error: str = ""
def to_dict(self) -> Metadata:
out: Metadata = {}
for k, v in self.__dict__.items():
if k == "model" or v not in (None, "", [], {}, 0, 0.0, False):
out[k] = v
return out
@classmethod
def from_dict(cls, raw: Metadata) -> "CommsLogEntry":
return cls(**_filter_known(raw, cls))
@dataclass(frozen=True, slots=True)
class HistoryMessage:
role: str = ""
content: Any = None
tool_calls: Any = None
tool_call_id: str = ""
name: str = ""
ts: str = ""
def to_dict(self) -> Metadata:
out: Metadata = {}
for k, v in self.__dict__.items():
if v not in (None, "", [], {}, 0, 0.0, False):
out[k] = v
return out
@classmethod
def from_dict(cls, raw: Metadata) -> "HistoryMessage":
return cls(**_filter_known(raw, cls))
@dataclass(frozen=True, slots=True)
class ToolDefinition:
name: str = ""
description: str = ""
parameters: Any = None
auto_start: bool = False
def to_dict(self) -> Metadata:
out: Metadata = {}
for k, v in self.__dict__.items():
if v not in (None, "", [], {}, 0, 0.0, False):
out[k] = v
return out
@classmethod
def from_dict(cls, raw: Metadata) -> "ToolDefinition":
return cls(**_filter_known(raw, cls))
@dataclass(frozen=True, slots=True)
class SessionInsights:
total_tokens: int = 0
call_count: int = 0
burn_rate: float = 0.0
session_cost: float = 0.0
completed_tickets: int = 0
efficiency: float = 0.0
def to_dict(self) -> Metadata:
return dict(self.__dict__)
@classmethod
def from_dict(cls, raw: Metadata) -> "SessionInsights":
return cls(**_filter_known(raw, cls))
@dataclass(frozen=True, slots=True)
class DiscussionSettings:
temperature: float = 0.7
top_p: float = 1.0
max_output_tokens: int = 0
def to_dict(self) -> Metadata:
return dict(self.__dict__)
@classmethod
def from_dict(cls, raw: Metadata) -> "DiscussionSettings":
return cls(**_filter_known(raw, cls))
@dataclass(frozen=True, slots=True)
class CustomSlice:
tag: str = ""
comment: str = ""
start_line: int = 0
end_line: int = 0
def to_dict(self) -> Metadata:
return dict(self.__dict__)
@classmethod
def from_dict(cls, raw: Metadata) -> "CustomSlice":
return cls(**_filter_known(raw, cls))
@dataclass(frozen=True, slots=True)
class MMAUsageStats:
model: str = "unknown"
input: int = 0
output: int = 0
def to_dict(self) -> Metadata:
out: Metadata = {}
for k, v in self.__dict__.items():
if v not in (None, "", [], {}, 0, 0.0, False):
out[k] = v
return out
@classmethod
def from_dict(cls, raw: Metadata) -> "MMAUsageStats":
return cls(**_filter_known(raw, cls))
@dataclass(frozen=True, slots=True)
class ProviderPayload:
script: str = ""
args: Metadata = field(default_factory=dict)
output: str = ""
source_tier: str = "main"
def to_dict(self) -> Metadata:
out: Metadata = {}
for k, v in self.__dict__.items():
if v not in (None, "", [], {}, 0, 0.0, False):
out[k] = v
return out
@classmethod
def from_dict(cls, raw: Metadata) -> "ProviderPayload":
return cls(**_filter_known(raw, cls))
@dataclass(frozen=True, slots=True)
class UIPanelConfig:
separate_message_panel: bool = False
separate_response_panel: bool = False
separate_tool_calls_panel: bool = False
def to_dict(self) -> Metadata:
return dict(self.__dict__)
@classmethod
def from_dict(cls, raw: Metadata) -> "UIPanelConfig":
return cls(**_filter_known(raw, cls))
@dataclass(frozen=True, slots=True)
class PathInfo:
logs_dir: str = ""
scripts_dir: str = ""
project_root: str = ""
def to_dict(self) -> Metadata:
return dict(self.__dict__)
@classmethod
def from_dict(cls, raw: Metadata) -> "PathInfo":
return cls(**_filter_known(raw, cls))
FileItem: TypeAlias = "models.FileItem"
ToolCall: TypeAlias = "openai_schemas.ToolCall"
ChatMessage: TypeAlias = "openai_schemas.ChatMessage"
CommsLog: TypeAlias = list[CommsLogEntry]
History: TypeAlias = list[HistoryMessage]
FileItems: TypeAlias = list[FileItem]
CommsLogCallback: TypeAlias = Callable[[CommsLogEntry], None]
JsonPrimitive: TypeAlias = str | int | float | bool | None
JsonValue: TypeAlias = JsonPrimitive | list["JsonValue"] | dict[str, "JsonValue"]
class FileItemsDiff(NamedTuple):
refreshed: FileItems
changed: FileItems
SAFETY (run after the edit):
uv run python -c "from src.type_aliases import CommsLogEntry, HistoryMessage, ToolDefinition, SessionInsights, DiscussionSettings, CustomSlice, MMAUsageStats, ProviderPayload, UIPanelConfig, PathInfo; print('OK')"
# Expect: OK
uv run python -c "from src.type_aliases import CommsLogEntry; e = CommsLogEntry(role='user', ts='2025-01-01'); print(e.role, e.ts, e.model)"
# Expect: user 2025-01-01 unknown
uv run python -c "from src.type_aliases import CommsLogEntry; e = CommsLogEntry.from_dict({'role': 'user', 'ts': '2025-01-01', 'unknown_field': 'x'}); print(e.role, e.ts)"
# Expect: user 2025-01-01 (unknown_field filtered)
uv run python -c "from src.type_aliases import CommsLogEntry; e = CommsLogEntry(role='user'); d = e.to_dict(); print(sorted(d.items()))"
# Expect: [('model', 'unknown'), ('role', 'user')]
uv run python -c "from src.type_aliases import Metadata; print(type(Metadata))"
# Expect: <class 'str'> # it's a TypeAlias, so the str representation
COMMIT: refactor(type_aliases): add 10 per-aggregate dataclasses (CommsLogEntry, HistoryMessage, ToolDefinition, SessionInsights, DiscussionSettings, CustomSlice, MMAUsageStats, ProviderPayload, UIPanelConfig, PathInfo)
GIT NOTE: Per-aggregate dataclasses added to src/type_aliases.py. Metadata: TypeAlias = dict[str, Any] UNCHANGED (the catch-all for collapsed codepaths). Existing dataclasses (FileItem, ToolCall, ChatMessage) re-exported as TypeAliases. No consumer migration yet.
ROLLBACK: git revert HEAD (one atomic commit). The 10 dataclasses are additive; no consumer code changes.
Task 0.2: Add RAGChunk dataclass to src/rag_engine.py
WHERE: src/rag_engine.py (insert after the existing class definitions; use manual-slop_py_add_def with anchor_type="bottom")
WHAT to add (exact text):
@dataclass(frozen=True, slots=True)
class RAGChunk:
document: str = ""
path: str = ""
score: float = 0.0
metadata: Metadata = field(default_factory=dict)
def to_dict(self) -> Metadata:
out: Metadata = {}
for k, v in self.__dict__.items():
if v not in (None, "", [], {}, 0.0, False):
out[k] = v
return out
@classmethod
def from_dict(cls, raw: Metadata) -> "RAGChunk":
valid = {f.name for f in fields(cls)}
return cls(**{k: v for k, v in raw.items() if k in valid})
Imports to add at top of src/rag_engine.py:
from dataclasses import dataclass, field, fields
from src.type_aliases import Metadata
HOW: manual-slop_py_add_def with anchor_type="bottom", new_content=<the above>. Then manual-slop_edit_file to add the imports (anchor on the existing import block).
SAFETY:
uv run python -c "from src.rag_engine import RAGChunk; c = RAGChunk(document='hi', path='/foo.py', score=0.95); print(c.document, c.path, c.score)"
# Expect: hi /foo.py 0.95
uv run python -c "from src.rag_engine import RAGChunk; c = RAGChunk.from_dict({'document': 'x', 'path': '/y', 'score': 0.5, 'extra': 'filtered'}); print(c.document)"
# Expect: x
uv run python scripts/audit_main_thread_imports.py
# Expect: exit 0 (verify the new imports don't break the main-thread-purity invariant)
COMMIT: feat(rag_engine): add RAGChunk dataclass
GIT NOTE: RAGChunk dataclass added to src/rag_engine.py. Per-aggregate type for RAG retrieval results. No consumer migration yet.
ROLLBACK: git revert HEAD.
Task 0.3: Add 2 NEW dataclasses to src/mcp_client.py (ToolDefinition proxy + ASTNode + SearchResult + MCPToolResult)
WHERE: src/mcp_client.py (insert at the bottom, before the final region markers; use manual-slop_py_add_def)
WHAT to add (exact text):
@dataclass(frozen=True, slots=True)
class ASTNode:
kind: str = ""
name: str = ""
indent: int = 0
start_line: int = 0
end_line: int = 0
full_path: str = ""
def to_dict(self) -> Metadata:
return dict(self.__dict__)
@classmethod
def from_dict(cls, raw: Metadata) -> "ASTNode":
valid = {f.name for f in fields(cls)}
return cls(**{k: v for k, v in raw.items() if k in valid})
@dataclass(frozen=True, slots=True)
class SearchResult:
title: str = ""
link: str = ""
snippet: str = ""
def to_dict(self) -> Metadata:
return dict(self.__dict__)
@classmethod
def from_dict(cls, raw: Metadata) -> "SearchResult":
valid = {f.name for f in fields(cls)}
return cls(**{k: v for k, v in raw.items() if k in valid})
@dataclass(frozen=True, slots=True)
class MCPToolResult:
content: tuple[Metadata, ...] = ()
tools: tuple[Metadata, ...] = ()
def to_dict(self) -> Metadata:
return {"content": list(self.content), "tools": list(self.tools)}
@classmethod
def from_dict(cls, raw: Metadata) -> "MCPToolResult":
return cls(content=tuple(raw.get("content", ())), tools=tuple(raw.get("tools", ())))
Imports to add (if not already present):
from dataclasses import dataclass, field, fields
Check the existing imports at the top of src/mcp_client.py; add only what's missing.
SAFETY:
uv run python -c "from src.mcp_client import ASTNode, SearchResult, MCPToolResult; n = ASTNode(kind='function', name='foo', indent=1, start_line=10, end_line=20, full_path='/foo.py'); print(n.kind, n.name, n.full_path)"
# Expect: function foo /foo.py
uv run python -c "from src.mcp_client import SearchResult; r = SearchResult(title='t', link='l', snippet='s'); print(r)"
# Expect: SearchResult(title='t', link='l', snippet='s')
uv run python scripts/audit_main_thread_imports.py
# Expect: exit 0
COMMIT: feat(mcp_client): add ASTNode, SearchResult, MCPToolResult dataclasses
GIT NOTE: Per-aggregate dataclasses added to src/mcp_client.py for AST traversal results, web search results, and MCP tool call results. No consumer migration yet.
ROLLBACK: git revert HEAD.
Task 0.4: Add 2 NEW dataclasses to src/performance_monitor.py and src/log_registry.py
WHERE:
src/performance_monitor.py— insert at the bottom:PerformanceMetricsdataclasssrc/log_registry.py— insert at the bottom:SessionInfoandSessionMetadatadataclasses
WHAT to add (exact text):
src/performance_monitor.py (at the bottom, before any final region markers):
@dataclass(frozen=True, slots=True)
class PerformanceMetrics:
fps: float = 0.0
frame_time_ms_avg: float = 0.0
def to_dict(self) -> Metadata:
return dict(self.__dict__)
@classmethod
def from_dict(cls, raw: Metadata) -> "PerformanceMetrics":
valid = {f.name for f in fields(cls)}
return cls(**{k: v for k, v in raw.items() if k in valid})
Imports to add if not present: from dataclasses import dataclass, field, fields; from src.type_aliases import Metadata.
src/log_registry.py (at the bottom):
@dataclass(frozen=True, slots=True)
class SessionInfo:
session_id: str = ""
path: str = ""
def to_dict(self) -> Metadata:
return dict(self.__dict__)
@classmethod
def from_dict(cls, raw: Metadata) -> "SessionInfo":
valid = {f.name for f in fields(cls)}
return cls(**{k: v for k, v in raw.items() if k in valid})
@dataclass(frozen=True, slots=True)
class SessionMetadata:
timestamp: str = ""
def to_dict(self) -> Metadata:
return dict(self.__dict__)
@classmethod
def from_dict(cls, raw: Metadata) -> "SessionMetadata":
valid = {f.name for f in fields(cls)}
return cls(**{k: v for k, v in raw.items() if k in valid})
Imports to add if not present: from dataclasses import dataclass, field, fields; from src.type_aliases import Metadata.
SAFETY:
uv run python -c "from src.performance_monitor import PerformanceMetrics; m = PerformanceMetrics(fps=60.0, frame_time_ms_avg=16.7); print(m)"
# Expect: PerformanceMetrics(fps=60.0, frame_time_ms_avg=16.7)
uv run python -c "from src.log_registry import SessionInfo, SessionMetadata; s = SessionInfo(session_id='abc', path='/x'); print(s)"
# Expect: SessionInfo(session_id='abc', path='/x')
COMMIT: feat(perf,log_registry): add PerformanceMetrics, SessionInfo, SessionMetadata dataclasses
GIT NOTE: Per-aggregate dataclasses added for performance telemetry and session metadata.
ROLLBACK: git revert HEAD.
Task 0.5: Complete ContextPreset schema in src/models.py
WHERE: src/models.py:932 (the ContextPreset class)
Read first (Tier 3 reads the current implementation):
git grep -n "class ContextPreset" src/models.py
Read the full class with manual-slop_get_file_slice (the slice tool).
Current state (per the data_structure_strengthening_20260606 spec §3.1): ContextPreset has name, files, screenshots minimum.
WHAT to add (extend the schema with all observed fields from src/gui_2.py:4181-4185,4333,4448):
Verify which fields are accessed:
preset.get('files', [])—src/gui_2.py:4184preset.get('screenshots', [])—src/gui_2.py:4185preset.name(string)len(preset.files),len(preset.screenshots)
The minimal schema is already correct (name, files, screenshots). The change is to add the @dataclass(frozen=True, slots=True) decorator (if not present) and ensure to_dict() / from_dict() round-trip is lossless.
Read the current class first with manual-slop_get_file_slice from line 932, length ~40 lines. Then determine if it's already a dataclass; if not, ADD the @dataclass(frozen=True, slots=True) decorator via manual-slop_edit_file. Preserve all existing fields and methods.
HOW: manual-slop_get_file_slice first, then manual-slop_edit_file to add the decorator if missing. Verify to_dict() / from_dict() exist; if not, add them using the canonical pattern from src/models.py:567 (FileItem.to_dict()).
SAFETY:
uv run python -c "from src.models import ContextPreset; cp = ContextPreset(name='test', files=[], screenshots=[]); print(cp.to_dict())"
# Expect: {'name': 'test', 'files': [], 'screenshots': []}
uv run python -c "from src.models import ContextPreset; cp = ContextPreset.from_dict({'name': 'test', 'files': [], 'screenshots': []}); print(cp)"
# Expect: ContextPreset(name='test', files=([],), screenshots=([],)) # may differ based on default_factory
uv run python -m pytest tests/test_context_presets_models.py -v
# Expect: all tests pass
COMMIT: refactor(models): complete ContextPreset schema (add @dataclass(frozen=True, slots=True) decorator if missing)
GIT NOTE: ContextPreset schema completed. The class is now a typed dataclass with to_dict() / from_dict() round-trip. No consumer migration yet.
ROLLBACK: git revert HEAD.
Task 0.6: Create 12 per-aggregate regression-guard test files
WHERE: NEW FILES in tests/:
tests/test_comms_log_entry.pytests/test_history_message.pytests/test_tool_definition.pytests/test_session_insights.pytests/test_discussion_settings.pytests/test_custom_slice.pytests/test_mma_usage_stats.pytests/test_provider_payload.pytests/test_ui_panel_config.pytests/test_path_info.pytests/test_rag_chunk.py(in addition to the 10 above)tests/test_metadata_dataclass_aux.py(for ASTNode, SearchResult, MCPToolResult, PerformanceMetrics, SessionInfo, SessionMetadata)
HOW: write_file per file. Each file has the SAME STRUCTURE (5 tests minimum per file):
# tests/test_comms_log_entry.py
from __future__ import annotations
import pytest
from src.type_aliases import CommsLogEntry
def test_default_constructor() -> None:
e = CommsLogEntry()
assert e.ts == ""
assert e.role == ""
assert e.model == "unknown"
assert e.source_tier == "main"
def test_constructor_with_kwargs() -> None:
e = CommsLogEntry(role="user", ts="2025-01-01", content="hi")
assert e.role == "user"
assert e.ts == "2025-01-01"
assert e.content == "hi"
def test_field_access_direct() -> None:
e = CommsLogEntry(role="user")
assert e.role == "user"
def test_frozen_raises() -> None:
import dataclasses
e = CommsLogEntry(role="user")
with pytest.raises(dataclasses.FrozenInstanceError):
e.role = "assistant"
def test_slots_no_dict() -> None:
e = CommsLogEntry()
with pytest.raises(AttributeError):
e.unknown_field = "x"
def test_to_dict_includes_non_empty() -> None:
e = CommsLogEntry(role="user", ts="2025-01-01")
d = e.to_dict()
assert d["role"] == "user"
assert d["ts"] == "2025-01-01"
assert "model" in d # model has default "unknown" — always included
def test_from_dict_filters_unknown() -> None:
e = CommsLogEntry.from_dict({"role": "user", "unknown_field": "x"})
assert e.role == "user"
assert e.ts == ""
Repeat the pattern for each of the 12 files, adapting imports and field names.
For tests/test_metadata_dataclass_aux.py (multi-class test file):
from __future__ import annotations
import pytest
from src.mcp_client import ASTNode, SearchResult, MCPToolResult
from src.performance_monitor import PerformanceMetrics
from src.log_registry import SessionInfo, SessionMetadata
def test_ast_node_constructor() -> None:
n = ASTNode(kind="function", name="foo", indent=1, start_line=10, end_line=20, full_path="/foo.py")
assert n.kind == "function"
assert n.full_path == "/foo.py"
def test_ast_node_to_from_dict_roundtrip() -> None:
n = ASTNode(kind="function", name="foo", indent=1, start_line=10, end_line=20, full_path="/foo.py")
d = n.to_dict()
n2 = ASTNode.from_dict(d)
assert n == n2
def test_search_result_constructor() -> None:
r = SearchResult(title="t", link="l", snippet="s")
assert r.title == "t"
def test_search_result_roundtrip() -> None:
r = SearchResult(title="t", link="l", snippet="s")
r2 = SearchResult.from_dict(r.to_dict())
assert r == r2
def test_mcp_tool_result_roundtrip() -> None:
tr = MCPToolResult(content=({"text": "hi"},), tools=({"name": "foo"},))
tr2 = MCPToolResult.from_dict(tr.to_dict())
assert tr == tr2
def test_performance_metrics_constructor() -> None:
m = PerformanceMetrics(fps=60.0, frame_time_ms_avg=16.7)
assert m.fps == 60.0
def test_performance_metrics_roundtrip() -> None:
m = PerformanceMetrics(fps=60.0, frame_time_ms_avg=16.7)
m2 = PerformanceMetrics.from_dict(m.to_dict())
assert m == m2
def test_session_info_constructor() -> None:
s = SessionInfo(session_id="abc", path="/x")
assert s.session_id == "abc"
def test_session_info_roundtrip() -> None:
s = SessionInfo(session_id="abc", path="/x")
s2 = SessionInfo.from_dict(s.to_dict())
assert s == s2
def test_session_metadata_constructor() -> None:
m = SessionMetadata(timestamp="2025-01-01T00:00:00")
assert m.timestamp == "2025-01-01T00:00:00"
def test_session_metadata_roundtrip() -> None:
m = SessionMetadata(timestamp="2025-01-01T00:00:00")
m2 = SessionMetadata.from_dict(m.to_dict())
assert m == m2
SAFETY:
uv run pytest tests/test_comms_log_entry.py tests/test_history_message.py tests/test_tool_definition.py tests/test_session_insights.py tests/test_discussion_settings.py tests/test_custom_slice.py tests/test_mma_usage_stats.py tests/test_provider_payload.py tests/test_ui_panel_config.py tests/test_path_info.py tests/test_rag_chunk.py tests/test_metadata_dataclass_aux.py -v
# Expect: all 12 files PASS; total 60+ tests
COMMIT: test(type_aliases): add per-aggregate dataclass regression-guard suite (60+ tests across 12 files)
GIT NOTE: 12 regression-guard test files added. The 11 NEW dataclasses are tested for: default constructor, kwargs constructor, field access, frozen, slots, to_dict/from_dict round-trip, from_dict filters unknown fields. Consumer migration is in subsequent phases; this commit only adds the tests.
ROLLBACK: git revert HEAD (no consumer code changed; tests are additive).
Task 0.7: Re-measure baseline (Phase 0 complete)
uv run python -c "
import sys
sys.path.insert(0, 'scripts/code_path_audit')
sys.path.insert(0, 'src')
from code_path_audit import build_pcg
from code_path_audit_ssdl import count_branches_in_function
pcg = build_pcg('src').data
metadata_consumers = pcg.consumers.get('Metadata', [])
total = sum(2 ** count_branches_in_function(f, 'src') for f in metadata_consumers)
print(f'Post-Phase-0 effective codepaths: {total:.3e}')
print(f'Metadata consumers: {len(metadata_consumers)}')
"
# Expect: ~4.014e+22 (no consumer migration yet; baseline unchanged)
Phase 0 introduces NO codepath reduction — it's purely design + tests. The reduction happens in Phases 1-10 as consumers migrate to direct field access.
End of Phase 0.
Phase 1: Migrate Ticket consumers (REUSED dataclass; remove legacy .get() method)
Focus: Ticket is already a dataclass at src/models.py:302 with 15 fields. The consumers currently use t.get('id', '') etc. via the legacy Ticket.get(key, default) method (line 348). After this phase, all consumers use direct field access (t.id, t.depends_on, t.manual_block) and the legacy get() method is REMOVED.
Acceptance: All 30+ t.get(...) and t['...'] access sites on Ticket consumers replaced with direct field access; legacy Ticket.get() method removed; all existing tests pass.
Task 1.1: Migrate src/gui_2.py Ticket access sites (read-only uses)
WHERE: src/gui_2.py:1366,1369,1387,1393,1399,1408,1418,1419,1427,1428,1436,1438,1439,1682,6852,6854,6860,6861,6870,7018,7022,7071,7096,7128,7156,7158,7162,7166,7171,7203,7204,7208,7215,7223,7233,7234,7248,7255,7256,7269,7270,7272,7294
Pattern A — t.get('id', '') → str(t.id):
Read the exact line via manual-slop_get_file_slice. Each occurrence uses manual-slop_edit_file with old_string and new_string.
Example for src/gui_2.py:1366:
- old_string:
id_to_idx = {str(t.get('id', '')): i for i, t in enumerate(new_tickets)} - new_string:
id_to_idx = {str(t.id): i for i, t in enumerate(new_tickets)}
Example for src/gui_2.py:1369:
- old_string:
deps = t.get('depends_on', []) - new_string:
deps = list(t.depends_on)(the dataclass field is a list; the call site uses it as a list, so wrap withlist()for type narrowness OR justt.depends_onif the downstream accepts the dataclass type)
Read the surrounding context with manual-slop_get_file_slice to determine the correct new_string for each line.
Pattern B — t.get('manual_block', False) → t.manual_block:
Example for src/gui_2.py:1428:
- old_string:
if t and t.get('manual_block', False): - new_string:
if t and t.manual_block:
Pattern C — t.get('status') → t.status:
Example for src/gui_2.py:1388,1394,1400,1410,1421,1429,1444:
- old_string:
if t: t['status'] = 'in_progress'(these are mutation sites; see Task 1.2)
Task 1.1 covers read-only sites only. Mutation sites are in Task 1.2.
PATTERN TABLE for Ticket fields (use manual-slop_edit_file per site):
| Site | old_string (access) | new_string (direct) |
|---|---|---|
gui_2.py:1366 |
str(t.get('id', '')) |
str(t.id) |
gui_2.py:1387 |
str(t.get('id', '')) |
str(t.id) |
gui_2.py:1393 |
str(t.get('id', '')) |
str(t.id) |
gui_2.py:1399 |
str(t.get('id', '')) |
str(t.id) |
gui_2.py:1408 |
str(ticket_id) (compare) |
str(ticket_id) (unchanged; the lookup key) |
gui_2.py:1418 |
for dep_id in t.get('depends_on', []): |
for dep_id in t.depends_on: |
gui_2.py:1419 |
str(x.get('id', '')) |
str(x.id) |
gui_2.py:1427 |
str(ticket_id) (compare) |
(unchanged) |
gui_2.py:1428 |
t.get('manual_block', False) |
t.manual_block |
gui_2.py:1436 |
t.get('status') == 'blocked' and not t.get('manual_block', False) |
t.status == 'blocked' and not t.manual_block |
gui_2.py:1438 |
for dep_id in t.get('depends_on', []): |
for dep_id in t.depends_on: |
gui_2.py:1439 |
str(x.get('id', '')) |
str(x.id) |
gui_2.py:1682 |
{'id': str(t.get('id', '')), 'depends_on': t.get('depends_on', [])} |
{'id': str(t.id), 'depends_on': list(t.depends_on)} |
gui_2.py:6852 |
str(t.get('id', '')) |
str(t.id) |
gui_2.py:6854 |
ticket.get('status', 'todo') |
ticket.status |
gui_2.py:6860 |
ticket.get('target_file', '') |
ticket.target_file or '' |
gui_2.py:6860 |
', '.join(ticket.get('depends_on', [])) |
', '.join(ticket.depends_on) |
gui_2.py:6861 |
ticket.get('persona_id', '') |
ticket.persona_id or '' |
gui_2.py:6870 |
str(t.get('id', '')) |
str(t.id) |
gui_2.py:7018 |
track.get('title', '') |
track.title (if Track is dataclass; if not, see Task 1.4) |
gui_2.py:7022 |
track.get('goal', '') |
track.goal (Track dataclass) |
gui_2.py:7071 |
str(t.get('id', '')) |
str(t.id) |
gui_2.py:7096 |
str(t.get('id', '')) |
str(t.id) |
gui_2.py:7128 |
t.get('priority', 'medium') |
t.priority |
gui_2.py:7156 |
t.get('status', 'todo') |
t.status |
gui_2.py:7158 |
t.get('status', 'todo') |
t.status |
gui_2.py:7162 |
t.get('description', '') |
t.description |
gui_2.py:7166 |
t.get('status', 'todo') |
t.status |
gui_2.py:7171 |
t.get('manual_block', False) |
t.manual_block |
gui_2.py:7203 |
str(t.get('id', '')) |
str(t.id) |
gui_2.py:7204 |
str(t.get('id', '')) |
str(t.id) |
gui_2.py:7208 |
str(t.get('id', '??')) |
str(t.id) if t.id else '??' |
gui_2.py:7215 |
t.get('status', 'todo') |
t.status |
gui_2.py:7223 |
t.get('target_file','') |
t.target_file or '' |
gui_2.py:7233 |
str(t.get('id', '??')) |
str(t.id) if t.id else '??' |
gui_2.py:7234 |
for dep in t.get('depends_on', []): |
for dep in t.depends_on: |
gui_2.py:7248 |
str(t.get('id', '')) |
str(t.id) |
gui_2.py:7255 |
str(t.get('id', '')) |
str(t.id) |
gui_2.py:7256 |
t.get('depends_on', []) |
t.depends_on |
gui_2.py:7269 |
str(t.get('id', '')) |
str(t.id) |
gui_2.py:7270 |
t.get('depends_on', []) |
t.depends_on |
gui_2.py:7294 |
t.get('id', '') |
t.id |
For Track (if not already a dataclass): Task 1.4 adds the @dataclass(frozen=True, slots=True) decorator to Track at src/models.py if it's not already a typed dataclass. Read it first; if it's a dict, skip Task 1.4 (Track is out of scope for Phase 1; the read-only sites in gui_2.py:7018-7024 will keep track.get('title', '') for now and become a follow-up track).
HOW: For each row, read the exact line context with manual-slop_get_file_slice (start_line=N-2, end_line=N+2), then use manual-slop_edit_file with the precise old_string and new_string. If the surrounding context has tabs vs spaces, preserve exactly.
SAFETY (run after EVERY 5 edits):
uv run python -m pytest tests/test_ticket_queue.py tests/test_per_ticket_model.py tests/test_manual_block.py tests/test_tiered_aggregation.py -x
# Expect: all tests pass (use -x to stop on first failure; revert immediately if any fail)
SAFETY (run after the full phase):
git grep -nE "\.get\('id'," -- 'src/gui_2.py' | grep -v "str(t.id)" | wc -l
# Expect: 0 (all .get('id', ...) sites in gui_2.py are migrated)
git grep -nE "t\.get\(" -- 'src/gui_2.py' | wc -l
# Expect: 0 (no t.get() calls remaining)
git grep -nE "\.get\('depends_on'," -- 'src/gui_2.py' | wc -l
# Expect: 0
COMMIT: refactor(gui_2): migrate Ticket access sites to direct field access (~40 sites)
GIT NOTE: Migrated ~40 Ticket access sites in src/gui_2.py from t.get('key', default) / t['key'] to direct field access (t.id, t.depends_on, t.manual_block, etc.). Verified by ticket test files. Ticket dataclass REUSED unchanged from src/models.py:302.
ROLLBACK: git revert HEAD. The mutations in Task 1.2 are in a separate commit; this commit is read-only.
Task 1.2: Migrate src/gui_2.py Ticket mutation sites
WHERE: src/gui_2.py:1388,1394,1400,1410,1411,1412,1421,1429,1430,1431,1444,6867,7137,7148,7151,7272
PATTERN: Mutations of frozen=True dataclass fields use dataclasses.replace():
# BEFORE:
if t: t['status'] = 'in_progress'
# AFTER:
import dataclasses
if t:
from src.type_aliases import dataclasses_replace_compat # if needed
t = dataclasses.replace(t, status='in_progress')
OR, since the mutation is inside an if-check on a next(...) result, replace the lookup with a direct construction:
# BEFORE:
t = next((t for t in app.active_tickets if str(t.get('id', '')) == tid), None)
if t: t['status'] = 'in_progress'
# AFTER (cleanest):
from src.mma_tickets import update_ticket_status # helper, or inline
match = next((i for i, t in enumerate(app.active_tickets) if str(t.id) == tid), None)
if match is not None:
app.active_tickets[match] = dataclasses.replace(app.active_tickets[match], status='in_progress')
EXACT migration table:
| Site | old_string | new_string |
|---|---|---|
gui_2.py:1388 |
if t: t['status'] = 'in_progress' |
if t: app.active_tickets[app.active_tickets.index(t)] = dataclasses.replace(t, status='in_progress') |
gui_2.py:1394 |
if t: t['status'] = 'completed' |
if t: app.active_tickets[app.active_tickets.index(t)] = dataclasses.replace(t, status='completed') |
gui_2.py:1400 |
if t: t['status'] = 'blocked' |
if t: app.active_tickets[app.active_tickets.index(t)] = dataclasses.replace(t, status='blocked') |
gui_2.py:1410-1412 |
t['status'] = 'blocked'; t['manual_block'] = True; t['blocked_reason'] = '[MANUAL] User blocked' |
use dataclasses.replace(t, status='blocked', manual_block=True, blocked_reason='[MANUAL] User blocked') |
gui_2.py:1421 |
t['status'] = 'blocked' |
dataclasses.replace(t, status='blocked') |
gui_2.py:1429-1431 |
t['status'] = 'todo'; t['manual_block'] = False; t['blocked_reason'] = None |
dataclasses.replace(t, status='todo', manual_block=False, blocked_reason=None) |
gui_2.py:1444 |
t['status'] = 'todo' |
dataclasses.replace(t, status='todo') |
gui_2.py:6858 |
ticket['priority'] = p_opt |
use dataclasses.replace(ticket, priority=p_opt) |
gui_2.py:6866 |
ticket['persona_id'] = None |
use dataclasses.replace(ticket, persona_id=None) |
gui_2.py:6867 |
ticket['status'] = 'done' |
use dataclasses.replace(ticket, status='done') |
gui_2.py:7137 |
t['priority'] = p_opt |
dataclasses.replace(t, priority=p_opt) |
gui_2.py:7148 |
t['model_override'] = None |
dataclasses.replace(t, model_override=None) |
gui_2.py:7151 |
t['model_override'] = model |
dataclasses.replace(t, model_override=model) |
gui_2.py:7272 |
t['depends_on'] = [dep for dep in deps if abs(hash(dep + "_" + tid)) != lid_val] |
dataclasses.replace(t, depends_on=[dep for dep in deps if abs(hash(dep + "_" + tid)) != lid_val]) |
Add the import at the top of src/gui_2.py (if not already present):
import dataclasses
SAFETY:
uv run python -m pytest tests/test_ticket_queue.py tests/test_per_ticket_model.py tests/test_manual_block.py tests/test_tiered_aggregation.py -x
# Expect: all pass
uv run python -m pytest tests/test_mma_step_mode_sim.py tests/test_spawn_interception_v2.py -x
# Expect: all pass (these tests exercise the mutation paths)
COMMIT: refactor(gui_2): migrate Ticket mutation sites to dataclasses.replace (~14 sites)
GIT NOTE: Migrated ~14 Ticket mutation sites in src/gui_2.py from t['key'] = value to dataclasses.replace(t, key=value). The frozen=True invariant is preserved.
ROLLBACK: git revert HEAD.
Task 1.3: Migrate src/app_controller.py Ticket access + mutation sites
WHERE: src/app_controller.py:4810,4820,4868 (mutation sites); also any read-only sites — run git grep -nE "t\.get\(|\['status'\]" -- 'src/app_controller.py' to enumerate.
PATTERN: Same as Task 1.1 + 1.2.
EXACT table (from grep):
app_controller.py:4810:t['status'] = 'todo'→ mutation via replaceapp_controller.py:4820:t['status'] = 'skipped'→ mutation via replaceapp_controller.py:4868:t['status'] = 'in_progress'→ mutation via replace
(Read the exact context with manual-slop_get_file_slice first to get the surrounding for loop / if conditions.)
SAFETY:
uv run python -m pytest tests/test_ticket_queue.py tests/test_conductor_engine_v2.py tests/test_phase6_engine.py -x
# Expect: all pass
COMMIT: refactor(app_controller): migrate Ticket mutation sites to dataclasses.replace (~3 sites)
GIT NOTE: Migrated ~3 Ticket mutation sites in src/app_controller.py.
ROLLBACK: git revert HEAD.
Task 1.4: Migrate src/conductor_tech_lead.py Ticket access sites
WHERE: src/conductor_tech_lead.py:125 (ticket_map = {t['id']: t for t in tickets})
EXACT migration:
- old_string:
ticket_map = {t['id']: t for t in tickets} - new_string:
ticket_map = {t.id: t for t in tickets}
Read the surrounding context with manual-slop_get_file_slice first; the t here is a Ticket dataclass (passed in from src/dag_engine.py:TrackDAG.get_executable_tickets which returns List[Ticket]).
SAFETY:
uv run python -m pytest tests/test_conductor_engine_v2.py tests/test_track_get_executable_tickets_complex.py -x
# Expect: all pass
COMMIT: refactor(conductor_tech_lead): migrate Ticket access site to direct field access
GIT NOTE: Migrated 1 Ticket access site.
ROLLBACK: git revert HEAD.
Task 1.5: Remove the legacy Ticket.get(key, default) method
WHERE: src/models.py:348 (the def get(self, key: str, default: Any = None) -> Any method)
Read first with manual-slop_get_file_slice to verify the exact line range. The method spans lines 348-363 approximately.
HOW: manual-slop_py_remove_def with name="Ticket.get". The tool will identify and remove the method body.
SAFETY (CRITICAL — run BEFORE removing):
git grep -nE "\.get\('id'," -- 'src/*.py' | wc -l
# Expect: 0 (no .get('id', default) calls remain on Ticket consumers)
git grep -nE "t\.get\(" -- 'src/*.py' | wc -l
# Expect: 0
uv run python -m pytest tests/test_ticket_queue.py tests/test_per_ticket_model.py tests/test_manual_block.py tests/test_tiered_aggregation.py tests/test_conductor_engine_v2.py -x
# Expect: all pass
If any t.get(...) call remains, REVERT this task and the previous tasks; investigate which consumer was missed.
After removal:
uv run python -m pytest tests/ -x --timeout=60 -q
# Expect: all pass (or the documented pre-existing failures only)
COMMIT: refactor(models): remove legacy Ticket.get() method (direct field access is now the only path)
GIT NOTE: Legacy compat method removed. All consumers migrated in Tasks 1.1-1.4. The Ticket dataclass is now purely typed; no dynamic-key fallback.
ROLLBACK: git revert HEAD. Note: if this revert is needed, the consumers in Tasks 1.1-1.4 will break (they use t.id not t.get('id', '')). Revert those commits too (in reverse order: 1.4, 1.3, 1.2, 1.1).
Task 1.6: Re-measure + verify Phase 1
# Effective codepaths after Phase 1
uv run python -c "
import sys
sys.path.insert(0, 'scripts/code_path_audit')
sys.path.insert(0, 'src')
from code_path_audit import build_pcg
from code_path_audit_ssdl import count_branches_in_function
pcg = build_pcg('src').data
metadata_consumers = pcg.consumers.get('Metadata', [])
total = sum(2 ** count_branches_in_function(f, 'src') for f in metadata_consumers)
print(f'Post-Phase-1 effective codepaths: {total:.3e}')
print(f'Metadata consumers: {len(metadata_consumers)}')
"
# Expect: < 1e+22 (significant drop from Ticket migrations; baseline 4.014e+22)
# VC4 partial: no .get('id', default) calls on Ticket consumers
git grep -nE "\.get\('id'," -- 'src/*.py' | wc -l
# Expect: 0
# All existing tests pass
uv run python scripts/run_tests_batched.py
# Expect: 10/11 PASS (RAG flake acceptable)
End of Phase 1.
Phase 2: Migrate FileItem consumers (REUSED dataclass)
Focus: FileItem is already a dataclass at src/models.py:533. Migrate consumers from fi.get('path', 'attachment') to fi.path, f['path'] to f.path.
Task 2.1: Migrate src/ai_client.py FileItem consumers
WHERE: src/ai_client.py:2565,2807,2898
EXACT migrations (all the same pattern):
- old_string:
fi.get('path', 'attachment') - new_string:
fi.path or 'attachment'
(Read each line with manual-slop_get_file_slice first to get the exact context; the fi variable name might differ at each site — it could be item, fi, file_item. Use the actual variable name in old_string.)
SAFETY:
uv run python -m pytest tests/test_ai_client.py tests/test_file_item_model.py -x --timeout=60
# Expect: all pass
COMMIT: refactor(ai_client): migrate FileItem access sites to direct field access (~3 sites)
GIT NOTE: Migrated 3 FileItem access sites in src/ai_client.py. FileItem dataclass REUSED unchanged from src/models.py:533.
ROLLBACK: git revert HEAD.
Task 2.2: Migrate src/app_controller.py FileItem consumer
WHERE: src/app_controller.py:3508 (file_paths = [f['path'] for f in file_items])
EXACT migration:
- old_string:
file_paths = [f['path'] for f in file_items] - new_string:
file_paths = [f.path for f in file_items]
SAFETY:
uv run python -m pytest tests/test_file_item_model.py tests/test_app_controller.py -x --timeout=60
# Expect: all pass
COMMIT: refactor(app_controller): migrate FileItem access site to direct field access
GIT NOTE: Migrated 1 FileItem access site.
ROLLBACK: git revert HEAD.
Task 2.3: Re-measure + verify Phase 2
# VC4 partial: no .get('path', ...) calls on FileItem consumers
git grep -nE "\.get\('path'," -- 'src/ai_client.py' | wc -l
# Expect: 0 (the 3 sites in src/ai_client.py are migrated; ProjectConfig's self.project.get('paths', {}) doesn't match this regex)
git grep -nE "fi\.get\(|f\['path'\]" -- 'src/*.py' | wc -l
# Expect: 0
End of Phase 2.
Phase 3: Migrate CommsLogEntry consumers (NEW dataclass from Phase 0)
Focus: CommsLogEntry was added in Phase 0 (src/type_aliases.py). Now wire it into the consumers.
Task 3.1: Migrate src/app_controller.py CommsLogEntry consumers
WHERE: src/app_controller.py:2277,2302,2310 (and any other sites in this file — search with git grep -nE "entry\.get\(" -- 'src/app_controller.py')
EXACT migrations:
app_controller.py:2277:'source_tier': entry.get('source_tier', 'main')→'source_tier': entry.source_tier(read full line context first; theentryis a CommsLogEntry dataclass)app_controller.py:2302:tier = entry.get('source_tier', 'main')→tier = entry.source_tierapp_controller.py:2310:'model': entry.get('model', 'unknown')→'model': entry.model
For each site, the entry variable MUST be a CommsLogEntry instance. If any site reads entry.get('model', 'unknown') where entry is actually a different aggregate (e.g., a UsageStats-like dict), STOP and report to Tier 2.
SAFETY:
uv run python -m pytest tests/test_session_logger_optimization.py tests/test_session_logger_reset.py tests/test_session_logging.py tests/test_logging_e2e.py tests/test_comms_log_entry.py -x --timeout=60
# Expect: all pass
COMMIT: refactor(app_controller): migrate CommsLogEntry access sites to direct field access
GIT NOTE: Migrated ~3 CommsLogEntry access sites in src/app_controller.py.
ROLLBACK: git revert HEAD.
Task 3.2: Migrate src/gui_2.py CommsLogEntry consumer
WHERE: src/gui_2.py:5803 (imgui.text_colored(C_SUB(), f"[{entry.get('source_tier', 'main')}]"))
EXACT migration:
- old_string:
f"[{entry.get('source_tier', 'main')}]" - new_string:
f"[{entry.source_tier}]"
SAFETY:
uv run python -m pytest tests/test_comms_log_entry.py tests/test_logging_e2e.py -x --timeout=60
# Expect: all pass
COMMIT: refactor(gui_2): migrate CommsLogEntry access site to direct field access
GIT NOTE: Migrated 1 CommsLogEntry access site in src/gui_2.py.
ROLLBACK: git revert HEAD.
Task 3.3: Re-measure + verify Phase 3
git grep -nE "entry\.get\('source_tier'," -- 'src/*.py' | wc -l
# Expect: 0
End of Phase 3.
Phase 4: Migrate HistoryMessage consumers (NEW dataclass)
Focus: HistoryMessage is the UI-layer discussion message (distinct from openai_schemas.ChatMessage which is provider-side).
Task 4.1: Migrate src/synthesis_formatter.py HistoryMessage consumers
WHERE: src/synthesis_formatter.py:24,37
EXACT migrations:
synthesis_formatter.py:24:f"{msg.get('role', 'unknown')}: {msg.get('content', '')}"→f"{msg.role}: {msg.content or ''}"(ormsg.contentif the field is always set)synthesis_formatter.py:37: same pattern
Read the full context first; the msg variable is a HistoryMessage instance.
SAFETY:
uv run python -m pytest tests/test_synthesis_formatter.py tests/test_history_message.py -x --timeout=60
# Expect: all pass (search for the actual test file name; may be tests/test_synthesis*.py)
COMMIT: refactor(synthesis_formatter): migrate HistoryMessage access sites to direct field access
GIT NOTE: Migrated 2 HistoryMessage access sites in src/synthesis_formatter.py.
ROLLBACK: git revert HEAD.
Task 4.2: Re-measure + verify Phase 4
git grep -nE "msg\.get\('role'," -- 'src/*.py' | wc -l
# Expect: 0
End of Phase 4.
Phase 5: Wire ChatMessage into per-vendor send paths
Focus: ChatMessage is already in src/openai_schemas.py:48. The per-vendor send paths (_send_anthropic, _send_deepseek, etc.) currently use the per-vendor history modules (provider_state.get_history(...)). Wire ChatMessage into the message construction.
Task 5.1: Migrate _send_anthropic and _send_deepseek (~9 sites)
WHERE: src/ai_client.py (the _send_anthropic and _send_deepseek methods)
Read first with manual-slop_get_file_slice to find the exact construction sites. Each provider builds a list of messages in a specific format.
HOW: The migration is provider-specific. Each provider's send method has a for msg in history: messages.append({...}) block. Replace the dict-construction with ChatMessage(role=msg.role, content=msg.content, ...).
EXACT pattern (for _send_anthropic):
# BEFORE:
for msg in anthropic_history:
if msg.get("role") == "user":
messages.append({"role": "user", "content": msg.get("content", "")})
# AFTER:
for msg in anthropic_history:
cm = ChatMessage.from_dict(msg) if isinstance(msg, dict) else msg
if cm.role == "user":
messages.append(cm.to_dict())
(Read each provider's send method first; the exact pattern depends on the provider's message schema.)
For each of the 8 send methods (_send_anthropic, _send_deepseek, _send_gemini, _send_gemini_cli, _send_minimax, _send_qwen, _send_llama, _send_grok):
- Read the method with
manual-slop_get_file_slice. - Identify the per-message dict-construction sites.
- Replace with
ChatMessageuse.
SAFETY:
uv run python -m pytest tests/test_ai_client.py tests/test_anthropic_provider.py tests/test_deepseek_provider.py tests/test_openai_schemas.py -x --timeout=120
# Expect: all pass
COMMIT (5.1, 5.2, 5.3): 3 atomic commits, one per provider pair
refactor(ai_client): wire ChatMessage into _send_anthropic and _send_deepseekrefactor(ai_client): wire ChatMessage into _send_gemini and _send_gemini_clirefactor(ai_client): wire ChatMessage into _send_minimax, _send_qwen, _send_llama, _send_grok
GIT NOTE: Wired ChatMessage (existing in src/openai_schemas.py:48) into the per-vendor send paths. The dataclass was already created; this phase wires it into the message construction.
ROLLBACK: git revert HEAD (one atomic commit at a time).
Task 5.4: Re-measure + verify Phase 5
git grep -nE "msg\.get\('role'," -- 'src/ai_client.py' | wc -l
# Expect: 0 (or only collapsed-codepath sites documented in Phase 11)
End of Phase 5.
Phase 6: Wire UsageStats into per-call usage aggregation
Task 6.1: Migrate src/app_controller.py:2299-2309 UsageStats access sites
WHERE: src/app_controller.py:2299-2309
Read first with manual-slop_get_file_slice.
EXACT migrations:
app_controller.py:2304:new_mma_usage[tier]['input'] += u.get('input_tokens', 0) or 0→new_mma_usage[tier] = dataclasses.replace(new_mma_usage[tier], input=new_mma_usage[tier].input + (u.input_tokens if hasattr(u, 'input_tokens') else u.get('input_tokens', 0)))(verifyuis a UsageStats instance first; if it's still a dict, this site is collapsed-codepath and stays)- Same pattern for
app_controller.py:2305,2308,2309
IMPORTANT: If u (or usage) is a dict (e.g., loaded from JSON), the migration is via UsageStats.from_dict(u). If it's already a dataclass instance, use direct attribute access.
SAFETY:
uv run python -m pytest tests/test_token_usage.py tests/test_usage_analytics_popout_sim.py tests/test_openai_schemas.py -x --timeout=60
# Expect: all pass
COMMIT: refactor(app_controller): wire UsageStats into per-call usage aggregation (~4 sites)
GIT NOTE: Wired UsageStats (existing in src/openai_schemas.py:68) into the per-call usage aggregation in src/app_controller.py.
ROLLBACK: git revert HEAD.
Task 6.2: Re-measure + verify Phase 6
git grep -nE "u\.get\('input_tokens'," -- 'src/app_controller.py' | wc -l
# Expect: 0
End of Phase 6.
Phase 7: Wire ToolCall into tool loop section
Task 7.1: Migrate src/ai_client.py tool loop section
WHERE: src/ai_client.py (the _dispatch_tool and tool loop methods)
Read first with manual-slop_get_file_slice. The migration is mechanical: replace tc.get('id'), tc.get('function', {}).get('name'), tc.get('function', {}).get('arguments') with tc.id, tc.function.name, tc.function.arguments.
EXACT pattern:
# BEFORE:
for tc in response.tool_calls:
tool_call_id = tc.get('id', '')
function_name = tc.get('function', {}).get('name', '')
arguments_str = tc.get('function', {}).get('arguments', '')
# AFTER:
for tc in response.tool_calls:
tool_call_id = tc.id
function_name = tc.function.name
arguments_str = tc.function.arguments
SAFETY:
uv run python -m pytest tests/test_ai_client.py tests/test_openai_schemas.py -x --timeout=60
# Expect: all pass
COMMIT: refactor(ai_client): wire ToolCall into tool loop section (~56 sites)
GIT NOTE: Wired ToolCall (existing in src/openai_schemas.py:32) into the tool loop in src/ai_client.py.
ROLLBACK: git revert HEAD.
Task 7.2: Verify src/mcp_client.py tool loop
WHERE: src/mcp_client.py:1707-1714 (the result['tools'] and result['content'] sites)
EXACT migrations:
mcp_client.py:1707:for t in result['tools']:→for t in result.tools:(after converting result toMCPToolResult.from_dict(result)if it's still a dict)mcp_client.py:1708:self.tools[t['name']] = t→self.tools[t.name] = tmcp_client.py:1714:return '\n'.join([c.get('text', '') for c in result['content'] if c.get('type') == 'text'])→return '\n'.join([c.get('text', '') for c in result.content if c.get('type') == 'text'])(thecontentis a tuple of dicts, not a list ofMCPToolResult; leave the innerc.getcalls as-is sincecis still aMetadatadict)
SAFETY:
uv run python -m pytest tests/test_mcp_client.py tests/test_metadata_dataclass_aux.py -x --timeout=60
# Expect: all pass
COMMIT: refactor(mcp_client): wire MCPToolResult into tool loop section (~3 sites)
GIT NOTE: Wired MCPToolResult (added in Phase 0) into the tool loop in src/mcp_client.py.
ROLLBACK: git revert HEAD.
End of Phase 7.
Phase 8: Migrate ToolDefinition consumers (NEW dataclass)
Task 8.1: Migrate src/mcp_client.py:1970 and src/gui_2.py:5876,5878
EXACT migrations:
mcp_client.py:1970:'description': tinfo.get('description', '')→'description': tinfo.description(aftertinfo = ToolDefinition.from_dict(...)if needed)gui_2.py:5876:imgui.text(tinfo.get('server', 'unknown'))→imgui.text(tinfo.server)(butToolDefinitiondoesn't haveserver; this is a different aggregate — likely aToolInfodict from a separate source. IfToolDefinition.from_dictdoesn't have aserverfield, STOP and report to Tier 2)gui_2.py:5878:imgui.text(tinfo.get('description', ''))→imgui.text(tinfo.description)
SAFETY:
uv run python -m pytest tests/test_mcp_client.py tests/test_tool_definition.py -x --timeout=60
# Expect: all pass
COMMIT: refactor(mcp_client,gui_2): migrate ToolDefinition access sites to direct field access
GIT NOTE: Migrated ~3 ToolDefinition access sites.
ROLLBACK: git revert HEAD.
End of Phase 8.
Phase 9: Migrate RAGChunk consumers (NEW dataclass)
Task 9.1: Migrate src/aggregate.py, src/ai_client.py, src/app_controller.py RAGChunk consumers
EXACT migrations:
aggregate.py:3259:chunk.get('document', '')→chunk.document(afterchunk = RAGChunk.from_dict(...)if chunk is a dict)app_controller.py:251: same patternapp_controller.py:4162: same patternai_client.py:3259: same pattern
Read each line with manual-slop_get_file_slice first to determine if chunk is already a dict or dataclass.
SAFETY:
uv run python -m pytest tests/test_rag_engine.py tests/test_aggregate.py tests/test_rag_chunk.py -x --timeout=120
# Expect: all pass
COMMIT: refactor(rag_engine,aggregate,app_controller,ai_client): migrate RAGChunk access sites to direct field access (~4 sites)
GIT NOTE: Migrated ~4 RAGChunk access sites across 4 files.
ROLLBACK: git revert HEAD.
End of Phase 9.
Phase 10: Migrate small-batch aggregates (8 aggregates)
Focus: SessionInsights, DiscussionSettings, CustomSlice, MMAUsageStats, ProviderPayload, UIPanelConfig, PathInfo, ToolDefinition. Batched because each has few sites.
Task 10.1: Migrate src/gui_2.py small-batch consumers
EXACT migrations (in src/gui_2.py):
| Site | Aggregate | old_string | new_string |
|---|---|---|---|
2199 |
MMAUsageStats | model = stats.get('model', 'unknown') |
model = stats.model (after stats = MMAUsageStats.from_dict(...) if needed) |
2200 |
MMAUsageStats | in_t = stats.get('input', 0) |
in_t = stats.input |
2201 |
MMAUsageStats | out_t = stats.get('output', 0) |
out_t = stats.output |
2216 |
MMAUsageStats | stats.get('model', '') |
stats.model |
3535 |
DiscussionSettings | entry.get('temperature', 0.7) |
entry.temperature (after entry = DiscussionSettings.from_dict(...)) |
4048 |
CustomSlice | slc.get('tag', '') |
slc.tag (after slc = CustomSlice.from_dict(...)) |
4054 |
CustomSlice | slc.get('comment', '') |
slc.comment |
4090 |
CustomSlice | slc.get('tag') == 'auto-ast' |
slc.tag == 'auto-ast' |
4269 |
FileStats (NEW) | stats.get('lines', 0), AST: {stats.get('ast_elements', 0)} |
stats.lines, AST: {stats.ast_elements} (FileStats is a NEW dataclass in Phase 10; if not added, keep as dict) |
4926-4931 |
SessionInsights | insights.get('total_tokens', 0), etc. |
insights.total_tokens, etc. (after insights = SessionInsights.from_dict(...)) |
5876 |
ToolDefinition | tinfo.get('server', 'unknown') |
(NOT a ToolDefinition field; report to Tier 2 if server is not in ToolDefinition) |
5878 |
ToolDefinition | tinfo.get('description', '') |
tinfo.description |
5953 |
CustomSlice | slc.get('tag', '') |
slc.tag |
5959 |
CustomSlice | slc.get('comment', '') |
slc.comment |
5980,5981 |
CustomSlice | slc.get('tag') == 'auto-ast', etc. |
slc.tag == 'auto-ast', etc. |
6610 |
MMAUsageStats | u.get('model','unknown'), u.get('input',0), u.get('output',0) |
u.model, u.input, u.output (after u = MMAUsageStats.from_dict(u) if u is a dict) |
6785-6787 |
MMAUsageStats | stats.get('model', 'unknown'), etc. |
stats.model, etc. |
At each site, FIRST verify the aggregate type. If the variable is already a dataclass instance, use direct field access. If it's still a dict[str, Any], call .from_dict() first OR classify as collapsed-codepath (in which case keep .get()).
SAFETY:
uv run python -m pytest tests/test_session_insights.py tests/test_discussion_settings.py tests/test_custom_slice.py tests/test_mma_usage_stats.py tests/test_provider_payload.py tests/test_ui_panel_config.py tests/test_path_info.py -x --timeout=60
# Expect: all pass
COMMIT (10.1): refactor(gui_2): migrate small-batch aggregates (SessionInsights, DiscussionSettings, CustomSlice, MMAUsageStats, ToolDefinition) to direct field access (~25 sites)
GIT NOTE: Migrated ~25 small-aggregate access sites in src/gui_2.py.
ROLLBACK: git revert HEAD.
Task 10.2: Migrate src/app_controller.py small-batch consumers
EXACT migrations (in src/app_controller.py):
| Site | Aggregate | old_string | new_string |
|---|---|---|---|
2068 |
UIPanelConfig | gui_cfg.get('separate_message_panel', False) |
gui_cfg.separate_message_panel |
2069 |
UIPanelConfig | gui_cfg.get('separate_response_panel', False) |
gui_cfg.separate_response_panel |
2070 |
UIPanelConfig | gui_cfg.get('separate_tool_calls_panel', False) |
gui_cfg.separate_tool_calls_panel |
2257,2258 |
MMAUsageStats | new_mma_usage[t]['input'] = 0, etc. |
new_mma_usage[t] = dataclasses.replace(new_mma_usage[t], input=0) (verify new_mma_usage is a dict[str, MMAUsageStats]; if it's still a dict[str, dict], leave as collapsed-codepath) |
2274,2287 |
ProviderPayload | payload.get('script'), payload.get('args', {}), payload.get('output', payload.get('content', '')) |
payload.script, payload.args, payload.output |
2304,2305 |
MMAUsageStats | new_mma_usage[tier]['input'] += u.get('input_tokens', 0) or 0, etc. |
mutation via replace (per Phase 6 pattern) |
Subscript sites (app_controller.py:1974,1978,1984,1985):
| Site | Aggregate | old_string | new_string |
|---|---|---|---|
1974 |
PathInfo | lpath = Path(proj_paths['logs_dir']) |
lpath = Path(proj_paths.logs_dir) (verify proj_paths is PathInfo after from_dict call) |
1978 |
PathInfo | spath = Path(proj_paths['scripts_dir']) |
spath = Path(proj_paths.scripts_dir) |
1984 |
PathInfo | path_info['logs_dir']['path'] |
path_info.logs_dir.path (if path_info.logs_dir is a PathInfo nested; otherwise leave) |
1985 |
PathInfo | path_info['scripts_dir']['path'] |
path_info.scripts_dir.path |
SAFETY:
uv run python -m pytest tests/test_ui_panel_config.py tests/test_provider_payload.py tests/test_path_info.py tests/test_app_controller.py -x --timeout=60
# Expect: all pass
COMMIT (10.2): refactor(app_controller): migrate ProviderPayload, UIPanelConfig, PathInfo, MMAUsageStats to direct field access (~10 sites)
GIT NOTE: Migrated ~10 small-aggregate access sites in src/app_controller.py.
ROLLBACK: git revert HEAD.
Task 10.3: Migrate src/multi_agent_conductor.py:638 and other small-batch sites
EXACT migration:
multi_agent_conductor.py:638:response_payload['stream_id']→response_payload.stream_id(after convertingresponse_payloadto a typed dataclass OR keeping as collapsed-codepath)
If response_payload is a dict from JSON, classify as collapsed-codepath and keep .get().
SAFETY:
uv run python -m pytest tests/test_multi_agent_conductor.py -x --timeout=60
# Expect: all pass
COMMIT (10.3): refactor(multi_agent_conductor): migrate ProviderPayload stream_id access to direct field access (if applicable; collapsed-codepath otherwise)
GIT NOTE: Migrated 1 site IF applicable; otherwise classified as collapsed-codepath.
ROLLBACK: git revert HEAD.
Task 10.4: Re-measure + verify Phase 10
git grep -nE "insights\.get\(|stats\.get\('model',|slc\.get\('tag',|slc\.get\('comment',|payload\.get\('script',|gui_cfg\.get\('separate_" -- 'src/*.py' | wc -l
# Expect: 0 (all migrated)
End of Phase 10.
Phase 11: Metadata collapsed-codepath audit (FR6)
Focus: Every remaining .get('key', default) and ['key'] site is classified as either (a) "promoted to per-aggregate dataclass → migrated" or (b) "collapsed codepath → keeps Metadata with documented justification."
Task 11.1: Audit and document remaining collapsed-codepath sites
Run the audit:
git grep -nE "\.get\('[a-z_]+'," -- 'src/*.py' > /tmp/remaining_get_sites.txt
wc -l /tmp/remaining_get_sites.txt
# Expect: < 30 (was 107; should be ~20 collapsed-codepath sites)
git grep -nE "\[[ ]*'[a-z_]+'[ ]*\]" -- 'src/*.py' > /tmp/remaining_subscript_sites.txt
wc -l /tmp/remaining_subscript_sites.txt
# Expect: ~80 (most of the 106 are genuinely dict access for collapsed codepaths)
For each remaining .get() site, classify and document:
| File:line | Current access | Classification | Justification |
|---|---|---|---|
app_controller.py:1972 |
self.project.get('paths', {}) |
collapsed | manual_slop.toml project config; shape unknown |
app_controller.py:2016 |
self.project.get('conductor', {}).get('dir', 'conductor') |
collapsed | TOML config |
app_controller.py:2033 |
self.project.get('project', {}).get('mcp_config_path') |
collapsed | TOML config |
gui_2.py:820 |
self.controller.project.get('context_presets', {}) |
collapsed | TOML config |
gui_2.py:4181 |
app.controller.project.get('context_presets', {}) |
collapsed | TOML config |
gui_2.py:4333 |
same | collapsed | TOML config |
gui_2.py:4448 |
app.controller.project.get('context_presets', {}).get(cp_name) |
collapsed | TOML config |
gui_2.py:5036 |
app.project.get('discussion', {}).get('discussions', {}) |
collapsed | TOML config (DiscussionStore) |
gui_2.py:5046,5047 |
same | collapsed | TOML config |
gui_2.py:5200,5217,5238 |
same | collapsed | TOML config |
synthesis_formatter.py:24,37 |
(already migrated in Phase 4) | n/a | n/a |
paths.py:262 |
data.get('conductor', {}).get('dir') |
collapsed | config.toml parsing; schema is opaque at this layer |
app_controller.py:2178 |
item['time'] |
collapsed | collated timeline item; aggregate is CollatedItem (NOT in scope; would require new dataclass) |
app_controller.py:2257,2258 |
new_mma_usage[t]['input'] = 0 |
collapsed | dict mutation; promotion would require a new MMAUsageMap type |
app_controller.py:2290,2294,2295 |
paired_tools[tid]['result'] |
collapsed | dict mutation; promotion would require a new PairedTools type |
app_controller.py:2299 |
u = payload['usage'] |
collapsed | JSON-deserialized payload |
app_controller.py:3508 |
(already migrated in Phase 2) | n/a | n/a |
gui_2.py:355-387 |
self.controller._predefined_callbacks['save_context_preset'] |
collapsed | handler-map; INTENTIONALLY a dict (it's the dispatch table, not data) |
gui_2.py:1388-1431 |
(Ticket mutations, already migrated in Phase 1.2) | n/a | n/a |
gui_2.py:2148-2150 |
usage['input_tokens'], etc. |
collapsed | UsageStats dict (promotion to dataclass requires from_dict at the JSON boundary) |
gui_2.py:3992-3995,4079,4080,4082,4085 |
node['indent'], node['kind'], etc. |
collapsed | AST node dict from tree-sitter; promotion requires new ASTNode use at the boundary (NOT done in Phase 0) |
gui_2.py:4033,4047,4053,4055 |
slice_data['tag'] = ..., etc. |
collapsed | CustomSlice mutation; promotion requires a typed list |
gui_2.py:4090,4091 |
slc.get('tag') == 'auto-ast' |
collapsed | CustomSlice in FileItem.custom_slices (list of dicts; promotion requires a typed list) |
gui_2.py:5921,5952,5958,5960,5980,5981 |
same pattern | collapsed | same |
gui_2.py:6318-6320 |
app.shader_uniforms['crt'] |
collapsed | shader uniform dict; NOT a sub-aggregate |
gui_2.py:6623 |
track_stats['percentage'] |
collapsed | TrackStats dict |
gui_2.py:7018,7020,7022,7024 |
track.get('title', ''), etc. |
collapsed | Track dict (the Track dataclass exists but these sites use track as a dict) |
gui_2.py:7020,7024 |
track['title'] = new_t, etc. |
collapsed | Track dict mutation |
log_pruner.py:53,54 |
session_info['session_id'], etc. |
collapsed | SessionInfo dict |
log_registry.py:174-179 |
new_session_data['start_time'], etc. |
collapsed | session mutation |
mcp_client.py:1045 |
r['title'], r['link'], r['snippet'] |
collapsed | SearchResult dict |
mcp_client.py:1707,1708 |
result['tools'], t['name'] |
collapsed | MCPToolResult dict (not yet converted to dataclass) |
mcp_client.py:1714 |
c.get('text', '') |
collapsed | content list of dicts |
models.py:976-978,989,991 |
data.get('command'), etc. |
collapsed | MCPServerConfig.from_dict (already a dataclass; these are the dict-input to the constructor) |
multi_agent_conductor.py:638 |
response_payload['stream_id'] |
collapsed | MMA response payload dict |
performance_monitor.py:27,28 |
metrics['fps'], metrics['frame_time_ms_avg'] |
collapsed | PerformanceMetrics dict |
project_manager.py:456 |
project_dict['discussion']['discussions'] |
collapsed | ProjectManager TOML dict |
synthesis_formatter.py:24,37 |
msg.get('role', 'unknown') |
collapsed (if msg is dict) or promoted (if dataclass) |
verify at the site |
app_controller.py:2178 |
item['time'] |
collapsed | collated timeline |
Write the classification as a doc:
cat > /tmp/collapsed_codepath_classification.md << 'EOF'
# Collapsed-codepath classification (Phase 11, FR6)
The following `.get('key', default)` and `['key']` sites REMAIN after Phases 0-10.
Each is classified as "collapsed codepath" with a documented justification.
## Why these are NOT promoted to per-aggregate dataclasses
Each site reads from a source where the shape is genuinely unknown at type level
(TOML config, JSON wire payload, polymorphic log entry, handler-map dispatch table).
The `Metadata: TypeAlias = dict[str, Any]` catch-all is the correct type here.
## Per-site classification
| File:line | Aggregate | Justification |
|---|---|---|
| (full table here) |
EOF
COMMIT: docs(audit): classify remaining .get() sites as collapsed-codepath (FR6)
GIT NOTE: Per-site classification of the remaining .get('key', default) and ['key'] sites. Each site is documented as "collapsed codepath" with a justification (TOML config, JSON wire, polymorphic log, handler-map). The Metadata: TypeAlias = dict[str, Any] catch-all is preserved for these sites.
ROLLBACK: git revert HEAD (no code changed; documentation only).
Task 11.2: Verify VC8 (no regression in audit gates)
uv run python scripts/audit_weak_types.py --strict
uv run python scripts/generate_type_registry.py --check
uv run python scripts/audit_main_thread_imports.py
uv run python scripts/audit_no_models_config_io.py
uv run python scripts/audit_code_path_audit_coverage.py --input-dir docs/reports/code_path_audit/latest --strict
uv run python scripts/audit_exception_handling.py --strict
uv run python scripts/audit_optional_in_3_files.py --strict
# Expect: all exit 0 (or only pre-existing failures documented in Phase 0)
End of Phase 11.
Phase 12: Verification + end-of-track (1 task, 3 commits)
Task 12.1: Run all 10 VCs + write TRACK_COMPLETION report
Run all VCs:
# VC1: Metadata unchanged
git grep "^Metadata:" src/type_aliases.py
# Expect: Metadata: TypeAlias = dict[str, Any]
# VC2: Each new dataclass is its OWN @dataclass(frozen=True, slots=True)
git grep -A 1 "^class CommsLogEntry\|^class HistoryMessage\|^class ToolDefinition\|^class RAGChunk\|^class SessionInsights\|^class DiscussionSettings\|^class CustomSlice\|^class MMAUsageStats\|^class ProviderPayload\|^class UIPanelConfig\|^class PathInfo" src/type_aliases.py src/rag_engine.py src/mcp_client.py src/performance_monitor.py src/log_registry.py
# Expect: each followed by @dataclass(frozen=True, slots=True)
# VC3: Existing dataclasses reused
git grep "class Ticket\|class FileItem\|class ToolCall\|class ChatMessage\|class UsageStats\|class ContextPreset\|class MCPServerConfig" src/models.py src/openai_schemas.py
# Expect: all exist
# VC4: 107 .get('key', ...) sites on known aggregates replaced
git grep -E "\.get\('[a-z_]+'," HEAD -- 'src/*.py' | wc -l
# Expect: < 30 (only collapsed-codepath sites from Phase 11)
# VC5: 106 ['key'] subscript sites on known aggregates replaced
git grep -E "\[[ ]*'[a-z_]+'[ ]*\]" HEAD -- 'src/*.py' | wc -l
# Expect: ~80 (only collapsed-codepath sites from Phase 11)
# VC6: 60+ tests pass
uv run pytest tests/test_comms_log_entry.py tests/test_history_message.py tests/test_tool_definition.py tests/test_rag_chunk.py tests/test_session_insights.py tests/test_discussion_settings.py tests/test_custom_slice.py tests/test_mma_usage_stats.py tests/test_provider_payload.py tests/test_ui_panel_config.py tests/test_path_info.py tests/test_metadata_dataclass_aux.py -v
# Expect: all pass (60+ tests across 12 files)
# VC7: Effective codepaths drops by >= 2 orders of magnitude
uv run python -c "
import sys
sys.path.insert(0, 'scripts/code_path_audit')
sys.path.insert(0, 'src')
from code_path_audit import build_pcg
from code_path_audit_ssdl import count_branches_in_function
pcg = build_pcg('src').data
metadata_consumers = pcg.consumers.get('Metadata', [])
total = sum(2 ** count_branches_in_function(f, 'src') for f in metadata_consumers)
print(f'Final effective codepaths: {total:.3e} (baseline 4.014e+22)')
"
# Expect: < 1e+20
# VC8: All 7 audit gates pass (re-run from Phase 11.2)
# Expect: all exit 0
# VC9: 10/11 batched test tiers PASS
uv run python scripts/run_tests_batched.py
# Expect: 10/11 PASS (RAG flake acceptable)
Write the TRACK_COMPLETION report at docs/reports/TRACK_COMPLETION_metadata_promotion_20260624.md:
cat > docs/reports/TRACK_COMPLETION_metadata_promotion_20260624.md << 'EOF'
# TRACK COMPLETION: metadata_promotion_20260624
**Date:** $(date -u +%Y-%m-%d)
**Track ID:** metadata_promotion_20260624
**Status:** SHIPPED
**Author:** Tier 2 (executed the Tier 1 exhaustive plan)
## Summary
Promoted 11 NEW per-aggregate `@dataclass(frozen=True, slots=True)` classes:
- CommsLogEntry, HistoryMessage, ToolDefinition, SessionInsights, DiscussionSettings, CustomSlice, MMAUsageStats, ProviderPayload, UIPanelConfig, PathInfo (in src/type_aliases.py)
- RAGChunk (in src/rag_engine.py)
- ASTNode, SearchResult, MCPToolResult (in src/mcp_client.py)
- PerformanceMetrics (in src/performance_monitor.py)
- SessionInfo, SessionMetadata (in src/log_registry.py)
Reused 8 EXISTING dataclasses unchanged: Ticket, FileItem, ToolCall, ChatMessage, UsageStats, ContextPreset, MCPServerConfig (in src/models.py and src/openai_schemas.py).
Migrated ~213 access sites across 9 consumer files from `.get('key', default)` / `['key']` to direct field access.
`Metadata: TypeAlias = dict[str, Any]` UNCHANGED — preserved as the catch-all for collapsed codepaths (TOML config, JSON wire, polymorphic log, handler-map).
## Verification results
| VC | Criterion | Result |
|---|---|---|
| VC1 | Metadata unchanged | PASS |
| VC2 | Each new dataclass is its OWN @dataclass | PASS |
| VC3 | Existing dataclasses reused | PASS |
| VC4 | 107 .get() sites on known aggregates replaced | PASS (< 30 remain; all classified as collapsed) |
| VC5 | 106 ['key'] subscript sites on known aggregates replaced | PASS (~80 remain; all classified as collapsed) |
| VC6 | 60+ regression-guard tests pass | PASS |
| VC7 | Effective codepaths < 1e+20 | PASS (was 4.014e+22) |
| VC8 | All 7 audit gates pass --strict | PASS |
| VC9 | 10/11 batched tiers PASS | PASS |
| VC10 | TRACK_COMPLETION written | PASS |
## Effective codepaths metric
Baseline: 4.014e+22
Final: $(<number>)e+<exp>
Drop: $(<number>) orders of magnitude
## Per-phase progress
[Phase-by-phase table here]
## Collapsed-codepath classification
[Reference to /tmp/collapsed_codepath_classification.md]
## Pre-existing failures
[Any pre-existing test failures documented at track start]
## Lessons learned
[Per-phase observations]
EOF
COMMIT (3 commits):
conductor(state): metadata_promotion_20260624 SHIPPED(updatesstate.tomltostatus = "completed",current_phase = "complete", all phasescompleted)docs(reports): TRACK_COMPLETION_metadata_promotion_20260624(the new report)conductor(tracks): update metadata_promotion_20260624 row(updatesconductor/tracks.md)
End of Phase 12. Track SHIPPED.
Tier 3 hard rules (DO NOT VIOLATE)
- Do NOT use
git restore,git checkout --, orgit reset— banned per AGENTS.md. If you need to revert, usegit revert <commit_sha>(one atomic commit per revert). - Do NOT use the native
edittool on Python files — it destroys 1-space indentation. Usemanual-slop_edit_file,manual-slop_py_update_definition,manual-slop_py_add_def, ormanual-slop_set_file_slice. - Do NOT add comments to source code — banned per AGENTS.md. Documentation lives in
/docs. - Do NOT create new
src/<thing>.pyfiles — banned per AGENTS.md hard rule. Helpers go in the parent module. - Do NOT skip a failing test with
@pytest.mark.skip— fix the bug instead. If you can't, report to Tier 2. - Do NOT batch commits — one atomic commit per task. Per-task commits enable precise rollback.
- Do NOT improvise decisions not in the plan — if the plan doesn't cover your situation, STOP and report to Tier 2.
- Do NOT exceed 5 nesting levels — extract to functions if you hit the limit.
- Do NOT modify
src/code_path_audit*.py— the audit infrastructure is correct. - Do NOT promote
Metadata: TypeAlias = dict[str, Any]itself — it's preserved as the catch-all. - NO-OP CLASSIFICATION BAN (added 2026-06-25 after a Tier 2 failure) — Do NOT classify any planned task as "no-op" without explicit user authorization in the same session. If a planned migration site exists in the plan's Task table, MIGRATE it. The plan is the contract; FR2 is a rule in the spec; the plan's tasks are what you execute. If FR2 and the plan conflict at a specific site:
- Read the site's variable type via
git grep -B 5 -A 5 <site>to determine if it's a dict or a dataclass instance. - If it's a dataclass instance: migrate to direct field access per the plan.
- If it's a dict at a genuinely collapsed codepath (TOML config, JSON wire, handler-map): migrate anyway by inserting
entry = <Aggregate>.from_dict(entry)(or equivalent) BEFORE the access site, OR classify the SPECIFIC site as collapsed-codepath and document in the per-site commit message (do NOT classify the whole phase). - If you can't decide: STOP. Write a 1-sentence question. Wait for the user. Do NOT invent a reconciliation.
- Read the site's variable type via
- METRIC REGRESSION REVERT (added 2026-06-25) — After every consumer-migration phase (Phases 1-10), run:
If the metric did NOT decrease vs. the previous measurement,
uv run python -c " import sys sys.path.insert(0, 'scripts/code_path_audit') sys.path.insert(0, 'src') from code_path_audit import build_pcg from code_path_audit_ssdl import count_branches_in_function pcg = build_pcg('src').data metadata_consumers = pcg.consumers.get('Metadata', []) total = sum(2 ** count_branches_in_function(f, 'src') for f in metadata_consumers) print(f'Post-Phase-N effective codepaths: {total:.3e}') "git revert <phase_commit_sha>IMMEDIATELY. Do NOT add a followup task to "investigate". Do NOT rationalize in the commit message or TRACK_COMPLETION. The 4.014e+22 → < 1e+20 drop is the success criterion; if it's not happening, the implementation is wrong; revert and report.
Per-phase Tier 2 review checklist
Before approving each phase, Tier 2 verifies:
- All tasks in the phase have commits.
- All test files for the new dataclasses exist and pass.
- The pre-phase git grep counts decreased by the expected amount (e.g., Phase 1 should remove ~50
.get('id', default)sites). - The audit gates (
audit_weak_types.py --strict,audit_main_thread_imports.py, etc.) still pass. - The batched test suite (
scripts/run_tests_batched.py) still passes 10/11 tiers. - The effective codepaths metric decreased (or held steady for design-only phases). If metric did NOT decrease after a consumer-migration phase (1-10), REVERT that phase commit immediately per Hard Rule #12. Do NOT approve.
If any check fails, Tier 2 REVERTS the phase commit and reports to the user.
Anti-pattern guard (per AGENTS.md)
If you observe any of these patterns in your own work, STOP and re-read AGENTS.md:
- The Deduction Loop: running a test 4+ times in one investigation. STOP after 2 failures.
- The Report-Instead-of-Fix Pattern: writing a 200-line status report instead of fixing.
- The Scope-Creep Track-Doc Pattern: writing a 5-phase spec for a 1-line fix.
- The Inherited-Cruft Pattern: trying to "fix" a broken file from a previous agent.
- No Diagnostic Noise in Production:
sys.stderr.writelines insrc/*.pyare technical debt. - The "I Am Not Going To Attempt Another Fix" Surrender: only surrender after 5-step protocol.
- The Verbose-Commit-Message Pattern: commit messages > 15 lines are reports.
- The Isolated-Pass Verification Fallacy: verifying in isolation but not in batch.
See also
conductor/tracks/metadata_promotion_20260624/spec.md— the corrected spec (rewritten 2026-06-25)conductor/tracks/metadata_promotion_20260624/metadata.json— the corrected metadataconductor/code_styleguides/type_aliases.md§2.5 — the new "per-aggregate dataclass" ruledocs/reports/PLANNING_CORRECTION_metadata_promotion_20260625.md— the planning-correction rationaleconductor/code_styleguides/data_oriented_design.md— canonical DOD referenceconductor/code_styleguides/error_handling.md—Result[T]conventionconductor/code_styleguides/python.md— Python style (1-space indent, CRLF, no comments)conductor/workflow.md— task workflow + commit disciplinesrc/openai_schemas.py— canonical per-aggregate dataclass patternsrc/models.py:533—FileItemcanonical in-module dataclass patternsrc/models.py:302—Ticketcanonical dataclass with legacy.get()removal exampledocs/reports/SSDL_CAMPAIGN_ABORTED_20260624.md— the post-mortem that established the type-dispatch thesis