Orchestrata Ultimate › Plans › codemod-engine-briefs › BRIEF-W2

BRIEF-W2: Architect + Reviewer (Qwen Max + Sonnet, Parallel)

Source: plans/codemod-engine-briefs/BRIEF-W2.md at 8a2cb42ad · last changed 2026-07-27 · first committed 2026-07-27 · 1057 lines.

Prose pending.

Plan text

BRIEF-W2: Architect + Reviewer (Qwen Max + Sonnet, Parallel)

Wave Overview

Wave: W2 (Architect + Reviewer)
Agents:

ARCHITECT (Qwen Max): passes.py (5 built-in transform passes), analysis/*.py (7 analysis modules)
REVIEWER (Sonnet): review.py (40 in-memory review checks), observability/*.py (5 observer classes)
Duration: ~20 minutes
Dependencies: W1 (workspace, CIG, txn)
Parallel Execution: Yes (ARCHITECT and REVIEWER work in parallel)

Objective

Implement the transform passes, analysis modules, review checks, and observability infrastructure:

ARCHITECT: Composable transform passes that mutate the CST forest, and read-only analysis passes that compute code intelligence metrics from the CIG
REVIEWER: 40 in-memory review checks organized into 7 categories, and 5 observer classes for structured observability

Scope

ARCHITECT Agent (Qwen Max)

Files to Implement

src/orchestrata_ultimate/refactor/codemod/passes.py (~600 LOC)
- TransformPass abstract base class (already skeleton from W0)
- ReplaceStringLiteralPass: Exact and regex modes with keep-pattern guard
- AddImportPass: Adds import statements to target modules
- RemoveUnusedImportPass: Identifies and removes unreferenced imports via CIG
- DocstringSweepPass: Vocabulary replacement in docstrings AND comments
- RenameSymbolPass: Consumes WorkspaceEditPlan from SPEC-109, applies LSP edits as LibCST mutations
src/orchestrata_ultimate/refactor/codemod/analysis/__init__.py (~50 LOC)
- Public API exports for all analysis passes
src/orchestrata_ultimate/refactor/codemod/analysis/blast_radius.py (~150 LOC)
- BlastRadiusAnalyzer class
- BFS blast radius computation over CALLS + INHERITS + IMPORTS edges
- Returns BlastRadiusResult dataclass with affected_nodes, affected_files, impact_score
src/orchestrata_ultimate/refactor/codemod/analysis/breaking_changes.py (~200 LOC)
- BreakingChangeDetector class
- Signature diff analysis (removed/added/reordered parameters, changed return types)
- Downstream consumer detection via CALLS + INHERITS edges
- Returns BreakingChangeResult dataclass with broken_consumers, incompatibilities
src/orchestrata_ultimate/refactor/codemod/analysis/complexity.py (~180 LOC)
- ComplexityAnalyzer class
- Cyclomatic complexity (count branching nodes: If/For/While/With/Try/ExceptHandler/BooleanOperation/IfExp/comp)
- Nesting depth (recursive DFS counting compound blocks)
- Cognitive complexity (accounts for nesting increments)
- Returns ComplexityResult dataclass with cyclomatic, nesting, cognitive per function/method
src/orchestrata_ultimate/refactor/codemod/analysis/coverage.py (~150 LOC)
- TestCoverageMapper class
- Test-to-source mapping via TESTED_BY edges
- Name heuristic (strip test_ prefix) for unmapped functions
- Returns CoverageResult dataclass with function_to_tests mapping
src/orchestrata_ultimate/refactor/codemod/analysis/abstraction.py (~120 LOC)
- AbstractionAnalyzer class
- Abstraction level (ratio of Protocol/ABC to concrete classes)
- Returns AbstractionResult dataclass with abstract_count, concrete_count, ratio
src/orchestrata_ultimate/refactor/codemod/analysis/coupling.py (~150 LOC)
- CouplingAnalyzer class
- Efferent coupling (outgoing dependencies)
- Afferent coupling (incoming dependencies)
- Instability metric (efferent / (efferent + afferent))
- Returns CouplingResult dataclass with efferent, afferent, instability
src/orchestrata_ultimate/refactor/codemod/analysis/dead_code.py (~180 LOC)
- DeadCodeDetector class
- Unreachable functions/classes via CIG reachability analysis
- Functions with no CALLS edges and no TESTED_BY edges
- Classes with no INHERITS edges and no instantiation
- Returns DeadCodeResult dataclass with dead_functions, dead_classes

Files to Test

tests/refactor/codemod/test_transform_passes.py (~400 LOC)
- 8 BDD scenarios from transform_passes.feature
- Test ReplaceStringLiteralPass (exact, regex, keep-pattern)
- Test AddImportPass (correct placement, group ordering)
- Test RemoveUnusedImportPass (CIG-based detection)
- Test DocstringSweepPass (docstrings AND comments)
- Test RenameSymbolPass (SPEC-109 interop)
- Test pass composition (N passes, one ChangeSet)
- Test pass cannot write to disk (AST grep enforcement)
tests/refactor/codemod/test_analysis_passes.py (~500 LOC)
- 13 BDD scenarios from analysis_passes.feature
- Test BlastRadiusAnalyzer (BFS, impact score)
- Test BreakingChangeDetector (removed param, changed return type)
- Test ComplexityAnalyzer (cyclomatic, nesting, cognitive)
- Test TestCoverageMapper (TESTED_BY edges, name heuristic)
- Test AbstractionAnalyzer (Protocol/ABC ratio)
- Test CouplingAnalyzer (efferent/afferent/instability)
- Test DeadCodeDetector (unreachable functions/classes)
- Test analysis passes are read-only (no mutations)
- Test analysis results have str and to_dict()

REVIEWER Agent (Sonnet)

Files to Implement

src/orchestrata_ultimate/refactor/codemod/review.py (~800 LOC)
- ReviewPass abstract base class (already skeleton from W0)
- InMemoryReviewEngine class (executes 40 checks in 3 phases)
- Complexity checks (4):
  - CyclomaticComplexityCheck: Count branching nodes (max 10)
  - NestingDepthCheck: Recursive DFS (max 4)
  - FunctionLengthCheck: Line count excluding docstring (max 40)
  - ParameterCountCheck: Max 5 excluding self/cls
- Style checks (6):
  - LineLengthCheck: Max 88 chars, URLs exempted
  - TrailingWhitespaceCheck: Detect trailing spaces
  - TabIndentationCheck: Must use spaces
  - FinalNewlineCheck: Must end with \n
  - NamingConventionCheck: PascalCase classes, snake_case functions
  - ImportOrderingCheck: stdlib < third-party < local
- Quality checks (4):
  - TypeHintsCheck: Public functions must have return type and param annotations
  - DocstringsCheck: Public classes and functions must have docstrings
  - MutableDefaultsCheck: No [], {}, set() as defaults
  - GodObjectCheck: Max 300 lines/class, max 15 public methods, max 500 lines/file
- Structure checks (6):
  - StubMethodCheck: No log-only, no-op methods in non-test code
  - PlaceholderCheck: No pass-only bodies, no raise NotImplementedError
  - FunctionArgumentsCheck: Max 5 parameters
  - FileLengthCheck: Max 500 lines
  - NestingDepthStructureCheck: Max 4 levels
  - ErrorHandlingCheck: No bare except:, no swallowed exceptions
- Safety checks (10):
  - BareOsEnvironCheck: os.environ["X"] must use .get()
  - MockPatternCheck: 14 regex patterns from checks_safety.py
  - EvalExecCheck: No eval()/exec() calls
  - HardcodedSecretCheck: password/token patterns
  - BannedLibraryCheck: From .claude/banned-libraries.txt
  - FlushDBCheck: No flushdb()/flushall() in tests
  - UnscopedDeleteCheck: No unscoped DETACH DELETE in tests
  - TODOCommentCheck: TODO/FIXME/HACK/XXX comments
  - BroadRaiseCheck: No raise Exception(...) or raise BaseException(...)
  - PrintStatementCheck: No print() outside __main__ guard
- Correctness checks (6):
  - RelativeImportCheck: No level > 0 imports
  - ImportedModuleExistenceCheck: Imported module must exist in forest
  - EntryPointGuardCheck: CLI files must have if __name__ == "__main__":
  - ThirdPartyImportDeclarationCheck: Must be in pyproject.toml
  - FileCompletenessCheck: All imports resolve
  - DependencyDeclarationCheck: All third-party imports declared
- Test isolation checks (4):
  - FlushDBIsolationCheck: No flushdb()/flushall() in tests
  - UnscopedDeleteIsolationCheck: No unscoped DETACH DELETE in tests
  - ModuleScopedFixtureCheck: No module-scoped fixtures for mutable state
  - GlobalMutableStateCheck: No global mutable state in test files
- DEFAULT_REVIEWS list (ParseIntegrity, BannedConstructs, DiffBudget, PrologSemanticReview)
src/orchestrata_ultimate/refactor/codemod/observability/__init__.py (~50 LOC)
- Public API exports for all observers
src/orchestrata_ultimate/refactor/codemod/observability/tracing.py (~150 LOC)
- TracingObserver class
- Writes structured JSON spans via core.tracing.Tracer
- Includes correlation_id context
- Each span: span_name, duration_ms, event_type, payload
src/orchestrata_ultimate/refactor/codemod/observability/metrics.py (~180 LOC)
- MetricsObserver class
- Counters: passes_run, files_mutated, findings_by_severity, commits_by_status
- Histograms: pass_duration_seconds, commit_duration_seconds, workspace_load_seconds
- Bucket configuration for histograms
src/orchestrata_ultimate/refactor/codemod/observability/debug_panel.py (~120 LOC)
- DebugPanelObserver class
- Writes NDJSON trace files to pipeline_output/traces/{correlation_id}.ndjson
- Each line: timestamp, event_type, payload
- Post-mortem analysis format
src/orchestrata_ultimate/refactor/codemod/observability/error_detector.py (~150 LOC)
- ErrorDetector class
- Pattern-matches error sequences (e.g., 3 consecutive parse failures)
- Emits alerts via EventBus
- Configurable patterns
src/orchestrata_ultimate/refactor/codemod/observability/graph_writer.py (~200 LOC)
- GraphWriterObserver class
- Async writes to graph DB via CodemodGraphRepository
- Batched, non-blocking
- Drops events on buffer overflow (does not block codemod)

Files to Test

tests/refactor/codemod/test_review_before_write.py (~400 LOC)
- 8 BDD scenarios from review_before_write.feature
- Test ParseIntegrity (valid mutations pass)
- Test BannedConstructs (bare os.environ, mock import, banned library)
- Test DiffBudget (exceeding budget aborts)
- Test error finding aborts, warning does not
- Test PrologSemanticReview (import cycle, blast radius)
tests/refactor/codemod/test_in_memory_review.py (~600 LOC)
- 40 BDD scenarios from in_memory_review.feature
- Test all 40 checks across 7 categories
- Test three-phase execution (fast structural, graph-dependent, Prolog)
- Test all phases run on mutated forest
- Test no subprocess calls, no external linters
tests/refactor/codemod/test_observability.py (~300 LOC)
- 9 BDD scenarios from observability.feature
- Test workspace emits events for all operations
- Test correlation ID links all events
- Test TracingObserver writes structured JSON spans
- Test GraphWriterObserver writes to graph DB asynchronously
- Test ErrorDetector pattern-matches error sequences
- Test MetricsObserver records counters and histograms
- Test DebugPanelObserver writes NDJSON trace files
- Test observers are optional and configurable
- Test custom observers can be provided

Instructions for ARCHITECT Agent

Step 1: Implement passes.py

Implement the 5 built-in transform passes. Each pass must:

Extend TransformPass abstract base class
Implement apply(workspace) -> ChangeSet for serial mode
Implement apply_file(path, tree, workspace) -> ChangeEntry | None for per-file parallel mode
Declare parallelism class variable ("serial" or "per_file")
Declare max_files and max_lines for DiffBudget review
Carry @traces decorator with correct spec/story/reqs

ReplaceStringLiteralPass

@traces(spec="110", story="US-110-002", reqs=["REQ-110-010"])
class ReplaceStringLiteralPass(TransformPass):
    """Replace string literals in exact or regex mode."""
    
    name = "ReplaceStringLiteralPass"
    parallelism = "per_file"
    
    def __init__(
        self,
        old: str,
        new: str,
        regex: bool = False,
        keep_patterns: Sequence[str] = (),
    ) -> None:
        self._old = old
        self._new = new
        self._regex = regex
        self._keep_patterns = keep_patterns
    
    def apply_file(
        self,
        path: str,
        tree: libcst.Module,
        ws: CodemodWorkspace,
    ) -> ChangeEntry | None:
        """Replace string literals in one file."""
        # Check keep_patterns (skip file if matches)
        if any(fnmatch.fnmatch(path, pat) for pat in self._keep_patterns):
            return None
        
        # Use CSTTransformer to walk and replace SimpleString/ConcatenatedString
        transformer = _StringLiteralTransformer(self._old, self._new, self._regex)
        mutated_tree = tree.visit(transformer)
        
        if mutated_tree.deep_equals(tree):
            return None
        
        return ChangeEntry(
            path=path,
            mutated_tree=mutated_tree,
            original_bytes=tree.code.encode('utf-8'),
            description=f"Replaced '{self._old}' with '{self._new}'",
            pass_name=self.name,
        )

AddImportPass

@traces(spec="110", story="US-110-007", reqs=["REQ-110-010"])
class AddImportPass(TransformPass):
    """Add import statement to target modules."""
    
    name = "AddImportPass"
    parallelism = "per_file"
    
    def __init__(self, module: str, name: str | None = None) -> None:
        self._module = module
        self._name = name
    
    def apply_file(
        self,
        path: str,
        tree: libcst.Module,
        ws: CodemodWorkspace,
    ) -> ChangeEntry | None:
        """Add import to one file."""
        # Check if import already exists
        if self._import_exists(tree):
            return None
        
        # Use LibCST AddImportVisitor pattern
        # Place in correct group (stdlib < third-party < local)
        mutated_tree = self._add_import(tree)
        
        return ChangeEntry(
            path=path,
            mutated_tree=mutated_tree,
            original_bytes=tree.code.encode('utf-8'),
            description=f"Added import: from {self._module} import {self._name}",
            pass_name=self.name,
        )

RemoveUnusedImportPass

@traces(spec="110", story="US-110-007", reqs=["REQ-110-010"])
class RemoveUnusedImportPass(TransformPass):
    """Remove imports not referenced in module body."""
    
    name = "RemoveUnusedImportPass"
    parallelism = "per_file"
    
    def apply_file(
        self,
        path: str,
        tree: libcst.Module,
        ws: CodemodWorkspace,
    ) -> ChangeEntry | None:
        """Remove unused imports from one file."""
        # Use CIG IMPORTS edges to detect unused imports
        # Or walk tree to find Name nodes referencing imported names
        unused_imports = self._find_unused_imports(tree, ws, path)
        
        if not unused_imports:
            return None
        
        mutated_tree = self._remove_imports(tree, unused_imports)
        
        return ChangeEntry(
            path=path,
            mutated_tree=mutated_tree,
            original_bytes=tree.code.encode('utf-8'),
            description=f"Removed {len(unused_imports)} unused imports",
            pass_name=self.name,
        )

DocstringSweepPass

@traces(spec="110", story="US-110-002", reqs=["REQ-110-010"])
class DocstringSweepPass(TransformPass):
    """Vocabulary replacement in docstrings AND comments."""
    
    name = "DocstringSweepPass"
    parallelism = "per_file"
    
    def __init__(self, replacements: dict[str, str]) -> None:
        self._replacements = replacements
    
    def apply_file(
        self,
        path: str,
        tree: libcst.Module,
        ws: CodemodWorkspace,
    ) -> ChangeEntry | None:
        """Replace vocabulary in docstrings and comments."""
        # LibCST preserves comments as EmptyLine nodes with comment field
        # Walk tree and replace in:
        # - SimpleString nodes (docstrings)
        # - EmptyLine.comment nodes (comments)
        transformer = _DocstringCommentTransformer(self._replacements)
        mutated_tree = tree.visit(transformer)
        
        if mutated_tree.deep_equals(tree):
            return None
        
        return ChangeEntry(
            path=path,
            mutated_tree=mutated_tree,
            original_bytes=tree.code.encode('utf-8'),
            description=f"Swept {len(self._replacements)} vocabulary replacements",
            pass_name=self.name,
        )

RenameSymbolPass

@traces(spec="110", story="US-110-009", reqs=["REQ-110-010", "REQ-110-020", "REQ-110-021"])
class RenameSymbolPass(TransformPass):
    """Consume WorkspaceEditPlan from SPEC-109, apply LSP edits as LibCST mutations."""
    
    name = "RenameSymbolPass"
    parallelism = "serial"  # cross-file dependencies
    
    def __init__(self, plan: WorkspaceEditPlan) -> None:
        self._plan = plan
    
    def apply(self, ws: CodemodWorkspace) -> ChangeSet:
        """Apply LSP text edits from SPEC-109 plan."""
        entries = []
        for file_edit in self._plan.files:
            path = str(file_edit.path.relative_to(ws._root))
            tree = ws.forest[path]
            
            # Convert LSP text edits to LibCST mutations
            wrapper = libcst.MetadataWrapper(tree)
            transformer = _LSPToCSTTransformer(file_edit.edits, wrapper)
            mutated_tree = wrapper.visit(transformer)
            
            entries.append(ChangeEntry(
                path=path,
                mutated_tree=mutated_tree,
                original_bytes=tree.code.encode('utf-8'),
                description=f"Applied {len(file_edit.edits)} LSP edits",
                pass_name=self.name,
            ))
        
        return ChangeSet(entries=tuple(entries))

Step 2: Implement Analysis Modules

Each analysis pass must:

Be read-only (no mutations to CST forest or CIG)
Return structured results (frozen dataclasses with slots=True)
Have __str__ method for human-readable output
Have to_dict() method for JSON serialization
Carry @traces decorator with correct spec/story/reqs

BlastRadiusAnalyzer

@dataclass(frozen=True, slots=True)
class BlastRadiusResult:
    """Result of blast radius analysis."""
    affected_nodes: frozenset[int]
    affected_files: frozenset[str]
    impact_score: float  # 0.0 to 1.0
    
    def __str__(self) -> str:
        return f"Blast radius: {len(self.affected_nodes)} nodes, {len(self.affected_files)} files, impact {self.impact_score:.2f}"
    
    def to_dict(self) -> dict[str, Any]:
        return {
            "affected_nodes": sorted(self.affected_nodes),
            "affected_files": sorted(self.affected_files),
            "impact_score": self.impact_score,
        }


@traces(spec="110", story="US-110-006", reqs=["REQ-110-031"])
class BlastRadiusAnalyzer:
    """BFS blast radius computation over CALLS + INHERITS + IMPORTS edges."""
    
    def analyze(self, cig: CodeIntelligenceGraph, node_id: int, hops: int = 3) -> BlastRadiusResult:
        """Compute blast radius for a node."""
        blast = cig.blast_radius(node_id, hops)
        return BlastRadiusResult(
            affected_nodes=blast.affected_nodes,
            affected_files=blast.affected_files,
            impact_score=blast.impact_score,
        )

BreakingChangeDetector

@dataclass(frozen=True, slots=True)
class BreakingChangeResult:
    """Result of breaking change detection."""
    broken_consumers: tuple[BreakingChange, ...]
    
    def __str__(self) -> str:
        return f"Breaking changes: {len(self.broken_consumers)} broken consumers"
    
    def to_dict(self) -> dict[str, Any]:
        return {
            "broken_consumers": [
                {
                    "consumer_path": bc.consumer_path,
                    "consumer_line": bc.consumer_line,
                    "incompatibility": bc.incompatibility,
                }
                for bc in self.broken_consumers
            ],
        }


@traces(spec="110", story="US-110-003", reqs=["REQ-110-031"])
class BreakingChangeDetector:
    """API breaking change detection via signature diff + downstream consumers."""
    
    def analyze(
        self,
        cig: CodeIntelligenceGraph,
        original_signature: FunctionSignature,
        mutated_signature: FunctionSignature,
    ) -> BreakingChangeResult:
        """Detect breaking changes between original and mutated signatures."""
        # Compute signature diff
        incompatibilities = self._diff_signatures(original_signature, mutated_signature)
        
        if not incompatibilities:
            return BreakingChangeResult(broken_consumers=())
        
        # Find downstream consumers via CALLS + INHERITS edges
        consumers = self._find_consumers(cig, original_signature.qualified_name)
        
        broken = []
        for consumer in consumers:
            for incompat in incompatibilities:
                broken.append(BreakingChange(
                    change_path=original_signature.module_path,
                    consumer_path=consumer.module_path,
                    consumer_line=consumer.line,
                    incompatibility=incompat,
                ))
        
        return BreakingChangeResult(broken_consumers=tuple(broken))

ComplexityAnalyzer

@dataclass(frozen=True, slots=True)
class ComplexityResult:
    """Result of complexity analysis."""
    cyclomatic: int
    nesting: int
    cognitive: int
    line_count: int
    param_count: int
    
    def __str__(self) -> str:
        return f"Complexity: cyclomatic={self.cyclomatic}, nesting={self.nesting}, cognitive={self.cognitive}"
    
    def to_dict(self) -> dict[str, Any]:
        return {
            "cyclomatic": self.cyclomatic,
            "nesting": self.nesting,
            "cognitive": self.cognitive,
            "line_count": self.line_count,
            "param_count": self.param_count,
        }


@traces(spec="110", story="US-110-006", reqs=["REQ-110-031"])
class ComplexityAnalyzer:
    """Cyclomatic + nesting + cognitive complexity per function/method."""
    
    def analyze(self, tree: libcst.Module) -> dict[str, ComplexityResult]:
        """Analyze complexity for all functions/methods in a module."""
        visitor = _ComplexityVisitor()
        tree.visit(visitor)
        return visitor.results

TestCoverageMapper

@dataclass(frozen=True, slots=True)
class CoverageResult:
    """Result of test coverage mapping."""
    function_to_tests: dict[str, frozenset[str]]
    unmapped_functions: frozenset[str]
    
    def __str__(self) -> str:
        mapped = len(self.function_to_tests)
        unmapped = len(self.unmapped_functions)
        return f"Coverage: {mapped} mapped, {unmapped} unmapped"
    
    def to_dict(self) -> dict[str, Any]:
        return {
            "function_to_tests": {k: sorted(v) for k, v in self.function_to_tests.items()},
            "unmapped_functions": sorted(self.unmapped_functions),
        }


@traces(spec="110", story="US-110-006", reqs=["REQ-110-031"])
class TestCoverageMapper:
    """Test-to-source mapping via TESTED_BY edges + name heuristic."""
    
    def analyze(self, cig: CodeIntelligenceGraph) -> CoverageResult:
        """Map functions to their tests."""
        # Use TESTED_BY edges from CIG
        # Fall back to name heuristic (strip test_ prefix)
        ...

AbstractionAnalyzer

@dataclass(frozen=True, slots=True)
class AbstractionResult:
    """Result of abstraction analysis."""
    abstract_count: int
    concrete_count: int
    ratio: float
    
    def __str__(self) -> str:
        return f"Abstraction: {self.abstract_count} abstract, {self.concrete_count} concrete, ratio {self.ratio:.2f}"
    
    def to_dict(self) -> dict[str, Any]:
        return {
            "abstract_count": self.abstract_count,
            "concrete_count": self.concrete_count,
            "ratio": self.ratio,
        }


@traces(spec="110", story="US-110-006", reqs=["REQ-110-031"])
class AbstractionAnalyzer:
    """Abstraction level (ratio of Protocol/ABC to concrete classes)."""
    
    def analyze(self, cig: CodeIntelligenceGraph, module_id: int) -> AbstractionResult:
        """Compute abstraction level for a module."""
        # Query CIG for classes in module
        # Count Protocol/ABC vs concrete
        ...

CouplingAnalyzer

@dataclass(frozen=True, slots=True)
class CouplingResult:
    """Result of coupling analysis."""
    efferent: int
    afferent: int
    instability: float
    
    def __str__(self) -> str:
        return f"Coupling: efferent={self.efferent}, afferent={self.afferent}, instability={self.instability:.2f}"
    
    def to_dict(self) -> dict[str, Any]:
        return {
            "efferent": self.efferent,
            "afferent": self.afferent,
            "instability": self.instability,
        }


@traces(spec="110", story="US-110-006", reqs=["REQ-110-031"])
class CouplingAnalyzer:
    """Efferent/afferent coupling + instability metric."""
    
    def analyze(self, cig: CodeIntelligenceGraph, module_id: int) -> CouplingResult:
        """Compute coupling metrics for a module."""
        # Query CIG for dependencies (efferent)
        # Query CIG for dependents (afferent)
        # Compute instability = efferent / (efferent + afferent)
        ...

DeadCodeDetector

@dataclass(frozen=True, slots=True)
class DeadCodeResult:
    """Result of dead code detection."""
    dead_functions: frozenset[str]
    dead_classes: frozenset[str]
    
    def __str__(self) -> str:
        return f"Dead code: {len(self.dead_functions)} functions, {len(self.dead_classes)} classes"
    
    def to_dict(self) -> dict[str, Any]:
        return {
            "dead_functions": sorted(self.dead_functions),
            "dead_classes": sorted(self.dead_classes),
        }


@traces(spec="110", story="US-110-006", reqs=["REQ-110-031"])
class DeadCodeDetector:
    """Unreachable functions/classes via CIG reachability analysis."""
    
    def analyze(self, cig: CodeIntelligenceGraph) -> DeadCodeResult:
        """Detect dead code in the codebase."""
        # Functions with no CALLS edges and no TESTED_BY edges
        # Classes with no INHERITS edges and no instantiation
        ...

Step 3: Write Tests

Write BDD scenarios from transform_passes.feature and analysis_passes.feature. Use pytest-bdd with @scenario decorator and TST/REQ docstrings.

Instructions for REVIEWER Agent

Step 1: Implement review.py

Implement the InMemoryReviewEngine with 40 checks organized into 7 categories. The engine executes checks in three phases:

Phase 1 — Fast structural checks (parallel per-file): Complexity, style, quality, structure checks. Each file is reviewed independently via ThreadPoolExecutor.
Phase 2 — Graph-dependent checks (require CIG): Correctness, test isolation checks. These checks query the CIG for cross-file relationships.
Phase 3 — Prolog semantic review (require Prolog facts): The PrologSemanticReview pass asserts mutations as Prolog facts, queries codemod_review.pl predicates.

@traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034", "REQ-110-035"])
class InMemoryReviewEngine:
    """Execute 40 checks on mutated CST forest in 3 phases."""
    
    def __init__(self, checks: Sequence[Check] | None = None) -> None:
        self._checks = checks or DEFAULT_CHECKS
    
    def review(
        self,
        ws: CodemodWorkspace,
        changes: ChangeSet,
    ) -> list[Finding]:
        """Execute all checks in 3 phases."""
        findings = []
        
        # Phase 1: Fast structural (parallel)
        with ThreadPoolExecutor(max_workers=os.cpu_count()) as pool:
            futures = [
                pool.submit(self._review_file, entry.mutated_tree, entry.path)
                for entry in changes.entries
            ]
            for f in futures:
                findings.extend(f.result())
        
        # Phase 2: Graph-dependent
        findings.extend(self._review_graph(ws, changes))
        
        # Phase 3: Prolog semantic
        findings.extend(self._review_prolog(changes))
        
        return findings

Each check must:

Extend Check abstract base class
Implement check_file(tree, path) -> list[Finding] for phase 1
Implement check_graph(ws, changes) -> list[Finding] for phase 2
Use CSTTransformer or CSTVisitor (no ast.parse, no subprocess.run)
Carry @traces decorator with correct spec/story/reqs

Example Check Implementation

@traces(spec="110", story="US-110-004", reqs=["REQ-110-040"])
class BareOsEnvironCheck(Check):
    """Detect bare os.environ["X"] subscript reads."""
    
    name = "BARE_OS_ENVIRON"
    severity = "error"
    
    def check_file(self, tree: libcst.Module, path: str) -> list[Finding]:
        """Check for bare os.environ subscript reads."""
        visitor = _OsEnvironVisitor()
        tree.visit(visitor)
        
        findings = []
        for line in visitor.violations:
            findings.append(Finding(
                severity=self.severity,
                path=path,
                line=line,
                rule=self.name,
                message="Bare os.environ['X'] subscript read; use os.environ.get('X', default)",
            ))
        
        return findings


class _OsEnvironVisitor(cst.CSTVisitor):
    """Visit CST to find os.environ subscript reads."""
    
    def __init__(self) -> None:
        self.violations: list[int] = []
    
    def visit_Subscript(self, node: cst.Subscript) -> None:
        """Check for os.environ["X"] pattern."""
        # Detect os.environ subscript
        ...

Step 2: Implement Observability Modules

Each observer must:

Implement Observer[CodemodEvent] protocol from core.patterns.observer
Implement on_event(event: CodemodEvent) -> None
Be optional and configurable
Follow log-and-continue policy (exceptions logged, not raised)
Carry @traces decorator with correct spec/story/reqs

TracingObserver

@traces(spec="110", story="US-110-005", reqs=["REQ-110-026"])
class TracingObserver:
    """Write structured JSON spans via core.tracing.Tracer."""
    
    def __init__(self) -> None:
        self._tracer = Tracer()
        self._logger = logging.getLogger(__name__)
    
    def on_event(self, event: CodemodEvent) -> None:
        """Write span for event."""
        try:
            self._tracer.start_span(
                name=event.span_name or event.event_type.value,
                attributes={
                    "correlation_id": event.correlation_id,
                    "event_type": event.event_type.value,
                    **event.payload,
                },
            )
            # Record duration if available
            if "duration_seconds" in event.payload:
                self._tracer.end_span(duration_ms=event.payload["duration_seconds"] * 1000)
        except Exception as e:
            self._logger.warning(f"TracingObserver failed: {e}")

MetricsObserver

@traces(spec="110", story="US-110-005", reqs=["REQ-110-026"])
class MetricsObserver:
    """Record counters and histograms."""
    
    def __init__(self) -> None:
        self._counters: dict[str, int] = {}
        self._histograms: dict[str, list[float]] = {}
    
    def on_event(self, event: CodemodEvent) -> None:
        """Increment counters and record histograms."""
        try:
            # Counters
            if event.event_type == CodemodEventType.PASS_COMPLETED:
                self._increment("codemod_passes_run_total", labels={"pass_name": event.payload.get("pass_name")})
                self._increment("codemod_files_mutated_total", labels={"pass_name": event.payload.get("pass_name")}, value=event.payload.get("file_count", 0))
            
            if event.event_type == CodemodEventType.FINDING_EMITTED:
                self._increment("codemod_findings_total", labels={"severity": event.payload.get("severity"), "rule": event.payload.get("rule")})
            
            # Histograms
            if event.event_type == CodemodEventType.PASS_COMPLETED:
                self._record("codemod_pass_duration_seconds", event.payload.get("duration_seconds", 0))
            
            if event.event_type == CodemodEventType.COMMIT_COMPLETED:
                self._increment("codemod_commits_total", labels={"status": "success"})
                self._record("codemod_commit_duration_seconds", event.payload.get("duration_seconds", 0))
        except Exception as e:
            logging.getLogger(__name__).warning(f"MetricsObserver failed: {e}")

DebugPanelObserver

@traces(spec="110", story="US-110-005", reqs=["REQ-110-026"])
class DebugPanelObserver:
    """Write NDJSON trace files for post-mortem analysis."""
    
    def __init__(self, output_dir: Path) -> None:
        self._output_dir = output_dir
        self._files: dict[str, Path] = {}
    
    def on_event(self, event: CodemodEvent) -> None:
        """Write NDJSON line for event."""
        try:
            # Create file per correlation_id
            if event.correlation_id not in self._files:
                self._output_dir.mkdir(parents=True, exist_ok=True)
                self._files[event.correlation_id] = self._output_dir / f"{event.correlation_id}.ndjson"
            
            # Write JSON line
            line = json.dumps({
                "timestamp": event.timestamp,
                "event_type": event.event_type.value,
                "payload": event.payload,
            })
            with self._files[event.correlation_id].open("a") as f:
                f.write(line + "\n")
        except Exception as e:
            logging.getLogger(__name__).warning(f"DebugPanelObserver failed: {e}")

ErrorDetector

@traces(spec="110", story="US-110-005", reqs=["REQ-110-026"])
class ErrorDetector:
    """Pattern-match error sequences and emit alerts."""
    
    def __init__(self, patterns: Sequence[str] = ("3_consecutive_parse_failures",)) -> None:
        self._patterns = patterns
        self._error_buffer: list[CodemodEvent] = []
        self._event_bus = EventBus()
    
    def on_event(self, event: CodemodEvent) -> None:
        """Detect error patterns."""
        try:
            if event.event_type == CodemodEventType.ERROR_DETECTED:
                self._error_buffer.append(event)
                self._check_patterns()
        except Exception as e:
            logging.getLogger(__name__).warning(f"ErrorDetector failed: {e}")
    
    def _check_patterns(self) -> None:
        """Check for error patterns."""
        if "3_consecutive_parse_failures" in self._patterns:
            if len(self._error_buffer) >= 3:
                # Check if last 3 are parse failures
                last_3 = self._error_buffer[-3:]
                if all(e.payload.get("error_type") == "parse_failure" for e in last_3):
                    self._event_bus.emit(SystemEvent(
                        id=f"codemod-error-{last_3[0].correlation_id}",
                        type="codemod.error",
                        source="codemod_engine",
                        timestamp=time.time(),
                        payload={"pattern": "3_consecutive_parse_failures", "files": [e.payload.get("path") for e in last_3]},
                        correlation_id=last_3[0].correlation_id,
                        severity="high",
                    ))

GraphWriterObserver

@traces(spec="110", story="US-110-010", reqs=["REQ-110-026"])
class GraphWriterObserver:
    """Async writes to graph DB via CodemodGraphRepository."""
    
    def __init__(self, repository: CodemodGraphRepository, batch_size: int = 100) -> None:
        self._repo = repository
        self._batch_size = batch_size
        self._buffer: list[CodemodEvent] = []
    
    def on_event(self, event: CodemodEvent) -> None:
        """Buffer events and flush on commit."""
        try:
            self._buffer.append(event)
            
            # Flush on commit completion
            if event.event_type == CodemodEventType.COMMIT_COMPLETED:
                self._flush()
            
            # Drop on buffer overflow (non-blocking)
            if len(self._buffer) > self._batch_size * 10:
                self._buffer = self._buffer[-self._batch_size:]
        except Exception as e:
            logging.getLogger(__name__).warning(f"GraphWriterObserver failed: {e}")
    
    def _flush(self) -> None:
        """Write buffered events to graph DB."""
        # Convert events to records
        # Call repository.save_run(), save_pass(), save_findings(), save_changes()
        ...

Step 3: Write Tests

Write BDD scenarios from review_before_write.feature, in_memory_review.feature, and observability.feature. Use pytest-bdd with @scenario decorator and TST/REQ docstrings.

Success Criteria

ARCHITECT Agent

[ ] passes.py implements 5 built-in transform passes
[ ] All passes carry @traces decorators with correct spec/story/reqs
[ ] ReplaceStringLiteralPass supports exact and regex modes with keep-pattern guard
[ ] AddImportPass places imports in correct group (stdlib < third-party < local)
[ ] RemoveUnusedImportPass uses CIG IMPORTS edges
[ ] DocstringSweepPass updates docstrings AND comments (LibCST preserves both)
[ ] RenameSymbolPass consumes WorkspaceEditPlan from SPEC-109 (no second resolver)
[ ] All 7 analysis modules implemented with structured results (dataclasses)
[ ] All analysis results have __str__ and to_dict() methods
[ ] All analysis passes are read-only (no mutations)
[ ] All BDD scenarios pass (transform_passes, analysis_passes)
[ ] Performance targets met (passes execute in <3s on 797 files)

REVIEWER Agent

[ ] review.py implements 40 checks organized into 7 categories
[ ] All checks carry @traces decorators with correct spec/story/reqs
[ ] All checks operate on LibCST trees (no ast.parse, no subprocess.run)
[ ] Three-phase execution (fast structural, graph-dependent, Prolog)
[ ] All phases run on mutated forest (not on-disk files)
[ ] DEFAULT_REVIEWS list includes ParseIntegrity, BannedConstructs, DiffBudget, PrologSemanticReview
[ ] All 5 observer classes implemented
[ ] All observers are optional and configurable
[ ] All observers follow log-and-continue policy
[ ] TracingObserver writes structured JSON spans via core.tracing.Tracer
[ ] MetricsObserver records counters and histograms
[ ] DebugPanelObserver writes NDJSON trace files
[ ] ErrorDetector pattern-matches error sequences
[ ] GraphWriterObserver writes to graph DB asynchronously (batched, non-blocking)
[ ] All BDD scenarios pass (review_before_write, in_memory_review, observability)
[ ] Review completes in <500ms (no subprocess overhead)

Deliverables

ARCHITECT Agent

src/orchestrata_ultimate/refactor/codemod/passes.py (5 transform passes)
src/orchestrata_ultimate/refactor/codemod/analysis/__init__.py (exports)
src/orchestrata_ultimate/refactor/codemod/analysis/blast_radius.py
src/orchestrata_ultimate/refactor/codemod/analysis/breaking_changes.py
src/orchestrata_ultimate/refactor/codemod/analysis/complexity.py
src/orchestrata_ultimate/refactor/codemod/analysis/coverage.py
src/orchestrata_ultimate/refactor/codemod/analysis/abstraction.py
src/orchestrata_ultimate/refactor/codemod/analysis/coupling.py
src/orchestrata_ultimate/refactor/codemod/analysis/dead_code.py
tests/refactor/codemod/test_transform_passes.py (8 BDD scenarios)
tests/refactor/codemod/test_analysis_passes.py (13 BDD scenarios)

REVIEWER Agent

src/orchestrata_ultimate/refactor/codemod/review.py (40 checks + InMemoryReviewEngine)
src/orchestrata_ultimate/refactor/codemod/observability/__init__.py (exports)
src/orchestrata_ultimate/refactor/codemod/observability/tracing.py
src/orchestrata_ultimate/refactor/codemod/observability/metrics.py
src/orchestrata_ultimate/refactor/codemod/observability/debug_panel.py
src/orchestrata_ultimate/refactor/codemod/observability/error_detector.py
src/orchestrata_ultimate/refactor/codemod/observability/graph_writer.py
tests/refactor/codemod/test_review_before_write.py (8 BDD scenarios)
tests/refactor/codemod/test_in_memory_review.py (40 BDD scenarios)
tests/refactor/codemod/test_observability.py (9 BDD scenarios)

Handoff to W3

W3 (Integrator) will implement:

scripts/codemod.py CLI with load-bench, run, report, analyze subcommands
SPEC-109 interop (RenameSymbolPass consumes WorkspaceEditPlan)
Integration tests (end-to-end codemod operations)
Wiring workspace.run() and workspace.commit() to use passes, reviews, and observers

W3 agents will use the passes, reviews, and observers from W2 to build the CLI and integration layer.

Notes

ARCHITECT and REVIEWER work in parallel (no dependencies between them)
ARCHITECT agent should use Qwen Max (strong at algorithms and code transformation)
REVIEWER agent should use Sonnet (strong at pattern matching and observability)
Both agents must use the workspace, CIG, and txn from W1
Both agents must use the skeleton data structures from W0 (ChangeEntry, ChangeSet, Finding, etc.)
All code must have @traces decorators with correct spec/story/reqs
All tests must use BDD scenarios from feature files
All checks must operate on LibCST trees (no ast.parse, no subprocess.run)
All observers must follow log-and-continue policy (exceptions logged, not raised)
All analysis passes must be read-only (no mutations to CST forest or CIG)
All analysis results must have __str__ and to_dict() methods for CLI output and JSON serialization

BRIEF-W2: Architect + Reviewer (Qwen Max + Sonnet, Parallel)

Plan text

BRIEF-W2: Architect + Reviewer (Qwen Max + Sonnet, Parallel)

Wave Overview

Objective

Scope

ARCHITECT Agent (Qwen Max)

Files to Implement

Files to Test

REVIEWER Agent (Sonnet)

Files to Implement

Files to Test

Instructions for ARCHITECT Agent

Step 1: Implement passes.py

ReplaceStringLiteralPass

AddImportPass

RemoveUnusedImportPass

DocstringSweepPass

RenameSymbolPass

Step 2: Implement Analysis Modules

BlastRadiusAnalyzer

BreakingChangeDetector

ComplexityAnalyzer

TestCoverageMapper

AbstractionAnalyzer

CouplingAnalyzer

DeadCodeDetector

Step 3: Write Tests

Instructions for REVIEWER Agent

Step 1: Implement review.py

Example Check Implementation

Step 2: Implement Observability Modules

TracingObserver

MetricsObserver

DebugPanelObserver

ErrorDetector

GraphWriterObserver

Step 3: Write Tests

Success Criteria

ARCHITECT Agent

REVIEWER Agent

Deliverables

ARCHITECT Agent

REVIEWER Agent

Handoff to W3

Notes

Related