scripts.generate_rtm_evidence

Reference › scripts › generate_rtm_evidence

scripts/generate_rtm_evidence.py generates the evidence-graded requirements traceability matrix (RTM): every requirement row carries the class of proof behind it. It is used by a human and by other scripts as a library: check_feature_loss and check_requirement_corpus import it, as does orchestrata_ultimate.factory.contracts.trace_validation. It builds on orchestrata_ultimate.core.tracing.scanner, orchestrata_ultimate.verification and orchestrata_ultimate.verification.rtm_constraints.

Three harvesters feed it: load_requirements reads requirement definitions from the spec files (requirement_ids_in, requirement_source_files), harvest_src_declarations collects @traces decorators in the source tree, harvest_web_declarations collects @traces comment tags from JS and TS roots (skipping text inside comment spans via _ts_comment_spans), and harvest_test_citations finds requirement ids cited by tests. build_matrix joins them into rows, and build_rtm_facts produces the facts for the integrity check.

main options: --specs (default specs), --src, --web or --ts (repeatable), --tests, --json and --md output paths, --verify (run every citing test module with plain pytest, never xdist, and grade rows TESTED or FAILING; --verify-timeout defaults to 600 seconds) and --check-constraints (list the SPEC-033 REQ-033-C4 integrity violations: untraced, unproven, dangling; exit 1 if any). It writes the JSON and markdown matrices.

At a glance

Source scripts/generate_rtm_evidence.py
Lines of code 788
Classes 0
Functions 16
Imports (internal) orchestrata_ultimate.core.atomic_json, orchestrata_ultimate.core.tracing.scanner, orchestrata_ultimate.verification, orchestrata_ultimate.verification.rtm_constraints
Imported by orchestrata_ultimate.factory.contracts.trace_validation, scripts.check_feature_loss, scripts.check_requirement_corpus
Tests that import it tests/scripts/test_check_feature_loss_corpus.py, tests/scripts/test_check_requirement_corpus.py, tests/scripts/test_rtm_evidence_paths.py

Class diagram

No classes; see Functions.

Dependency diagram

Dependency diagram scripts.generate_rtm_evidence

API

Constants

Name Type Value
WEB_SOURCE_SUFFIXES tuple[str, ...] ('.ts', '.tsx', '.js', '.jsx')
WEB_SOURCE_ROOTS tuple[str, ...] ('opencode', 'vizint-ui')

Functions

Function Signature Summary
__getattr__ __getattr__(name: str) -> Any
requirement_ids_in requirement_ids_in(text: str) -> dict[str, str] id -> title for every requirement DEFINITION in text.
requirement_source_files requirement_source_files(specs_dir: Path) -> list[Path] The normative requirement-definition files — checklist, spec.md, addendum.
load_requirements load_requirements(specs_dir: Path) -> dict[str, dict] id -> {spec, title, file}.
_claim_positions _claim_positions(text: str, tree: ast.Module | None = None) -> tuple[list[tuple[int, int]], dict[int, int]] Where authored CLAIMS can live in a file: docstring line spans and comment start-columns per line.
_ts_comment_spans _ts_comment_spans(text: str) -> list[tuple[int, int]] Character-offset spans of TypeScript // and /* */ comments.
harvest_web_declarations harvest_web_declarations(root: Path) -> dict[str, list[dict]] req id -> [{file, line, name}] from @traces comment tags in JS/TS.
harvest_test_citations harvest_test_citations(tests_dir: Path, corpus: ModuleCorpus | None = None) -> dict[str, list[dict]] req id -> [{file, line, kind}] for authored evidence a test claims to verify that requirement.
harvest_src_declarations harvest_src_declarations(src_dir: Path) -> dict[str, list[dict]] req id -> [{file, line, name}] from @traces decorators in src.
run_citing_tests run_citing_tests(citing_files: list[str], timeout: int) -> dict[str, bool] Run each citing test module (plain pytest, NEVER xdist) and return file -> passed.
build_rtm_facts build_rtm_facts(reqs: dict[str, dict], cites: dict[str, list[dict]], test_results: dict[str, bool] | None) -> RtmFacts Build the SPEC-033 REQ-033-C4 constraint fact set from evidence this script already computed (load_requirements/harvest_test_citations).
format_violation format_violation(v: RtmViolation) -> str One line per violation: kind + the violating identifiers.
run_constraint_check run_constraint_check(reqs: dict[str, dict], cites: dict[str, list[dict]], test_results: dict[str, bool] | None) -> int Enumerate SPEC-033 REQ-033-C4 violations, print each, return the exit code contribution (1 if any violation was found, else 0).
build_matrix build_matrix(reqs: dict[str, dict], cites: dict[str, list[dict]], decls: dict[str, list[dict]], test_results: dict[str, bool] | None) -> list[dict]
render_markdown render_markdown(rows: list[dict], verified: bool) -> str
main main() -> int

