scripts.affected_tests

Reference › scripts › affected_tests

Resolves a set of changed files to the test modules that could cover them. It implements Phase 3, step 3.1 of the workflow-improvements plan and is the largest script in this group, at about 2,400 lines. test_green_ledger imports it, and two test modules in tests/scripts cover it: the main behaviour and the manifest rules.

Invocation: main takes either --since REV (changed files from git diff --name-only REV) or --files FILE... (an explicit repo-relative list), the two being mutually exclusive, plus --json for machine-readable output and --max-depth for the transitive search limit. Without --json it prints a # VERDICT: line followed by one row per changed file.

It reads the serialized source and test ASTs and a reference-edge index, regenerating any that _is_stale or _is_stale_index finds out of date (regenerate_ast, regenerate_ref_edges). From the source AST it builds a module-to-file map and a reverse dependency graph (build_module_file_map, build_reverse_dep_graph); from the test AST it builds a test-to-source import map (build_test_import_map). transitive_dependents walks the reverse graph up to the depth limit, so a test that imports a dependent of the changed module is selected. Further sources of tests are a rules manifest (load_manifest, manifest_tests), package test trees, a stem naming convention that maps a source file to its expected test path, a suite-wide conftest rule, and key-level matching for environment-endpoint config changes. Several classifiers mark files that need no tests: documentation, dependency or environment changes, empty __init__ files, files transferred to codetools, and generated artifacts. Each file becomes an AffectedRow, and _aggregate_verdict folds the rows into the overall verdict held by AffectedResult. The script writes nothing to the repository except regenerated index files.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

AffectedRow (dataclass)

Decorators: dataclass · Line 139

One changed file's resolution: what it was, how it resolved.

AffectedResult (dataclass)

Decorators: dataclass · Line 156

The resolved answer for a set of changed files.

ReferenceIndex (dataclass)

Decorators: dataclass · Line 599

Reverse view of reference-edges.json: artifact -> referencing files.

_ConftestResolution (dataclass)

Decorators: dataclass · Line 1465

The fixture-aware answer for one changed conftest.

Functions

Source

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

#!/usr/bin/env python3
"""Affected-test selection — the SPEC (``plans/workflow-improvements-plan.md``
Phase 3, Step 3.1) fix for "too long, max wait 2 minutes" / "run only the
part that fails": resolve a set of changed files to the subset of test
modules that could plausibly cover them, instead of the full ~27k-test suite.

Three data sources, combined:

1. **The import closure** — ``plans/ast-ultimate.json`` (src) and
   ``plans/ast-tests.json`` (tests), both produced by
   ``scripts/serialize_codebase_ast.py serialize``. For a changed
   ``src/orchestrata_ultimate/**`` module, walk the SRC dependency graph
   BACKWARDS (who imports this module, transitively, depth <= 3) to find
   every module that could break, then look up which test modules import
   ANY of those modules.
2. **Reference edges** (``plans/reference-edges.json``) — extracted from
   the tree by ``scripts/extract_reference_edges.py``: subprocess argv
   literals (``subprocess.run(["bash", hook.sh])`` — how hook tests bind to
   ``.claude/hooks/*.sh``), pytest-bdd ``scenario(...)`` bindings, Prolog
   ``consult(...)`` paths, and content reads of module-level path
   constants. Exact for the invocation/reference territory the import
   closure cannot see — machine-extracted, NOT hand-maintained
   (2026-08-11: a 2-file deploy/hook change escalated to the full 32k
   suite because these edges were invisible).
3. **The manifest** (``plans/affected-tests-manifest.json``) — demoted to
   genuinely DYNAMIC leftovers: runtime-computed paths a static walk
   cannot resolve, plus the ``scripts/X.py -> tests/scripts/test_X.py``
   stem convention.

**Hard rule (fail-closed, non-negotiable):** a changed file outside
``src/`` and ``tests/`` with NO reference edges and NO manifest coverage
is an UNTESTED-file defect to report (``_UNTESTED_IS_NOT_A_FULL_SUITE``),
not a suite to run — the full suite contains no test for it either.
Documentation (.md/.rst/.txt/.adoc) with zero reference edges is certified
inert and costs only the smoke floor.

**Escalation doctrine (Mohamed, 2026-09-21, verbatim):** "No test that
passes must run twice. Tests MUST only run once unless the code changed
for that specific test. If the hook escalates to a full run then the logic
is wrong." Escalation to the full suite is now reserved for changes whose
HONEST blast radius is the whole suite: a src/ file that exists but sits
outside the serialized closure, a deleted src module whose former
importers cannot be recovered from the range base, a coverage claim a
manifest/stem rule made that is not on disk, and a module-level change to
the ROOT conftest (its scope subtree IS the whole suite — see
``_resolve_conftest_change``). Everything else is bounded: deleted src
modules resolve to their former importers at the range base, zero-coverage
src files are reported untested, and conftest changes are resolved
fixture-aware (import-only edits select nothing; fixture-body edits select
the fixture's consumers; module-level edits bound to the conftest's scope
subtree).

**RUL-032 (Mohamed, 2026-09-22)** sharpened this further: the ``env/*.env``
endpoint contract (RUL-019) no longer escalates either. It used to — every
test inherits it, and no import/reference edge can see the f-string load in
``tests/conftest.py`` — but a blanket full-suite run over a one-line value
change was exactly the "logic is wrong" case above. It now resolves to the
``KEY`` names the edit actually changed, and from there to the test/src
modules whose text names those keys (plus the smoke floor) — see the
``env_files`` handling in ``compute_affected``.

CLI:
    python scripts/affected_tests.py --since HEAD~1
    python scripts/affected_tests.py --files src/orchestrata_ultimate/gateway/audit.py
    python scripts/affected_tests.py --files scripts/foo.py --json

Default text output: one test path (file or directory) per line, always
including the ``tests/infrastructure`` smoke floor, headed by a ``# VERDICT:``
line and one ``# resolved|inert|named_full|refused:`` row per changed file.
``--json`` emits ``{"verdict": "bounded"|"named_full"|"refused", "reasons":
[...], "rows": [...], "tests": [...], "untested": [...], "changed": [...]}``
for programmatic callers (``scripts/run_tests.py --affected``).
"""

from __future__ import annotations

import argparse
import ast
import fnmatch
import functools
import json
import os
import re
import subprocess
import sys
import time
from collections.abc import Callable, Sequence
from dataclasses import asdict, dataclass, field, replace
from pathlib import Path
from typing import Any, cast

#: The repo root (containing plans/, scripts/, tests/, .git). Deliberately
#: NOT `core.paths.get_project_root()`: that resolver walks up looking for
#: a `prolog/` directory and stops at `src/orchestrata_ultimate/prolog/`
#: (the canonical rules dir also happens to satisfy its marker check) --
#: one directory short of the actual repo root this script needs.
REPO_ROOT = Path(__file__).resolve().parent.parent

#: Any src module could, in principle, ripple through this many hops of
#: "who imports the thing that imports the thing I changed" before the
#: signal gets too diffuse to be worth tracking (plan-mandated cap).
MAX_DEPTH = 3

#: Always run this floor regardless of what changed (plan-mandated).
SMOKE_FLOOR: tuple[str, ...] = ("tests/infrastructure",)

#: Modules whose dotted path is looked up with this prefix stripped.
SRC_IMPORT_PREFIX = "orchestrata_ultimate."

#: An AST map older than this is regenerated before use (plan-mandated).
STALE_SECONDS = 86400


#: The three verdicts a resolution can carry (RUL-040: the measured reasons ARE
#: the verdict; a full run is never the silent fallback).
#:
#: ``bounded``    — every changed file resolved to a real test subset (or was
#:                  certified inert). Run that subset.
#: ``named_full`` — at least one change's honest blast radius IS the whole
#:                  suite, NAMED: a dependency/environment/pytest-config file,
#:                  or a root-conftest module-level edit. The gate operator
#:                  sees the name and decides; never an implicit full run.
#: ``refused``    — at least one change has NO resolution at all (nothing tests
#:                  it, a promised test is absent, a deleted module's former
#:                  importers are unrecoverable, or a src file sits outside the
#:                  serialized closure). The run aborts and demands an
#:                  explicit decision.
VERDICT_BOUNDED = "bounded"
VERDICT_NAMED_FULL = "named_full"
VERDICT_REFUSED = "refused"

#: Per-row verdicts (``AffectedRow.verdict``): the three above plus ``inert``
#: for a documentation-class file certified inert (smoke floor only). ``inert``
#: aggregates as ``bounded`` (the smoke floor is a real, bounded selection).
ROW_INERT = "inert"


@dataclass
class AffectedRow:
    """One changed file's resolution: what it was, how it resolved.

    ``verdict`` is ``bounded`` / ``inert`` / ``named_full`` / ``refused``.
    ``tests`` is the resolved test set — empty for ``named_full`` (the name IS
    the whole-suite justification, never an auto-run) and for ``refused``
    (which has none). ``reason`` carries the human-readable classification,
    prefix-tagged for the named_full/refused shapes the gate operator must
    see (RUL-040)."""

    path: str
    verdict: str
    reason: str
    tests: list[str] = field(default_factory=list)


@dataclass
class AffectedResult:
    """The resolved answer for a set of changed files."""

    verdict: str = VERDICT_BOUNDED
    reasons: list[str] = field(default_factory=list)
    tests: list[str] = field(default_factory=list)
    #: Changed files that NOTHING tests — no reference edge, no manifest rule,
    #: no test on disk, and not a dependency/env change. These are a defect to
    #: fix, not a suite to run: see ``_UNTESTED_IS_NOT_A_FULL_SUITE``.
    untested: list[str] = field(default_factory=list)
    #: Per-file resolution rows (see ``AffectedRow``). The aggregate ``verdict``
    #: is the most severe row verdict (refused > named_full > bounded/inert).
    rows: list[AffectedRow] = field(default_factory=list)
    #: The changed file list this resolution was computed for — carried in the
    #: output so a consumer aborts NAMING the changed set, not just the facts.
    changed: list[str] = field(default_factory=list)


def _aggregate_verdict(rows: list[AffectedRow]) -> str:
    """The aggregate verdict: refused > named_full > bounded (inert counts as
    bounded — the smoke floor is a real, bounded selection)."""
    kinds = {row.verdict for row in rows}
    if VERDICT_REFUSED in kinds:
        return VERDICT_REFUSED
    if VERDICT_NAMED_FULL in kinds:
        return VERDICT_NAMED_FULL
    return VERDICT_BOUNDED


# ---------------------------------------------------------------------------
# AST map loading (regenerate-if-stale)
# ---------------------------------------------------------------------------


def _is_stale(path: Path, *, now: float | None = None) -> bool:
    if not path.exists():
        return True
    return ((now if now is not None else time.time()) - path.stat().st_mtime) > (
        STALE_SECONDS
    )


#: Source roots the generated indexes cover (extract_reference_edges.py scans
#: exactly these; keep in sync with regenerate_ref_edges' --source list).
_INDEX_SOURCE_ROOTS = ("src", "scripts", "tests", ".claude")


def _newest_source_mtime(repo_root: Path) -> float:
    """Newest mtime across the indexed source roots.

    Staleness for a generated index must be CONTENT-age, not wall-clock age:
    a fresh `git checkout` stamps every committed file with checkout time, so
    the wall-clock `_is_stale` above sees a committed reference-edges.json
    (content from Aug) as fresh for STALE_SECONDS after ANY checkout —
    measured 2026-09-21 in a push-staging worktree: tests written an hour
    earlier were invisible to the selector because the tracked index's mtime
    was the checkout's, and 3 changed files were judged UNTESTED by an index
    that simply did not know them. Comparing against the newest source mtime
    makes a fresh checkout regenerate exactly once, on first use.
    """
    newest = 0.0
    for root in _INDEX_SOURCE_ROOTS:
        root_path = repo_root / root
        if not root_path.exists():
            continue
        for dirpath, dirnames, filenames in os.walk(root_path):
            dirnames[:] = [
                d for d in dirnames if d not in ("__pycache__", ".git", "node_modules")
            ]
            for filename in filenames:
                try:
                    mtime = (Path(dirpath) / filename).stat().st_mtime
                except OSError:
                    continue
                if mtime > newest:
                    newest = mtime
    return newest


