scripts.derive_rtm

Reference › scripts › derive_rtm

Derives requirements-traceability-matrix attribution over the merged static and runtime call graph. Starting from the code that declares @traces for a requirement, it follows calls to attribute the requirement to the code reachable from it. tests/scripts/test_derive_rtm.py imports it; it uses trace_intel and writes JSON through atomic_json. See also build_runtime_callgraph, which supplies the runtime half of the graph.

Invocation: main has one subcommand, build, which writes a JSON result and a Markdown report (_cmd_build, build); the paths of the structural and traceability fact files can be overridden. With no subcommand it prints help and exits 1.

The flow: parse_call_graph reads the structural facts into an adjacency map and a map from function to module. parse_traceability reads the traceability facts into the functions that declare each requirement and the specs that define each requirement (_spec_name_map resolves spec names). compute_anchors combines these into the anchor functions for each requirement. compute_attribution runs a breadth-first search (_bfs) outward from the anchors over the call graph and records which requirements reach which functions and modules. write_report renders the outcome.

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

#!/usr/bin/env python3
"""scripts/derive_rtm.py — derived RTM attribution over the MERGED static +
runtime call graph (reachability-attribution follow-on to the SPEC-023
runtime call-graph capture).

The traceability facts under ``.cache/trace_intel/`` (owned by
``scripts/trace_intel.py``) now carry both static edges
(``calls_resolved/3``, from AST call-site resolution) and runtime edges
(``runtime_call/2``, from the ``sys.monitoring`` capture merged in by
``scripts/build_runtime_callgraph.py`` + ``scripts/code_intel.py reindex``).
This script walks the UNION of those two edge sets to propose ``@traces``
attributions the static graph alone would miss (DI/dispatch/registry calls
routed through ``self.x()``, a factory, or a lookup table).

Method
------
1. Parse ``calls_resolved/3`` ∪ ``runtime_call/2`` from
   ``codebase_facts.pl`` into one merged adjacency ``caller -> {callees}``,
   and ``symbol/7`` (class/function/method rows only) into a
   ``qname -> module`` map.
2. Parse ``traces('Q', 'REQ')`` and ``spec_requirement('SPEC', 'REQ', ...)``
   from ``traceability_facts.pl``.
3. A spec's ANCHORS are the functions already ``@traces``-linked to one of
   its requirements. A spec's CLOSURE is everything merged-reachable
   (forward BFS over the merged adjacency) from its anchors.
4. ATTRIBUTED(spec) = untraced functions in the spec's closure that are
   reachable from EXACTLY ONE spec's closure — the discriminating filter
   that separates spec-specific code from shared infrastructure code
   incidentally reachable from many specs.
5. ALIGNED(spec) ⊆ ATTRIBUTED(spec) = the subset that ALSO lives in a
   module the spec already owns an anchor in — the high-confidence tier,
   dropping shared utilities that land in one closure by chance rather
   than by ownership.

The per-spec "reached by exactly one spec's closure" discriminator needs
every spec's closure materialized at once, which is why this lives in
Python rather than the corpus-global Prolog rules in ``traceability_db.pl``
(see that file's Layer-3 comment) — this script is the reference-numbers
producer those rules point to.

Usage::

    uv run python scripts/derive_rtm.py build

Writes ``plans/derived-rtm.json`` (deterministic, sorted — no timestamps)
and ``reports/derived-rtm.md`` (human digest), and prints a compact JSON
summary to stdout. Exits 1 with a clear stderr message if the structural
facts predate the runtime-call-graph merge (no ``runtime_call(`` fact at
all) rather than silently falling back to a static-only result.
"""

from __future__ import annotations

import argparse
import json
import re
import sys
from collections import defaultdict, deque
from pathlib import Path
from typing import Any

# Repo root on sys.path so `scripts` (imported below) resolves regardless of
# how this file is invoked (direct `python scripts/derive_rtm.py`, `-m`, or
# as a test import) — same pattern as scripts/check_wiring.py.
_REPO_ROOT = Path(__file__).resolve().parent.parent
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))

from scripts import trace_intel as t  # noqa: E402

DEFAULT_OUT_JSON = _REPO_ROOT / "plans" / "derived-rtm.json"
DEFAULT_REPORT = _REPO_ROOT / "reports" / "derived-rtm.md"