Source

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

"""Evidence-graded RTM generator — every row carries its proof class.

Replaces the fabrication-prone "a decorator claim is a link" model
(docs/rtm.{json,md}, measured 55% +/- 5% accurate at W6, 2026-07-09) with a
matrix where each requirement is graded by the STRONGEST evidence that
exists for it, and absence of evidence is reported as absence:

  TESTED    a test module that cites the req id ran GREEN under --verify
            (execution evidence, this run)
  FAILING   a citing test module ran RED under --verify (the claim exists
            and is currently broken — the most actionable row in the file)
  CITED     a test file cites the req id (human-authored, executable claim;
            not executed in this run)
  DECLARED  only @traces decorators in src/ (or @traces comment tags in
            opencode/**/*.ts) claim the req (unverified — the 55% caveat
            applies; see AGENTS.md "@traces Conventions")
  UNLINKED  no evidence anywhere (the honest majority)

Usage:
  python scripts/generate_rtm_evidence.py                     # static grades
  python scripts/generate_rtm_evidence.py --verify            # run citing tests
  python scripts/generate_rtm_evidence.py --json out.json --md out.md

Deterministic: no LLM, no judgment. A row upgrades from DECLARED to
CITED/TESTED only when someone writes a test that names the requirement —
the only way a retroactively-annotated codebase grows a REAL matrix.
"""

from __future__ import annotations

import argparse
import ast
import importlib.util
import re
import subprocess
import sys
from collections import defaultdict
from pathlib import Path
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.core.tracing.scanner import (
    ModuleCorpus,
    RequirementMentionFilter,
    TraceScanner,
)

# Ensure the repo root is on sys.path when this file is run directly (its own
# usage line: `python scripts/generate_rtm_evidence.py`) — a script run by
# file path gets its OWN directory as sys.path[0], not the repo root, so
# `scripts` (the package) wouldn't otherwise resolve for the transitive
# `orchestrata_ultimate.verification` -> `scripts.parse_spec_requirements`
# import below (mirrors the src-on-path bootstrap in kg_schema.py et al.).
_repo_root = Path(__file__).resolve().parent.parent
if str(_repo_root) not in sys.path:
    sys.path.insert(0, str(_repo_root))

# RtmFacts/RtmViolation/check_rtm are imported LAZILY (inside the
# --check-constraints code paths, and via the module __getattr__ below for
# attribute access): the verification package transitively imports
# scripts.parse_spec_requirements at import time, and the default no-flag
# invocation of this script must not depend on that chain.
if TYPE_CHECKING:
    from orchestrata_ultimate.verification.rtm_constraints import (
        RtmFacts,
        RtmViolation,
    )

_LAZY_RTM_EXPORTS = ("RtmFacts", "RtmViolation", "check_rtm")


def __getattr__(name: str) -> Any:  # noqa: N807 — PEP 562 lazy re-export
    # Any is unavoidable here: this returns whichever of two classes or one
    # function `rtm_constraints` exports, by name, without importing that
    # module (and its heavy transitive chain) eagerly.
    if name in _LAZY_RTM_EXPORTS:
        from orchestrata_ultimate.verification import rtm_constraints

        return getattr(rtm_constraints, name)
    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")


# Same ground-truth shapes trace-lint accepts: "### REQ-..." headings and
# "**REQ-...**" checklist bold spans, title text captured to end of line.
_REQ_DEF_RE = re.compile(
    r"(?:^###\s+|\*\*)(REQ-[A-Za-z0-9][A-Za-z0-9-]*)(?:\*\*)?:?\s*(.*)$",
    re.MULTILINE,
)
_REQ_MENTION_RE = re.compile(r"REQ-[0-9][A-Za-z0-9-]*")

# Corpus-absent citation triage (backlog 504). A referenced id the corpus
# does not define is a GHOST row ONLY when it is a complete, plausible
# requirement id (3-digit spec number, then segments each ending in a digit)
# outside the reserved fixture space. Without this split the traceability
# tooling's own parser-rejection fixtures ("nope", "REQ-001-", "REQ-033-C")
# and sentinel fixtures manufacture ghost rows, and GHOST stops being a
# defect count. Specs never reach number 900: ids with spec >= 900 are the
# reserved FIXTURE space for tooling tests (grade FIXTURE, never GHOST);
# tokens that cannot be a requirement id at all grade MALFORMED.
_PLAUSIBLE_REQ_RE = re.compile(r"^REQ-(\d{3})(-[A-Za-z]*\d+)+$")
FIXTURE_SPEC_FLOOR = 900

