orchestrata_ultimate.semantic_graph.adapters.memgraph

Reference › orchestrata_ultimate › semantic_graph › adapters › memgraph

MemgraphSemanticGraph is the production implementation of the SemanticGraph protocol. It reads and writes the shared semantic/RAG graph, the memgraph database, and never the operational orchkg graph. The official Neo4j Bolt driver is used because Memgraph speaks Bolt; that is client naming only. The driver is synchronous and is run in worker threads to keep the async repository interface and avoid event-loop binding problems in tests. repository registers it as the memgraph backend, and the in-memory twin for tests is memory.

• Connection. connect asks _resolve_uri for the current MAIN, which calls main_discovery.resolve_main from main_discovery with the adapter's own config fields and no caching, because a reconnect after failure is exactly when a cached MAIN would be stale. It opens the driver (explicitly unauthenticated unless a username or password is configured), verifies connectivity and calls _ensure_schema for constraints and lookup indexes. It returns False and logs a warning on failure.
• Graph side. upsert_entities, upsert_relationships and get_neighbors, written in batches of write_batch_size.
• Vector side. add_chunks creates the native vector index lazily, sized from the first chunk's embedding (_ensure_vector_index, _wait_for_vector_index, with checks of existing dimension and indexes), and search_similar queries it through the vector_search.search module.
• Retrieval. retrieve takes the nearest chunks and, for a graph depth above zero, expands a matching neighbourhood in one batched query into a GraphRAGContext. search_keyword, matching_module_names, chunk_prominence and link_chunks_to_modules supply the keyword, module and prominence signals that ingestion fuses for hybrid search.
• Maintenance. clear removes this project's chunk label and index in batches, clear_prefixed is the test-isolated cleanup, delete_chunks and chunks_with_id_prefix serve incremental updates, and find_stale_chunk_nodes with prune_stale_chunk_nodes (dry run unless apply is set) removes nodes under stale chunk labels.

Failures raise ConnectionError, QueryError or VectorSearchError from exceptions; a failed query triggers a reconnect.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

MemgraphSemanticGraph

Line 56

Memgraph implementation of the SemanticGraph interface.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/semantic_graph/adapters/memgraph.py

"""Memgraph-backed semantic graph adapter.

Uses the official Neo4j Bolt driver because Memgraph speaks the Bolt protocol.
The synchronous driver is used internally and wrapped in threads to preserve the
async repository interface; this avoids event-loop binding issues in tests.

Vector search uses Memgraph's native HNSW vector index via the
``vector_search.search`` query module. The index is created lazily on the first
``add_chunks`` call using the dimension of the first chunk's embedding.
"""

from __future__ import annotations

import json
import logging
import time
from collections.abc import Callable
from typing import Any, TypeVar

import anyio
import anyio.to_thread

from orchestrata_ultimate.core.graph.retry import _PERMANENT_MARKERS
from orchestrata_ultimate.semantic_graph import main_discovery
from orchestrata_ultimate.semantic_graph.config import SemanticGraphConfig
from orchestrata_ultimate.semantic_graph.exceptions import (
    ConnectionError,
    QueryError,
    VectorSearchError,
)
from orchestrata_ultimate.semantic_graph.models import (
    Chunk,
    Entity,
    GraphRAGContext,
    Relationship,
)

logger = logging.getLogger(__name__)

T = TypeVar("T")

# Stale chunk labels from earlier embedding backends (audit #3). ``Chunk`` is
# also used by the unrelated wiki graph, so any scan/prune over these labels
# MUST additionally scope by the orchestrata chunk-id shape — see
# MemgraphSemanticGraph.find_stale_chunk_nodes.
STALE_CHUNK_LABELS: tuple[str, ...] = ("SemanticChunk", "QwenChunk", "Chunk")

# Row cap for one bulk-write transaction (whole-label delete, chunk→Module
# linking). Prod's SYNC replica must apply every committed transaction inside
# the main's fixed 5 s replication RPC wait; one unbounded DETACH DELETE over
# every 1536-dim chunk took ~54 s to apply there and failed the 2026-10-19
# promote. Bulk writes therefore loop in transactions of at most this many rows.
WRITE_BATCH_SIZE = 500