_CALL_EDGE_RE = re.compile(r"(?:calls_resolved|runtime_call)\('([^']+)',\s*'([^']+)'")
_SYMBOL_RE = re.compile(
    r"symbol\('([^']+)',\s*'([^']+)',\s*(?:class|function|method)\b"
)
_TRACES_RE = re.compile(r"traces\('([^']+)',\s*'([^']+)'\)")
_SPEC_REQ_RE = re.compile(r"spec_requirement\('([^']+)',\s*'([^']+)'")
_SPEC_DIR_RE = re.compile(r"^(\d{3})-(.+)$")

_MISSING_RUNTIME_CALL_MSG = (
    "ERROR: no runtime_call/2 facts found in {facts} -- the structural KB "
    "predates the runtime call-graph merge, and attribution over a "
    "static-only graph would silently under-count DI/dispatch edges. Run "
    "`python scripts/code_intel.py reindex` then "
    "`python scripts/trace_intel.py extract` to regenerate the merged "
    "facts, then re-run this command."
)


# ---------------------------------------------------------------------------
# Fact parsing
# ---------------------------------------------------------------------------


def parse_call_graph(
    structural_facts_text: str,
) -> tuple[dict[str, set[str]], dict[str, str]]:
    """Parse ``calls_resolved/3`` ∪ ``runtime_call/2`` into a merged
    adjacency (``caller -> {callees}``) and ``symbol/7`` class/function/
    method rows into a ``qname -> module`` map."""
    adj: dict[str, set[str]] = defaultdict(set)
    for m in _CALL_EDGE_RE.finditer(structural_facts_text):
        adj[m.group(1)].add(m.group(2))
    module_of: dict[str, str] = {}
    for m in _SYMBOL_RE.finditer(structural_facts_text):
        module_of[m.group(1)] = m.group(2)
    return adj, module_of


def parse_traceability(
    traceability_text: str,
) -> tuple[dict[str, set[str]], dict[str, set[str]]]:
    """Parse ``traces('Q', 'REQ')`` (``qname -> {req}``) and
    ``spec_requirement('SPEC', 'REQ', ...)`` (``req -> {spec}``)."""
    traces: dict[str, set[str]] = defaultdict(set)
    for m in _TRACES_RE.finditer(traceability_text):
        traces[m.group(1)].add(m.group(2))
    req_specs: dict[str, set[str]] = defaultdict(set)
    for m in _SPEC_REQ_RE.finditer(traceability_text):
        req_specs[m.group(2)].add(m.group(1))
    return traces, req_specs


def _spec_name_map(specs_root: Path) -> dict[str, str]:
    """``specs/<id>-<name>/`` directory names -> ``{id: name}``."""
    mapping: dict[str, str] = {}
    if not specs_root.is_dir():
        return mapping
    for entry in sorted(specs_root.iterdir()):
        if not entry.is_dir():
            continue
        m = _SPEC_DIR_RE.match(entry.name)
        if m:
            mapping[m.group(1)] = m.group(2)
    return mapping


# ---------------------------------------------------------------------------
# Attribution
# ---------------------------------------------------------------------------


def compute_anchors(
    traces: dict[str, set[str]], req_specs: dict[str, set[str]]
) -> dict[str, set[str]]:
    """A spec's anchors: every ``@traces``-linked qname whose requirement
    belongs to that spec."""
    anchors: dict[str, set[str]] = defaultdict(set)
    for qname, reqs in traces.items():
        for req in reqs:
            for spec in req_specs.get(req, ()):
                anchors[spec].add(qname)
    return anchors


def _bfs(seeds: set[str], adj: dict[str, set[str]]) -> set[str]:
    """Forward BFS over the merged adjacency, seeded from ``seeds``."""
    seen: set[str] = set()
    queue: deque[str] = deque(seeds)
    while queue:
        node = queue.popleft()
        if node in seen:
            continue
        seen.add(node)
        for nxt in adj.get(node, ()):
            if nxt not in seen:
                queue.append(nxt)
    return seen