# A test file containing this pragma opts out of citation harvest entirely:
# every REQ mention in it — docstring, comment, or @traces decorator — is
# fixture data or historical narrative about ids, never a coverage claim.
# For the RTM tooling's own test modules, whose re-pin narratives must cite
# dead ids verbatim and whose fixtures need real-looking ids sub-900.
FIXTURE_IDS_PRAGMA = "rtm-fixture-ids"


def requirement_ids_in(text: str) -> dict[str, str]:
    """id -> title for every requirement DEFINITION in ``text``.

    THE single definition of "this text defines requirement X", shared by the
    RTM loader below and by ``scripts/check_spec_checklists.py``. It is one
    function on purpose: a coverage gate that decided id-hood even slightly
    differently from the loader would certify a checklist as complete while the
    loader still could not see half of it — a pass that means nothing, which is
    the failure mode this repo keeps paying for.

    Definition = a ``### REQ-...`` heading or a ``**REQ-...**`` bold span. A
    bare prose mention does NOT define an id; SPEC-110 has a live instance of
    exactly that (``REQ-110-140`` appears unbolded in its checklist and is
    invisible to the loader).
    """
    out: dict[str, str] = {}
    for m in _REQ_DEF_RE.finditer(text):
        rid, title = m.group(1), m.group(2).strip().rstrip("*").strip()
        out.setdefault(rid, title[:160])
    return out


def requirement_source_files(specs_dir: Path) -> list[Path]:
    """The normative requirement-definition files — checklist, spec.md, addendum.

    Delegates to the AST toolkit's ``_requirement_source_files``, which is the
    RATIFIED definition of the requirement corpus in this repo, loaded by path
    via the same pattern ``ast_report.py::_load_tool`` and
    ``check_spec_authority.py::load_ast_toolkit`` already use.

    Why delegate rather than re-implement: this generator used to glob
    ``*/checklists/requirements.md`` and nothing else. The toolkit had already
    hit that exact bug and fixed it — its docstring records that the narrow scan
    made 43 real ``@traces`` look structurally dangling — but the fix was never
    propagated here. The result was two loaders in one repo disagreeing about
    what the spec corpus contains: ``trace-lint`` reported those citations
    clean while the RTM graded them GHOST. Re-implementing the scan would
    recreate exactly that divergence, so there is one definition and this reads
    it.

    Measured at adoption (2026-08-13): the corpus widens from 5,497 ids to
    5,565 (+68, 0 lost) across 6 files — 034/spec.md (22), 110/spec.md (22),
    033/spec.md (10), 001-kernel/pipeline-event-type-addendum.md (7),
    097/redis-key-namespacing-addendum.md (5), 024/spec.md (2) — and every one
    of the 21 production dangling ``@traces`` resolves to a real requirement.
    """
    tool_path = _repo_root / "scripts" / "serialize_codebase_ast.py"
    spec = importlib.util.spec_from_file_location("serialize_codebase_ast", tool_path)
    if spec is None or spec.loader is None:
        raise ImportError(f"cannot load AST toolkit from {tool_path}")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return list(mod._requirement_source_files(specs_dir))


def load_requirements(specs_dir: Path) -> dict[str, dict]:
    """id -> {spec, title, file}. First definition wins for title/file;
    dual-defined ids record every defining spec."""
    reqs: dict[str, dict] = {}
    for req_file in requirement_source_files(specs_dir):
        # A source is either <spec>/checklists/requirements.md or a file
        # sitting directly in <spec>/ (spec.md, *-addendum.md), so the owning
        # spec dir is one level up in the first shape and zero in the second.
        # Deriving it as parent.parent unconditionally would attribute every
        # spec.md requirement to "specs" and silently mislabel 68 ids.
        spec_dir = (
            req_file.parent.parent
            if req_file.parent.name == "checklists"
            else req_file.parent
        )
        spec_name = spec_dir.name
        try:
            text = req_file.read_text(encoding="utf-8")
        except OSError, UnicodeDecodeError:
            continue
        for rid, title in requirement_ids_in(text).items():
            entry = reqs.setdefault(
                rid,
                {"spec": spec_name, "title": title, "specs": []},
            )
            if spec_name not in entry["specs"]:
                entry["specs"].append(spec_name)
    return reqs


