Orchestrata Ultimate / Specs / 110-codemod-engine

110-codemod-engine: In-Memory Code Intelligence Platform

SPEC-110 defines a complete in-memory code intelligence platform that loads an entire Python codebase as a lossless Concrete Syntax Tree (CST) forest, builds a multihop-queryable Code Intelligence Graph (CIG), executes composable transform and review passes entirely in RAM, and commits changes atomically with full rollback capability. The platform observes all operations through an event-driven architecture, persists provenance to a graph database via the repository pattern, and...

At a glance

Reading this spec

This spec defines a complete in-memory code intelligence platform for Python. It loads a whole codebase as a lossless Concrete Syntax Tree (CST) forest, builds a Code Intelligence Graph (CIG) that supports multihop queries, runs composable transform and review passes entirely in RAM, and commits changes atomically with full rollback. The spec says it is more than a codemod engine: every analysis, review, graph query and transform works on the same in-memory representation, and nothing touches disk until the final commit. The reason is the gw to slot incident of spec 109, where a text-based, file-by-file rename missed four Redis slot parsers and caused 54 test failures and more than a day of recovery. The project runs on free-threaded Python 3.14, where a measured 2.66x parallel speedup on 4 threads makes parallel parsing and per-file passes worthwhile.

Work happens in three phases. Load and index parses every .py file into a LibCST module with byte-perfect round trip, builds the CIG (modules, classes, functions, methods, parameters, tests and imports, linked by imports, calls, inheritance, definition, tested-by and override edges) and builds compressed sparse row adjacency arrays. Transform and review runs passes over the forest, then runs review passes over the mutated forest so verdicts come before bytes, emits events to observers, and aborts the commit on any error-severity finding. Commit snapshots the pre-commit bytes, writes atomically through temporary files and rename, batches directory fsyncs and re-parses to verify.

The user stories (US-110-001 onward) include a codebase-wide symbol rename, a vocabulary migration over 500 or more files, breaking-change detection, a 40-check in-memory review, composing transforms in one atomic commit, persisting provenance to the graph database, and a warm daemon. The 59 requirement IDs (REQ-110-NNN) cover loading and incremental index updates, the rule that a pass never writes to disk and write-back happens only through workspace.commit(), dry-run commits that produce a diff and review report (the CLI defaults to dry-run), five built-in observers, analysis passes for blast radius, breaking changes and complexity, the daemon with its JSON-RPC query methods, byte-splice edits that keep every byte outside an edit span identical, and quality gates with at least 90% coverage per file.

The implementation spans the codemod package, with the most citations in the workspace (17 requirements), followed by the transaction layer and the daemon. The requirements traceability matrix lists each ID. The spec's own status line reads Complete and it tracks 20 files. Measured coverage is 43 of 59 requirements cited by at least one @traces in src/ (72.9%), spread over 36 modules; the module citing the most of them is orchestrata_ultimate/refactor/codemod/workspace (17 requirements). In tests/, 28 requirements are cited by a test-side @traces, thinner than the code side (28 against 43). 16 requirements have no citing symbol; the behaviour may still exist in code that was never annotated.

Requirements

Checkbox is the state recorded in the requirements checklist; Traced by is the count of src/ symbols whose @traces names the ID.

Code that traces this spec

Checklists

checklists/constitutional.md

SPEC-110 Constitutional Compliance Checklist

This checklist verifies that SPEC-110: In-Memory Code Intelligence Platform complies with the project's 12 constitutional articles.

Status Legend:

• [ ] Not verified
• [~] Partially compliant
• [x] Fully compliant
• [!] Non-compliant (see notes)

Article 1: Human Autonomy

Principle: The system must preserve human control over critical decisions.

• [x] Dry-run default: CLI defaults to dry-run mode; human must explicitly pass --apply to write changes
• [x] Review-before-write: Review passes run before commit; human can inspect findings before applying
• [x] Rollback capability: Full rollback on any failure; human can revert changes
• [x] No automatic suggestions: Engine executes user-specified transforms, not AI-generated suggestions
• [x] Explicit pass selection: Human chooses which passes to run via CLI --pass flag

Compliance: ✅ Fully compliant

Article 2: Transparency

Principle: The system must make its operations visible and understandable.

• [x] CommitReport: Machine-readable JSON report with passes run, files written, findings, duration
• [x] Unified diff: Full diff of all changes included in report
• [x] Observability: 14 event types emitted via SyncObservable[CodemodEvent]
• [x] Tracing: Structured JSON spans via core.tracing.Tracer with correlation_id
• [x] Debug panel: NDJSON trace files for post-mortem analysis
• [x] Metrics: Counters and histograms for performance monitoring
• [x] Graph provenance: Full operation history persisted to graph database

Compliance: ✅ Fully compliant

Article 3: Harm Prevention

Principle: The system must prevent damage to code, data, and infrastructure.

• [x] Atomic writes: Tempfile + fsync + rename; no partial writes
• [x] Post-write verification: Re-parse every written file; rollback on failure
• [x] Concurrent-edit guard: mtime + content hash check prevents overwriting concurrent edits
• [x] Symlink attack prevention: O_NOFOLLOW flag on all file opens
• [x] Advisory locking: fcntl.flock(LOCK_EX | LOCK_NB) prevents concurrent writes
• [x] Path confinement: _resolve_in_repo() prevents path traversal attacks
• [x] No disk I/O in passes: Transform passes cannot write to disk (enforced by construction)
• [x] Review abort: Error-severity findings abort commit; nothing written

Compliance: ✅ Fully compliant

Article 4: Privacy

Principle: The system must respect user privacy and data confidentiality.

• [x] No telemetry without opt-in: Observers are optional and configurable
• [x] Local operation: All operations run locally; no external API calls (except graph DB if configured)
• [x] No user data collection: Engine operates on code, not user data
• [x] Correlation ID: UUID for linking events; no personally identifiable information
• [x] Graph DB opt-in: Graph persistence is optional; can be disabled

Compliance: ✅ Fully compliant

Article 5: Fairness

Principle: The system must treat all inputs and users equitably.

• [x] No bias in pass execution: Passes execute in user-specified order; no implicit prioritization
• [x] Deterministic behavior: Same inputs produce same outputs (no randomness)
• [x] No discrimination: Engine treats all files equally; no file-type bias
• [x] Fair resource allocation: ThreadPoolExecutor distributes work evenly across cores
• [x] No preferential treatment: All modules parsed and transformed with same algorithms

Compliance: ✅ Fully compliant

Article 6: Accountability

Principle: The system must enable tracing of actions to their origins.

• [x] Graph DB provenance: CodemodRun, CodemodPass, CodemodFinding, CodeChange nodes
• [x] Correlation ID: Links all events in a single codemod run
• [x] @traces decorator: Every public class/method tagged with REQ-110-NNN
• [x] CommitReport: Machine-readable record of all operations
• [x] Audit trail: Graph database records who ran what, when, and why
• [x] Observer pattern: All operations emit events for external auditing

Compliance: ✅ Fully compliant

Article 7: Security

Principle: The system must protect against malicious attacks and unauthorized access.

• [x] Path confinement: _resolve_in_repo() with Path.resolve() + relative_to() check
• [x] Symlink attack prevention: O_NOFOLLOW flag on all file opens
• [x] Advisory locking: fcntl.flock(LOCK_EX | LOCK_NB) prevents concurrent writes
• [x] No disk I/O in passes: Transform passes cannot write to disk (enforced by AST grep)
• [x] Parameterized Cypher: No string interpolation in graph DB queries (prevents injection)
• [x] Prolog fact isolation: Facts asserted from trusted ChangeSet, not user input
• [x] Threat model: Documented in spec.md §11.5

Compliance: ✅ Fully compliant

Article 8: Sustainability

Principle: The system must use resources efficiently and minimize environmental impact.

• [x] Demand-driven parse: Files parsed only when accessed; reduces RSS by ~80%
• [x] Parallel parse: ThreadPoolExecutor exploits 12 cores; 6x speedup
• [x] Per-file parallel dispatch: Transform passes with parallelism == "per_file" use all cores
• [x] Batch fsync: 5 directory fsyncs instead of 100 file fsyncs; 20x I/O reduction
• [x] CSR adjacency arrays: O(1) lookups; contiguous memory layout
• [x] Incremental index update: _update_index(paths) re-derives only mutated modules
• [x] RSS guard: Memory guard prevents excessive memory usage (< 4 GiB)

Compliance: ✅ Fully compliant

Article 9: Dignity

Principle: The system must respect human dignity and avoid degrading interactions.

• [x] No condescending error messages: Clear, actionable error messages
• [x] No blame: Errors attributed to system state, not user incompetence
• [x] Respectful CLI output: Professional, concise output; no emoji or slang
• [x] No manipulation: Engine does not nudge or coerce user decisions
• [x] N/A (no human-facing UI): CLI-only interface; no chatbot or conversational UI

Compliance: ✅ Fully compliant (N/A for most criteria)

Article 10: Rights

Principle: The system must respect user rights and intellectual property.

• [x] No user data collection: Engine operates on code, not user data
• [x] No surveillance: No tracking of user behavior or preferences
• [x] No content filtering: Engine transforms all code equally; no content-based discrimination
• [x] Respects licenses: Engine does not modify license files or copyright notices
• [x] N/A (no user data): Most criteria not applicable

Compliance: ✅ Fully compliant (N/A for most criteria)

Article 11: Continuous Improvement

Principle: The system must enable learning and improvement over time.

• [x] Observability: 14 event types enable performance tuning
• [x] Metrics: Counters and histograms identify bottlenecks
• [x] Debug panel: NDJSON trace files enable post-mortem analysis
• [x] Graph provenance: Historical data enables trend analysis
• [x] Benchmark tests: Performance budgets tracked over time
• [x] Coverage tracking: ≥90% per file ensures test quality
• [x] Gate commands: Automated quality gates prevent regression

Compliance: ✅ Fully compliant

Article 12: Plan Fidelity

Principle: The system must adhere to its stated specifications and requirements.

• [x] @traces decorator: Every public class/method tagged with REQ-110-NNN
• [x] 132 requirements: All REQ-110-NNN tracked in requirements.md checklist
• [x] 10 acceptance tests: TST-110-001 through TST-110-010 verify core functionality
• [x] 81 BDD scenarios: Gherkin feature files specify behavioral requirements
• [x] 18-section spec: spec.md follows SPEC_STANDARD.md template
• [x] Design document: plans/codemod-engine-design.md documents architecture decisions
• [x] Contract files: .mmd and .contract.json specify class structure and metadata

Compliance: ✅ Fully compliant

Summary

Overall Compliance: ✅ 12/12 articles fully compliant

Verification Methods

Automated Verification

• [x] @traces decorator: Verified by scripts/serialize_codebase_ast.py trace-lint
• [x] Requirements checklist: Verified by scripts/serialize_codebase_ast.py check-traces
• [x] Test coverage: Verified by pytest --cov (≥90% per file)
• [x] Gate commands: Verified by pre-commit hooks (ruff, black, standards, check-deps, test-review, pyright_ratchet)

Manual Verification

• [ ] Constitutional review: Human review of spec.md §12 (Governance) section
• [ ] Threat model review: Human review of spec.md §11.5 (Threat Model) table
• [ ] Observer review: Human review of observability architecture (spec.md §9.7)
• [ ] Graph provenance review: Human review of Cypher MERGE patterns (spec.md §9.8)

Non-Compliance Resolution

No non-compliance issues identified.

If non-compliance is discovered during implementation or review:

1. Document the issue: Add entry to this checklist with [!] status and description
2. Assess severity: Determine if issue is blocker or can be deferred
3. Propose fix: Update spec.md, design document, or implementation as needed
4. Verify fix: Re-run verification methods to confirm compliance
5. Update checklist: Change status to [x] and add resolution notes

References

• Constitutional articles: Defined in project governance documentation
• SPEC_STANDARD.md: Template for spec documentation (18 sections)
• SPEC_REVIEW_CHECKLIST.md: 19-section review checklist with IDs
• spec.md §11 (Security): Threat model and security requirements
• spec.md §12 (Governance): Prolog integration and constitutional compliance
• spec.md §15 (Observability): Event types, observers, tracing, metrics

checklists/requirements.md

The requirement list above comes from this file; 56 of 57 checkbox items are ticked.

checklists/testing.md

SPEC-110 Testing Checklist

This checklist maps test scenarios (TST-110-NNN) to Gherkin feature files and tracks implementation status.

Status Legend:

• [ ] Not started
• [~] In progress
• [x] Complete
• [!] Blocked (see notes)

Acceptance Tests (TST-110-001 through TST-110-010)

TST-110-001: Round-Trip Fidelity

Feature File: workspace_load.feature
Scenario: "Round-trip fidelity — module.code equals on-disk bytes"
Requirements: REQ-110-002

• [ ] Load real orchestrata-ultimate codebase (797 files)
• [ ] For every loaded module, verify module.code.encode('utf-8') == on_disk_bytes
• [ ] Unparseable files recorded in workspace.unparsable with path and error message
• [ ] Zero silent corruption

Implementation Notes:

• Use test_round_trip_fidelity BDD scenario
• Measure parse time (single-threaded and parallel)
• Record RSS delta via psutil.Process().memory_info().rss

TST-110-002: Comment Preservation

Feature File: workspace_load.feature
Scenario: "Comment preservation — only the mutated line changes"
Requirements: REQ-110-002

• [ ] Load fixture file dense with comments, blank lines, odd quoting
• [ ] Apply ReplaceStringLiteralPass changing one literal
• [ ] Unified diff contains ONLY the line with that literal
• [ ] All comments and blank lines preserved unchanged

Implementation Notes:

• Use test_comment_preservation BDD scenario
• Verify LibCST preserves EmptyLine nodes with comment field
• Compare diff output line-by-line

TST-110-003: Compose Multiple Passes

Feature File: transform_passes.feature
Scenario: "Compose — rename + sweep + add-import in one run"
Requirements: REQ-110-008, REQ-110-009

• [ ] Run three passes in order: RenameSymbolPass, DocstringSweepPass, AddImportPass
• [ ] Composed ChangeSet contains entries for all mutated modules
• [ ] All three changes visible in final forest
• [ ] One CommitReport covers all three passes

Implementation Notes:

• Use test_compose_multiple_passes BDD scenario
• Verify workspace.run(passes) applies passes sequentially
• Each pass sees previous pass's mutations

TST-110-004: Review-Abort on Banned Construct

Feature File: review_before_write.feature
Scenario: "BannedConstructs — bare os.environ aborts commit"
Requirements: REQ-110-012, REQ-110-013

