scripts.enrich_semantic_graph_specs

Reference › scripts › enrich_semantic_graph_specs

Enriches the orchestrata-ultimate semantic graph with the specs and the code traces. A human runs it; no test imports it. It builds its configuration from config and loads the environment through env_bootstrap. The graph itself is described in the semantic_graph narrative.

Invocation: main is asynchronous and takes --with-tests, which also covers the tests tree. It returns 1 if the specs directory or the source directory is missing, if the semantic graph backend cannot be reached or if enrichment fails. main creates the graph from SemanticGraphConfig.from_env, connects and calls enrich_project_with_specs_and_traces over the specs, source and (optionally) tests roots, prints the summary as a dictionary and closes the connection in a finally block.

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/enrich_semantic_graph_specs.py

"""Enrich the orchestrata-ultimate semantic graph with specs and code traces.

Parses the ``specs/`` tree and ``@traces`` annotations from ``src/``, then writes
Spec, Story, Requirement, and CodeElement nodes plus their relationships into the
Memgraph semantic graph. This lets GraphRAG answer requirement-aware questions
such as "what implements REQ-004-US-001-001?".

With ``--with-tests`` (the default), it also scans ``tests/`` for @traces'd test
functions and heuristically-mapped test modules, writing TestCase/TestModule
nodes and TESTS edges (audit item 5) -- extending the Spec->Story->Requirement
->CodeElement chain all the way to what actually exercises the code.
"""

from __future__ import annotations

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

from orchestrata_ultimate.core.env_bootstrap import load_env_contract
from orchestrata_ultimate.semantic_graph.config import SemanticGraphConfig
from orchestrata_ultimate.semantic_graph.repository import create_semantic_graph
from orchestrata_ultimate.semantic_graph.spec_trace_enricher import (
    enrich_project_with_specs_and_traces,
)

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
SPECS_DIR = PROJECT_ROOT / "specs"
SOURCE_DIR = PROJECT_ROOT / "src" / "orchestrata_ultimate"
TESTS_DIR = PROJECT_ROOT / "tests"
PROJECT_ID = os.environ.get("PROJECT_ID", "orchestrata-ultimate")


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

    Delegates to ``core.env_bootstrap.load_env_contract`` — THE canonical
    loader. What stood here loaded ``.env`` and stopped, and ``.env`` is the
    prod-BASE layer: it carries ``CXP_GRPC_HOST=localhost`` for the host
    user-units, so a script that stops after phase 1 resolves PRODUCTION
    endpoints while believing it is on the test fleet.

    ``override=False`` was the right instinct aimed at the wrong half. It was
    added on 2026-07-26 after a bootstrap intended for the test node wrote 592
    chunks into the production cluster, because ``.env``'s value had replaced
    an explicitly-exported one. The canonical loader keeps that property for
    phase 1 (``.env`` only fills keys the environment has not already set) and
    then applies the ``ORCHESTRATA_ENV_PROFILE`` overlay, which is the half
    that actually 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 _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--with-tests",
        dest="with_tests",
        action="store_true",
        default=True,
        help="Scan tests/ and emit TestCase/TestModule TESTS edges (default: on).",
    )
    parser.add_argument(
        "--no-with-tests",
        dest="with_tests",
        action="store_false",
        help="Skip tests/ scanning -- only Spec/Story/Requirement/CodeElement.",
    )
    return parser.parse_args(argv)


async def main(argv: list[str] | None = None) -> int:
    """Parse specs/traces and write them to the semantic graph."""
    args = _parse_args(argv)
    _load_dotenv()

    if not SPECS_DIR.is_dir():
        logger.error("Specs directory not found: %s", SPECS_DIR)
        return 1
    if not SOURCE_DIR.is_dir():
        logger.error("Source directory not found: %s", SOURCE_DIR)
        return 1

    config = SemanticGraphConfig.from_env()
    graph = create_semantic_graph(config)

    connected = await graph.connect()
    if not connected:
        logger.error("Failed to connect to semantic graph backend")
        return 1

    try:
        summary = await enrich_project_with_specs_and_traces(
            graph=graph,
            specs_root=SPECS_DIR,
            source_root=SOURCE_DIR,
            project_id=PROJECT_ID,
            tests_root=TESTS_DIR,
            with_tests=args.with_tests,
        )
        logger.info("Enrichment complete: %s", summary.to_dict())
        print(summary.to_dict())
        return 0
    except Exception:
        logger.exception("Enrichment failed")
        return 1
    finally:
        await graph.close()


if __name__ == "__main__":
    sys.exit(asyncio.run(main(sys.argv[1:])))