def _claim_positions(
    text: str, tree: ast.Module | None = None
) -> tuple[list[tuple[int, int]], dict[int, int]]:
    """Where authored CLAIMS can live in a file: docstring line spans and
    comment start-columns per line. A REQ id inside an in-code string
    literal is test DATA (e.g. traceability tests feeding sample ids into
    the machinery under test), not a claim that the test verifies it."""
    spans: list[tuple[int, int]] = []
    if tree is None:
        try:
            tree = ast.parse(text)
        except SyntaxError:
            return [], {}
    for node in ast.walk(tree):
        if isinstance(
            node, (ast.Module, ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)
        ):
            body = node.body
            if (
                body
                and isinstance(body[0], ast.Expr)
                and isinstance(body[0].value, ast.Constant)
                and isinstance(body[0].value.value, str)
            ):
                spans.append((body[0].lineno, body[0].end_lineno or body[0].lineno))
    import io
    import tokenize

    comment_cols: dict[int, int] = {}
    try:
        for tok in tokenize.generate_tokens(io.StringIO(text).readline):
            if tok.type == tokenize.COMMENT:
                comment_cols[tok.start[0]] = tok.start[1]
    except tokenize.TokenError, IndentationError:
        pass
    return spans, comment_cols


def _ts_comment_spans(text: str) -> list[tuple[int, int]]:
    """Character-offset spans of TypeScript `//` and `/* */` comments.

    A tiny hand-rolled scanner (not a full TS lexer) that tracks single/
    double/template-quoted strings so a `//` inside a URL string (e.g.
    `"http://localhost:3456/health"`, common in this codebase) is never
    mistaken for a line-comment start. Good enough for finding @traces
    tags deterministically; not a general-purpose TS parser.
    """
    spans: list[tuple[int, int]] = []
    i, n = 0, len(text)
    in_string: str | None = None
    while i < n:
        ch = text[i]
        if in_string:
            if ch == "\\":
                i += 2
                continue
            if ch == in_string:
                in_string = None
            i += 1
            continue
        if ch in ("'", '"', "`"):
            in_string = ch
            i += 1
            continue
        if ch == "/" and i + 1 < n and text[i + 1] == "/":
            nl = text.find("\n", i)
            end = nl if nl != -1 else n
            spans.append((i, end))
            i = end
            continue
        if ch == "/" and i + 1 < n and text[i + 1] == "*":
            close = text.find("*/", i + 2)
            end = close + 2 if close != -1 else n
            spans.append((i, end))
            i = end
            continue
        i += 1
    return spans


# THE canonical output paths for this generator, named so the three components
# that must agree about them cannot silently re-split. They did split once:
# ca71ec06 moved the .md from docs/ to reports/ and added reports/*.json, but
# left the argparse defaults and scripts/project_status.py pointing at docs/.
# The result ran for ten days — every session's `rtm:` status line reported a
# STALE docs/ copy (5,284 rows) while the fresh matrix sat in reports/ (5,340).
# reports/ is where generated run output lives (run-tests-latest.json,
# derived-rtm.md, code-intel-slo.json); docs/ is @import-able documentation,
# which a 2.3 MB machine-written matrix is not.
# Pinned by tests/scripts/test_rtm_evidence_paths.py across producer (here),
# consumer (project_status.get_rtm_summary) and the RUL-030 artifact gate.
DEFAULT_JSON_OUT = "reports/rtm-evidence.json"
DEFAULT_MD_OUT = "reports/rtm-evidence.md"

WEB_SOURCE_SUFFIXES: tuple[str, ...] = (".ts", ".tsx", ".js", ".jsx")

# Every web tree in this repo that can carry `@traces` comment tags. The harvest
# root used to be a single `--ts opencode`, so widening the glob to `.js` bought
# nothing on its own: vizint-ui was still never walked, and SPEC-017's tags could
# not reach the RTM no matter what anyone wrote in them. A root that does not
# exist is skipped by rglob, so listing an optional tree here is harmless.
WEB_SOURCE_ROOTS: tuple[str, ...] = ("opencode", "vizint-ui")


