scripts.sync_traces_to_graph

Reference › scripts › sync_traces_to_graph

scripts/sync_traces_to_graph.py syncs @traces decorator metadata into the knowledge graph as TRACES_TO edges from code nodes (classes, functions, methods) to Requirement nodes. A human or a graph-sync step runs it. It uses the scanner and traceability helpers in orchestrata_ultimate.core.tracing.scanner and orchestrata_ultimate.core.tracing.traceability, and the connection in orchestrata_ultimate.core.graph_factory. See also trace_intel and generate_rtm_evidence.

run has three steps. harvest_trace_edges scans the source tree for @traces decorators and returns deduplicated pairs of qualified name and requirement id (_module_dotted_path derives the dotted path from the file path). _match_existing_symbols and _match_existing_requirements keep only the pairs whose code node and canonical requirement node already exist in the graph. _write_edges then MERGEs the edges, so a re-run is idempotent. format_report renders the counts. Options: --source (default src/orchestrata_ultimate), --project (scopes the code-node match), --dry-run (harvest and print counts, write nothing), and --uri, --user and --password connection overrides. An exception is printed as ERROR: and exits 1.

At a glance

Class diagram

Dependency diagram

API

Classes

TraceEdge (dataclass)

Decorators: dataclass(frozen=True, order=True) · Line 86

One (decorated symbol, requirement id) pair harvested from source.

SyncReport (dataclass)

Decorators: dataclass · Line 94

Summary of one harvest+map(+write) run.

Functions

Source

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

#!/usr/bin/env python3
"""Sync ``@traces`` decorator metadata into the knowledge graph.

Thousands of ``@traces(spec=..., story=..., reqs=[...])`` decorators live on
classes/functions/methods across ``src/`` (see
``orchestrata_ultimate.core.base.traces``), but today only a handful of
``Class``/``Function`` nodes in the live graph carry a ``TRACES_TO`` edge --
the SPEC-106 selector's ``trace_degree`` ranking signal and the BUILD context
template are both starved of this signal. This script closes that gap.

Pipeline
--------
1. HARVEST -- ``orchestrata_ultimate.core.tracing.scanner.collect_traces``
   walks ``--source`` and returns one ``TraceInfo`` per (decorated symbol,
   requirement id) pair. This script does *not* re-implement that AST walk;
   it only re-derives the symbol's dotted module path (the scanner's own
   ``TraceInfo.module`` is relative to the source root's *parent*, which
   would carry a leading ``orchestrata_ultimate.`` segment the live graph's
   ``qualified_name`` convention does not use -- see ``_module_dotted_path``).
2. MAP -- each harvested symbol is matched against the canonical KG by its
   ``qualified_name`` property (never by parsing node ``id`` strings -- the
   ``{project}:{qualified_name}`` id format has changed before and the
   8b072d7d regression is exactly this mistake), scoped by ``project``, and
   may live under the ``Class``, ``Function``, or ``Method`` label.
3. WRITE -- ``MERGE (code)-[:TRACES_TO]->(ri:RequirementItem {id:
   "{project}:{req_id}"})``. Per the canonical ontology
   (``knowledge_graph/canonical_schema.py``: TRACES_TO always targets
   RequirementItem, never Requirement directly; spec-tier ids are
   project-qualified since the 2026-07-12 multi-project ruling).
   RequirementItem nodes are matched, never created (ontology-before-import
   doctrine: the spec tier is owned by the kg_sync ``sync_specs`` stage, not
   this writer) -- a referenced requirement id with no matching node is
   reported as "unresolved", not fabricated.

Every write is a MERGE, so re-running this script is safe and idempotent.

Connection defaults to the ``NEO4J_URI``, ``NEO4J_USERNAME``, and
``NEO4J_PASSWORD`` environment variables; ``--uri``/``--user``/``--password``
override them.

Examples
--------
    python scripts/sync_traces_to_graph.py --dry-run

    python scripts/sync_traces_to_graph.py \\
        --source src/orchestrata_ultimate --project orchestrata-ultimate

    python scripts/sync_traces_to_graph.py \\
        --uri neo4j://10.53.1.11:7687 --user neo4j --password changeme
"""

from __future__ import annotations

import argparse
import os
import sys
from collections.abc import Mapping
from dataclasses import dataclass, field
from pathlib import Path

from neo4j import Driver, GraphDatabase

from orchestrata_ultimate.core.graph_factory import resolve_kg_database
from orchestrata_ultimate.core.tracing.scanner import collect_traces
from orchestrata_ultimate.core.tracing.traceability import TraceInfo

_BATCH_SIZE = 500
_DISPLAY_CAP = 20

# Per-label scoped subquery (the house pattern from SearchEngine's
# _contains_search): a label-less `(c) WHERE c:Class OR ...` predicate cannot
# use indexes and degrades to a node-store scan PER UNWIND ROW. Each branch
# here is served by the {label}_qualified_name RANGE indexes (schema.py).
_CODE_NODE_SUBQUERY = (
    "CALL (qn) { "
    "MATCH (c:Class {qualified_name: qn}) WHERE c.project = $project RETURN c "
    "UNION MATCH (c:Function {qualified_name: qn}) "
    "  WHERE c.project = $project RETURN c "
    "UNION MATCH (c:Method {qualified_name: qn}) "
    "  WHERE c.project = $project RETURN c "
    "} "
)