def _is_stale_index(path: Path, repo_root: Path, *, now: float | None = None) -> bool:
    """Wall-clock stale OR older than the newest indexed source."""
    if _is_stale(path, now=now):
        return True
    return path.stat().st_mtime < _newest_source_mtime(repo_root)


def _run_regen(cmd: list[str], repo_root: Path) -> None:
    """Run a snapshot regeneration with its chatter kept OFF stdout.

    In ``--json`` mode this script's stdout IS its contract: the caller does
    ``json.loads(result.stdout)``. A child launched with inherited stdout writes
    onto that same channel, and the serializer prints a human banner
    ("Serialized orchestrata-ultimate: Modules: 867 ...") before the JSON is
    ever emitted, so the payload arrives spliced and unparseable.

    Measured 2026-08-21 in a ship run: ``run_tests.py`` reported
    ``affected_tests.py produced invalid JSON: Expecting value: line 2 column
    1`` and fell back to running the requested paths unmodified -- the whole
    1666-module tree instead of the affected subset. It fails SAFE (more tests,
    never fewer), which is exactly why it went unnoticed; the only symptom is
    an eight-minute run where a thirty-second one was asked for.

    ``check=True`` catches a child that FAILS. Nothing caught a child that
    succeeded noisily. The banner one line above this call already routes
    itself to stderr; the child now gets the same treatment. stderr is used
    directly when it has a real file descriptor so output still streams live;
    under pytest capture it does not, so the output is collected and relayed.
    """
    try:
        stream: Any = sys.stderr.fileno()
    except Exception:
        stream = subprocess.PIPE
    result = subprocess.run(cmd, cwd=str(repo_root), check=True, stdout=stream)
    if stream is subprocess.PIPE and result.stdout:
        sys.stderr.write(result.stdout.decode("utf-8", "replace"))


def regenerate_ast(source: str, name: str, output: Path, *, repo_root: Path) -> None:
    """Regenerate a serialized-AST JSON via ``serialize_codebase_ast.py serialize``.

    Uses ``--incremental`` when a previous copy exists (warm-starts from it,
    reparsing only changed files) — falls back to a full serialize
    automatically on any mismatch, so the output is correct either way.
    """
    cmd = [
        sys.executable,
        str(repo_root / "scripts" / "serialize_codebase_ast.py"),
        "serialize",
        "--source",
        source,
        "--name",
        name,
        "--output",
        str(output),
    ]
    if output.exists():
        cmd.append("--incremental")
    print(f"[affected_tests] regenerating {output} (stale) ...", file=sys.stderr)
    _run_regen(cmd, repo_root)


def _load_ast(
    path: Path,
    source: str,
    name: str,
    *,
    repo_root: Path,
    regenerate: Callable[..., None] = regenerate_ast,
) -> dict[str, Any]:
    if _is_stale(path):
        # NEVER regenerate the canonical committed artifact in place:
        # gates.py's ast-refresh is the ONLY allowed writer of
        # plans/ast-ultimate.json (DESIGN_WAIVER 2026-08-09) — its
        # invocation carries extra_roots + the runtime-callgraph reaugment,
        # while this resolver's plain serialize produces neither. Measured
        # 2026-08-11: a stale-mtime regen shrank the snapshot by 1.8M lines
        # (extra_roots and runtime_calls destroyed) and turned 3
        # snapshot-derived test modules red. Regenerate to a gitignored
        # scratch sidecar and read THAT; the artifact stays gates-owned.
        scratch = path.with_name(path.name + ".affected-scratch")
        if _is_stale(scratch):
            regenerate(source, name, scratch, repo_root=repo_root)
        path = scratch
    with path.open(encoding="utf-8") as fh:
        # json.load's return is Any (stdlib is untyped past the top level);
        # the serialized AST is always a JSON object, so dict[str, Any] is
        # a real narrowing, not a widened escape hatch.
        return cast(dict[str, Any], json.load(fh))


# ---------------------------------------------------------------------------
# Dependency-graph indexing (pure — operates on already-loaded AST dicts)
# ---------------------------------------------------------------------------


def build_module_file_map(
    ast_src: dict, *, repo_root: Path | None = None
) -> dict[str, str]:
    """``{file_path: dotted module path}`` for every serialized src module.

    ``file_path`` is normalized to repo-relative when ``repo_root`` is given
    and the serialized value is absolute -- ``serialize_codebase_ast.py``
    has emitted both conventions depending on invocation (observed live,
    2026-07-26: an ``--incremental`` regen produced absolute paths where a
    full serialize had produced relative ones). ``changed_files`` from
    ``git diff``/``--files`` are always repo-relative, so without this
    normalization a real, freshly-regenerated map can silently stop
    matching anything.
    """
    mapping: dict[str, str] = {}
    for m in ast_src.get("modules", []):
        file_path = m["file_path"]
        if repo_root is not None:
            file_path = _rebase_serialized_path(file_path, repo_root)
        mapping[file_path] = m["path"]
    return mapping


# Top-level directories every serialized path passes through. Used to recover
# a repo-relative path from an index serialized in a DIFFERENT checkout.
_REPO_ANCHORS = ("src", "tests", "scripts", "rtm")


def _rebase_serialized_path(file_path: str, repo_root: Path) -> str:
    """Repo-relative form of a serialized ``file_path``, wherever it came from.

    ``plans/ast-*.json`` is committed but bakes in the ABSOLUTE root of the
    checkout that produced it, so in any other checkout -- a git worktree, a
    fresh clone, CI -- every path is outside ``repo_root``. The old code left
    those untouched with "won't match anyway", which made the whole selector
    match NOTHING and return a plausible-looking but wrong test set instead of
    failing: exactly the false-green an affected-tests gate must not produce.
    Recovering the tail from the last top-level anchor makes the shared index
    relocatable without regenerating it per checkout.
    """
    p = Path(file_path)
    if not p.is_absolute():
        return file_path
    try:
        return str(p.relative_to(repo_root))
    except ValueError:
        pass
    parts = p.parts
    for anchor in _REPO_ANCHORS:
        if anchor in parts:
            idx = len(parts) - 1 - parts[::-1].index(anchor)
            return str(Path(*parts[idx:]))
    return file_path


def build_reverse_dep_graph(ast_src: dict) -> dict[str, set[str]]:
    """Invert ``dependency_graph`` (module -> its imports) into
    (module -> modules that import IT), restricted to internal (src) modules
    -- the dependency_graph's own keys are exactly the internal module set,
    so anything in a deps list that is also a key is internal; anything
    else (``logging``, ``__future__``, third-party) is dropped.
    """
    dg: dict[str, list[str]] = ast_src.get("dependency_graph", {})
    internal = set(dg.keys())
    reverse: dict[str, set[str]] = {mod: set() for mod in internal}
    for mod, deps in dg.items():
        for dep in deps:
            if dep in internal:
                reverse.setdefault(dep, set()).add(mod)
    return reverse


def build_test_import_map(
    ast_tests: dict,
    *,
    src_prefix: str = SRC_IMPORT_PREFIX,
    internal_modules: frozenset[str] = frozenset(),
) -> dict[str, set[str]]:
    """``{src dotted module: {test file paths that import it}}``.

    Matches both ``from orchestrata_ultimate.x import Y`` and
    ``import orchestrata_ultimate.x`` -- either way the AST serializer
    records the imported module's dotted string in ``imports[].module``.

    ``internal_modules`` -- normally the live src AST's ``dependency_graph``
    keys, i.e. every internal module path -- additionally registers a
    ``from pkg import submodule`` test import under the dotted
    ``pkg.submodule`` key when that name is ITSELF a known internal module.
    The serializer records only the parent package in ``imports[].module``
    plus the imported names in ``imports[].names``, so a test doing
    ``from orchestrata_ultimate.tools import ast_pipeline`` was previously
    registered under "tools" only -- ``resolve_src_file``'s BFS starting at
    "tools.ast_pipeline" (the changed file's own module) never found it (72
    occurrences measured 2026-08-14, e.g. tests/tools/test_ast_pipeline.py
    invisible to changes of src/orchestrata_ultimate/tools/ast_pipeline.py).
    The parent-package key is still registered too -- this is additive, not
    a replacement.

    ``src_prefix`` is NOT the only way in. The serialized closure covers
    extra roots as well (``src/orchestrata_ast``, ``src/orchestrata_tools``,
    ``scripts``, ``claude``), and modules under those are numbered by their
    own directory -- ``orchestrata_ast.analyzer``, not
    ``orchestrata_ultimate.orchestrata_ast.analyzer`` -- so the prefix test
    rejects every one of them. Measured 2026-08-20: a change to
    ``src/orchestrata_ast/analyzer.py`` selected 14 tests, ALL of them
    reached indirectly through downstream ``orchestrata_ultimate`` modules,
    while the three modules that test it directly
    (``tests/test_orchestrata_ast_analyzer.py`` among them) were structurally
    invisible. Worse, the selection was non-empty, so the fail-closed
    "zero test import coverage" escalation never fired and the run reported
    ``escalate: false`` -- a bounded blast radius certified as PROVEN over a
    map that could not see the tests. Any module the closure already knows
    is admitted on its own name.
    """
    mapping: dict[str, set[str]] = {}
    for m in ast_tests.get("modules", []):
        test_path = m.get("file_path")
        if not test_path:
            continue
        for imp in m.get("imports", []):
            mod = imp.get("module") or ""
            if mod.startswith(src_prefix):
                dotted = mod[len(src_prefix) :]
            elif mod in internal_modules:
                # Extra-root module: its snapshot path IS the dotted key.
                dotted = mod
            else:
                continue
            if not dotted:
                continue
            mapping.setdefault(dotted, set()).add(test_path)
            for name in imp.get("names") or []:
                submodule_key = f"{dotted}.{name}"
                if submodule_key in internal_modules:
                    mapping.setdefault(submodule_key, set()).add(test_path)
    return mapping


def transitive_dependents(
    start: str, reverse_dep: dict[str, set[str]], max_depth: int = MAX_DEPTH
) -> set[str]:
    """BFS up to ``max_depth`` hops through ``reverse_dep``, starting at
    ``start`` (included at depth 0)."""
    seen = {start}
    frontier = {start}
    for _ in range(max_depth):
        nxt: set[str] = set()
        for mod in frontier:
            nxt |= reverse_dep.get(mod, set())
        nxt -= seen
        if not nxt:
            break
        seen |= nxt
        frontier = nxt
    return seen


def resolve_src_file(
    rel_path: str,
    module_file_map: dict[str, str],
    reverse_dep: dict[str, set[str]],
    test_import_map: dict[str, set[str]],
    max_depth: int = MAX_DEPTH,
) -> set[str]:
    """Test paths affected by a single changed src ``.py`` file."""
    dotted = module_file_map.get(rel_path)
    if dotted is None:
        return set()
    dependents = transitive_dependents(dotted, reverse_dep, max_depth)
    tests: set[str] = set()
    for mod in dependents:
        tests |= test_import_map.get(mod, set())
    return tests


# ---------------------------------------------------------------------------
# Manifest (non-import edges) — Rule 2
# ---------------------------------------------------------------------------


def load_manifest(path: Path) -> list[dict]:
    """``[]`` when the manifest is missing — it is a supplement, not required."""
    if not path.exists():
        return []
    with path.open(encoding="utf-8") as fh:
        data = json.load(fh)
    rules = data.get("rules", [])
    return rules if isinstance(rules, list) else []


def manifest_tests(rel_path: str, rules: list[dict]) -> set[str]:
    """Test paths/dirs a manifest rule maps ``rel_path`` to.

    ``fnmatch`` (not ``pathlib`` globbing) so ``*`` matches across ``/`` --
    a pattern like ``*/prolog/*.pl`` matches
    ``src/orchestrata_ultimate/prolog/governance.pl`` regardless of how many
    directories precede ``prolog/``. Each ``tests`` entry may use a
    ``{stem}`` placeholder for the changed file's basename-without-suffix
    (the ``scripts/X.py -> tests/scripts/test_X.py`` pattern rule).
    """
    stem = Path(rel_path).stem
    return {
        t.format(stem=stem)
        for rule in rules
        if (pattern := rule.get("pattern", "")) and fnmatch.fnmatch(rel_path, pattern)
        for t in rule.get("tests", [])
    }


