scripts.sync_architecture_semantic_twin

Reference › scripts › sync_architecture_semantic_twin

scripts/sync_architecture_semantic_twin.py syncs the Memgraph semantic twin of the ratified architecture ontology corpus: the ArchitectureDecision nodes and design-document Artifact nodes, which it reads from the operational graph, are embedded and stored as chunks so that ADR and design prose can be retrieved with GraphRAG (see semantic_search). It is the semantic counterpart of sync_architecture_to_graph. A human runs it. It uses orchestrata_ultimate.semantic_graph.architecture_twin, the Memgraph adapter, orchestrata_ultimate.core.graph and orchestrata_ultimate.core.env_bootstrap.

Options: --project (the project slug, default PROJECT) and --dry-run, which prints the chunk label, vector index name, embedding model and dimension without connecting to either graph or building an embedder. _twin_graph_config builds the twin's Memgraph configuration, and _sync connects, syncs, prints a summary and always closes both connections. A real run needs OPENROUTER_API_KEY in the environment and exits 1 with a message if it is not set.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/sync_architecture_semantic_twin.py

"""Sync the Memgraph semantic twin of the ratified architecture ontology corpus.

Thin CLI over ``semantic_graph.architecture_twin.ArchitectureTwin`` (wave
ontology-phase2, ruling 4 of ``plans/architecture-ontology-draft.md``):
reads the ``ArchitectureDecision`` + design-doc ``Artifact`` nodes Phase 2a
(``scripts/sync_architecture_to_graph.py --adrs``) wrote into the graph, chunks
their prose, embeds each chunk with the same embedder/model used for code
ingestion (``scripts/ingest_orchestrata_ultimate.py``), and MERGEs them into
Memgraph under a dedicated chunk label (``ArchitectureChunk`` -- deliberately
NOT ``SemanticChunk``, which is on the pruner's stale-label kill list; see
``adapters/memgraph.py::STALE_CHUNK_LABELS``) so ADR/design-doc prose is
retrievable via GraphRAG (e.g. "find decisions about queue back-pressure").

Idempotent -- ``add_chunks`` MERGEs by id, and every sync prunes twin-owned
chunks whose source record fell out of the live graph set.

Mirrors ``sync_architecture_to_graph.py``'s CLI shape: manual .env loading,
a ``--dry-run`` that constructs NOTHING infra-side (no graph/Memgraph
connection, no embedder), and the same "missing OPENROUTER_API_KEY is a
loud exit 1" behavior as ``ingest_orchestrata_ultimate.py``.
"""

from __future__ import annotations

import argparse
import asyncio
import dataclasses
import logging
import os
import sys
from pathlib import Path

# Ensure repo src is on path when run directly (mirrors sync_architecture_to_graph.py).
_repo_src = Path(__file__).resolve().parent.parent / "src"
if str(_repo_src) not in sys.path:
    sys.path.insert(0, str(_repo_src))

from orchestrata_ultimate.core.env_bootstrap import load_env_contract  # noqa: E402
from orchestrata_ultimate.core.graph import (  # noqa: E402
    GraphRepositoryConfig,
    SyncGraphRepository,
    create_graph_repository,
)
from orchestrata_ultimate.semantic_graph.adapters.memgraph import (  # noqa: E402
    MemgraphSemanticGraph,
)
from orchestrata_ultimate.semantic_graph.architecture_twin import (  # noqa: E402
    ArchitectureTwin,
)
from orchestrata_ultimate.semantic_graph.config import (  # noqa: E402
    SemanticGraphConfig,
)
from orchestrata_ultimate.semantic_graph.embedders import (  # noqa: E402
    OpenRouterEmbedder,
)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)-8s %(name)s: %(message)s",
)
logger = logging.getLogger(__name__)

PROJECT_ROOT = Path(__file__).resolve().parent.parent
PROJECT = "orchestrata-ultimate"

# Same embedder/model as the code ingest (scripts/ingest_orchestrata_ultimate.py)
# so the ADR/design-doc vector space is consistent with the code vector space.
EMBED_MODEL = "mistralai/codestral-embed-2505"
EMBED_DIMENSION = 1536

# The twin's own dedicated Memgraph label/index -- ratified design, never
# SemanticChunk (the pruner's stale-label kill list) or CodestralChunk (the
# code-ingest's label).
CHUNK_LABEL = "ArchitectureChunk"
VECTOR_INDEX_NAME = "architecture_chunk_embedding_idx"