@dataclass(frozen=True, order=True)
class TraceEdge:
    """One (decorated symbol, requirement id) pair harvested from source."""

    qualified_name: str
    req_id: str


@dataclass
class SyncReport:
    """Summary of one harvest+map(+write) run."""

    trace_hits: int
    decorated_symbols: int
    distinct_req_ids: int
    edges_merged: int
    unresolved_reqs: list[str] = field(default_factory=list)
    unmatched_symbols: list[str] = field(default_factory=list)
    dry_run: bool = False


def _module_dotted_path(file_path: str, source_root: Path) -> str:
    """Dotted module path for *file_path*, relative to *source_root*.

    Matches the canonical KG's ``qualified_name`` convention -- a dotted
    path relative to the package root (e.g. ``knowledge_graph.search`` for
    ``src/orchestrata_ultimate/knowledge_graph/search.py`` when
    ``source_root`` is ``src/orchestrata_ultimate``). This is deliberately
    NOT ``TraceInfo.module`` (the scanner computes that relative to the
    source root's *parent*, so it carries a leading
    ``orchestrata_ultimate.`` segment the live graph's nodes do not have).
    """
    rel = Path(file_path).resolve().relative_to(source_root.resolve())
    parts = list(rel.with_suffix("").parts)
    if parts and parts[-1] == "__init__":
        parts = parts[:-1]
    return ".".join(parts)


def harvest_trace_edges(source: str) -> list[TraceEdge]:
    """Scan *source* for ``@traces`` decorators and return the deduplicated,
    sorted set of (qualified_name, req_id) edges they declare."""
    source_root = Path(source)

    def _qualified_name(info: TraceInfo) -> str:
        module = _module_dotted_path(info.file_path, source_root)
        segments = [module] + [p for p in (info.class_name, info.method) if p]
        return ".".join(s for s in segments if s)

    edges: set[TraceEdge] = {
        TraceEdge(qualified_name=_qualified_name(info), req_id=req_id)
        for info in collect_traces(source)
        for req_id in info.reqs
    }
    return sorted(edges)


def _resolve_neo4j_connection(
    args: argparse.Namespace, env: Mapping[str, str] | None = None
) -> tuple[str, str, str]:
    """Resolve graph connection info -- CLI flags override env vars.

    *env* defaults to ``os.environ`` and exists purely as a test injection
    seam (the project's no-monkeypatch policy); production callers never
    pass it.
    """
    if env is None:
        env = os.environ
    uri = args.uri or (env.get("KG_ROUTING_URI") or env.get("KG_URI"))
    user = args.user or ""
    password = args.password or ""
    # Only the ADDRESS is required. This also demanded user AND password;
    # both are now always empty (every remaining graph endpoint negotiates no
    # auth, and NEO4J_USERNAME/NEO4J_PASSWORD were retired with the Neo4j
    # containers on 2026-07-27), so the old check raised on every invocation.
    if not uri:
        raise ValueError(
            "graph connection info missing: set KG_ROUTING_URI or KG_URI, "
            "or pass --uri."
        )
    return uri, user, password


def _match_existing_symbols(
    driver: Driver, project: str, qualified_names: list[str]
) -> set[str]:
    """Which of *qualified_names* have a Class/Function/Method node with
    that ``qualified_name`` in *project*."""
    existing: set[str] = set()
    for i in range(0, len(qualified_names), _BATCH_SIZE):
        batch = qualified_names[i : i + _BATCH_SIZE]
        with driver.session(database=resolve_kg_database()) as session:
            result = session.run(
                "UNWIND $names AS qn "
                f"{_CODE_NODE_SUBQUERY}"
                "RETURN DISTINCT qn AS qn",
                names=batch,
                project=project,
            )
            existing.update(record["qn"] for record in result)
    return existing


def _match_existing_requirements(
    driver: Driver, project: str, req_ids: list[str]
) -> set[str]:
    """Which BARE *req_ids* have a matching canonical
    ``RequirementItem {id: "{project}:{req_id}"}`` node."""
    existing: set[str] = set()
    for i in range(0, len(req_ids), _BATCH_SIZE):
        batch = req_ids[i : i + _BATCH_SIZE]
        with driver.session(database=resolve_kg_database()) as session:
            result = session.run(
                "UNWIND $ids AS rid "
                "MATCH (ri:RequirementItem {id: $project + ':' + rid}) "
                "RETURN DISTINCT rid AS rid",
                ids=batch,
                project=project,
            )
            existing.update(record["rid"] for record in result)
    return existing