@functools.cache
def _package_test_tree(pkg: str, root: str) -> tuple[str, ...] | None:
    """The test files of ``packages/<pkg>/tests``, repo-relative — or None.

    ADR-0105: the landed component packages live outside ``src/`` and
    ``tests/`` and their parity suites run in the package's own venv, so
    neither the import closure nor the stem convention can see them. This
    is the packages resolution source: the whole tree is one cached walk
    (thousands of changed files must not re-walk it per file).
    """
    tree = Path(root) / "packages" / pkg / "tests"
    if not tree.is_dir():
        return None
    return tuple(
        sorted(str(p.relative_to(root)) for p in tree.rglob("test_*.py") if p.is_file())
    )


def _package_tests_for(rel_path: str, repo_root: Path) -> set[str] | None:
    """ADR-0105: map ``packages/<pkg>/**`` to the package's parity test tree.

    None when the file is not under ``packages/`` or its package has no
    tests tree on disk — packages without one (design-scripts, review-tools,
    tools, …) are manifest territory and fall through to the untested
    report. A changed test file maps to itself; anything else under the
    package maps to the whole tree (the parity suite covers the package's
    public surface — real imports, one test per claimed requirement).
    """
    parts = rel_path.split("/")
    if len(parts) < 3 or parts[0] != "packages":
        return None
    pkg = parts[1]
    tree = _package_test_tree(pkg, str(repo_root))
    if tree is None:
        return None
    if (
        rel_path.startswith(f"packages/{pkg}/tests/")
        and rel_path.endswith(".py")
        and Path(rel_path).name.startswith("test_")
    ):
        return {rel_path}
    return set(tree)


# ---------------------------------------------------------------------------
# Reference edges (non-import, AST-extracted) — Rule 1b
# ---------------------------------------------------------------------------
#
# plans/reference-edges.json is produced by scripts/extract_reference_edges.py
# walking CALL nodes: subprocess argv literals ("invoke"), pytest-bdd
# scenario bindings ("bdd"), Prolog consults ("consult"), and content reads
# of path constants ("read"). Where the import closure is exact for Python
# imports, these edges are exact for the invocation/reference territory the
# closure cannot see — extracted from the tree, NOT hand-maintained (the
# manifest below is demoted to genuinely dynamic leftovers: runtime-computed
# paths a static walk cannot resolve).


@dataclass
class ReferenceIndex:
    """Reverse view of reference-edges.json: artifact -> referencing files."""

    exact: dict[str, set[str]] = field(default_factory=dict)
    prefixes: list[tuple[str, str]] = field(default_factory=list)
    feature_basenames: dict[str, set[str]] = field(default_factory=dict)

    def referrers(self, rel_path: str) -> set[str]:
        """Every file that references ``rel_path`` — exactly, by directory
        prefix (``CONST / <computed>`` dir edges), or by feature basename
        (pytest-bdd scenario paths are base-dir relative, not repo
        relative)."""
        hits = set(self.exact.get(rel_path, set()))
        for prefix, from_path in self.prefixes:
            if rel_path.startswith(prefix):
                hits.add(from_path)
        if rel_path.endswith(".feature"):
            hits |= self.feature_basenames.get(Path(rel_path).name, set())
        return hits


def regenerate_ref_edges(output: Path, *, repo_root: Path) -> None:
    """Regenerate reference-edges.json via the extractor CLI."""
    cmd = [
        sys.executable,
        str(repo_root / "scripts" / "extract_reference_edges.py"),
        "--source",
        "src",
        "scripts",
        "tests",
        ".claude",
        "--output",
        str(output),
    ]
    print(f"[affected_tests] regenerating {output} (stale) ...", file=sys.stderr)
    _run_regen(cmd, repo_root)


def _load_ref_edges(
    path: Path,
    *,
    repo_root: Path,
    regenerate: Callable[..., None] = regenerate_ref_edges,
) -> dict[str, Any]:
    if _is_stale_index(path, repo_root):
        if (repo_root / "scripts" / "extract_reference_edges.py").exists():
            # Same canonical-artifact rule as _load_ast: never rewrite the
            # committed map in place — regenerate to a scratch sidecar.
            scratch = path.with_name(path.name + ".affected-scratch")
            if _is_stale_index(scratch, repo_root):
                regenerate(scratch, repo_root=repo_root)
            path = scratch
        elif not path.exists():
            # A checkout without the extractor (the synthetic repos
            # affected_tests' own tests build) gets an EMPTY index: every
            # outside file then escalates, which is the fail-CLOSED
            # direction — never a silent skip.
            return {"version": 1, "edges": []}
    with path.open(encoding="utf-8") as fh:
        return cast(dict[str, Any], json.load(fh))


def build_reference_index(data: dict[str, Any]) -> ReferenceIndex:
    index = ReferenceIndex()
    for edge in data.get("edges", []):
        target, from_path, kind = edge.get("target", ""), edge.get("from", ""), ""
        if not target or not from_path:
            continue
        kind = edge.get("kind", "")
        if target.endswith("/"):
            index.prefixes.append((target, from_path))
        elif kind == "bdd":
            index.feature_basenames.setdefault(Path(target).name, set()).add(from_path)
        else:
            index.exact.setdefault(target, set()).add(from_path)
    return index


#: Documentation-class files: with ZERO reference edges they are certified
#: inert (nothing executes or asserts on them), so they cost only the smoke
#: floor instead of escalating. Code-adjacent formats (.feature, .sh, .pl,
#: .py, .json, .toml, .yaml...) are NEVER inert — an unreferenced one stays
#: UNKNOWN and escalates, fail-closed.
_DOC_SUFFIXES = (".md", ".rst", ".txt", ".adoc")

#: Generated-output roots: artifacts a run WRITES (test reports, duration
#: sinks). With zero reference edges they are inert BY CONSTRUCTION — an
#: output nothing reads can break nothing. Measured 2026-08-11: the tracked
#: reports/run-tests-latest.json is rewritten by every suite run, so every
#: subsequent --affected run escalated to the full suite over an OUTPUT.
_INERT_GENERATED_ROOTS = ("reports/",)

#: Agent/config directories that are NON-RUNTIME source. ``.claude/`` is
#: handled by its own directory convention (tests/claude/), so it is NOT
#: listed here; ``.opencode/opencode.json`` is an agent config no test
#: exercises and there is none to write.
_INERT_CONFIG_ROOTS = (".opencode/",)

#: Scripts whose orchestrata coverage TRANSFERRED to codetools (MOVE-109
#: rename engine, ADR-0118 tier-1 pyright_autofix/pyright_ratchet). The
#: remaining file is a thin shim over the deployed codetools CLI; its real
#: tests moved WITH it, so "nothing tests it" is the MOVE, not a defect. The
#: trailing patterns name the tests that left with them, so ``_drop_absent_tests``
#: also treats their deletion as a move, not a coverage gap. fnmatch (`*`
#: crosses ``/``).
_TRANSFERRED_TO_CODETOOLS = (
    "scripts/rename_engine.py",
    "scripts/pyright_autofix.py",
    "scripts/pyright_ratchet.py",
    "tests/refactor/rename*",
    "tests/scripts/test_pyright_*",
    "tests/scripts/test_rename_engine*",
    "tests/refactor/codemod/test_spec109_interop.py",
)

#: Build-infra scripts with no test BY DESIGN: a hatchling build hook runs in
#: the isolated build env (never imported at runtime) and is exercised by
#: ``uv build`` itself, not by a test module.
_INERT_BUILD_SCRIPTS = ("scripts/vendor_members_hook.py",)


def _generated_artifacts() -> frozenset[str]:
    """Every path the RUL-030 registry declares as generated.

    Read from ``scripts/check_generated_artifacts.py`` rather than restated
    here. That registry already names plans/ast-ultimate.json,
    plans/ast-tests.json, plans/ast-deep-comparison.json, docs/CODEBASE_MAP*.md,
    docs/LEVEL0.md and reports/run-tests-latest.json — and every one of them was
    escalating this selector to the full 33k suite while a hardcoded
    ``("reports/",)`` sat here covering exactly one of them.

    Deriving it means an artifact added to the registry is inert here the same
    day, with nobody having to notice this file exists. Measured 2026-08-14: a
    16-test change escalated to 33,104 tests, three of the five causes being
    registry artifacts a run had just rewritten.

    ``runtime_input`` artifacts are EXCLUDED: generated, yes, but the running
    suite consumes them — ``uv.lock`` is the versions every test imports,
    ``cxp_pb2.py`` is the codec the CXP tests exercise. Certifying those inert
    was this function's first defect (caught same day it shipped): a
    dependency bump would have run the smoke floor and called itself green.
    They escalate like any other unknown source change.
    """
    try:
        import importlib.util

        path = Path(__file__).resolve().parent / "check_generated_artifacts.py"
        spec = importlib.util.spec_from_file_location("_cga_registry", path)
        if spec is None or spec.loader is None:
            return frozenset()
        module = importlib.util.module_from_spec(spec)
        # Registered BEFORE exec: the registry declares @dataclass(frozen=True),
        # and dataclasses resolves sys.modules[cls.__module__].__dict__ to look
        # for KW_ONLY. An unregistered spec-loaded module makes that lookup
        # return None and the decorator raises — so this silently fell into the
        # fail-closed branch below and certified nothing.
        sys.modules.setdefault("_cga_registry", module)
        spec.loader.exec_module(module)
        return frozenset(
            a.path
            for a in module.GENERATED_ARTIFACTS
            if not getattr(a, "runtime_input", False)
        )
    except Exception:
        # Fail CLOSED: an unreadable registry means nothing is certified
        # inert, so unknown files escalate exactly as they did before.
        return frozenset()


#: Suffixes the ``X.py -> tests/**/test_X.py`` naming convention is entitled
#: to guess at -- code the convention was ever meant to describe. Without
#: this bound the stem check fires on ANY basename collision: measured
#: 2026-08-14, ``claude/commands/session.md`` stem-matched
#: tests/test_session.py (and 3 more test modules) purely by filename
#: coincidence -- zero real coverage -- and
#: ``specs/**/checklists/requirements.md`` coincidence-matched
#: tests/verification/test_requirements.py the same way. A file outside this
#: tuple that reaches the check now returns nothing and falls through to
#: ``_is_doc_inert`` instead (doc-inert, smoke floor -- or the manifest,
#: which is consulted BEFORE the stem convention and so already wins for any
#: specs/** pattern rule).
_STEM_CONVENTION_SUFFIXES = (".py", ".sh", ".pl")


#: Spellings the stem convention resolves, in addition to the bare
#: ``test_<stem>.py``. ``test_<stem>_cli.py`` is not a guess: it is this
#: repo's established name for the test module covering a CLI shim script,
#: used by **26** of them (factory_*.py, gate.py, refactor.py,
#: schedule_tests.py, …) as of 2026-08-19. Every one of those scripts was
#: invisible to this resolver, so a one-line change to any of them escalated
#: the whole 1664-module suite — measured on this very repo, where a diff
#: touching scripts/gate.py and scripts/factory_decision.py turned an
#: affected run into a full 18-minute one.
#:
#: **Note:** ``rename_engine.py`` moved to codetools/ on 2026-10-20 (MOVE-109).
#:
#: Existence still decides: an extra spelling widens where coverage may be
#: FOUND, never what counts as proof it exists.
_STEM_TEST_SPELLINGS = ("test_{stem}.py", "test_{stem}_cli.py")


def _stem_convention_tests(rel_path: str, repo_root: Path) -> set[str]:
    """tests/**/test_<stem>.py (or test_<stem>_cli.py), when one EXISTS.

    The manifest already declares this for ``scripts/*.py``. Declaring it once
    per directory means every new tool directory escalates the whole suite
    until somebody notices and adds a rule — and nobody notices, because the
    escalation still passes. Measured 2026-08-14: ``claude/hooks/*.py`` has had
    real coverage in ``tests/claude/`` the whole time and was escalating a
    16-test change to 33,104 tests.

    Existence is the whole safety condition. A file that does NOT exist proves
    nothing, so this resolves only against real coverage on disk and otherwise
    returns nothing and lets the caller escalate.

    Restricted to ``_STEM_CONVENTION_SUFFIXES`` -- see that constant's
    docstring for the filename-coincidence defects this bound closes. The
    accepted test spellings are ``_STEM_TEST_SPELLINGS``.
    """
    if not rel_path.endswith(_STEM_CONVENTION_SUFFIXES):
        return set()
    stem = Path(rel_path).stem
    if not stem:
        return set()
    tests_root = repo_root / "tests"
    if not tests_root.is_dir():
        return set()
    return {
        str(hit.relative_to(repo_root))
        for spelling in _STEM_TEST_SPELLINGS
        for hit in tests_root.rglob(spelling.format(stem=stem))
        if hit.is_file()
    }


