orchestrata_ultimate.knowledge_graph.synchronizer

Reference › orchestrata_ultimate › knowledge_graph › synchronizer

FileSynchronizer keeps the operational graph (orchkg on orchgraph) in step with source files on disk, one file or one project at a time (Spec 008, REQ-008-022 and REQ-008-025). It follows the goal-oriented, idempotent, unidirectional pattern named in its docstring. The kernel's pipeline_kg and file_watcher import it; the bulk, AST-JSON path is sync.

The constructor takes an optional GraphClient, a code_parser CodeParser, a SyncConfig, a synchronous repository and a project id. When a project id is set, file node ids are project-qualified (_file_node_id). sync_file parses a file and compares its hash with the stored one; if unchanged it returns early, otherwise _persist_to_graph writes the File node and, through _write_defines_batch, merged class and function nodes with their File-[:DEFINES]->node edges inside one unit of work. A retry decorator, retry_on_mvcc_conflict, covers Memgraph write conflicts. sync_project walks eligible files, while sync_project_with_resolution is two-phase: phase one parses and persists every file, phase two resolves call, inheritance, import and method edges across all analyses and writes them with _persist_resolution_edges and a ResolutionResult.

Incremental behaviour rests on a per-project hash state file that _save_state and _load_state persist, so unchanged files are skipped across restarts; needs_sync, get_file_hash and ensure_fresh (sync only changed files before a task, within the configured deadline) use it. watch returns a FileWatcher and stop_watching ends it. sync_with_audit syncs and then runs the anomaly_detector, optionally auto-correcting. clear resets state, and sync_count and tracked_files report counters.

At a glance

Class diagram

Dependency diagram

API

Classes

FileSynchronizer

Decorators: traces(spec='008', story='US-001', reqs=['REQ-008-022', 'REQ-008-025']) · Line 40

Synchronizes source files with the knowledge graph.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/knowledge_graph/synchronizer.py

# mypy: ignore-errors
"""File synchronizer for the knowledge graph.

Watches files, parses them, and updates the graph.
Follows XOS pattern: goal-oriented, idempotent, unidirectional.
"""

from __future__ import annotations

import asyncio
import json
import logging
import time
from pathlib import Path
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph.retry import retry_on_mvcc_conflict
from orchestrata_ultimate.knowledge_graph.code_parser import CodeParser
from orchestrata_ultimate.knowledge_graph.config import SyncConfig
from orchestrata_ultimate.knowledge_graph.types import (
    FileAnalysis,
    FileNode,
    ResolutionResult,
    SyncResult,
    SyncStatus,
)

if TYPE_CHECKING:
    from orchestrata_ultimate.core.graph import SyncGraphRepository
    from orchestrata_ultimate.knowledge_graph.file_watcher import FileWatcher
    from orchestrata_ultimate.knowledge_graph.graph_client import GraphClient

_logger = logging.getLogger(__name__)

logger = logging.getLogger(__name__)