• [ ] Pass introduces os.environ["X"] into mutated tree
• [ ] BannedConstructs review produces error finding for rule "BARE_OS_ENVIRON"
• [ ] Commit aborted
• [ ] No files written to disk

Implementation Notes:

• Use test_review_abort_banned_construct BDD scenario
• Verify workspace.commit(reviews=[...]) runs reviews FIRST
• Error-severity finding aborts commit

TST-110-005: Rollback on Post-Write Parse Failure

Feature File: transactional_writeback.feature
Scenario: "Rollback — post-write parse failure restores all files"
Requirements: REQ-110-014

• [ ] Pass mutates 3 files, one of which will fail post-write parse
• [ ] Commit raises verification error
• [ ] All 3 files restored to pre-commit bytes exactly
• [ ] Byte comparison proves identical restoration

Implementation Notes:

• Use test_rollback_post_write_failure BDD scenario
• Snapshot pre-commit bytes in ChangeEntry.original_bytes
• Restore via Path.write_bytes(original_bytes) on failure

TST-110-006: Concurrent-Edit Guard

Feature File: transactional_writeback.feature
Scenario: "Concurrent-edit guard — file changed after load"
Requirements: REQ-110-015

• [ ] Pass mutates file A
• [ ] File A modified on disk after workspace loaded
• [ ] Commit refused
• [ ] Refusal message names stale path

Implementation Notes:

• Use test_concurrent_edit_guard BDD scenario
• Capture mtime + content hash (xxhash3_64) at load time
• Check before write, refuse if changed

TST-110-007: Incremental Index Update

Feature File: workspace_index.feature
Scenario: "Incremental index — add-import updates dependents without reload"
Requirements: REQ-110-006

• [ ] Workspace loaded with modules A and B
• [ ] Apply AddImportPass adding "import B" to module A
• [ ] dependents(B) includes A without reloading workspace

Implementation Notes:

• Use test_incremental_index_update BDD scenario
• _update_index(paths) re-derives dependencies for specified modules only
• CSR adjacency arrays updated incrementally

TST-110-008: Benchmark — Full-Repo Load

Feature File: workspace_load.feature
Scenario: "Parallel parse completes within time budget"
Requirements: REQ-110-003, REQ-110-004, REQ-110-005

• [ ] Load real orchestrata-ultimate source tree
• [ ] Parallel parse completes in < 40 seconds (12 cores)
• [ ] Forest contains at least 700 modules
• [ ] RSS delta < 4 GiB

Implementation Notes:

• Use test_benchmark_full_repo_load BDD scenario
• Measure with time.perf_counter() and psutil.Process().memory_info().rss
• Compare against single-threaded baseline (< 120 seconds)

TST-110-009: CLI — Subprocess Dry-Run and Apply

Feature File: cli.feature
Scenario: "Dry-run produces diff and report without writing"
Requirements: REQ-110-018, REQ-110-019

• [ ] python scripts/codemod.py run --pass replace-string --arg old=NEO4J --arg new=GRAPH (dry-run default)
• [ ] Exit code 0
• [ ] Diff printed to stdout
• [ ] CommitReport JSON written to reports/codemod-*.json
• [ ] No files written to disk

Implementation Notes:

• Use test_cli_dry_run BDD scenario
• Verify exit codes: 0 success, 1 review-abort, 2 usage error
• Parse CommitReport JSON and verify structure

TST-110-010: SPEC-109 Interop

Feature File: spec109_interop.feature
Scenario: "RenameSymbolPass consumes WorkspaceEditPlan"
Requirements: REQ-110-020, REQ-110-021

• [ ] Generate WorkspaceEditPlan from SPEC-109 SymbolRenamer
• [ ] Apply via RenameSymbolPass in codemod engine
• [ ] Apply same plan via SPEC-109 EditApplier
• [ ] Results identical (byte-for-byte comparison)

Implementation Notes:

• Use test_spec109_interop BDD scenario — RETIRED 2026-10-20 with the SPEC-109 move (orchestrata ADR-0117/ADR-0118): the scenario needed EditApplier and RenameSymbolPass in one process, which the move makes impossible, and it was red in BOTH repos (orchestrata: producer deleted; codetools: transitively requires the undeclared orchestrata_ast). The four checkboxes above are therefore unverifiable in-process and stay unchecked; the owed replacement is a cross-repo contract test through the deployed codetools CLI, gated on the queued console-script//opt/codetools/bin exposure step. See specs/110-codemod-engine/spec.md US-110-009.
• Verify no second rename resolver in codemod engine
• Cross-check with the EditApplier scenarios, which moved to codetools: tests/refactor/rename/test_applier.py there (was tests/refactor/rename/test_applier.py here, deleted 2026-10-20)

FR-013: Daemon Architecture

TST-110-011: Daemon Lifecycle

Feature File: daemon_lifecycle.feature Scenario: "Daemon start creates socket, writes PID, parses workspace; stop shuts down gracefully" Requirements: REQ-110-013

• [ ] Start daemon, verify socket created, PID written, workspace parsed
• [ ] Stop daemon, verify graceful shutdown

Implementation Notes:

• Use test_daemon_lifecycle BDD scenario
• Verify PID file contains correct process ID
• Verify Unix socket exists at expected path
• Verify workspace fully parsed during startup

TST-110-012: IPC Protocol

Feature File: ipc_protocol.feature Scenario: "JSON-RPC request via Unix socket returns correct response within latency budget" Requirements: REQ-110-013

• [ ] Send JSON-RPC request via Unix socket, verify response format and latency <100ms

Implementation Notes:

• Use test_ipc_protocol BDD scenario
• Verify JSON-RPC 2.0 envelope (id, result/error fields)
• Measure round-trip latency with time.perf_counter()

TST-110-013: File Watching

Feature File: file_watching.feature Scenario: "inotify triggers incremental reload within 500ms of file modification" Requirements: REQ-110-013

• [ ] Modify file, verify inotify triggers incremental reload in <500ms

Implementation Notes:

• Use test_file_watching BDD scenario
• Use inotify or watchdog for filesystem events
• Measure time from modification to reload completion

TST-110-014: Concurrency

Feature File: concurrency.feature Scenario: "Parallel read requests succeed; write requests are serialized" Requirements: REQ-110-013

• [ ] 10 clients send parallel read requests, verify all succeed
• [ ] 2 clients send write requests, verify serialized

Implementation Notes:

• Use test_concurrency BDD scenario
• Use ThreadPoolExecutor for parallel client simulation
• Verify write serialization via lock or queue ordering

Feature File Inventory

Total: 16 feature files, ~126 scenarios

Non-BDD Tests

Note: Daemon tests (TST-110-011 through TST-110-014) are BDD scenarios, not plain pytest. They live in the feature files above, not in this table.

Total: 15 plain pytest tests

Coverage Targets

REQ-110-132: Coverage ≥90% per file

pytest tests/refactor/codemod/ \
  --cov=orchestrata_ultimate.refactor.codemod \
  --cov-report=term-missing \
  -o addopts="" \
  -p orchestrata_ultimate.core.threading.pytest_live_observer

Coverage Status:

• [ ] workspace.py ≥90%
• [ ] passes.py ≥90%
• [ ] review.py ≥90%
• [ ] txn.py ≥90%
• [ ] report.py ≥90%
• [ ] _cig.py ≥90%
• [ ] _cig_builder.py ≥90%
• [ ] _csr_index.py ≥90%
• [ ] analysis/*.py ≥90%
• [ ] observability/*.py ≥90%
• [ ] persistence/*.py ≥90%

Gate Commands

All gates must pass before every commit:

.venv/bin/ruff check src tests scripts
.venv/bin/black --check src/orchestrata_ultimate/refactor/codemod tests/refactor/codemod scripts/codemod.py
PYTHONPATH=src .venv/bin/python scripts/serialize_codebase_ast.py standards
PYTHONPATH=src .venv/bin/python scripts/serialize_codebase_ast.py check-deps --include-tests
PYTHONPATH=src .venv/bin/python scripts/test-review.py tests/refactor/codemod/ --stdout
PYTHONPATH=src .venv/bin/python scripts/pyright_ratchet.py --max 283

Gate Status:

• [ ] ruff check clean
• [ ] black --check clean
• [ ] standards clean
• [ ] check-deps clean
• [ ] test-review 0 issues
• [ ] pyright_ratchet ≤283 errors

Test Infrastructure

BDD Configuration

Feature file location: specs/110-codemod-engine/features/

Step definitions: tests/refactor/codemod/conftest.py (shared) + per-test-file (specific)

Scenario binding: @scenario per scenario (explicit, with TST/REQ docstrings)

Tags: @REQ_110_NNN, @TST_110_NNN, @FR_NNN on scenarios (registered as pytest markers)

Fixtures

# tests/refactor/codemod/conftest.py

@pytest.fixture
def codemod_repo(tmp_path: Path) -> Path:
    """Real git repository under tmp_path with src/ layout."""
    subprocess.run(["git", "init", "-q"], cwd=tmp_path, check=True)
    subprocess.run(["git", "config", "user.email", "test@example.com"], cwd=tmp_path, check=True)
    subprocess.run(["git", "config", "user.name", "Test"], cwd=tmp_path, check=True)
    return tmp_path

@pytest.fixture
def fixture_package(codemod_repo: Path) -> Callable[[Mapping[str, str]], Path]:
    """Factory: write src/-rooted files, return repo root."""
    def make(files: Mapping[str, str]) -> Path:
        src_root = codemod_repo / "src"
        for relative, text in files.items():
            target = src_root / relative
            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_text(text, encoding="utf-8")
        return codemod_repo
    return make

Shared Step Definitions

# tests/refactor/codemod/conftest.py

from pytest_bdd import given, when, then, parsers

@given(parsers.parse("the fixture package contains {count:d} Python files"), target_fixture="repo_root")
def fixture_with_files(fixture_package, count):
    files = {}
    for i in range(count):
        files[f"pkg/mod_{i}.py"] = f"# Module {i}\nx_{i} = {i}\n"
    files["pkg/__init__.py"] = ""
    return fixture_package(files)

@when("I load the workspace", target_fixture="workspace")
def workspace_loaded(repo_root):
    return CodemodWorkspace.load(repo_root, include=("src",))

@then(parsers.parse("the forest contains exactly {count:d} modules"))
def forest_count(workspace, count):
    assert len(workspace.forest) == count

Test Execution

Running BDD Tests

# Run all BDD scenarios
pytest tests/refactor/codemod/ -v

# Run specific feature file
pytest tests/refactor/codemod/test_workspace_load.py -v

# Run scenarios with specific tag
pytest tests/refactor/codemod/ -m REQ_110_001 -v

# Run with Gherkin terminal reporter
pytest tests/refactor/codemod/ --gherkin-terminal-reporter

Running Non-BDD Tests

# Run benchmark tests
pytest tests/refactor/codemod/test_benchmark_load.py -v

# Run CSR index unit tests
pytest tests/refactor/codemod/test_csr_index.py -v

# Run C extension parity tests
pytest tests/refactor/codemod/test_c_extension.py -v

Running Full Suite

# Via batched executor (recommended)
python scripts/run_tests.py tests/refactor/codemod/ --skip-gates

# Via streaming runner (development)
python scripts/run_tests_streaming.py tests/refactor/codemod/ -j 8

Implementation Notes

Test Organization

• BDD tests (126 scenarios): Behavioral specification, living documentation
• Non-BDD tests (15 tests): Performance benchmarks, data structure correctness, C extension parity
• Total: 141 tests

Test Data

• Fixture packages: Small synthetic codebases under tmp_path (5-20 files)
• Real codebase: orchestrata-ultimate (797 files, 9.2 MB) for benchmark and round-trip tests
• Git repositories: Real git init repos for transactional write-back tests

Test Isolation

• Filesystem: tmp_path fixture (pytest built-in)
• Git: Real git init in tmp_path, no shared state
• Graph DB: In-memory InMemoryCodemodRepository for unit tests, real BoltSyncCodemodRepository for integration tests
• Prolog: Predicate-prefix isolation via test_id (UUID fixture)

Test Parallelism

• BDD scenarios: Run via batched executor (scripts/run_tests.py) or streaming runner (scripts/run_tests_streaming.py)
• No pytest-xdist: Banned repo-wide (deadlocks ~78% of the time)
• Per-file parallel dispatch: Transform passes with parallelism == "per_file" use ThreadPoolExecutor internally

Deferred Tests

• Property-based testing (Hypothesis): CSR index and CIG query correctness with random graphs — deferred to Phase 2
• Fuzz testing: Parser and transformer with malformed/adversarial inputs — deferred to Phase 2
• Integration tests with real graph DB: Full pipeline including BoltSyncCodemodRepository — deferred to Wave 4
• Incremental codemod tests: Re-running on changed codebase — deferred to Phase 2

Success Criteria

• [ ] All 14 acceptance tests (TST-110-001 through TST-110-014) pass
• [ ] All 126 BDD scenarios pass
• [ ] All 15 non-BDD tests pass
• [ ] Coverage ≥90% per file
• [ ] All gates pass (ruff, black, standards, check-deps, test-review, pyright_ratchet)
• [ ] Test execution time < 5 minutes (via batched executor)
• [ ] Zero test flakes (no retry needed)

Other tracked files

• specs/110-codemod-engine/features/analysis_passes.feature
• specs/110-codemod-engine/features/cli.feature
• specs/110-codemod-engine/features/code_intelligence_graph.feature
• specs/110-codemod-engine/features/concurrency.feature
• specs/110-codemod-engine/features/daemon_lifecycle.feature
• specs/110-codemod-engine/features/file_watching.feature
• specs/110-codemod-engine/features/graph_persistence.feature
• specs/110-codemod-engine/features/in_memory_review.feature
• specs/110-codemod-engine/features/ipc_protocol.feature
• specs/110-codemod-engine/features/observability.feature
• specs/110-codemod-engine/features/review_before_write.feature
• specs/110-codemod-engine/features/spec109_interop.feature
• specs/110-codemod-engine/features/transactional_writeback.feature
• specs/110-codemod-engine/features/transform_passes.feature
• specs/110-codemod-engine/features/workspace_index.feature
• specs/110-codemod-engine/features/workspace_load.feature

Full spec text

Verbatim from specs/110-codemod-engine/spec.md; headings are pushed down two levels.

SPEC-110: In-Memory Code Intelligence Platform

Metadata

• Spec ID: 110-codemod-engine
• Title: In-Memory Code Intelligence Platform (Lossless CST Forest, Multihop Graph Queries, Transactional Write-Back)
• Author: Mohamed (with Claude Opus 4.6 orchestration)
• Status: Complete
• Created: 2026-07-27
• Last Updated: 2026-07-27
• Priority: P0
• Version: 2.0.0
• Extends: SPEC-109 (rename-engine)
• Components: refactor.codemod, refactor.codemod.analysis, refactor.codemod.observability, refactor.codemod.persistence
• Dependencies: libcst>=1.8.6, xxhash>=3.0.0, psutil>=7.2.2, janus-swi>=1.5.3

1. Overview

What

SPEC-110 defines a complete in-memory code intelligence platform that loads an entire Python codebase as a lossless Concrete Syntax Tree (CST) forest, builds a multihop-queryable Code Intelligence Graph (CIG), executes composable transform and review passes entirely in RAM, and commits changes atomically with full rollback capability. The platform observes all operations through an event-driven architecture, persists provenance to a graph database via the repository pattern, and integrates Prolog-based semantic review for dependency safety analysis.

This is not merely a codemod engine. It is a code intelligence substrate that happens to also transform code. Every analysis, every review, every graph query, every transform pass operates on the same in-memory representation. Nothing touches disk until the final atomic commit.

Why

The project's existing rename engine (SPEC-109) demonstrated that half-applied refactors are catastrophic: a gw→slot rename that missed 4 Redis slot parsers produced 54 test failures and required >1 day of manual recovery across 3 entangled waves. The root cause was text-based, file-by-file editing without a unified view of the codebase.

Existing tools fail to solve this:

• jscodeshift (JavaScript): Single-threaded, AST-based (loses comments/formatting), no graph queries
• OpenRewrite (JVM): Heavy runtime, Java-centric, no Prolog integration, no free-threading
• ast-grep (Rust): Read-only pattern matching, no transforms, no transactional writes
• LibCST codemods (Python): Single-threaded, no graph intelligence, no observability

The project runs on Python 3.14.6 free-threaded with the GIL genuinely disabled (measured 2.66x parallel speedup on 4 threads, 12 CPU cores available). This is an unfair advantage no other codemod engine possesses. SPEC-110 exploits this with parallel CST parsing, per-file parallel transform passes, and concurrent observer dispatch.

How

The platform operates in three phases:

Phase 1 — Load & Index (RAM-only)

1. Parse all .py files under configured roots into LibCST Module trees (lossless, byte-perfect round-trip)
2. Build the Code Intelligence Graph (CIG): nodes for Module/Class/Function/Method/Parameter/Test/Import, edges for IMPORTS/CALLS/INHERITS/DEFINES/TESTED_BY/OVERRIDES
3. Construct CSR (Compressed Sparse Row) adjacency arrays for O(1) dependency queries
4. Index symbols, imports, and call sites for multihop traversal

Phase 2 — Transform & Review (RAM-only)

1. Execute composable transform passes over the CST forest (parallel per-file where safe)
2. Execute review passes over the mutated forest (verdicts before bytes)
3. Emit CodemodEvent observations to all subscribed observers
4. If any review pass produces an error-severity finding, abort the commit

Phase 3 — Commit (disk I/O)

1. Snapshot pre-commit bytes for all mutated files
2. Write mutated trees to disk via atomic tempfile + os.rename
3. Batch fsync directory file descriptors (not per-file)
4. Post-write re-parse verification
5. If verification fails, restore all files from snapshot (byte-exact rollback)
6. Persist provenance to graph database via CodemodGraphRepository

Non-Goals

• Repos >10,000 files: The platform targets codebases up to ~5,000 files. The current codebase (2,500 files) fits comfortably. Larger repos require sharding or incremental parse (deferred to Phase 2).
• Cross-language support: Python-only. Tree-sitter integration for multi-language is out of scope.
• IDE integration: No LSP server or editor plugins. CLI-only interface.
• Real-time collaboration: No CRDT or operational transform for concurrent editors.
• Automatic refactoring suggestions: The platform executes user-specified transforms, not AI-generated suggestions.

2. Terminology

3. User Scenarios

US-110-001: Rename a symbol across the entire codebase with full safety

As a developer performing a large-scale rename
I want to rename a class, function, or variable and have every reference updated atomically
So that I never produce a half-applied rename that breaks tests

Acceptance Criteria:

• The rename pass consumes a WorkspaceEditPlan from SPEC-109's SymbolRenamer
• All references (imports, call sites, type annotations, string literals) are updated
• The CIG detects breaking changes (downstream consumers with incompatible signatures)
• Review passes verify no banned constructs are introduced
• The commit is atomic: either all files are updated, or none are
• The graph database records the rename operation with full provenance

US-110-002: Apply a vocabulary migration across 500+ files

As a developer migrating from neo4j to graph terminology
I want to replace all occurrences of "Neo4j" with "Graph" in code, comments, and docstrings
So that the codebase reflects the technology-agnostic naming policy

Acceptance Criteria:

• The ReplaceStringLiteralPass handles exact and regex modes
• The DocstringSweepPass updates vocabulary in docstrings and comments (LibCST preserves these)
• Per-file parallel dispatch exploits 12 cores for <3s execution on 797 files
• The BannedConstructs review pass catches any accidental os.environ["NEO4J_URI"] introductions
• The commit produces a unified diff showing all changes
• The graph database records the migration with file-level granularity

US-110-003: Detect breaking API changes before committing

As a library maintainer changing a public function signature
I want to know which downstream consumers will break
So that I can update them in the same commit or defer the change

Acceptance Criteria:

• The BreakingChangeDetector analyzes signature diffs (removed/added/reordered parameters, changed return types)
• The CIG's blast_radius() method computes transitive dependents via CALLS + INHERITS edges
• Findings include the downstream consumer's file path, line number, and the specific incompatibility
• Error-severity findings abort the commit
• The graph database records the breaking change analysis with proof trees

US-110-004: Run 40-check in-memory code review without subprocess calls

As a CI pipeline running pre-commit checks
I want to execute all style, quality, safety, and correctness checks in-process
So that review completes in <500ms instead of 5s with subprocess overhead

Acceptance Criteria:

• All 40 checks operate on LibCST trees (no ast.parse, no subprocess.run)
• Checks include: complexity, nesting, style, quality, safety, correctness, test isolation
• The PrologSemanticReview check asserts mutations as Prolog facts and queries codemod_review.pl
• Findings are emitted as CodemodEvent observations
• Error-severity findings abort the commit
• The graph database records all findings with rule IDs and severity

US-110-005: Debug a failed codemod run with full observability

As a developer investigating why a codemod failed
I want to see a structured trace of all operations, timings, and errors
So that I can identify the root cause without re-running the entire operation

Acceptance Criteria:

• The TracingObserver writes structured JSON spans via core.tracing.Tracer
• The DebugPanelObserver writes NDJSON trace files to pipeline_output/traces/
• The ErrorDetector pattern-matches error sequences and emits alerts
• The MetricsObserver records counters and histograms (passes run, files mutated, findings by severity)
• All events carry a correlation_id linking them to a single codemod run
• The graph database persists the full event stream as CodemodRun → CodemodPass → CodemodFinding nodes

US-110-006: Query code intelligence without reloading the workspace

As a developer exploring the impact of a proposed change
I want to ask "what depends on this module?" or "what tests cover this function?"
So that I can make informed refactoring decisions

Acceptance Criteria:

• The CIG supports multihop queries: dependencies(), dependents(), blast_radius(), affected_tests()
• Queries execute in <1ms via CSR adjacency arrays
• The CodeIntelligenceGraph API is accessible from Python code and the CLI
• Query results include node properties (file path, line number, qualified name)
• The graph database records query patterns for usage analytics

US-110-007: Compose multiple transforms in a single atomic commit

As a developer performing a multi-step refactor (rename + add import + update docstrings)
I want to apply all transforms in sequence and commit once
So that the codebase is never in an inconsistent intermediate state

Acceptance Criteria:

• The workspace.run(passes) method applies passes sequentially, each seeing the previous pass's mutations
• The composed ChangeSet is committed atomically
• Review passes run on the final mutated forest (after all transforms)
• The CommitReport lists all passes run and their individual contributions
• The graph database records the composition as a single CodemodRun with multiple CodemodPass children

US-110-008: Verify LibCST round-trip fidelity on a real codebase

As a platform maintainer validating the CST substrate
I want to prove that every file's module.code equals the on-disk bytes
So that I can guarantee no silent corruption during transforms

Acceptance Criteria:

• The test_round_trip_fidelity BDD scenario loads the real orchestrata-ultimate codebase (797 files)
• For every loaded module, module.code.encode('utf-8') == on_disk_bytes
• Unparseable files are recorded in workspace.unparsable with path and error message
• The benchmark completes in <30s single-threaded, <5s parallel (12 cores)
• RSS delta is <4 GiB

US-110-009: Integrate with SPEC-109 rename engine without forking resolvers

As a developer using the rename engine's LSP-based resolution
I want to feed WorkspaceEditPlan objects into the codemod engine
So that I get transactional writes and observability without duplicating rename logic

Acceptance Criteria:

• The RenameSymbolPass consumes WorkspaceEditPlan from the SPEC-109 rename engine — since 2026-10-20 that engine lives in codetools (codetools.refactor.rename.models, moved under orchestrata ADR-0117/ADR-0118), and the consumption is STRUCTURAL: RenameSymbolPass duck-types the plan it is handed and imports nothing from the rename engine, so this criterion still holds across the repo boundary
• LSP text edits are converted to LibCST mutations via libcst.metadata.PositionProvider
• The codemod engine adds NO second rename resolver
• Write-back reuses the applier's safety invariants (atomicity, post-parse, rollback, path-escape guard)
• The test_spec109_interop BDD scenario verifies identical results between EditApplier and RenameSymbolPass — RETIRED 2026-10-20: that scenario asserted the two appliers agree in one process, which the SPEC-109 move makes impossible (EditApplier is now in codetools, RenameSymbolPass here; the codetools copy could only import this lane through orchestrata's pinned git dep, which transitively requires orchestrata_ast — a package codetools does not declare, so it was red in both repos). Deleted from both repos by ruling. OWED REPLACEMENT: re-establish the equivalence as a cross-repo CONTRACT test driven through the deployed codetools CLI, gated on the queued console-script//opt/codetools/bin exposure step (ADR-0117's "orchestrata consumes the deployed CLI"). Until that lands, EditApplier↔RenameSymbolPass equivalence is UNTESTED — recorded, not silently dropped.

US-110-010: Persist codemod provenance to the operational graph database

As a project maintainer auditing code changes
I want to query the graph database for "what codemod runs touched this file?"
So that I can trace changes back to their originating operations

Acceptance Criteria:

• The BoltSyncCodemodRepository writes CodemodRun, CodemodPass, CodemodFinding, CodeChange nodes
• Edges: HAS_CODEMOD_RUN, HAS_PASS, HAS_FINDING, ABOUT, PRODUCED_BY, BREAKING_CHANGE, CHANGED_IN
• Writes use SyncGraphRepository.unit_of_work() for transactional consistency
• The ontology extends canonical_schema.py with new labels and relationships
• The test_graph_persistence BDD scenario verifies Cypher MERGE patterns against a real graph DB

US-110-013: Keep the workspace warm across invocations via a daemon

As a developer running repeated codemod queries against a large workspace I want to pay the 120s parse cost once rather than on every invocation So that interactive use is possible at all

Acceptance Criteria:

• A long-running daemon holds the parsed workspace in memory and serves requests over a Unix socket (REQ-110-133, REQ-110-134)
• Lifecycle is managed by codemod daemon start|stop|status|restart (REQ-110-135)
• File changes trigger incremental re-parse via inotify in <500ms (REQ-110-136)
• Concurrent clients read in parallel; writes serialize; reload takes an exclusive lock (REQ-110-137)
• The JSON-RPC surface and its Python client are specified in §11.2 (REQ-110-138, REQ-110-139, REQ-110-140)

Provenance. The daemon capability and its requirements REQ-110-133..140 were specified in §11 from the outset, but no US-110-0NN heading was ever written for them — so the four modules implementing it (daemon.py, client.py, server.py, watcher.py) have carried story="US-110-013" against a story that did not exist. This heading backs the id those decorators already cite; the story number is 013 as shipped rather than renumbered, because renumbering would invalidate working citations to fix nothing. 011 and 012 are unused, as REQ-110 ids are likewise sparse (001-035, 101-108, 133-140).

4. Data Models

4.1 Core Data Types

The platform's data model is built on frozen dataclasses with slots=True for memory efficiency and immutability. All types are serializable to JSON for CommitReport persistence.

# refactor/codemod/report.py

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Literal, Any

import libcst


@dataclass(frozen=True, slots=True)
class ChangeEntry:
    """One mutated CST tree with metadata."""
    path: str                          # repo-relative path (e.g., "src/pkg/mod.py")
    mutated_tree: libcst.Module        # the modified CST
    original_bytes: bytes              # pre-mutation on-disk bytes (for rollback)
    description: str                   # human-readable (e.g., "Renamed Foo to Bar")
    pass_name: str                     # which pass produced this entry


@dataclass(frozen=True, slots=True)
class ChangeSet:
    """Immutable collection of ChangeEntry objects."""
    entries: tuple[ChangeEntry, ...]

    @property
    def file_count(self) -> int:
        return len(self.entries)

    @property
    def paths(self) -> frozenset[str]:
        return frozenset(e.path for e in self.entries)

    def merge(self, other: ChangeSet) -> ChangeSet:
        """Combine two ChangeSets (later entries override earlier for same path)."""
        merged = {e.path: e for e in self.entries}
        merged.update({e.path: e for e in other.entries})
        return ChangeSet(entries=tuple(merged.values()))


@dataclass(frozen=True, slots=True)
class Finding:
    """One review finding (error or warning)."""
    severity: Literal["error", "warning"]
    path: str
    line: int | None
    rule: str                          # e.g., "BARE_OS_ENVIRON", "IMPORT_CYCLE"
    message: str
    proof_tree: dict[str, Any] | None = None  # from Prolog explain:explain_whynot/2


@dataclass(frozen=True, slots=True)
class WorkspaceStats:
    """Workspace load statistics."""
    file_count: int
    total_bytes: int
    rss_delta_bytes: int
    parse_time_seconds: float
    unparsable_count: int


@dataclass(frozen=True, slots=True)
class CommitReport:
    """Machine-readable commit report (JSON-serializable)."""
    passes_run: tuple[str, ...]
    files_written: tuple[str, ...]
    diffstat: str                      # unified diff summary
    findings: tuple[Finding, ...]
    duration_seconds: float
    dry_run: bool
    correlation_id: str                # links all CodemodEvent observations

    def to_json(self) -> str:
        """Serialize to JSON string."""
        ...

4.2 Code Intelligence Graph (CIG) Node Types

The CIG is an in-memory property graph with the following node types:

# refactor/codemod/_cig.py

from __future__ import annotations

from dataclasses import dataclass


@dataclass(frozen=True, slots=True)
class CIGModule:
    node_id: int                       # CSR index
    path: str                          # repo-relative path
    qualified_name: str                # dotted module name
    line_count: int
    byte_count: int
    content_hash: str                  # xxhash3_64


@dataclass(frozen=True, slots=True)
class CIGClass:
    node_id: int
    name: str
    qualified_name: str
    module_path: str
    line: int
    is_abstract: bool
    is_dataclass: bool
    is_protocol: bool
    is_enum: bool
    bases: tuple[str, ...]             # base class qualified names


@dataclass(frozen=True, slots=True)
class CIGFunction:
    node_id: int
    name: str
    qualified_name: str
    module_path: str
    line: int
    is_async: bool
    return_type: str | None
    complexity: int                    # cyclomatic complexity
    nesting_depth: int
    line_count: int
    param_count: int


@dataclass(frozen=True, slots=True)
class CIGMethod:
    node_id: int
    name: str
    qualified_name: str
    class_name: str
    module_path: str
    line: int
    is_async: bool
    is_classmethod: bool
    is_staticmethod: bool
    is_property: bool
    is_abstract: bool
    return_type: str | None
    complexity: int
    nesting_depth: int


@dataclass(frozen=True, slots=True)
class CIGParameter:
    node_id: int
    name: str
    type_annotation: str | None
    default: str | None
    position: int                      # 0-based parameter index
    parent_id: int                     # CIGFunction or CIGMethod node_id


@dataclass(frozen=True, slots=True)
class CIGTest:
    node_id: int
    name: str
    qualified_name: str
    module_path: str
    line: int
    target_name: str                   # stripped "test_" prefix for TESTED_BY matching


@dataclass(frozen=True, slots=True)
class CIGImport:
    node_id: int
    module_path: str                   # importing module
    imported_module: str               # imported module (dotted name)
    imported_name: str | None          # specific name (from X import Y) or None (import X)
    alias: str | None                  # "as" alias
    is_relative: bool
    line: int

4.3 Code Intelligence Graph (CIG) Edge Types

# refactor/codemod/_cig.py (continued)

from enum import StrEnum


class CIGEdgeType(StrEnum):
    """Edge types in the Code Intelligence Graph."""
    IMPORTS = "IMPORTS"                # Module → Module
    CALLS = "CALLS"                    # Function/Method → Function/Method
    INHERITS = "INHERITS"              # Class → Class
    DEFINES = "DEFINES"                # Module → Class/Function; Class → Method
    HAS_PARAMETER = "HAS_PARAMETER"    # Function/Method → Parameter
    TESTED_BY = "TESTED_BY"            # Function/Method → Test
    OVERRIDES = "OVERRIDES"            # Method → Method (same name, different class)

4.4 CSR (Compressed Sparse Row) Index

The CSR index stores graph adjacency as contiguous array.array('i') buffers for O(1) neighbor lookups:

# refactor/codemod/_csr_index.py

from __future__ import annotations

import array


class _CSRIndex:
    """Compressed Sparse Row adjacency index.

    Stores a directed graph as two CSR matrices (forward and transpose)
    for O(1) neighbor lookups in both directions.

    Memory layout:
        _offsets[i] : start index in _neighbors for node i
        _offsets[i+1] : end index (exclusive)
        _neighbors[_offsets[i]:_offsets[i+1]] : neighbor node_ids

    Example:
        Node 0 has neighbors [2, 5, 7]
        _offsets[0] = 0, _offsets[1] = 3
        _neighbors[0:3] = [2, 5, 7]
    """

    __slots__ = ("_offsets", "_neighbors", "_transpose_offsets", "_transpose_neighbors", "_node_count")

    def __init__(self, node_count: int, edges: list[tuple[int, int]]) -> None:
        self._node_count = node_count
        self._offsets, self._neighbors = self._build_csr(node_count, edges)
        self._transpose_offsets, self._transpose_neighbors = self._build_csr(
            node_count, [(to, frm) for frm, to in edges]
        )

    @staticmethod
    def _build_csr(node_count: int, edges: list[tuple[int, int]]) -> tuple[array.array, array.array]:
        """Build CSR arrays from edge list."""
        # Count out-degree per node
        degree = [0] * node_count
        for frm, _ in edges:
            degree[frm] += 1

        # Compute offsets (prefix sum)
        offsets = array.array("i", [0] * (node_count + 1))
        for i in range(node_count):
            offsets[i + 1] = offsets[i] + degree[i]

        # Fill neighbors
        neighbors = array.array("i", [0] * len(edges))
        cursor = list(offsets)  # mutable copy for fill cursor
        for frm, to in edges:
            neighbors[cursor[frm]] = to
            cursor[frm] += 1

        return offsets, neighbors

    def dependencies(self, node_id: int) -> list[int]:
        """Forward neighbors (what this node depends on)."""
        start = self._offsets[node_id]
        end = self._offsets[node_id + 1]
        return list(self._neighbors[start:end])

    def dependents(self, node_id: int) -> list[int]:
        """Reverse neighbors (what depends on this node)."""
        start = self._transpose_offsets[node_id]
        end = self._transpose_offsets[node_id + 1]
        return list(self._transpose_neighbors[start:end])

    def transitive_closure(self, node_id: int, direction: str = "forward", max_depth: int = -1) -> frozenset[int]:
        """BFS transitive closure with optional depth limit."""
        ...

    def update(self, node_id: int, new_edges: list[tuple[int, int]]) -> None:
        """Incrementally update edges for one node (rebuild CSR)."""
        ...

4.5 Codemod Event Types

# refactor/codemod/observability/events.py

from __future__ import annotations

from dataclasses import dataclass
from enum import StrEnum
from typing import Any


class CodemodEventType(StrEnum):
    """Observable event types during codemod operations."""
    WORKSPACE_LOADED = "workspace_loaded"
    INDEX_BUILT = "index_built"
    PASS_STARTED = "pass_started"
    PASS_COMPLETED = "pass_completed"
    FILE_MUTATED = "file_mutated"
    REVIEW_STARTED = "review_started"
    FINDING_EMITTED = "finding_emitted"
    REVIEW_COMPLETED = "review_completed"
    COMMIT_STARTED = "commit_started"
    FILE_WRITTEN = "file_written"
    COMMIT_COMPLETED = "commit_completed"
    COMMIT_ABORTED = "commit_aborted"
    ROLLBACK_TRIGGERED = "rollback_triggered"
    ERROR_DETECTED = "error_detected"


@dataclass(frozen=True, slots=True)
class CodemodEvent:
    """One observable event during codemod operations."""
    event_type: CodemodEventType
    timestamp: float                   # time.monotonic()
    correlation_id: str                # UUID linking all events in one run
    payload: dict[str, Any]            # event-specific data
    span_name: str | None = None       # for TracingObserver (e.g., "CodemodWorkspace.load")

4.6 Graph Persistence Records

# refactor/codemod/persistence/repository.py

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime


@dataclass(frozen=True, slots=True)
class CodemodRunRecord:
    """Graph database record of a codemod run."""
    id: str                            # "{project}:codemod_run:{correlation_id}"
    project: str
    started_at: datetime
    completed_at: datetime | None
    status: str                        # "success", "aborted", "rolled_back"
    dry_run: bool
    pass_count: int
    finding_count: int
    file_count: int
    duration_ms: int


@dataclass(frozen=True, slots=True)
class CodemodPassRecord:
    """Graph database record of a transform/review pass."""
    id: str                            # "{run_id}:pass:{pass_name}"
    run_id: str
    pass_name: str
    pass_type: str                     # "transform" or "review"
    file_count: int
    duration_ms: int


@dataclass(frozen=True, slots=True)
class CodemodFindingRecord:
    """Graph database record of a review finding."""
    id: str                            # "{run_id}:finding:{rule}:{path}:{line}"
    run_id: str
    severity: str                      # "error" or "warning"
    rule: str
    path: str
    line: int | None
    message: str
    module_id: str | None              # CIGModule node_id for ABOUT edge


@dataclass(frozen=True, slots=True)
class CodeChangeRecord:
    """Graph database record of a code change."""
    id: str                            # "{run_id}:change:{path}"
    run_id: str
    pass_name: str
    path: str
    description: str
    diff_hash: str                     # xxhash3_64 of unified diff
    entity_id: str | None              # CIGModule/Class/Function node_id for CHANGED_IN edge

4.7 Data Model Diagram

5. Functional Requirements

FR-001: Lossless CST Forest Load

REQ-110-001: The CodemodWorkspace.load(root, include, parallel) method SHALL parse every .py file under the configured include roots into a libcst.Module tree, keyed by repo-relative path.

REQ-110-002: For every loaded module, module.code.encode('utf-8') SHALL equal the on-disk bytes at load time (lossless round-trip fidelity). A module that fails to parse SHALL be recorded in workspace.unparsable as a (path, error_message) tuple and excluded from the forest.

REQ-110-003 (amended 2026-08-11): Loading the full orchestrata-ultimate tree (~2,500 files across src/scripts/tests, ~780K LOC) with parallel=True SHALL use a ThreadPoolExecutor with max_workers=os.cpu_count() and complete in <120 seconds on 12 cores. The original "<360 seconds single-threaded" clause is RETIRED: it was never a user mandate (introduced by the SPEC-110 implementation commit as spec completeness) and production (the codemod daemon, _daemon_entry.py) always loads with parallel=True. Serial remains the signature default for small/CLI loads; its performance is informational only.

REQ-110-004 (amended 2026-08-24): The workspace.stats() method SHALL report WorkspaceStats with file count, total bytes, RSS delta (via psutil.Process().memory_info().rss), parse time, and unparsable count. RSS delta SHALL remain <3.5 GiB for the full codebase (2,500 files, 780K LOC, 30 MB on disk), measured in parallel=True mode — the configuration production actually runs (see REQ-110-003 amendment).

FR-002: Module + Dependency Index

REQ-110-005: The workspace SHALL expose modules() (bidirectional mapping of dotted names ↔ paths), dependencies(module) (modules this module imports), and dependents(module) (modules that import this module), all derived from the CST forest's import statements.

REQ-110-006: The index SHALL update incrementally when a pass mutates imports. The _update_index(paths) method SHALL re-derive dependencies and dependents for the specified modules only, not the entire forest.

REQ-110-007: The symbol_table() method SHALL map top-level defined names (classes, functions, module-level variables) per module, sufficient for a pass to resolve "who defines X" without disk access.

FR-003: Transform Passes

REQ-110-008: A transform pass SHALL be a class implementing TransformPass.apply(workspace) -> ChangeSet. The ChangeSet SHALL record, per module: the mutated CST tree, a human-readable description, and the pass name.

REQ-110-009: Passes SHALL compose: workspace.run(passes: list[TransformPass]) SHALL apply them in order over the same forest. Pass N SHALL see pass N-1's mutations. The composed ChangeSet union SHALL be what write-back serializes — one transaction for N passes.

REQ-110-010: The engine SHALL ship 5 built-in transform passes:

• RenameSymbolPass: Consumes a WorkspaceEditPlan from SPEC-109's SymbolRenamer, applies LSP text edits as LibCST mutations. NO second rename resolver.
• ReplaceStringLiteralPass: Exact and regex modes with keep-pattern guard. Uses LibCST CSTTransformer to walk and replace SimpleString/ConcatenatedString nodes.
• AddImportPass: Adds an import statement to target modules via LibCST AddImportVisitor pattern.
• RemoveUnusedImportPass: Identifies imports not referenced in the module's body via CIG IMPORTS edge analysis.
• DocstringSweepPass: Vocabulary replacement in docstrings AND comments (LibCST preserves comments as EmptyLine nodes with comment field).

REQ-110-011: A transform pass SHALL NOT write to disk. The API SHALL give passes no file handles. Write-back SHALL be exclusively workspace.commit(). This SHALL be enforced by construction (no open(), Path.write_text(), or os.rename in pass code) and asserted by a test that greps the pass base-class surface.

FR-004: Review-Before-Write

REQ-110-012: A review pass SHALL be a class implementing ReviewPass.review(workspace, changes) -> list[Finding]. Review passes SHALL operate on the mutated forest (the ChangeSet entries' mutated_tree), not on diffs or disk files.

The engine SHALL ship 4 built-in review passes:

• ParseIntegrity: Every mutated module SHALL still parse (libcst.parse_module(entry.mutated_tree.code)) and round-trip (module.code == entry.mutated_tree.code).
• BannedConstructs: Re-hosts check_env_access (bare os.environ["X"]), check_no_mocks (14 mock patterns), and banned-library imports from .claude/banned-libraries.txt — all applied to LibCST trees.
• DiffBudget: A pass declaring max_files/max_lines SHALL fail review if its ChangeSet exceeds them (fat-finger guard).
• PrologSemanticReview: Asserts ChangeSet mutations as Prolog facts, queries codemod_review.pl predicates (pass_safe/3, blast_radius_ok/3, import_cycle_free/2, rename_safe/3, string_replace_safe/3), uses explain:explain_whynot/2 for proof trees.

REQ-110-013: The workspace.commit(reviews=[...]) method SHALL run reviews FIRST. Any Finding with severity "error" SHALL abort the commit with the findings report — nothing is written to disk. This is the in-RAM analogue of SPEC-109's residue abort.

FR-005: Transactional Write-Back

REQ-110-014: The commit() method SHALL write only mutated modules. Each file SHALL be written via: (1) write to tempfile in same directory, (2) os.fsync(tempfile_fd), (3) os.rename(tempfile_path, target_path). After ALL writes, every written file SHALL be re-parsed from disk (libcst.parse_module(written_bytes)). Any failure SHALL trigger rollback of every written file to its pre-commit bytes (kept in ChangeEntry.original_bytes).

REQ-110-015: The commit() method SHALL refuse to run if any target file's on-disk mtime or content hash (xxhash3_64) changed since load() (concurrent-edit guard). The refusal SHALL list the stale paths. The guard SHALL use O_NOFOLLOW when opening files to prevent symlink attacks, and fcntl.flock(LOCK_EX | LOCK_NB) for advisory locking during the write window.

REQ-110-016: A commit(dry_run=True) SHALL produce the full unified diff and the review report without writing. The CLI SHALL default to dry-run (house convention).

REQ-110-017: Every commit SHALL emit a machine-readable CommitReport (JSON): passes run, files written, diffstat, review findings, duration, correlation_id. The report path SHALL be printed on completion.

FR-006: CLI

REQ-110-018: The scripts/codemod.py CLI SHALL provide subcommands:

• load-bench: Measure workspace load time, RSS delta, file count (REQ-110-003/004 numbers)
• run --pass <name> [--arg k=v]... [--apply]: Execute named built-in passes; dry-run default
• report <path>: Pretty-print a CommitReport JSON file
• analyze --query <type> [--module <path>]: Execute CIG queries (dependencies, dependents, blast_radius, cycles, topological_order)

The CLI SHALL use os.environ.get only (no bare os.environ["X"]), absolute imports, and read nothing but the repo tree and its own args.

REQ-110-019: Exit codes SHALL be: 0 success/clean-dry-run, 1 review-abort or verification failure, 2 usage error. These SHALL be asserted by CLI subprocess tests.

FR-007: SPEC-109 Integration

REQ-110-020: The RenameSymbolPass SHALL consume SPEC-109 artifacts (WorkspaceEditPlan from orchestrata_ultimate.refactor.rename.models) rather than reimplementing resolution. The engine SHALL add NO second rename resolver.

REQ-110-021: Write-back SHALL reuse the applier's safety invariants. If code reuse of EditApplier is impractical (CST writes whole modules, not spans), the INVARIANTS (atomicity, post-parse, rollback, path-escape guard) SHALL be reimplemented against whole-module writes and cross-tested with the same scenarios as tests/refactor/rename/test_applier.py.

FR-008: Code Intelligence Graph

REQ-110-022: The CodeIntelligenceGraph SHALL be built from the CST forest by _cig_builder.py. It SHALL contain nodes for Module, Class, Function, Method, Parameter, Test, Import, and edges for IMPORTS, CALLS, INHERITS, DEFINES, TESTED_BY, OVERRIDES.

REQ-110-023: The CIG SHALL support multihop queries:

• dependencies(module_id): Direct imports (O(1) via CSR)
• dependents(module_id): Direct reverse imports (O(1) via CSR transpose)
• blast_radius(node_id, hops=3): BFS transitive closure over CALLS + INHERITS + IMPORTS edges
• breaking_changes(change): Signature diff + downstream consumer detection via CALLS + INHERITS edges
• cycles(): Tarjan's SCC algorithm for import cycle detection
• topological_order(): Kahn's algorithm for dependency-ordered module list
• affected_tests(changed_functions): Reverse TESTED_BY edges
• coupling_score(module_id): Efferent/afferent coupling + instability metric
• abstraction_level(module_id): Ratio of abstract definitions (Protocol/ABC) to concrete
• centrality(top_k=20): Betweenness centrality (bottleneck modules)
• impact_score(node_id): PageRank-style importance based on incoming edges

REQ-110-024: All CIG queries SHALL execute in <2ms for the target codebase (2,500 files, ~15,000 edges) via CSR adjacency arrays.

FR-009: Observability

REQ-110-025: The workspace SHALL compose a SyncObservable[CodemodEvent] and emit events for all operations: workspace loaded, index built, pass started/completed, file mutated, review started/completed, finding emitted, commit started/completed/aborted, rollback triggered, error detected.

REQ-110-026: The engine SHALL ship 5 built-in observers:

• TracingObserver: Writes structured JSON spans via core.tracing.Tracer with correlation_id context
• GraphWriterObserver: Async writes to graph DB via CodemodGraphRepository (batched, non-blocking, drops on buffer overflow)
• ErrorDetector: Pattern-matches error sequences (e.g., 3 consecutive parse failures) and emits alerts via EventBus
• MetricsObserver: Counters (passes run, files mutated, findings by severity) and histograms (pass duration, commit duration)
• DebugPanelObserver: Writes NDJSON trace files to pipeline_output/traces/{correlation_id}.ndjson for post-mortem analysis

REQ-110-027: All observers SHALL be optional and configurable. The workspace.run() and workspace.commit() methods SHALL accept an observers parameter to override defaults.

FR-010: Graph Persistence

REQ-110-028: The CodemodGraphRepository Protocol SHALL define methods: save_run(), save_pass(), save_findings(), save_changes(), link_changes_to_code(), query_runs(), query_findings(), query_affected_code().

REQ-110-029: The BoltSyncCodemodRepository SHALL implement the Protocol using SyncGraphRepository.unit_of_work() for transactional writes. It SHALL write CodemodRun, CodemodPass, CodemodFinding, CodeChange nodes with edges: HAS_CODEMOD_RUN, HAS_PASS, HAS_FINDING, ABOUT, PRODUCED_BY, BREAKING_CHANGE, CHANGED_IN.

REQ-110-030: The ontology SHALL extend canonical_schema.py with new labels (CodemodRun, CodemodPass, CodemodFinding, CodeChange) and relationships. The CodeChange label SHALL reuse the existing Change label as a sub-label.

FR-011: Analysis Passes

REQ-110-031: The engine SHALL ship analysis passes that compute code intelligence metrics from the CIG:

• BlastRadiusAnalyzer: BFS blast radius computation (REQ-110-023)
• BreakingChangeDetector: API breaking change detection (signature diff + downstream consumers)
• ComplexityAnalyzer: Cyclomatic + nesting + cognitive complexity per function/method
• TestCoverageMapper: Test-to-source mapping via TESTED_BY edges + name heuristic (strip test_ prefix)
• AbstractionAnalyzer: Abstraction level (ratio of Protocol/ABC to concrete classes)
• CouplingAnalyzer: Efferent/afferent coupling + instability metric
• DeadCodeDetector: Unreachable functions/classes via CIG reachability analysis

REQ-110-032: Analysis passes SHALL be read-only (no mutations) and SHALL return structured results (dataclasses) suitable for CLI output and graph persistence.

FR-012: In-Memory Review (40 Checks)

REQ-110-033: The InMemoryReviewEngine SHALL execute 40 checks on the mutated CST forest, organized into 7 categories:

Complexity (4 checks):

• Cyclomatic complexity (count branching nodes: If/For/While/With/Try/ExceptHandler/BooleanOperation/IfExp/comp)
• Nesting depth (recursive DFS counting compound blocks, max 4)
• Function length (line count excluding docstring, max 40)
• Parameter count (max 5 excluding self/cls)

Style (6 checks):

• Line length (max 88 chars, URLs exempted)
• Trailing whitespace
• Tab indentation (must use spaces)
• Final newline (must end with \n)
• Naming conventions (PascalCase for classes, snake_case for functions)
• Import ordering (stdlib < third-party < local)

Quality (4 checks):

• Type hints (public functions must have return type and parameter annotations)
• Docstrings (public classes and functions must have docstrings)
• Mutable defaults (no [], {}, set() as default arguments)
• God objects (max 300 lines/class, max 15 public methods, max 500 lines/file)

Structure (6 checks):

• Stub methods (no log-only, no-op methods in non-test code)
• Placeholders (no pass-only bodies, no raise NotImplementedError)
• Function arguments (max 5 parameters)
• File length (max 500 lines)
• Nesting depth (max 4 levels)
• Error handling (no bare except:, no swallowed exceptions)

Safety (10 checks):

• Bare os.environ["X"] subscript reads (must use .get())
• Mock patterns (14 regex patterns from checks_safety.py)
• eval()/exec() calls
• Hardcoded secrets (password/token patterns)
• Banned library imports (from .claude/banned-libraries.txt)
• flushdb()/flushall() in tests
• Unscoped DETACH DELETE in tests
• TODO/FIXME/HACK/XXX comments
• Broad raise Exception(...) or raise BaseException(...)
• print() outside __main__ guard

Correctness (6 checks):

• Relative imports (level > 0)
• Imported local module existence
• Entry point guard (if __name__ == "__main__": for CLI files)
• Third-party import declaration (in pyproject.toml)
• File completeness (all imports resolve)
• Dependency declaration (all third-party imports declared)

Test Isolation (4 checks):

• flushdb()/flushall() in tests
• Unscoped DETACH DELETE in tests
• Module-scoped fixtures for mutable state
• Global mutable state in test files

REQ-110-034: All 40 checks SHALL operate on the engine's in-memory lossless layer: the ctok token array (LosslessModule / TokenStream) and the CPython ast overlay the forest already holds (NodeIndex). No check SHALL re-parse source from disk, call subprocess.run, or invoke an external linter. Structural checks SHALL be implemented as ast.NodeVisitor subclasses over the MUTATED forest's ast overlay; lexical checks (line length, whitespace, tabs, final newline, comment and pattern scans) SHALL be implemented as scanners over the MUTATED forest's token array. A check SHALL reuse the matching verification/review/ check_* function where one exists rather than re-implement it. (Amended by ADR-0068, Accepted 2026-10-06; the original clause named LibCST CSTTransformer/CSTVisitor trees, which REQ-110-120..127 superseded.)

REQ-110-035: The InMemoryReviewEngine SHALL execute checks in three phases:

1. Fast structural checks (parallel per-file): complexity, style, quality, structure
2. Graph-dependent checks (require CIG): correctness, test isolation
3. Prolog semantic review (require Prolog facts): PrologSemanticReview pass

All phases SHALL run on the mutated forest (changes applied in-memory).

6. Error Conditions

7. API Contracts

7.1 CodemodWorkspace

# refactor/codemod/workspace.py

from __future__ import annotations

from collections.abc import Mapping, Sequence
from pathlib import Path
from typing import ClassVar

import libcst

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.patterns.observer import SyncObservable
from orchestrata_ultimate.refactor.codemod.passes import TransformPass
from orchestrata_ultimate.refactor.codemod.review import ReviewPass
from orchestrata_ultimate.refactor.codemod.report import (
    ChangeSet,
    CommitReport,
    WorkspaceStats,
    CodemodEvent,
)
from orchestrata_ultimate.refactor.codemod._cig import CodeIntelligenceGraph


@traces(spec="110", story="US-110-001", reqs=["REQ-110-001", "REQ-110-002", "REQ-110-003", "REQ-110-004"])
class CodemodWorkspace:
    """In-memory CST forest with demand-driven parse and parallel load."""

    DEFAULT_INCLUDE: ClassVar[tuple[str, ...]] = ("src", "scripts", "tests")

    def __init__(self, root: Path, include: tuple[str, ...], parallel: bool) -> None:
        ...

    @classmethod
    def load(
        cls,
        root: Path,
        include: tuple[str, ...] = DEFAULT_INCLUDE,
        parallel: bool = False,
    ) -> CodemodWorkspace:
        """Parse all .py files under include roots into CST forest.

        Args:
            root: Repository root directory.
            include: Tuple of directory names to include (e.g., ("src", "tests")).
            parallel: If True, use ThreadPoolExecutor for parallel parse.

        Returns:
            CodemodWorkspace with populated forest and index.

        Raises:
            ERR-110-001: Files that fail to parse are recorded in unparsable, not raised.
        """
        ...

    @property
    def forest(self) -> Mapping[str, libcst.Module]:
        """Repo-relative path → CST Module (demand-driven parse)."""
        ...

    @property
    def unparsable(self) -> Sequence[tuple[str, str]]:
        """(path, error_message) for files that failed to parse."""
        ...

    def modules(self) -> Mapping[str, str]:
        """Bidirectional mapping: dotted name ↔ repo-relative path."""
        ...

    def dependencies(self, module: str) -> frozenset[str]:
        """Modules this module imports (direct only)."""
        ...

    def dependents(self, module: str) -> frozenset[str]:
        """Modules that import this module (direct only)."""
        ...

    def symbol_table(self) -> Mapping[str, frozenset[str]]:
        """Module path → top-level defined names (classes, functions, variables)."""
        ...

    def stats(self) -> WorkspaceStats:
        """Workspace load statistics."""
        ...

    @traces(spec="110", story="US-110-003", reqs=["REQ-110-008", "REQ-110-009"])
    def run(self, passes: Sequence[TransformPass]) -> ChangeSet:
        """Apply transform passes sequentially over the forest.

        Pass N sees pass N-1's mutations. Returns composed ChangeSet.

        Args:
            passes: Sequence of TransformPass instances.

        Returns:
            ChangeSet with all mutations from all passes.

        Raises:
            ERR-110-007: If a pass attempts disk I/O.
        """
        ...

    @traces(spec="110", story="US-110-005", reqs=["REQ-110-013", "REQ-110-014", "REQ-110-015", "REQ-110-016", "REQ-110-017"])
    def commit(
        self,
        changes: ChangeSet,
        reviews: Sequence[ReviewPass] | None = None,
        dry_run: bool = True,
        observers: Sequence[Observer[CodemodEvent]] | None = None,
    ) -> CommitReport:
        """Review mutated forest, then write changes atomically.

        Args:
            changes: ChangeSet from workspace.run().
            reviews: Review passes to run before write. Defaults to DEFAULT_REVIEWS.
            dry_run: If True, produce diff and report without writing.
            observers: Override default observers for this commit.

        Returns:
            CommitReport with passes run, files written, findings, duration.

        Raises:
            ERR-110-002: If any review produces error-severity finding (introduced by the change, ADR-0073; `absolute` baseline aborts on any).
            ERR-110-003: If post-write re-parse fails (triggers rollback).
            ERR-110-004: If concurrent-edit guard detects stale file.
            ERR-110-005: If symlink attack detected.
            ERR-110-006: If advisory lock fails.
        """
        ...

    def _build_index(self) -> None:
        """Build CIG and CSR index from forest (called by load())."""
        ...

    def _update_index(self, paths: Sequence[str]) -> None:
        """Incrementally update index for mutated modules."""
        ...

7.2 TransformPass

# refactor/codemod/passes.py

from __future__ import annotations

from abc import ABC, abstractmethod
from typing import ClassVar, Literal

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.refactor.codemod.workspace import CodemodWorkspace
from orchestrata_ultimate.refactor.codemod.report import ChangeSet


@traces(spec="110", story="US-110-003", reqs=["REQ-110-008", "REQ-110-011"])
class TransformPass(ABC):
    """Abstract base for composable code transforms."""

    name: ClassVar[str]                # e.g., "ReplaceStringLiteralPass"
    parallelism: ClassVar[Literal["serial", "per_file"]] = "serial"
    max_files: ClassVar[int | None] = None  # for DiffBudget review
    max_lines: ClassVar[int | None] = None

    @abstractmethod
    def apply(self, ws: CodemodWorkspace) -> ChangeSet:
        """Apply transform to workspace, return ChangeSet.

        MUST NOT write to disk (enforced by ERR-110-007).
        """
        ...

    def apply_file(
        self,
        path: str,
        tree: libcst.Module,
        ws: CodemodWorkspace,
    ) -> ChangeEntry | None:
        """Per-file parallel dispatch (override for embarrassingly-parallel passes).

        Only called when parallelism == "per_file".
        Return None if file is unchanged.
        """
        raise NotImplementedError

7.3 ReviewPass

# refactor/codemod/review.py

from __future__ import annotations

from abc import ABC, abstractmethod

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.refactor.codemod.workspace import CodemodWorkspace
from orchestrata_ultimate.refactor.codemod.report import ChangeSet, Finding


@traces(spec="110", story="US-110-004", reqs=["REQ-110-012", "REQ-110-013"])
class ReviewPass(ABC):
    """Abstract base for composable code reviews."""

    name: ClassVar[str]                # e.g., "BannedConstructs"

    @abstractmethod
    def review(self, ws: CodemodWorkspace, changes: ChangeSet) -> list[Finding]:
        """Review mutated forest, return findings.

        Operates on ChangeSet entries' mutated_tree, not on disk files.
        """
        ...

7.4 CodeIntelligenceGraph

# refactor/codemod/_cig.py

from __future__ import annotations

from dataclasses import dataclass

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.refactor.codemod._csr_index import _CSRIndex


@dataclass(frozen=True, slots=True)
class BlastRadius:
    """Result of blast_radius() query."""
    affected_nodes: frozenset[int]
    affected_files: frozenset[str]
    impact_score: float                # 0.0 to 1.0


@dataclass(frozen=True, slots=True)
class BreakingChange:
    """Result of breaking_changes() query."""
    change_path: str
    consumer_path: str
    consumer_line: int
    incompatibility: str               # e.g., "removed parameter 'x'"


@dataclass(frozen=True, slots=True)
class CouplingScore:
    """Result of coupling_score() query."""
    efferent: int                      # outgoing dependencies
    afferent: int                      # incoming dependencies
    instability: float                 # efferent / (efferent + afferent)


@dataclass(frozen=True, slots=True)
class AbstractionLevel:
    """Result of abstraction_level() query."""
    abstract_count: int                # Protocol/ABC classes
    concrete_count: int
    ratio: float                       # abstract / (abstract + concrete)


@traces(spec="110", story="US-110-006", reqs=["REQ-110-022", "REQ-110-023", "REQ-110-024"])
class CodeIntelligenceGraph:
    """In-memory property graph with multihop traversal."""

    def __init__(self, csr: _CSRIndex, nodes: dict[int, CIGModule | CIGClass | CIGFunction | ...]) -> None:
        ...

    def dependencies(self, node_id: int) -> list[int]:
        """Direct imports (O(1) via CSR)."""
        ...

    def dependents(self, node_id: int) -> list[int]:
        """Direct reverse imports (O(1) via CSR transpose)."""
        ...

    def blast_radius(self, node_id: int, hops: int = 3) -> BlastRadius:
        """BFS transitive closure over CALLS + INHERITS + IMPORTS edges."""
        ...

    def breaking_changes(self, change: CodeChange) -> list[BreakingChange]:
        """Detect API-breaking changes via signature diff + downstream consumers."""
        ...

    def dependency_path(self, from_id: int, to_id: int) -> list[int] | None:
        """Shortest dependency path (BFS)."""
        ...

    def cycles(self) -> list[list[int]]:
        """All import cycles via iterative Tarjan's SCC."""
        ...

    def topological_order(self) -> list[int]:
        """Kahn's algorithm — modules in dependency order."""
        ...

    def affected_tests(self, changed_functions: frozenset[int]) -> frozenset[int]:
        """Tests covering changed functions via TESTED_BY reverse edges."""
        ...

    def coupling_score(self, node_id: int) -> CouplingScore:
        """Efferent/afferent coupling + instability metric."""
        ...

    def abstraction_level(self, node_id: int) -> AbstractionLevel:
        """Ratio of abstract definitions (Protocol/ABC) to concrete."""
        ...

    def centrality(self, top_k: int = 20) -> list[tuple[int, float]]:
        """Betweenness centrality — bottleneck modules."""
        ...

    def impact_score(self, node_id: int) -> float:
        """PageRank-style importance based on incoming edges."""
        ...

7.5 CodemodGraphRepository

# refactor/codemod/persistence/repository.py

from __future__ import annotations

from collections.abc import Sequence
from typing import Protocol, runtime_checkable

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.refactor.codemod.persistence.records import (
    CodemodRunRecord,
    CodemodPassRecord,
    CodemodFindingRecord,
    CodeChangeRecord,
)


@traces(spec="110", story="US-110-010", reqs=["REQ-110-028"])
@runtime_checkable
class CodemodGraphRepository(Protocol):
    """Repository for codemod provenance in the operational graph DB."""

    def save_run(self, run: CodemodRunRecord) -> None:
        """MERGE CodemodRun node + HAS_CODEMOD_RUN edge to Project."""
        ...

    def save_pass(self, run_id: str, pass_record: CodemodPassRecord) -> None:
        """MERGE CodemodPass node + HAS_PASS edge to CodemodRun."""
        ...

    def save_findings(self, run_id: str, findings: Sequence[CodemodFindingRecord]) -> None:
        """UNWIND batch MERGE CodemodFinding nodes + HAS_FINDING + ABOUT edges."""
        ...

    def save_changes(self, run_id: str, changes: Sequence[CodeChangeRecord]) -> None:
        """UNWIND batch MERGE CodeChange nodes + PRODUCED_BY + CHANGED_IN edges."""
        ...

    def link_changes_to_code(self, changes: Sequence[CodeChangeRecord]) -> None:
        """MERGE BREAKING_CHANGE edges for detected breaking changes."""
        ...

    def query_runs(self, project: str, limit: int = 10) -> list[CodemodRunRecord]:
        """Query recent codemod runs for a project."""
        ...

    def query_findings(self, run_id: str) -> list[CodemodFindingRecord]:
        """Query all findings for a codemod run."""
        ...

    def query_affected_code(self, run_id: str) -> list[str]:
        """Query all code entity IDs affected by a codemod run."""
        ...

8. Non-Functional Requirements

REQ-110-101: No mocks, stubs, fakes, monkeypatch, or test skips in tests. Real files in tmp_path, real git init repos, real containers. Function-scope fixtures only.

REQ-110-102: Every public class and method SHALL carry @traces(spec="110", story="US-110-NNN", reqs=["REQ-110-NNN"]) citing SPEC-110 requirement IDs.

REQ-110-103: The engine SHALL pass all gates: ruff check, black --check, serialize_codebase_ast.py standards, check-deps --include-tests, test-review.py --stdout (0 issues), pyright_ratchet.py --max 283.

REQ-110-104: No pytest-xdist. Tests SHALL run via the batched executor (scripts/run_tests.py) or streaming runner (scripts/run_tests_streaming.py).

REQ-110-105: New package: src/orchestrata_ultimate/refactor/codemod/ (sibling of refactor/rename/). Tests in tests/refactor/codemod/. No edits to refactor/rename/*, agents/tool_registry.py, or refactor/__init__.py (wiring is a later step, same policy as SPEC-109).

REQ-110-106: Free-threading compatibility. The C extension (_codemod_fast) SHALL declare Py_mod_gil = Py_MOD_GIL_NOT_USED in its PyModuleDef. The tests/conftest.py GIL probe SHALL fast-fail if the GIL is re-enabled.

REQ-110-107: LibCST FT-compatibility. If LibCST 1.8.6's native Rust parser has GIL issues on 3.14t, the engine SHALL fall back to LIBCST_PARSER_TYPE=pure (pure-Python parser) and document which mode is active in benchmark output.

REQ-110-108: Coverage ≥90% per file. Measured via pytest --cov=orchestrata_ultimate.refactor.codemod --cov-report=term-missing -o addopts="".

9. Technical Design

9.1 System Architecture

9.2 CST Forest Structure

The CST forest is a dictionary mapping repo-relative paths to LibCST Module trees. Each Module is a lossless representation of the source file, preserving:

• Comments: Stored as EmptyLine nodes with comment field
• Whitespace: Preserved in IndentedBlock and EmptyLine nodes
• Formatting: Quote styles, trailing commas, line breaks all preserved
• Byte-perfect round-trip: module.code.encode('utf-8') == on_disk_bytes

Demand-driven parse: The forest is initialized with None sentinels for all discovered files. When a pass or query accesses workspace.forest[path], the file is parsed on-demand. This reduces RSS by ~80% for typical passes that touch <20% of files.

Parallel parse: When parallel=True, the workspace uses ThreadPoolExecutor(max_workers=os.cpu_count()) to parse files concurrently. On Python 3.14t with GIL disabled, this achieves ~6x speedup on 12 cores (measured 2.66x on 4 threads, extrapolated linearly).

9.3 Code Intelligence Graph

The CIG is built by _cig_builder.py in a single pass over the CST forest:

1. Node creation: Walk each Module.body to extract ClassDef, FunctionDef, Import, ImportFrom nodes. Assign sequential node_id integers.
2. Edge extraction: For each import, resolve the target module path and create an IMPORTS edge. For each call expression, resolve the callee and create a CALLS edge. For each class base, create an INHERITS edge.
3. CSR construction: Pass the edge list to _CSRIndex, which builds forward and transpose adjacency arrays.

Multihop queries use BFS over the CSR arrays:

def blast_radius(self, node_id: int, hops: int = 3) -> BlastRadius:
    """BFS with depth limit over CALLS + INHERITS + IMPORTS edges."""
    visited = {node_id}
    frontier = [node_id]
    for _ in range(hops):
        next_frontier = []
        for nid in frontier:
            # Forward edges (CALLS, IMPORTS)
            next_frontier.extend(self._csr.dependencies(nid))
            # Reverse edges (INHERITS — subclasses affected by base changes)
            next_frontier.extend(self._csr.dependents(nid))
        next_frontier = [n for n in next_frontier if n not in visited]
        visited.update(next_frontier)
        frontier = next_frontier
    affected_files = frozenset(self._nodes[nid].module_path for nid in visited if isinstance(self._nodes[nid], (CIGFunction, CIGMethod)))
    return BlastRadius(affected_nodes=frozenset(visited), affected_files=affected_files, impact_score=len(visited) / len(self._nodes))

9.4 Transform Pass Execution

Transform passes execute in two modes:

Serial mode (parallelism == "serial"): The pass's apply(workspace) method is called once. The pass iterates over the forest internally. Used for RenameSymbolPass (cross-file dependencies require global view).

Per-file parallel mode (parallelism == "per_file"): The workspace dispatches apply_file(path, tree, workspace) to a ThreadPoolExecutor. Each thread processes one file independently. Used for ReplaceStringLiteralPass, AddImportPass, RemoveUnusedImportPass, DocstringSweepPass (embarrassingly parallel).

def run(self, passes: Sequence[TransformPass]) -> ChangeSet:
    composed = ChangeSet(entries=())
    for pass_ in passes:
        if pass_.parallelism == "per_file":
            with ThreadPoolExecutor(max_workers=os.cpu_count()) as pool:
                futures = [
                    pool.submit(pass_.apply_file, path, tree, self)
                    for path, tree in self.forest.items()
                ]
                entries = [f.result() for f in futures if f.result() is not None]
        else:
            entries = pass_.apply(self).entries
        composed = composed.merge(ChangeSet(entries=tuple(entries)))
        self._update_index(composed.paths)  # incremental index update
    return composed

9.5 Review Pipeline

The review pipeline executes in three phases:

Phase 1 — Fast structural checks (parallel per-file): Complexity, style, quality, structure checks. Each file is reviewed independently via ThreadPoolExecutor. These checks operate on individual CST trees without cross-file context.

Phase 2 — Graph-dependent checks (require CIG): Correctness, test isolation checks. These checks query the CIG for cross-file relationships (e.g., "does this import resolve to a file in the forest?").

Phase 3 — Prolog semantic review (require Prolog facts): The PrologSemanticReview pass asserts ChangeSet mutations as Prolog facts, then queries codemod_review.pl predicates. Uses explain:explain_whynot/2 for proof trees when a safety check fails.

def review(self, ws: CodemodWorkspace, changes: ChangeSet) -> list[Finding]:
    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

9.6 Transactional Write-Back

The commit process follows a strict sequence to ensure atomicity:

Batch fsync optimization: Instead of fsync-ing each file individually (100 fsyncs for 100 files), the engine opens directory file descriptors and fsyncs each directory once. This reduces I/O overhead by ~20x (5 directories instead of 100 files).

def _commit_atomic(self, changes: ChangeSet, repo_root: Path, snapshot: dict[str, bytes]) -> None:
    dir_fds = {}
    written = []
    try:
        for entry in changes.entries:
            path = repo_root / entry.path
            parent = path.parent
            if parent not in dir_fds:
                dir_fds[parent] = os.open(parent, os.O_RDONLY | os.O_DIRECTORY)
            tmp = path.with_suffix('.codemod_tmp')
            tmp.write_bytes(entry.mutated_tree.code.encode('utf-8'))
            os.fsync(tmp.open('rb').fileno())
            os.rename(tmp, path)
            written.append(path)

        for fd in dir_fds.values():
            os.fsync(fd)

        # Post-write verification
        for path in written:
            libcst.parse_module(path.read_text(encoding='utf-8'))
    except Exception:
        self._rollback(snapshot, repo_root)
        raise
    finally:
        for fd in dir_fds.values():
            os.close(fd)

9.7 Observer Event Flow

The observer architecture uses SyncObservable[CodemodEvent] from core.patterns.observer. Events are emitted synchronously during operations, and observers react in subscription order.

Log-and-continue policy: If an observer raises an exception, the observable logs the exception and continues dispatching to other observers. This prevents a failing observer from aborting the codemod operation.

9.8 Graph Persistence Model

The BoltSyncCodemodRepository writes provenance to the operational graph database using Cypher MERGE patterns:

// CodemodRun node + HAS_CODEMOD_RUN edge
MERGE (r:CodemodRun {id: $id})
SET r.started_at = datetime($started_at),
    r.completed_at = datetime($completed_at),
    r.status = $status,
    r.dry_run = $dry_run,
    r.pass_count = $pass_count,
    r.finding_count = $finding_count,
    r.file_count = $file_count,
    r.duration_ms = $duration_ms
WITH r
MATCH (p:Project {slug: $project})
MERGE (p)-[:HAS_CODEMOD_RUN]->(r)

// CodemodPass nodes + HAS_PASS edges (batched)
UNWIND $passes AS pass
MERGE (cp:CodemodPass {id: pass.id})
SET cp.pass_name = pass.pass_name,
    cp.pass_type = pass.pass_type,
    cp.file_count = pass.file_count,
    cp.duration_ms = pass.duration_ms
WITH cp, pass
MATCH (r:CodemodRun {id: pass.run_id})
MERGE (r)-[:HAS_PASS]->(cp)

// CodemodFinding nodes + HAS_FINDING + ABOUT edges (batched)
UNWIND $findings AS f
MERGE (cf:CodemodFinding {id: f.id})
SET cf.severity = f.severity,
    cf.rule = f.rule,
    cf.path = f.path,
    cf.line = f.line,
    cf.message = f.message
WITH cf, f
MATCH (r:CodemodRun {id: f.run_id})
MERGE (r)-[:HAS_FINDING]->(cf)
WITH cf, f
OPTIONAL MATCH (m:Module {id: f.module_id})
FOREACH (_ IN CASE WHEN m IS NOT NULL THEN [1] ELSE [] END |
    MERGE (cf)-[:ABOUT]->(m))

// CodeChange nodes + PRODUCED_BY + CHANGED_IN edges (batched)
UNWIND $changes AS c
MERGE (cc:CodeChange {id: c.id})
SET cc.pass_name = c.pass_name,
    cc.description = c.description,
    cc.diff_hash = c.diff_hash
WITH cc, c
MATCH (r:CodemodRun {id: c.run_id})
MERGE (cc)-[:PRODUCED_BY]->(r)
WITH cc, c
MATCH (entity {id: c.entity_id})
MERGE (entity)-[:CHANGED_IN]->(cc)

9.9 Prolog Integration

The PrologSemanticReview pass integrates with the project's Prolog infrastructure via janus-swi:

1. Assert mutations as facts: For each ChangeEntry, assert codemod_mutated(ModulePath, PassName) and codemod_affected(Symbol, ModulePath) facts.
2. Query safety predicates: Call codemod_review:pass_safe(PassType, Context, Safe) to check if the pass is safe given the current mutations.
3. Explain failures: If a safety check fails, call explain:explain_whynot(codemod_review:pass_safe(...), ProofTree) to get a proof tree explaining why.

% src/orchestrata_ultimate/prolog/codemod_review.pl

:- module(codemod_review, [
    pass_safe/3,
    blast_radius_ok/3,
    import_cycle_free/2,
    rename_safe/3,
    string_replace_safe/3
]).

:- use_module(library(lists)).

% Dynamic facts asserted by Python
:- dynamic codemod_affected/2.
:- dynamic codemod_imports/2.
:- dynamic codemod_mutated/2.
:- dynamic codemod_string_use/3.

% A pass is safe if all its mutations pass safety checks
pass_safe(PassName, Module, Safe) :-
    codemod_mutated(Module, PassName),
    \+ unsafe_mutation(PassName, Module),
    Safe = true.
pass_safe(_, _, false).

% Blast radius: affected files must not exceed limit
blast_radius_ok(Symbol, MaxFiles, Ok) :-
    findall(A, codemod_affected(Symbol, A), Affected),
    length(Affected, Count),
    (Count =< MaxFiles -> Ok = true ; Ok = false).

% Import cycle detection
import_cycle_free(TargetModule, ImportedModule) :-
    \+ transitive_depends(ImportedModule, TargetModule).

transitive_depends(A, B) :- codemod_imports(A, B).
transitive_depends(A, B) :- codemod_imports(A, C), transitive_depends(C, B).

% Rename safety: no affected symbol in a different package
rename_safe(Symbol, _NewName, Safe) :-
    findall(A, codemod_affected(Symbol, A), Affected),
    forall(member(M, Affected), same_package(Symbol, M)),
    Safe = true.
rename_safe(_, _, false).

% String replacement safety
string_replace_safe(Module, Old, Safe) :-
    \+ codemod_string_use(Module, Old, fstring),
    \+ codemod_string_use(Module, Old, format_call),
    Safe = true.
string_replace_safe(_, _, false).

9.10 C Extension Design

The _codemod_fast C extension follows the janus-swi vendored-fork pattern:

Sub-package structure:

scripts/_codemod_fast/
  pyproject.toml          # setuptools build-backend
  setup.py                # Extension('codemod_fast._native', ...)
  codemod_fast/
    __init__.py           # Python fallback: try import _native, fall back to pure Python
    csr_index.c           # CSR build + BFS in C (10x faster than Python array.array)
    content_hash.c        # xxhash3 wrapper (5 GB/s vs SHA-256 500 MB/s)
    mmap_loader.c         # mmap file loading (zero-copy parse source)
    module.c              # PyModuleDef with Py_mod_gil = Py_MOD_GIL_NOT_USED

Critical: The module.c MUST declare Py_MOD_GIL_NOT_USED or the entire process loses free-threading. This is enforced by tests/conftest.py's GIL probe.

Python fallback pattern:

# codemod_fast/__init__.py

try:
    from codemod_fast._native import csr_build, csr_bfs, xxhash3_64, mmap_load
    _HAS_NATIVE = True
except ImportError:
    _HAS_NATIVE = False
    # Pure Python implementations below
    def csr_build(node_count, edges):
        ...
    def csr_bfs(offsets, neighbors, start, max_depth):
        ...
    def xxhash3_64(data):
        import hashlib
        return hashlib.sha256(data).hexdigest()[:16]
    def mmap_load(path):
        return Path(path).read_bytes()

Wiring in root pyproject.toml:

[tool.uv.sources]
codemod-fast = { path = "scripts/_codemod_fast" }

[tool.uv]
no-build-isolation-package = ["janus-swi", "codemod-fast"]

9.11 BDD Test Strategy

The test suite uses pytest-bdd 8.1.0 for behavioral specification:

Feature file location: specs/110-codemod-engine/features/ (co-located with spec.md)

Step definitions: tests/refactor/codemod/conftest.py (shared) + per-test-file (specific)

Scenario binding: @scenario per scenario (explicit, with TST/REQ docstrings)

Tags: @REQ_110_NNN, @TST_110_NNN, @FR_NNN on scenarios (registered as pytest markers)

Non-BDD tests: Benchmark, CSR unit, C extension parity tests stay as plain pytest (performance and data-structure tests don't benefit from Given/When/Then)

Example feature file:

@FR_001
Feature: Lossless CST Forest Load
  As a developer running a codemod
  I want the entire repository loaded as a lossless CST forest in RAM
  So that every subsequent operation works on a single, consistent, byte-perfect tree

  Background:
    Given a fixture package with Python source files

  @REQ_110_001 @TST_110_001
  Scenario: Load parses every .py file under included roots
    Given the fixture package contains 5 Python files across 2 directories
    When I load the workspace with include roots "src" and "tests"
    Then the forest contains exactly 5 modules
    And the unparsable list is empty

  @REQ_110_002 @TST_110_001
  Scenario: Round-trip fidelity — module.code equals on-disk bytes
    Given the fixture package contains a file with comments, blank lines, and odd quoting
    When I load the workspace
    Then for every loaded module, module.code equals the on-disk bytes decoded as UTF-8

Example step definitions:

# tests/refactor/codemod/conftest.py

from pytest_bdd import given, when, then, parsers

@given(
    parsers.parse("the fixture package contains {count:d} Python files"),
    target_fixture="repo_root",
)
def fixture_with_files(fixture_package, count):
    files = {}
    for i in range(count):
        files[f"pkg/mod_{i}.py"] = f"# Module {i}\nx_{i} = {i}\n"
    files["pkg/__init__.py"] = ""
    return fixture_package(files)

@when("I load the workspace", target_fixture="workspace")
def workspace_loaded(repo_root):
    return CodemodWorkspace.load(repo_root, include=("src",))

@then(parsers.parse("the forest contains exactly {count:d} modules"))
def forest_count(workspace, count):
    assert len(workspace.forest) == count

10. Examples

10.1 CLI Usage

# Measure workspace load performance
python scripts/codemod.py load-bench
# Output:
# Load time (serial): 28.3s
# Load time (parallel): 4.7s (12 cores)
# Files loaded: 797
# Total bytes: 9,437,184
# RSS delta: 1.2 GiB
# Unparsable: 0

# Dry-run a string replacement (default)
python scripts/codemod.py run --pass replace-string --arg old=Neo4j --arg new=Graph
# Output:
# Pass: ReplaceStringLiteralPass
# Files mutated: 152
# Findings: 0 errors, 3 warnings
# Diff:
# --- a/src/orchestrata_ultimate/knowledge_graph/neo4j_client.py
# +++ b/src/orchestrata_ultimate/knowledge_graph/neo4j_client.py
# @@ -42,7 +42,7 @@
# -class Neo4jClient:
# +class GraphClient:
# ...
# Commit report: reports/codemod-2026-07-27T14-32-01.json

# Apply the changes (write to disk)
python scripts/codemod.py run --pass replace-string --arg old=Neo4j --arg new=Graph --apply
# Output:
# Pass: ReplaceStringLiteralPass
# Files written: 152
# Findings: 0 errors, 3 warnings
# Duration: 2.3s
# Commit report: reports/codemod-2026-07-27T14-32-01.json

# Query code intelligence
python scripts/codemod.py analyze --query dependencies --module orchestrata_ultimate.knowledge_graph.sync
# Output:
# Dependencies of orchestrata_ultimate.knowledge_graph.sync:
#   - orchestrata_ultimate.knowledge_graph.canonical_schema
#   - orchestrata_ultimate.knowledge_graph.schema
#   - orchestrata_ultimate.core.graph.interfaces
#   - orchestrata_ultimate.core.neo4j_factory

# Query blast radius
python scripts/codemod.py analyze --query blast_radius --module orchestrata_ultimate.core.base.traces --hops 3
# Output:
# Blast radius of orchestrata_ultimate.core.base.traces (3 hops):
#   Affected nodes: 342
#   Affected files: 187
#   Impact score: 0.43

10.2 Python API Usage

from pathlib import Path
from orchestrata_ultimate.refactor.codemod.workspace import CodemodWorkspace
from orchestrata_ultimate.refactor.codemod.passes import (
    ReplaceStringLiteralPass,
    AddImportPass,
    DocstringSweepPass,
)
from orchestrata_ultimate.refactor.codemod.review import DEFAULT_REVIEWS

# Load workspace
ws = CodemodWorkspace.load(Path("."), include=("src", "tests"), parallel=True)
print(f"Loaded {ws.stats().file_count} files in {ws.stats().parse_time_seconds:.1f}s")

# Compose multiple passes
passes = [
    ReplaceStringLiteralPass(old="Neo4j", new="Graph", regex=False),
    DocstringSweepPass(replacements={"neo4j": "graph", "Neo4j": "Graph"}),
    AddImportPass(module="__future__", name="annotations"),
]
changes = ws.run(passes)
print(f"Mutated {changes.file_count} files")

# Review and commit (dry-run by default)
report = ws.commit(changes, reviews=DEFAULT_REVIEWS, dry_run=True)
print(f"Findings: {len([f for f in report.findings if f.severity == 'error'])} errors, "
      f"{len([f for f in report.findings if f.severity == 'warning'])} warnings")

# Apply changes (write to disk)
if not any(f.severity == "error" for f in report.findings):
    report = ws.commit(changes, reviews=DEFAULT_REVIEWS, dry_run=False)
    print(f"Written {len(report.files_written)} files in {report.duration_seconds:.1f}s")
    print(f"Commit report: {report.to_json()}")

10.3 CIG Query Usage

from orchestrata_ultimate.refactor.codemod.workspace import CodemodWorkspace

ws = CodemodWorkspace.load(Path("."), include=("src",))
cig = ws._cig  # CodeIntelligenceGraph instance

# Query dependencies
module_id = cig._node_ids["orchestrata_ultimate.knowledge_graph.sync"]
deps = cig.dependencies(module_id)
print(f"Dependencies: {[cig._nodes[d].qualified_name for d in deps]}")

# Query blast radius
blast = cig.blast_radius(module_id, hops=3)
print(f"Blast radius: {len(blast.affected_nodes)} nodes, {len(blast.affected_files)} files")
print(f"Impact score: {blast.impact_score:.2f}")

# Query breaking changes
from orchestrata_ultimate.refactor.codemod.report import CodeChange
change = CodeChange(path="src/pkg/api.py", pass_name="RenameSymbolPass", description="Removed parameter 'x' from foo()", diff_hash="abc123", entity_id=module_id)
breaking = cig.breaking_changes(change)
for bc in breaking:
    print(f"Breaking change: {bc.consumer_path}:{bc.consumer_line} — {bc.incompatibility}")

# Query import cycles
cycles = cig.cycles()
for cycle in cycles:
    print(f"Cycle: {' -> '.join(cig._nodes[nid].qualified_name for nid in cycle)}")

# Query topological order
order = cig.topological_order()
print(f"Topological order: {[cig._nodes[nid].qualified_name for nid in order[:10]]}...")

11. Daemon Architecture

11.1 Problem: Cold Start Cost

The engine parses 2,500 files (120s, 3 GiB RSS) on every invocation. This is unacceptable for:

• Interactive use (IDE plugins, REPL sessions)
• Multiple sequential operations (query → transform → query again)
• CI/CD pipelines running multiple codemods

11.2 Solution: Long-Running Daemon

REQ-110-133: The engine SHALL be deployed as a long-running daemon process that keeps the workspace in memory between invocations. The daemon SHALL parse the workspace once at startup (120s) and serve subsequent requests via Unix socket IPC (<100ms round-trip).

REQ-110-134: The daemon SHALL create a Unix socket at .codemod_daemon/socket in the repo root. The socket SHALL use JSON-RPC 2.0 protocol for client communication. The daemon SHALL write its PID to .codemod_daemon/pid and acquire a lock file at .codemod_daemon/lock to prevent multiple daemons per repo.

REQ-110-135: The daemon SHALL support lifecycle management via codemod daemon start|stop|status|restart subcommands. The start command SHALL fork to background, parse the workspace, create the Unix socket, and start an inotify watcher thread. The stop command SHALL send SIGTERM, wait 5s for graceful shutdown, and force kill if necessary.

REQ-110-136: The daemon SHALL watch for file changes using inotify. When a file is modified, created, deleted, or moved, the daemon SHALL trigger incremental re-parse of the affected file(s) and update the CIG and CSR index. Incremental reload SHALL complete in <500ms.

REQ-110-137: The daemon SHALL support concurrent clients. Multiple clients MAY connect simultaneously and execute read operations (queries) in parallel. Write operations (run_pass, commit) SHALL be serialized via mutex. Reload operations SHALL acquire exclusive lock and block all read/write operations.

REQ-110-138: The daemon SHALL expose the following JSON-RPC methods:

• query_dependencies(node_id) → list of node IDs
• query_dependents(node_id) → list of node IDs
• query_blast_radius(node_id, hops) → BlastRadius object
• query_breaking_changes(change) → list of BreakingChange objects
• query_cycles() → list of cycles
• query_topological_order() → list of node IDs
• query_affected_tests(function_ids) → list of test IDs
• query_coupling_score(node_id) → CouplingScore object
• query_abstraction_level(node_id) → float
• query_centrality(top_k) → list of (node_id, score) tuples
• query_impact_score(node_id) → float
• run_pass(pass_name, params) → ChangeSet object
• commit(changes, reviews, dry_run) → CommitReport object
• reload() → force workspace reload (bypass inotify)
• status() → daemon status (uptime, memory, clients)

REQ-110-139: The daemon SHALL provide a Python client library (CodemodClient) that connects to the Unix socket and exposes the JSON-RPC methods as Python functions. The client SHALL support context manager protocol for automatic connection cleanup.

REQ-110-140: The CLI SHALL auto-connect to the daemon if running, or start the daemon if not running (with --auto-start flag). If the daemon is not running and --auto-start is not set, the CLI SHALL fall back to library mode (parse on every invocation) and emit a warning.

11.3 Performance Targets (Updated)

11.4 Example Usage

Daemon lifecycle:

# Start daemon (one-time cost: 120s)
codemod daemon start

# Subsequent CLI invocations (fast: <100ms)
codemod query dependencies pkg.module.Class
codemod query blast-radius pkg.module.Class --hops 3
codemod run --pass replace-string --arg old=foo --arg new=bar

# Stop daemon
codemod daemon stop

Python client:

from pathlib import Path
from orchestrata_ultimate.refactor.codemod.client import CodemodClient

with CodemodClient.connect(Path(".")) as client:
    # Query operations (fast, <100ms)
    deps = client.query_dependencies("pkg.module.Class")
    blast = client.query_blast_radius("pkg.module.Class", hops=3)
    
    # Transform operations (slower, 1-10s)
    changes = client.run_pass("ReplaceStringLiteralPass", {"old": "foo", "new": "bar"})
    report = client.commit(changes, dry_run=True)

11.5 Alternative Considered: Library with Disk Caching

Rejected because:

• Disk caching still has cold-start cost on first invocation (120s parse + 5s serialize)
• Cache invalidation requires checking all file mtimes/hashes (~100ms for 2,500 files)
• No support for incremental reload (must re-parse entire workspace on cache miss)
• No support for concurrent clients (each process has its own cache)
• Daemon provides better performance (100ms vs 2s) and more features (file watching, concurrency)

When library mode would be appropriate:

• One-off CI/CD jobs (no daemon overhead)
• Environments where daemons are not allowed (containerized builds)
• Simple scripts that don't need interactive performance

These use cases are supported via --no-daemon flag (falls back to library mode).

11.6 Path Confinement (SR-110-001)

Every file path SHALL be resolved and confined to the declared repo_root before any I/O. The _resolve_in_repo(path) method SHALL raise PathEscapeError if the path resolves outside repo_root. This SHALL run before snapshots, before writes, and before post-write verification.

def _resolve_in_repo(self, path: Path) -> Path:
    resolved = Path(path).resolve()
    try:
        resolved.relative_to(self._repo_root)
    except ValueError as exc:
        raise PathEscapeError(
            f"{path} resolves to {resolved}, outside repo_root {self._repo_root}"
        ) from exc
    return resolved

11.7 Symlink Attack Prevention (SR-110-002)

The concurrent-edit guard SHALL use O_NOFOLLOW when opening files to prevent symlink attacks. If open() fails with ELOOP (symlink detected), the commit SHALL be refused and a security alert logged.

import os

def _check_concurrent_edits(self, changes: ChangeSet, repo_root: Path) -> None:
    for entry in changes.entries:
        path = repo_root / entry.path
        # O_NOFOLLOW prevents following symlinks
        fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW)
        try:
            stat = os.fstat(fd)
            if stat.st_mtime_ns != self._snapshots[entry.path].mtime_ns:
                raise ConcurrentEditError(f"File {path} modified since load")
            content = os.read(fd, stat.st_size)
            if xxhash3_64(content) != self._snapshots[entry.path].content_hash:
                raise ConcurrentEditError(f"File {path} content changed since load")
        finally:
            os.close(fd)

11.8 Advisory File Locking (SR-110-003)

The commit process SHALL acquire advisory locks (fcntl.flock(LOCK_EX | LOCK_NB)) on all target files before writing. If a lock fails with EWOULDBLOCK, the commit SHALL be refused and the locked paths listed.

import fcntl

def _acquire_locks(self, changes: ChangeSet, repo_root: Path) -> list[int]:
    fds = []
    for entry in changes.entries:
        path = repo_root / entry.path
        fd = os.open(path, os.O_RDWR | os.O_NOFOLLOW)
        try:
            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
            fds.append(fd)
        except OSError as exc:
            if exc.errno == 11:  # EWOULDBLOCK
                for fd in fds:
                    os.close(fd)
                raise ConcurrentEditError(f"File {path} locked by another process")
            raise
    return fds

11.9 No Disk I/O in Passes (SR-110-004)

Transform passes SHALL NOT write to disk. This SHALL be enforced by construction (the TransformPass API provides no file handles) and asserted by a test that greps the pass base-class surface for open(), Path.write_text(), os.rename, etc.

def test_pass_cannot_write_to_disk():
    """REQ-110-011: TransformPass API has no disk I/O capability."""
    import ast
    source = inspect.getsource(TransformPass)
    tree = ast.parse(source)
    for node in ast.walk(tree):
        if isinstance(node, ast.Call):
            if isinstance(node.func, ast.Name) and node.func.id in ("open", "write_text", "rename"):
                raise AssertionError(f"TransformPass contains disk I/O: {node.func.id}")

11.10 Threat Model

12. Governance

12.1 Prolog Semantic Review

The PrologSemanticReview pass integrates with the project's governance infrastructure:

• codebase_db.pl: Reuses the deductive code analysis database for dependency queries
• codemod_review.pl: Custom predicates for codemod-specific safety checks
• explain.pl: ADR-0031 proof-tree meta-interpreter for explaining why a safety check failed

12.2 Constitutional Compliance

The engine SHALL comply with the project's 12 constitutional articles:

13. Dependencies

13.1 Required Dependencies

13.2 Reused Project Infrastructure

13.3 Rejected Alternatives

14. Integration Points

14.1 BDD Test Strategy

The test suite uses pytest-bdd 8.1.0 for behavioral specification. Feature files live in specs/110-codemod-engine/features/ (co-located with spec.md). Step definitions live in tests/refactor/codemod/conftest.py (shared) and per-test-file (specific).

Feature file inventory (12 files, ~81 scenarios):

14.2 Non-BDD Tests

14.3 Coverage Targets

REQ-110-108: Coverage ≥90% per file. Measured via:

pytest tests/refactor/codemod/ \
  --cov=orchestrata_ultimate.refactor.codemod \
  --cov-report=term-missing \
  -o addopts="" \
  -p orchestrata_ultimate.core.threading.pytest_live_observer

14.4 Gate Commands

All gates SHALL pass before every commit:

.venv/bin/ruff check src tests scripts
.venv/bin/black --check src/orchestrata_ultimate/refactor/codemod tests/refactor/codemod scripts/codemod.py
PYTHONPATH=src .venv/bin/python scripts/serialize_codebase_ast.py standards
PYTHONPATH=src .venv/bin/python scripts/serialize_codebase_ast.py check-deps --include-tests
PYTHONPATH=src .venv/bin/python scripts/test-review.py tests/refactor/codemod/ --stdout
PYTHONPATH=src .venv/bin/python scripts/pyright_ratchet.py --max 283

15. Observability

15.1 Event Types

The CodemodEventType enum defines 14 observable events:

15.2 Observer Configuration

Observers are configured per-commit via the observers parameter:

from orchestrata_ultimate.refactor.codemod.observability import (
    TracingObserver,
    GraphWriterObserver,
    ErrorDetector,
    MetricsObserver,
    DebugPanelObserver,
)

observers = [
    TracingObserver(),
    GraphWriterObserver(repository=graph_repo, batch_size=100),
    ErrorDetector(patterns=["3_consecutive_parse_failures"]),
    MetricsObserver(),
    DebugPanelObserver(output_dir=Path("pipeline_output/traces")),
]

report = ws.commit(changes, reviews=DEFAULT_REVIEWS, dry_run=False, observers=observers)

15.3 Tracing Integration

The TracingObserver writes structured JSON spans via core.tracing.Tracer:

{
  "timestamp": "2026-07-27T14:32:01.123Z",
  "correlation_id": "codemod-2026-07-27T14-32-01",
  "span_name": "CodemodWorkspace.load",
  "duration_ms": 4723,
  "event_type": "WORKSPACE_LOADED",
  "payload": {
    "file_count": 797,
    "parse_time_seconds": 4.7,
    "rss_delta_bytes": 1288490188
  }
}

15.4 Metrics

The MetricsObserver records counters and histograms:

Counters:

• codemod_passes_run_total{pass_name, pass_type}
• codemod_files_mutated_total{pass_name}
• codemod_findings_total{severity, rule}
• codemod_commits_total{status} (success, aborted, rolled_back)

Histograms:

• codemod_pass_duration_seconds{pass_name} (buckets: 0.1, 0.5, 1, 5, 10, 30, 60)
• codemod_commit_duration_seconds (buckets: 0.1, 0.5, 1, 5, 10, 30, 60)
• codemod_workspace_load_seconds (buckets: 1, 5, 10, 30, 60, 120)

15.5 Debug Panel

The DebugPanelObserver writes NDJSON trace files to pipeline_output/traces/{correlation_id}.ndjson:

{"timestamp": 1722089521.123, "event_type": "WORKSPACE_LOADED", "payload": {"file_count": 797}}
{"timestamp": 1722089521.456, "event_type": "INDEX_BUILT", "payload": {"node_count": 5234}}
{"timestamp": 1722089522.789, "event_type": "PASS_STARTED", "payload": {"pass_name": "ReplaceStringLiteralPass"}}
{"timestamp": 1722089523.012, "event_type": "FILE_MUTATED", "payload": {"path": "src/pkg/mod.py"}}
{"timestamp": 1722089523.234, "event_type": "PASS_COMPLETED", "payload": {"file_count": 152}}
{"timestamp": 1722089523.567, "event_type": "FINDING_EMITTED", "payload": {"severity": "warning", "rule": "TODO_COMMENT"}}
{"timestamp": 1722089524.890, "event_type": "COMMIT_COMPLETED", "payload": {"files_written": 152}}

16. Integration Points

16.1 SPEC-109 Rename Engine

The RenameSymbolPass consumes WorkspaceEditPlan from orchestrata_ultimate.refactor.rename.models. LSP text edits are converted to LibCST mutations via libcst.metadata.PositionProvider:

class RenameSymbolPass(TransformPass):
    name = "RenameSymbolPass"
    parallelism = "serial"  # cross-file dependencies

    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]
            # 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))

16.2 Knowledge Graph Ontology

The ontology extends canonical_schema.py with new labels and relationships:

New labels:

• CodemodRun (merge key: id)
• CodemodPass (merge key: id)
• CodemodFinding (merge key: id)
• CodeChange (sub-label of existing Change)

New relationships:

• HAS_CODEMOD_RUN (Project → CodemodRun)
• HAS_PASS (CodemodRun → CodemodPass)
• HAS_FINDING (CodemodRun → CodemodFinding)
• ABOUT (CodemodFinding → Module/Class/Function)
• PRODUCED_BY (CodeChange → CodemodPass)
• BREAKING_CHANGE (CodeChange → Module/Class/Function)
• CHANGED_IN (Module/Class/Function → CodeChange, reuses existing relationship)

16.3 Prolog Infrastructure

The PrologSemanticReview pass integrates with:

• codebase_db.pl: Deductive code analysis database (dependency queries, call graph, impact analysis)
• codemod_review.pl: Custom predicates for codemod-specific safety checks
• explain.pl: ADR-0031 proof-tree meta-interpreter for explaining failures

16.4 Event Bus

The ErrorDetector observer emits alerts via kernel.event_bus.EventBus:

class ErrorDetector:
    def on_event(self, event: CodemodEvent) -> None:
        if event.event_type == CodemodEventType.ERROR_DETECTED:
            self._event_bus.emit(SystemEvent(
                id=f"codemod-error-{event.correlation_id}",
                type="codemod.error",
                source="codemod_engine",
                timestamp=event.timestamp,
                payload=event.payload,
                correlation_id=event.correlation_id,
                severity="high",
            ))

16.5 Logging

The TracingObserver uses core.logging.StructuredFormatter for JSON logging:

class TracingObserver:
    def on_event(self, event: CodemodEvent) -> None:
        self._logger.log(
            level=logging.INFO,
            msg=event.event_type.value,
            extra={
                "correlation_id": event.correlation_id,
                "event_type": event.event_type.value,
                "payload": event.payload,
                "duration_ms": event.payload.get("duration_seconds", 0) * 1000,
            },
        )

17. Definition of Done

• [ ] refactor/codemod/ implements FR-001 through FR-012 with @traces(reqs=[...]) citing REQ-110-*
• [ ] scripts/codemod.py CLI (FR-006) with load-bench, run, report, analyze subcommands
• [ ] TST-110-001 through TST-110-010 pass via BDD scenarios (pytest-bdd)
• [ ] 12 Gherkin feature files in specs/110-codemod-engine/features/ with ~81 scenarios
• [ ] Step definitions in tests/refactor/codemod/conftest.py (shared) + per-test-file (specific)
• [ ] Coverage ≥90% per file (measured via pytest --cov)
• [ ] All gates pass: ruff, black, standards, check-deps, test-review, pyright_ratchet
• [ ] plans/contracts/spec-110-codemod-engine.mmd Mermaid classDiagram contract
• [ ] plans/contracts/spec-110-codemod-engine.contract.json machine-readable metadata
• [ ] src/orchestrata_ultimate/prolog/codemod_review.pl Prolog semantic review predicates
• [ ] Builder briefs (BRIEF-W0 through BRIEF-W5) in plans/codemod-engine-briefs/
• [ ] C extension (scripts/_codemod_fast/) with Py_MOD_GIL_NOT_USED declaration
• [ ] Graph ontology additions in canonical_schema.py (new labels and relationships)
• [ ] BDD infrastructure wiring: pyproject.toml updates, existing feature file wiring
• [ ] Registry wiring (refactor/init.py, tool_registry.py) is a separate, later step — not in scope here

18. Revision History

Related

• RTM page for 110-codemod-engine
• All specs
• RTM overview
• Gates