orchestrata_ultimate.semantic_graph.architecture_twin

Reference › orchestrata_ultimate › semantic_graph › architecture_twin

The semantic twin of the architecture ontology (ruling 4). The architecture decisions and design-document artifacts live in the operational graph (orchkg on orchgraph), written by architecture_writer. This module reads that corpus and builds a twin in the shared semantic graph (memgraph), so decision and design prose is retrievable by GraphRAG, for example "find decisions about queue back-pressure", keyed by the operational node id.

ArchitectureTwin takes a SemanticGraph, an Embedder, an optional synchronous repository for the operational graph, a project (default orchestrata-ultimate) and a repository root. sync reads the live decision and design-doc rows (_read_adr_rows, _read_design_doc_rows), builds chunks, embeds them, merges them into the memgraph chunk label (_embed_and_store) and prunes the twin's own chunks for the project whose operational ids no longer exist (_prune), returning counts. Two pure helpers do the construction and need no database: build_adr_chunk makes the single chunk of an ArchitectureDecision row, and chunk_design_doc splits a design document into overlapping line windows (60 lines with 10 of overlap by default). The embedder is the same one used for code ingestion, so the vector spaces stay consistent. The label comes from the injected SemanticGraphConfig and is never SemanticChunk, which the adapter's pruner treats as stale. The sync_architecture_semantic_twin script imports it.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

ArchitectureTwin

Line 176

Sync the ratified operational-KG architecture corpus into its Memgraph twin.

Functions

Source

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

"""Memgraph semantic twin of the ratified architecture ontology (ruling 4).

``plans/architecture-ontology-draft.md`` ruling 4 (session 242, 2026-07-19)
graduated ``ArchitectureDecision``/``Artifact`` design records into the
operational KG as first-class operational data (see
``knowledge_graph/architecture_writer.py`` and
``scripts/sync_architecture_to_graph.py``). This module builds the
Memgraph *semantic* twin over that same corpus so ADR/design-doc prose is
retrievable via GraphRAG (e.g. "find decisions about queue back-pressure"),
keyed by the operational-KG node id.

The twin chunks each ``ArchitectureDecision`` and each design-doc
``Artifact``, embeds the chunks with the same embedder/model used for code
ingestion (``scripts/ingest_orchestrata_ultimate.py``) so the vector spaces
stay consistent, and MERGEs them into a Memgraph chunk label dedicated to
this twin (never ``SemanticChunk`` -- that label is on
:data:`orchestrata_ultimate.semantic_graph.adapters.memgraph.STALE_CHUNK_LABELS`,
the pruner's kill list). The label is config-driven (``chunk_label`` on the
injected :class:`~orchestrata_ultimate.semantic_graph.config.SemanticGraphConfig`)
so tests can point at an isolated label without touching this module.

Two pure helpers -- :func:`build_adr_chunk` and :func:`chunk_design_doc` --
do the id/content/metadata construction and are unit-testable without any
infrastructure, mirroring the ``CodeChunker``/``SpecTraceEnricher`` split in
``ingestion.py``/``spec_trace_enricher.py``.
"""

from __future__ import annotations

import logging
from collections.abc import Callable, Mapping, Sequence
from pathlib import Path
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.core.paths import resolve_project_root
from orchestrata_ultimate.semantic_graph.embedders import Embedder
from orchestrata_ultimate.semantic_graph.interfaces import SemanticGraph
from orchestrata_ultimate.semantic_graph.models import Chunk

if TYPE_CHECKING:
    from orchestrata_ultimate.core.graph import SyncGraphRepository

logger = logging.getLogger(__name__)

# The two operational-KG labels this twin reads from (canonical_schema.py).
_ARCHITECTURE_DECISION_LABEL = "ArchitectureDecision"
_ARTIFACT_LABEL = "Artifact"
_DESIGN_DOC_ARTIFACT_TYPE = "design-doc"