#: WHY AN UNTESTED FILE NO LONGER ESCALATES (2026-08-26).
#:
#: Escalation exists to stop a SELECTION from silently skipping tests that
#: would have caught a regression. That reasoning requires the tests to exist.
#: When a changed file has no reference edge, no manifest rule and no test on
#: disk, the full suite contains ZERO tests for it — so running all ~1835
#: modules verifies exactly as much as running none of them, at ~8 minutes a
#: go. Measured: scripts/apply_docstring_traces.py had no test, escalated a
#: 3-file change to the full suite, and the suite exercised that script not
#: once. The escalation was not fail-closed; it was a no-op wearing the costume
#: of one, and it hid the real defect (the missing test) behind a green run.
#:
#: So the outcome is now the truth: NOTHING tests this file. That is reported
#: as a defect naming the file — actionable in seconds — instead of being
#: laundered into an expensive run that proves nothing. This is strictly more
#: fail-closed, not less: the old path could pass, this one cannot.
#:
#: Escalation is untouched wherever it still buys coverage: env endpoint
#: contracts, src/ files that EXIST outside the serialized closure, and
#: coverage claims not on disk still escalate. What changed (2026-09-21
#: ruling, see the module docstring): zero-coverage src modules joined this
#: list's own logic -- an unexercised file is EXACTLY the case above, the
#: full suite would not exercise it either -- and deleted src modules /
#: conftest edits got honest bounds instead of blanket escalation.
_UNTESTED_IS_NOT_A_FULL_SUITE = True


def _expected_stem_test_path(rel_path: str) -> str | None:
    """The stem-convention path where this file's test belongs, if it has one.

    Advisory only — it names where to write the missing test so the report is
    actionable. Returns ``None`` for paths the stem convention does not cover.
    """
    if not rel_path.endswith(_STEM_CONVENTION_SUFFIXES):
        return None
    path = Path(rel_path)
    if not path.stem or path.stem == "__init__":
        # tests/<area>/test___init__.py is not a real convention -- a
        # package marker's coverage lives in the package's test modules.
        return None
    area = path.parent.name or "scripts"
    return f"tests/{area}/test_{path.stem}.py"


def _is_doc_inert(rel_path: str) -> bool:
    artifacts = _generated_artifacts()
    return (
        rel_path.endswith(_DOC_SUFFIXES)
        or rel_path.startswith(_INERT_GENERATED_ROOTS)
        or rel_path in artifacts
        # Glob-shaped registry entries (plugins/orchestrata/**). Exact
        # matching alone silently ignored every glob artifact, so a
        # regenerated plugin mirror escalated the FULL suite on each regen —
        # a permanent tax on the very generation the plugin-drift gate
        # enforces. fnmatch's `*` crosses `/`, which is exactly the registry's
        # `**` intent here. runtime_input globs never reach this set, so
        # generated TESTS (which the suite runs) still escalate.
        or any(
            fnmatch.fnmatch(rel_path, pattern)
            for pattern in artifacts
            if "*" in pattern
        )
    )


#: File basenames whose change is a DEPENDENCY or ENVIRONMENT change: every
#: test inherits them, so the honest blast radius IS the whole suite — NAMED
#: (RUL-040), never a silent full run, never a false "NOTHING tests it". The
#: root ``pyproject.toml``/``uv.lock`` are already bounded by manifest rules
#: (their contract tests); this catches the dependency/config files nothing
#: else bounds — a new package's pyproject.toml/uv.lock, a pytest.ini, a
#: setup.cfg, a tox.ini. ``env/*.env`` never reaches this check (RUL-032
#: resolves it to the keys it changed first).
_DEPENDENCY_ENV_NAMES = (
    "pyproject.toml",
    "uv.lock",
    "pytest.ini",
    "setup.cfg",
    "tox.ini",
)


def _is_dependency_or_env_change(rel_path: str) -> bool:
    """True for a dependency/environment/pytest-config change. Such a file is
    ``named_full``, not ``untested``: a dependency change affects EVERY test,
    so "NOTHING tests it" would be false — everything inherits it."""
    return Path(rel_path).name in _DEPENDENCY_ENV_NAMES


def _is_transferred_to_codetools(rel_path: str) -> bool:
    """True for a file whose coverage moved to codetools (MOVE-109 / ADR-0118)."""
    return any(
        fnmatch.fnmatch(rel_path, pattern) for pattern in _TRANSFERRED_TO_CODETOOLS
    )


def _is_empty_init(rel_path: str, repo_root: Path) -> bool:
    """An ``__init__.py`` with nothing but a docstring is a package marker, not
    behaviour — its coverage lives in the package's test modules, never in a
    test of the marker itself."""
    if Path(rel_path).name != "__init__.py":
        return False
    try:
        text = (repo_root / rel_path).read_text(encoding="utf-8")
    except OSError:
        return False
    if not text.strip():
        return True
    import ast

    try:
        tree = ast.parse(text)
    except SyntaxError:
        return False
    return all(
        isinstance(n, ast.Expr)
        and isinstance(n.value, ast.Constant)
        and isinstance(n.value.value, str)
        for n in tree.body
    )


def _is_inert_non_source(rel_path: str, repo_root: Path) -> bool:
    """True for a changed file no test covers BY DESIGN — documentation,
    generated artifacts, an agent/config file, an empty package marker, a
    build hook, or code whose coverage transferred to codetools. Inert → smoke
    floor only, never a refusal (there is no test to write)."""
    return (
        _is_doc_inert(rel_path)
        or rel_path.startswith(_INERT_CONFIG_ROOTS)
        or _is_transferred_to_codetools(rel_path)
        or rel_path in _INERT_BUILD_SCRIPTS
        or _is_empty_init(rel_path, repo_root)
    )


def _is_under_claude(rel_path: str) -> bool:
    """A changed ``claude/**/*.py`` or ``claude/**/*.sh`` file — the one
    source tree outside src/scripts/tests, with no import closure and no
    per-file stem-test naming convention of its own."""
    return rel_path.startswith("claude/") and rel_path.endswith((".py", ".sh"))


def _claude_directory_convention_tests(repo_root: Path) -> set[str]:
    """``tests/claude`` as a directory-level fallback for ``claude/`` changes,
    UNIONED in on top of whatever the more specific rules (reference edges,
    stem convention, manifest) already produced — never gated behind "the
    other rules found nothing".

    Measured 2026-08-14: a 3-file change to ``claude/hooks/intercept_{bash,
    write,edit}.py`` escalated to the full suite. Real coverage exists —
    ``tests/claude/test_intercept_guards.py`` drives all three hooks as
    subprocesses — but it is a MANY-TO-ONE test module (one file covering
    three hooks under different basenames), which the stem convention
    (``test_<stem>.py``) cannot see by construction, and reference-edge
    extraction may not resolve every binding shape (module-level
    string-literal walking has its own reach limits). Rather than chase
    every future many-to-one shape one-off, ``tests/claude/`` -- ~60
    tests/seconds, 500x cheaper than the full suite -- is an honest
    over-approximation: any ``claude/`` change lands somewhere in that one
    small directory's blast radius, because that IS where claude/ coverage
    lives.

    Making this ADDITIVE (rather than a last-resort elif) closes a second,
    worse defect also measured 2026-08-14: the reference-edge extractor's
    basename-literal rule minted coincidental edges from basenames as common
    as ``__init__.py`` (75 test modules contain that literal), so 10 of 96
    claude/ files had a wrong-but-non-empty referrer set that SUPPRESSED
    this fallback entirely under the old elif ladder. Unioning it in always
    means a wrong edge can shrink nothing — it can only ever add.

    Existence AND non-emptiness are the safety condition, mirroring
    ``_stem_convention_tests``: an empty or absent ``tests/claude/`` proves
    nothing, so this returns nothing and lets the caller escalate exactly as
    it did before this convention existed. "Non-empty" means it actually
    holds collectible test modules (``test_*.py``), not just a stray
    ``__pycache__/`` or an ``__init__.py``.
    """
    claude_tests_dir = repo_root / "tests" / "claude"
    if not claude_tests_dir.is_dir():
        return set()
    if not any(claude_tests_dir.rglob("test_*.py")):
        return set()
    return {"tests/claude"}


# ---------------------------------------------------------------------------
# Escalation — Rule 1
# ---------------------------------------------------------------------------


def is_in_scope(rel_path: str) -> bool:
    """``True`` when ``rel_path`` is under ``src/`` or ``tests/`` — the only
    territory the import-closure map can reason about at all."""
    return rel_path.startswith("src/") or rel_path.startswith("tests/")


def _is_env_endpoint_config(rel_path: str) -> bool:
    """``env/*.env`` — the infra endpoint contract every test in the suite
    inherits: ``tests/conftest.py`` loads ``env/{profile}.env`` through an
    f-string (``ORCHESTRATA_ENV_PROFILE``-selected), which is statically
    invisible by design — no import edge, no reference edge worth trusting.

    Before this check, ``env/test.env`` resolved through an INCIDENTAL
    directory-prefix reference edge (``env/`` -> a test that happens to read
    the env directory for an unrelated reason) to exactly one test — a
    wrong-endpoint edit would have smoke-floored green. This must be
    checked, and must win, BEFORE reference-edge/stem/doc-inert resolution
    gets anywhere near the file.

    ``.env`` at the repo root would carry the same argument if it were
    tracked — verified untracked (``git ls-files .env`` empty) at fix time,
    so no rule is needed for it; a future ``git add .env`` would need one.

    RUL-032 (Mohamed, 2026-09-22) replaced the blanket full-suite escalation
    this check used to trigger with key-name resolution: the changed
    ``KEY=value`` names are diffed out and resolved to the test/src modules
    that name them (see the ``env_files`` handling in ``compute_affected``).
    This check still must run, and still must win, first — only the
    CONSEQUENCE changed.
    """
    return fnmatch.fnmatch(rel_path, "env/*.env")


#: KEY=value lines, POSIX env-file convention (env/*.env's own shape).
_ENV_KEY_RE = re.compile(r"^\s*([A-Za-z_][A-Za-z0-9_]*)\s*=")


def _extract_env_keys(lines: list[tuple[int, str]]) -> set[str]:
    """``KEY`` names out of ``(line_number, text)`` pairs — a diff payload
    (``_diff_changed_lines``' removed/added lists) or a plain file's own
    lines (``enumerate``d) — whose text matches the env-file convention."""
    keys: set[str] = set()
    for _, text in lines:
        m = _ENV_KEY_RE.match(text)
        if m:
            keys.add(m.group(1))
    return keys


def _env_changed_keys(
    rel_path: str, *, repo_root: Path, base_rev: str | None
) -> set[str]:
    """The ``KEY`` names that changed in ``rel_path`` since ``base_rev``
    (RUL-032: an ``env/*.env`` edit resolves to the keys it touched, not a
    full-suite escalation). Without a usable diff (no ``base_rev``, not a
    git checkout, the file is new to this range) falls back to every key
    the file CURRENTLY declares — still bounded, never the full suite.
    """
    if base_rev is not None:
        diff = _diff_changed_lines(rel_path, base_rev, repo_root=repo_root)
        if diff is not None:
            removed, added = diff
            return _extract_env_keys(removed) | _extract_env_keys(added)
    try:
        lines = (repo_root / rel_path).read_text(encoding="utf-8").splitlines()
    except OSError:
        return set()
    return _extract_env_keys(list(enumerate(lines, 1)))


#: A maximal run of identifier characters. A key is NAMED by a file iff it
#: equals one of the file's runs: no letter, digit or ``_`` on either side.
_IDENTIFIER_RUN_RE = re.compile(r"[A-Za-z0-9_]+")


