orchestrata_ultimate.knowledge_graph.sync

Reference › orchestrata_ultimate › knowledge_graph › sync

GraphSyncer is the bulk ingestion engine: it takes the serialized AST JSON that the repository's AST serializer produces and materializes it in the operational knowledge graph, the orchkg database on the orchgraph engine (a Memgraph fork). The module docstring describes 12 node types and 18 relationship types written in a nodes-first, edges-second pattern (Spec 008, REQ-008-023). The kernel CLI and the ast_live_sync script import it. It inherits the read-only inspection queries of _sync_queries. The neo4j names in its docstrings are Bolt-client naming only. The package story is on the knowledge_graph module page; the incremental, file-level path is synchronizer.

• Connection and schema. The constructor takes a URI, optional credentials and database, a no_auth flag and a dialect. connect opens a BoltSyncRepository, create_schema applies constraints and indexes idempotently using canonical_schema and schema, and close releases the repository.
• Full sync. full_sync clears the project's existing data (_clear_project) and calls _sync_ast, which runs batched _sync_packages, _sync_modules, _sync_classes, _sync_functions, _sync_methods, _sync_parameters and _sync_requirements (from __traces__ decorators), then the edge passes: package and module CONTAINS, DEFINES, INHERITS, IMPORTS, HAS_PARAMETER and CALLS. Call edges are resolved with a local-name index and an import-name index per module. It returns a count per category.
• Namespaces. An optional namespace tags the facts owned by the synced modules and prefixes their ids, so several trees can share one project; a trunk sentinel namespace is recognised.
• Incremental sync. incremental_sync re-syncs named modules only: delete_modules removes those modules and everything they own, then the nodes and edges are recreated and call edges repaired.
• Tests and mappings. sync_tests loads a test AST, creates Test nodes and TESTED_BY edges by matching test names to function and method names, and map_classes writes MAPS_TO edges from a class-mapping file.

Writes use batched MERGE statements so a repeated sync does not duplicate nodes.

At a glance

Class diagram

Dependency diagram

API

Classes

GraphSyncer

Bases: QueryMixin · Decorators: traces(spec='008', story='US-001', reqs=['REQ-008-023']) · Line 58

Sync AST JSON to knowledge graph.

Functions

Source

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

# mypy: ignore-errors
"""GraphSyncer — sync serialized AST JSON into knowledge graph.

12 node types, 18 relationship types, nodes-first-edges-second pattern.
"""

from __future__ import annotations

import logging
import os
from collections.abc import Sequence
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph import BoltSyncRepository, GraphRepositoryConfig
from orchestrata_ultimate.core.graph_factory import NO_AUTH, resolve_kg_dialect
from orchestrata_ultimate.knowledge_graph._sync_queries import QueryMixin
from orchestrata_ultimate.knowledge_graph.canonical_schema import (
    DERIVED_LABELS,
    LABEL_PARAMETER,
    LABEL_PROJECT,
    REL_CONTAINS,
    REL_HAS_PARAMETER,
    REL_HAS_REQUIREMENT,
    project_qualified_id,
)
from orchestrata_ultimate.knowledge_graph.exceptions import SyncError
from orchestrata_ultimate.knowledge_graph.query_engine import bump_graph_version

_logger = logging.getLogger(__name__)

# ── Node types (12) ──────────────────────────────────────────────────
# Project, Package, Module, Class, Function, Method, Parameter,
# Requirement, Spec, UserStory, ImportNode, DecoratorNode

# ── Relationship types (18) ──────────────────────────────────────────
# CONTAINS, DEFINES, DEFINES, HAS_PARAMETER, INHERITS,
# IMPORTS, CALLS, DECORATED_BY, BELONGS_TO_PACKAGE, BELONGS_TO_MODULE,
# TRACES_TO, MAPS_TO, HAS_BASE, NEXT, PREVIOUS, ASSOCIATED_WITH,
# IMPLEMENTS, EXTENDS


def _is_trunk_namespace(namespace: str) -> bool:
    """True if ``namespace`` is the ADR-0004 TRUNK sentinel key (B4a.2).

    Mirrors ``factory.promotion.models.FactNamespace.trunk(commit).key``'s
    format (``"trunk@{commit}"``) by convention rather than importing that
    module -- ``knowledge_graph`` sits below ``factory`` in this codebase's
    layering, and the reverse import would invert it. Both sides pin this
    exact format via tests (``FactNamespace.trunk`` and this module's own
    namespace-qualification tests).
    """
    return namespace.startswith("trunk@")