@traces(spec="008", story="US-001", reqs=["REQ-008-022", "REQ-008-025"])
class FileSynchronizer:
    """Synchronizes source files with the knowledge graph.

    Parses files and creates/updates nodes and relationships
    in the graph. Supports full project sync and incremental
    file-level sync.
    """

    def __init__(
        self,
        graph_client: GraphClient | None = None,
        parser: CodeParser | None = None,
        config: SyncConfig | None = None,
        repo: SyncGraphRepository | None = None,
        project_id: str | None = None,
    ) -> None:
        self._client = graph_client
        self._parser = parser or CodeParser()
        self._config = config or SyncConfig()
        self._repo = repo
        self._project_id = project_id
        self._file_hashes: dict[str, str] = {}
        self._sync_count = 0
        self._watching = False
        self._watcher = None

    @traces(spec="008", story="US-001", reqs=["REQ-008-021", "REQ-008-023"])
    def _file_node_id(self, file_path: str) -> str:
        """Canonical File node id: project-qualified when a project_id is set."""
        if self._project_id:
            return f"{self._project_id}:{file_path}"
        return file_path

    async def connect(self) -> None:
        """Ensure the underlying graph client is connected."""
        if (
            self._client is not None
            and hasattr(self._client, "connect")
            and not self._client.connected
        ):
            await self._client.connect()

    @traces(spec="008", story="US-001", reqs=["REQ-008-021", "REQ-008-023"])
    def sync_file(self, file_path: str) -> SyncResult:
        """Sync a single file to the knowledge graph."""
        start = time.perf_counter()
        path = Path(file_path)

        if not path.exists():
            return SyncResult(
                file_path=file_path,
                status=SyncStatus.FAILED,
                error=f"File not found: {file_path}",
            )

        # Check if file has changed
        current_hash = self._parser.get_file_hash(file_path)
        if current_hash == self._file_hashes.get(file_path):
            return SyncResult(
                file_path=file_path,
                status=SyncStatus.SKIPPED,
                duration_ms=(time.perf_counter() - start) * 1000,
            )

        # Parse file
        analysis = self._parser.parse_file(file_path)
        if not analysis.success:
            return SyncResult(
                file_path=file_path,
                status=SyncStatus.FAILED,
                error="; ".join(analysis.errors),
                duration_ms=(time.perf_counter() - start) * 1000,
            )

        # Write to the graph if client is available
        nodes_created = 0
        rels_created = 0

        # Persist when EITHER a GraphClient or a sync repository is supplied.
        # `_persist_to_graph` writes via `_acquire_sync_repo()` (which prefers
        # `self._repo`), so a `repo=`-only synchronizer must still write —
        # gating on `self._client` alone silently dropped every node.
        if self._client is not None or self._repo is not None:
            try:
                nodes_created, rels_created = self._persist_with_retry(
                    file_path, path, analysis, current_hash
                )
            except Exception as exc:
                _logger.warning("Graph persist failed: %s", exc)
                return SyncResult(
                    file_path=file_path,
                    status=SyncStatus.FAILED,
                    error=str(exc),
                    duration_ms=(time.perf_counter() - start) * 1000,
                )
        else:
            # No client — count only (backwards compat)
            nodes_created = (
                1
                + len(analysis.classes)
                + len(analysis.functions)
                + len(analysis.predicates)
            )
            rels_created = len(analysis.imports)

        # Update hash
        self._file_hashes[file_path] = current_hash
        self._sync_count += 1

        elapsed = (time.perf_counter() - start) * 1000

        return SyncResult(
            file_path=file_path,
            status=SyncStatus.COMPLETED,
            nodes_created=nodes_created,
            nodes_updated=0,
            relationships_created=rels_created,
            duration_ms=elapsed,
        )

    @traces(spec="008", story="US-001", reqs=["REQ-008-021", "REQ-008-023"])
    def _sync_file_from_analysis(
        self, file_path: str, analysis: FileAnalysis
    ) -> SyncResult:
        """Sync a file using an already-parsed FileAnalysis.

        Same logic as sync_file() but skips re-parsing — uses the
        provided analysis so that node IDs are consistent between
        persistence and resolution.
        """
        start = time.perf_counter()
        path = Path(file_path)

        if not path.exists():
            return SyncResult(
                file_path=file_path,
                status=SyncStatus.FAILED,
                error=f"File not found: {file_path}",
            )

        if not analysis.success:
            return SyncResult(
                file_path=file_path,
                status=SyncStatus.FAILED,
                error="; ".join(analysis.errors),
                duration_ms=(time.perf_counter() - start) * 1000,
            )

        current_hash = self._parser.get_file_hash(file_path)
        if current_hash == self._file_hashes.get(file_path):
            return SyncResult(
                file_path=file_path,
                status=SyncStatus.SKIPPED,
                duration_ms=(time.perf_counter() - start) * 1000,
            )

        nodes_created = 0
        rels_created = 0

        # Persist when EITHER a GraphClient or a sync repository is supplied.
        # `_persist_to_graph` writes via `_acquire_sync_repo()` (which prefers
        # `self._repo`), so a `repo=`-only synchronizer must still write —
        # gating on `self._client` alone silently dropped every node.
        if self._client is not None or self._repo is not None:
            try:
                nodes_created, rels_created = self._persist_with_retry(
                    file_path, path, analysis, current_hash
                )
            except Exception as exc:
                _logger.warning("Graph persist failed: %s", exc)
                return SyncResult(
                    file_path=file_path,
                    status=SyncStatus.FAILED,
                    error=str(exc),
                    duration_ms=(time.perf_counter() - start) * 1000,
                )
        else:
            nodes_created = (
                1
                + len(analysis.classes)
                + len(analysis.functions)
                + len(analysis.predicates)
            )
            rels_created = len(analysis.imports)

        self._file_hashes[file_path] = current_hash
        self._sync_count += 1
        elapsed = (time.perf_counter() - start) * 1000

        return SyncResult(
            file_path=file_path,
            status=SyncStatus.COMPLETED,
            nodes_created=nodes_created,
            nodes_updated=0,
            relationships_created=rels_created,
            duration_ms=elapsed,
        )

    @traces(spec="008", story="US-001", reqs=["REQ-008-021", "REQ-008-023"])
    def _persist_with_retry(
        self,
        file_path: str,
        path: Path,
        analysis: FileAnalysis,
        current_hash: str,
    ) -> tuple[int, int]:
        """``_persist_to_graph`` with the same 3-attempt transient-error
        retry ``GraphClient`` gives its own queries.

        The repository ``unit_of_work`` path bypasses the client, so a
        single transient lock/leader error under parallel-suite load used
        to fail the whole sync (surfaced by the anomaly-detector
        auto-correction test: ``corrected=False`` on a transient).
        Non-transient errors raise immediately.
        """
        from orchestrata_ultimate.knowledge_graph.graph_client import GraphClient

        last_exc: Exception | None = None
        for attempt in range(3):
            try:
                return self._persist_to_graph(file_path, path, analysis, current_hash)
            except Exception as exc:
                if not GraphClient._is_transient_error(str(exc)):
                    raise
                last_exc = exc
                _logger.warning(
                    "transient graph persist error (attempt %d/3): %s",
                    attempt + 1,
                    exc,
                )
                time.sleep(0.2 * (attempt + 1))
        raise last_exc  # type: ignore[misc]

    @traces(spec="008", story="US-001", reqs=["REQ-008-021", "REQ-008-023"])
    def _persist_to_graph(
        self,
        file_path: str,
        path: Path,
        analysis: FileAnalysis,
        current_hash: str,
    ) -> tuple[int, int]:
        """Write parsed entities to the graph via the repository seam.

        Uses the repository's ``unit_of_work`` so the File node plus every
        entity batch write in one transaction — the same one-transaction
        guarantee the pre-seam single ``driver.session()`` block gave.

        Args:
            file_path: Relative file path.
            path: Path object.
            analysis: Parsed file analysis.
            current_hash: File content hash.

        Returns:
            (nodes_created, relationships_created)
        """
        from orchestrata_ultimate.knowledge_graph.nodes import file_node_to_cypher

        repo, should_close = self._acquire_sync_repo()

        file_id = self._file_node_id(file_path)
        file_node = FileNode(
            id=file_id,
            path=file_path,
            name=path.name,
            language=analysis.language,
            line_count=analysis.line_count,
            size_bytes=path.stat().st_size if path.exists() else 0,
            hash=current_hash,
            project_id=self._project_id,
        )

        # Prepare all entity data with file_path baked in
        class_rows = self._entity_rows(analysis.classes, "Class", file_path)
        func_rows = self._entity_rows(analysis.functions, "Function", file_path)
        pred_rows = self._entity_rows(analysis.predicates, "Predicate", file_path)
        rule_rows = self._entity_rows(analysis.rules, "Rule", file_path)

        nc = 0
        rc = 0

        _, file_props = file_node_to_cypher(file_node)

        with repo.unit_of_work() as uow:
            # 1. File node — single MERGE keyed by canonical id
            uow.run(
                "MERGE (f:File {id: $id}) SET f = $props",
                {"id": file_id, "props": file_props},
            )
            nc += 1

            # 2-5. Classes, Functions, Predicates, Rules — batched UNWIND,
            #      node + DEFINES from File in one query each.
            for rows, label, var in (
                (class_rows, "Class", "c"),
                (func_rows, "Function", "fn"),
                (pred_rows, "Predicate", "p"),
                (rule_rows, "Rule", "rl"),
            ):
                written = self._write_defines_batch(uow, rows, label, var)
                nc += written
                rc += written

        if should_close:
            repo.close()
        return nc, rc

    @traces(spec="008", story="US-001", reqs=["REQ-008-021", "REQ-008-023"])
    def _acquire_sync_repo(self) -> tuple[SyncGraphRepository, bool]:
        """Return a sync graph repository and whether the caller must close it.

        Reuses ``self._repo`` if present (shared — never closed here, its
        lifecycle belongs to whoever injected it). Otherwise builds a fresh,
        OWNED ``BoltSyncRepository`` from the async client's config
        (uri/auth) — the repository's own owned-driver semantics replace the
        old ``(driver, should_close)`` manual-close protocol, but the caller
        still must close THIS repository after use since it isn't retained
        on ``self``.
        """
        from orchestrata_ultimate.core.graph import (
            BoltSyncRepository,
            GraphRepositoryConfig,
        )

        if self._repo is not None:
            return self._repo, False

        client_config = getattr(self._client, "_config", None)
        if client_config is None:
            raise RuntimeError("Graph client has no _config")

        repo = BoltSyncRepository(
            GraphRepositoryConfig(
                uri=client_config.uri,
                auth=(client_config.username, client_config.password),
            )
        )
        return repo, True

    @traces(spec="008", story="US-001", reqs=["REQ-008-021", "REQ-008-023"])
    def _entity_rows(
        self, entities: list[Any], label: str, file_path: str
    ) -> list[dict[str, Any]]:
        """Serialize parsed entities to Cypher prop dicts.

        Class/Function/Predicate rows get ``_file_path`` baked in so the
        DEFINES edge can MATCH the owning File. Rule rows are left unchanged
        (rule_node_to_cypher already determines their own scope), preserving
        the original behavior where Rules carry no injected ``_file_path``.
        """
        from orchestrata_ultimate.knowledge_graph.nodes import (
            class_node_to_cypher,
            function_node_to_cypher,
            predicate_node_to_cypher,
            rule_node_to_cypher,
        )

        to_cypher = {
            "Class": class_node_to_cypher,
            "Function": function_node_to_cypher,
            "Predicate": predicate_node_to_cypher,
            "Rule": rule_node_to_cypher,
        }[label]

        rows: list[dict[str, Any]] = []
        for entity in entities:
            _, props = to_cypher(entity)
            if label != "Rule":
                props["_file_path"] = file_path
                props["_file_id"] = self._file_node_id(file_path)
            rows.append(props)
        return rows

    @traces(spec="008", story="US-001", reqs=["REQ-008-021", "REQ-008-023"])
    @staticmethod
    def _write_defines_batch(
        uow: Any, rows: list[dict[str, Any]], label: str, var: str
    ) -> int:
        """MERGE a batch of nodes and their File-[:DEFINES]->node edges.

        ``uow`` is a ``core.graph.interfaces.UnitOfWork`` — one statement in
        the caller's shared transaction.

        Returns the number of nodes written (0 when ``rows`` is empty).
        """
        if not rows:
            return 0
        uow.run(
            f"UNWIND $rows AS r "
            f"MERGE ({var}:{label} {{id: r.id}}) SET {var} = r "
            f"WITH {var}, r "
            f"MATCH (f:File {{id: r._file_id}}) "
            f"MERGE (f)-[:DEFINES]->({var})",
            {"rows": rows},
        )
        return len(rows)

    @traces(spec="008", story="US-001", reqs=["REQ-008-024", "REQ-008-050"])
    def sync_project(
        self,
        project_path: str,
        extensions: set[str] | None = None,
    ) -> list[SyncResult]:
        """Sync all files in a project directory.

        Loads persisted hash state at the start and saves it at the end so
        subsequent ``ensure_fresh`` calls only re-sync changed files.
        """
        exts = extensions or {".py", ".pl"}
        ignore = set(self._config.ignore_patterns)
        project = Path(project_path)

        if not project.is_dir():
            return [
                SyncResult(
                    file_path=project_path,
                    status=SyncStatus.FAILED,
                    error="Not a directory",
                )
            ]

        self._load_state(project_path)
        results: list[SyncResult] = []
        file_count = 0

        for source_file in sorted(project.rglob("*")):
            if source_file.suffix not in exts:
                continue
            if any(part in ignore for part in source_file.parts):
                continue
            if file_count >= self._config.max_files:
                break

            result = self.sync_file(str(source_file))
            results.append(result)
            file_count += 1

        self._save_state(project_path)
        return results

    @traces(
        spec="008",
        story="US-004",
        reqs=[
            "REQ-008-056",
            "REQ-008-057",
            "REQ-008-062",
            "REQ-008-063",
            "REQ-008-067",
            "REQ-008-068",
            "REQ-008-072",
            "REQ-008-073",
        ],
    )
    def sync_project_with_resolution(
        self,
        project_path: str,
        extensions: set[str] | None = None,
    ) -> ResolutionResult:
        """Two-phase sync: load all nodes first, then resolve edges.

        Phase 1 — Parse & persist nodes (FK-first):
            For each file: parse → create File, Class, Function,
            Predicate, Rule nodes with DEFINES edges. Collect all
            FileAnalysis objects.

        Phase 2 — Resolve cross-references:
            Run resolve_calls(), resolve_inheritance(),
            resolve_imports() on collected analyses. Batch UNWIND
            all edges into the graph.

        Args:
            project_path: Root directory to sync.
            extensions: File extensions to include.

        Returns:
            ResolutionResult with sync results and edge counts.
        """
        start = time.perf_counter()

        project = Path(project_path)
        if not project.is_dir():
            return ResolutionResult(
                sync_results=[
                    SyncResult(
                        file_path=project_path,
                        status=SyncStatus.FAILED,
                        error="Not a directory",
                    )
                ],
            )

        # Phase 1: Parse all eligible files and persist nodes.
        sync_results, all_analyses = self._parse_and_persist_project(
            project, extensions
        )

        # Phase 2: Resolve cross-references
        call_edges = self._parser.resolve_calls(all_analyses)
        inherit_edges = self._parser.resolve_inheritance(all_analyses)
        import_edges = self._parser.resolve_imports(all_analyses)
        method_edges = self._collect_method_edges(all_analyses)

        # Collect file paths synced in this batch (for scoped resolution)
        batch_file_paths = [
            r.file_path for r in sync_results if r.status != SyncStatus.FAILED
        ]

        # Persist edges to the graph
        edges_created = 0
        persist_failed = False
        persist_error: str | None = None
        if self._client is not None:
            try:
                # The retry wraps this CALL, not unit_of_work itself.
                # bolt_sync.SyncGraphRepository.unit_of_work is deliberately
                # NOT MVCC-retried (see its docstring at
                # core/graph/adapters/bolt_sync.py:225-233): "retrying a
                # multi-statement transaction correctly requires re-running
                # the CALLER's business logic inside the `with` block, not
                # just the transaction machinery". call_edges/inherit_edges/
                # import_edges/method_edges were computed above, BEFORE any
                # transaction opens, so re-invoking _persist_resolution_edges
                # here re-runs exactly that business logic — and every
                # statement it issues is a MERGE, so re-running it on an
                # MVCC abort is idempotent.
                edges_created = retry_on_mvcc_conflict(
                    lambda: self._persist_resolution_edges(
                        call_edges,
                        inherit_edges,
                        import_edges,
                        method_edges,
                        batch_file_paths,
                    )
                )
            except Exception as exc:
                persist_failed = True
                persist_error = str(exc)
                _logger.error("Resolution edge persist failed: %s", exc)

        elapsed = (time.perf_counter() - start) * 1000

        return ResolutionResult(
            sync_results=sync_results,
            call_edges=len(call_edges),
            inherit_edges=len(inherit_edges),
            import_edges=len(import_edges),
            method_edges=len(method_edges),
            total_edges_created=edges_created,
            duration_ms=elapsed,
            persist_failed=persist_failed,
            persist_error=persist_error,
        )

    @traces(spec="008", story="US-004", reqs=["REQ-008-021", "REQ-008-023"])
    def _parse_and_persist_project(
        self,
        project: Path,
        extensions: set[str] | None,
    ) -> tuple[list[SyncResult], list[FileAnalysis]]:
        """Phase 1: walk eligible files, parse each once, persist its nodes.

        Each file is parsed a single time and the same FileAnalysis is reused
        for both persistence and later cross-reference resolution.

        Returns:
            (sync_results, successful FileAnalysis objects).
        """
        exts = extensions or {".py", ".pl"}
        ignore = set(self._config.ignore_patterns)

        sync_results: list[SyncResult] = []
        all_analyses: list[FileAnalysis] = []
        file_count = 0

        for source_file in sorted(project.rglob("*")):
            if source_file.suffix not in exts:
                continue
            if any(part in ignore for part in source_file.parts):
                continue
            if file_count >= self._config.max_files:
                break

            file_path_str = str(source_file)
            analysis = self._parser.parse_file(file_path_str)
            if analysis.success:
                all_analyses.append(analysis)

            result = self._sync_file_from_analysis(file_path_str, analysis)
            sync_results.append(result)
            file_count += 1

        return sync_results, all_analyses

    @staticmethod
    @traces(spec="008", story="US-004", reqs=["REQ-008-073"])
    def _collect_method_edges(
        all_analyses: list[FileAnalysis],
    ) -> list[dict[str, str]]:
        """Collect DEFINES edges from functions carrying a class_id."""
        method_edges: list[dict[str, str]] = []
        for analysis in all_analyses:
            for func in analysis.functions:
                if func.class_id:
                    method_edges.append(
                        {"class_id": func.class_id, "function_id": func.id}
                    )
        return method_edges

    @traces(
        spec="008",
        story="US-004",
        reqs=["REQ-008-057", "REQ-008-063", "REQ-008-068", "REQ-008-073"],
    )
    def _persist_resolution_edges(
        self,
        call_edges: list[dict],
        inherit_edges: list[dict],
        import_edges: list[dict],
        method_edges: list[dict],
        batch_file_paths: list[str] | None = None,
    ) -> int:
        """Batch UNWIND every resolved edge, tiered by precedence.

        Two-tier resolution — precedence plus provenance. The precise
        resolution wins; the approximate one fills only the gaps it alone
        can reach, and every edge records which tier produced it via
        ``r.resolution``:

        1. Tier 1 (scoped) — DEFINES, CALLS, and INHERITS edges the Python
           resolver already computed over this batch's FileAnalysis
           objects are persisted via UNWIND, matching BOTH endpoints BY
           NODE ID. No name ever appears in these queries. CALLS edges get
           ``resolution='scoped'`` plus the resolver's per-edge
           ``confidence``; INHERITS edges get ``resolution='scoped'``
           only — resolve_inheritance computes no confidence to carry.
        2. Tier 2 (name-match) — a Cypher fallback, scoped to batch files
           on the caller/child side but resolving the callee/parent side
           by NAME across the whole graph. This is the only way to reach
           a callee/parent that is already in the graph but absent from
           this batch (the Python resolver's name index is built solely
           from ``all_analyses``, i.e. this batch). Tier 2 is narrowed to
           skip every ``(caller_id, call_name)`` / ``(child_id,
           base_name)`` pair Tier 1 already resolved — it may only touch
           names the scoped resolver could not reach at all. Edges it
           creates get ``resolution='name-match'``, and a non-zero count
           is logged at WARNING so the approximation is visible in
           operation, not silent.

        Args:
            call_edges: CALLS edges from the Python resolver — dicts with
                caller_id, callee_id, confidence, call_name.
            inherit_edges: INHERITS edges from the Python resolver — dicts
                with child_id, parent_id, base_name.
            import_edges: IMPORTS edges from Python resolver.
            method_edges: DEFINES edges from Python resolver.
            batch_file_paths: File paths synced in this batch. Used to
                scope graph-wide resolution queries to avoid scanning
                the entire graph.

        Returns:
            Edges ASSERTED — MERGE counts matched-or-created, so this is
            not strictly "newly created" (re-syncing unchanged source can
            report the same edges again). Tier 1 and Tier 2 never
            double-count the same pair: Tier 2 explicitly excludes
            whatever Tier 1 already resolved.
        """
        repo, should_close = self._acquire_sync_repo()

        total = 0
        paths = batch_file_paths or []
        file_ids = [self._file_node_id(p) for p in paths]

        import_rows = import_edges
        if import_edges:
            import_rows = [
                {
                    **r,
                    "source_file_id": self._file_node_id(r["source_file"]),
                    "target_file_id": self._file_node_id(r["target_file"]),
                }
                for r in import_edges
            ]

        # Pairs Tier 1 already resolved. Tier 2 must not re-approximate
        # these by name — that would either double-count the edge or,
        # worse, silently overwrite a precise resolution with a guess.
        resolved_call_pairs = [
            f"{r['caller_id']}::{r['call_name']}" for r in call_edges
        ]
        resolved_inherit_pairs = [
            f"{r['child_id']}::{r['base_name']}" for r in inherit_edges
        ]

        name_matched_calls = 0
        name_matched_inherits = 0

        with repo.unit_of_work() as uow:
            # ── Tier 1: scoped edges (UNWIND from the Python resolver,
            #    matched by node id — never by name) ──

            # DEFINES edges — Class -> Function (class_id is set during
            # parsing so we know both endpoints)
            if method_edges:
                uow.run(
                    "UNWIND $rows AS r "
                    "MATCH (c:Class {id: r.class_id}) "
                    "MATCH (fn:Function {id: r.function_id}) "
                    "MERGE (c)-[:DEFINES]->(fn)",
                    {"rows": method_edges},
                )
                total += len(method_edges)

            # CALLS — both endpoints matched by id; the resolver's
            # per-edge confidence is carried through onto the relationship.
            if call_edges:
                result = uow.run(
                    "UNWIND $rows AS r "
                    "MATCH (caller:Function {id: r.caller_id}) "
                    "MATCH (callee:Function {id: r.callee_id}) "
                    "MERGE (caller)-[rel:CALLS]->(callee) "
                    "SET rel.confidence = r.confidence, rel.resolution = 'scoped' "
                    "RETURN count(rel) AS cnt",
                    {"rows": call_edges},
                )
                if result:
                    total += result[0]["cnt"]

            # INHERITS — both endpoints matched by id.
            if inherit_edges:
                result = uow.run(
                    "UNWIND $rows AS r "
                    "MATCH (child:Class {id: r.child_id}) "
                    "MATCH (parent:Class {id: r.parent_id}) "
                    "MERGE (child)-[rel:INHERITS]->(parent) "
                    "SET rel.resolution = 'scoped' "
                    "RETURN count(rel) AS cnt",
                    {"rows": inherit_edges},
                )
                if result:
                    total += result[0]["cnt"]

            # ── Tier 2: name-match fallback (batch files only), narrowed
            #    to the pairs Tier 1 could not reach ──

            if file_ids:
                # CALLS: scope callers to batch files, resolve callees
                # against all Function nodes (including pre-existing) —
                # but skip any (caller, call_name) Tier 1 already handled.
                result = uow.run(
                    "UNWIND $file_ids AS fid "
                    "MATCH (f:File {id: fid})-[:DEFINES]->(caller:Function) "
                    "WHERE caller.calls IS NOT NULL "
                    "  AND size(caller.calls) > 0 "
                    "UNWIND caller.calls AS call_name "
                    "WITH caller, call_name, "
                    "     caller.id + '::' + call_name AS pair_key "
                    "WHERE NOT pair_key IN $resolved_pairs "
                    "MATCH (callee:Function {name: call_name}) "
                    "WHERE caller.id <> callee.id "
                    "MERGE (caller)-[r:CALLS]->(callee) "
                    "SET r.resolution = 'name-match' "
                    "RETURN count(r) AS cnt",
                    {"file_ids": file_ids, "resolved_pairs": resolved_call_pairs},
                )
                if result:
                    name_matched_calls = result[0]["cnt"]
                    total += name_matched_calls

                # INHERITS: scope children to batch files, resolve parents
                # against all Class nodes (including pre-existing) — but
                # skip any (child, base_name) Tier 1 already handled.
                result = uow.run(
                    "UNWIND $file_ids AS fid "
                    "MATCH (f:File {id: fid})-[:DEFINES]->(child:Class) "
                    "WHERE child.base_classes IS NOT NULL "
                    "  AND size(child.base_classes) > 0 "
                    "UNWIND child.base_classes AS base_name "
                    "WITH child, base_name, "
                    "     child.id + '::' + base_name AS pair_key "
                    "WHERE NOT pair_key IN $resolved_pairs "
                    "MATCH (parent:Class {name: base_name}) "
                    "WHERE child.id <> parent.id "
                    "MERGE (child)-[r:INHERITS]->(parent) "
                    "SET r.resolution = 'name-match' "
                    "RETURN count(r) AS cnt",
                    {"file_ids": file_ids, "resolved_pairs": resolved_inherit_pairs},
                )
                if result:
                    name_matched_inherits = result[0]["cnt"]
                    total += name_matched_inherits

            # IMPORTS: resolve via import_path matching.
            # Each File node stores its path. ImportNodes are not persisted
            # as separate nodes — they exist in FileAnalysis only. So we
            # use the batch-local import_edges for IMPORTS (these already
            # include module->file resolution from Python-side).
            if import_rows:
                uow.run(
                    "UNWIND $rows AS r "
                    "MATCH (src:File {id: r.source_file_id}) "
                    "MATCH (tgt:File {id: r.target_file_id}) "
                    "MERGE (src)-[i:IMPORTS]->(tgt) "
                    "SET i.imported_names = r.imported_names, "
                    "    i.line_number = r.line_number, "
                    "    i.is_relative = r.is_relative",
                    {"rows": import_rows},
                )
                total += len(import_rows)

        if name_matched_calls or name_matched_inherits:
            _logger.warning(
                "Name-match fallback asserted %d CALLS + %d INHERITS edge(s) "
                "(project=%s) that the scoped resolver could not reach — "
                "the callee/parent was outside this batch.",
                name_matched_calls,
                name_matched_inherits,
                self._project_id or "<unset>",
            )

        if should_close:
            repo.close()
        return total

    @traces(spec="008", story="US-001", reqs=["REQ-008-019"])
    def get_file_hash(self, file_path: str) -> str | None:
        """Get the stored hash for a file."""
        return self._file_hashes.get(file_path)

    @traces(spec="008", story="US-001", reqs=["REQ-008-019"])
    def needs_sync(self, file_path: str) -> bool:
        """Check if a file needs to be synced."""
        current_hash = self._parser.get_file_hash(file_path)
        return current_hash != self._file_hashes.get(file_path)

    @traces(spec="008", story="US-001", reqs=["REQ-008-019"])
    def _state_file(self, project_path: str) -> Path:
        """Path to the persisted hash-state file for a project."""
        return Path(project_path) / ".orchestrata" / "kg_sync_state.json"

    @traces(spec="008", story="US-001", reqs=["REQ-008-019"])
    def _load_state(self, project_path: str) -> None:
        """Load per-file hashes from disk so incremental sync survives restarts."""
        state_file = self._state_file(project_path)
        if not state_file.exists():
            return
        try:
            data = json.loads(state_file.read_text(encoding="utf-8"))
            self._file_hashes.update(data.get("hashes", {}))
            _logger.debug("Loaded KG sync state for %r", project_path)
        except Exception as exc:
            _logger.warning("Failed to load KG sync state: %s", exc)

    @traces(spec="008", story="US-001", reqs=["REQ-008-019"])
    def _save_state(self, project_path: str) -> None:
        """Persist per-file hashes to disk."""
        state_file = self._state_file(project_path)
        try:
            state_file.parent.mkdir(parents=True, exist_ok=True)
            state_file.write_text(
                json.dumps(
                    {"hashes": self._file_hashes, "updated_at": time.time()},
                    indent=2,
                    sort_keys=True,
                ),
                encoding="utf-8",
            )
            _logger.debug("Saved KG sync state for %r", project_path)
        except Exception as exc:
            _logger.warning("Failed to save KG sync state: %s", exc)

    @traces(spec="008", story="US-005", reqs=["REQ-008-026"])
    def watch(self, project_path: str = "") -> FileWatcher:
        """Create and return a FileWatcher for the given project.

        Args:
            project_path: Root directory to watch.

        Returns:
            A FileWatcher instance ready to be started.
        """
        from orchestrata_ultimate.knowledge_graph.file_watcher import FileWatcher

        self._watching = True
        watcher = FileWatcher(
            project_path=project_path,
            synchronizer=self,
            graph_client=self._client,
            config=self._config,
        )
        self._watcher = watcher
        return watcher

    @traces(spec="008", story="US-005", reqs=["REQ-008-026"])
    def stop_watching(self) -> None:
        """Stop watching for file changes."""
        self._watching = False
        self._watcher = None

    @property
    def is_watching(self) -> bool:
        return self._watching

    @property
    @traces(spec="002", story="US-008", reqs=["REQ-002-US-008-008"])
    def sync_count(self) -> int:
        return self._sync_count

    @property
    def tracked_files(self) -> int:
        return len(self._file_hashes)

    @traces(spec="008", story="FR-010", reqs=["REQ-008-025"])
    async def sync_with_audit(
        self,
        project_path: str,
        project_id: str,
        auto_correct: bool = False,
        event_loop: asyncio.AbstractEventLoop | None = None,
    ) -> tuple:
        """Sync project then run anomaly audit.

        Performs a full project sync followed by anomaly detection.
        Optionally auto-corrects any detected anomalies.

        Args:
            project_path: Root directory to sync and audit.
            project_id: Graph project node ID.
            auto_correct: If True, auto-correct anomalies.
            event_loop: Optional event loop for async ops.

        Returns:
            Tuple of (list[SyncResult], AnomalyReport).

        Raises:
            RuntimeError: If no graph client is connected (audit requires
                a live connection; unlike ``sync_file``/``sync_project``
                there is no client-less counting fallback for anomaly
                detection).
        """
        from orchestrata_ultimate.knowledge_graph.anomaly_detector import (
            AnomalyDetector,
        )

        if self._client is None:
            raise RuntimeError(
                "Graph client is not connected; cannot run anomaly audit"
            )

        sync_results = self.sync_project(project_path)
        mode = "auto" if auto_correct else "report"
        detector = AnomalyDetector(
            graph_client=self._client,
            synchronizer=self,
            project_id=project_id,
            event_loop=event_loop,
        )
        report = await detector.audit(project_path, mode=mode)
        return (sync_results, report)

    @traces(spec="008", story="US-005", reqs=["REQ-008-022", "REQ-008-056"])
    def ensure_fresh(self, project_path: str) -> int:
        """Sync only changed files before a task.

        Lightweight pre-task hook: iterates project files, checks hashes,
        re-syncs only files that changed since last sync. Does NOT run
        full resolution — that is deferred to sync_project_with_resolution.

        Bounded like ``sync_project``: honours ``SyncConfig.max_files``
        (files considered, not just re-synced) and a wall-clock deadline
        (``SyncConfig.ensure_fresh_deadline_seconds``, checked every
        ``SyncConfig.batch_size`` files). On a cold cache — no persisted
        hash state — every eligible file needs a real graph write, so
        max_files alone does not bound wall time; exceeding either budget
        degrades to a partial sync instead of hanging, since this hook
        feeds context injection, an optimisation for which partial
        context is correct and a multi-minute stall is not.

        Hash state is loaded from disk at the start and saved at the end so
        incremental sync survives process restarts — including a sync that
        stopped early on a budget: whatever was re-synced before the break
        is still persisted.

        Args:
            project_path: Root directory to check.

        Returns:
            Number of files re-synced.
        """
        exts = self._config.watch_extensions or {".py", ".pl"}
        ignore = set(self._config.ignore_patterns)
        project = Path(project_path)
        if not project.is_dir():
            return 0

        self._load_state(project_path)
        resynced = 0
        file_count = 0
        start = time.monotonic()
        deadline = start + self._config.ensure_fresh_deadline_seconds
        budget_exceeded = False
        for source_file in sorted(project.rglob("*")):
            if source_file.suffix not in exts:
                continue
            if any(part in ignore for part in source_file.parts):
                continue
            if file_count >= self._config.max_files:
                break
            # Checked EVERY file, not once per batch. Measured 2026-08-12:
            # sync_file does a real network write to the graph (~0.48s/file
            # against the live cluster), so a batch-gated check overshot a
            # 10.0s budget to 24.2s — worst case one whole batch of real
            # work past the deadline. time.monotonic() costs tens of
            # nanoseconds against a ~480ms loop body, so the batching it
            # was meant to save is not a saving at all. Overshoot is now
            # bounded by ONE file, which is the floor without cancelling a
            # write already in flight.
            if time.monotonic() >= deadline:
                budget_exceeded = True
                break

            file_path = str(source_file)
            if self.needs_sync(file_path):
                result = self.sync_file(file_path)
                if result.status == SyncStatus.COMPLETED:
                    resynced += 1
            file_count += 1

        if budget_exceeded:
            _logger.warning(
                "kg sync budget exceeded after %.1fs, continuing with "
                "partial context (%d files considered, %d re-synced, "
                "project=%r)",
                time.monotonic() - start,
                file_count,
                resynced,
                project_path,
            )

        self._save_state(project_path)
        return resynced

    def clear(self) -> None:
        """Clear all sync state."""
        self._file_hashes.clear()
        self._sync_count = 0