def _files_naming_keys(keys: set[str], *, repo_root: Path) -> dict[str, set[str]]:
    """``{key: {repo-relative .py paths that NAME it}}`` — ONE walk of
    ``tests/**/*.py`` and ``src/**/*.py`` regardless of how many keys are
    being resolved (RUL-032: the selector must not re-walk the tree once per
    changed env key).

    A file names a key only where the key stands as a WHOLE identifier:
    ``CC`` is named by ``env["CC"]`` and ``CC=/usr/bin/gcc``, not by
    ``SUCCESS``; ``REDIS_HOST`` is not named by ``REDIS_HOST_OVERRIDE``,
    which is a different key. Until 2026-10-11 the test was ``key in text``:
    `affected_tests --files env/toolchain.env` (keys CC, CXX) then selected
    598 test modules, because ``CC`` occurs inside SUCCESS, ACCESS and
    ACCEPT — a near-full suite under the name "no escalation".

    Each file is split into its identifier runs ONCE and intersected with
    the key set. A per-key regex search gives the same answer and timed out
    (>120 s) on the real env/test.env's key count, measured 2026-10-11."""
    hits: dict[str, set[str]] = {k: set() for k in keys}
    if not keys:
        return hits
    for root_name in ("tests", "src"):
        base = repo_root / root_name
        if not base.is_dir():
            continue
        for path in base.rglob("*.py"):
            try:
                text = path.read_text(encoding="utf-8", errors="replace")
            except OSError:
                continue
            rel = path.relative_to(repo_root).as_posix()
            for key in keys & set(_IDENTIFIER_RUN_RE.findall(text)):
                hits[key].add(rel)
    return hits


def _is_suite_wide_conftest(rel_path: str) -> bool:
    """Any ``conftest.py`` under ``tests/`` — routed to the fixture-aware
    resolver (``_resolve_conftest_change``), NOT to blanket escalation.

    Before this check existed, ``tests/conftest.py`` fell through EVERY
    branch silently: ``is_in_scope`` short-circuits it out of the
    outside-scope resolution ladder (it starts with "tests/"), and it
    never lands in ``test_file_hits`` either (its basename doesn't start
    with "test_") — escalate=False, reasons=[], smoke floor only, with
    nothing printed to explain why. The first fix escalated EVERY conftest
    to the full suite; the 2026-09-21 ruling ("if the hook escalates to a
    full run then the logic is wrong") measured that as wrong in the other
    direction: a 3-conftest push whose hunks were all import-PATH-only
    edits (downstream of a src relocation codemod) escalated 33k tests to
    verify what the moved modules' own blast radius already covered.
    """
    return rel_path.startswith("tests/") and Path(rel_path).name == "conftest.py"


# ---------------------------------------------------------------------------
# Fixture-aware conftest resolution + deleted-module resolution (2026-09-21)
# ---------------------------------------------------------------------------


def _git_show(spec: str, *, repo_root: Path) -> str | None:
    """Content of a git object (``<rev>:<path>``), ``None`` when unreadable.

    Read-only by construction: the base snapshot a deleted module's former
    importers are recovered from is a COMMITTED blob, immutable and
    gates-owned — the regenerate-if-stale/canonical-artifact rules
    ``_load_ast`` honors for the working-tree snapshot are satisfied by
    never writing anything at all.
    """
    result = subprocess.run(
        ["git", "show", spec],
        cwd=str(repo_root),
        capture_output=True,
        text=True,
        check=False,
    )
    return result.stdout if result.returncode == 0 else None


def _resolve_base_rev(rev: str | None, *, repo_root: Path) -> str | None:
    """A single commit SHA usable as the diff/show base for ``rev``.

    ``--since`` takes whatever the caller passes — the pre-push hook hands
    it a 3-dot range (``origin/master...HEAD``), which names a SET of
    commits, not one. ``git show <rev>:path`` and ``git diff <rev> --
    path`` both need a single commit-ish, so a range is resolved to its
    merge-base (exactly what the 3-dot diff itself compares against); a
    2-dot range resolves to its left endpoint. ``None`` means undiffable
    (unborn branch, missing base object) — callers bound wider, never
    narrower.
    """
    if not rev:
        return None
    if "..." in rev:
        left, _, right = rev.partition("...")
        result = subprocess.run(
            ["git", "merge-base", left or "HEAD", right or "HEAD"],
            cwd=str(repo_root),
            capture_output=True,
            text=True,
            check=False,
        )
        base = result.stdout.strip()
        return base or None
    base = rev.split("..", 1)[0] or "HEAD"
    result = subprocess.run(
        ["git", "rev-parse", "--verify", f"{base}^{{commit}}"],
        cwd=str(repo_root),
        capture_output=True,
        text=True,
        check=False,
    )
    return result.stdout.strip() if result.returncode == 0 else None


def _resolve_deleted_src_file(
    rel_path: str,
    *,
    repo_root: Path,
    base_rev: str | None,
    base_ast: dict[str, Any] | None,
    test_import_map: dict[str, set[str]],
    max_depth: int,
) -> tuple[set[str], str, bool]:
    """Blast radius of a src module DELETED in the diff range (Bug A).

    Returns ``(tests, reason, refused)``. Measured 2026-09-21:
    ``src/orchestrata_ultimate/cxp/packet.py`` was deleted by a move to
    ``core/packet/``; the post-move snapshot has no node for the old path,
    so the selector escalated a relocation to the full suite as "cannot
    bound". But a deleted module's blast radius is its FORMER importers,
    and those are readable from the snapshot AT THE RANGE BASE — and are
    typically themselves in the changed set (the move's codemod rewrote
    them), so the correct selection is usually a no-op beyond what the
    importers' own changes already select. Never a full suite.

    The CURRENT test import map is used for the former dependents: any
    test module that imported the deleted module directly must itself be
    in this diff (it would not even collect otherwise), and changed test
    files self-select via ``test_file_hits``. The conservative direction
    is kept: no diff base, an unreadable base snapshot, or a base snapshot
    that never knew the module all still escalate — ambiguity bounds
    wider, never silent-skip.
    """
    if base_rev is None:
        return (
            set(),
            f"{rel_path} was deleted in this range but no diff base is "
            "available to recover its former importers — cannot bound its "
            "blast radius",
            True,
        )
    if base_ast is None:
        return (
            set(),
            f"{rel_path} was deleted in this range and the AST snapshot at "
            f"the range base ({base_rev[:12]}) is unreadable — cannot bound "
            "its former importers",
            True,
        )
    base_map = build_module_file_map(base_ast, repo_root=repo_root)
    dotted = base_map.get(rel_path)
    if dotted is None:
        return (
            set(),
            f"{rel_path} was deleted in this range and the range-base "
            "snapshot never serialized it — its former importers are "
            "invisible, cannot bound its blast radius",
            True,
        )
    base_reverse = build_reverse_dep_graph(base_ast)
    dependents = transitive_dependents(dotted, base_reverse, max_depth) - {dotted}
    tests: set[str] = set()
    for mod in dependents:
        tests |= test_import_map.get(mod, set())
    return (
        tests,
        f"{rel_path} was deleted (moved) in this range — blast radius = its "
        f"{len(dependents)} former importer(s) at the range base "
        f"({', '.join(sorted(dependents)[:3])}"
        f"{'…' if len(dependents) > 3 else ''}), whose own changes resolve "
        "separately",
        False,
    )


#: A diff payload line that cannot change runtime behavior on its own:
#: blank, a comment, or an import. Conftest class 1 (2026-09-21 ruling): a
#: conftest whose hunks touch ONLY such lines is DOWNSTREAM of a module
#: move — the moved module's own blast radius (already selected from the
#: src changes) covers it, and the smoke floor collection-checks that the
#: conftest still imports. Measured on the push range that motivated the
#: ruling: 3 conftest changes, every hunk an ``orchestrator.X`` ->
#: ``infra.X`` import-path rewrite inside a fixture body.
_INERT_DIFF_LINE_RE = re.compile(r"^\s*(?:$|#|import\s+[\w.]|from\s+[\w.]+\s+import\b)")

_HUNK_HEADER_RE = re.compile(r"@@ -(\d+)(?:,\d+)? \+(\d+)(?:,\d+)? @@")


