scripts.attribute_traces_runtime

Reference › scripts › attribute_traces_runtime

Attributes requirements to code by what actually ran: runtime execution evidence for the requirements traceability matrix. It reads a combined coverage database in which each executed line is tagged with the test that ran it. A human or the RTM pipeline runs it; it imports harvest_src_declarations from the generate_rtm_evidence script at run time.

Invocation: --coverage-db is required. Others are --specs (default specs), --src (default src/orchestrata_ultimate), --tests (default tests), --json-out (default docs/trace-attribution.json), --top-n and --ubiquity-cap.

verify_ctrace_core checks that the coverage data was recorded with the compiled tracer core and refuses with a CoverageCoreError (exit 1) if not. test_context_citations maps each test to the requirement ids it cites, executed_defs_by_context maps each test context to the public definitions it executed (public_defs, span), and build_attribution joins these with the @traces declarations in source. Two measures keep results meaningful: breadth (how many tests execute a definition) and spread, used with the ubiquity cap so code that every test touches does not claim every requirement. Each requirement gets a tier, main counts the tiers, and the result is written as JSON.

At a glance

Class diagram

Dependency diagram

No internal imports and no importers.

API

Classes

CoverageCoreError

Bases: RuntimeError · Line 96

Raised when a coverage db does not carry the ctrace re-recording signature this tool depends on (report §3.2's hard gate).

Functions

Source

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

"""Runtime execution-evidence attribution for the RTM (BDD runtime-tracing
Phase 1, ``plans/bdd-runtime-tracing-plan.md``).

Joins per-test coverage attribution (which lines each test's ``|run`` phase
actually executed) against the citation harvest (which requirements each
test claims to verify, ``generate_rtm_evidence.harvest_test_citations``) to
produce deterministic, zero-LLM (requirement -> def) edges with provenance.
This is strictly execution evidence: no judgment, no keyword matching.

Evidence chain (report §5.1, ``reports/research-bdd-runtime-tracing-rtm.md``):

    requirement R --cited by--> test T   (docstring/comment or @traces-on-test
                                          decorator; GHOST ids excluded)
    test T --executed (|run phase)--> def M   (coverage dynamic context,
                                          line -> enclosing PUBLIC def via AST)
    => (M, R) candidate edge, provenance=runtime

HARD GATE -- ctrace core only (report §3.2): coverage.py does NOT persist
which core (ctrace/sysmon/pytrace) collected a ``.coverage`` db -- there is
no stored flag to read back. But sysmon's PEP 669 per-line event disabling
means it never re-records a line under a NEW dynamic context once ANY
context has recorded it, so any code shared by more than one test silently
loses attribution for every test after the first. That is a structural
fingerprint we CAN read back deterministically: if the db has >=2 '|run'
contexts but no line anywhere is ever recorded under more than one of them,
the db was not collected under ctrace (or is otherwise unusable) and this
tool refuses. Verified empirically this session (scratchpad covtest2/): a
helper called by 3 tests recorded 3 contexts under ctrace and exactly 1
(the first caller) under sysmon.

Scoring (report §5.2), for candidate edge (M, R):

    support(M,R)  = # distinct test contexts citing R whose |run phase
                    executed M
    breadth(M)    = # distinct reqs whose citing tests executed M   (IDF)
    spread(R)     = # distinct defs executed by R's citing tests    (IDF)
    ubiquity(M)   = fraction of ALL |run contexts (cited or not) executing M

    score(M,R) = support / (log2(1+breadth) * log2(1+spread))

Hard filters (all from the report, validated against the pilot):
  1. Public defs only -- private helpers/dunders inherit their class/module
     trace (``plans/trace-remediation-plan.md`` W5 ruling).
  2. Infra lane excluded -- reuse ``_INFRA_LANE`` from
     ``scripts/add_traces_from_evidence.py``.
  3. Ubiquity cap: drop M when ubiquity(M) > 0.5 (kills conftest/fixture
     plumbing every test touches).
  4. Phase filter: ``|run`` contexts only.
  5. Top-N=3 proposals per requirement (most-discriminative defs by score).

Output tiers, per edge:
  CONFIRMED           static @traces id on M ∩ a runtime edge (M, R).
  RUNTIME-CANDIDATE    runtime edge (M, R), no static id on M for R, score
                        surviving the filters -- a nomination, not a write.
  STATIC-UNCONFIRMED   R has a static @traces id (on some def D) and citing
                        tests exist in the coverage db, but none of them
                        executed D. NOTE: pass/fail is not an input to this
                        join (the coverage db alone doesn't carry it) --
                        combine with ``generate_rtm_evidence.py --verify``'s
                        FAILING grade to fully match report §5.3's
                        "citing tests all passed" wording. Flag-only, never
                        auto-stripped (D3 pending).
  STATIC-UNTESTABLE    R has a static @traces id but no citing tests at all
                        in this harvest -- not evidence of anything.

Usage:
  python scripts/attribute_traces_runtime.py --coverage-db .coverage
  python scripts/attribute_traces_runtime.py --coverage-db .coverage \\
      --json-out docs/trace-attribution.json --top-n 3
"""

from __future__ import annotations

import argparse
import ast
import json
import math
import sys
from collections import defaultdict
from pathlib import Path

from coverage import CoverageData  # noqa: E402  (third-party; declared in pyproject)

sys.path.insert(0, str(Path(__file__).parent))

from add_traces_from_evidence import _INFRA_LANE  # noqa: E402
from generate_rtm_evidence import (
    harvest_test_citations,
    load_requirements,
)  # noqa: E402

_UBIQUITY_CAP = 0.5
_TOP_N = 3


class CoverageCoreError(RuntimeError):
    """Raised when a coverage db does not carry the ctrace re-recording
    signature this tool depends on (report §3.2's hard gate)."""


def verify_ctrace_core(data: CoverageData) -> None:
    """Refuse coverage data with no evidence it was collected under
    ``COVERAGE_CORE=ctrace``.

    coverage.py stores no "which core" flag -- this checks the one thing
    that structurally differs between cores: whether ANY line, anywhere in
    the measured files, was ever recorded under more than one '|run'
    context. ctrace re-records every context that executes a line; sysmon
    (PEP 669) records only the first. A real suite with >=2 tests that
    share any code path (helpers, base classes, fixtures) will show
    overlap under ctrace; sysmon shows none, ever.

    Raises:
        CoverageCoreError: fewer than 2 '|run' contexts to compare, or no
            line anywhere was ever shared by more than one '|run' context.
    """
    run_contexts = {c for c in data.measured_contexts() if c.endswith("|run")}
    if len(run_contexts) < 2:
        raise CoverageCoreError(
            "coverage db has fewer than 2 '|run' test contexts -- cannot "
            "verify the ctrace re-recording signature (need >=2 contexts "
            "touching shared code). Re-collect with --cov-context=test "
            "over more than one test."
        )
    for filename in data.measured_files():
        contexts_by_line = data.contexts_by_lineno(filename)
        for contexts in contexts_by_line.values():
            if sum(1 for c in contexts if c.endswith("|run")) > 1:
                return  # proof of cross-context re-recording: ctrace confirmed
    raise CoverageCoreError(
        "coverage db looks like it was collected under COVERAGE_CORE=sysmon "
        "(or an equivalent non-re-recording core): no line anywhere in the "
        "measured files was ever recorded under more than one '|run' "
        "context. sysmon (PEP 669) silently drops per-test attribution for "
        "any code exercised by more than one test (report §3.2). Re-collect "
        "with COVERAGE_CORE=ctrace."
    )


def public_defs(py_file: Path) -> list[tuple[str, int, int]]:
    """[(qualname, start_line, end_line)] for public module-level functions
    and public methods of public classes in ``py_file``.

    Public = name does not start with "_" (W5 ruling: private helpers and
    dunders inherit their class/module trace, never attributed on their
    own). ``start_line`` includes any decorator lines so an executed
    decorator call still counts as "inside" the def.
    """
    try:
        tree = ast.parse(py_file.read_text(encoding="utf-8"))
    except OSError, SyntaxError, UnicodeDecodeError:
        return []
    out: list[tuple[str, int, int]] = []

    def span(node: ast.FunctionDef | ast.AsyncFunctionDef) -> tuple[int, int]:
        start = min([node.lineno] + [d.lineno for d in node.decorator_list])
        return start, node.end_lineno or node.lineno

    for node in tree.body:
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            if node.name.startswith("_"):
                continue
            start, end = span(node)
            out.append((node.name, start, end))
        elif isinstance(node, ast.ClassDef) and not node.name.startswith("_"):
            for item in node.body:
                if isinstance(
                    item, (ast.FunctionDef, ast.AsyncFunctionDef)
                ) and not item.name.startswith("_"):
                    start, end = span(item)
                    out.append((f"{node.name}.{item.name}", start, end))
    return out


def test_context_citations(
    tests_dir: Path, reqs: dict[str, dict]
) -> dict[str, set[str]]:
    """{nodeid (no '|run' suffix): {req ids cited by that test}}.

    Built from ``harvest_test_citations`` (docstring + decorator provenance,
    already merged there) via the same enclosing-span logic
    ``add_traces_from_evidence.enclosing_test_functions`` uses, reused
    rather than reimplemented. GHOST ids (cited but defined in no spec) are
    dropped here -- they can never be adjudicated by this join.
    """
    from add_traces_from_evidence import enclosing_test_functions

    cites = harvest_test_citations(tests_dir)
    lines_by_file: dict[str, dict[int, set[str]]] = defaultdict(
        lambda: defaultdict(set)
    )
    for rid, occs in cites.items():
        if rid not in reqs:
            continue
        for occ in occs:
            lines_by_file[occ["file"]][occ["line"]].add(rid)

    out: dict[str, set[str]] = {}
    for test_file, line_map in lines_by_file.items():
        for (tname, tcls), rids in enclosing_test_functions(
            Path(test_file), line_map
        ).items():
            qual = f"{test_file}::{tcls}::{tname}" if tcls else f"{test_file}::{tname}"
            out[qual] = rids
    return out


def executed_defs_by_context(
    data: CoverageData, src_dir: Path
) -> tuple[dict[str, set[str]], set[str]]:
    """(def_qualname -> {context nodeid, ...}, all |run nodeids).

    Returns ``executed_by[def] = {nodeid, ...}`` (which test contexts
    executed that public def's lines) and the set of ALL distinct |run
    nodeids in the db (for the ubiquity denominator). Infra-lane files
    (``_INFRA_LANE``) are never indexed -- hard filter #2.
    """
    executed_by: dict[str, set[str]] = defaultdict(set)
    all_run_nodeids: set[str] = set()
    src_root = src_dir.resolve()
    cwd = Path.cwd()
    for filename in sorted(data.measured_files()):
        abspath = Path(filename).resolve()
        try:
            abspath.relative_to(src_root)
        except ValueError:
            continue  # outside --src entirely, not a candidate
        try:
            rel_to_cwd = str(abspath.relative_to(cwd))
        except ValueError:
            rel_to_cwd = str(abspath)
        # _INFRA_LANE prefixes are repo-root-relative (e.g.
        # "src/orchestrata_ultimate/core/base/"), not --src-relative.
        if any(rel_to_cwd.startswith(lane) for lane in _INFRA_LANE):
            continue
        defs = public_defs(abspath)
        if not defs:
            continue
        contexts_by_line = data.contexts_by_lineno(filename)
        for lineno, contexts in contexts_by_line.items():
            run_ctxs = {c.rsplit("|", 1)[0] for c in contexts if c.endswith("|run")}
            all_run_nodeids |= run_ctxs
            if not run_ctxs:
                continue
            for qualname, start, end in defs:
                if start <= lineno <= end:
                    key = f"{abspath}:{qualname}"
                    executed_by[key] |= run_ctxs
    return dict(executed_by), all_run_nodeids


def build_attribution(
    reqs: dict[str, dict],
    decls: dict[str, list[dict]],
    test_reqs: dict[str, set[str]],
    executed_by: dict[str, set[str]],
    all_run_nodeids: set[str],
    top_n: int = _TOP_N,
    ubiquity_cap: float = _UBIQUITY_CAP,
) -> dict[str, dict]:
    """The scored join (report §5.2). Deterministic: sorted inputs in,
    sorted outputs out."""
    total_contexts = len(all_run_nodeids) or 1
    # citing_contexts[R] = {nodeid, ...} whose citation set contains R --
    # restricted to nodeids that actually ran as PART OF THIS coverage db.
    # A test that cites R but never executed here (e.g. this run measured
    # only a package slice) gives no execution evidence either way for R;
    # counting it as "cited" would produce false STATIC-UNCONFIRMED flags
    # any time the coverage db doesn't happen to cover the whole suite.
    citing_contexts: dict[str, set[str]] = defaultdict(set)
    for nodeid, rids in test_reqs.items():
        if nodeid not in all_run_nodeids:
            continue
        for rid in rids:
            citing_contexts[rid].add(nodeid)

    # ubiquity filter, applied once
    survivors = {
        d: ctxs
        for d, ctxs in executed_by.items()
        if len(ctxs) / total_contexts <= ubiquity_cap
    }

    def breadth(defname: str) -> int:
        ctxs = survivors.get(defname, set())
        return len({rid for c in ctxs for rid in test_reqs.get(c, ())})

    def spread(rid: str) -> int:
        ctxs = citing_contexts.get(rid, set())
        return len({d for d, dctxs in survivors.items() if ctxs & dctxs})

    # Normalize to the SAME resolved-absolute-path form ``executed_by`` keys
    # use (harvest_src_declarations reports file paths relative to however
    # --src was invoked; coverage records its own, possibly-absolute form --
    # without resolving both, CONFIRMED would never match).
    declared_defs: dict[str, set[str]] = defaultdict(set)
    for rid, occs in decls.items():
        for occ in occs:
            declared_defs[rid].add(f"{Path(occ['file']).resolve()}:{occ['name']}")

    result: dict[str, dict] = {}
    for rid in sorted(reqs):
        cited = citing_contexts.get(rid, set())
        declared = declared_defs.get(rid, set())
        if not cited and not declared:
            continue
        if not cited:
            result[rid] = {"tier": "STATIC-UNTESTABLE", "edges": []}
            continue
        sp = spread(rid) or 1
        candidates: list[dict] = []
        for defname, ctxs in survivors.items():
            support = len(cited & ctxs)
            if support == 0:
                continue
            br = breadth(defname) or 1
            score = support / (math.log2(1 + br) * math.log2(1 + sp))
            candidates.append(
                {
                    "def": defname,
                    "tests": sorted(cited & ctxs),
                    "support": support,
                    "breadth": br,
                    "spread": sp,
                    "score": round(score, 6),
                }
            )
        candidates.sort(key=lambda e: (-e["score"], e["def"]))
        top = candidates[:top_n]
        confirmed = [e for e in top if e["def"] in declared]
        if confirmed:
            result[rid] = {"tier": "CONFIRMED", "edges": confirmed}
        elif declared:
            # static id exists, citing tests ran, none executed the
            # declared def(s) -- the wrong-trace / STATIC-UNCONFIRMED flag.
            result[rid] = {
                "tier": "STATIC-UNCONFIRMED",
                "edges": [{"def": d, "tests": sorted(cited)} for d in sorted(declared)],
            }
        elif top:
            result[rid] = {"tier": "RUNTIME-CANDIDATE", "edges": top}
        else:
            result[rid] = {
                "tier": "STATIC-UNTESTABLE" if declared else "UNLINKED",
                "edges": [],
            }
    return result


def main() -> int:
    ap = argparse.ArgumentParser(description=(__doc__ or "").splitlines()[0])
    ap.add_argument("--coverage-db", required=True, help="Combined .coverage db path")
    ap.add_argument("--specs", default="specs")
    ap.add_argument("--src", default="src/orchestrata_ultimate")
    ap.add_argument("--tests", default="tests")
    ap.add_argument("--json-out", default="docs/trace-attribution.json")
    ap.add_argument("--top-n", type=int, default=_TOP_N)
    ap.add_argument("--ubiquity-cap", type=float, default=_UBIQUITY_CAP)
    args = ap.parse_args()

    data = CoverageData(basename=args.coverage_db)
    data.read()
    try:
        verify_ctrace_core(data)
    except CoverageCoreError as exc:
        print(f"REFUSED: {exc}", file=sys.stderr)
        return 1

    from generate_rtm_evidence import harvest_src_declarations

    reqs = load_requirements(Path(args.specs))
    decls = harvest_src_declarations(Path(args.src))
    test_reqs = test_context_citations(Path(args.tests), reqs)
    executed_by, all_run_nodeids = executed_defs_by_context(data, Path(args.src))

    attribution = build_attribution(
        reqs,
        decls,
        test_reqs,
        executed_by,
        all_run_nodeids,
        top_n=args.top_n,
        ubiquity_cap=args.ubiquity_cap,
    )

    tiers: dict[str, int] = defaultdict(int)
    for row in attribution.values():
        tiers[row["tier"]] += 1

    out = {
        "coverage_db": args.coverage_db,
        "run_contexts_total": len(all_run_nodeids),
        "requirements": attribution,
        "tier_counts": dict(sorted(tiers.items())),
    }
    Path(args.json_out).write_text(json.dumps(out, indent=1, sort_keys=True))
    print(
        f"{len(attribution)} requirements attributed -> "
        + ", ".join(f"{t}={n}" for t, n in sorted(tiers.items()))
    )
    print(f"wrote {args.json_out}")
    return 0


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