def harvest_web_declarations(root: Path) -> dict[str, list[dict]]:
    """req id -> [{file, line, name}] from `@traces` comment tags in JS/TS.

    Neither JS nor TS has a decorator equivalent to Python's `@traces(...)`,
    so the convention (documented in AGENTS.md "TS-side @traces convention")
    is a plain comment tag: `// @traces REQ-021-065 REQ-021-066` or the same
    inside a `/** */` block. Only ids on the SAME LINE as the `@traces`
    marker are claimed by that tag (a repeated tag line claims a separate
    group). `node_modules` is excluded everywhere in the tree, not just at
    the top level (vendored packages can nest their own).

    Covers `.js`/`.jsx` as well as `.ts`/`.tsx`. It used to glob `*.ts`
    alone, which excluded vizint-ui entirely -- 95 `.js` files against
    opencode/plugins' 25 `.ts` -- so SPEC-017 and SPEC-022 could not be
    traced at all no matter what anyone wrote in them (REQ-022 measured
    39/39 UNLINKED). `_ts_comment_spans` needs nothing added for JS: the
    comment and string-literal forms it scans are the same in both.
    """
    decls: dict[str, list[dict]] = defaultdict(list)
    sources = sorted(
        p for suffix in WEB_SOURCE_SUFFIXES for p in root.rglob(f"*{suffix}")
    )
    for ts in sources:
        if "node_modules" in ts.parts:
            continue
        try:
            text = ts.read_text(encoding="utf-8")
        except OSError, UnicodeDecodeError:
            continue
        for start, end in _ts_comment_spans(text):
            comment = text[start:end]
            for tag in re.finditer(r"@traces\b", comment):
                nl = comment.find("\n", tag.end())
                line_end = nl if nl != -1 else len(comment)
                tail = comment[tag.end() : line_end]
                lineno = text.count("\n", 0, start + tag.start()) + 1
                for m in _REQ_MENTION_RE.finditer(tail):
                    rid = m.group(0)
                    if rid == "REQ-000-000":
                        continue
                    decls[rid].append(
                        {"file": str(ts), "line": lineno, "name": ts.stem}
                    )
    return decls


def harvest_test_citations(
    tests_dir: Path, corpus: ModuleCorpus | None = None
) -> dict[str, list[dict]]:
    """req id -> [{file, line, kind}] for authored evidence a test claims to
    verify that requirement.

    Two citation forms, both blessed as canonical (R-D5,
    AGENTS.md "@traces Conventions"), each carrying its own provenance:

    - ``kind="docstring"``: a REQ id mentioned in a docstring or comment
      (a REQ id in an in-code string literal is test data, not a claim —
      traceability/RTM tests use real-looking ids as fixtures). Placeholders
      are excluded.
    - ``kind="decorator"``: an ``@traces(reqs=[...])`` decorator directly on
      a test function/class, harvested via the shared AST scanner
      (``core.tracing.scanner.collect_traces`` — the same walk src-side
      declarations use) rather than reimplemented here.

    A file containing ``FIXTURE_IDS_PRAGMA`` is skipped for BOTH forms:
    every REQ mention in it is fixture data or narrative about ids, never a
    coverage claim (backlog 504 — the RTM tooling's own test modules were
    manufacturing GHOST rows).

    *corpus* is the shared parse (WI-FLOSS, tasks 578/581): when given, both
    passes consume its text/tree and nothing under *tests_dir* is read or
    parsed again — ``check_feature_loss.run_check`` builds one corpus for the
    whole run. Without it a ``RequirementMentionFilter`` corpus is built here,
    so the CLI is unchanged. Before this seam the docstring pass and
    ``collect_traces`` each parsed the tests tree (2026-10-08, ~2,060
    modules: 185s of a 206s ``run_check`` under host load).
    """
    cites: dict[str, list[dict]] = defaultdict(list)
    pragma_files: set[str] = set()
    if corpus is None:
        corpus = ModuleCorpus.load(tests_dir, RequirementMentionFilter())
    for pm in corpus.modules():
        py, text = pm.path, pm.text
        if FIXTURE_IDS_PRAGMA in text:
            pragma_files.add(str(py.resolve()))
            continue
        if _REQ_MENTION_RE.search(text) is None:
            # No REQ id anywhere in the file: there is nothing to classify,
            # so skip the ast.parse + tokenize that ``_claim_positions`` costs.
            # Measured 2026-10-08: this harvest took 85s solo / 183s on a
            # loaded host over the ~1,850-module tests tree, with the parse
            # and tokenize of every module dominating — most of them mention
            # no requirement at all.
            continue
        doc_spans, comment_cols = _claim_positions(text, pm.tree)
        doc_lines: set[int] = set()
        for a, b in doc_spans:
            doc_lines.update(range(a, b + 1))
        for lineno, line in enumerate(text.splitlines(), 1):
            in_doc = lineno in doc_lines
            ccol = comment_cols.get(lineno)
            for m in _REQ_MENTION_RE.finditer(line):
                rid = m.group(0)
                if rid == "REQ-000-000":
                    continue
                if in_doc or (ccol is not None and m.start() >= ccol):
                    cites[rid].append(
                        {"file": str(py), "line": lineno, "kind": "docstring"}
                    )
    for trace in TraceScanner().collect_from(corpus):
        if str(Path(trace.file_path).resolve()) in pragma_files:
            continue
        for rid in trace.reqs:
            if rid == "REQ-000-000":
                continue
            cites[rid].append(
                {
                    "file": trace.file_path,
                    "line": trace.line_number,
                    "kind": "decorator",
                }
            )
    return cites