# Chunk-id prefixes owned by this twin -- pruning never touches anything else,
# even within the twin's own (already dedicated) chunk label.
ADR_CHUNK_ID_PREFIX = "adr:"
DESIGN_DOC_CHUNK_ID_PREFIX = "designdoc:"
CHUNK_ID_PREFIXES: tuple[str, ...] = (ADR_CHUNK_ID_PREFIX, DESIGN_DOC_CHUNK_ID_PREFIX)

_ADR_READ_QUERY = (
    f"MATCH (a:{_ARCHITECTURE_DECISION_LABEL}) WHERE a.project = $project "
    "RETURN a.id AS neo4j_id, a.project AS project, a.title AS title, "
    "a.status AS status, a.context AS context, a.decision AS decision, "
    "a.alternatives AS alternatives, a.consequences AS consequences "
    "ORDER BY a.id"
)

_DESIGN_DOC_READ_QUERY = (
    f"MATCH (a:{_ARTIFACT_LABEL}) "
    "WHERE a.project = $project AND a.artifact_type = $artifact_type "
    "RETURN a.id AS neo4j_id, a.project AS project, a.name AS name, a.path AS path "
    "ORDER BY a.id"
)

# Reader seam: either a full SyncGraphRepository or a plain callable with the
# same (query, params) -> rows shape (ArchitectureWriter._read's signature),
# so pure tests can inject canned rows without touching the graph at all.
ReaderFn = Callable[[str, "Mapping[str, Any] | None"], "list[dict[str, Any]]"]


def build_adr_chunk(row: Mapping[str, Any]) -> Chunk:
    """Build the one semantic chunk for an ``ArchitectureDecision`` row.

    ``row`` is a graph read result matching :data:`_ADR_READ_QUERY`'s RETURN
    clause: ``neo4j_id``/``project``/``title``/``status``/``context``/
    ``decision``/``alternatives``/``consequences``.

    The chunk id is ``f"adr:{neo4j_id}"``; the content is the title followed
    by each non-empty body field under a section header ("Context: ...\\n
    Decision: ..."), so an ADR with e.g. no recorded alternatives still
    produces a coherent chunk instead of a blank section.
    """
    neo4j_id = row["neo4j_id"]
    title = row.get("title") or ""
    project = row.get("project") or ""
    status = row.get("status") or ""

    sections = (
        ("Context", row.get("context") or ""),
        ("Decision", row.get("decision") or ""),
        ("Alternatives", row.get("alternatives") or ""),
        ("Consequences", row.get("consequences") or ""),
    )
    parts: list[str] = [title] if title else []
    for header, text in sections:
        if text:
            parts.append(f"{header}: {text}")
    content = "\n".join(parts)

    return Chunk(
        id=f"{ADR_CHUNK_ID_PREFIX}{neo4j_id}",
        content=content,
        embedding=None,
        metadata={
            "neo4j_id": neo4j_id,
            "source_label": _ARCHITECTURE_DECISION_LABEL,
            "project": project,
            "title": title,
            "status": status,
        },
    )


def chunk_design_doc(
    text: str,
    neo4j_id: str,
    *,
    project: str = "",
    name: str = "",
    window: int = 60,
    overlap: int = 10,
) -> list[Chunk]:
    """Chunk a design-doc's file text into ~``window``-line overlapping windows.

    Mirrors ``ingestion.CodeChunker``'s line-window algorithm. Chunk ids are
    ``f"designdoc:{neo4j_id}:{n:04d}"`` (zero-padded, 0-based). An empty (or
    whitespace-only) ``text`` returns ``[]`` rather than a single blank chunk.

    Raises:
        ValueError: ``window`` < 1, or ``overlap`` is not in ``[0, window)``.
    """
    if window < 1:
        raise ValueError("window must be >= 1")
    if overlap < 0 or overlap >= window:
        raise ValueError("overlap must be >= 0 and < window")

    if not text.strip():
        return []

    lines = text.splitlines()
    step = max(1, window - overlap)
    chunks: list[Chunk] = []
    index = 0

    for start in range(0, len(lines), step):
        end = min(start + window, len(lines))
        content = "\n".join(lines[start:end])
        if content.strip():
            chunks.append(
                Chunk(
                    id=f"{DESIGN_DOC_CHUNK_ID_PREFIX}{neo4j_id}:{index:04d}",
                    content=content,
                    embedding=None,
                    metadata={
                        "neo4j_id": neo4j_id,
                        "source_label": _ARTIFACT_LABEL,
                        "project": project,
                        "name": name,
                        "chunk_index": index,
                    },
                )
            )
            index += 1
        if end == len(lines):
            break

    return chunks


