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

BRIEF-W3: Integrator (Qwen Max Agent)

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

Prose pending.

Plan text

BRIEF-W3: Integrator (Qwen Max Agent)

Wave Overview

Wave: W3 (Integrator)
Agent: Qwen Max (CLI design, interop glue, benchmark harness)
Duration: ~15 minutes
Dependencies: W1 (Forest + Sentinel) + W2 (Architect + Reviewer)
Parallel Execution: No (single agent — integrates all W1+W2 components)

Objective

Implement the user-facing integration layer for the codemod engine:

1. CLI (scripts/codemod.py): load-bench, run, report, analyze subcommands with correct exit codes
2. SPEC-109 interop (RenameSymbolPass): consumes WorkspaceEditPlan from the rename engine, converts LSP text edits to LibCST mutations
3. Benchmark tests: measure parse time, RSS delta, and query latency against spec performance targets

Scope

Files to Implement

1. scripts/codemod.py (~350 LOC)

   • main() entry point with argparse subcommands
   • load-bench subcommand: measure workspace load performance
   • run subcommand: execute named passes with --pass, --arg, --apply, --dry-run flags
   • report subcommand: pretty-print a CommitReport JSON file
   • analyze subcommand: execute CIG queries (--query, --module, --hops)
   • --auto-start flag: auto-connect to daemon or start it
   • Exit codes: 0 (success/clean dry-run), 1 (review-abort/verification failure), 2 (usage error)

2. tests/refactor/codemod/test_cli.py (~300 LOC)

   • 7 BDD scenarios from cli.feature
   • Subprocess tests asserting exit codes 0, 1, 2
   • Verify dry-run default (no disk writes)
   • Verify --apply writes files
   • Verify report pretty-prints JSON
   • Verify analyze executes CIG queries

3. tests/refactor/codemod/test_spec109_interop.py (~250 LOC)

   • 4 BDD scenarios from spec109_interop.feature
   • Verify RenameSymbolPass consumes WorkspaceEditPlan
   • Verify no second rename resolver exists (AST grep on pass code)
   • Verify write-back reuses applier safety invariants
   • Verify byte-identical results between EditApplier and RenameSymbolPass

4. tests/refactor/codemod/test_benchmark_load.py (~150 LOC)

   • Plain pytest (not BDD — performance tests)
   • Measure serial parse time (target: <360s for full codebase)
   • Measure parallel parse time (target: <120s on 12 cores)
   • Measure RSS delta (target: <3 GiB)
   • Measure CIG query latency (target: <2ms per query)

Files to Modify

5. src/orchestrata_ultimate/refactor/codemod/passes.py (add RenameSymbolPass, ~120 LOC)
   • RenameSymbolPass class consuming WorkspaceEditPlan
   • _LSPToCSTTransformer converting LSP text edits to LibCST mutations via PositionProvider

Files NOT to Modify