def harvest_src_declarations(src_dir: Path) -> dict[str, list[dict]]:
    """req id -> [{file, line, name}] from @traces decorators in src."""
    decls: dict[str, list[dict]] = defaultdict(list)
    for py in sorted(src_dir.rglob("*.py")):
        try:
            tree = ast.parse(py.read_text(encoding="utf-8"))
        except OSError, SyntaxError, UnicodeDecodeError:
            continue
        for node in ast.walk(tree):
            if not isinstance(
                node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)
            ):
                continue
            for dec in node.decorator_list:
                if not (
                    isinstance(dec, ast.Call)
                    and (
                        (isinstance(dec.func, ast.Name) and dec.func.id == "traces")
                        or (
                            isinstance(dec.func, ast.Attribute)
                            and dec.func.attr == "traces"
                        )
                    )
                ):
                    continue
                for kw in dec.keywords:
                    if kw.arg != "reqs" or not isinstance(
                        kw.value, (ast.List, ast.Tuple)
                    ):
                        continue
                    for elt in kw.value.elts:
                        if (
                            isinstance(elt, ast.Constant)
                            and isinstance(elt.value, str)
                            and elt.value != "REQ-000-000"
                        ):
                            decls[elt.value].append(
                                {
                                    "file": str(py),
                                    "line": node.lineno,
                                    "name": node.name,
                                }
                            )
    return decls


def run_citing_tests(citing_files: list[str], timeout: int) -> dict[str, bool]:
    """Run each citing test module (plain pytest, NEVER xdist) and return
    file -> passed. One invocation per module keeps attribution exact."""
    results: dict[str, bool] = {}
    for f in citing_files:
        proc = subprocess.run(
            [
                sys.executable,
                "-m",
                "pytest",
                f,
                "-q",
                "--timeout",
                "120",
                "-p",
                "no:xdist",
                "-p",
                "no:cacheprovider",
            ],
            capture_output=True,
            text=True,
            timeout=timeout,
        )
        results[f] = proc.returncode == 0
        print(f"  {'PASS' if proc.returncode == 0 else 'FAIL'}  {f}", flush=True)
    return results


def build_rtm_facts(
    reqs: dict[str, dict],
    cites: dict[str, list[dict]],
    test_results: dict[str, bool] | None,
) -> RtmFacts:
    """Build the SPEC-033 REQ-033-C4 constraint fact set from evidence this
    script already computed (``load_requirements``/``harvest_test_citations``).

    Only test CITATIONS become ``trace`` facts — a bare ``@traces`` decorator
    claim (``decls``, the DECLARED grade) is not "reached by an asserting
    test" (ADR-0007) and is not proof. A requirement with citations but no
    execution evidence (no ``--verify``) is graded the default minimum
    (``cited``, matching the CITED grade); under ``--verify`` a citing unit
    that ran green/red is graded ``tested``/``failing`` instead, so a
    requirement whose citing tests are ALL currently red surfaces as an
    ``unproven`` violation.
    """
    traces: set[tuple[str, str]] = set()
    test_units: set[str] = set()
    proof_classes: dict[str, str] = {}
    for rid, citations in cites.items():
        for c in citations:
            unit = c["file"]
            traces.add((rid, unit))
            test_units.add(unit)
            if test_results is not None and unit in test_results:
                proof_classes[unit] = "tested" if test_results[unit] else "failing"
            elif unit not in proof_classes:
                proof_classes[unit] = "cited"
    from orchestrata_ultimate.verification.rtm_constraints import RtmFacts

    return RtmFacts(
        requirements=frozenset(reqs),
        traces=frozenset(traces),
        test_units=frozenset(test_units),
        proof_classes=proof_classes,
        min_classes={},
    )