class ArchitectureTwin:
    """Sync the ratified operational-KG architecture corpus into its Memgraph twin.

    Ownership: when ``repo`` and ``reader`` are both None, ``_read`` lazily
    builds and owns a :class:`SyncGraphRepository` via
    ``create_graph_repository(GraphRepositoryConfig.from_env())`` -- the same
    lazy-construction stance as ``ArchitectureWriter`` (resolved at first
    read, not at import time). Injecting ``reader`` (a plain
    ``(query, params) -> rows`` callable) lets pure tests supply canned rows
    without touching the graph at all; injecting ``repo`` lets callers share an
    already-open repository. ``reader`` takes precedence over ``repo`` if
    both are given.
    """

    def __init__(
        self,
        graph: SemanticGraph,
        embedder: Embedder,
        *,
        repo: SyncGraphRepository | None = None,
        reader: ReaderFn | None = None,
        project: str = "orchestrata-ultimate",
        repo_root: Path | None = None,
        embed_batch_size: int = 32,
    ) -> None:
        self._graph = graph
        self._embedder = embedder
        self._repo = repo
        self._reader = reader
        self._project = project
        # resolve_project_root, not a __file__ parent count: installed, this
        # module sits in site-packages, three parents below .venv/lib.
        self._repo_root = repo_root or resolve_project_root()
        self._embed_batch_size = max(1, embed_batch_size)

    def close(self) -> None:
        """Close the twin's repository (mirrors ``ArchitectureWriter.close()``
        -- a no-op for an injected/shared repo, since ``BoltSyncRepository
        .close()`` is itself a no-op for a shared driver; a real close for
        one this twin lazily built itself)."""
        if self._repo is not None:
            self._repo.close()
        self._repo = None

    def _get_repo(self) -> SyncGraphRepository:
        if self._repo is None:
            from orchestrata_ultimate.core.graph import (
                GraphRepositoryConfig,
                create_graph_repository,
            )

            self._repo = create_graph_repository(GraphRepositoryConfig.from_env())
        return self._repo

    def _read(
        self, query: str, params: Mapping[str, Any] | None = None
    ) -> list[dict[str, Any]]:
        if self._reader is not None:
            return self._reader(query, params)
        return self._get_repo().execute_read(query, dict(params or {}))

    def _read_adr_rows(self) -> list[dict[str, Any]]:
        return self._read(_ADR_READ_QUERY, {"project": self._project})

    def _read_design_doc_rows(self) -> list[dict[str, Any]]:
        return self._read(
            _DESIGN_DOC_READ_QUERY,
            {"project": self._project, "artifact_type": _DESIGN_DOC_ARTIFACT_TYPE},
        )

    def _load_design_doc_chunks(
        self, design_doc_rows: Sequence[Mapping[str, Any]]
    ) -> list[Chunk]:
        chunks: list[Chunk] = []
        for row in design_doc_rows:
            neo4j_id = row["neo4j_id"]
            path_str = row.get("path") or ""
            if not path_str:
                logger.warning(
                    "SKIP design doc %s: no path recorded in the graph", neo4j_id
                )
                continue
            full_path = self._repo_root / path_str
            if not full_path.is_file():
                logger.warning(
                    "SKIP design doc %s: file not found at %s", neo4j_id, full_path
                )
                continue
            text = full_path.read_text(encoding="utf-8", errors="replace")
            chunks.extend(
                chunk_design_doc(
                    text,
                    neo4j_id,
                    project=row.get("project") or "",
                    name=row.get("name") or "",
                )
            )
        return chunks

    async def _embed_and_store(self, chunks: list[Chunk]) -> None:
        if not chunks:
            return
        for i in range(0, len(chunks), self._embed_batch_size):
            batch = chunks[i : i + self._embed_batch_size]
            texts = [chunk.content for chunk in batch]
            embeddings = await self._embedder.embed(texts, prompt_type="document")
            if len(embeddings) != len(batch):
                raise RuntimeError(
                    f"Embedder returned {len(embeddings)} vectors for {len(batch)} chunks"
                )
            for chunk, embedding in zip(batch, embeddings):
                chunk.embedding = embedding
            await self._graph.add_chunks(batch)

    async def _prune(self, live_neo4j_ids: set[str]) -> int:
        """Delete this twin's own chunks for THIS project whose ``neo4j_id``
        fell out of the live set just read from the graph.

        Scoped to chunks whose id starts with one of :data:`CHUNK_ID_PREFIXES`
        -- i.e. only chunks this twin itself wrote, within its own configured
        chunk label -- AND whose ``project`` metadata matches
        ``self._project``. The project scope matters: the chunk label/index
        are shared across every project that syncs a twin (multi-project
        invariant), so an unscoped prune of "not in this call's live set"
        would delete every OTHER project's chunks on their very first sync.
        Requires the injected graph to expose ``chunks_with_id_prefix``/
        ``delete_chunks`` (duck-typed, mirroring ``ingestion.py``'s
        ``hasattr(self._graph, "search_keyword")`` idiom); adapters that
        lack them (e.g. a minimal test double) simply prune 0.
        """
        # Duck-typed extras (not on the SemanticGraph Protocol, same stance as
        # ingestion.py's hasattr(self._graph, "search_keyword")): fetched via
        # getattr so a plain-Protocol-typed self._graph still type-checks.
        list_fn = getattr(self._graph, "chunks_with_id_prefix", None)
        delete_fn = getattr(self._graph, "delete_chunks", None)
        if list_fn is None or delete_fn is None:
            return 0

        existing = await list_fn(list(CHUNK_ID_PREFIXES))
        stale_ids = [
            chunk.id
            for chunk in existing
            if chunk.metadata.get("project") == self._project
            and str(chunk.metadata.get("neo4j_id", "")) not in live_neo4j_ids
        ]
        if stale_ids:
            await delete_fn(stale_ids)
        return len(stale_ids)

    async def sync(self) -> dict[str, int]:
        """Read the live ADR/design-doc corpus from the graph, embed, MERGE into
        Memgraph, and prune chunks for records no longer live.

        Returns:
            ``{"adr_chunks": N, "designdoc_chunks": M, "pruned": K}``.
        """
        adr_rows = self._read_adr_rows()
        design_doc_rows = self._read_design_doc_rows()

        adr_chunks = [build_adr_chunk(row) for row in adr_rows]
        design_doc_chunks = self._load_design_doc_chunks(design_doc_rows)

        live_neo4j_ids = {row["neo4j_id"] for row in adr_rows} | {
            row["neo4j_id"] for row in design_doc_rows
        }

        await self._embed_and_store(adr_chunks + design_doc_chunks)
        pruned = await self._prune(live_neo4j_ids)

        summary = {
            "adr_chunks": len(adr_chunks),
            "designdoc_chunks": len(design_doc_chunks),
            "pruned": pruned,
        }
        logger.info(
            "Architecture semantic twin synced for %s: %s", self._project, summary
        )
        return summary