@traces(spec="008", story="US-001", reqs=["REQ-008-023"])
class GraphSyncer(QueryMixin):
    """Sync AST JSON to knowledge graph."""

    def __init__(
        self,
        uri: str | None = None,
        username: str | None = None,
        password: str | None = None,
        database: str | None = None,
        *,
        no_auth: bool = False,
        dialect: str | None = None,
    ) -> None:
        # Infra endpoints come from .env with NO hardcoded fallback (a silent
        # localhost default masks a missing config); connect() fails fast.
        # KG_ROUTING_URI/KG_URI: the NEO4J_* keys were retired with the Neo4j
        # containers on 2026-07-27.
        self._uri = (
            uri or os.environ.get("KG_ROUTING_URI") or os.environ.get("KG_URI") or ""
        )
        self._user = username or ""
        self._password = password or ""
        # Step 3.3 re-derives the AST subgraph onto orchgraph, which serves
        # multiple databases from one endpoint and negotiates no auth at all.
        # `no_auth` was explicit rather than inferred from empty credentials
        # because NEO4J_USERNAME/NEO4J_PASSWORD were then set in .env for the
        # real Neo4j, so a syncer merely pointed at orchgraph would have
        # presented Neo4j's credentials to a server that wants none. Those keys
        # were retired with the containers on 2026-07-27, so there are no
        # credentials left to leak; the flag is kept because callers still pass
        # it and an explicit "this endpoint has no auth" beats inferring it.
        self._database = database or None
        self._no_auth = no_auth
        # Resolved at construction from KG_ENGINE unless the caller names a
        # dialect. It used to default to the literal "neo4j", so after the
        # cutover every caller that did not pass `dialect=` emitted DDL the
        # live engine rejects outright (`CREATE TEXT INDEX ... IF NOT EXISTS`).
        self._dialect = resolve_kg_dialect(dialect)
        self._repo: BoltSyncRepository | None = None

    @property
    def _connected_repo(self) -> BoltSyncRepository:
        """Type-narrowing accessor for the private sync/query helpers.

        They are only ever reached via a public entrypoint (``full_sync``,
        ``create_schema``, ...) that already raised "Not connected" if
        ``self._repo`` was None, so this never actually raises in practice —
        it exists so pyright can see a non-Optional repository inside the
        helpers without re-guarding every one of them.
        """
        if self._repo is None:
            raise RuntimeError("Not connected. Call connect() first.")
        return self._repo

    def connect(self) -> None:
        """Establish the graph driver connection."""
        if not self._uri:
            raise RuntimeError(
                "No graph URI is set — set KG_ROUTING_URI (or KG_URI) in the "
                "active env profile, or pass uri= explicitly. The NEO4J_URI key "
                "this message used to name was retired with the Neo4j "
                "containers on 2026-07-27, so setting it fixes nothing."
            )
        repo = BoltSyncRepository(
            GraphRepositoryConfig(
                uri=self._uri,
                # NO_AUTH is the sentinel `create_graph`/`_resolve_auth` turn
                # into a real `auth=None`; a plain None here means "unset —
                # fall back to NEO4J_USERNAME/NEO4J_PASSWORD", which is the
                # opposite of what an unauthenticated cluster needs.
                auth=NO_AUTH if self._no_auth else (self._user, self._password),
                database=self._database,
            )
        )
        repo.verify_connectivity()
        self._repo = repo

    def close(self) -> None:
        """Close Neo4j driver connection."""
        if self._repo:
            self._repo.close()
            self._repo = None

    # ── Schema ─────────────────────────────────────────────────────────

    # AST-sync-specific DDL on top of the unified canon (get_unified_ddl):
    # labels only this syncer writes, plus its load-bearing lookup indexes
    # (module_path backs the IMPORTS edge MATCH; project_name backs browse).
    # The `project_id` (Project.id) backstop constraint was RETIRED
    # 2026-07-24: `_sync_ast` always writes `id` and `slug` as the same
    # value, so the canon's `project_slug` constraint already guarantees
    # `id` uniqueness transitively — and shipping the extra constraint here
    # re-created, on every test run, the wrong-key drift that
    # canonical_conformance drops and the kg_schema check flags.
    # Package.id was ADOPTED as the canonical key 2026-07-12 (bare
    # Package.path collides across projects), so package_id below now
    # duplicates the canon — kept for statement-name stability.
    #
    # ``method_id`` was RETIRED 2026-08-06: Method graduated into
    # ``canonical_schema.CANONICAL_MERGE_KEYS`` (REUSING this exact name),
    # so ``get_unified_ddl()``'s canonical generator now supplies it — a
    # second, independent definer of the same fact here would re-create the
    # split-brain this canon exists to eliminate.
    #
    # ``import_node_id``/``decorator_node_id`` were RETIRED 2026-08-06: dead
    # schema. ImportNode/DecoratorNode are never persisted as graph nodes —
    # they exist only as in-memory dataclasses during AST parsing (see
    # ``code_parser.py``, ``synchronizer.py``'s "ImportNodes are not
    # persisted as separate nodes" comment) — confirmed 0 nodes under either
    # label across all 32 databases on the shared test cluster, measured
    # 2026-08-06. The two constraints sat here, live, protecting a label no
    # writer anywhere ever creates.
    SYNC_DDL: list[str] = [
        "CREATE CONSTRAINT package_id IF NOT EXISTS FOR (p:Package) REQUIRE p.id IS UNIQUE",
        "CREATE INDEX project_name IF NOT EXISTS FOR (p:Project) ON (p.name)",
        "CREATE INDEX module_path IF NOT EXISTS FOR (m:Module) ON (m.path)",
    ]

    def create_schema(self, statements: list[str] | None = None) -> None:
        """Create constraints and indexes (idempotent).

        Applies the unified DDL (SPEC-106 canonical identity constraints +
        schema.py's lookup layer, via ``schema.get_unified_ddl``) plus this
        syncer's own ``SYNC_DDL``. Only genuine ``*AlreadyExists`` conflicts
        are tolerated; any other DDL failure RAISES — the old blanket
        warning-swallow could leave the sync graph without its constraints
        and load-bearing indexes while looking successful (same defect class
        as ensure_schema, commit 246cad25).

        Args:
            statements: Override the DDL set (tests / scoped migrations),
                mirroring ``GraphClient.ensure_schema``.
        """
        if not self._repo:
            raise RuntimeError("Not connected. Call connect() first.")

        from orchestrata_ultimate.knowledge_graph.schema import get_unified_ddl

        if statements is None:
            statements = get_unified_ddl(extra=self.SYNC_DDL, dialect=self._dialect)
        for cql in statements:
            try:
                # execute_autocommit, not execute_write: Memgraph rejects
                # constraint/index manipulation inside an explicit/managed
                # transaction ("Constraint manipulation is not allowed in
                # multicommand transactions"), which made this method fail
                # outright against orchgraph — the cluster Step 3.3 re-derives
                # onto. Neo4j accepts either, so the auto-commit path is the
                # portable one. It is still MVCC-retried.
                self._connected_repo.execute_autocommit(cql)
            except Exception as exc:
                if "AlreadyExists" in (getattr(exc, "code", "") or ""):
                    _logger.debug("Schema DDL already satisfied: %s", exc)
                    continue
                _logger.error("Schema DDL failed: %s -- %s", cql, exc)
                raise

    # ── Full Sync ──────────────────────────────────────────────────────

    @traces(spec="008", story="US-002", reqs=["REQ-008-002"])
    def full_sync(
        self,
        ast_data: dict[str, Any],
        project: str,
        packages_meta: dict[str, Any] | None = None,
        *,
        namespace: str | None = None,
    ) -> dict[str, int]:
        """Clear existing project data and re-sync from AST JSON.

        Args:
            ast_data: AST JSON to sync.
            project: Project name.
            packages_meta: Optional package metadata (reserved).
            namespace: SPEC-033 §4 / ADR-0004 fact-namespace key (e.g.
                ``FactNamespace.key`` from
                ``factory.promotion.models``). ``None`` (the default)
                preserves today's un-namespaced trunk behavior byte-for-byte
                -- every existing caller omits this and is unaffected. When
                given, every fact node synced by this call additionally gets
                a ``namespace`` property set to this value (see
                ``_tag_namespace``), so parallel-worktree gate reads can
                scope to their own branch's facts (B4a).

        Returns:
            Dict with nodes_created and relationships_created counts.
        """
        if not self._repo:
            raise RuntimeError("Not connected. Call connect() first.")

        # Clear existing project data, then (re)build the whole graph.
        self._clear_project(project, namespace=namespace)
        return self._sync_ast(
            ast_data,
            project,
            packages_meta,
            sync_project_node=True,
            namespace=namespace,
        )

    @traces(spec="008", story="US-005", reqs=["REQ-008-021", "REQ-008-024"])
    def _sync_ast(
        self,
        ast_data: dict[str, Any],
        project: str,
        packages_meta: dict[str, Any] | None = None,
        *,
        sync_project_node: bool = True,
        namespace: str | None = None,
        context_modules: list[dict[str, Any]] | None = None,
    ) -> dict[str, int]:
        """Create/MERGE all nodes and edges for ``ast_data`` without clearing.

        Shared by ``full_sync`` (after a project-wide clear) and
        ``incremental_sync`` (after deleting only the targeted modules). All
        nodes are MERGEd on stable ids, so re-syncing a subset updates those
        modules without touching the rest of the project.

        Args:
            ast_data: AST JSON — the full graph or a filtered module subset.
            project: Project name.
            packages_meta: Optional package metadata (reserved).
            sync_project_node: When ``False`` (incremental), the Project node is
                not rewritten — its ``total_loc``/``module_count`` describe the
                whole project, not the subset being re-synced.
            namespace: SPEC-033 §4 fact-namespace key (B4a). ``None`` (default)
                is a pure no-op -- none of the node-writing helpers below are
                touched, so the un-namespaced path is byte-identical to
                before this parameter existed. When given, ``_tag_namespace``
                runs ONE additional write after the normal sync, scoped to
                exactly the modules just processed (see below).
            context_modules: Full-project module pool used ONLY to resolve
                cross-module CALLS edges (``_sync_calls_edges``) when
                ``ast_data["modules"]`` is a filtered subset (incremental
                sync). ``None`` (default) leaves ``full_sync`` byte-identical
                to before this parameter existed, since there ``ast_data``
                already is the whole project.

        Returns:
            Dict with nodes_created and relationships_created counts.
        """
        if not self._repo:
            raise RuntimeError("Not connected. Call connect() first.")

        # Extract packages from module paths
        packages: dict[str, dict[str, Any]] = {}
        for mod in ast_data.get("modules", []):
            path = mod.get("path", "")
            parts = path.split(".")
            for i in range(len(parts)):
                pkg_path = ".".join(parts[: i + 1])
                if pkg_path not in packages:
                    packages[pkg_path] = {
                        "id": f"{project}:{pkg_path}",
                        "name": parts[i],
                        "path": pkg_path,
                        "depth": i,
                    }

        # Create Project node with useful properties (skipped for incremental)
        modules = ast_data.get("modules", [])
        if sync_project_node:
            total_loc = sum(m.get("line_count", 0) for m in modules)
            # Project's canonical merge key is `slug` (canonical_schema
            # CANONICAL_MERGE_KEYS / SYNC_DDL's own comment above documents
            # this syncer as the divergent `id`-keyed writer). `id` and
            # `slug` are always the same value here (both = `project`), so
            # this is a same-value rename, not a behaviour change -- but it
            # aligns the MERGE key with the live `project_slug` constraint,
            # closing the ConstraintValidationFailed window under
            # parallel/sequential batch writes (2026-07-22 incident class).
            # `id` is kept via SET (never drop data; `_clear_project` still
            # MATCHes on it).
            self._connected_repo.execute_write(
                "MERGE (p:Project {slug: $slug}) "
                "SET p.id = $id, "
                "    p.name = $name, "
                "    p.project_type = $project_type, "
                "    p.total_loc = $total_loc, "
                "    p.root_path = $root_path, "
                "    p.synced_at = datetime(), "
                "    p.module_count = $module_count",
                {
                    "id": project,
                    "name": project,
                    "slug": project,
                    "project_type": "library",
                    "total_loc": total_loc,
                    "root_path": ast_data.get("root_path", ""),
                    "module_count": len(modules),
                },
            )
            # Any cached query results are now stale because the graph has
            # changed; bump the global version so QueryEngine cache keys miss.
            bump_graph_version()

        # Phase 1: Nodes (packages, modules, classes, functions, methods, params)
        nodes_created = self._sync_packages(project, packages)
        nodes_created += self._sync_modules(project, ast_data.get("modules", []))
        nodes_created += self._sync_classes(project, ast_data.get("modules", []))
        nodes_created += self._sync_functions(project, ast_data.get("modules", []))
        nodes_created += self._sync_methods(project, ast_data.get("modules", []))
        nodes_created += self._sync_parameters(project, ast_data.get("modules", []))
        nodes_created += self._sync_requirements(project, ast_data.get("modules", []))

        # Phase 2: Edges
        rels_created = self._sync_contains_edges(project, packages)
        rels_created += self._sync_module_edges(
            project, ast_data.get("modules", []), packages
        )
        rels_created += self._sync_defines_edges(project, ast_data.get("modules", []))
        rels_created += self._sync_inheritance_edges(
            project, ast_data.get("inheritance_graph", {})
        )
        rels_created += self._sync_dependency_edges(
            project, ast_data.get("dependency_graph", {})
        )
        rels_created += self._sync_has_method_edges(
            project, ast_data.get("modules", [])
        )
        rels_created += self._sync_parameter_edges(project, ast_data.get("modules", []))
        rels_created += self._sync_calls_edges(
            project,
            ast_data.get("modules", []),
            ast_data.get("root_path", ""),
            context_modules=context_modules,
        )

        # B4a/ADR-0004: additive-only fact-namespace tagging. Runs AFTER the
        # normal sync above and touches nothing about it; skipped entirely
        # (byte-identical to pre-B4a behavior) when namespace is None.
        if namespace is not None:
            module_ids = [f"{project}:module:{m.get('path', '')}" for m in modules]
            self._tag_namespace(module_ids, namespace)

        return {"nodes_created": nodes_created, "relationships_created": rels_created}

    def _tag_namespace(self, module_ids: list[str], namespace: str) -> None:
        """Tag every fact node owned by ``module_ids`` with ``namespace``.

        SPEC-033 §4 (ADR-0004): AST/trace facts are branch-dependent state,
        so a promotable fact carries the ``(worktree, commit)`` namespace key
        it was built under. This is the ONLY behavior ``_sync_ast`` gains
        from a non-``None`` ``namespace`` -- it runs as one extra write
        AFTER the existing node/edge sync, using the same
        ``DEFINES|HAS_PARAMETER`` ownership traversal ``_clear_project``/
        ``incremental_sync`` already use to reach everything a Module owns
        (classes, functions, methods, parameters), so none of the ~15
        existing ``_sync_*`` writer methods needed to change.

        Downstream promotion/GC over these tags is
        ``factory.promotion.store.PromotionStore`` (B4a), which matches on
        the ``namespace`` property alone (label-agnostic) — see that
        module's docstring for the promotion/GC contract.

        B4a.2 (ADR-0004): unless ``namespace`` is the TRUNK sentinel (see
        ``_is_trunk_namespace``), this ALSO qualifies every tagged node's
        ``id`` with an ``"ns::{namespace}::"`` prefix
        (``_qualify_namespace_ids``) -- see that method's docstring for why
        this is required (two concurrent ``full_sync`` calls under
        different namespaces would otherwise MERGE onto the same
        unqualified node ids).
        """
        if not module_ids:
            return
        self._connected_repo.execute_write(
            "UNWIND $ids AS mid "
            "MATCH (m:Module {id: mid}) "
            "SET m.namespace = $namespace",
            {"ids": module_ids, "namespace": namespace},
        )
        self._connected_repo.execute_write(
            "UNWIND $ids AS mid "
            "MATCH (m:Module {id: mid})-[:DEFINES|HAS_PARAMETER*1..]->(n) "
            "SET n.namespace = $namespace",
            {"ids": module_ids, "namespace": namespace},
        )
        if not _is_trunk_namespace(namespace):
            self._qualify_namespace_ids(module_ids, namespace)

    def _qualify_namespace_ids(self, module_ids: list[str], namespace: str) -> None:
        """Prefix every namespace-tagged node's ``id`` with ``"ns::{namespace}::"``.

        SPEC-033 §4 (ADR-0004), B4a.2. The hazard this closes: two
        CONCURRENT ``full_sync(..., namespace=A)`` / ``(..., namespace=B)``
        calls for the SAME project previously MERGEd onto the SAME
        unqualified node ids (e.g. ``"{project}:module:{path}"``), so the
        last writer's ``namespace`` tag won and the loser's facts vanished
        from its own namespace. Qualifying the ``id`` itself -- not just the
        ``namespace`` property -- means a subsequent MERGE by the (now
        freed) unqualified id creates a fresh node instead of colliding with
        the already-qualified one, and ``PromotionStore.facts_in``/
        ``promote`` (label-agnostic, matching purely on ``namespace``) see
        genuinely disjoint id sets per namespace.

        Implementation choice (read the code before picking): this project's
        ~15 ``_sync_*`` writer methods build ids inline via f-strings at
        many different call sites (``_sync_modules``, ``_sync_classes``,
        ``_sync_functions``, ``_sync_methods``, ``_build_parameter_rows``,
        plus every edge writer that re-derives the same id strings to MATCH
        by them) -- there is no single id-builder choke point to consult a
        ``self._id_prefix`` through without threading a new prefix parameter
        through all of them (the larger diff). Reusing ``_tag_namespace``'s
        EXISTING ``DEFINES|HAS_PARAMETER`` ownership traversal for a
        single additive post-pass is the smaller diff and needs no writer
        signature to change.

        Descendants are rewritten FIRST, matched by ``mid`` (the Module's
        STILL-unqualified id at this point in the call); Module nodes are
        rewritten LAST, because renaming ``Module.id`` first would
        invalidate ``mid`` for the descendant ``MATCH`` above. Neo4j
        relationships attach to node identity, not the ``id`` property, so a
        plain ``SET n.id = ...`` here does not disturb any DEFINES/
        HAS_PARAMETER/INHERITS/IMPORTS/CALLS edge created earlier in this
        same ``_sync_ast`` call -- only PROMOTION
        (``factory.promotion.store.PromotionStore.promote``), which MERGEs
        onto a DIFFERENT (trunk) node and deletes the qualified one, must
        explicitly recreate them (see that method's docstring).
        """
        prefix = f"ns::{namespace}::"
        self._connected_repo.execute_write(
            "UNWIND $ids AS mid "
            "MATCH (m:Module {id: mid})-[:DEFINES|HAS_PARAMETER*1..]->(n) "
            "SET n.id = $prefix + n.id",
            {"ids": module_ids, "prefix": prefix},
        )
        self._connected_repo.execute_write(
            "UNWIND $ids AS mid "
            "MATCH (m:Module {id: mid}) "
            "SET m.id = $prefix + m.id",
            {"ids": module_ids, "prefix": prefix},
        )

    @traces(spec="008", story="US-005", reqs=["REQ-008-023"])
    def _clear_project(self, project: str, *, namespace: str | None = None) -> None:
        """Delete all nodes and relationships for a project.

        Traverses every ownership edge -- CONTAINS (project/package/module
        hierarchy), DEFINES (module->class/function), DEFINES
        (class->method) and HAS_PARAMETER (->parameter) -- so that classes,
        functions, methods and parameters are not left orphaned (they hang off
        modules via DEFINES/DEFINES, not CONTAINS).

        Parameter nodes are ADDITIONALLY reclaimed by their ``parent_id``
        property, independent of the traversal above: every sync before this
        fix wrote Parameter nodes without the HAS_PARAMETER edge the
        traversal (and this method's own docstring) assumed existed, so they
        accumulated as graph-orphaned islands unreachable from Project
        (kg-schema-gaps: 9.4k such orphans found live, 2026-07-18). This
        second pass is idempotent -- a no-op once every Parameter carries a
        real HAS_PARAMETER edge -- so it is safe to keep even after
        ``_sync_parameter_edges`` closes the gap going forward.

        Scope: DERIVED labels only, and the Project hub is never deleted.
        Both were live hazards, not tidying. The traversal used ``*0..``,
        whose zero-hop case matches the Project node itself, and a trailing
        statement deleted it outright -- so a full sync ``DETACH DELETE``d
        the hub. That was survivable while this graph held only the AST
        subgraph. It is not survivable in the migrated multi-tenant
        ``orchkg``, where the same hub carries HAS_SESSION / HAS_THREAD /
        HAS_VOLK / HAS_FACT_NAMESPACE: measured on the live graph
        2026-07-26, running a full sync there would have severed 8,341 hub
        edges and violated the session-254 root-connectivity invariant for
        the entire record-of-fact subgraph, while ``_sync_ast`` recreated a
        bare Project node that made the damage look like success.

        Not deleting the hub is safe because ``_sync_ast`` MERGEs it on
        ``slug`` rather than creating it, so a surviving node is found and
        updated in place.

        Args:
            namespace: SPEC-033 §4 fact-namespace key (B4a.2). ``None``
                (default) is byte-identical to before this parameter
                existed -- every existing caller omits it. When given, the
                traversal-based delete additionally excludes nodes already
                tagged with a DIFFERENT namespace: a namespace-scoped
                ``full_sync`` must be able to coexist with a sibling
                worktree's already-synced facts under the SAME project
                (ADR-0004) rather than wiping them out from under it. Trunk/
                untagged facts (``namespace IS NULL``) and this namespace's
                OWN prior facts are still cleared and rebuilt, matching
                today's "resync from scratch" semantics.
        """
        if namespace is not None:
            self._connected_repo.execute_write(
                "MATCH (p:Project {id: $project})"
                "-[:CONTAINS|DEFINES|HAS_PARAMETER*1..]->(n) "
                "WHERE (n.namespace IS NULL OR n.namespace = $namespace) "
                "AND any(l IN labels(n) WHERE l IN $derived) "
                "DETACH DELETE n",
                {
                    "project": project,
                    "namespace": namespace,
                    "derived": sorted(DERIVED_LABELS),
                },
            )
            self._connected_repo.execute_write(
                "MATCH (param:Parameter) WHERE param.parent_id STARTS WITH $prefix "
                "AND (param.namespace IS NULL OR param.namespace = $namespace) "
                "DETACH DELETE param",
                {"prefix": f"{project}:", "namespace": namespace},
            )
        else:
            self._connected_repo.execute_write(
                "MATCH (p:Project {id: $project})"
                "-[:CONTAINS|DEFINES|HAS_PARAMETER*1..]->(n) "
                "WHERE any(l IN labels(n) WHERE l IN $derived) "
                "DETACH DELETE n",
                {"project": project, "derived": sorted(DERIVED_LABELS)},
            )
            self._connected_repo.execute_write(
                "MATCH (param:Parameter) WHERE param.parent_id STARTS WITH $prefix "
                "DETACH DELETE param",
                {"prefix": f"{project}:"},
            )

    @traces(spec="008", story="US-004", reqs=["REQ-008-021"])
    def _sync_packages(self, project: str, packages: dict[str, Any]) -> int:
        """Batch sync Package nodes."""
        if not packages:
            return 0
        rows = [
            {
                "id": v["id"],
                "name": v["name"],
                "path": v["path"],
                "depth": v["depth"],
                "project": project,
            }
            for v in packages.values()
        ]
        self._connected_repo.execute_write(
            "UNWIND $rows AS r "
            "MERGE (p:Package {id: r.id}) "
            "SET p.name = r.name, p.path = r.path, p.depth = r.depth, p.project = r.project",
            {"rows": rows},
        )
        return len(rows)

    @traces(spec="008", story="US-004", reqs=["REQ-008-021"])
    def _sync_modules(self, project: str, modules: list[dict]) -> int:
        """Batch sync Module nodes."""
        if not modules:
            return 0
        rows = []
        for mod in modules:
            path = mod.get("path", "")
            rows.append(
                {
                    "id": f"{project}:module:{path}",
                    "path": path,
                    "qualified_name": path,
                    "file_path": mod.get("file_path", ""),
                    "line_count": mod.get("line_count", 0),
                    "project": project,
                }
            )
        self._connected_repo.execute_write(
            "UNWIND $rows AS r "
            "MERGE (m:Module {id: r.id}) "
            "SET m.path = r.path, m.qualified_name = r.qualified_name, "
            "    m.file_path = r.file_path, m.line_count = r.line_count, m.project = r.project",
            {"rows": rows},
        )
        return len(rows)

    @traces(spec="008", story="US-004", reqs=["REQ-008-021"])
    def _sync_classes(self, project: str, modules: list[dict]) -> int:
        """Batch sync Class nodes.

        ``id`` is project-qualified (``"{project}:{qualified_name}"``): a bare
        ``qualified_name`` (e.g. ``"pkg.module.Class"``) legitimately recurs
        across projects sharing a package layout, so it is unsound as a global
        MERGE key in a multi-project graph (2026-07-12 multi-project ruling).
        """
        rows = []
        for mod in modules:
            mod_path = mod.get("path", "")
            for cls in mod.get("classes", []):
                qname = cls.get("qualified_name") or f"{mod_path}.{cls['name']}"
                rows.append(
                    {
                        "id": f"{project}:{qname}",
                        "name": cls.get("name", ""),
                        "qualified_name": qname,
                        "module_path": cls.get("module_path", mod_path),
                        "file_path": cls.get("file_path", ""),
                        "line": cls.get("line", 0),
                        "is_abstract": cls.get("is_abstract", False),
                        "is_dataclass": cls.get("is_dataclass", False),
                        "is_pydantic": cls.get("is_pydantic", False),
                        "is_protocol": cls.get("is_protocol", False),
                        "is_enum": cls.get("is_enum", False),
                        "bases": cls.get("bases", []),
                        "project": project,
                    }
                )
        if not rows:
            return 0
        self._connected_repo.execute_write(
            "UNWIND $rows AS r "
            "MERGE (c:Class {id: r.id}) "
            "SET c.name = r.name, c.qualified_name = r.qualified_name, "
            "c.module_path = r.module_path, c.file_path = r.file_path, "
            "c.line = r.line, c.is_abstract = r.is_abstract, "
            "c.is_dataclass = r.is_dataclass, c.is_pydantic = r.is_pydantic, "
            "c.is_protocol = r.is_protocol, c.is_enum = r.is_enum, "
            "c.bases = r.bases, c.project = r.project",
            {"rows": rows},
        )
        return len(rows)

    @traces(spec="008", story="US-004", reqs=["REQ-008-021"])
    def _sync_functions(self, project: str, modules: list[dict]) -> int:
        """Batch sync standalone Function nodes.

        ``id`` is project-qualified via :func:`canonical_schema.project_qualified_id`
        (the shared helper other Function-id writers, e.g.
        ``test_results_store.py``'s COVERS writer, also call so the scheme
        cannot drift) — see ``_sync_classes`` docstring.
        """
        rows = []
        for mod in modules:
            mod_path = mod.get("path", "")
            for func in mod.get("functions", []):
                qname = func.get("qualified_name") or f"{mod_path}.{func['name']}"
                rows.append(
                    {
                        "id": project_qualified_id(project, qname),
                        "name": func.get("name", ""),
                        "qualified_name": qname,
                        "module_path": func.get("module_path", mod_path),
                        "file_path": func.get("file_path", ""),
                        "line": func.get("line", 0),
                        "is_async": func.get("is_async", False),
                        "return_type": func.get("return_type", ""),
                        "class_name": func.get("class_name", ""),
                        "project": project,
                    }
                )
        if not rows:
            return 0
        self._connected_repo.execute_write(
            "UNWIND $rows AS r "
            "MERGE (f:Function {id: r.id}) "
            "SET f.name = r.name, f.qualified_name = r.qualified_name, "
            "f.module_path = r.module_path, f.file_path = r.file_path, "
            "f.line = r.line, f.is_async = r.is_async, f.return_type = r.return_type, "
            "f.project = r.project",
            {"rows": rows},
        )
        return len(rows)

    @traces(spec="008", story="US-004", reqs=["REQ-008-021"])
    def _sync_methods(self, project: str, modules: list[dict]) -> int:
        """Batch sync Method nodes (functions inside classes).

        ``id`` is project-qualified — see ``_sync_classes`` docstring.
        """
        rows = []
        for mod in modules:
            mod_path = mod.get("path", "")
            for cls in mod.get("classes", []):
                for method in cls.get("methods", []):
                    qname = method.get("qualified_name") or (
                        f"{mod_path}.{cls['name']}.{method['name']}"
                    )
                    rows.append(
                        {
                            "id": f"{project}:{qname}",
                            "name": method.get("name", ""),
                            "qualified_name": qname,
                            "module_path": method.get("module_path", mod_path),
                            "file_path": method.get("file_path", ""),
                            "line": method.get("line", 0),
                            "is_async": method.get("is_async", False),
                            "is_classmethod": method.get("is_classmethod", False),
                            "is_staticmethod": method.get("is_staticmethod", False),
                            "is_property": method.get("is_property", False),
                            "is_abstract": method.get("is_abstract", False),
                            "return_type": method.get("return_type", ""),
                            "class_name": method.get("class_name", cls["name"]),
                            "project": project,
                        }
                    )
        if not rows:
            return 0
        self._connected_repo.execute_write(
            "UNWIND $rows AS r "
            "MERGE (m:Method {id: r.id}) "
            "SET m.name = r.name, m.qualified_name = r.qualified_name, "
            "m.module_path = r.module_path, m.file_path = r.file_path, "
            "m.line = r.line, m.is_async = r.is_async, "
            "m.is_classmethod = r.is_classmethod, m.is_staticmethod = r.is_staticmethod, "
            "m.is_property = r.is_property, m.is_abstract = r.is_abstract, "
            "m.return_type = r.return_type, m.class_name = r.class_name, "
            "m.project = r.project",
            {"rows": rows},
        )
        return len(rows)

    @traces(spec="008", story="US-004", reqs=["REQ-008-021"])
    @staticmethod
    def _build_parameter_rows(
        project: str, modules: list[dict]
    ) -> list[dict[str, str]]:
        """Build one row per Parameter across ``modules``.

        Shared by ``_sync_parameters`` (nodes) and ``_sync_parameter_edges``
        (the HAS_PARAMETER edge) so both stay off one source of truth for
        parameter identity — ``id``/``parent_id`` are project-qualified, see
        ``_sync_classes`` docstring; ``parent_id`` must match the (also
        qualified) owning Function/Method's ``id``.
        """
        rows: list[dict[str, str]] = []
        for mod in modules:
            mod_path = mod.get("path", "")
            rows.extend(
                {
                    "id": f"{parent_id}:param:{i}",
                    "name": param.get("name", ""),
                    "type": param.get("type", ""),
                    "default": param.get("default", ""),
                    "parent_id": parent_id,
                }
                for func in mod.get("functions", [])
                for func_qname in [
                    func.get("qualified_name") or f"{mod_path}.{func['name']}"
                ]
                for parent_id in [f"{project}:{func_qname}"]
                for i, param in enumerate(func.get("parameters", []))
            )
            rows.extend(
                {
                    "id": f"{parent_id}:param:{i}",
                    "name": param.get("name", ""),
                    "type": param.get("type", ""),
                    "default": param.get("default", ""),
                    "parent_id": parent_id,
                }
                for cls in mod.get("classes", [])
                for method in cls.get("methods", [])
                for method_qname in [
                    method.get("qualified_name")
                    or f"{mod_path}.{cls['name']}.{method['name']}"
                ]
                for parent_id in [f"{project}:{method_qname}"]
                for i, param in enumerate(method.get("parameters", []))
            )
        return rows

    @traces(spec="008", story="US-004", reqs=["REQ-008-021"])
    def _sync_parameters(self, project: str, modules: list[dict]) -> int:
        """Batch sync Parameter nodes from functions and methods.

        Node creation only — the owning HAS_PARAMETER edge is written
        separately by ``_sync_parameter_edges`` (Phase 2, edges), keeping
        this file's established nodes-first-edges-second pattern.
        """
        rows = self._build_parameter_rows(project, modules)
        if not rows:
            return 0
        self._connected_repo.execute_write(
            "UNWIND $rows AS r "
            "MERGE (p:Parameter {id: r.id}) "
            "SET p.name = r.name, p.type = r.type, p.default = r.default, "
            "p.parent_id = r.parent_id",
            {"rows": rows},
        )
        return len(rows)

    _PARAMETER_EDGE_BATCH_SIZE = 2000

    @traces(spec="008", story="US-004", reqs=["REQ-008-006"])
    def _sync_parameter_edges(self, project: str, modules: list[dict]) -> int:
        """Create HAS_PARAMETER edges: Function|Method -> Parameter.

        Without this edge, Parameter nodes are graph-orphaned islands
        reachable only via an unindexed ``parent_id`` property scan, and
        ``_clear_project``'s cleanup traversal (which assumes HAS_PARAMETER
        exists) can never reach them on re-sync.

        Session-254 root-connectivity ruling: a prior single-transaction
        version of this pass silently under-delivered (its returned count
        was never compared against ``len(rows)``), leaving 5044 live
        Parameter nodes with a resolvable ``parent_id`` and no edge — the
        failure signature matches Memgraph's ``storage.access_timeout``
        aborting an oversized write. This version batches the write and
        raises loudly on any shortfall instead of swallowing it.
        """
        rows = self._build_parameter_rows(project, modules)
        if not rows:
            return 0
        created = 0
        for start in range(0, len(rows), self._PARAMETER_EDGE_BATCH_SIZE):
            batch = rows[start : start + self._PARAMETER_EDGE_BATCH_SIZE]
            # One single-label pass per owner label — NEVER the multi-label
            # `(owner:Function|Method {id: ...})` form. orchgraph 3.11
            # planner defect, measured 2026-08-16: an UNWIND-driven
            # multi-label indexed lookup collapses to ONE result row per
            # DISTINCT key per transaction once both per-label id indexes
            # exist (25,517/40,330 delivered; all 21 batches returned
            # exactly their distinct-parent count; the same rows split into
            # single-label passes sum to exactly 40,330). An id resolves
            # under exactly one of the two labels, so the passes are
            # disjoint and their counts add.
            for owner_label in ("Function", "Method"):
                result = self._connected_repo.execute_write(
                    "UNWIND $rows AS r "
                    f"MATCH (owner:{owner_label} {{id: r.parent_id}}) "
                    f"MATCH (param:{LABEL_PARAMETER} {{id: r.id}}) "
                    f"MERGE (owner)-[:{REL_HAS_PARAMETER}]->(param) "
                    "RETURN count(*) AS cnt",
                    {"rows": batch},
                )
                created += result[0]["cnt"] if result else 0
        if created < len(rows):
            raise SyncError(
                "HAS_PARAMETER edge pass under-delivered: "
                f"{created}/{len(rows)} edges for project {project!r} — "
                "Parameter nodes would be left as unreachable islands "
                "(session-254 root-connectivity invariant).",
                file_path=project,
            )
        return created

    @traces(spec="008", story="US-004", reqs=["REQ-008-021"])
    def _sync_requirements(self, project: str, modules: list[dict]) -> int:
        """Extract and sync Requirement nodes from __traces__ decorators.

        ``id`` is project-qualified (``"{project}:{req}"``): the bare
        requirement tag (e.g. ``"REQ-008-023"``) is a per-project trace
        label, not a globally unique one (2026-07-12 multi-project ruling).
        The bare tag is kept under ``req_id`` (matching the property name
        ``knowledge_graph.context_injector``'s ``coalesce(req.req_id,
        req.id)`` and ``kernel.pipeline_kg.sync_decisions_to_kg`` already use
        for the same purpose) for display/lookup.

        Session-254 root-connectivity ruling: this pass used to MERGE
        ``:Requirement`` with no relationship at all, surviving only because
        other writers happen to add incoming ``TRACES_TO`` edges. That is
        luck, not an anchor, so this now also MERGEs the ``:Project`` hub
        directly -- these are code-derived (``__traces__``) Requirements,
        not the design-time Spec/UserStory/Requirement tree ``kg_design.py``
        anchors via ``REL_HAS_REQUIREMENT`` from ``:UserStory``; reusing the
        same relationship type from the Project hub keeps one canonical
        edge NAME for "this hub owns this Requirement" rather than inventing
        a second one.
        """
        rows: list[dict[str, str]] = []
        seen: set[str] = set()

        def collect(entity: dict) -> None:
            for req in entity.get("requirements", []):
                if req and req not in seen:
                    seen.add(req)
                    rows.append({"id": f"{project}:{req}", "name": req, "req_id": req})

        for mod in modules:
            for cls in mod.get("classes", []):
                collect(cls)
            for func in mod.get("functions", []):
                collect(func)
            for cls in mod.get("classes", []):
                for method in cls.get("methods", []):
                    collect(method)
        if not rows:
            return 0
        self._connected_repo.execute_write(
            "UNWIND $rows AS r "
            "MERGE (req:Requirement {id: r.id}) "
            "SET req.name = r.name, req.req_id = r.req_id "
            "WITH req "
            f"MERGE (p:{LABEL_PROJECT} {{slug: $project}}) "
            f"MERGE (p)-[:{REL_HAS_REQUIREMENT}]->(req)",
            {"rows": rows, "project": project},
        )
        return len(rows)

    # ── Edges ──────────────────────────────────────────────────────────

    @traces(spec="008", story="US-004", reqs=["REQ-008-006"])
    def _sync_contains_edges(self, project: str, packages: dict[str, Any]) -> int:
        """Create CONTAINS edges: Project → Package, Package → Package."""
        rows = []
        # Project -> top-level packages
        for pkg_path, pkg in packages.items():
            if pkg["depth"] == 0:
                rows.append(
                    {"from_id": project, "to_id": pkg["id"], "type": REL_CONTAINS}
                )
            else:
                parent_path = ".".join(pkg_path.split(".")[:-1])
                parent_id = packages.get(parent_path, {}).get("id")
                if parent_id:
                    rows.append(
                        {"from_id": parent_id, "to_id": pkg["id"], "type": REL_CONTAINS}
                    )

        # Labels are load-bearing: an unlabeled MATCH on {id} cannot use
        # the per-label unique constraints and plans an AllNodesScan per
        # row (~768K-node KG → multi-minute sync). And the label must be
        # SINGLE: the multi-label `(a:Project|Package {id})` form hits the
        # orchgraph 3.11 UNWIND cardinality collapse (one row per distinct
        # key — see _sync_parameter_edges). The builder above knows the
        # parent kind (depth-0 rows hang off the Project), so split there.
        project_rows = [r for r in rows if r["from_id"] == project]
        package_rows = [r for r in rows if r["from_id"] != project]
        created = 0
        for label, subset in (("Project", project_rows), ("Package", package_rows)):
            if not subset:
                continue
            result = self._connected_repo.execute_write(
                "UNWIND $rows AS r "
                f"MATCH (a:{label} {{id: r.from_id}}) "
                "MATCH (b:Package {id: r.to_id}) "
                f"MERGE (a)-[:{REL_CONTAINS}]->(b) "
                "RETURN count(*) AS cnt",
                {"rows": subset},
            )
            created += result[0]["cnt"] if result else 0
        return created

    @traces(spec="008", story="US-004", reqs=["REQ-008-006"])
    def _sync_module_edges(
        self, project: str, modules: list[dict], packages: dict[str, Any]
    ) -> int:
        """Create CONTAINS edges from Package to Module."""
        rows = []
        for mod in modules:
            path = mod.get("path", "")
            # Find the deepest package that matches this module's path
            pkg_path = None
            for candidate in sorted(packages.keys(), key=len, reverse=True):
                if path.startswith(candidate + ".") or path == candidate:
                    pkg_path = candidate
                    break
            if pkg_path:
                pkg_id = packages[pkg_path]["id"]
                mod_id = f"{project}:module:{path}"
                rows.append({"from_id": pkg_id, "to_id": mod_id})

        if not rows:
            return 0

        result = self._connected_repo.execute_write(
            "UNWIND $rows AS r "
            "MATCH (a:Package {id: r.from_id}) "
            "MATCH (b:Module {id: r.to_id}) "
            "MERGE (a)-[:CONTAINS]->(b) "
            "RETURN count(*) AS cnt",
            {"rows": rows},
        )
        return result[0]["cnt"] if result else 0

    @traces(spec="008", story="US-004", reqs=["REQ-008-006"])
    def _sync_defines_edges(self, project: str, modules: list[dict]) -> int:
        """Create DEFINES edges: Module → Class/Function/Method.

        ``to_id`` is project-qualified to match the (also qualified)
        Class/Function/Method ``id`` written by ``_sync_classes`` et al.
        """
        rows = []
        for mod in modules:
            mod_id = f"{project}:module:{mod['path']}"
            rows.extend(
                row
                for cls in mod.get("classes", [])
                for row in (
                    {
                        "from_id": mod_id,
                        "to_id": f"{project}:{cls.get('qualified_name', '')}",
                        "rel": "DEFINES",
                    },
                    *(
                        {
                            "from_id": mod_id,
                            "to_id": f"{project}:{method.get('qualified_name', '')}",
                            "rel": "DEFINES",
                        }
                        for method in cls.get("methods", [])
                    ),
                )
            )
            rows.extend(
                {
                    "from_id": mod_id,
                    "to_id": f"{project}:{func.get('qualified_name', '')}",
                    "rel": "DEFINES",
                }
                for func in mod.get("functions", [])
            )

        result = self._connected_repo.execute_write(
            "UNWIND $rows AS r "
            "MATCH (a:Module {id: r.from_id}) "
            "MATCH (b:Class|Method|Function {id: r.to_id}) "
            "MERGE (a)-[:DEFINES]->(b) "
            "RETURN count(*) AS cnt",
            {"rows": rows},
        )
        return result[0]["cnt"] if result else 0

    @traces(spec="008", story="US-004", reqs=["REQ-008-006"])
    def _sync_inheritance_edges(
        self, project: str, inheritance_graph: dict[str, list[str]]
    ) -> int:
        """Create INHERITS edges from inheritance graph.

        ``child``/``base`` are project-qualified to match Class ``id``.
        """
        if not inheritance_graph:
            return 0
        rows = [
            {"child": f"{project}:{child}", "base": f"{project}:{base}"}
            for child, bases in inheritance_graph.items()
            for base in bases
        ]
        result = self._connected_repo.execute_write(
            "UNWIND $rows AS r "
            "MATCH (child:Class {id: r.child}) "
            "MATCH (base:Class {id: r.base}) "
            "MERGE (child)-[:INHERITS]->(base) "
            "RETURN count(*) AS cnt",
            {"rows": rows},
        )
        return result[0]["cnt"] if result else 0

    @traces(spec="008", story="US-004", reqs=["REQ-008-006"])
    def _sync_dependency_edges(
        self, project: str, dependency_graph: dict[str, list[str]]
    ) -> int:
        """Create IMPORTS edges from dependency graph.

        Matches Module by its project-qualified ``id`` (``"{project}:module:
        {path}"``), not the bare ``path`` property: two projects can share a
        module path (e.g. ``"core"``), and a bare-path match here would wire
        IMPORTS edges across unrelated projects' modules.
        """
        if not dependency_graph:
            return 0
        rows = [
            {
                "source": f"{project}:module:{source}",
                "target": f"{project}:module:{target}",
            }
            for source, targets in dependency_graph.items()
            for target in targets
        ]
        result = self._connected_repo.execute_write(
            "UNWIND $rows AS r "
            "MATCH (src:Module {id: r.source}) "
            "MATCH (tgt:Module {id: r.target}) "
            "MERGE (src)-[:IMPORTS]->(tgt) "
            "RETURN count(*) AS cnt",
            {"rows": rows},
        )
        return result[0]["cnt"] if result else 0

    @traces(spec="008", story="US-004", reqs=["REQ-008-057", "REQ-008-058"])
    def _sync_calls_edges(
        self,
        project: str,
        modules: list[dict],
        root_path: str = "",
        *,
        context_modules: list[dict] | None = None,
    ) -> int:
        """Create CALLS edges from method/function call lists.

        Resolves each call name to a Method/Function id in two scopes, in
        priority order, mirroring real Python name resolution rather than
        an unscoped project-wide "matches something, somewhere" guess
        (which would wire calls to unrelated same-named functions):

        1. **Module-local scope** — functions/methods defined in the
           caller's own module (``_build_local_name_index``).
        2. **Explicit import scope** — names brought into the caller's
           module via ``from <internal-module> import <name>``, resolved to
           the qualified name that internal module locally defines
           (``_build_import_name_index``). This is what closes the
           cross-module CALLS gap (2026-07-18 audit: module-local-only
           resolution wrote 1160/3946 = 29% of CALLS edges even on a fresh
           full sync — most real calls are cross-module).

        Calls that resolve to neither scope are intentionally NOT synced —
        a documented policy, not a silent drop: built-ins, external-library
        calls, and attribute calls through a receiver (``self.x()``,
        ``obj.method()``, plain ``import pkg; pkg.func()``) whose target
        module isn't statically resolvable from a bare call name are out of
        reach of this pass; they need type/scope inference this syncer does
        not do.

        ``context_modules`` is the pool used ONLY to build the name indices
        (local + import) that resolution reads from; the edges actually
        WRITTEN are still restricted to callers in ``modules``. On a
        ``full_sync``, ``modules`` already is the whole project so callers
        leave this ``None`` and get identical behavior. On an
        ``incremental_sync``, ``modules`` is a filtered subset — passing the
        full project's modules here lets ``_build_import_name_index`` see
        import targets OUTSIDE that subset, so cross-module CALLS edges are
        no longer silently dropped on a partial resync (2026-08 fix; P2.29).
        Without it, ``mod_paths`` below would only ever contain the subset's
        own paths and every ``from <module-outside-subset> import <name>``
        would resolve to nothing.
        """
        root_pkg_prefix = f"{Path(root_path).name}." if root_path else ""
        index_modules = context_modules if context_modules is not None else modules
        local_indices = {
            mod.get("path", ""): self._build_local_name_index(mod)
            for mod in index_modules
        }
        mod_paths = set(local_indices)

        rows: list[dict[str, str]] = []
        for mod in modules:
            mod_path = mod.get("path", "")
            local_names = local_indices.get(mod_path) or self._build_local_name_index(
                mod
            )
            import_names = self._build_import_name_index(
                mod, local_indices, mod_paths, root_pkg_prefix
            )

            # Standalone functions
            rows.extend(
                call
                for func in mod.get("functions", [])
                for call in self._calls_from_entity(
                    project, func, local_names, import_names
                )
            )

            # Methods
            rows.extend(
                call
                for cls in mod.get("classes", [])
                for method in cls.get("methods", [])
                for call in self._calls_from_entity(
                    project, method, local_names, import_names
                )
            )

        if not rows:
            return 0
        # Dedupe: an entity calling the same name twice (or two scopes
        # resolving the same pair) would otherwise inflate the MERGE-based
        # "cnt" beyond the number of distinct edges actually written.
        deduped = list({(r["caller"], r["target"]): r for r in rows}.values())
        # Four single-label passes instead of `(:Method|Function {id})` on
        # each endpoint: the multi-label form hits the orchgraph 3.11 UNWIND
        # cardinality collapse (one row per distinct key — see
        # _sync_parameter_edges), which here would drop every call an entity
        # makes beyond its first. An id resolves under exactly one label, so
        # the four label-combination passes are disjoint and their counts add.
        created = 0
        for label_a in ("Function", "Method"):
            for label_b in ("Function", "Method"):
                result = self._connected_repo.execute_write(
                    "UNWIND $rows AS r "
                    f"MATCH (a:{label_a} {{id: r.caller}}) "
                    f"MATCH (b:{label_b} {{id: r.target}}) "
                    "MERGE (a)-[:CALLS]->(b) "
                    "RETURN count(*) AS cnt",
                    {"rows": deduped},
                )
                created += result[0]["cnt"] if result else 0
        return created

    @traces(spec="008", story="US-004", reqs=["REQ-008-056"])
    @staticmethod
    def _build_local_name_index(mod: dict) -> dict[str, str]:
        """Map call-resolvable names → qualified_name for one module.

        Includes bare function names, bare method names, and
        ``Class.method`` qualified method names. Values stay bare
        (unqualified) here — the project prefix is applied in
        ``_calls_from_entity`` when the edge row is built.
        """
        local_names: dict[str, str] = {}
        for func in mod.get("functions", []):
            local_names[func["name"]] = func.get("qualified_name", "")
        for cls in mod.get("classes", []):
            for method in cls.get("methods", []):
                local_names[method["name"]] = method.get("qualified_name", "")
                local_names[f"{cls['name']}.{method['name']}"] = method.get(
                    "qualified_name", ""
                )
        return local_names

    @traces(spec="008", story="US-004", reqs=["REQ-008-056"])
    @staticmethod
    def _build_import_name_index(
        mod: dict,
        local_indices: dict[str, dict[str, str]],
        mod_paths: set[str],
        root_pkg_prefix: str,
    ) -> dict[str, str]:
        """Map call-resolvable names → qualified_name via ``from X import Y``.

        For every ``from <module> import <name>`` in ``mod``'s own import
        list, resolves ``<module>`` to an internal module path (stripping
        the root-package prefix the same way ``CodebaseAnalyzer`` normalizes
        ``dependency_graph``, e.g. ``"orchestrata_ultimate.agents.constants"``
        -> ``"agents.constants"``) and, only if that target module itself
        locally defines a function/method named ``<name>``, maps the call
        name to THAT module's qualified name. This resolves real cross-
        module calls without guessing across unrelated same-named symbols
        project-wide.
        """
        idx: dict[str, str] = {}
        for imp in mod.get("imports", []):
            if not imp.get("is_from"):
                continue
            target = imp.get("module", "") or ""
            if (
                target not in mod_paths
                and root_pkg_prefix
                and target.startswith(root_pkg_prefix)
            ):
                stripped = target[len(root_pkg_prefix) :]
                if stripped in mod_paths:
                    target = stripped
            if target not in mod_paths:
                continue
            target_local = local_indices.get(target, {})
            for name in imp.get("names", []) or []:
                if name in target_local:
                    idx[name] = target_local[name]
        return idx

    @traces(spec="008", story="US-004", reqs=["REQ-008-058"])
    @staticmethod
    def _calls_from_entity(
        project: str,
        entity: dict,
        local_names: dict[str, str],
        import_names: dict[str, str],
    ) -> list[dict[str, str]]:
        """Resolve one entity's call list to CALLS edge rows (skip self-calls).

        Module-local names take priority over the import scope (matching
        Python's own name resolution: a local definition shadows an
        imported one of the same bare name). ``caller``/``target`` are
        project-qualified to match Method/Function ``id``.
        """
        caller = entity.get("qualified_name", "")
        rows: list[dict[str, str]] = []
        for call_name in entity.get("calls", []):
            target = local_names.get(call_name) or import_names.get(call_name)
            if target and target != caller:
                rows.append(
                    {"caller": f"{project}:{caller}", "target": f"{project}:{target}"}
                )
        return rows

    def sync_tests(self, test_ast_path: str, project: str) -> dict[str, int]:
        """Sync test AST and create TESTED_BY relationships.

        Args:
            test_ast_path: Path to a JSON file containing test AST data.
            project: Project name — scopes Test ids and the Method/Function
                match targets (a bare test/entity qualified_name recurs
                across projects, 2026-07-12 multi-project ruling).

        Returns:
            Dict with test_nodes_created and tested_by_edges counts.
        """
        import json as _json

        path = Path(test_ast_path)
        if not path.exists():
            raise FileNotFoundError(f"Test AST file not found: {test_ast_path}")

        with path.open(encoding="utf-8") as fh:
            test_ast = _json.load(fh)

        all_targets = self._load_match_targets(project)

        test_nodes: list[dict[str, Any]] = []
        tested_by_edges: list[dict[str, str]] = []
        for mod in test_ast.get("modules", []):
            mod_path = mod.get("path", "")
            # Only process modules that look like test modules
            if "test" not in mod_path and not mod_path.startswith("tests."):
                continue

            for func in mod.get("functions", []):
                qname = func.get("qualified_name") or f"{mod_path}.{func['name']}"
                self._collect_test_entity(
                    func,
                    f"{project}:{qname}",
                    mod_path,
                    all_targets,
                    test_nodes,
                    tested_by_edges,
                )

            for cls in mod.get("classes", []):
                for method in cls.get("methods", []):
                    qname = method.get("qualified_name") or (
                        f"{mod_path}.{cls['name']}.{method['name']}"
                    )
                    self._collect_test_entity(
                        method,
                        f"{project}:{qname}",
                        mod_path,
                        all_targets,
                        test_nodes,
                        tested_by_edges,
                    )

        return {
            "test_nodes_created": self._persist_test_nodes(test_nodes),
            "tested_by_edges": self._persist_tested_by_edges(tested_by_edges),
        }

    def _load_match_targets(self, project: str) -> dict[str, list[str]]:
        """Map Method/Function node name → list of ids, for test matching.

        Scoped to ``project`` (via the project-qualified ``id`` prefix): an
        unscoped match here would let a test in one project wire a
        TESTED_BY edge to another project's same-named method (2026-07-12
        multi-project ruling).
        """
        prefix = f"{project}:"
        all_targets: dict[str, list[str]] = {}
        for record in self._connected_repo.execute_read(
            "MATCH (m:Method) WHERE m.id STARTS WITH $prefix "
            "RETURN m.name AS name, m.id AS id",
            {"prefix": prefix},
        ):
            all_targets.setdefault(record["name"], []).append(record["id"])
        for record in self._connected_repo.execute_read(
            "MATCH (f:Function) WHERE f.id STARTS WITH $prefix "
            "RETURN f.name AS name, f.id AS id",
            {"prefix": prefix},
        ):
            all_targets.setdefault(record["name"], []).append(record["id"])
        return all_targets

    @staticmethod
    def _test_target_name(test_name: str) -> str | None:
        """Derive the matched target name from a test name (``test_`` heuristic)."""
        if test_name.startswith("test_"):
            return test_name[5:]
        if test_name.startswith("test"):
            return test_name[4:]
        return None

    def _collect_test_entity(
        self,
        entity: dict,
        test_id: str,
        mod_path: str,
        all_targets: dict[str, list[str]],
        test_nodes: list[dict[str, Any]],
        tested_by_edges: list[dict[str, str]],
    ) -> None:
        """Append the Test node and any TESTED_BY edges for one test entity."""
        test_nodes.append(
            {
                "id": test_id,
                "name": entity.get("name", ""),
                "qualified_name": entity.get("qualified_name", ""),
                "module_path": mod_path,
                "file_path": entity.get("file_path", ""),
                "line": entity.get("line", 0),
            }
        )
        target_name = self._test_target_name(entity.get("name", ""))
        if target_name and target_name in all_targets:
            for target_id in all_targets[target_name]:
                tested_by_edges.append({"test_id": test_id, "target_id": target_id})

    def _persist_test_nodes(self, test_nodes: list[dict[str, Any]]) -> int:
        """MERGE Test nodes; returns count created (0 when empty)."""
        if not test_nodes:
            return 0
        self._connected_repo.execute_write(
            "UNWIND $rows AS r "
            "MERGE (t:Test {id: r.id}) "
            "SET t.name = r.name, t.qualified_name = r.qualified_name, "
            "t.module_path = r.module_path, t.file_path = r.file_path, "
            "t.line = r.line",
            {"rows": test_nodes},
        )
        return len(test_nodes)

    def _persist_tested_by_edges(self, tested_by_edges: list[dict[str, str]]) -> int:
        """MERGE TESTED_BY edges; returns edges actually created."""
        if not tested_by_edges:
            return 0
        # Two single-label passes instead of `(:Method|Function {id})`:
        # the multi-label form hits the orchgraph 3.11 UNWIND cardinality
        # collapse (one row per distinct key — see _sync_parameter_edges),
        # which here would keep at most one TESTED_BY edge per target.
        created = 0
        for label in ("Function", "Method"):
            result = self._connected_repo.execute_write(
                "UNWIND $rows AS r "
                f"MATCH (m:{label} {{id: r.target_id}}) "
                "MATCH (t:Test {id: r.test_id}) "
                "MERGE (m)-[:TESTED_BY]->(t) "
                "RETURN count(*) AS cnt",
                {"rows": tested_by_edges},
            )
            created += result[0]["cnt"] if result else 0
        return created

    @traces(spec="008", story="US-004", reqs=["REQ-008-006"])
    def _sync_has_method_edges(self, project: str, modules: list[dict]) -> int:
        """Create DEFINES edges: Class → Method.

        ``class_id``/``method_id`` are project-qualified to match Class/
        Method ``id``.
        """
        rows = []
        for mod in modules:
            for cls in mod.get("classes", []):
                class_id = f"{project}:{cls.get('qualified_name', '')}"
                rows.extend(
                    {
                        "class_id": class_id,
                        "method_id": f"{project}:{method.get('qualified_name', '')}",
                    }
                    for method in cls.get("methods", [])
                )
        if not rows:
            return 0
        result = self._connected_repo.execute_write(
            "UNWIND $rows AS r "
            "MATCH (c:Class {id: r.class_id}) "
            "MATCH (m:Method {id: r.method_id}) "
            "MERGE (c)-[:DEFINES]->(m) "
            "RETURN count(*) AS cnt",
            {"rows": rows},
        )
        return result[0]["cnt"] if result else 0

    # ── Map Classes ────────────────────────────────────────────────────

    @traces(spec="008", story="US-003", reqs=["REQ-008-003"])
    def map_classes(
        self,
        mapping: dict[str, Any],
        source_project: str,
        target_project: str,
    ) -> int:
        """Create MAPS_TO edges from class-mapping.json.

        Args:
            mapping: Parsed class-mapping.json dict.
            source_project: Source project name.
            target_project: Target project name.

        Returns:
            Number of MAPS_TO edges created.
        """
        if not self._repo:
            raise RuntimeError("Not connected. Call connect() first.")

        class_mappings = mapping.get("class_mappings", {})
        rows = [
            {"source": source_class, "target": target_class}
            for source_class, target_class in class_mappings.items()
        ]

        if not rows:
            return 0

        result = self._connected_repo.execute_write(
            "UNWIND $rows AS r "
            "MATCH (src:Class {name: r.source}) "
            "MATCH (tgt:Class {name: r.target}) "
            "MERGE (src)-[:MAPS_TO {source_project: $sp, target_project: $tp}]->(tgt) "
            "RETURN count(*) AS cnt",
            {"rows": rows, "sp": source_project, "tp": target_project},
        )
        return result[0]["cnt"] if result else 0

    # ── Incremental Sync ───────────────────────────────────────────────

    @traces(spec="008", story="US-004", reqs=["REQ-008-004"])
    def delete_modules(self, project: str, modules: Sequence[str]) -> None:
        """Delete these modules and everything they own, nothing else.

        Extracted from :meth:`incremental_sync` (2026-07-26) so a caller that
        needs only the DELETE half can have it. ``ast_live_sync`` is that
        caller: when a source file is deleted there is no AST entry left to
        re-sync, so ``incremental_sync`` cannot express the operation at all
        (it raises on a module path the AST does not contain) -- and without
        this, a deleted file's Module/Class/Function/Method/Parameter nodes
        stayed in the graph forever while the sync happily reported
        "1 removed".

        Args:
            project: Project the modules belong to.
            modules: DOTTED module paths (``knowledge_graph.sync``), not file
                paths -- ids are built as ``"{project}:module:{path}"``.
        """
        for mod_path in modules:
            mod_id = f"{project}:module:{mod_path}"
            # `*0..` is correct HERE (unlike in `_clear_project`): the anchor
            # is the Module itself, so the zero-hop case deletes the module
            # node, which is exactly what is wanted.
            self._connected_repo.execute_write(
                "MATCH (m:Module {id: $id})"
                "-[:DEFINES|HAS_PARAMETER*0..]->(n) "
                "DETACH DELETE n",
                {"id": mod_id},
            )
            # Parameter nodes are ADDITIONALLY reclaimed by parent_id prefix
            # (same reasoning as _clear_project): legacy syncs never wrote
            # HAS_PARAMETER, so this module's Parameters may be unreachable
            # via the traversal above. Idempotent once every Parameter has
            # a real edge.
            self._connected_repo.execute_write(
                "MATCH (param:Parameter) WHERE param.parent_id STARTS WITH $prefix "
                "DETACH DELETE param",
                {"prefix": f"{project}:{mod_path}."},
            )

    @traces(spec="008", story="US-004", reqs=["REQ-008-004"])
    def incremental_sync(
        self,
        ast_data: dict[str, Any],
        project: str,
        modules: list[str],
        *,
        namespace: str | None = None,
    ) -> dict[str, int]:
        """Re-sync specific modules (delete + recreate + repair cross edges).

        Args:
            ast_data: Full AST JSON data.
            project: Project name.
            modules: List of module paths to re-sync.
            namespace: SPEC-033 §4 fact-namespace key (B4a) -- see
                ``full_sync``'s ``namespace`` doc. ``None`` (default) is
                byte-identical to pre-B4a behavior.

        Returns:
            Dict with nodes_created and relationships_created counts.
        """
        if not self._repo:
            raise RuntimeError("Not connected. Call connect() first.")

        module_set = set(modules)
        filtered_modules = [
            mod
            for mod in ast_data.get("modules", [])
            if mod.get("path", "") in module_set
        ]
        missing = module_set - {mod.get("path", "") for mod in filtered_modules}
        if missing:
            raise ValueError(
                "modules not present in AST data (pass DOTTED module paths "
                f"like 'knowledge_graph.sync', not file paths): {sorted(missing)}"
            )

        self.delete_modules(project, modules)

        # Re-sync only the filtered modules. This deliberately does NOT route
        # through full_sync (which clears the whole project) and does not
        # rewrite the Project node's project-wide totals.
        ast_subset = {
            **ast_data,
            "modules": filtered_modules,
        }
        # context_modules carries the FULL project's modules (not the
        # filtered subset) so _sync_calls_edges can resolve `from <module>
        # import <name>` targets that live outside this partial resync --
        # otherwise cross-module CALLS edges are silently dropped (P2.29).
        # Nodes/edges are still only WRITTEN for `filtered_modules`; this is
        # read-only context, not a widening of what gets synced.
        return self._sync_ast(
            ast_subset,
            project,
            sync_project_node=False,
            namespace=namespace,
            context_modules=ast_data.get("modules", []),
        )