def format_violation(v: RtmViolation) -> str:
    """One line per violation: kind + the violating identifiers."""
    parts = [v.kind, v.req]
    if v.unit is not None:
        parts.append(f"unit={v.unit}")
    if v.needed_class is not None:
        parts.append(f"needed={v.needed_class}")
    return " ".join(parts)


def run_constraint_check(
    reqs: dict[str, dict],
    cites: dict[str, list[dict]],
    test_results: dict[str, bool] | None,
) -> int:
    """Enumerate SPEC-033 REQ-033-C4 violations, print each, return the exit
    code contribution (1 if any violation was found, else 0)."""
    from orchestrata_ultimate.verification.rtm_constraints import check_rtm

    facts = build_rtm_facts(reqs, cites, test_results)
    violations = check_rtm(facts)
    for v in violations:
        print(format_violation(v))
    print(f"{len(violations)} RTM integrity-constraint violation(s)")
    return 1 if violations else 0


def build_matrix(
    reqs: dict[str, dict],
    cites: dict[str, list[dict]],
    decls: dict[str, list[dict]],
    test_results: dict[str, bool] | None,
) -> list[dict]:
    rows = []
    for rid in sorted(reqs):
        cited = cites.get(rid, [])
        declared = decls.get(rid, [])
        if test_results is not None and cited:
            files = {c["file"] for c in cited}
            outcomes = [test_results[f] for f in files if f in test_results]
            grade = (
                "TESTED"
                if outcomes and all(outcomes)
                else "FAILING" if outcomes else "CITED"
            )
        elif cited:
            grade = "CITED"
        elif declared:
            grade = "DECLARED"
        else:
            grade = "UNLINKED"
        rows.append(
            {
                "req": rid,
                "spec": reqs[rid]["spec"],
                "specs": reqs[rid]["specs"],
                "title": reqs[rid]["title"],
                "grade": grade,
                "cited_by": cited,
                "declared_by": declared,
            }
        )
    # Fabrication signal: ids cited/declared but defined in NO spec.
    # Triage (backlog 504): GHOST only for plausible ids outside the
    # reserved fixture space; parser-rejection fixtures grade MALFORMED and
    # sentinel-space ids grade FIXTURE so GHOST stays a clean defect count.
    unknown_ids = (set(cites) | set(decls)) - set(reqs)
    for rid in sorted(unknown_ids):
        m = _PLAUSIBLE_REQ_RE.match(rid)
        if m is None:
            grade = "MALFORMED"
        elif int(m.group(1)) >= FIXTURE_SPEC_FLOOR:
            grade = "FIXTURE"
        else:
            grade = "GHOST"
        rows.append(
            {
                "req": rid,
                "spec": None,
                "specs": [],
                "title": "",
                "grade": grade,
                "cited_by": cites.get(rid, []),
                "declared_by": decls.get(rid, []),
            }
        )
    return rows


def render_markdown(rows: list[dict], verified: bool) -> str:
    counts: dict[str, int] = defaultdict(int)
    by_spec: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int))
    for r in rows:
        counts[r["grade"]] += 1
        if r["spec"]:
            by_spec[r["spec"]][r["grade"]] += 1
    total = sum(counts.values())
    lines = [
        "# RTM — evidence-graded",
        "",
        "> Generated by `scripts/generate_rtm_evidence.py`. Every row carries",
        "> its proof class; a row upgrades ONLY when a test cites the",
        "> requirement (and, under `--verify`, runs green). DECLARED rows are",
        "> unverified decorator claims — measured 55% +/- 5% accurate (W6,",
        "> 2026-07-09); do not treat them as ground truth.",
        "",
        f"Execution evidence this run: {'YES (--verify)' if verified else 'NO (static; CITED not executed)'}",
        "",
        "## Totals",
        "",
        "| grade | requirements | share |",
        "|---|---|---|",
    ]
    for g in (
        "TESTED",
        "FAILING",
        "CITED",
        "DECLARED",
        "UNLINKED",
        "GHOST",
        "FIXTURE",
        "MALFORMED",
    ):
        if counts.get(g):
            lines.append(f"| {g} | {counts[g]} | {100 * counts[g] / total:.1f}% |")
    lines += [
        "",
        "## By spec",
        "",
        "| spec | TESTED | FAILING | CITED | DECLARED | UNLINKED |",
        "|---|---|---|---|---|---|",
    ]
    for spec in sorted(by_spec):
        c = by_spec[spec]
        lines.append(
            f"| {spec} | {c.get('TESTED', 0)} | {c.get('FAILING', 0)} "
            f"| {c.get('CITED', 0)} | {c.get('DECLARED', 0)} | {c.get('UNLINKED', 0)} |"
        )
    lines += ["", "## Evidence rows (TESTED / FAILING / CITED)", ""]
    for r in rows:
        if r["grade"] in ("TESTED", "FAILING", "CITED"):
            tests = sorted({c["file"] for c in r["cited_by"]})
            lines.append(
                f"- **{r['req']}** [{r['grade']}] {r['title']}  \n"
                f"  tests: {', '.join(tests)}"
            )
    ghosts = [r for r in rows if r["grade"] == "GHOST"]
    if ghosts:
        lines += ["", "## GHOST ids (referenced but defined in no spec)", ""]
        for r in ghosts:
            src = r["cited_by"] + r["declared_by"]
            where = f"{src[0]['file']}:{src[0]['line']}" if src else "?"
            lines.append(f"- {r['req']} (first ref: {where})")
    malformed = [r for r in rows if r["grade"] == "MALFORMED"]
    if malformed:
        lines += [
            "",
            "## MALFORMED citation tokens (cannot be a requirement id)",
            "",
        ]
        for r in malformed:
            src = r["cited_by"] + r["declared_by"]
            where = f"{src[0]['file']}:{src[0]['line']}" if src else "?"
            lines.append(f"- `{r['req']}` (first ref: {where})")
    lines.append("")
    return "\n".join(lines)