def compute_attribution(
    adj: dict[str, set[str]],
    module_of: dict[str, str],
    anchors: dict[str, set[str]],
    traced: set[str],
) -> tuple[dict[str, set[str]], dict[str, list[str]], dict[str, list[str]]]:
    """Returns ``(closure, attributed, aligned)`` per the module docstring's
    method section — ``closure`` is the full merged-reachable set per spec
    (traced and untraced), ``attributed``/``aligned`` are keyed by spec and
    hold only the untraced qnames each tier proposes."""
    closure = {spec: _bfs(anch, adj) for spec, anch in anchors.items()}
    spec_mods = {
        spec: {module_of.get(a) for a in anch} for spec, anch in anchors.items()
    }

    reach_count: dict[str, int] = defaultdict(int)
    for cl in closure.values():
        for q in cl:
            if q not in traced:
                reach_count[q] += 1

    attributed: dict[str, list[str]] = defaultdict(list)
    aligned: dict[str, list[str]] = defaultdict(list)
    for spec, cl in closure.items():
        for q in cl:
            if q not in traced and reach_count[q] == 1:
                attributed[spec].append(q)
                if module_of.get(q) in spec_mods[spec]:
                    aligned[spec].append(q)
    return closure, attributed, aligned


_METHOD_SUMMARY = (
    "Merged static (calls_resolved/3) ∪ runtime (runtime_call/2) call "
    "graph; a spec's anchors are its @traces-linked functions, its closure "
    "is forward-BFS reachability from those anchors; ATTRIBUTED = untraced "
    "closure functions reachable from exactly one spec's closure (the "
    "discriminator separating spec-specific code from shared infra); "
    "ALIGNED (high-confidence tier) = the ATTRIBUTED subset that also lives "
    "in a module the spec already anchors."
)


def build(
    out_json: Path = DEFAULT_OUT_JSON,
    out_report: Path = DEFAULT_REPORT,
    structural_facts: Path | None = None,
    traceability_facts: Path | None = None,
    specs_root: Path | None = None,
) -> dict[str, Any]:
    """Compute the attribution and write both output artifacts.

    ``structural_facts``/``traceability_facts``/``specs_root`` default to
    ``trace_intel``'s own constants; overridable so tests can point at a
    synthetic fixture without touching the real repo KB."""
    structural_path = (
        structural_facts if structural_facts is not None else t.STRUCTURAL_FACTS
    )
    traceability_path = (
        traceability_facts if traceability_facts is not None else t.TRACEABILITY_FACTS
    )
    specs_root_path = specs_root if specs_root is not None else t.SPECS_ROOT

    if structural_facts is None and traceability_facts is None:
        # Reading the real repo KB: refresh it first. Without this, build()
        # answers from whatever facts sit on disk — a specs/ edit that never
        # triggered a trace_intel query left the KB pre-edit for four ship
        # suites (2026-08-15), so the acceptance numbers reported the state
        # BEFORE the checklist restoration and the +55 surfaced mid-ship.
        t.ensure_kb(no_refresh=False)

    structural_text = structural_path.read_text(encoding="utf-8")
    if "runtime_call(" not in structural_text:
        print(
            _MISSING_RUNTIME_CALL_MSG.format(facts=structural_path),
            file=sys.stderr,
        )
        raise SystemExit(1)

    adj, module_of = parse_call_graph(structural_text)
    traceability_text = traceability_path.read_text(encoding="utf-8")
    traces, req_specs = parse_traceability(traceability_text)
    traced = set(traces)
    anchors = compute_anchors(traces, req_specs)
    closure, attributed, aligned = compute_attribution(adj, module_of, anchors, traced)

    spec_names = _spec_name_map(specs_root_path)

    per_spec: list[dict[str, Any]] = []
    for spec in closure:
        aligned_qnames = sorted(aligned.get(spec, []))
        per_spec.append(
            {
                "spec": spec,
                "spec_name": spec_names.get(spec, ""),
                "aligned_count": len(aligned_qnames),
                "closure_count": len(attributed.get(spec, [])),
                "aligned": aligned_qnames,
            }
        )
    per_spec.sort(key=lambda row: (-row["aligned_count"], row["spec"]))

    all_specs_with_reqs: set[str] = {s for specs in req_specs.values() for s in specs}
    specs_without_anchors = sorted(all_specs_with_reqs - set(closure))

    result: dict[str, Any] = {
        "method": _METHOD_SUMMARY,
        "totals": {
            "specs_with_anchors": len(closure),
            "total_attributed": sum(len(v) for v in attributed.values()),
            "total_aligned": sum(len(v) for v in aligned.values()),
        },
        "per_spec": per_spec,
        "specs_without_anchors": specs_without_anchors,
    }

    # Atomic: `open(..., "w")` truncates at OPEN and then streams the document
    # over many writes, so a concurrent reader sees a shrinking file and a
    # concurrent writer splices with it. Byte-identical output (indent=2 plus
    # the trailing newline written explicitly before).
    from orchestrata_ultimate.core.atomic_json import dump_json_atomic

    dump_json_atomic(out_json, result, indent=2, trailing_newline=True)

    write_report(out_report, result)
    return result