def _diff_changed_lines(
    rel_path: str, base_rev: str, *, repo_root: Path
) -> tuple[list[tuple[int, str]], list[tuple[int, str]]] | None:
    """``(removed, added)`` payload lines as ``(line_number, text)`` pairs.

    Line numbers index the OLD file for removed lines and the NEW file for
    added ones, so each side can be classified against its own AST.
    ``None`` when the diff cannot be produced or parsed — the caller's
    conservative direction (bound wider) takes over.
    """
    result = subprocess.run(
        ["git", "diff", "--no-color", "-U0", base_rev, "--", rel_path],
        cwd=str(repo_root),
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        return None
    removed: list[tuple[int, str]] = []
    added: list[tuple[int, str]] = []
    old_ln = new_ln = 0
    for line in result.stdout.splitlines():
        header = _HUNK_HEADER_RE.match(line)
        if header:
            old_ln, new_ln = int(header.group(1)), int(header.group(2))
        elif line.startswith("-") and not line.startswith("---"):
            removed.append((old_ln, line[1:]))
            old_ln += 1
        elif line.startswith("+") and not line.startswith("+++"):
            added.append((new_ln, line[1:]))
            new_ln += 1
        elif line.startswith(("diff ", "index ", "---", "+++", "\\", "@@")):
            continue
        else:
            # A context line — none expected under -U0, but counting it
            # keeps the line numbers honest if the format ever widens.
            old_ln += 1
            new_ln += 1
    return removed, added


def _decorator_dotted(dec: ast.expr) -> str:
    """Dotted name of a decorator target (``fixture``, ``pytest.fixture``,
    ``pytest.mark.usefixtures``); "" for anything more exotic."""
    parts: list[str] = []
    node = dec
    while isinstance(node, ast.Attribute):
        parts.append(node.attr)
        node = node.value
    if isinstance(node, ast.Name):
        parts.append(node.id)
    return ".".join(reversed(parts))


def _fixture_name_and_span(
    node: ast.FunctionDef | ast.AsyncFunctionDef,
) -> tuple[str, tuple[int, int]] | None:
    """``(provided_name, (first_line, last_line))`` when ``node`` carries a
    ``@pytest.fixture`` decorator, else ``None``. The provided name is the
    decorator's ``name=`` override when present — that, not the function
    name, is what test signatures request."""
    for dec in node.decorator_list:
        target = dec.func if isinstance(dec, ast.Call) else dec
        if not _decorator_dotted(target).endswith("fixture"):
            continue
        name = node.name
        if isinstance(dec, ast.Call):
            for kw in dec.keywords:
                if (
                    kw.arg == "name"
                    and isinstance(kw.value, ast.Constant)
                    and isinstance(kw.value.value, str)
                ):
                    name = kw.value.value
        return name, (node.lineno, node.end_lineno or node.lineno)
    return None


def _conftest_fixture_spans(source: str) -> dict[str, tuple[int, int]]:
    """``{fixture name: (first_line, last_line)}`` for every fixture def in
    a conftest source. Unparseable source yields ``{}`` — the caller's
    conservative direction treats what it cannot see as module-level."""
    try:
        tree = ast.parse(source)
    except SyntaxError, ValueError:
        return {}
    spans: dict[str, tuple[int, int]] = {}
    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            found = _fixture_name_and_span(node)
            if found is not None:
                spans[found[0]] = found[1]
    return spans


def _fixture_consumers(
    fixture_names: set[str], scope: str, *, repo_root: Path
) -> set[str]:
    """Test modules under ``scope`` that request any of ``fixture_names``
    by name — in a test-function signature, or via
    ``@pytest.mark.usefixtures``. Conftest fixtures CHAIN: a conftest
    fixture whose own signature requests an affected fixture is itself
    affected, so the affected set is widened to a fixpoint (a fixture that
    provides ``b`` and requests affected ``a`` makes ``b``'s requesters
    consumers too). A textual pre-filter keeps the AST pass off files that
    cannot possibly request any affected name — for the root conftest the
    scope is the whole ``tests/`` tree.

    A fixture chain deeper than the iteration cap falls back to the whole
    scope subtree: ambiguity bounds wider, never silent-skip.
    """
    scope_dir = repo_root / scope
    if not scope_dir.is_dir():
        return set()
    affected = set(fixture_names)
    consumers: set[str] = set()
    for _ in range(10):
        widened = False
        for path in scope_dir.rglob("*.py"):
            is_conftest = path.name == "conftest.py"
            if not is_conftest and not path.name.startswith("test_"):
                continue
            try:
                text = path.read_text(encoding="utf-8")
            except OSError:
                continue
            if not any(name in text for name in affected):
                continue
            rel = str(path.relative_to(repo_root))
            try:
                tree = ast.parse(text)
            except SyntaxError, ValueError:
                # Unparseable file that textually names an affected fixture
                # — it MIGHT request it. Count it (bounds wider).
                if not is_conftest:
                    consumers.add(rel)
                continue
            for node in ast.walk(tree):
                if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                    continue
                requested = {a.arg for a in (*node.args.args, *node.args.kwonlyargs)}
                for dec in node.decorator_list:
                    if not isinstance(dec, ast.Call):
                        continue
                    if not _decorator_dotted(dec.func).endswith("usefixtures"):
                        continue
                    requested |= {
                        arg.value
                        for arg in dec.args
                        if isinstance(arg, ast.Constant) and isinstance(arg.value, str)
                    }
                if not (requested & affected):
                    continue
                if is_conftest:
                    found = _fixture_name_and_span(node)
                    if found is not None and found[0] not in affected:
                        affected.add(found[0])
                        widened = True
                else:
                    consumers.add(rel)
        if not widened:
            return consumers
    return {scope}


@dataclass
class _ConftestResolution:
    """The fixture-aware answer for one changed conftest."""

    tests: set[str] = field(default_factory=set)
    reason: str = ""
    #: ``VERDICT_NAMED_FULL`` for the root conftest's module-level change (its
    #: scope subtree IS the whole suite — an honest, NAMED full justification),
    #: ``""`` otherwise (a bounded resolution: import-only, fixture-body
    #: consumers, or a nested conftest's subtree).
    verdict: str = ""


def _conftest_scope(rel_path: str) -> str:
    """The pytest visibility subtree of a conftest: the whole ``tests/``
    tree for the ROOT conftest, its own directory subtree otherwise."""
    if rel_path == "tests/conftest.py":
        return "tests"
    return str(Path(rel_path).parent)


def _resolve_conftest_change(
    rel_path: str, *, repo_root: Path, base_rev: str | None
) -> _ConftestResolution:
    """Fixture-aware blast radius of a changed conftest (Bug C).

    Three classes, per the 2026-09-21 ruling:

    1. Hunks touch only import/blank/comment lines -> the edit is
       DOWNSTREAM of a module move; the moved module's own blast radius
       (already selected from the src changes) covers it. Select nothing
       extra — the smoke floor collection-checks the conftest import.
    2. Hunks touch a fixture-decorated def's body -> select the CONSUMERS
       of those fixtures: test modules whose signatures request them by
       name (root conftest: all of ``tests/``; directory conftest: that
       subtree).
    3. Hunks touch module-level non-import code (helpers, constants) ->
       every fixture/test in the conftest's scope subtree is the honest
       blast radius. For the ROOT conftest that subtree IS the whole
       suite — reported as an escalation because that is what running
       everything means to callers, with the reason naming it as the
       honest radius, not a heuristic give-up.

    Conservative direction kept throughout: anything undiffable or
    unparseable degrades toward class 3, never toward silent-skip.
    """
    scope = _conftest_scope(rel_path)

    def subtree(why: str) -> _ConftestResolution:
        root = scope == "tests"
        return _ConftestResolution(
            tests={scope},
            reason=(
                f"{rel_path}: {why} — bounding widest: the conftest's whole "
                f"scope subtree ({scope}/)"
                + (
                    " — for the ROOT conftest that IS the whole suite; this "
                    "is the honest blast radius, not a heuristic escalation"
                    if root
                    else ""
                )
            ),
            verdict=VERDICT_NAMED_FULL if root else "",
        )

    if base_rev is None:
        return subtree("no diff base available (--files mode)")
    diff = _diff_changed_lines(rel_path, base_rev, repo_root=repo_root)
    if diff is None:
        return subtree(f"could not diff against the range base ({base_rev[:12]})")
    removed, added = diff
    removed_active = [(n, t) for n, t in removed if not _INERT_DIFF_LINE_RE.match(t)]
    added_active = [(n, t) for n, t in added if not _INERT_DIFF_LINE_RE.match(t)]
    if not removed_active and not added_active:
        return _ConftestResolution(
            tests=set(),
            reason=(
                f"{rel_path}: import/whitespace-only conftest edit — "
                "downstream of a module move; the moved module's own blast "
                "radius (selected from the src changes) covers it, and the "
                "smoke floor collection-checks the import"
            ),
        )
    old_source = _git_show(f"{base_rev}:{rel_path}", repo_root=repo_root)
    new_path = repo_root / rel_path
    new_source = None
    if new_path.is_file():
        try:
            new_source = new_path.read_text(encoding="utf-8")
        except OSError:
            new_source = None
    if old_source is None or new_source is None:
        return subtree("conftest added or deleted in this range")
    old_fixtures = _conftest_fixture_spans(old_source)
    new_fixtures = _conftest_fixture_spans(new_source)
    affected: set[str] = set()
    for lines, fixtures in (
        (removed_active, old_fixtures),
        (added_active, new_fixtures),
    ):
        for lineno, _text in lines:
            hits = {
                name
                for name, (first, last) in fixtures.items()
                if first <= lineno <= last
            }
            if hits:
                affected |= hits
            else:
                return subtree("module-level (non-fixture, non-import) code changed")
    consumers = _fixture_consumers(affected, scope, repo_root=repo_root)
    return _ConftestResolution(
        tests=consumers,
        reason=(
            f"{rel_path}: fixture body change ({', '.join(sorted(affected))}) "
            f"— selected {len(consumers)} consumer test module(s) that "
            "request the fixture(s) by name"
        ),
    )


# ---------------------------------------------------------------------------
# Orchestration
# ---------------------------------------------------------------------------


def compute_affected(
    changed_files: list[str],
    *,
    repo_root: Path,
    ast_src_path: Path | None = None,
    ast_tests_path: Path | None = None,
    manifest_path: Path | None = None,
    ref_edges_path: Path | None = None,
    max_depth: int = MAX_DEPTH,
    smoke_floor: tuple[str, ...] = SMOKE_FLOOR,
    base_rev: str | None = None,
    load_ast: Callable[..., dict[str, Any]] = _load_ast,
    load_ref_edges: Callable[..., dict[str, Any]] = _load_ref_edges,
) -> AffectedResult:
    """Resolve ``changed_files`` (repo-relative POSIX paths) to the set of
    test paths that could cover them, per the two hard rules in the module
    docstring. Always includes ``smoke_floor``.

    ``base_rev`` is the single commit the diff range is computed against
    (``_resolve_base_rev`` output). It powers the two base-relative
    resolutions: a DELETED src module's former importers (read from the
    snapshot at the base) and fixture-aware conftest diffing. Without it
    both degrade conservatively -- wider bounds, never silent-skip.
    """
    ast_src_path = ast_src_path or (repo_root / "plans" / "ast-ultimate.json")
    ast_tests_path = ast_tests_path or (repo_root / "plans" / "ast-tests.json")
    manifest_path = manifest_path or (
        repo_root / "plans" / "affected-tests-manifest.json"
    )
    ref_edges_path = ref_edges_path or (repo_root / "plans" / "reference-edges.json")

    ref_index = build_reference_index(
        load_ref_edges(ref_edges_path, repo_root=repo_root)
    )

    reasons: list[str] = []
    untested: list[str] = []
    #: Per-file resolution rows (RUL-040): the measured reasons ARE the verdict.
    rows: list[AffectedRow] = []
    #: Per-file bounded test sets, for the "bounded" rows (reference edges,
    #: manifest, stem, claude, package, import closure, env keys, conftest).
    bounded_files: dict[str, set[str]] = {}
    rules = load_manifest(manifest_path)
    #: Test modules selected by the fixture-aware conftest resolver.
    conftest_hits: set[str] = set()
    #: Package parity-suite hits (ADR-0105) — packages/<pkg>/tests paths,
    #: executed by run_tests.py in the package's own venv.
    package_hits: set[str] = set()
    # Referrers routing buckets — filled while classifying outside files,
    # resolved through the import closure / manifest below.
    referrer_src_files: list[str] = []
    referrer_manifest_hits: set[str] = set()
    referrer_test_files: set[str] = set()
    files_with_referrers: set[str] = set()
    #: (rel_path, changed KEY names) for every env/*.env file in this
    #: change set — collected here (still BEFORE the reference-edge lookup,
    #: for the same reason the old unconditional-escalate branch had to run
    #: first: an env file must never fall through to the incidental
    #: directory-prefix edge _is_env_endpoint_config's docstring documents)
    #: but resolved AFTER this loop, once the import-closure maps it needs
    #: are loaded (RUL-032).
    env_files: list[tuple[str, set[str]]] = []
    for f in changed_files:
        if _is_env_endpoint_config(f):
            env_files.append(
                (f, _env_changed_keys(f, repo_root=repo_root, base_rev=base_rev))
            )
            continue
        if is_in_scope(f):
            if _is_suite_wide_conftest(f):
                resolution = _resolve_conftest_change(
                    f, repo_root=repo_root, base_rev=base_rev
                )
                reasons.append(resolution.reason)
                if resolution.verdict:
                    # Root conftest module-level change: its scope subtree IS
                    # the whole suite — an honest, NAMED full justification,
                    # never a silently-run full run (RUL-040). Do NOT merge the
                    # whole-suite marker into the bounded selection; the name
                    # IS the verdict, and the gate operator sees it and decides.
                    rows.append(
                        AffectedRow(
                            path=f,
                            verdict=VERDICT_NAMED_FULL,
                            reason=(
                                f"named_full: dependency-or-environment-change:"
                                f"{f} — {resolution.reason}"
                            ),
                            tests=[],
                        )
                    )
                else:
                    conftest_hits |= resolution.tests
                    bounded_files[f] = set(resolution.tests)
            continue
        # Every source below is tried INDEPENDENTLY and the results are
        # UNIONED -- resolution is no longer exclusive if/elif consumption.
        # Measured 2026-08-14 (deploy/ship.sh): tests/deploy/test_ship.py
        # anchors the script via `Path(__file__).resolve().parents[2]`, a
        # shape reference-edge extraction cannot see, so its only edge was
        # ONE unrelated invoke-edge (tests/claude/test_unshipped_guard.py).
        # Under the old `if referrers: ... elif ...` ladder that single
        # partial edge exclusively CONSUMED the file, skipping the
        # stem-convention branch that would have found the real test sitting
        # right there on disk. A file now counts as resolved when ANY source
        # produces tests, and the selected set is the union of all of them —
        # this only ever GROWS a selection, never shrinks one.
        referrers = ref_index.referrers(f)
        stem_hits = _stem_convention_tests(f, repo_root)
        manifest_hits_for_f = manifest_tests(f, rules)
        claude_hits = (
            _claude_directory_convention_tests(repo_root)
            if _is_under_claude(f)
            else set()
        )
        # Per-file bounded test set (for this file's "bounded" row). Manifest
        # hits are added in the phantom-filtering loop below (filtered there),
        # so this starts with the other resolution sources.
        f_bounded: set[str] = set(stem_hits) | set(claude_hits)

        # A referrer edge counts as a RESOLUTION SOURCE only when it
        # actually CONTRIBUTES coverage -- not merely when one exists.
        # Two shapes are real, non-speculative evidence and always count:
        # the referrer IS a test module (it literally is the coverage), or
        # the referrer is a src/ module (a concrete import-closure entry
        # point; whether IT is covered is resolved -- and, if uncovered,
        # escalated -- downstream by the src_changed loop). The remaining
        # shape is a HEURISTIC guess (stem-expand the referrer's own
        # basename against the scripts/*.py convention) and must survive
        # existence-filtering to count. Measured 2026-08-14: treating mere
        # referrer PRESENCE as "resolved" (regardless of whether the
        # heuristic expansion survived filtering) converted a loud
        # escalation into a silent smoke-floor-only under-selection for any
        # file whose ONLY source is a referrer with an entirely phantom
        # expansion (synthetic repro: scripts/code-review.py --read-->
        # deploy/probe_only.conf) -- the existence filter correctly dropped
        # the phantom TEST PATH but the file was still marked "resolved"
        # by the referrer's mere existence.
        referrer_contributed = False
        if referrers:
            for r in sorted(referrers):
                if r.startswith("tests/") and Path(r).name.startswith("test_"):
                    referrer_test_files.add(r)
                    f_bounded.add(r)
                    referrer_contributed = True
                elif r.startswith("src/") and r.endswith(".py"):
                    referrer_src_files.append(r)
                    referrer_contributed = True
                else:
                    # scripts/*.py, fixtures, helpers — the manifest's stem
                    # convention (scripts/X.py -> tests/scripts/test_X.py)
                    # and its declared rules own this territory. The
                    # REFERRER r, not the changed file f, is what gets
                    # stem-expanded here: a phantom from THAT expansion
                    # describes a file that did NOT change, so it is dropped
                    # silently -- no escalation, no reason line. Measured
                    # 2026-08-14: scripts/graph_test_preflight.py is
                    # referenced by scripts/code-review.py, whose hyphenated
                    # stem manufactures the never-existed
                    # tests/scripts/test_code-review.py; the per-changed-file
                    # phantom filter below only ever sees `f`, never a
                    # referrer, so this phantom reached _drop_absent_tests
                    # and wrongly escalated a change with real coverage.
                    hits = manifest_tests(r, rules)
                    filtered_hits = {h for h in hits if (repo_root / h).exists()}
                    referrer_manifest_hits |= filtered_hits
                    f_bounded |= filtered_hits
                    if filtered_hits:
                        referrer_contributed = True

        resolved = bool(
            referrer_contributed or stem_hits or manifest_hits_for_f or claude_hits
        )

        # ADR-0105: the packages resolution source. A changed file under a
        # landed component package with a tests tree on disk is covered by
        # that package's parity suite (run in the package's own venv, so no
        # other source can see it). Union semantics like every other source:
        # it only ever ADDS tests, never shrinks a selection.
        pkg_hits_for_f = _package_tests_for(f, repo_root)
        if pkg_hits_for_f is not None:
            package_hits |= pkg_hits_for_f
            f_bounded |= pkg_hits_for_f
            resolved = True
            reasons.append(
                f"{f}: covered by its package parity suite "
                f"(packages/{f.split('/')[1]}/tests) — runs in the package's "
                "own venv (ADR-0105)"
            )

        if referrer_contributed:
            files_with_referrers.add(f)
            reasons.append(
                f"{f}: resolved via {len(referrers)} reference edge(s) "
                f"({', '.join(sorted(referrers)[:3])}{'…' if len(referrers) > 3 else ''})"
            )
        elif referrers:
            # Edges exist but none survived contribution-gating -- honest
            # about the difference between "no edge at all" and "an edge
            # that turned out to name nothing on disk", since the escalate
            # reason below (if this is the only source) would otherwise say
            # "no reference edges" about a file that plainly has one.
            reasons.append(
                f"{f}: {len(referrers)} reference edge(s) found "
                f"({', '.join(sorted(referrers)[:3])}{'…' if len(referrers) > 3 else ''}) "
                "but none contributed usable coverage — heuristic stem-expansion "
                "of the referrer resolved to nothing on disk, so this is not "
                "counted as a resolution source"
            )

        if manifest_hits_for_f:
            # The manifest DECLARES coverage for this outside file (the
            # scripts/*.py stem convention, .pl rule files, ...). Rule 1's
            # blanket escalation used to fire anyway — measured 2026-08-11,
            # scripts/run_tests.py escalated a push to 32k tests despite the
            # stem rule mapping it exactly. (The actual tests are gathered,
            # phantom-filtered, in the manifest_hits loop below — this only
            # records that a source resolved f and why.)
            reasons.append(f"{f}: resolved via manifest-declared coverage")

        if stem_hits:
            referrer_test_files |= stem_hits
            reasons.append(
                f"{f}: resolved by the stem convention to "
                f"{', '.join(sorted(stem_hits))} (exists on disk)"
            )

        if claude_hits:
            # ADDITIVE, unconditionally, for every claude/ file — never
            # gated behind "nothing else resolved". Measured 2026-08-14:
            # basename-literal collisions bound claude/**/__init__.py (and
            # other basenames) to dozens of unrelated test modules purely by
            # filename coincidence, and because the old fallback was an
            # elif of last resort, that wrong-but-non-empty referrer set
            # silently SUPPRESSED the tests/claude/ guarantee for 10 of 96
            # claude/ files. A wrong edge can no longer drop this fallback.
            referrer_test_files |= claude_hits
            reasons.append(
                f"{f}: resolved by the claude/ directory convention to "
                f"{', '.join(sorted(claude_hits))} (tests/claude/ exists and is "
                "non-empty)"
            )

        if not resolved:
            if _is_inert_non_source(f, repo_root):
                reason = (
                    f"{f}: documentation with zero reference edges — certified "
                    "inert, smoke floor only"
                )
                reasons.append(reason)
                rows.append(
                    AffectedRow(
                        path=f,
                        verdict=ROW_INERT,
                        reason="certified inert: smoke floor",
                        tests=list(smoke_floor),
                    )
                )
            elif _is_dependency_or_env_change(f):
                reason = (
                    f"{f}: dependency or environment change — every test "
                    "inherits it, so the honest blast radius IS the whole suite"
                )
                reasons.append(reason)
                rows.append(
                    AffectedRow(
                        path=f,
                        verdict=VERDICT_NAMED_FULL,
                        reason=f"named_full: dependency-or-environment-change:{f}",
                        tests=[],
                    )
                )
            else:
                # No resolution at all — REFUSED (RUL-040). See
                # _UNTESTED_IS_NOT_A_FULL_SUITE: the full suite holds nothing
                # that exercises this file either, so running it proves nothing.
                # The run aborts and demands an explicit decision.
                untested.append(f)
                want = _expected_stem_test_path(f)
                fix = f" — write {want}" if want else ""
                if referrers:
                    # A referrer edge existed (reported above) but contributed
                    # nothing usable -- an entirely phantom heuristic
                    # expansion is not a resolution.
                    reason = (
                        f"{f} is outside src/ and tests/ and its only "
                        "reference edge(s) contributed no usable coverage: "
                        f"NOTHING tests it{fix}"
                    )
                else:
                    reason = (
                        f"{f} is outside src/ and tests/ with no reference "
                        "edges and no test on disk: NOTHING tests it — the "
                        "full suite would not exercise it either"
                        f"{fix}"
                    )
                reasons.append(reason)
                rows.append(
                    AffectedRow(
                        path=f,
                        verdict=VERDICT_REFUSED,
                        reason=f"refused: NOTHING tests {f}",
                        tests=[],
                    )
                )
        else:
            bounded_files[f] = f_bounded

    manifest_hits: set[str] = set()
    for f in changed_files:
        hits = manifest_tests(f, rules)
        # Phantom-stem discipline: the {stem} convention promises
        # tests/<area>/test_<stem>.py whether or not that file exists. A
        # phantom promise must ESCALATE when it is the file's only coverage
        # (the 2026-08-10 _irc under-selection — kept, fail-closed, via
        # _drop_absent_tests), but must NOT veto coverage the file provably
        # has through reference edges or another manifest rule (measured
        # 2026-08-11: scripts/serialize_codebase_ast.py resolved via 4
        # reference edges and STILL escalated to 32k tests because the stem
        # file tests/scripts/test_serialize_codebase_ast.py has never
        # existed; scripts/_irc_send.py escalates despite its declared
        # rule). Phantoms are filtered HERE, at compute time, per file —
        # _drop_absent_tests keeps its job for everything that survives.
        existing = {h for h in hits if (repo_root / h).exists()}
        phantoms = hits - existing
        if phantoms and (existing or f in files_with_referrers):
            reasons.append(
                f"{f}: dropped phantom stem-convention path(s) "
                f"({', '.join(sorted(phantoms))}) — real coverage exists"
            )
            hits = existing
        manifest_hits |= hits
        if existing:
            # On-disk manifest coverage is real per-file bounded coverage,
            # contributed here (post phantom-filter) for this file's row.
            bounded_files.setdefault(f, set()).update(existing)

    src_changed = [
        f for f in changed_files if f.startswith("src/") and f.endswith(".py")
    ]
    # Referrers under src/ (a hook/script/feature is read or invoked by a src
    # module) route through the same import-closure resolution as a direct
    # src change — the referrer IS the blast-radius entry point.
    src_changed.extend(f for f in referrer_src_files if f not in src_changed)
    import_hits: set[str] = set()
    env_hits: set[str] = set()
    # The import-closure maps are expensive enough to build once, not once
    # per caller: a direct src/ change AND an env/*.env key resolution
    # (RUL-032) both need module_file_map/reverse_dep/test_import_map, so
    # they are loaded here, the first either one needs them.
    module_file_map: dict[str, str] = {}
    reverse_dep: dict[str, set[str]] = {}
    test_import_map: dict[str, set[str]] = {}
    if src_changed or env_files:
        ast_src = load_ast(
            ast_src_path,
            "src/orchestrata_ultimate",
            "orchestrata-ultimate",
            repo_root=repo_root,
        )
        ast_tests = load_ast(
            ast_tests_path,
            "tests",
            "orchestrata-ultimate-tests",
            repo_root=repo_root,
        )
        module_file_map = build_module_file_map(ast_src, repo_root=repo_root)
        reverse_dep = build_reverse_dep_graph(ast_src)
        test_import_map = build_test_import_map(
            ast_tests,
            internal_modules=frozenset(ast_src.get("dependency_graph", {})),
        )
    if src_changed:
        # Lazily-loaded snapshot AT THE RANGE BASE, for deleted modules
        # (Bug A): parsed at most once, only when a deleted src file needs
        # its former importers recovered. None stays None when unreadable.
        base_ast: dict[str, Any] | None = None
        base_ast_loaded = False
        for f in src_changed:
            if f not in module_file_map:
                if not (repo_root / f).exists():
                    # DELETED in this range (e.g. a relocation). The current
                    # snapshot has no node for the old path BY CONSTRUCTION;
                    # its blast radius is its former importers, readable
                    # from the snapshot at the range base. Measured
                    # 2026-09-21: cxp/packet.py (moved to core/packet/)
                    # escalated a relocation push to the full suite.
                    if not base_ast_loaded:
                        base_ast_loaded = True
                        if base_rev is not None:
                            try:
                                ast_rel = ast_src_path.relative_to(repo_root)
                            except ValueError:
                                ast_rel = None
                            raw = (
                                _git_show(f"{base_rev}:{ast_rel}", repo_root=repo_root)
                                if ast_rel is not None
                                else None
                            )
                            if raw is not None:
                                try:
                                    parsed = json.loads(raw)
                                    base_ast = (
                                        parsed if isinstance(parsed, dict) else None
                                    )
                                except json.JSONDecodeError:
                                    base_ast = None
                    d_tests, d_reason, d_refused = _resolve_deleted_src_file(
                        f,
                        repo_root=repo_root,
                        base_rev=base_rev,
                        base_ast=base_ast,
                        test_import_map=test_import_map,
                        max_depth=max_depth,
                    )
                    import_hits |= d_tests
                    reasons.append(d_reason)
                    if d_refused:
                        rows.append(
                            AffectedRow(
                                path=f,
                                verdict=VERDICT_REFUSED,
                                reason=f"refused: {d_reason}",
                                tests=[],
                            )
                        )
                    else:
                        bounded_files[f] = set(d_tests)
                    continue
                # A src/ .py file the serialized closure has NO node for
                # (e.g. src/orchestrata_ast/**, src/orchestrata_tools/** —
                # ast-ultimate.json covers src/orchestrata_ultimate only).
                # resolve_src_file would silently contribute zero tests:
                # UNSAFE under-selection (found 2026-08-02 when a change to
                # orchestrata_ast selected only the smoke floor). REFUSED.
                reason = (
                    f"{f} is under src/ but outside the serialized import "
                    "closure (ast-ultimate.json covers src/orchestrata_"
                    "ultimate only) — cannot bound its blast radius"
                )
                reasons.append(reason)
                rows.append(
                    AffectedRow(
                        path=f,
                        verdict=VERDICT_REFUSED,
                        reason=f"refused: {reason}",
                        tests=[],
                    )
                )
                continue
            f_hits = resolve_src_file(
                f, module_file_map, reverse_dep, test_import_map, max_depth
            )
            if not f_hits:
                # Bug B (2026-09-21 ruling): this branch used to ESCALATE
                # (added 2026-08-14 to close a silent fall-through; seven
                # non-__init__.py modules and 62 __init__.py were live
                # instances). But an unexercised file is EXACTLY the
                # _UNTESTED_IS_NOT_A_FULL_SUITE case: the full suite
                # contains no test that imports it or anything within
                # depth<=3 of it, so the escalation paid ~1.8k modules to
                # verify exactly as much as the smoke floor does. Measured
                # 2026-09-21: src/orchestrata_ultimate/infra/__init__.py
                # (a new package marker from a relocation) escalated the
                # whole suite over zero coverage. Reclassified to the
                # untested list -- the defect is the missing test, said
                # out loud, with the same write-the-test hint convention
                # used for outside-src files. Still fail-closed: run_tests
                # turns a non-empty untested list into a non-zero exit.
                untested.append(f)
                want = _expected_stem_test_path(f)
                fix = f" — write {want}" if want else ""
                reason = (
                    f"{f} has zero test import coverage within the "
                    f"serialized closure (depth<={max_depth}): NOTHING "
                    "tests it — the full suite would not exercise it "
                    f"either{fix}"
                )
                reasons.append(reason)
                rows.append(
                    AffectedRow(
                        path=f,
                        verdict=VERDICT_REFUSED,
                        reason=f"refused: NOTHING tests {f}",
                        tests=[],
                    )
                )
                continue
            import_hits |= f_hits
            bounded_files[f] = set(f_hits)

    if env_files:
        # RUL-032: resolve every env/*.env file's changed KEY names to the
        # test/src modules that name them — one walk of tests/**/*.py and
        # src/**/*.py for every key across every env file in this change
        # set, not one walk per file.
        all_keys: set[str] = set()
        for _, keys in env_files:
            all_keys |= keys
        key_files = _files_naming_keys(all_keys, repo_root=repo_root)
        for f, keys in env_files:
            file_test_hits: set[str] = set()
            file_src_hits: set[str] = set()
            for key in keys:
                for rel in key_files.get(key, set()):
                    if rel.startswith("tests/") and Path(rel).name.startswith("test_"):
                        file_test_hits.add(rel)
                    elif rel.startswith("src/"):
                        file_src_hits.add(rel)
            resolved_from_src: set[str] = set()
            for src_rel in file_src_hits:
                resolved_from_src |= resolve_src_file(
                    src_rel, module_file_map, reverse_dep, test_import_map, max_depth
                )
            f_tests = file_test_hits | resolved_from_src
            env_hits |= f_tests
            bounded_files[f] = set(f_tests)
            if keys:
                reasons.append(
                    f"{f}: keys {', '.join(sorted(keys))} changed — resolved "
                    f"to {len(f_tests)} test(s) that name them "
                    "(RUL-032: no escalation)"
                )
            else:
                reasons.append(
                    f"{f}: no keys changed — smoke floor only "
                    "(RUL-032: no escalation)"
                )
            unresolved = sorted(key for key in keys if not key_files.get(key))
            if unresolved:
                reasons.append(
                    f"{f}: key(s) {', '.join(unresolved)} named by nothing — "
                    "smoke floor only (RUL-032: no escalation)"
                )

    test_file_hits = {
        f
        for f in changed_files
        if f.startswith("tests/")
        and Path(f).name.startswith("test_")
        and f.endswith(".py")
    }
    for f in test_file_hits:
        # A changed test file is its own bounded coverage.
        bounded_files.setdefault(f, set()).add(f)

    all_tests = (
        set(smoke_floor)
        | import_hits
        | manifest_hits
        | referrer_manifest_hits
        | referrer_test_files
        | test_file_hits
        | conftest_hits
        | env_hits
        | package_hits
    )

    # Build the per-file rows: bounded resolutions first, then the decisive
    # inert/named_full/refused rows already accumulated above.
    for f in sorted(bounded_files):
        if f in {row.path for row in rows}:
            continue
        rows.append(
            AffectedRow(
                path=f,
                verdict=VERDICT_BOUNDED,
                reason=f"{f}: resolved",
                tests=sorted(bounded_files[f]),
            )
        )

    return AffectedResult(
        verdict=_aggregate_verdict(rows),
        reasons=reasons,
        tests=sorted(all_tests),
        untested=sorted(set(untested)),
        rows=rows,
        changed=list(changed_files),
    )


def changed_files_since(rev: str, *, repo_root: Path) -> list[str]:
    """``git diff --name-only <rev>`` (working tree vs. ``rev``), repo-relative."""
    result = subprocess.run(
        ["git", "diff", "--name-only", rev],
        cwd=str(repo_root),
        capture_output=True,
        text=True,
        check=True,
    )
    return [line.strip() for line in result.stdout.splitlines() if line.strip()]


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


def _drop_absent_tests(
    result: AffectedResult,
    *,
    repo_root: Path,
    protected: tuple[str, ...] = SMOKE_FLOOR,
    changed: Sequence[str] = (),
) -> AffectedResult:
    """Remove emitted test paths that are not on disk, and REFUSE when any was.

    ``compute_affected`` is pure over the closure and the manifest; it has no
    business stat-ing the filesystem. But the manifest's ``{stem}`` rules are a
    naming CONVENTION expanded by string substitution -- ``scripts/*.py ->
    tests/scripts/test_{stem}.py`` -- and nothing there checks whether the
    convention was honoured.

    Measured 2026-08-10: a change to ``scripts/_irc_conversation.py`` resolved
    to ``tests/scripts/test__irc_conversation.py``, which has never existed.
    The module that actually covers it is
    ``tests/scripts/test_irc_conversation_capture.py``, and no rule reached it
    until one was added.

    A path that is not there is worse than no path: pytest errors on it, or a
    caller reads a non-empty list and believes the change is covered. And an
    absent path is a COVERAGE CLAIM THAT FAILED, so it escalates for the same
    reason a src/ file outside the serialized closure does -- a rule whose
    target does not exist has not bounded anything.

    ``protected`` is exempt, and that boundary is the point of the check rather
    than an exception to it. What is being validated is a claim RESOLVED from
    the diff: a manifest rule said "this change is covered THERE", and there
    turned out to be nothing. The smoke floor is not resolved from anything --
    it is an unconditional constant every run carries, so its absence says
    something about the checkout, not about whether this change was bounded.
    Measured 2026-08-10: without this exemption the check escalated inside
    test_run_tests.py`s synthetic repo, which deliberately contains only the
    files affected_tests needs and no tests/ tree at all.
    """
    exempt = set(protected)
    absent = {
        t for t in result.tests if t not in exempt and not (repo_root / t).exists()
    }
    if not absent:
        return result
    # Every absent path leaves the selection, refused or not. A test whose
    # coverage moved to codetools (MOVE-109) is a move, so it is not REFUSED --
    # but it is not on disk either, and handed to pytest it is "file or
    # directory not found", which aborts its whole batch. Measured 2026-10-22:
    # the 16 deleted tests/refactor/rename/* paths stayed selected and sank 70
    # modules that never ran.
    tests = [t for t in result.tests if t not in absent]
    missing = sorted(t for t in absent if not _is_transferred_to_codetools(t))
    if not missing:
        return replace(result, tests=tests)

    # Two different causes, same symptom. Both still REFUSE — coverage that is
    # not on disk has bounded nothing either way — but the REASON has to name the
    # right one, because the remedies are opposites: a phantom means write the
    # missing test or fix the rule; a deletion means nothing is wrong at all.
    #
    # Measured 2026-08-13: moving tests/test_gate_parallel_safety.py into
    # tests/scripts/ (so the scripts/X.py -> tests/scripts/test_X.py convention
    # would find it) made the gate report the OLD path as "a naming-convention
    # rule promised coverage that is absent". No rule promised it: it was a
    # deleted file in the diff, mapped to itself. The message sent a reader
    # hunting for a broken convention rule that was working correctly.
    deleted = {t for t in missing if t in set(changed)}
    phantom = [t for t in missing if t not in deleted]

    reasons = list(result.reasons)
    refused_rows: list[AffectedRow] = []
    if phantom:
        reasons.append(
            "mapped to test path(s) that do not exist on disk: "
            + ", ".join(phantom)
            + " — a naming-convention rule promised coverage that is absent, so "
            "the affected set cannot be trusted to bound this change"
        )
        refused_rows.append(
            AffectedRow(
                path=", ".join(phantom),
                verdict=VERDICT_REFUSED,
                reason=(
                    "refused: naming-convention rule promised coverage that is "
                    "absent (" + ", ".join(phantom) + ")"
                ),
                tests=[],
            )
        )
    if deleted:
        reasons.append(
            "test path(s) deleted or moved by this change: "
            + ", ".join(sorted(deleted))
            + " — the coverage they provided is gone from this diff's base, so the "
            "affected set cannot bound it. Nothing is misconfigured: if this was a "
            "rename, the new path resolves separately, and this reason disappears "
            "once the rename is in the compared base"
        )
        refused_rows.append(
            AffectedRow(
                path=", ".join(sorted(deleted)),
                verdict=VERDICT_REFUSED,
                reason=(
                    "refused: test path(s) deleted or moved by this change ("
                    + ", ".join(sorted(deleted))
                    + ")"
                ),
                tests=[],
            )
        )

    return AffectedResult(
        verdict=_aggregate_verdict(list(result.rows) + refused_rows),
        reasons=reasons,
        tests=tests,
        # Carried through: a rebuild that drops this silently loses the
        # untested-file finding, which is the whole point of computing it.
        untested=list(result.untested),
        rows=list(result.rows) + refused_rows,
        changed=list(result.changed),
    )


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description=(
            "Resolve changed files to the test modules that could cover "
            "them (plans/workflow-improvements-plan.md Phase 3 Step 3.1)."
        )
    )
    group = parser.add_mutually_exclusive_group()
    group.add_argument(
        "--since", metavar="REV", help="Changed files via `git diff --name-only REV`."
    )
    group.add_argument(
        "--files",
        nargs="+",
        metavar="FILE",
        help="Explicit changed file list (repo-relative).",
    )
    parser.add_argument(
        "--json", action="store_true", help="Machine-readable JSON output."
    )
    parser.add_argument("--max-depth", type=int, default=MAX_DEPTH)
    args = parser.parse_args(argv)

    repo_root = REPO_ROOT

    if args.files:
        changed = list(args.files)
        base_rev = None
    else:
        changed = changed_files_since(args.since or "HEAD", repo_root=repo_root)
        base_rev = _resolve_base_rev(args.since, repo_root=repo_root)

    result = compute_affected(
        changed, repo_root=repo_root, max_depth=args.max_depth, base_rev=base_rev
    )
    result = _drop_absent_tests(result, repo_root=repo_root, changed=changed)

    if args.json:
        print(json.dumps(asdict(result), indent=2))
        return 0

    print(f"# VERDICT: {result.verdict}")
    for row in result.rows:
        tag = {
            VERDICT_BOUNDED: "resolved",
            ROW_INERT: "inert",
            VERDICT_NAMED_FULL: "named_full",
            VERDICT_REFUSED: "refused",
        }.get(row.verdict, row.verdict)
        tests = f" -> {', '.join(row.tests)}" if row.tests else ""
        print(f"# {tag}: {row.path}{tests} ({row.reason})")
    for reason in result.reasons:
        print(f"# resolved: {reason}")
    for f in result.untested:
        # Named separately from the refused rows because the action is
        # different in kind: this is not "run more tests", it is "no test
        # exists for this file". The full suite would not cover it either.
        print(f"# UNTESTED: {f}")
    for t in result.tests:
        print(t)
    return 0


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