class MemgraphSemanticGraph:
    """Memgraph implementation of the SemanticGraph interface."""

    def __init__(
        self,
        config: SemanticGraphConfig | None = None,
        write_batch_size: int = WRITE_BATCH_SIZE,
    ) -> None:
        if write_batch_size < 1:
            raise ValueError(f"write_batch_size must be >= 1, got {write_batch_size}")
        self._config = config or SemanticGraphConfig.from_env()
        self._write_batch_size = write_batch_size
        self._driver: Any | None = None
        self._current_uri: str | None = None
        self._vector_index_dimension: int | None = None

    async def connect(self) -> bool:
        """Open the Bolt driver and verify connectivity.

        If an HA controller URL is configured, discover the current MAIN URI from
        the controller before connecting.
        """
        uri = await self._resolve_uri()
        if uri is None:
            logger.warning("No Memgraph URI available from controller or config")
            return False

        def _connect() -> Any:
            from neo4j import basic_auth

            from orchestrata_ultimate.core.graph_factory import NO_AUTH, create_graph

            # Memgraph nodes normally run unauthenticated: NO_AUTH means
            # "explicitly no auth", unlike None which would fall back to the
            # NEO4J_USERNAME/NEO4J_PASSWORD env credentials.
            auth = NO_AUTH
            if self._config.username or self._config.password:
                auth = basic_auth(self._config.username, self._config.password)
            driver = create_graph(uri=uri, auth=auth)
            driver.verify_connectivity()
            return driver

        driver: Any | None = None
        try:
            driver = await anyio.to_thread.run_sync(_connect)
            self._driver = driver
            self._current_uri = uri
            await self._ensure_schema()
            return True
        except Exception as exc:
            logger.warning("Memgraph connection failed: %s", exc)
            if driver is not None:
                try:
                    await anyio.to_thread.run_sync(driver.close)
                except Exception as close_exc:  # pragma: no cover - defensive cleanup
                    logger.debug(
                        "Failed to close failed Memgraph driver: %s", close_exc
                    )
            self._driver = None
            return False

    async def _resolve_uri(self) -> str | None:
        """Return a Bolt URI to connect to -- the current Memgraph MAIN.

        Delegates to :func:`main_discovery.resolve_main`, built from this
        adapter's own config fields (not the process environment) so a
        caller with a custom ``SemanticGraphConfig`` (e.g. a non-default env
        key, or a config built directly in tests) is honored exactly.

        Always resolves fresh (``cache=False``): ``connect()`` runs once at
        startup and again -- via ``_execute_with_reconnect`` -- after a
        connection failure, which is exactly when a stale cached MAIN would
        be wrong (a restart failing MAIN over is the scenario this exists
        to survive; see the ``main_discovery`` module docstring).
        """
        env = {
            "MEMGRAPH_HA_CONTROLLER_URL": self._config.ha_controller_url or "",
            "MEMGRAPH_URIS": ",".join(self._config.uris),
            "MEMGRAPH_URI": self._config.uri or "",
        }

        def _resolve() -> str:
            return main_discovery.resolve_main(env, cache=False)

        try:
            return await anyio.to_thread.run_sync(_resolve)
        except ConnectionError as exc:
            logger.warning("Memgraph MAIN discovery failed: %s", exc)
            return None

    async def close(self) -> None:
        """Close the Bolt driver."""
        if self._driver is not None:
            await anyio.to_thread.run_sync(self._driver.close)
            self._driver = None

    @property
    def _chunk_label(self) -> str:
        """Return the configured node label for chunks."""
        return self._config.chunk_label

    def _session(self) -> Any:
        """Open a driver session, targeting ``config.database`` when set.

        Default (None) keeps the historic server-default-database behavior;
        a configured database gives the caller its own index/label universe
        on a multi-tenant cluster (test worker isolation, 2026-08-11).
        """
        driver = self._driver
        if driver is None:
            raise ConnectionError(
                "Memgraph driver is not connected — call connect() first"
            )
        if self._config.database:
            return driver.session(database=self._config.database)
        return driver.session()

    async def _ensure_schema(self) -> None:
        """Create uniqueness constraints, lookup indexes, and detect the
        vector index.

        The explicit indexes are load-bearing: in Memgraph a uniqueness
        constraint does NOT back lookups, and without a label(+property)
        index every ``MATCH (e:Entity {id: ...})`` scans the whole graph —
        ~742K nodes on the shared cluster turned a seconds-long batched
        UNWIND update into a >120s timeout (2026-07-08).
        """
        label = self._chunk_label
        statements = [
            # Memgraph dialect, not Neo4j's: this adapter only ever talks to
            # Memgraph, and "CREATE CONSTRAINT <name> IF NOT EXISTS FOR ...
            # REQUIRE ..." is a parse error there — so both uniqueness
            # constraints raised on every call and were swallowed by the
            # debug-level except below. Neither has ever existed on the
            # cluster. Memgraph constraints are unnamed and re-creation is a
            # silent no-op (measured live 2026-07-26).
            "CREATE CONSTRAINT ON (e:Entity) ASSERT e.id IS UNIQUE",
            f"CREATE CONSTRAINT ON (c:{label}) ASSERT c.id IS UNIQUE",
            # CREATE INDEX has no IF NOT EXISTS in Memgraph; duplicates raise
            # and are absorbed by the try/except below.
            "CREATE INDEX ON :Entity",
            "CREATE INDEX ON :Entity(id)",
            f"CREATE INDEX ON :{label}",
            f"CREATE INDEX ON :{label}(id)",
            # Module(name) exists for ONE reason: `link_chunks_to_modules`
            # does `MERGE (m:Module {name: c.module})`, and MERGE matches on
            # the exact property pattern it is given. Without this index that
            # MERGE label-scans every Module once per chunk -- measured
            # 2026-08-17 on prod at 7188 chunks x 789 modules, a single
            # transaction that ran ~15 minutes and blocked a promotion.
            #
            # An INDEX, deliberately, not a constraint: the docstring on
            # link_chunks_to_modules explains at length why Memgraph's Module
            # identity is NOT unified with canonical Neo4j's `id` key, and
            # that reasoning stands. This changes no MERGE semantics and
            # declares no key -- it only stops the scan.
            "CREATE INDEX ON :Module",
            "CREATE INDEX ON :Module(name)",
        ]
        for cypher in statements:
            try:
                await self._run_write(cypher)
            except (
                Exception
            ) as exc:  # pragma: no cover - already-present constraint/index
                logger.debug(
                    "Schema statement skipped (likely already present): %s", exc
                )

        await self._load_vector_index_dimension()

    async def upsert_entities(self, entities: list[Entity]) -> None:
        if not entities:
            return

        def _upsert(driver: Any, entity: Entity) -> None:
            props = dict(entity.properties)
            props.pop("id", None)
            props.pop("labels", None)
            with self._session() as session:
                session.run(
                    """
                    MERGE (e:Entity {id: $id})
                    SET e.labels = $labels, e += $props
                    """,
                    {
                        "id": entity.id,
                        "labels": json.dumps(entity.labels),
                        "props": props,
                    },
                ).consume()

        for entity in entities:
            await self._execute_with_reconnect(_upsert, entity)

    async def upsert_relationships(self, relationships: list[Relationship]) -> None:
        if not relationships:
            return

        def _upsert(driver: Any, rel: Relationship) -> None:
            props = dict(rel.properties)
            props.pop("id", None)
            cypher = (
                f"MATCH (a:Entity {{id: $source_id}}), (b:Entity {{id: $target_id}}) "
                f"MERGE (a)-[r:{rel.type} {{id: $id}}]->(b) "
                f"SET r += $props"
            )
            with self._session() as session:
                session.run(
                    cypher,
                    {
                        "source_id": rel.source_id,
                        "target_id": rel.target_id,
                        "id": rel.id,
                        "props": props,
                    },
                ).consume()

        for rel in relationships:
            await self._execute_with_reconnect(_upsert, rel)

    async def get_neighbors(
        self, entity_id: str, depth: int = 1
    ) -> tuple[list[Entity], list[Relationship]]:
        if not isinstance(depth, int):
            raise QueryError("depth must be an integer between 1 and 10")
        if depth < 1:
            return ([], [])
        if depth > 10:
            raise QueryError("depth must be an integer between 1 and 10")

        # Memgraph does not support parameterizing the upper bound of a
        # variable-length pattern, so the depth literal is interpolated safely
        # after the integer validation above. We use OPTIONAL MATCH so that the
        # start entity is always returned even when it has no outgoing edges,
        # matching the behaviour of the in-memory adapter. We also return
        # ``start`` and ``end`` explicitly because Memgraph only populates node
        # properties for nodes returned in the query, which we need to build
        # relationship IDs.
        cypher = (
            f"MATCH (start:Entity {{id: $id}}) "
            f"OPTIONAL MATCH (start)-[r*1..{depth}]->(end:Entity) "
            f"RETURN DISTINCT start, end, r AS rels"
        )

        def _fetch(driver: Any) -> tuple[list[Entity], list[Relationship]]:
            with self._session() as session:
                result = session.run(cypher, {"id": entity_id})
                return _records_to_entities_relationships(result)

        return await self._execute_with_reconnect(_fetch)

    async def add_chunks(self, chunks: list[Chunk]) -> None:
        if not chunks:
            return

        first_embedding = next((c.embedding for c in chunks if c.embedding), None)
        if first_embedding is not None:
            await self._ensure_vector_index(len(first_embedding))

        label = self._chunk_label

        def _add(driver: Any, chunk: Chunk) -> None:
            props = dict(chunk.metadata)
            props.pop("id", None)
            props.pop("content", None)
            props.pop("embedding", None)
            with self._session() as session:
                session.run(
                    f"""
                    MERGE (c:{label} {{id: $id}})
                    SET c.content = $content,
                        c.embedding = $embedding,
                        c += $props
                    """,
                    {
                        "id": chunk.id,
                        "content": chunk.content,
                        "embedding": chunk.embedding,
                        "props": props,
                    },
                ).consume()

        for chunk in chunks:
            await self._execute_with_reconnect(_add, chunk)

    async def _load_vector_index_dimension(self) -> None:
        """Set ``_vector_index_dimension`` from an existing index, if any."""
        info = await self._vector_index_info()
        if info is not None:
            self._vector_index_dimension = info["dimension"]

    async def _vector_index_info(self) -> dict[str, Any] | None:
        """Return metadata for the configured vector index, or None if absent."""

        def _fetch(driver: Any) -> dict[str, Any] | None:
            with self._session() as session:
                result = session.run(
                    "CALL vector_search.show_index_info() YIELD index_name, dimension "
                    "WHERE index_name = $index_name "
                    "RETURN dimension",
                    {"index_name": self._config.vector_index_name},
                )
                record = result.single()
                if record is None:
                    return None
                return {"dimension": record["dimension"]}

        return await self._execute_with_reconnect(_fetch)

    async def _ensure_vector_index(self, dimension: int) -> None:
        """Create the native vector index lazily if it does not yet exist."""
        if self._vector_index_dimension == dimension:
            return

        label = self._chunk_label
        existing = await self._all_vector_indexes()
        logger.debug("Existing vector indexes: %s", existing)
        for index in existing:
            if index["label"] == label and index["property"] == "embedding":
                if index["dimension"] == dimension:
                    self._vector_index_dimension = dimension
                    return
                raise QueryError(
                    f"Vector index {index['name']} has dimension {index['dimension']}, "
                    f"but chunks have dimension {dimension}. "
                    "Clear the graph or use a different chunk_label before changing embedders."
                )

        def _create(driver: Any) -> None:
            cypher = (
                "CREATE VECTOR INDEX "
                f"{self._config.vector_index_name} ON :{label}(embedding) "
                "WITH CONFIG {"
                f"dimension: {dimension}, "
                f"capacity: {self._config.vector_index_capacity}, "
                f"metric: '{self._config.vector_index_metric}'"
                "}"
            )
            with self._session() as session:
                session.run(cypher).consume()

        logger.info(
            "Creating vector index %s with dimension %s",
            self._config.vector_index_name,
            dimension,
        )
        await self._execute_with_reconnect(_create, max_storage_retries=20)
        await self._wait_for_vector_index()
        self._vector_index_dimension = dimension

    async def _wait_for_vector_index(self) -> None:
        """Poll until the configured vector index is reported by Memgraph.

        Memgraph builds vector indexes asynchronously, so a freshly issued
        ``CREATE VECTOR INDEX`` may not be queryable immediately. Polling
        prevents race conditions between index creation and search.
        """
        deadline = time.monotonic() + self._config.vector_index_timeout
        delay = 0.05
        while time.monotonic() < deadline:
            info = await self._vector_index_info()
            if info is not None:
                logger.debug("Vector index %s is ready", self._config.vector_index_name)
                return
            await anyio.sleep(delay)
            delay = min(delay * 2, 0.5)
        # Memgraph builds vector indexes synchronously in practice; this path
        # only fires if index creation hangs.
        raise QueryError(  # pragma: no cover
            f"Vector index {self._config.vector_index_name} was not ready within "
            f"{self._config.vector_index_timeout}s"
        )

    async def _all_vector_indexes(self) -> list[dict[str, Any]]:
        """Return metadata for all vector indexes in the database."""

        def _fetch(driver: Any) -> list[dict[str, Any]]:
            with self._session() as session:
                result = session.run(
                    "CALL vector_search.show_index_info() "
                    "YIELD index_name, label, property, dimension "
                    "RETURN index_name, label, property, dimension"
                )
                return [
                    {
                        "name": record["index_name"],
                        "label": record["label"].lstrip(":"),
                        "property": record["property"],
                        "dimension": record["dimension"],
                    }
                    for record in result
                ]

        return await self._execute_with_reconnect(_fetch)

    async def _drop_vector_index_by_name(self, name: str) -> None:
        """Drop a vector index by its exact name."""

        def _drop(driver: Any) -> None:
            with self._session() as session:
                try:
                    session.run(f"DROP VECTOR INDEX {name}").consume()
                except Exception as exc:
                    # Tolerate a genuinely-absent index only. Transient
                    # storage contention MUST propagate so
                    # _execute_with_reconnect backs off and retries —
                    # swallowing it here leaked 148 __TEST_ vector indexes
                    # onto the shared cluster (found 2026-07-12).
                    if self._is_storage_contention(exc):
                        raise
                    logger.debug("Vector index drop failed for %s: %s", name, exc)

        await self._execute_with_reconnect(_drop, max_storage_retries=20)
        self._vector_index_dimension = None

    async def search_similar(
        self, embedding: list[float], top_k: int = 5
    ) -> list[Chunk]:
        if not embedding:
            return []
        if self._vector_index_dimension is None:
            # No chunks have been added yet, so the index does not exist.
            return []
        if len(embedding) != self._vector_index_dimension:
            raise VectorSearchError(
                f"Query embedding has dimension {len(embedding)}, "
                f"but the vector index {self._config.vector_index_name!r} "
                f"has dimension {self._vector_index_dimension}. "
                "Use an embedder that produces vectors matching the index's dimension."
            )

        cypher = (
            "CALL vector_search.search($index_name, $top_k, $query) "
            "YIELD node, similarity "
            "RETURN node "
            "ORDER BY similarity DESC"
        )

        def _search(driver: Any) -> list[Chunk]:
            with self._session() as session:
                current_top_k = top_k
                while current_top_k >= 1:
                    try:
                        result = session.run(
                            cypher,
                            {
                                "index_name": self._config.vector_index_name,
                                "top_k": current_top_k,
                                "query": embedding,
                            },
                        )
                        return [_node_to_chunk(record["node"]) for record in result]
                    except Exception as exc:
                        # Defensive retry for a transient Memgraph bug where a
                        # deleted object is returned during vector search. This
                        # is impossible to trigger deterministically in tests.
                        error_message = str(exc).lower()
                        if (
                            "deleted object" not in error_message or current_top_k <= 1
                        ):  # pragma: no cover
                            raise
                        logger.warning(  # pragma: no cover
                            "Vector search returned deleted object with top_k=%s, retrying with top_k=%s",
                            current_top_k,
                            current_top_k - 1,
                        )
                        current_top_k -= 1  # pragma: no cover
                return []  # pragma: no cover

        return await self._execute_with_reconnect(_search)

    async def retrieve(
        self,
        query_embedding: list[float],
        top_k: int = 5,
        graph_depth: int = 1,
    ) -> GraphRAGContext:
        chunks = await self.search_similar(query_embedding, top_k)
        entities: dict[str, Entity] = {}
        relationships: dict[str, Relationship] = {}

        if graph_depth > 0 and chunks:
            ents, rels = await self._matching_neighborhood(
                [chunk.content for chunk in chunks], graph_depth
            )
            for e in ents:
                entities[e.id] = e
            for r in rels:
                relationships[r.id] = r

        return GraphRAGContext(
            entities=list(entities.values()),
            relationships=list(relationships.values()),
            chunks=chunks,
        )

    async def _matching_neighborhood(
        self, texts: list[str], depth: int
    ) -> tuple[list[Entity], list[Relationship]]:
        """Retrieve's entity match + graph expansion as ONE batched query.

        Replaces the old per-entity fan-out — a full-graph ``_all_entities()``
        fetch (0.90s) plus one ``get_neighbors`` Bolt round-trip PER matched
        entity (measured 2026-08-12: 10,964 round-trips, 16.19s for a single
        ``retrieve(top_k=5, depth=1)`` on the real 13,747-entity graph). The
        substring candidate heuristic is UNCHANGED — an entity matches when
        its id or any of its string property values appears in a chunk's text
        — but the scan is pushed server-side into the same query as the
        variable-depth expansion, so the whole graph side of a retrieve costs
        one round-trip.

        The ``valueType`` guards matter: the shared graph holds imported
        nodes with non-string ``id``/property values, and ``CONTAINS`` on a
        non-string raises a server error (same lesson as ``clear_test_data``).
        """
        if not isinstance(depth, int) or depth < 1 or depth > 10:
            raise QueryError("depth must be an integer between 1 and 10")

        # Depth literal interpolated for the same reason as get_neighbors:
        # Memgraph cannot parameterize a variable-length pattern's bound,
        # and the integer validation above keeps the interpolation safe.
        cypher = (
            "MATCH (start:Entity) "
            "WHERE any(text IN $texts WHERE "
            "(valueType(start.id) = 'STRING' AND text CONTAINS start.id) "
            "OR any(k IN keys(start) "
            "WHERE valueType(start[k]) = 'STRING' AND text CONTAINS start[k])) "
            f"OPTIONAL MATCH (start)-[r*1..{depth}]->(end:Entity) "
            "RETURN DISTINCT start, end, r AS rels"
        )

        def _fetch(driver: Any) -> tuple[list[Entity], list[Relationship]]:
            with self._session() as session:
                result = session.run(cypher, {"texts": texts})
                return _records_to_entities_relationships(result)

        return await self._execute_with_reconnect(_fetch)

    async def _all_entities(self) -> list[Entity]:
        """Return all entities in the graph."""

        def _fetch(driver: Any) -> list[Entity]:
            with self._session() as session:
                result = session.run("MATCH (e:Entity) RETURN e")
                return [_node_to_entity(record["e"]) for record in result]

        return await self._execute_with_reconnect(_fetch)

    async def clear(self) -> None:
        """Clear THIS project's data: the vector index and its chunk-label
        nodes. Deliberately does NOT touch ``:Entity`` nodes.

        The Memgraph cluster is SHARED (knowledge-wiki's enhancement layer
        stores ``Entity{Spec/Requirement/Story/CodeElement}`` + IMPLEMENTS
        edges here). An earlier unscoped ``MATCH (e:Entity) DETACH DELETE e``
        meant every production ingest run wiped the wiki project's entities
        (found + ruled 2026-07-10). Entities are MERGE-by-id upserts, so
        re-ingest overwrites in place without any delete; stale-orphan
        pruning is a separate, deliberate, scoped operation
        (``clear_prefixed`` / ``clear_test_data``).
        """
        # Drop the vector index first so stale index entries do not leak into
        # subsequent searches after the nodes are deleted.
        await self._drop_vector_index()
        await self._delete_chunk_label_batched()

    async def _delete_chunk_label_batched(self) -> int:
        """DETACH DELETE every chunk-label node, ``_write_batch_size`` per
        transaction (see ``WRITE_BATCH_SIZE``). Returns the rows deleted."""
        return await self._run_batched_write(
            f"MATCH (c:{self._chunk_label}) WITH c LIMIT $batch "
            "DETACH DELETE c RETURN count(*) AS n"
        )

    async def clear_prefixed(self, prefix: str) -> None:
        """Test-isolated cleanup: drop index, delete chunks by label, and delete
        only entities whose id starts with ``prefix``.

        This avoids parallel test workers racing on a shared ``Entity`` label.
        """
        await self._drop_vector_index()
        await self._delete_chunk_label_batched()
        await self._run_write(
            "MATCH (e:Entity) WHERE e.id STARTS WITH $prefix DETACH DELETE e",
            {"prefix": prefix},
        )

    async def drop_chunk_label_schema(self) -> None:
        """Drop the per-chunk-label constraint and lookup indexes that
        ``_ensure_schema`` created for ``self._chunk_label``.

        Test-teardown counterpart to ``_ensure_schema``: ``clear_prefixed``
        deletes NODES and the vector index but Memgraph keeps the (empty)
        label's constraint + indexes forever, and boot recovery replays every
        one of them — 3,379 leaked ``*__TEST_*`` labels (6,713 indexes +
        3,323 constraints) cost the shared cluster ~17 minutes of pegged CPU
        per data-node restart before the 2026-07-15 purge. Call this AFTER
        the last ``clear_prefixed`` of a test, never before the test body
        (the schema is load-bearing while the test runs). Deliberately never
        touches the shared ``:Entity`` schema. A production label's schema is
        recreated by the next ``connect()``, but there is no reason to call
        this outside test teardown.
        """
        label = self._chunk_label
        statements = [
            f"DROP CONSTRAINT ON (c:{label}) ASSERT c.id IS UNIQUE",
            f"DROP INDEX ON :{label}(id)",
            f"DROP INDEX ON :{label}",
        ]
        for cypher in statements:
            try:
                await self._run_write(cypher)
            except Exception as exc:
                # Absent constraint/index is benign (mirrors _ensure_schema's
                # duplicate-absorb). Contention that escapes _run_write has
                # exhausted its retries — propagate so the leak is LOUD, not
                # silent (see _drop_vector_index_by_name).
                if self._is_storage_contention(exc):
                    raise
                logger.debug("Schema drop skipped (likely absent): %s", exc)

    async def link_chunks_to_modules(self) -> None:
        """Create Module nodes and BELONGS_TO edges from chunk metadata.

        Chunks must have a ``module`` metadata key. The operation is idempotent.

        NOT a MERGE-key/constraint-key mismatch (checked 2026-07-22, twin-bug
        audit of ``canonical_schema.CANONICAL_MERGE_KEYS`` vs live MERGE
        sites): ``Module.name`` here targets MEMGRAPH (this adapter's own
        graph, the shared semantic/RAG store), not Neo4j. ``canonical_schema``
        constraints are Neo4j-only DDL (``apply()`` takes a
        ``SyncGraphRepository`` over the OPERATIONAL graph); this adapter's
        own ``_ensure_schema`` never creates a ``Module`` constraint on
        Memgraph at all (only ``Entity``/the configured chunk label) --
        canonical Neo4j's ``module_id`` constraint simply does not reach
        here, so there is no divergent-key collision to fix. If Memgraph
        Module identity is ever unified with the canonical `id` scheme, that
        is a real design decision (a project-qualified id + a Memgraph-side
        constraint), not a mechanical MERGE-key rename.

        Bounded (2026-10-19): each transaction links at most
        ``_write_batch_size`` not-yet-linked chunks and the loop stops when a
        pass links none, so a re-run over a linked corpus is one empty pass.
        """
        await self._run_batched_write(f"""
            MATCH (c:{self._chunk_label})
            WHERE c.module IS NOT NULL AND NOT (c)-[:BELONGS_TO]->(:Module)
            WITH c LIMIT $batch
            MERGE (m:Module {{name: c.module}})
            MERGE (c)-[:BELONGS_TO]->(m)
            RETURN count(*) AS n
            """)

    async def search_keyword(
        self, terms: list[str], limit_per_term: int = 50
    ) -> list[tuple[Chunk, int]]:
        """Return chunks whose content contains any of ``terms``.

        One case-insensitive ``CONTAINS`` query is issued per term (capped at
        ``limit_per_term`` matches each) and results are unioned in Python,
        deduplicated by chunk id, with a per-chunk count of how many distinct
        terms matched (used as the keyword-ranking signal for fusion).
        """
        if not terms:
            return []

        label = self._chunk_label

        def _search(driver: Any) -> list[tuple[Chunk, int]]:
            counts: dict[str, int] = {}
            chunks: dict[str, Chunk] = {}
            with self._session() as session:
                for term in terms:
                    result = session.run(
                        f"""
                        MATCH (c:{label})
                        WHERE toLower(c.content) CONTAINS toLower($term)
                        RETURN c
                        LIMIT $limit
                        """,
                        {"term": term, "limit": limit_per_term},
                    )
                    for record in result:
                        chunk = _node_to_chunk(record["c"])
                        counts[chunk.id] = counts.get(chunk.id, 0) + 1
                        chunks[chunk.id] = chunk
            return [(chunks[cid], counts[cid]) for cid in chunks]

        return await self._execute_with_reconnect(_search)

    async def matching_module_names(self, terms: list[str]) -> set[str]:
        """Return lower-cased ``Module.name`` values that match any of ``terms``."""
        if not terms:
            return set()

        lowered_terms = [t.lower() for t in terms]

        def _fetch(driver: Any) -> set[str]:
            with self._session() as session:
                result = session.run(
                    "MATCH (m:Module) WHERE toLower(m.name) IN $terms "
                    "RETURN toLower(m.name) AS name",
                    {"terms": lowered_terms},
                )
                return {record["name"] for record in result}

        return await self._execute_with_reconnect(_fetch)

    async def chunk_prominence(self, chunk_ids: list[str]) -> dict[str, float]:
        """Return ``chunk_id -> prominence`` for chunks linked to a prominent Entity.

        Follows ``BELONGS_TO``/``MENTIONS`` edges from a chunk to an
        ``:Entity`` node carrying a numeric ``prominence`` property. When a
        chunk links to more than one such entity, the maximum prominence wins.
        """
        if not chunk_ids:
            return {}

        label = self._chunk_label

        def _fetch(driver: Any) -> dict[str, float]:
            prominences: dict[str, float] = {}
            with self._session() as session:
                result = session.run(
                    f"""
                    MATCH (c:{label})-[:BELONGS_TO|MENTIONS]->(e:Entity)
                    WHERE c.id IN $ids AND e.prominence IS NOT NULL
                    RETURN c.id AS cid, e.prominence AS prominence
                    """,
                    {"ids": chunk_ids},
                )
                for record in result:
                    cid = record["cid"]
                    prominence = float(record["prominence"])
                    if cid not in prominences or prominence > prominences[cid]:
                        prominences[cid] = prominence
            return prominences

        return await self._execute_with_reconnect(_fetch)

    async def chunks_with_id_prefix(self, prefixes: list[str]) -> list[Chunk]:
        """Return chunks (of this adapter's configured label) whose id starts
        with any of ``prefixes``.

        Generic, label-scoped id-prefix scan -- used by
        :class:`orchestrata_ultimate.semantic_graph.architecture_twin.ArchitectureTwin`
        to find its own stale chunks without touching any other label or id
        shape, but reusable by any caller that owns a dedicated chunk label
        and wants to enumerate a subset of it by id prefix.
        """
        if not prefixes:
            return []

        label = self._chunk_label

        def _fetch(driver: Any) -> list[Chunk]:
            with self._session() as session:
                result = session.run(
                    f"""
                    MATCH (c:{label})
                    WHERE any(p IN $prefixes WHERE c.id STARTS WITH p)
                    RETURN c
                    """,
                    {"prefixes": prefixes},
                )
                return [_node_to_chunk(record["c"]) for record in result]

        return await self._execute_with_reconnect(_fetch)

    async def delete_chunks(self, ids: list[str]) -> None:
        """Delete chunks (of this adapter's configured label) by exact id."""
        if not ids:
            return

        label = self._chunk_label

        def _delete(driver: Any) -> None:
            with self._session() as session:
                session.run(
                    f"MATCH (c:{label}) WHERE c.id IN $ids DETACH DELETE c",
                    {"ids": ids},
                ).consume()

        await self._execute_with_reconnect(_delete)

    async def find_stale_chunk_nodes(
        self, labels: list[str] | None = None
    ) -> list[dict[str, Any]]:
        """Return nodes with a stale chunk label that carry the orchestrata
        chunk-id shape (audit #3).

        ``CodestralChunk`` is the canonical, fully-embedded chunk label.
        Older ingestion runs left behind ``SemanticChunk``/``QwenChunk``/
        ``Chunk`` nodes. The bare ``Chunk`` label is ALSO used by the
        unrelated wiki graph (``chunk_id`` property + ``DERIVED_FROM``->
        ``Page`` edges), so this scopes matches to nodes that have an
        orchestrata-style ``id`` property, no ``chunk_id`` property, and no
        outgoing ``DERIVED_FROM`` edge — never the wiki graph's nodes.
        """
        labels = labels or list(STALE_CHUNK_LABELS)

        def _fetch(driver: Any) -> list[dict[str, Any]]:
            with self._session() as session:
                found: list[dict[str, Any]] = [
                    {"id": record["id"], "label": label}
                    for label in labels
                    for record in session.run(f"""
                        MATCH (c:{label})
                        WHERE c.id IS NOT NULL
                          AND c.chunk_id IS NULL
                          AND NOT (c)-[:DERIVED_FROM]->()
                        RETURN c.id AS id
                        """)
                ]
            return found

        return await self._execute_with_reconnect(_fetch)

    async def prune_stale_chunk_nodes(
        self, labels: list[str] | None = None, apply: bool = False
    ) -> int:
        """Report (or, with ``apply=True``, delete) stale chunk nodes.

        Returns the count of stale nodes found (dry-run) or deleted
        (``apply=True``). Never touches wiki-graph ``Chunk`` nodes — see
        :meth:`find_stale_chunk_nodes` for the exact scoping rule.
        """
        labels = labels or list(STALE_CHUNK_LABELS)
        stale = await self.find_stale_chunk_nodes(labels)
        if not apply or not stale:
            return len(stale)

        def _delete(driver: Any) -> None:
            with self._session() as session:
                for label in labels:
                    session.run(f"""
                        MATCH (c:{label})
                        WHERE c.id IS NOT NULL
                          AND c.chunk_id IS NULL
                          AND NOT (c)-[:DERIVED_FROM]->()
                        DETACH DELETE c
                        """).consume()

        await self._execute_with_reconnect(_delete)
        return len(stale)

    async def _drop_vector_index(self) -> None:
        """Drop the native vector index so the next test starts clean."""

        def _drop(driver: Any) -> None:
            with self._session() as session:
                try:
                    session.run(
                        f"DROP VECTOR INDEX {self._config.vector_index_name}"
                    ).consume()
                except Exception as exc:
                    # See _drop_vector_index_by_name: contention must reach
                    # the retry wrapper; only absent-index errors are benign.
                    if self._is_storage_contention(exc):
                        raise
                    logger.debug("Vector index drop failed (may not exist): %s", exc)

        await self._execute_with_reconnect(_drop, max_storage_retries=20)
        self._vector_index_dimension = None

    async def clear_test_data(self, prefix: str) -> None:
        """Remove nodes whose id starts with ``prefix`` (used by tests).

        The ``CASE`` guard matters: the shared graph holds imported
        operational nodes, and ``STARTS WITH`` raises a server error when it
        meets a non-string ``id`` — one bad node poisons the whole sweep.
        A plain ``valueType(...) = 'STRING' AND ...`` conjunct does NOT help:
        Memgraph's planner gives no short-circuit guarantee (verified
        2026-07-08); only ``CASE`` forces the type check to gate evaluation.
        """
        await self._run_write(
            "MATCH (n) WHERE CASE valueType(n.id) WHEN 'STRING' "
            "THEN n.id STARTS WITH $prefix ELSE false END DETACH DELETE n",
            {"prefix": prefix},
        )

    # Memgraph aborts any query that cannot acquire storage access within
    # ``storage.access_timeout_sec``. On the shared HA cluster the periodic
    # snapshots (and concurrent writers) transiently hold that access, so a
    # write/read surfaces a RETRYABLE ``TransientError`` — "Cannot get unique/
    # shared access to the storage. Try stopping other parallel queries. Note:
    # Periodic snapshots also hold storage access." This is expected
    # steady-state behaviour on the shared cluster, not a fault: the snapshot /
    # competing writer is short-lived, so back off and retry the SAME driver.
    _STORAGE_CONTENTION_MARKERS = (
        "cannot get unique access to the storage",
        "cannot get shared access to the storage",
        "storage.access_timeout",
        "try stopping other parallel queries",
    )

    @classmethod
    def _is_storage_contention(cls, exc: Exception) -> bool:
        """True when *exc* is Memgraph's transient storage-access contention.

        The permanent-marker guard MUST run BEFORE the type-name check:
        Memgraph reports a nonexistent database (gql status 50N42, "Tried to
        retrieve an unknown database") as a ``TransientError``, so the bare
        ``type name == "TransientError"`` disjunct below would otherwise burn
        the whole storage-retry budget (~75s of backoff PER operation, and
        ``connect()`` to a nonexistent db never failed within 250s — measured
        2026-08-12) on a failure no retry can ever clear. The marker list is
        shared with ``core.graph.retry`` (``_PERMANENT_MARKERS``), not
        duplicated here.
        """
        message = str(exc).lower()
        if any(marker.lower() in message for marker in _PERMANENT_MARKERS):
            return False
        return type(exc).__name__ == "TransientError" or any(
            marker in message for marker in cls._STORAGE_CONTENTION_MARKERS
        )

    async def _execute_with_reconnect(
        self,
        operation: Callable[..., T],
        *args: Any,
        max_retries: int = 1,
        max_storage_retries: int = 12,
    ) -> T:
        """Run a sync operation in a thread with two independent retry classes.

        A *connection* error reconnects to the (possibly new) MAIN, bounded by
        ``max_retries``. A transient *storage-access-contention* error backs off
        and retries the same driver, bounded by ``max_storage_retries`` — the
        snapshot or competing writer that held storage access clears quickly.
        """
        conn_attempts = 0
        storage_attempts = 0
        while True:
            if self._driver is None:
                connected = await self.connect()
                if not connected:
                    raise ConnectionError("Unable to connect to Memgraph")
            try:
                return await anyio.to_thread.run_sync(operation, self._driver, *args)
            except Exception as exc:
                error_message = str(exc).lower()
                is_storage_contention = self._is_storage_contention(exc)
                if is_storage_contention and storage_attempts < max_storage_retries:
                    storage_attempts += 1
                    # exponential backoff capped at 10s: total window across
                    # 12 retries ~75s, comfortably outlasting a periodic
                    # snapshot's storage-access hold on the shared cluster.
                    backoff = min(0.25 * (2 ** (storage_attempts - 1)), 10.0)
                    logger.warning(
                        "Memgraph storage-access contention (attempt %d/%d), "
                        "backing off %.2fs: %s",
                        storage_attempts,
                        max_storage_retries,
                        backoff,
                        exc,
                    )
                    await anyio.sleep(backoff)
                    continue
                # Failover reconnect is defensive code for transient network
                # partitions; forcing a real connection failure mid-operation is
                # not practical in unit tests.
                is_connection_error = any(  # pragma: no cover
                    marker in error_message
                    for marker in (
                        "connection refused",
                        "failed to establish connection",
                        "serviceunavailable",
                        "sessionexpired",
                        "broken pipe",
                        "connection reset",
                    )
                )
                if (
                    conn_attempts >= max_retries or not is_connection_error
                ):  # pragma: no cover
                    raise
                conn_attempts += 1
                logger.warning(  # pragma: no cover
                    "Memgraph operation failed, attempting failover reconnect: %s", exc
                )
                await self.close()  # pragma: no cover

    async def _run_write(
        self, cypher: str, params: dict[str, Any] | None = None
    ) -> None:
        """Run a write query in a thread."""

        def _write(driver: Any) -> None:
            with self._session() as session:
                # consume() so server-side failures raise here; an unconsumed
                # write result is silently discarded on session close and the
                # failure never surfaces (verified against Memgraph 2026-07-08).
                session.run(cypher, params or {}).consume()

        await self._execute_with_reconnect(_write)

    async def _run_batched_write(self, cypher: str) -> int:
        """Repeat a ``LIMIT $batch`` write, one transaction per pass, until a
        pass reports ``n == 0``. ``cypher`` must bind ``$batch`` and return a
        single ``count(*) AS n`` row. Returns the total rows written."""
        bound = {"batch": self._write_batch_size}

        def _write(driver: Any) -> int:
            with self._session() as session:
                record = session.run(cypher, bound).single()
                return int(record["n"]) if record is not None else 0

        total = 0
        while True:
            written = await self._execute_with_reconnect(_write)
            if written == 0:
                return total
            total += written