# ---------------------------------------------------------------------------
# reports/derived-rtm.md
# ---------------------------------------------------------------------------


def write_report(path: Path, result: dict[str, Any]) -> None:
    totals = result["totals"]
    lines = [
        "# Derived RTM — merged static+runtime attribution",
        "",
        "Generated by `scripts/derive_rtm.py build` over the merged static "
        "(`calls_resolved/3`) + runtime (`runtime_call/2`) call graph in "
        "`.cache/trace_intel/codebase_facts.pl`.",
        "",
        "## Method",
        "",
        result["method"],
        "",
        "## Corpus totals",
        "",
        f"- Specs with anchors: {totals['specs_with_anchors']}",
        f"- Total attributed (untraced, exactly-one-spec closure): "
        f"{totals['total_attributed']}",
        f"- Total aligned (high-confidence, module-matched): "
        f"{totals['total_aligned']}",
        "",
        "## Per-spec proposed @traces",
        "",
    ]
    for row in result["per_spec"]:
        title = f"### SPEC-{row['spec']}"
        if row["spec_name"]:
            title += f" ({row['spec_name']})"
        title += (
            f" — {row['aligned_count']} aligned / {row['closure_count']} attributed"
        )
        lines.append(title)
        lines.append("")
        if not row["aligned"]:
            lines.append("_No high-confidence (module-aligned) attributions._")
        else:
            for qname in row["aligned"]:
                lines.append(f"- [ ] `{qname}`")
        lines.append("")
    lines.append("## Gaps")
    lines.append("")
    if result["specs_without_anchors"]:
        lines.append(
            "Specs with requirements but zero anchors (no `@traces`-linked "
            "code to seed a closure from):"
        )
        lines.append("")
        for spec in result["specs_without_anchors"]:
            lines.append(f"- SPEC-{spec}")
    else:
        lines.append("No specs with requirements are anchor-less.")
    lines.append("")
    lines.append(
        "Abstract Protocol/interface/model code is correctly unreachable "
        "from any anchor closure and therefore excluded from attribution "
        "-- that is expected, not a gap."
    )
    lines.append("")
    from orchestrata_ultimate.core.atomic_json import dump_text_atomic

    dump_text_atomic(path, "\n".join(lines) + "\n")


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------


def _cmd_build(args: argparse.Namespace) -> int:
    out_json = Path(args.out_json)
    out_report = Path(args.out_report)
    result = build(out_json=out_json, out_report=out_report)
    summary = {
        "specs_with_anchors": result["totals"]["specs_with_anchors"],
        "total_attributed": result["totals"]["total_attributed"],
        "total_aligned": result["totals"]["total_aligned"],
        "out_json": str(out_json),
        "out_report": str(out_report),
    }
    print(json.dumps(summary, indent=2))
    return 0


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="derive_rtm",
        description=(
            "Derive an RTM attribution over the merged static+runtime call "
            "graph and propose @traces for untraced, spec-discriminated code."
        ),
    )
    sub = parser.add_subparsers(dest="command", required=True)
    b = sub.add_parser(
        "build",
        help="Compute attribution; write plans/derived-rtm.json + reports/derived-rtm.md.",
    )
    b.add_argument(
        "--out-json",
        default=str(DEFAULT_OUT_JSON),
        help=f"Derived-RTM JSON output path (default: {DEFAULT_OUT_JSON}).",
    )
    b.add_argument(
        "--out-report",
        default=str(DEFAULT_REPORT),
        help=f"Markdown digest output path (default: {DEFAULT_REPORT}).",
    )
    return parser


def main(argv: list[str] | None = None) -> None:
    parser = _build_parser()
    args = parser.parse_args(argv)
    if args.command == "build":
        sys.exit(_cmd_build(args))
    parser.print_help()
    sys.exit(1)


if __name__ == "__main__":
    main()