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

BRIEF-W4: Persistence (Qwen Max Agent)

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

Prose pending.

Plan text

BRIEF-W4: Persistence (Qwen Max Agent)

Wave Overview

Wave: W4 (Persistence)
Agent: Qwen Max (graph database patterns, Cypher queries)
Duration: ~15 minutes
Dependencies: W1 (workspace, CIG), W2 (passes, review), W3 (observability, CLI)
Parallel Execution: No (single agent)

Objective

Implement graph database persistence for codemod provenance. This wave writes codemod run metadata (runs, passes, findings, changes) to the operational graph database using the repository pattern, enabling queries like "what codemod runs touched this file?" and "trace this change back to its origin."

Scope

Files to Implement

1. src/orchestrata_ultimate/refactor/codemod/persistence/repository.py (~200 LOC)

   • CodemodGraphRepository Protocol (8 methods)
   • Import record dataclasses from records.py (W0 skeleton)

2. src/orchestrata_ultimate/refactor/codemod/persistence/bolt_repository.py (~350 LOC)

   • BoltSyncCodemodRepository class (implements Protocol)
   • Cypher MERGE patterns for all node types
   • UNWIND batch operations for passes, findings, changes
   • Transactional writes via SyncGraphRepository.unit_of_work()

3. tests/refactor/codemod/test_graph_persistence.py (~400 LOC)

   • 11 BDD scenarios from graph_persistence.feature
   • Test Protocol method signatures
   • Test Cypher MERGE patterns against real graph DB
   • Test query methods (query_runs, query_findings, query_affected_code)

Files to Modify

1. src/orchestrata_ultimate/knowledge_graph/canonical_schema.py (~30 LOC added)
   • Add new labels: LABEL_CODEMOD_RUN, LABEL_CODEMOD_PASS, LABEL_CODEMOD_FINDING, LABEL_CODE_CHANGE
   • Add new relationships: REL_HAS_CODEMOD_RUN, REL_HAS_PASS, REL_HAS_FINDING, REL_ABOUT, REL_PRODUCED_BY, REL_BREAKING_CHANGE
   • CodeChange reuses existing LABEL_CHANGE as sub-label (no new label constant needed for Change)

Requirements

This wave implements FR-010: Graph Persistence from spec.md:

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

Instructions

Step 1: Extend canonical_schema.py

Add the following constants to src/orchestrata_ultimate/knowledge_graph/canonical_schema.py:

# ---------------------------------------------------------------------------
# Codemod engine labels (SPEC-110 FR-010, REQ-110-030)
# ---------------------------------------------------------------------------

LABEL_CODEMOD_RUN = "CodemodRun"
LABEL_CODEMOD_PASS = "CodemodPass"
LABEL_CODEMOD_FINDING = "CodemodFinding"
# Note: CodeChange uses LABEL_CHANGE (already declared) as sub-label
# MERGE (cc:CodeChange:Change {id: $id})

# ---------------------------------------------------------------------------
# Codemod engine relationships (SPEC-110 FR-010, REQ-110-030)
# ---------------------------------------------------------------------------

REL_HAS_CODEMOD_RUN = "HAS_CODEMOD_RUN"  # (:Project)-[:HAS_CODEMOD_RUN]->(:CodemodRun)
REL_HAS_PASS = "HAS_PASS"  # (:CodemodRun)-[:HAS_PASS]->(:CodemodPass)
REL_HAS_FINDING = "HAS_FINDING"  # (:CodemodRun)-[:HAS_FINDING]->(:CodemodFinding)
REL_ABOUT = "ABOUT"  # (:CodemodFinding)-[:ABOUT]->(:Module)
REL_PRODUCED_BY = "PRODUCED_BY"  # (:CodeChange)-[:PRODUCED_BY]->(:CodemodPass)
REL_BREAKING_CHANGE = "BREAKING_CHANGE"  # (:CodeChange)-[:BREAKING_CHANGE]->(:Module|:Class|:Function)
# Note: REL_CHANGED_IN already exists (line 232) — (:entity)-[:CHANGED_IN]->(:CodeChange)

Important:

• LABEL_CODE_CHANGE is NOT added as a separate constant. Instead, CodeChange nodes use LABEL_CHANGE as a sub-label: MERGE (cc:CodeChange:Change {id: $id}).
• REL_CHANGED_IN already exists at line 232. Do NOT duplicate it.
• Add these constants after the existing Tier I (conversation) labels and before Tier J (architecture ontology).

Step 2: Implement CodemodGraphRepository Protocol

Create src/orchestrata_ultimate/refactor/codemod/persistence/repository.py:

"""Repository protocol for codemod provenance persistence."""

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.
    
    Implementations write codemod run metadata (runs, passes, findings, changes)
    to the graph database, enabling provenance queries like "what codemod runs
    touched this file?" and "trace this change back to its origin."
    """

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

    def save_pass(self, run_id: str, pass_record: CodemodPassRecord) -> None:
        """MERGE CodemodPass node + HAS_PASS edge to CodemodRun.
        
        Args:
            run_id: The parent codemod run ID.
            pass_record: The pass record to persist.
        """
        ...

    def save_findings(self, run_id: str, findings: Sequence[CodemodFindingRecord]) -> None:
        """UNWIND batch MERGE CodemodFinding nodes + HAS_FINDING + ABOUT edges.
        
        Args:
            run_id: The parent codemod run ID.
            findings: The finding records to persist (batched).
        """
        ...

    def save_changes(self, run_id: str, changes: Sequence[CodeChangeRecord]) -> None:
        """UNWIND batch MERGE CodeChange nodes + PRODUCED_BY + CHANGED_IN edges.
        
        Args:
            run_id: The parent codemod run ID.
            changes: The change records to persist (batched).
        """
        ...

    def link_changes_to_code(self, changes: Sequence[CodeChangeRecord]) -> None:
        """MERGE BREAKING_CHANGE edges for detected breaking changes.
        
        Args:
            changes: The change records with breaking_change=True.
        """
        ...

    def query_runs(self, project: str, limit: int = 10) -> list[CodemodRunRecord]:
        """Query recent codemod runs for a project.
        
        Args:
            project: The project slug.
            limit: Maximum number of runs to return.
            
        Returns:
            List of codemod run records, ordered by started_at DESC.
        """
        ...

    def query_findings(self, run_id: str) -> list[CodemodFindingRecord]:
        """Query all findings for a codemod run.
        
        Args:
            run_id: The codemod run ID.
            
        Returns:
            List of finding records for the run.
        """
        ...

    def query_affected_code(self, run_id: str) -> list[str]:
        """Query all affected entity IDs for a codemod run.
        
        Args:
            run_id: The codemod run ID.
            
        Returns:
            List of entity IDs (Module/Class/Function node IDs) affected by the run.
        """
        ...

Key points:

• Use @runtime_checkable to enable isinstance() checks.
• All methods have ... (Ellipsis) bodies — this is a Protocol, not an ABC.
• Import record dataclasses from records.py (created in W0 skeleton).
• Add @traces decorator with correct spec/story/reqs.

Step 3: Implement BoltSyncCodemodRepository

Create src/orchestrata_ultimate/refactor/codemod/persistence/bolt_repository.py:

"""Bolt-based synchronous implementation of CodemodGraphRepository."""

from __future__ import annotations

from collections.abc import Sequence
from typing import TYPE_CHECKING

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.knowledge_graph import canonical_schema as cs
from orchestrata_ultimate.refactor.codemod.persistence.records import (
    CodemodRunRecord,
    CodemodPassRecord,
    CodemodFindingRecord,
    CodeChangeRecord,
)
from orchestrata_ultimate.refactor.codemod.persistence.repository import (
    CodemodGraphRepository,
)

if TYPE_CHECKING:
    from orchestrata_ultimate.core.graph import SyncGraphRepository


@traces(spec="110", story="US-110-010", reqs=["REQ-110-029"])
class BoltSyncCodemodRepository:
    """Bolt-based synchronous implementation of CodemodGraphRepository.
    
    Uses SyncGraphRepository.unit_of_work() for transactional writes.
    Writes CodemodRun, CodemodPass, CodemodFinding, CodeChange nodes with
    edges: HAS_CODEMOD_RUN, HAS_PASS, HAS_FINDING, ABOUT, PRODUCED_BY,
    BREAKING_CHANGE, CHANGED_IN.
    """

    def __init__(self, repo: SyncGraphRepository, project: str) -> None:
        """Initialize the repository.
        
        Args:
            repo: The underlying graph repository (Neo4j or Memgraph).
            project: The project slug (used for Project node lookup).
        """
        self._repo = repo
        self._project = project

    def save_run(self, run: CodemodRunRecord) -> None:
        """MERGE CodemodRun node + HAS_CODEMOD_RUN edge to Project."""
        cypher = f"""
        MERGE (r:{cs.LABEL_CODEMOD_RUN} {{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:{cs.LABEL_PROJECT} {{slug: $project}})
        MERGE (p)-[:{cs.REL_HAS_CODEMOD_RUN}]->(r)
        """
        params = {
            "id": run.id,
            "started_at": run.started_at.isoformat(),
            "completed_at": run.completed_at.isoformat() if run.completed_at else None,
            "status": run.status,
            "dry_run": run.dry_run,
            "pass_count": run.pass_count,
            "finding_count": run.finding_count,
            "file_count": run.file_count,
            "duration_ms": run.duration_ms,
            "project": self._project,
        }
        with self._repo.unit_of_work() as uow:
            uow.run(cypher, params)

    def save_pass(self, run_id: str, pass_record: CodemodPassRecord) -> None:
        """MERGE CodemodPass node + HAS_PASS edge to CodemodRun."""
        cypher = f"""
        MERGE (cp:{cs.LABEL_CODEMOD_PASS} {{id: $id}})
        SET cp.pass_name = $pass_name,
            cp.pass_type = $pass_type,
            cp.file_count = $file_count,
            cp.duration_ms = $duration_ms
        WITH cp
        MATCH (r:{cs.LABEL_CODEMOD_RUN} {{id: $run_id}})
        MERGE (r)-[:{cs.REL_HAS_PASS}]->(cp)
        """
        params = {
            "id": pass_record.id,
            "run_id": run_id,
            "pass_name": pass_record.pass_name,
            "pass_type": pass_record.pass_type,
            "file_count": pass_record.file_count,
            "duration_ms": pass_record.duration_ms,
        }
        with self._repo.unit_of_work() as uow:
            uow.run(cypher, params)

    def save_findings(self, run_id: str, findings: Sequence[CodemodFindingRecord]) -> None:
        """UNWIND batch MERGE CodemodFinding nodes + HAS_FINDING + ABOUT edges."""
        if not findings:
            return
        
        cypher = f"""
        UNWIND $findings AS f
        MERGE (cf:{cs.LABEL_CODEMOD_FINDING} {{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:{cs.LABEL_CODEMOD_RUN} {{id: f.run_id}})
        MERGE (r)-[:{cs.REL_HAS_FINDING}]->(cf)
        WITH cf, f
        OPTIONAL MATCH (m:{cs.LABEL_MODULE} {{id: f.module_id}})
        FOREACH (_ IN CASE WHEN m IS NOT NULL THEN [1] ELSE [] END |
            MERGE (cf)-[:{cs.REL_ABOUT}]->(m))
        """
        params = {
            "findings": [
                {
                    "id": f.id,
                    "run_id": f.run_id,
                    "severity": f.severity,
                    "rule": f.rule,
                    "path": f.path,
                    "line": f.line,
                    "message": f.message,
                    "module_id": f.module_id,
                }
                for f in findings
            ]
        }
        with self._repo.unit_of_work() as uow:
            uow.run(cypher, params)

    def save_changes(self, run_id: str, changes: Sequence[CodeChangeRecord]) -> None:
        """UNWIND batch MERGE CodeChange nodes + PRODUCED_BY + CHANGED_IN edges."""
        if not changes:
            return
        
        # CodeChange uses LABEL_CHANGE as sub-label (REQ-110-030)
        cypher = f"""
        UNWIND $changes AS c
        MERGE (cc:{cs.LABEL_CHANGE}: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 (cp:{cs.LABEL_CODEMOD_PASS} {{id: c.run_id + ':pass:' + c.pass_name}})
        MERGE (cc)-[:{cs.REL_PRODUCED_BY}]->(cp)
        WITH cc, c
        OPTIONAL MATCH (entity {{id: c.entity_id}})
        FOREACH (_ IN CASE WHEN entity IS NOT NULL THEN [1] ELSE [] END |
            MERGE (entity)-[:{cs.REL_CHANGED_IN}]->(cc))
        """
        params = {
            "changes": [
                {
                    "id": c.id,
                    "run_id": c.run_id,
                    "pass_name": c.pass_name,
                    "description": c.description,
                    "diff_hash": c.diff_hash,
                    "entity_id": c.entity_id,
                }
                for c in changes
            ]
        }
        with self._repo.unit_of_work() as uow:
            uow.run(cypher, params)

    def link_changes_to_code(self, changes: Sequence[CodeChangeRecord]) -> None:
        """MERGE BREAKING_CHANGE edges for detected breaking changes."""
        if not changes:
            return
        
        # Filter to changes with breaking_change=True
        breaking = [c for c in changes if c.breaking_change]
        if not breaking:
            return
        
        cypher = f"""
        UNWIND $changes AS c
        MATCH (cc:{cs.LABEL_CHANGE}:CodeChange {{id: c.id}})
        MATCH (entity {{id: c.entity_id}})
        MERGE (cc)-[:{cs.REL_BREAKING_CHANGE}]->(entity)
        """
        params = {
            "changes": [
                {
                    "id": c.id,
                    "entity_id": c.entity_id,
                }
                for c in breaking
            ]
        }
        with self._repo.unit_of_work() as uow:
            uow.run(cypher, params)

    def query_runs(self, project: str, limit: int = 10) -> list[CodemodRunRecord]:
        """Query recent codemod runs for a project."""
        cypher = f"""
        MATCH (p:{cs.LABEL_PROJECT} {{slug: $project}})-[:{cs.REL_HAS_CODEMOD_RUN}]->(r:{cs.LABEL_CODEMOD_RUN})
        RETURN r.id AS id,
               r.started_at AS started_at,
               r.completed_at AS completed_at,
               r.status AS status,
               r.dry_run AS dry_run,
               r.pass_count AS pass_count,
               r.finding_count AS finding_count,
               r.file_count AS file_count,
               r.duration_ms AS duration_ms
        ORDER BY r.started_at DESC
        LIMIT $limit
        """
        params = {"project": project, "limit": limit}
        results = self._repo.execute_read(cypher, params)
        
        return [
            CodemodRunRecord(
                id=row["id"],
                project=project,
                started_at=row["started_at"],
                completed_at=row["completed_at"],
                status=row["status"],
                dry_run=row["dry_run"],
                pass_count=row["pass_count"],
                finding_count=row["finding_count"],
                file_count=row["file_count"],
                duration_ms=row["duration_ms"],
            )
            for row in results
        ]

    def query_findings(self, run_id: str) -> list[CodemodFindingRecord]:
        """Query all findings for a codemod run."""
        cypher = f"""
        MATCH (r:{cs.LABEL_CODEMOD_RUN} {{id: $run_id}})-[:{cs.REL_HAS_FINDING}]->(cf:{cs.LABEL_CODEMOD_FINDING})
        RETURN cf.id AS id,
               cf.severity AS severity,
               cf.rule AS rule,
               cf.path AS path,
               cf.line AS line,
               cf.message AS message
        """
        params = {"run_id": run_id}
        results = self._repo.execute_read(cypher, params)
        
        return [
            CodemodFindingRecord(
                id=row["id"],
                run_id=run_id,
                severity=row["severity"],
                rule=row["rule"],
                path=row["path"],
                line=row["line"],
                message=row["message"],
                module_id=None,  # Not returned in query
            )
            for row in results
        ]

    def query_affected_code(self, run_id: str) -> list[str]:
        """Query all affected entity IDs for a codemod run."""
        cypher = f"""
        MATCH (r:{cs.LABEL_CODEMOD_RUN} {{id: $run_id}})-[:{cs.REL_HAS_CODEMOD_RUN}]->(cc:{cs.LABEL_CHANGE}:CodeChange)
        MATCH (entity)-[:{cs.REL_CHANGED_IN}]->(cc)
        RETURN DISTINCT entity.id AS entity_id
        """
        params = {"run_id": run_id}
        results = self._repo.execute_read(cypher, params)
        
        return [row["entity_id"] for row in results if row["entity_id"]]

Key points:

• Use SyncGraphRepository.unit_of_work() for all writes (transactional).
• Use execute_read() for all queries (read-only).
• Cypher patterns match spec.md section 9.8 exactly.
• CodeChange nodes use LABEL_CHANGE:CodeChange (sub-label pattern).
• UNWIND batch operations for passes, findings, changes (performance).
• Import canonical_schema as cs for all label/relationship constants.
• Add @traces decorator with correct spec/story/reqs.

Step 4: Implement BDD Tests

Create tests/refactor/codemod/test_graph_persistence.py:

"""BDD tests for graph persistence (FR-010, REQ-110-028 through REQ-110-030)."""

from __future__ import annotations

from datetime import datetime
from pathlib import Path

import pytest

from orchestrata_ultimate.refactor.codemod.persistence.records import (
    CodemodRunRecord,
    CodemodPassRecord,
    CodemodFindingRecord,
    CodeChangeRecord,
)
from orchestrata_ultimate.refactor.codemod.persistence.repository import (
    CodemodGraphRepository,
)
from orchestrata_ultimate.refactor.codemod.persistence.bolt_repository import (
    BoltSyncCodemodRepository,
)


@pytest.fixture
def graph_repo(test_id: str) -> BoltSyncCodemodRepository:
    """Create a BoltSyncCodemodRepository with real graph DB connection."""
    from orchestrata_ultimate.core.graph_factory import create_sync_repository
    
    repo = create_sync_repository()
    project = f"test-project-{test_id}"
    
    # Create Project node for testing
    repo.execute_write(
        f"MERGE (p:Project {{slug: $project}})",
        {"project": project},
    )
    
    return BoltSyncCodemodRepository(repo, project)


@pytest.fixture
def sample_run(test_id: str) -> CodemodRunRecord:
    """Create a sample codemod run record."""
    return CodemodRunRecord(
        id=f"test-project-{test_id}:codemod_run:test-123",
        project=f"test-project-{test_id}",
        started_at=datetime.now(),
        completed_at=datetime.now(),
        status="success",
        dry_run=False,
        pass_count=3,
        finding_count=5,
        file_count=10,
        duration_ms=1500,
    )


# ============================================================================
# REQ-110-028: Protocol defines all required methods
# ============================================================================


def test_protocol_defines_save_run():
    """CodemodGraphRepository Protocol defines save_run()."""
    assert hasattr(CodemodGraphRepository, "save_run")
    assert callable(getattr(CodemodGraphRepository, "save_run"))


def test_protocol_defines_save_pass():
    """CodemodGraphRepository Protocol defines save_pass()."""
    assert hasattr(CodemodGraphRepository, "save_pass")
    assert callable(getattr(CodemodGraphRepository, "save_pass"))


def test_protocol_defines_save_findings():
    """CodemodGraphRepository Protocol defines save_findings()."""
    assert hasattr(CodemodGraphRepository, "save_findings")
    assert callable(getattr(CodemodGraphRepository, "save_findings"))


def test_protocol_defines_save_changes():
    """CodemodGraphRepository Protocol defines save_changes()."""
    assert hasattr(CodemodGraphRepository, "save_changes")
    assert callable(getattr(CodemodGraphRepository, "save_changes"))


def test_protocol_defines_link_changes_to_code():
    """CodemodGraphRepository Protocol defines link_changes_to_code()."""
    assert hasattr(CodemodGraphRepository, "link_changes_to_code")
    assert callable(getattr(CodemodGraphRepository, "link_changes_to_code"))


def test_protocol_defines_query_runs():
    """CodemodGraphRepository Protocol defines query_runs()."""
    assert hasattr(CodemodGraphRepository, "query_runs")
    assert callable(getattr(CodemodGraphRepository, "query_runs"))


def test_protocol_defines_query_findings():
    """CodemodGraphRepository Protocol defines query_findings()."""
    assert hasattr(CodemodGraphRepository, "query_findings")
    assert callable(getattr(CodemodGraphRepository, "query_findings"))


def test_protocol_defines_query_affected_code():
    """CodemodGraphRepository Protocol defines query_affected_code()."""
    assert hasattr(CodemodGraphRepository, "query_affected_code")
    assert callable(getattr(CodemodGraphRepository, "query_affected_code"))


def test_bolt_repository_implements_protocol():
    """BoltSyncCodemodRepository implements CodemodGraphRepository Protocol."""
    assert isinstance(BoltSyncCodemodRepository, type)
    # runtime_checkable allows isinstance check on instances
    # (not on the class itself)


# ============================================================================
# REQ-110-029: BoltSyncCodemodRepository writes nodes
# ============================================================================


def test_save_run_merges_codemod_run_node(graph_repo, sample_run):
    """BoltSyncCodemodRepository writes CodemodRun node."""
    graph_repo.save_run(sample_run)
    
    # Query the graph DB to verify the node was created
    cypher = """
    MATCH (r:CodemodRun {id: $id})
    RETURN r.started_at AS started_at,
           r.status AS status,
           r.pass_count AS pass_count
    """
    results = graph_repo._repo.execute_read(cypher, {"id": sample_run.id})
    
    assert len(results) == 1
    assert results[0]["status"] == "success"
    assert results[0]["pass_count"] == 3


def test_save_run_creates_has_codemod_run_edge(graph_repo, sample_run):
    """HAS_CODEMOD_RUN edge connects Project to CodemodRun."""
    graph_repo.save_run(sample_run)
    
    cypher = """
    MATCH (p:Project {slug: $project})-[:HAS_CODEMOD_RUN]->(r:CodemodRun {id: $id})
    RETURN count(r) AS count
    """
    results = graph_repo._repo.execute_read(
        cypher, {"project": graph_repo._project, "id": sample_run.id}
    )
    
    assert results[0]["count"] == 1


def test_save_pass_merges_codemod_pass_node(graph_repo, sample_run):
    """BoltSyncCodemodRepository writes CodemodPass node."""
    graph_repo.save_run(sample_run)
    
    pass_record = CodemodPassRecord(
        id=f"{sample_run.id}:pass:RenameSymbolPass",
        run_id=sample_run.id,
        pass_name="RenameSymbolPass",
        pass_type="transform",
        file_count=5,
        duration_ms=200,
    )
    graph_repo.save_pass(sample_run.id, pass_record)
    
    cypher = """
    MATCH (cp:CodemodPass {id: $id})
    RETURN cp.pass_name AS pass_name,
           cp.pass_type AS pass_type,
           cp.file_count AS file_count
    """
    results = graph_repo._repo.execute_read(cypher, {"id": pass_record.id})
    
    assert len(results) == 1
    assert results[0]["pass_name"] == "RenameSymbolPass"
    assert results[0]["pass_type"] == "transform"


def test_save_pass_creates_has_pass_edge(graph_repo, sample_run):
    """HAS_PASS edge connects CodemodRun to CodemodPass."""
    graph_repo.save_run(sample_run)
    
    pass_record = CodemodPassRecord(
        id=f"{sample_run.id}:pass:RenameSymbolPass",
        run_id=sample_run.id,
        pass_name="RenameSymbolPass",
        pass_type="transform",
        file_count=5,
        duration_ms=200,
    )
    graph_repo.save_pass(sample_run.id, pass_record)
    
    cypher = """
    MATCH (r:CodemodRun {id: $run_id})-[:HAS_PASS]->(cp:CodemodPass {id: $pass_id})
    RETURN count(cp) AS count
    """
    results = graph_repo._repo.execute_read(
        cypher, {"run_id": sample_run.id, "pass_id": pass_record.id}
    )
    
    assert results[0]["count"] == 1


def test_save_findings_batch_merges_nodes(graph_repo, sample_run):
    """BoltSyncCodemodRepository writes CodemodFinding nodes (batched)."""
    graph_repo.save_run(sample_run)
    
    findings = [
        CodemodFindingRecord(
            id=f"{sample_run.id}:finding:banned_construct:src/a.py:10",
            run_id=sample_run.id,
            severity="error",
            rule="banned_construct",
            path="src/a.py",
            line=10,
            message="Found banned construct",
            module_id=None,
        ),
        CodemodFindingRecord(
            id=f"{sample_run.id}:finding:complexity:src/b.py:20",
            run_id=sample_run.id,
            severity="warning",
            rule="complexity",
            path="src/b.py",
            line=20,
            message="High complexity",
            module_id=None,
        ),
    ]
    graph_repo.save_findings(sample_run.id, findings)
    
    cypher = """
    MATCH (r:CodemodRun {id: $run_id})-[:HAS_FINDING]->(cf:CodemodFinding)
    RETURN count(cf) AS count
    """
    results = graph_repo._repo.execute_read(cypher, {"run_id": sample_run.id})
    
    assert results[0]["count"] == 2


def test_save_findings_creates_about_edge_when_module_exists(graph_repo, sample_run):
    """ABOUT edge connects CodemodFinding to Module (if module_id provided)."""
    graph_repo.save_run(sample_run)
    
    # Create a Module node first
    module_id = f"{graph_repo._project}:src/test.py::TestClass"
    graph_repo._repo.execute_write(
        "MERGE (m:Module {id: $id})",
        {"id": module_id},
    )
    
    findings = [
        CodemodFindingRecord(
            id=f"{sample_run.id}:finding:test:src/test.py:1",
            run_id=sample_run.id,
            severity="warning",
            rule="test",
            path="src/test.py",
            line=1,
            message="Test finding",
            module_id=module_id,
        ),
    ]
    graph_repo.save_findings(sample_run.id, findings)
    
    cypher = """
    MATCH (cf:CodemodFinding {id: $finding_id})-[:ABOUT]->(m:Module {id: $module_id})
    RETURN count(m) AS count
    """
    results = graph_repo._repo.execute_read(
        cypher, {"finding_id": findings[0].id, "module_id": module_id}
    )
    
    assert results[0]["count"] == 1


def test_save_changes_batch_merges_nodes(graph_repo, sample_run):
    """BoltSyncCodemodRepository writes CodeChange nodes (batched)."""
    graph_repo.save_run(sample_run)
    
    # Create a pass first (PRODUCED_BY edge requires it)
    pass_record = CodemodPassRecord(
        id=f"{sample_run.id}:pass:RenameSymbolPass",
        run_id=sample_run.id,
        pass_name="RenameSymbolPass",
        pass_type="transform",
        file_count=2,
        duration_ms=100,
    )
    graph_repo.save_pass(sample_run.id, pass_record)
    
    changes = [
        CodeChangeRecord(
            id=f"{sample_run.id}:change:src/a.py",
            run_id=sample_run.id,
            pass_name="RenameSymbolPass",
            path="src/a.py",
            description="Renamed foo to bar",
            diff_hash="abc123",
            entity_id=None,
            breaking_change=False,
        ),
        CodeChangeRecord(
            id=f"{sample_run.id}:change:src/b.py",
            run_id=sample_run.id,
            pass_name="RenameSymbolPass",
            path="src/b.py",
            description="Updated import",
            diff_hash="def456",
            entity_id=None,
            breaking_change=False,
        ),
    ]
    graph_repo.save_changes(sample_run.id, changes)
    
    cypher = """
    MATCH (cc:CodeChange:Change)
    WHERE cc.id STARTS WITH $prefix
    RETURN count(cc) AS count
    """
    results = graph_repo._repo.execute_read(
        cypher, {"prefix": f"{sample_run.id}:change:"}
    )
    
    assert results[0]["count"] == 2


def test_link_changes_to_code_creates_breaking_change_edge(graph_repo, sample_run):
    """BREAKING_CHANGE edge connects CodeChange to affected entity."""
    graph_repo.save_run(sample_run)
    
    # Create a pass and change first
    pass_record = CodemodPassRecord(
        id=f"{sample_run.id}:pass:RenameSymbolPass",
        run_id=sample_run.id,
        pass_name="RenameSymbolPass",
        pass_type="transform",
        file_count=1,
        duration_ms=100,
    )
    graph_repo.save_pass(sample_run.id, pass_record)
    
    # Create a Module node (the affected entity)
    module_id = f"{graph_repo._project}:src/api.py::PublicAPI"
    graph_repo._repo.execute_write(
        "MERGE (m:Module {id: $id})",
        {"id": module_id},
    )
    
    change = CodeChangeRecord(
        id=f"{sample_run.id}:change:src/api.py",
        run_id=sample_run.id,
        pass_name="RenameSymbolPass",
        path="src/api.py",
        description="Removed parameter from public API",
        diff_hash="xyz789",
        entity_id=module_id,
        breaking_change=True,
    )
    graph_repo.save_changes(sample_run.id, [change])
    graph_repo.link_changes_to_code([change])
    
    cypher = """
    MATCH (cc:CodeChange {id: $change_id})-[:BREAKING_CHANGE]->(m:Module {id: $module_id})
    RETURN count(m) AS count
    """
    results = graph_repo._repo.execute_read(
        cypher, {"change_id": change.id, "module_id": module_id}
    )
    
    assert results[0]["count"] == 1


# ============================================================================
# REQ-110-030: Ontology extends canonical_schema.py
# ============================================================================


def test_canonical_schema_defines_codemod_labels():
    """canonical_schema.py defines codemod labels."""
    from orchestrata_ultimate.knowledge_graph import canonical_schema as cs
    
    assert hasattr(cs, "LABEL_CODEMOD_RUN")
    assert cs.LABEL_CODEMOD_RUN == "CodemodRun"
    
    assert hasattr(cs, "LABEL_CODEMOD_PASS")
    assert cs.LABEL_CODEMOD_PASS == "CodemodPass"
    
    assert hasattr(cs, "LABEL_CODEMOD_FINDING")
    assert cs.LABEL_CODEMOD_FINDING == "CodemodFinding"
    
    # CodeChange uses LABEL_CHANGE as sub-label (no separate constant)
    assert hasattr(cs, "LABEL_CHANGE")
    assert cs.LABEL_CHANGE == "Change"


def test_canonical_schema_defines_codemod_relationships():
    """canonical_schema.py defines codemod relationships."""
    from orchestrata_ultimate.knowledge_graph import canonical_schema as cs
    
    assert hasattr(cs, "REL_HAS_CODEMOD_RUN")
    assert cs.REL_HAS_CODEMOD_RUN == "HAS_CODEMOD_RUN"
    
    assert hasattr(cs, "REL_HAS_PASS")
    assert cs.REL_HAS_PASS == "HAS_PASS"
    
    assert hasattr(cs, "REL_HAS_FINDING")
    assert cs.REL_HAS_FINDING == "HAS_FINDING"
    
    assert hasattr(cs, "REL_ABOUT")
    assert cs.REL_ABOUT == "ABOUT"
    
    assert hasattr(cs, "REL_PRODUCED_BY")
    assert cs.REL_PRODUCED_BY == "PRODUCED_BY"
    
    assert hasattr(cs, "REL_BREAKING_CHANGE")
    assert cs.REL_BREAKING_CHANGE == "BREAKING_CHANGE"
    
    # REL_CHANGED_IN already exists (line 232)
    assert hasattr(cs, "REL_CHANGED_IN")
    assert cs.REL_CHANGED_IN == "CHANGED_IN"


# ============================================================================
# REQ-110-030: Query methods
# ============================================================================


def test_query_runs_returns_recent_runs(graph_repo):
    """query_runs() returns recent codemod runs."""
    # Create 5 runs
    for i in range(5):
        run = CodemodRunRecord(
            id=f"{graph_repo._project}:codemod_run:run-{i}",
            project=graph_repo._project,
            started_at=datetime(2026, 1, 1, i, 0, 0),
            completed_at=datetime(2026, 1, 1, i, 0, 0),
            status="success",
            dry_run=False,
            pass_count=1,
            finding_count=0,
            file_count=1,
            duration_ms=100,
        )
        graph_repo.save_run(run)
    
    # Query with limit=3
    results = graph_repo.query_runs(graph_repo._project, limit=3)
    
    assert len(results) == 3
    # Should be ordered by started_at DESC (most recent first)
    assert results[0].id.endswith("run-4")
    assert results[1].id.endswith("run-3")
    assert results[2].id.endswith("run-2")


def test_query_findings_returns_all_findings_for_run(graph_repo, sample_run):
    """query_findings() returns all findings for a codemod run."""
    graph_repo.save_run(sample_run)
    
    # Create 10 findings
    findings = [
        CodemodFindingRecord(
            id=f"{sample_run.id}:finding:rule{i}:src/file.py:{i}",
            run_id=sample_run.id,
            severity="warning",
            rule=f"rule{i}",
            path="src/file.py",
            line=i,
            message=f"Finding {i}",
            module_id=None,
        )
        for i in range(10)
    ]
    graph_repo.save_findings(sample_run.id, findings)
    
    results = graph_repo.query_findings(sample_run.id)
    
    assert len(results) == 10


def test_query_affected_code_returns_entity_ids(graph_repo, sample_run):
    """query_affected_code() returns all affected entity IDs."""
    graph_repo.save_run(sample_run)
    
    # Create a pass
    pass_record = CodemodPassRecord(
        id=f"{sample_run.id}:pass:TestPass",
        run_id=sample_run.id,
        pass_name="TestPass",
        pass_type="transform",
        file_count=3,
        duration_ms=100,
    )
    graph_repo.save_pass(sample_run.id, pass_record)
    
    # Create 3 Module nodes
    module_ids = []
    for i in range(3):
        module_id = f"{graph_repo._project}:src/module{i}.py::Class{i}"
        graph_repo._repo.execute_write(
            "MERGE (m:Module {id: $id})",
            {"id": module_id},
        )
        module_ids.append(module_id)
    
    # Create changes with entity_ids
    changes = [
        CodeChangeRecord(
            id=f"{sample_run.id}:change:src/module{i}.py",
            run_id=sample_run.id,
            pass_name="TestPass",
            path=f"src/module{i}.py",
            description=f"Changed module {i}",
            diff_hash=f"hash{i}",
            entity_id=module_ids[i],
            breaking_change=False,
        )
        for i in range(3)
    ]
    graph_repo.save_changes(sample_run.id, changes)
    
    results = graph_repo.query_affected_code(sample_run.id)
    
    assert len(results) == 3
    assert set(results) == set(module_ids)

Key points:

• Use test_id fixture for isolation (UUID-based).
• Create Project node in graph_repo fixture for testing.
• Test all 8 Protocol methods (REQ-110-028).
• Test Cypher MERGE patterns against real graph DB (REQ-110-029).
• Test query methods (REQ-110-030).
• Test canonical_schema.py extensions (REQ-110-030).
• Use graph_repo._repo to access underlying SyncGraphRepository for setup/verification.

Step 5: Update persistence/init.py

Update src/orchestrata_ultimate/refactor/codemod/persistence/__init__.py:

"""Codemod provenance persistence (graph database)."""

from orchestrata_ultimate.refactor.codemod.persistence.records import (
    CodemodRunRecord,
    CodemodPassRecord,
    CodemodFindingRecord,
    CodeChangeRecord,
)
from orchestrata_ultimate.refactor.codemod.persistence.repository import (
    CodemodGraphRepository,
)
from orchestrata_ultimate.refactor.codemod.persistence.bolt_repository import (
    BoltSyncCodemodRepository,
)

__all__ = [
    "CodemodRunRecord",
    "CodemodPassRecord",
    "CodemodFindingRecord",
    "CodeChangeRecord",
    "CodemodGraphRepository",
    "BoltSyncCodemodRepository",
]

Step 6: Verify Implementation

# Check that all files exist
ls -la src/orchestrata_ultimate/refactor/codemod/persistence/*.py

# Check that code is syntactically valid
uv run python -m py_compile src/orchestrata_ultimate/refactor/codemod/persistence/repository.py
uv run python -m py_compile src/orchestrata_ultimate/refactor/codemod/persistence/bolt_repository.py

# Check that imports work
uv run python -c "from orchestrata_ultimate.refactor.codemod.persistence import CodemodGraphRepository, BoltSyncCodemodRepository"

# Check that Protocol is runtime_checkable
uv run python -c "from orchestrata_ultimate.refactor.codemod.persistence import CodemodGraphRepository; print('Protocol:', CodemodGraphRepository)"

# Check that canonical_schema.py has new constants
uv run python -c "from orchestrata_ultimate.knowledge_graph import canonical_schema as cs; print('LABEL_CODEMOD_RUN:', cs.LABEL_CODEMOD_RUN)"

# Run the BDD tests
uv run pytest tests/refactor/codemod/test_graph_persistence.py -v

Success Criteria

• [ ] CodemodGraphRepository Protocol implemented with all 8 methods (REQ-110-028)
• [ ] BoltSyncCodemodRepository implements Protocol using unit_of_work() (REQ-110-029)
• [ ] Cypher MERGE patterns match spec.md section 9.8 exactly
• [ ] UNWIND batch operations for passes, findings, changes (performance)
• [ ] canonical_schema.py extended with new labels and relationships (REQ-110-030)
• [ ] CodeChange reuses LABEL_CHANGE as sub-label (REQ-110-030)
• [ ] All 11 BDD scenarios pass (test_graph_persistence.py)
• [ ] Query methods return correct results (query_runs, query_findings, query_affected_code)
• [ ] Code has @traces decorators with correct spec/story/reqs
• [ ] No hardcoded IPs or ports (use .env via create_sync_repository())

Deliverables

1. src/orchestrata_ultimate/refactor/codemod/persistence/repository.py (Protocol)
2. src/orchestrata_ultimate/refactor/codemod/persistence/bolt_repository.py (implementation)
3. src/orchestrata_ultimate/refactor/codemod/persistence/__init__.py (exports)
4. src/orchestrata_ultimate/knowledge_graph/canonical_schema.py (extended)
5. tests/refactor/codemod/test_graph_persistence.py (BDD tests)

Handoff to W5

W5 (C Extension) will implement:

• _codemod_fast sub-package with C extensions for performance-critical operations
• _csr_index_fast (C implementation of CSR adjacency arrays)
• _cig_fast (C implementation of CIG queries)
• _parse_fast (C implementation of parallel CST parsing)

W5 will use the persistence layer from W4 to record provenance for C extension operations.

Notes

• Single agent: This wave uses a single Qwen Max agent (no parallel execution).
• Uses SyncGraphRepository: Import from orchestrata_ultimate.core.graph.interfaces.
• Cypher patterns: Follow spec.md section 9.8 exactly (MERGE, UNWIND, OPTIONAL MATCH).
• CodeChange sub-label: Uses LABEL_CHANGE:CodeChange pattern (not a separate label).
• REL_CHANGED_IN already exists: Do NOT duplicate it in canonical_schema.py (line 232).
• Real graph DB: Tests require a real graph database connection (Neo4j or Memgraph).
• Transactional writes: All writes use unit_of_work() for atomicity.
• Batch operations: UNWIND for passes, findings, changes (performance optimization).
• Error handling: ERR-110-012 specifies log-and-continue for graph DB write failures (non-blocking observer pattern in W3).

Related

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