def _node_to_entity(node: Any) -> Entity:
    """Convert a Memgraph node to an Entity model."""
    labels_raw = node.get("labels", "[]")
    try:
        labels = (
            json.loads(labels_raw) if isinstance(labels_raw, str) else list(labels_raw)
        )
    except (
        json.JSONDecodeError
    ):  # pragma: no cover - defensive fallback for corrupt labels
        labels = []
    props = {k: v for k, v in dict(node).items() if k not in ("id", "labels")}
    return Entity(id=node["id"], labels=labels, properties=props)


def _node_to_chunk(node: Any) -> Chunk:
    """Convert a Memgraph node to a Chunk model."""
    props = dict(node)
    embedding = props.pop("embedding", None)
    content = props.pop("content", "")
    node_id = props.pop("id", "")
    return Chunk(id=node_id, content=content, embedding=embedding, metadata=props)


def _rel_to_relationship(
    rel: Any, source_id: str | None = None, target_id: str | None = None
) -> Relationship:
    """Convert a Memgraph relationship to a Relationship model."""
    props = dict(rel)
    rel_id = props.pop("id", "")
    rel_type = rel.type if hasattr(rel, "type") else props.pop("type", "")
    # Memgraph's Bolt driver returns relationships without populated endpoint
    # nodes in most queries, so callers pass explicit source/target IDs. This
    # fallback is defensive and cannot be exercised with real Memgraph data.
    if source_id is None and hasattr(rel, "start_node"):  # pragma: no cover
        source_id = rel.start_node.get("id", "")
    if target_id is None and hasattr(rel, "end_node"):  # pragma: no cover
        target_id = rel.end_node.get("id", "")
    return Relationship(
        id=rel_id,
        type=rel_type,
        source_id=source_id or "",
        target_id=target_id or "",
        properties=props,
    )


def _records_to_entities_relationships(
    records: Any,
) -> tuple[list[Entity], list[Relationship]]:
    """Assemble deduplicated entities/relationships from expansion records.

    Consumes the ``RETURN DISTINCT start, end, r AS rels`` record shape shared
    by ``get_neighbors`` and ``_matching_neighborhood``.
    """
    entities: dict[str, Entity] = {}
    relationships: dict[str, Relationship] = {}
    for record in records:
        start_node = record["start"]
        entities[start_node["id"]] = _node_to_entity(start_node)
        end_node = record["end"]
        if end_node is None:
            continue
        entities[end_node["id"]] = _node_to_entity(end_node)
        rels = record["rels"] or []
        for rel in rels:
            rel_id = rel.get("id")
            if rel_id and rel_id not in relationships:
                relationships[rel_id] = _rel_to_relationship(
                    rel,
                    source_id=start_node["id"],
                    target_id=end_node["id"],
                )
    return (list(entities.values()), list(relationships.values()))