def main() -> int:
    ap = argparse.ArgumentParser(description=(__doc__ or "").splitlines()[0])
    ap.add_argument("--specs", default="specs")
    ap.add_argument("--src", default="src/orchestrata_ultimate")
    ap.add_argument(
        "--web",
        "--ts",
        dest="web",
        action="append",
        help=(
            "Root to harvest JS/TS `@traces` comment tags from; repeatable. "
            "Defaults to every root in WEB_SOURCE_ROOTS."
        ),
    )
    ap.add_argument("--tests", default="tests")
    ap.add_argument("--json", dest="json_out", default=DEFAULT_JSON_OUT)
    ap.add_argument("--md", dest="md_out", default=DEFAULT_MD_OUT)
    ap.add_argument(
        "--verify",
        action="store_true",
        help="Run every citing test module and grade TESTED/FAILING",
    )
    ap.add_argument("--verify-timeout", type=int, default=600)
    ap.add_argument(
        "--check-constraints",
        action="store_true",
        help=(
            "Enumerate SPEC-033 REQ-033-C4 RTM integrity-constraint violations "
            "(untraced/unproven/dangling) from the evidence this run computed, "
            "print each one, and exit 1 if any were found"
        ),
    )
    args = ap.parse_args()

    reqs = load_requirements(Path(args.specs))
    cites = harvest_test_citations(Path(args.tests))
    decls = harvest_src_declarations(Path(args.src))
    for web_root in args.web or WEB_SOURCE_ROOTS:
        for rid, occurrences in harvest_web_declarations(Path(web_root)).items():
            decls[rid].extend(occurrences)

    test_results = None
    if args.verify:
        citing_files = sorted({c["file"] for cs in cites.values() for c in cs})
        print(f"verifying {len(citing_files)} citing test modules...", flush=True)
        test_results = run_citing_tests(citing_files, args.verify_timeout)

    rows = build_matrix(reqs, cites, decls, test_results)
    # Atomic: reports/rtm-evidence.json is 2.9 MB and regenerated by the gate
    # battery; a truncate-in-place rewrite SIGBUSes any reader holding it
    # mmap'd. No trailing newline, matching the previous writer byte for byte.
    from orchestrata_ultimate.core.atomic_json import (
        dump_json_atomic,
        dump_text_atomic,
    )

    dump_json_atomic(
        Path(args.json_out),
        {
            "verified": bool(args.verify),
            "requirements": len(reqs),
            "rows": rows,
        },
        indent=1,
    )
    dump_text_atomic(Path(args.md_out), render_markdown(rows, bool(args.verify)))
    counts: dict[str, int] = defaultdict(int)
    for r in rows:
        counts[r["grade"]] += 1
    print(
        f"{len(reqs)} requirements -> "
        + ", ".join(f"{g}={n}" for g, n in sorted(counts.items()))
    )
    print(f"wrote {args.json_out} + {args.md_out}")

    if args.check_constraints:
        return run_constraint_check(reqs, cites, test_results)

    return 0


if __name__ == "__main__":
    sys.exit(main())