def _write_edges(driver: Driver, project: str, rows: list[TraceEdge]) -> int:
    """MERGE ``TRACES_TO`` edges for already-resolved (qualified_name,
    req_id) pairs. Idempotent -- MERGE only, never creates code or
    Requirement nodes."""
    for i in range(0, len(rows), _BATCH_SIZE):
        batch = rows[i : i + _BATCH_SIZE]
        with driver.session(database=resolve_kg_database()) as session:
            session.run(
                "UNWIND $rows AS row "
                "WITH row, row.qname AS qn "
                f"{_CODE_NODE_SUBQUERY}"
                "WITH row, c "
                "MATCH (ri:RequirementItem {id: $project + ':' + row.req_id}) "
                "MERGE (c)-[:TRACES_TO]->(ri)",
                rows=[{"qname": e.qualified_name, "req_id": e.req_id} for e in batch],
                project=project,
            ).consume()
    return len(rows)


def run(args: argparse.Namespace) -> SyncReport:
    """Harvest, map, and (unless ``--dry-run``) write TRACES_TO edges."""
    edges = harvest_trace_edges(args.source)
    decorated_symbols = sorted({e.qualified_name for e in edges})
    distinct_req_ids = sorted({e.req_id for e in edges})

    uri, user, password = _resolve_neo4j_connection(args)
    driver = GraphDatabase.driver(uri, auth=(user, password))
    try:
        existing_symbols = _match_existing_symbols(
            driver, args.project, decorated_symbols
        )
        existing_reqs = _match_existing_requirements(
            driver, args.project, distinct_req_ids
        )

        resolved_rows = [
            e
            for e in edges
            if e.qualified_name in existing_symbols and e.req_id in existing_reqs
        ]

        if args.dry_run:
            edges_merged = len(resolved_rows)
        else:
            edges_merged = _write_edges(driver, args.project, resolved_rows)
    finally:
        driver.close()

    return SyncReport(
        trace_hits=len(edges),
        decorated_symbols=len(decorated_symbols),
        distinct_req_ids=len(distinct_req_ids),
        edges_merged=edges_merged,
        unresolved_reqs=sorted(set(distinct_req_ids) - existing_reqs),
        unmatched_symbols=sorted(set(decorated_symbols) - existing_symbols),
        dry_run=args.dry_run,
    )


def format_report(report: SyncReport) -> str:
    """Render a ``SyncReport`` as the human-readable CLI summary."""
    lines = [
        f"Trace hits scanned: {report.trace_hits} "
        f"({report.decorated_symbols} distinct decorated symbol(s), "
        f"{report.distinct_req_ids} distinct requirement id(s))",
    ]
    if report.dry_run:
        lines.append(f"DRY RUN -- edges that would be merged: {report.edges_merged}")
    else:
        lines.append(f"Edges merged: {report.edges_merged}")

    lines.append(
        "Unresolved requirements (referenced by @traces, no canonical "
        f"RequirementItem node -- NOT created): {len(report.unresolved_reqs)}"
    )
    for rid in report.unresolved_reqs[:_DISPLAY_CAP]:
        lines.append(f"  - {rid}")
    if len(report.unresolved_reqs) > _DISPLAY_CAP:
        lines.append(f"  ... and {len(report.unresolved_reqs) - _DISPLAY_CAP} more")

    lines.append(
        "Unmatched symbols (decorated, no canonical Class/Function/Method "
        f"node): {len(report.unmatched_symbols)}"
    )
    for qn in report.unmatched_symbols[:_DISPLAY_CAP]:
        lines.append(f"  - {qn}")
    if len(report.unmatched_symbols) > _DISPLAY_CAP:
        lines.append(f"  ... and {len(report.unmatched_symbols) - _DISPLAY_CAP} more")

    return "\n".join(lines)


def _build_parser() -> argparse.ArgumentParser:
    """Build the argument parser."""
    parser = argparse.ArgumentParser(
        prog="sync_traces_to_graph",
        description=(
            "Sync @traces decorator metadata into the graph as TRACES_TO edges "
            "from code nodes to Requirement nodes."
        ),
    )
    parser.add_argument(
        "--source",
        default="src/orchestrata_ultimate",
        help="Source tree to scan for @traces decorators (default: %(default)s).",
    )
    parser.add_argument(
        "--project",
        default="orchestrata-ultimate",
        help="Project name to scope code-node matching (default: %(default)s).",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        default=False,
        help="Harvest and print counts without writing any edges.",
    )
    parser.add_argument(
        "--uri",
        default=None,
        help="Neo4j Bolt URI (default: NEO4J_URI env var).",
    )
    parser.add_argument(
        "--user",
        default=None,
        help="Neo4j username (default: NEO4J_USERNAME env var).",
    )
    parser.add_argument(
        "--password",
        default=None,
        help="Neo4j password (default: NEO4J_PASSWORD env var).",
    )
    return parser


def main(argv: list[str] | None = None) -> None:
    """Entry point."""
    parser = _build_parser()
    args = parser.parse_args(argv)

    try:
        report = run(args)
    except Exception as exc:  # noqa: BLE001
        print(f"ERROR: {exc}", file=sys.stderr)
        sys.exit(1)

    print(format_report(report))
    sys.exit(0)


if __name__ == "__main__":
    main()