def _load_dotenv() -> None:
    """Load the RUL-019 two-phase environment contract.

    Delegates to ``core.env_bootstrap.load_env_contract`` — THE canonical
    loader — instead of parsing ``.env`` here. The hand-rolled copy this
    replaced (byte-identical in four scripts) stopped after phase 1, and
    ``.env`` is the prod-BASE layer: it carries ``CXP_GRPC_HOST=localhost``
    for the host user-units, so a script that loads it and stops resolves
    PRODUCTION endpoints while believing it is on the test fleet. The
    canonical loader then applies the ``ORCHESTRATA_ENV_PROFILE`` overlay,
    which is the half that points the process at the right infrastructure,
    and honours ``declare_owned`` so it cannot reset a test worker's
    per-slot isolation.
    """
    load_env_contract()


def _twin_graph_config() -> SemanticGraphConfig:
    """The twin's Memgraph config: base env config, chunk label/index
    overridden to the twin's own dedicated values."""
    base = SemanticGraphConfig.from_env()
    return dataclasses.replace(
        base, chunk_label=CHUNK_LABEL, vector_index_name=VECTOR_INDEX_NAME
    )


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description=(
            "Sync the Memgraph semantic twin of the ratified architecture "
            "ontology corpus (ArchitectureDecision + design-doc Artifact "
            "nodes, read from the graph) so ADR/design-doc prose is retrievable "
            "via GraphRAG."
        ),
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Print what would sync without connecting to either graph "
        "and without constructing an embedder.",
    )
    parser.add_argument(
        "--project",
        default=PROJECT,
        help=f"Project slug to sync (default: {PROJECT!r}).",
    )
    return parser


async def _sync(
    graph: MemgraphSemanticGraph,
    embedder: OpenRouterEmbedder,
    repo: SyncGraphRepository,
    project: str,
    connect_retries: int = 2,
    retry_delay: float = 3.0,
) -> int:
    """Connect, sync, print/log a summary, always close both connections.

    Split out from ``_run`` (mirrors ``ingest_orchestrata_ultimate.py``'s
    ``_ingest(graph, embedder)``) so a caller -- production ``_run`` or a
    test -- can inject an already-configured ``graph``/``repo`` instead of
    this module always reaching for the real production chunk label/Neo4j.

    ``connect_retries``/``retry_delay``: the session-start hook can fire
    inside a Memgraph HA failover window (observed 2026-07-21: MAIN moved
    hosts and the router had no READ server for a moment -- the bolt PORT
    answered the hook's probe, but routing wasn't settled). A couple of
    short retries lets that boot sync self-heal instead of skipping until
    the next session.
    """
    try:
        connected = await graph.connect()
        for attempt in range(connect_retries):
            if connected:
                break
            logger.warning(
                "Memgraph connect failed (attempt %d/%d) -- likely an HA "
                "failover window; retrying in %.0fs",
                attempt + 1,
                connect_retries + 1,
                retry_delay,
            )
            await asyncio.sleep(retry_delay)
            connected = await graph.connect()
        if not connected:
            logger.error(
                "Failed to connect to Memgraph after %d attempts",
                connect_retries + 1,
            )
            return 1

        twin = ArchitectureTwin(graph, embedder, repo=repo, project=project)
        summary = await twin.sync()
        logger.info("Architecture semantic twin synced: %s", summary)
        print(f"Architecture semantic twin synced: {summary}")
        return 0
    finally:
        repo.close()
        await graph.close()


async def _run(project: str, api_key: str) -> int:
    embedder = OpenRouterEmbedder(
        api_key=api_key, model=EMBED_MODEL, dimension=EMBED_DIMENSION
    )
    graph = MemgraphSemanticGraph(_twin_graph_config())
    repo = create_graph_repository(GraphRepositoryConfig.from_env())
    return await _sync(graph, embedder, repo, project)


def main(argv: list[str] | None = None) -> int:
    args = _build_parser().parse_args(argv)
    _load_dotenv()

    if args.dry_run:
        print(
            f"Would sync Architecture semantic twin for project {args.project!r}: "
            f"ArchitectureDecision + design-doc Artifact nodes from the graph -> "
            f"Memgraph label {CHUNK_LABEL!r} (vector index {VECTOR_INDEX_NAME!r}), "
            f"embedded with {EMBED_MODEL!r} ({EMBED_DIMENSION} dims)."
        )
        return 0

    api_key = os.environ.get("OPENROUTER_API_KEY", "")
    if not api_key:
        logger.error("OPENROUTER_API_KEY is not set")
        print("OPENROUTER_API_KEY is not set", file=sys.stderr)
        return 1

    return asyncio.run(_run(args.project, api_key))


if __name__ == "__main__":
    raise SystemExit(main())