• workspace.py, txn.py, _cig.py, _csr_index.py — implemented in W1
• review.py, analysis/*.py, observability/*.py — implemented in W2
• refactor/rename/* — frozen, no edits to SPEC-109 source
• refactor/__init__.py, agents/tool_registry.py — wiring is a later step

Instructions

Step 1: Implement RenameSymbolPass in passes.py

The RenameSymbolPass bridges SPEC-109's LSP-based rename resolution with SPEC-110's LibCST-based transform pipeline. It consumes a WorkspaceEditPlan (produced by SPEC-109's SymbolRenamer) and converts LSP text edits into LibCST mutations.

Key design constraint: The pass adds NO second rename resolver. All symbol resolution is done by SPEC-109; this pass only applies the pre-computed edits.

# Add to src/orchestrata_ultimate/refactor/codemod/passes.py

from orchestrata_ultimate.refactor.rename.models import WorkspaceEditPlan
import libcst
import libcst.metadata


class RenameSymbolPass(TransformPass):
    """Apply SPEC-109 WorkspaceEditPlan as LibCST mutations.

    Consumes pre-resolved LSP text edits — does NOT reimplement resolution.
    """

    name: ClassVar[str] = "RenameSymbolPass"
    parallelism: ClassVar[Literal["serial", "per_file"]] = "serial"

    def __init__(self, plan: WorkspaceEditPlan) -> None:
        self._plan = plan

    def apply(self, ws: CodemodWorkspace) -> ChangeSet:
        entries = []
        for file_edit in self._plan.files:
            path = str(file_edit.path.relative_to(ws._root))
            tree = ws.forest[path]
            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))


class _LSPToCSTTransformer(libcst.CSTTransformer):
    """Convert LSP text edits (line/col spans) to LibCST node mutations.

    Uses libcst.metadata.PositionProvider to map LSP line/character ranges
    to LibCST nodes, then applies the replacement text.
    """

    METADATA_DEPENDENCIES = (libcst.metadata.PositionProvider,)

    def __init__(self, edits, wrapper):
        self._edits = edits
        self._wrapper = wrapper
        self._replacements = {}
        self._build_replacement_map()

    def _build_replacement_map(self):
        """Map LibCST nodes to replacement text based on LSP edit ranges."""
        for edit in self._edits:
            # edit.range is (start_line, start_col, end_line, end_col)
            # edit.new_text is the replacement string
            # Find the LibCST node at this position and record the replacement
            ...

    def on_leave(self, original_node, updated_node):
        """Apply replacement if this node matches an edit range."""
        pos = self.get_metadata(libcst.metadata.PositionProvider, original_node)
        # Check if any edit range covers this node
        # Return updated_node with replacement applied applied
        return updated_node

Critical implementation details:

1. PositionProvider mapping: LSP edits use 0-based line/character offsets. LibCST's PositionProvider uses 1-based CodeRange(start=CodePosition(line, column), end=CodePosition(line, column)). Convert carefully.

2. Edit ordering: LSP edits may overlap or be out of order. Sort edits by start position (descending) and apply from end to start to avoid offset drift.

3. Granularity: LSP edits target text spans, not AST nodes. The transformer must find the smallest LibCST node whose position fully contains the edit range, then replace the relevant portion of its text.

4. Validation: Raise InvalidPlanError if the WorkspaceEditPlan has missing fields or references files not in the workspace forest.

Step 2: Implement scripts/codemod.py CLI

#!/usr/bin/env python3
"""Codemod engine CLI — load-bench, run, report, analyze subcommands."""

from __future__ import annotations

import argparse
import json
import os
import sys
import time
from pathlib import Path


def main() -> int:
    parser = argparse.ArgumentParser(
        prog="codemod",
        description="In-memory code intelligence platform",
    )
    parser.add_argument(
        "--auto-start",
        action="store_true",
        help="Auto-connect to daemon or start it if not running",
    )
    parser.add_argument(
        "--no-daemon",
        action="store_true",
        help="Force library mode (parse on every invocation)",
    )

    subparsers = parser.add_subparsers(dest="command", required=True)

    # --- load-bench ---
    subparsers.add_parser(
        "load-bench",
        help="Measure workspace load time, RSS delta, file count",
    )

    # --- run ---
    run_parser = subparsers.add_parser(
        "run",
        help="Execute named transform passes",
    )
    run_parser.add_argument("--pass", dest="pass_name", required=True)
    run_parser.add_argument("--arg", action="append", default=[])
    run_parser.add_argument("--apply", action="store_true")
    run_parser.add_argument("--dry-run", action="store_true", default=True)

    # --- report ---
    report_parser = subparsers.add_parser(
        "report",
        help="Pretty-print a CommitReport JSON file",
    )
    report_parser.add_argument("path", type=Path)

    # --- analyze ---
    analyze_parser = subparsers.add_parser(
        "analyze",
        help="Execute CIG queries",
    )
    analyze_parser.add_argument(
        "--query",
        required=True,
        choices=["dependencies", "dependents", "blast_radius", "cycles", "topological_order"],
    )
    analyze_parser.add_argument("--module", type=str)
    analyze_parser.add_argument("--hops", type=int, default=3)

    try:
        args = parser.parse_args()
    except SystemExit as e:
        return 2 if e.code != 0 else 0

    if args.command == "load-bench":
        return _cmd_load_bench(args)
    elif args.command == "run":
        return _cmd_run(args)
    elif args.command == "report":
        return _cmd_report(args)
    elif args.command == "analyze":
        return _cmd_analyze(args)
    return 2


def _cmd_load_bench(args) -> int:
    """Measure workspace load performance."""
    from orchestrata_ultimate.refactor.codemod.workspace import CodemodWorkspace

    root = Path(os.environ.get("CODEMOD_ROOT", "."))

    # Serial load
    t0 = time.monotonic()
    ws_serial = CodemodWorkspace.load(root, parallel=False)
    serial_time = time.monotonic() - t0

    # Parallel load
    t0 = time.monotonic()
    ws_parallel = CodemodWorkspace.load(root, parallel=True)
    parallel_time = time.monotonic() - t0

    stats = ws_parallel.stats()
    print(f"Load time (serial):   {serial_time:.1f}s")
    print(f"Load time (parallel): {parallel_time:.1f}s ({os.cpu_count()} cores)")
    print(f"Files loaded:         {stats.file_count}")
    print(f"Total bytes:          {stats.total_bytes:,}")
    print(f"RSS delta:            {stats.rss_delta_bytes / (1024**3):.2f} GiB")
    print(f"Unparsable:           {stats.unparsable_count}")
    return 0


def _cmd_run(args) -> int:
    """Execute transform passes."""
    from orchestrata_ultimate.refactor.codemod.workspace import CodemodWorkspace

    root = Path(os.environ.get("CODEMOD_ROOT", "."))
    ws = CodemodWorkspace.load(root, parallel=True)

    # Parse --arg k=v pairs
    pass_args = {}
    for arg in args.arg:
        k, v = arg.split("=", 1)
        pass_args[k] = v

    # Resolve pass by name
    pass_instance = _resolve_pass(args.pass_name, pass_args)

    # Execute
    changes = ws.run([pass_instance])

    # Commit (dry-run by default, --apply to write)
    dry_run = not args.apply
    report = ws.commit(changes, dry_run=dry_run)

    # Output
    if report.diffstat:
        print(report.diffstat)

    error_count = len([f for f in report.findings if f.severity == "error"])
    warn_count = len([f for f in report.findings if f.severity == "warning"])
    print(f"Findings: {error_count} errors, {warn_count} warnings")

    # Write report JSON
    report_dir = root / "reports"
    report_dir.mkdir(exist_ok=True)
    report_path = report_dir / f"codemod-{report.correlation_id}.json"
    report_path.write_text(report.to_json())
    print(f"Commit report: {report_path}")

    if error_count > 0 and not dry_run:
        return 1
    return 0


def _cmd_report(args) -> int:
    """Pretty-print a CommitReport JSON file."""
    data = json.loads(args.path.read_text())
    print(f"Passes run:     {', '.join(data['passes_run'])}")
    print(f"Files written:  {len(data['files_written'])}")
    print(f"Duration:       {data['duration_seconds']:.1f}s")
    print(f"Dry run:        {data['dry_run']}")
    print(f"Correlation ID: {data['correlation_id']}")
    if data["findings"]:
        print(f"\nFindings ({len(data['findings'])}):")
        for f in data["findings"]:
            print(f"  [{f['severity']}] {f['rule']} at {f['path']}:{f.get('line', '?')} — {f['message']}")
    return 0


def _cmd_analyze(args) -> int:
    """Execute CIG queries."""
    from orchestrata_ultimate.refactor.codemod.workspace import CodemodWorkspace

    root = Path(os.environ.get("CODEMOD_ROOT", "."))
    ws = CodemodWorkspace.load(root, parallel=True)
    cig = ws._cig

    if args.query in ("dependencies", "dependents", "blast_radius"):
        if not args.module:
            print(f"Error: --module required for {args.query}", file=sys.stderr)
            return 2
        node_id = cig._node_ids.get(args.module)
        if node_id is None:
            print(f"Error: module '{args.module}' not found", file=sys.stderr)
            return 1

    if args.query == "dependencies":
        deps = cig.dependencies(node_id)
        print(f"Dependencies of {args.module}:")
        for d in deps:
            print(f"  - {cig._nodes[d].qualified_name}")

    elif args.query == "dependents":
        deps = cig.dependents(node_id)
        print(f"Dependents of {args.module}:")
        for d in deps:
            print(f"  - {cig._nodes[d].qualified_name}")

    elif args.query == "blast_radius":
        blast = cig.blast_radius(node_id, hops=args.hops)
        print(f"Blast radius of {args.module} ({args.hops} hops):")
        print(f"  Affected nodes: {len(blast.affected_nodes)}")
        print(f"  Affected files: {len(blast.affected_files)}")
        print(f"  Impact score:   {blast.impact_score:.2f}")

    elif args.query == "cycles":
        cycles = cig.cycles()
        if cycles:
            print(f"Import cycles found: {len(cycles)}")
            for cycle in cycles:
                names = [cig._nodes[nid].qualified_name for nid in cycle]
                print(f"  {' -> '.join(names)}")
        else:
            print("No import cycles found.")

    elif args.query == "topological_order":
        order = cig.topological_order()
        print(f"Topological order ({len(order)} modules):")
        for nid in order[:20]:
            print(f"  - {cig._nodes[nid].qualified_name}")
        if len(order) > 20:
            print(f"  ... and {len(order) - 20} more")

    return 0


def _resolve_pass(name: str, args: dict):
    """Resolve a pass by name and construct with args."""
    from orchestrata_ultimate.refactor.codemod.passes import (
        RenameSymbolPass,
        ReplaceStringLiteralPass,
        AddImportPass,
        RemoveUnusedImportPass,
        DocstringSweepPass,
    )

    registry = {
        "rename-symbol": RenameSymbolPass,
        "replace-string": ReplaceStringLiteralPass,
        "add-import": AddImportPass,
        "remove-unused-import": RemoveUnusedImportPass,
        "docstring-sweep": DocstringSweepPass,
    }

    cls = registry.get(name)
    if cls is None:
        print(f"Error: unknown pass '{name}'. Available: {', '.join(registry)}", file=sys.stderr)
        sys.exit(2)

    return cls(**args)


if __name__ == "__main__":
    sys.exit(main())

Key CLI requirements:

1. os.environ.get only — no bare os.environ["X"]. Use os.environ.get("CODEMOD_ROOT", ".") for repo root.
2. Absolute imports — all imports use orchestrata_ultimate.refactor.codemod.*, never relative.
3. Dry-run default — run subcommand defaults to dry_run=True. --apply enables writes.
4. Exit codes — 0 (success/clean dry-run), 1 (review-abort or verification failure), 2 (usage error). Asserted by subprocess tests.
5. --auto-start — if daemon is running, connect via Unix socket. If not running and --auto-start is set, start daemon and connect. If not running and --auto-start is NOT set, fall back to library mode and emit a warning to stderr.
6. Report output — every run writes a CommitReport JSON to reports/codemod-{correlation_id}.json and prints the path.

Step 3: Implement test_cli.py

Write BDD scenarios from specs/110-codemod-engine/features/cli.feature. Tests use subprocess.run to invoke the CLI and assert exit codes and output.

Key test patterns:

import subprocess
import sys
from pathlib import Path

def run_cli(*args, cwd: Path) -> subprocess.CompletedProcess:
    """Run the CLI as a subprocess."""
    return subprocess.run(
        [sys.executable, "scripts/codemod.py", *args],
        capture_output=True,
        text=True,
        cwd=cwd,
    )

# Scenario: load-bench measures workspace load performance
def test_load_bench(codemod_repo):
    result = run_cli("load-bench", cwd=codemod_repo)
    assert result.returncode == 0
    assert "Load time (serial)" in result.stdout
    assert "Files loaded" in result.stdout
    assert "RSS delta" in result.stdout

# Scenario: run — dry-run default produces diff and report
def test_run_dry_run(codemod_repo):
    result = run_cli(
        "run", "--pass", "replace-string",
        "--arg", "old=function_a", "--arg", "new=function_renamed",
        cwd=codemod_repo,
    )
    assert result.returncode == 0
    assert "Commit report:" in result.stdout
    # Verify no files modified on disk
    original = (codemod_repo / "src" / "module_a.py").read_text()
    assert "function_a" in original  # unchanged

# Scenario: Exit code 2 — usage error
def test_usage_error(codemod_repo):
    result = run_cli("run", "--invalid-flag", cwd=codemod_repo)
    assert result.returncode == 2

Important: CLI tests must use subprocess.run, not import-and-call. This validates exit codes, argument parsing, and stdout/stderr separation.

Step 4: Implement test_spec109_interop.py

Write BDD scenarios from specs/110-codemod-engine/features/spec109_interop.feature.

Key test patterns:

from orchestrata_ultimate.refactor.rename.models import WorkspaceEditPlan
from orchestrata_ultimate.refactor.codemod.passes import RenameSymbolPass

# Scenario: RenameSymbolPass consumes WorkspaceEditPlan
def test_rename_symbol_pass(codemod_repo, sample_workspace):
    plan = WorkspaceEditPlan(...)  # rename "OldName" to "NewName" across 3 files
    pass_ = RenameSymbolPass(plan)
    changes = pass_.apply(sample_workspace)
    assert changes.file_count == 3
    for entry in changes.entries:
        assert "NewName" in entry.mutated_tree.code
        assert "OldName" not in entry.mutated_tree.code

# Scenario: No second rename resolver
def test_no_second_resolver():
    import ast
    import inspect
    source = inspect.getsource(RenameSymbolPass)
    tree = ast.parse(source)
    # Verify no import resolution logic (no SymbolRenamer, no resolve_* calls)
    for node in ast.walk(tree):
        if isinstance(node, ast.Attribute) and node.attr.startswith("resolve"):
            raise AssertionError("RenameSymbolPass contains rename resolution logic")

# Scenario: Identical results between EditApplier and RenameSymbolPass
def test_identical_results(codemod_repo, sample_workspace):
    plan = WorkspaceEditPlan(...)
    # Apply via EditApplier (SPEC-109)
    applier_result = apply_via_edit_applier(plan, codemod_repo)
    # Apply via RenameSymbolPass (SPEC-110)
    pass_ = RenameSymbolPass(plan)
    changes = pass_.apply(sample_workspace)
    codemod_result = {e.path: e.mutated_tree.code for e in changes.entries}
    # Byte-for-byte identical
    for path in applier_result:
        assert applier_result[path] == codemod_result[path]

Step 5: Implement test_benchmark_load.py

Plain pytest tests (not BDD) measuring performance targets from the spec.

import time
import psutil
from pathlib import Path
from orchestrata_ultimate.refactor.codemod.workspace import CodemodWorkspace

# REQ-110-003: Serial parse <360s for full codebase
def test_serial_parse_time(real_codebase_root):
    t0 = time.monotonic()
    ws = CodemodWorkspace.load(real_codebase_root, parallel=False)
    elapsed = time.monotonic() - t0
    assert elapsed < 360, f"Serial parse took {elapsed:.1f}s (target: <360s)"

# REQ-110-003: Parallel parse <120s on 12 cores
def test_parallel_parse_time(real_codebase_root):
    t0 = time.monotonic()
    ws = CodemodWorkspace.load(real_codebase_root, parallel=True)
    elapsed = time.monotonic() - t0
    assert elapsed < 120, f"Parallel parse took {elapsed:.1f}s (target: <120s)"

# REQ-110-004: RSS delta <3 GiB
def test_rss_delta(real_codebase_root):
    process = psutil.Process()
    rss_before = process.memory_info().rss
    ws = CodemodWorkspace.load(real_codebase_root, parallel=True)
    rss_after = process.memory_info().rss
    delta_gib = (rss_after - rss_before) / (1024 ** 3)
    assert delta_gib < 3.0, f"RSS delta {delta_gib:.2f} GiB (target: <3 GiB)"

# REQ-110-024: CIG queries <2ms
def test_cig_query_latency(real_codebase_root):
    ws = CodemodWorkspace.load(real_codebase_root, parallel=True)
    cig = ws._cig
    node_id = next(iter(cig._node_ids.values()))
    # Warm up
    cig.dependencies(node_id)
    # Measure
    t0 = time.monotonic()
    for _ in range(100):
        cig.dependencies(node_id)
    avg_ms = (time.monotonic() - t0) / 100 * 1000
    assert avg_ms < 2.0, f"Avg query latency {avg_ms:.2f}ms (target: <2ms)"

Note: Benchmark tests use real_codebase_root fixture pointing to the actual orchestrata-ultimate repo (~2,500 files). For CI environments without the full repo, use codemod_repo (the small fixture) and scale targets proportionally.

Step 6: Wire --auto-start Daemon Support

The CLI must support connecting to a running daemon or starting one:

def _try_daemon_client(root: Path, auto_start: bool):
    """Try to connect to daemon, optionally start it."""
    from orchestrata_ultimate.refactor.codemod.client import CodemodClient

    socket_path = root / ".codemod_daemon" / "socket"
    if socket_path.exists():
        try:
            return CodemodClient.connect(root)
        except ConnectionError:
            pass

    if auto_start:
        import subprocess
        subprocess.run(
            [sys.executable, "scripts/codemod.py", "daemon", "start"],
            cwd=root,
            check=True,
        )
        # Wait for socket
        for _ in range(50):  # 5s timeout
            if socket_path.exists():
                return CodemodClient.connect(root)
            time.sleep(0.1)
        print("Warning: daemon failed to start, falling back to library mode", file=sys.stderr)
    else:
        print("Warning: daemon not running, using library mode (slow). Use --auto-start to start daemon.", file=sys.stderr)

    return None

Note: The daemon itself (daemon start|stop|status|restart subcommands) is implemented in a later wave. For W3, the CLI should gracefully fall back to library mode when no daemon is available.

Step 7: Run Gate Checks

# Lint
uv run ruff check scripts/codemod.py src/orchestrata_ultimate/refactor/codemod/passes.py
uv run black --check scripts/codemod.py src/orchestrata_ultimate/refactor/codemod/passes.py

# Test
uv run pytest tests/refactor/codemod/test_cli.py -v -o addopts=""
uv run pytest tests/refactor/codemod/test_spec109_interop.py -v -o addopts=""
uv run pytest tests/refactor/codemod/test_benchmark_load.py -v -o addopts=""

# Coverage
uv run pytest tests/refactor/codemod/test_cli.py tests/refactor/codemod/test_spec109_interop.py \
  --cov=orchestrata_ultimate.refactor.codemod.passes \
  --cov-report=term-missing \
  -o addopts=""

# Standards
PYTHONPATH=src uv run python scripts/serialize_codebase_ast.py standards
PYTHONPATH=src uv run python scripts/test-review.py tests/refactor/codemod/test_cli.py --stdout
PYTHONPATH=src uv run python scripts/test-review.py tests/refactor/codemod/test_spec109_interop.py --stdout

Success Criteria

• [ ] scripts/codemod.py CLI works with all 4 subcommands (load-bench, run, report, analyze)
• [ ] Exit codes: 0 (success), 1 (review-abort), 2 (usage error) — asserted by subprocess tests
• [ ] Dry-run is default; --apply writes to disk
• [ ] RenameSymbolPass consumes WorkspaceEditPlan from SPEC-109 without reimplementing resolution
• [ ] LSP text edits correctly converted to LibCST mutations via PositionProvider
• [ ] Byte-identical results between EditApplier (SPEC-109) and RenameSymbolPass (SPEC-110)
• [ ] All 7 CLI BDD scenarios pass
• [ ] All 4 SPEC-109 interop BDD scenarios pass
• [ ] Benchmark tests pass (serial <360s, parallel <120s, RSS <3 GiB, queries <2ms)
• [ ] CLI uses os.environ.get only (no bare os.environ["X"])
• [ ] CLI supports --auto-start flag for daemon (graceful fallback to library mode)
• [ ] All gates pass (ruff, black, standards, test-review)
• [ ] Coverage >=90% for passes.py (RenameSymbolPass additions)

Deliverables

1. scripts/codemod.py — CLI with 4 subcommands + daemon support
2. src/orchestrata_ultimate/refactor/codemod/passes.py — RenameSymbolPass + _LSPToCSTTransformer additions
3. tests/refactor/codemod/test_cli.py — 7 BDD scenarios
4. tests/refactor/codemod/test_spec109_interop.py — 4 BDD scenarios
5. tests/refactor/codemod/test_benchmark_load.py — 4 performance tests

Handoff to W4

W4 (Persistence) will implement:

• persistence/repository.py — BoltSyncCodemodRepository (graph DB writes)
• persistence/records.py — CodemodRunRecord, CodemodPassRecord, CodemodFindingRecord, CodeChangeRecord
• observability/graph_writer.py — GraphWriterObserver (async batched writes)
• canonical_schema.py additions — new labels (CodemodRun, CodemodPass, CodemodFinding, CodeChange) and relationships
• tests/refactor/codemod/test_graph_persistence.py — 5 BDD scenarios

W4 agents will use the CLI, passes, and workspace from W1+W2+W3 to implement graph database persistence.

Notes

• Single agent: W3 uses one agent (Qwen Max) because it integrates components from both W1 and W2. Parallel execution would create merge conflicts in passes.py.
• Uses all W1+W2 components: The CLI exercises CodemodWorkspace.load(), workspace.run(), workspace.commit(), CIG queries, and transform passes.
• SPEC-109 is frozen: Never modify refactor/rename/*. The RenameSymbolPass only imports from it.
• CLI must support --auto-start: Even though the daemon is implemented later, the CLI must gracefully handle the daemon-not-running case.
• Benchmark tests are plain pytest: Performance tests don't benefit from BDD Given/When/Then. Use direct assertions with timing measurements.
• Feature files are already written: cli.feature (7 scenarios) and spec109_interop.feature (4 scenarios) exist in specs/110-codemod-engine/features/. Implement step definitions to match them.
• All @traces decorators: RenameSymbolPass must carry @traces(spec="110", story="US-110-009", reqs=["REQ-110-020", "REQ-110-021"]).
• No mocks: All tests use real files in tmp_path, real git init repos, real WorkspaceEditPlan objects. Function-scope fixtures only.

Related

• Plans
• Orchestrata Ultimate — project index
• Workflow
• Tasks