scripts.test-review

Reference › scripts › test-review

scripts/test-review.py is the test file reviewer: it checks test files against the orchestrata-ultimate test standards and reports findings with severity ERROR, WARNING or INFO. A human or a review gate runs it; it has no internal imports. See audit_test_purity, test_isolation_audit and gate_parallel_safety for related checks.

find_test_files expands the paths given, and TestFileChecker.check_file parses each file with ast and runs a set of rule methods, each recording issues in a TestReport. The rules cover imports (including banned libraries and relative imports), fixtures, skips, mocking, monkeypatch, os.environ use, try-except-pass, module-level mutable state, hard-coded ports, xdist use, test-id usage, anyio marks and async tests, graph-write and Redis-write isolation, teardown of per-test DDL and chunk-label schema, bare driver sessions, and environment-key drift against registries built from the repository. A # gate-ok: comment exempts a line and the next. String literals and comments are blanked first (_strip_strings_and_comments), and code samples that carry Cypher as an inner literal are recognised so they are not flagged.

Options: positional paths, --output or -o (report file), --stdout, --min-severity (default WARNING) and --worklist-baseline (a JSON file whose rows are downgraded to INFO as named, non-blocking debt, while new findings keep their severity). With no test files it exits 0. Otherwise it exits 1 when any ERROR was found.

At a glance

Class diagram

Dependency diagram

No internal imports and no importers.

API

Constants

Classes

Severity (enum)

Bases: StrEnum · Line 178

TestIssue (dataclass)

Decorators: dataclass · Line 185

TestReport (dataclass)

Decorators: dataclass · Line 194

TestFileChecker

Line 221

AST-based checker for test file compliance.

Functions

Source

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

#!/usr/bin/env python3
"""Test file reviewer — checks orchestrata-ultimate test standards.

Usage:
    python test-review.py tests/test_foo.py
    python test-review.py --format text --stdout tests/test_foo.py
    python test-review.py end2end/harness/
"""

from __future__ import annotations

import argparse
import ast
import re
import sys
from collections.abc import Iterable
from dataclasses import dataclass, field
from enum import StrEnum
from pathlib import Path


def _strip_strings_and_comments(lines: list[str]) -> list[str]:
    """Return *lines* with every string literal and comment blanked out.

    Line-regex rules that scan raw source cannot tell a real call from the
    same text quoted inside a docstring or inside test DATA — modules that
    assert on code samples carry things like ``"return redis.Redis(host='x')"``
    as a plain string. Blanking those regions first keeps the rules matching
    code only. Tokenizing preserves line numbering: each token's span is
    replaced in place with spaces.

    Falls back to the original lines when the source does not tokenize (a
    syntax error is reported by other rules, not swallowed here).
    """
    import io
    import tokenize

    blanked = list(lines)
    try:
        tokens = list(tokenize.generate_tokens(io.StringIO("\n".join(lines)).readline))
    except tokenize.TokenError, IndentationError, SyntaxError:
        return blanked

    for tok in tokens:
        if tok.type not in (tokenize.STRING, tokenize.COMMENT):
            continue
        (srow, scol), (erow, ecol) = tok.start, tok.end
        for row in range(srow, erow + 1):
            idx = row - 1
            if idx >= len(blanked):
                break
            line = blanked[idx]
            start = scol if row == srow else 0
            end = ecol if row == erow else len(line)
            blanked[idx] = line[:start] + " " * (end - start) + line[end:]
    return blanked


# A Cypher WRITE is a clause applied to something, not the bare keyword.
# Matching the word alone (the form until 2026-08-06) fired on `'CREATE'` used
# as a plain operation-name value in test expectations, and on docstring prose
# such as "Fix: Use MERGE or DELETE existing node first" — the artifact-vs-
# description confusion again, in a rule that then reported the module as
# writing to the graph. CREATE/MERGE must be followed by a node/edge pattern;
# SET/DELETE/REMOVE by an identifier.
# Case-SENSITIVE on purpose: Cypher clauses are written uppercase in this
# codebase, and matching case-insensitively turns every ordinary Python
# `create(...)` / `.set(...)` call into a graph write (measured: T10 went
# 12 -> 31 findings the moment IGNORECASE was added).
_CYPHER_WRITE = re.compile(
    r"\b(CREATE|MERGE)\s*\("
    r"|\b(SET|DELETE|REMOVE)\s+[\w`]"
    r"|\bDROP\s+(INDEX|CONSTRAINT)\b"
)
# Calls that RECORD Cypher rather than execute it. A write-Cypher string handed
# to one of these is fixture DATA (written to a temp file for the tool under
# test to scan, or shelled out), never a graph write.
_CYPHER_DATA_SINKS = frozenset(
    {"write_text", "write_bytes", "write", "writelines", "dump", "dumps"}
)

# A string that is itself PYTHON SOURCE is a code sample handed to a checker,
# not a query. tests/scripts/test_code_review_tool.py feeds the reviewer whole
# function bodies like ``def test_wipe():\n    session.run("MATCH (n) DETACH
# DELETE n")`` -- the Cypher there is quoted INSIDE the sample and never runs.
_PY_SOURCE_MARKERS = re.compile(r"^\s*(?:def |class |import |from \w+ import )", re.M)


def _is_code_sample(text: str) -> bool:
    """True when ``text`` is Python source carrying Cypher as an inner literal."""
    return "\n" in text and bool(_PY_SOURCE_MARKERS.search(text))


# Generic verbs (set/delete/expire) need a redis-ish receiver — a bare
# `.set(...)` matches far too much. The rest are distinctive enough to match any
# receiver, and every one of them mutates.
_REDIS_WRITE = re.compile(
    r"\bredis\w*\.(set|delete|expire|rename|incr|decr)\("
    r"|\.(setex|psetex|getset|setnx|msetnx|mset|hset|hsetnx|hdel"
    r"|hincrby|lpush|rpush|lpop|rpop|linsert|lset|ltrim|sadd|srem"
    r"|spop|zadd|zrem|zincrby|pexpire|persist|flushdb|flushall"
    r"|xadd|xtrim|xdel)\(",
    re.I,
)


def _gate_ok_lines(lines: list[str]) -> set[int]:
    """Lines covered by a ``# gate-ok:`` declaration (the marker and the next).

    The project's established way of saying "this Cypher is DATA, not a query
    anyone runs". Six gates honour it; this one did not, so a module whose
    fixtures are Cypher STRINGS the tooling scans was reported as writing to
    the graph.
    """
    marked = {n for n, line in enumerate(lines, start=1) if "gate-ok:" in line}
    return marked | {n + 1 for n in marked}


def _has_graph_writes(code: list[str], tree: ast.AST, raw: list[str]) -> bool:
    """True when the module both calls the driver and carries write Cypher.

    Requiring both keeps read-only query modules out: the call alone says
    nothing about whether anything is mutated.

    ``code`` is the comment/string-blanked source used for the call scan;
    ``raw`` is the original, because the ``# gate-ok:`` markers live in
    comments and are blanked out of ``code`` by construction.
    """
    exempt = _gate_ok_lines(raw)
    return any(
        _executes_cypher_write(call, exempt)
        for call in ast.walk(tree)
        if isinstance(call, ast.Call)
    )


def _executes_cypher_write(call: ast.Call, exempt: set[int]) -> bool:
    """True when write-Cypher is passed INTO ``call`` and ``call`` runs it.

    Keyed on where the Cypher GOES, not on what the receiver is named. Two
    earlier forms both failed on this codebase:

    * matching a bare ``.run(`` classed ``subprocess.run(...)`` as a driver call,
      so a module whose Cypher is fixture text (test_graph_test_preflight.py)
      read as writing to the graph;
    * allow-listing driver method names missed project helpers -- ``session.run``
      reached through ``def w(cypher)`` (test_requirement_resolution.py) and
      ``syncer._run("MERGE ...")`` (kg_sync/test_stage_session.py) are real
      writes under names no allow-list predicts.

    Passing Cypher to a call means something executes it, unless the callee is a
    known data sink or a subprocess spawn.
    """
    func = call.func
    name = func.attr if isinstance(func, ast.Attribute) else getattr(func, "id", "")
    if name in _CYPHER_DATA_SINKS:
        return False
    if (
        isinstance(func, ast.Attribute)
        and getattr(func.value, "id", "") == "subprocess"
    ):
        return False
    return any(
        isinstance(node, ast.Constant)
        and isinstance(node.value, str)
        and _CYPHER_WRITE.search(node.value)
        and not _is_code_sample(node.value)
        and getattr(node, "lineno", 0) not in exempt
        for arg in [*call.args, *(kw.value for kw in call.keywords)]
        for node in ast.walk(arg)
    )


def _has_redis_writes(code: list[str]) -> bool:
    return any(_REDIS_WRITE.search(line) for line in code)


class Severity(StrEnum):
    ERROR = "ERROR"
    WARNING = "WARNING"
    INFO = "INFO"


@dataclass
class TestIssue:
    file: str
    line: int
    rule: str
    message: str
    severity: Severity


@dataclass
class TestReport:
    files_analyzed: int = 0
    issues: list[TestIssue] = field(default_factory=list)

    def add(
        self, file: str, line: int, rule: str, message: str, severity: Severity
    ) -> None:
        self.issues.append(TestIssue(file, line, rule, message, severity))


def find_test_files(paths: list[str]) -> list[Path]:
    files: list[Path] = []
    for p in paths:
        path = Path(p)
        if path.is_file() and path.suffix == ".py":
            # Always include explicit .py files (for T21 env-key rule)
            files.append(path)
        elif path.is_dir():
            # Directories only include test-named files (T1-T20 scoping)
            files.extend(
                f
                for f in path.rglob("*.py")
                if f.name.startswith("test_") or f.name.endswith("_test.py")
            )
    return sorted(set(files))


class TestFileChecker:
    """AST-based checker for test file compliance."""

    RULES = {
        "T1": "No mocking — no unittest.mock, MagicMock, AsyncMock, Mock, patch",
        "T2": "No stubs or fakes — no FakeX, StubX, DummyX classes",
        "T3": "No try/except/pass blocks",
        "T4": "Function-scope fixtures only — module scope solely for annotated IMMUTABLE precomputed results ('# T4-immutable-module-scope:'), class/session never",
        "T5": "No os.environ mutations",
        "T6": "No hardcoded ports",
        "T7": "No pytest.mark.skip or skipif",
        "T8": "No monkeypatch usage",
        "T9": "No global mutable state at module level",
        "T10": "Use test_id fixture for DB/graph (Memgraph) ops",
        "T11": "Use isolation fixture for file/DB resources",
        "T12": "No bare os.environ['X'] reads — use os.environ.get()",
        "T13": "No relative imports",
        "T14": "Async tests need pytest.mark.asyncio or asyncio_mode",
        "T15": "No pytest-xdist (-n/--numprocesses/--dist) — it deadlocks this suite; use the threadpool runner",
        "T16": "No banned-library imports — see .claude/banned-libraries.txt",
        "T17": "A uuid-scoped Memgraph chunk_label must drop its schema in teardown — clear() leaves the constraint + indexes behind forever",
        "T18": "Verify through the store/repo's own session() — a bare `.driver.session()` reads the cluster default database, not the one under test",
        "T19": "DDL a test CREATEs, the test must DROP — node sweeps delete nodes only, so a constraint/index outlives the run forever and is replayed at boot recovery",
        "T21": "Env key drift/fabrication — compare string literals against declared constants (names ending in _ENV); flag similar-but-not-equal keys (ERROR) and exact literals outside declaring module (WARNING)",
    }

    @staticmethod
    def _is_test_named(path: Path) -> bool:
        """True when path is test-named (test_*.py or *_test.py).

        This predicate matches find_test_files' directory-scanning logic and
        determines which files are subject to T1-T20 rules. The T21 env-key
        rule runs on ALL explicit .py files regardless of naming.
        """
        return path.name.startswith("test_") or path.name.endswith("_test.py")

    def __init__(
        self, report: TestReport, registry_roots: list[Path] | None = None
    ) -> None:
        self.report = report
        # Nodes of the file under check — populated once per file by
        # check_file() and read by every rule (see the note there).
        self._nodes: list[ast.AST] = []
        # Banned distribution names (.claude/banned-libraries.txt), discovered
        # lazily from the first test file's repo. None = not yet loaded.
        self._banned_libs: list[str] | None = None
        # Registry of declared env key constants, built once per review run.
        # If registry_roots is None, use the default src/orchestrata_ultimate + scripts.
        self._registry_roots = registry_roots
        self._env_registry: dict[str, tuple[str, str, int]] | None = None
        self._general_registry: dict[str, tuple[str, str, int]] | None = None

    def check_file(self, path: Path) -> None:
        self.report.files_analyzed += 1
        source = path.read_text(encoding="utf-8")
        lines = source.splitlines()

        try:
            tree = ast.parse(source)
        except SyntaxError as exc:
            self.report.add(
                str(path),
                exc.lineno or 1,
                "SYNTAX",
                f"SyntaxError: {exc.msg}",
                Severity.ERROR,
            )
            return

        # ONE walk per file, shared by every check. Each check used to call
        # ast.walk(tree) itself — 17 full traversals of the same tree, on
        # every one of ~1550 test modules. Measured 2026-08-06: 36.07s wall /
        # 35.69s CPU over tests/ before this cache.
        self._nodes = list(ast.walk(tree))

        # T1-T20 rules are gated behind test-name scoping (same predicate as
        # find_test_files for directory scanning). T21 (env-key rule) runs on
        # ALL explicit .py files regardless of naming.
        if self._is_test_named(path):
            self._check_imports(path, tree, lines)
            self._check_fixtures(path, tree, lines)
            self._check_skips(path, tree, lines)
            self._check_mocking(path, tree, lines)
            self._check_try_except_pass(path, tree, lines)
            doc_lines = self._string_stmt_lines(tree)
            self._check_global_state(path, tree, lines)
            self._check_monkeypatch(path, tree, lines)
            self._check_os_environ(path, tree, lines)
            self._check_hardcoded_ports(path, lines, doc_lines)
            self._check_no_xdist(path, tree)
            self._check_banned_libraries(path, lines, doc_lines)
            self._check_test_id_usage(path, tree, lines)
            self._check_relative_imports(path, tree, lines)
            self._check_async_tests(path, tree, lines)
            self._check_anyio_marks(path)
            self._check_chunk_label_schema_teardown(path, tree)
            self._check_ddl_teardown(path, tree)
            self._check_bare_driver_session(path, tree)
        self._check_env_key_drift(path, tree, lines)

    # ------------------------------------------------------------------
    # Individual checks
    # ------------------------------------------------------------------

    def _string_stmt_lines(self, tree: ast.AST) -> set[int]:
        """Line numbers spanned by bare string-literal statements — i.e. module/
        class/function docstrings and standalone string blocks. Trigger words
        (monkeypatch, ``-n auto``) inside these are documentation, not code, so
        the text-based checks skip them. Assignments/calls that embed a string
        (e.g. ``cmd = "pytest -n auto"``) are NOT bare Expr statements and remain
        checked, so real usage is still caught.
        """
        out: set[int] = set()
        for node in self._nodes:
            if (
                isinstance(node, ast.Expr)
                and isinstance(node.value, ast.Constant)
                and isinstance(node.value.value, str)
            ):
                end = getattr(node, "end_lineno", node.lineno) or node.lineno
                out.update(range(node.lineno, end + 1))
        return out

    def _check_imports(self, path: Path, tree: ast.AST, lines: list[str]) -> None:
        for node in self._nodes:
            if isinstance(node, ast.Import):
                for alias in node.names:
                    name = alias.name
                    if name in ("unittest.mock", "mock"):
                        self._add(
                            path,
                            node,
                            "T1",
                            f"Import of '{name}' is forbidden (no mocking)",
                        )
            elif isinstance(node, ast.ImportFrom):
                module = node.module or ""
                if "unittest.mock" in module or "mock" in module.split("."):
                    self._add(
                        path,
                        node,
                        "T1",
                        f"Import from '{module}' is forbidden (no mocking)",
                    )
                names = [a.name for a in node.names]
                for n in names:
                    if n in ("MagicMock", "AsyncMock", "Mock", "patch", "mocker"):
                        self._add(
                            path,
                            node,
                            "T1",
                            f"Import of '{n}' is forbidden (no mocking)",
                        )

    def _check_fixtures(self, path: Path, tree: ast.AST, lines: list[str]) -> None:
        # T4 harmonized with the CLAUDE.md Key Rule wording ("no module/class
        # scope for MUTABLE state"): scope='module' is allowed ONLY for an
        # immutable precomputed result, opted in explicitly with a
        # '# T4-immutable-module-scope: <why>' comment directly above the
        # decorator AND a return-only body (no yield — a teardown implies a
        # lifecycle, which implies state). class/session scope stay forbidden
        # unconditionally.
        for node in self._nodes:
            if not isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef):
                continue
            for dec in node.decorator_list:
                if not isinstance(dec, ast.Call):
                    continue
                func = dec.func
                name = (
                    func.attr
                    if isinstance(func, ast.Attribute)
                    else getattr(func, "id", "")
                )
                if name != "fixture":
                    continue
                for kw in dec.keywords:
                    if kw.arg == "scope" and isinstance(kw.value, ast.Constant):
                        # ast.Constant.value is Any, so a bytes literal
                        # (scope=b"session") would render as "b'session'" in
                        # the message below. Only a str scope is meaningful to
                        # pytest; anything else cannot name a forbidden scope.
                        if not isinstance(kw.value.value, str):
                            continue
                        scope: str = kw.value.value
                        if scope == "session" and self._gate_ok(node, lines):
                            # Session-scoped CONNECTION infra (the conftest
                            # neo4j_driver precedent) carries no mutable test
                            # state and is what makes the shared-index test
                            # files affordable. gate_parallel_safety.py already
                            # accepts a reasoned '# gate-ok:' waiver on exactly
                            # these lines; honouring the SAME marker here keeps
                            # one waiver vocabulary across both reviewers
                            # instead of one tool contradicting the other.
                            continue
                        if scope in ("class", "session"):
                            self._add(
                                path,
                                node,
                                "T4",
                                f"Fixture with scope='{scope}' is forbidden — use function scope only"
                                + (
                                    " (session-scoped CONNECTION infra may opt out with a reasoned '# gate-ok: <why>' comment directly above the decorator)"
                                    if scope == "session"
                                    else ""
                                ),
                            )
                        elif scope == "module" and not self._t4_immutable_ok(
                            node, lines
                        ):
                            self._add(
                                path,
                                node,
                                "T4",
                                "Fixture with scope='module' is forbidden — use function scope, or opt in an IMMUTABLE precomputed result with '# T4-immutable-module-scope: <why>' directly above the decorator (return-only, no yield)",
                            )

    @staticmethod
    def _gate_ok(
        node: ast.FunctionDef | ast.AsyncFunctionDef, lines: list[str]
    ) -> bool:
        """True when a reasoned ``# gate-ok: <why>`` sits above the decorator.

        Same marker and same "a reason is mandatory, a bare marker does not
        waive" rule as ``scripts/gate_parallel_safety.py`` (``_WAIVER_RE``), so
        one annotation satisfies both reviewers rather than each tool inventing
        its own vocabulary.
        """
        first_line = min([d.lineno for d in node.decorator_list] + [node.lineno])
        for offset in (2, 3):  # the line above the decorator, and the one above that
            if first_line >= offset:
                if re.search(r"#\s*gate-ok:\s*\S", lines[first_line - offset]):
                    return True
        return False

    @staticmethod
    def _line_gate_ok(lineno: int, lines: list[str]) -> bool:
        """Reasoned ``# gate-ok: <why>`` on the 1-based *lineno* or the line
        above — the adjacency rule ``test_isolation_audit.py::_is_waived``
        uses, so per-line waivers are portable between the two gates.
        """
        for idx in (lineno - 1, lineno - 2):
            if 0 <= idx < len(lines) and re.search(r"#\s*gate-ok:\s*\S", lines[idx]):
                return True
        return False

    @staticmethod
    def _t4_immutable_ok(
        node: ast.FunctionDef | ast.AsyncFunctionDef, lines: list[str]
    ) -> bool:
        """The module-scope opt-in: annotation comment + return-only body."""
        first_line = min([d.lineno for d in node.decorator_list] + [node.lineno])
        above = lines[first_line - 2] if first_line >= 2 else ""
        if "T4-immutable-module-scope:" not in above:
            return False
        return not any(isinstance(n, ast.Yield | ast.YieldFrom) for n in ast.walk(node))

    def _check_skips(self, path: Path, tree: ast.AST, lines: list[str]) -> None:
        for node in self._nodes:
            if isinstance(node, ast.Attribute):
                if node.attr in ("skip", "skipif"):
                    self._add(
                        path, node, "T7", f"Use of pytest.mark.{node.attr} is forbidden"
                    )

    def _check_mocking(self, path: Path, tree: ast.AST, lines: list[str]) -> None:
        for node in self._nodes:
            if isinstance(node, ast.Name):
                if node.id in ("MagicMock", "AsyncMock", "Mock"):
                    self._add(
                        path,
                        node,
                        "T1",
                        f"Use of '{node.id}' is forbidden (no mocking)",
                    )
            elif isinstance(node, ast.ClassDef):
                if any(node.name.startswith(p) for p in ("Fake", "Stub", "Dummy")):
                    self._add(
                        path,
                        node,
                        "T2",
                        f"Class '{node.name}' appears to be a stub/fake (no stubs/fakes)",
                    )

    def _check_try_except_pass(
        self, path: Path, tree: ast.AST, lines: list[str]
    ) -> None:
        for node in self._nodes:
            if isinstance(node, ast.Try):
                for handler in node.handlers:
                    if len(handler.body) == 1 and isinstance(handler.body[0], ast.Pass):
                        self._add(
                            path, handler, "T3", "try/except/pass block is forbidden"
                        )

    # Names that are shared mutable state whatever they are bound to.
    _STATE_NAMES = frozenset({"_cache", "_state", "_registry", "_instances"})

    # Names RESERVED BY PYTEST at module level — pytest's own documented API,
    # not state this suite owns. ``pytestmark`` is read by the collector at
    # import time to apply markers to every test in the module; nothing
    # mutates it, and nothing carries data between tests through it. The
    # mutable-container heuristic below could not tell pytest's protocol from
    # a suite-authored list and flagged all 62 modules that use the idiom —
    # a rule with no model of the framework it reviews.
    _PYTEST_RESERVED = frozenset(
        {"pytestmark", "collect_ignore", "collect_ignore_glob"}
    )

    @staticmethod
    def _mutable_literal(node: ast.AST) -> bool:
        """ATOM — the expression constructs a mutable container."""
        if isinstance(node, (ast.List, ast.Dict, ast.Set, ast.ListComp, ast.DictComp)):
            return True
        return (
            isinstance(node, ast.Call)
            and isinstance(node.func, ast.Name)
            and node.func.id in ("list", "dict", "set", "defaultdict", "Counter")
        )

    @classmethod
    def _is_module_state(cls, name: str, value: ast.AST) -> bool:
        """FACT — a module-level binding that is STATE rather than a constant.

        Two ways to qualify: a known shared-state name, or a mutable container
        bound to a non-UPPER_CASE name. The name list alone caught exactly four
        spellings and missed every other module-level mutable — UPPER_CASE is
        the project's constant convention, so anything else holding a mutable
        container is state.
        """
        if name in cls._PYTEST_RESERVED:
            return False
        return name in cls._STATE_NAMES or (
            not name.isupper() and cls._mutable_literal(value)
        )

    def _check_global_state(
        self, path: Path, tree: ast.Module, lines: list[str]
    ) -> None:
        """T9 — no module-level mutable state."""
        bindings = (
            (node, target.id)
            for node in tree.body
            if isinstance(node, ast.Assign)
            for target in node.targets
            if isinstance(target, ast.Name)
        )
        for node, name in bindings:
            if self._is_module_state(name, node.value):
                self._add(
                    path,
                    node,
                    "T9",
                    f"Module-level mutable state '{name}' is forbidden",
                )

    def _check_monkeypatch(self, path: Path, tree: ast.AST, lines: list[str]) -> None:
        """AST-based: only a real ``monkeypatch.<attr>`` access on a Name
        literally bound to ``monkeypatch`` counts. A line-based/regex scan
        (the prior implementation) false-positived on any line merely
        *containing* the substring "monkeypatch" — e.g. an identifier like
        a fixture named ``monkeypatch_free_chdir`` (declaring it does NOT
        use monkeypatch), or a string literal used as test data for a
        meta-test of a pattern-detector (e.g. ``lines = ["monkeypatch.setenv(...)\\n"]``
        fed into ``verification.review.checks_safety`` or this very tool's
        own test suite). Walking ``ast.Attribute`` nodes only matches actual
        attribute access in real code, never substrings or string contents.
        """
        for node in self._nodes:
            if not isinstance(node, ast.Attribute):
                continue
            if not (
                isinstance(node.value, ast.Name) and node.value.id == "monkeypatch"
            ):
                continue
            # monkeypatch.setenv/delenv is allowed in tests per AGENTS.md.
            if node.attr in ("setenv", "delenv"):
                continue
            self._add(
                path,
                node,
                "T8",
                "Use of monkeypatch is forbidden (only monkeypatch.setenv/delenv allowed)",
            )

    def _check_os_environ(self, path: Path, tree: ast.AST, lines: list[str]) -> None:
        """AST-based: only a real ``os.environ[...]`` subscript — an
        ``ast.Subscript`` node whose value is the attribute access
        ``os.environ`` — counts. A text/regex scan over raw lines (the prior
        implementation) false-positived on any line merely *containing* the
        substring, including inert string-literal fixture data belonging to
        OTHER tools' tests — e.g. Python source text fed to
        ``verification.review.checks_safety.check_env_access`` via
        ``ast.parse(source)``, which is not real code in THIS file, just a
        string constant. Walking ``ast.Subscript`` nodes only matches genuine
        subscript syntax; a string can never masquerade as one.

        A T5 mutation accepts a reasoned ``# gate-ok: <why>`` on the flagged
        line or the line above — the SAME waiver channel (marker, mandatory
        reason, adjacency rule) as ``test_isolation_audit.py``'s
        OS_ENVIRON_MUTATION check, so one annotation satisfies both gates.
        Two gates disagreeing about the same save/restore seam is worse than
        either being absent (see the T4 cross-reference note above).
        """
        for node in self._nodes:
            if not isinstance(node, ast.Subscript):
                continue
            if not self._is_os_environ(node.value):
                continue
            if isinstance(node.ctx, ast.Store | ast.Del):
                if self._line_gate_ok(node.lineno, lines):
                    continue
                self._add(path, node, "T5", "Mutation of os.environ[...] is forbidden")
            else:
                self._add(
                    path,
                    node,
                    "T12",
                    "Use os.environ.get(...) instead of os.environ[...]",
                    severity=Severity.WARNING,
                )

    @staticmethod
    def _is_os_environ(node: ast.AST) -> bool:
        return (
            isinstance(node, ast.Attribute)
            and node.attr == "environ"
            and isinstance(node.value, ast.Name)
            and node.value.id == "os"
        )

    def _check_hardcoded_ports(
        self, path: Path, lines: list[str], doc_lines: set[int]
    ) -> None:
        for i, line in enumerate(lines, 1):
            if i in doc_lines or line.lstrip().startswith("#"):
                continue
            if re.search(r"port\s*=\s*5\d{4}", line):
                self.report.add(
                    str(path),
                    i,
                    "T6",
                    "Hardcoded port detected — use dynamic allocation",
                    Severity.ERROR,
                )

    # SHOWSTOPPER: pytest-xdist (-n / --numprocesses / --dist) deadlocks this
    # suite (~78% stall). Use the threadpool runner instead
    # (scripts/run_tests_threadpool.py / core.threading.test_executor).
    _XDIST_TEXT_RE = re.compile(
        r"--dist\b|--numprocesses\b|pytest-xdist|(?<![\w-])-n[ =]*(auto|\d)"
    )
    # Callables that actually spawn/execute a command — the only place an
    # xdist CLI flag could ever really run pytest in xdist mode.
    _INVOCATION_ATTRS = {
        "run",
        "call",
        "check_call",
        "check_output",
        "Popen",
        "system",
        "popen",
    }

    def _check_no_xdist(self, path: Path, tree: ast.AST) -> None:
        """AST-based: only string arguments to an actual process invocation
        (``subprocess.run``/``Popen``/``call``/..., ``os.system``, or
        ``pytest.main``) count. A text/regex scan over raw lines (the prior
        implementation) false-positived on inert string-literal fixture data
        belonging to OTHER tools' tests — e.g. a fake ``.gitlab-ci.yml`` body
        ("head -n 100", "--dist-dir=dist") written via ``write_text`` for
        scripts/validate_ci.py's own gate tests, which is YAML text, never
        actually invoked as a command here. Restricting to real invocation
        call sites means a string can never masquerade as a genuine one.
        """
        for node in self._nodes:
            if not isinstance(node, ast.Call) or not self._is_invocation_call(
                node.func
            ):
                continue
            command = " ".join(self._iter_call_string_args(node))
            if not command or "ORCHESTRATA_WORKER_SLOT" in command:
                continue
            if self._XDIST_TEXT_RE.search(command):
                self._add(
                    path,
                    node,
                    "T15",
                    "pytest-xdist (-n/--numprocesses/--dist) is forbidden — it deadlocks this "
                    "suite (~78%); use scripts/run_tests_threadpool.py (core.threading.test_executor)",
                )

    @classmethod
    def _is_invocation_call(cls, func: ast.expr) -> bool:
        if isinstance(func, ast.Attribute):
            if func.attr in cls._INVOCATION_ATTRS:
                return True
            return (
                func.attr == "main"
                and isinstance(func.value, ast.Name)
                and func.value.id == "pytest"
            )
        if isinstance(func, ast.Name):
            return func.id in cls._INVOCATION_ATTRS
        return False

    @staticmethod
    def _iter_call_string_args(node: ast.Call) -> list[str]:
        out: list[str] = []

        def collect(n: ast.AST) -> None:
            if isinstance(n, ast.Constant) and isinstance(n.value, str):
                out.append(n.value)
            elif isinstance(n, ast.List | ast.Tuple):
                for elt in n.elts:
                    collect(elt)

        for arg in node.args:
            collect(arg)
        for kw in node.keywords:
            if kw.value is not None:
                collect(kw.value)
        return out

    def _load_banned_libs(self, anchor: Path) -> list[str]:
        """Find and parse .claude/banned-libraries.txt by walking up from a test
        file (works whether test-review runs in-repo or globally-installed).
        Cached after the first lookup."""
        if self._banned_libs is not None:
            return self._banned_libs
        self._banned_libs = []
        for parent in [anchor.resolve(), *anchor.resolve().parents]:
            candidate = parent / ".claude" / "banned-libraries.txt"
            if candidate.exists():
                self._banned_libs = [
                    ln.strip()
                    for ln in candidate.read_text(encoding="utf-8").splitlines()
                    if ln.strip() and not ln.strip().startswith("#")
                ]
                break
        return self._banned_libs

    def _check_banned_libraries(
        self, path: Path, lines: list[str], doc_lines: set[int]
    ) -> None:
        # A test file must not import a banned library. Derives the import module
        # from each banned distribution name (pytest-xdist -> xdist). xdist's
        # command flags are already covered by T15; this catches the import.
        banned = self._load_banned_libs(path)
        if not banned:
            return
        import_names = {b.removeprefix("pytest-").replace("-", "_"): b for b in banned}
        pattern = re.compile(
            r"^\s*(?:import|from)\s+("
            + "|".join(re.escape(n) for n in import_names)
            + r")\b"
        )
        # Scan with string literals and comments blanked (same treatment as
        # T10 below): an ``import xdist`` inside a payload string is test DATA
        # that never runs — gate-test fixtures carry banned source as strings.
        code = _strip_strings_and_comments(lines)
        for i, line in enumerate(code, 1):
            if i in doc_lines or line.lstrip().startswith("#"):
                continue
            m = pattern.match(line)
            if m:
                dist = import_names[m.group(1)]
                self.report.add(
                    str(path),
                    i,
                    "T16",
                    f"banned library '{dist}' imported — see "
                    ".claude/banned-libraries.txt",
                    Severity.ERROR,
                )

    def _check_test_id_usage(self, path: Path, tree: ast.AST, lines: list[str]) -> None:
        """T10 — a module that WRITES to Redis/Memgraph must namespace by test_id.

        The point of ``test_id`` is collision-freedom between parallel workers
        sharing one real database, so only WRITES can violate it. Reads have
        nothing to namespace, and three things used to be flagged that cannot
        collide with anything:

        - read-only Cypher (``session.run("RETURN 1 AS ok")``);
        - ``.pipeline(`` on any receiver — this project's own ``PipelineRunner``
          and volk pipelines match it far more often than redis-py does;
        - ``redis.Redis(`` appearing inside a docstring or a string literal
          that IS the test data (modules that assert on code samples).

        The scan therefore runs over source with strings and comments blanked
        out, and requires an actual write: a redis write command, or a Cypher
        write clause in one of the module's string constants.
        """
        code = _strip_strings_and_comments(lines)
        writes = _has_graph_writes(code, tree, lines) or _has_redis_writes(code)

        # `test_id` is one way to namespace; a per-test `uuid4()` prefix is the
        # other, and it buys the same collision-freedom. Checking only for the
        # fixture NAME flagged modules that already isolate correctly.
        has_namespacing = any(
            "test_id" in line or re.search(r"\buuid\d?\(|\buuid\.uuid\d\(", line)
            for line in code
        )
        has_conftest = path.name == "conftest.py"
        if writes and not has_namespacing and not has_conftest:
            self.report.add(
                str(path),
                1,
                "T10",
                "File WRITES to Redis/Memgraph without namespacing — use the "
                "test_id fixture or a per-test uuid4() prefix",
                Severity.WARNING,
            )

    def _check_relative_imports(
        self, path: Path, tree: ast.AST, lines: list[str]
    ) -> None:
        for node in self._nodes:
            if isinstance(node, ast.ImportFrom):
                if node.level > 0:
                    self._add(
                        path,
                        node,
                        "T13",
                        "Relative import is forbidden — use absolute imports",
                    )

    @staticmethod
    def _anyio_mark(node: ast.AST) -> ast.AST | None:
        """Return the `pytest.mark.anyio` decorator on *node*, or None.

        Matches both the bare form (`@pytest.mark.anyio`) and the called form
        (`@pytest.mark.anyio(...)`), on functions and on classes.
        """
        decorators = getattr(node, "decorator_list", [])
        for dec in decorators:
            target = dec.func if isinstance(dec, ast.Call) else dec
            if isinstance(target, ast.Attribute) and target.attr == "anyio":
                inner = target.value
                if isinstance(inner, ast.Attribute) and inner.attr == "mark":
                    return dec
        return None

    def _check_anyio_marks(self, path: Path) -> None:
        """T20 — no `@pytest.mark.anyio` while asyncio_mode is configured.

        Under `asyncio_mode = auto`, pytest-asyncio already owns every async
        test and async fixture. Adding the anyio mark puts two plugins in the
        same regime, and they do not agree about which event loop a FIXTURE
        runs on versus the TEST that consumes it. That split is what broke
        tests/gateway/handlers/test_projects.py on Python 3.14.6t (backlog
        P2.14); 55 further files carried the same latent shape and were
        de-marked. This rule is why they stay de-marked.
        """
        if not self._has_asyncio_mode():
            return
        for node in self._nodes:
            if not isinstance(
                node, ast.AsyncFunctionDef | ast.FunctionDef | ast.ClassDef
            ):
                continue
            dec = self._anyio_mark(node)
            if dec is not None:
                self._add(
                    path,
                    dec,
                    "T20",
                    "@pytest.mark.anyio is forbidden — asyncio_mode is configured, so "
                    "pytest-asyncio owns async tests and fixtures; running both plugins "
                    "splits the fixture loop from the test loop (P2.14). Delete the mark.",
                )

    def _check_async_tests(self, path: Path, tree: ast.AST, lines: list[str]) -> None:
        # Check project-wide asyncio_mode config
        has_asyncio_mode = self._has_asyncio_mode()
        for node in self._nodes:
            if isinstance(node, ast.AsyncFunctionDef):
                if node.name.startswith("test_"):
                    # Check if decorated with pytest.mark.asyncio
                    has_marker = any(
                        isinstance(d, ast.Call)
                        and isinstance(d.func, ast.Attribute)
                        and d.func.attr == "asyncio"
                        for d in node.decorator_list
                    )
                    if not has_marker and not has_asyncio_mode:
                        self._add(
                            path,
                            node,
                            "T14",
                            f"Async test '{node.name}' needs @pytest.mark.asyncio or asyncio_mode in config",
                        )

    def _check_chunk_label_schema_teardown(self, path: Path, tree: ast.AST) -> None:
        """T17 — a per-test Memgraph ``chunk_label`` must drop its schema.

        ``connect()`` creates a uniqueness constraint and two lookup indexes
        for whatever ``chunk_label`` names. ``clear()``/``clear_prefixed()``
        delete NODES only; the schema survives, and Memgraph replays every
        entry at boot recovery. When the label carries the ``test_id`` uuid
        that is unbounded growth: 3,379 leaked ``*__TEST_*`` labels (6,713
        indexes + 3,323 constraints guarding 9 nodes) cost the shared cluster
        ~17 minutes of pegged CPU per data-node restart in the 2026-07-15
        incident, and two fixtures had regrown the same leak by 2026-07-26.

        Fixed per-WORKER labels (``f"TestChunk{worker_slot}"``) are bounded by
        the worker count and deliberately keep their schema so the warm-run
        ``connect()`` short-circuit stays effective — this rule only fires on
        labels interpolating ``test_id``.
        """
        for node in self._nodes:
            if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                continue

            uuid_scoped_label: ast.AST | None = None
            for sub in ast.walk(node):
                if not isinstance(sub, ast.keyword) or sub.arg != "chunk_label":
                    continue
                if not isinstance(sub.value, ast.JoinedStr):
                    continue
                names = {
                    v.value.id
                    for v in sub.value.values
                    if isinstance(v, ast.FormattedValue)
                    and isinstance(v.value, ast.Name)
                }
                if "test_id" in names:
                    uuid_scoped_label = sub.value
                    break

            if uuid_scoped_label is None:
                continue

            # Only functions that OWN the graph's lifecycle can drop its
            # schema. A pure config factory (builds a SemanticGraphConfig and
            # returns it) has no teardown to put the drop in -- its consumers
            # do, and they are checked on their own. Ownership evidence: the
            # function yields the resource, or calls a lifecycle method on it.
            attrs = {
                sub.attr for sub in ast.walk(node) if isinstance(sub, ast.Attribute)
            }
            owns_lifecycle = bool(
                attrs & {"clear", "clear_prefixed", "close", "connect"}
            ) or any(
                isinstance(sub, (ast.Yield, ast.YieldFrom)) for sub in ast.walk(node)
            )
            if not owns_lifecycle:
                continue

            drops = "drop_chunk_label_schema" in attrs
            if not drops:
                self._add(
                    path,
                    node,
                    "T17",
                    f"'{node.name}' builds a uuid-scoped chunk_label but never calls "
                    "drop_chunk_label_schema() — its constraint + 2 indexes leak on "
                    "every run and Memgraph replays them at boot recovery",
                )

    # Cypher DDL as it appears inside a test's query strings. The label token
    # is captured VERBATIM: for an interpolated label the capture is the
    # placeholder itself (``{label}``), which is precisely what lets a CREATE
    # pair with its DROP without resolving the expression.
    _DDL_CREATE = re.compile(
        r"CREATE\s+CONSTRAINT\s+ON\s*\(\s*\w+\s*:\s*(?P<label>[^)\s]+)"
        r"|CREATE\s+(?:\w+\s+)?INDEX\s+ON\s*:\s*(?P<ilabel>[^(\s]+)",
        re.IGNORECASE,
    )
    _DDL_DROP = re.compile(
        r"DROP\s+CONSTRAINT\s+ON\s*\(\s*\w+\s*:\s*(?P<label>[^)\s]+)"
        r"|DROP\s+(?:\w+\s+)?INDEX\s+ON\s*:\s*(?P<ilabel>[^(\s]+)",
        re.IGNORECASE,
    )

    # Cheap prefilter: only strings mentioning CREATE or DROP can carry DDL.
    _DDL_WORD = re.compile(r"CREATE|DROP", re.IGNORECASE)

    @staticmethod
    def _norm_label(token: str) -> str:
        """Label token, comparable between a CREATE and its DROP."""
        return token.strip().strip("`").strip()

    @classmethod
    def _token(cls, match: re.Match[str]) -> str:
        """Normalised label from a DDL match, whichever alternative fired."""
        return cls._norm_label(match.group("label") or match.group("ilabel") or "")

    @staticmethod
    def _fstring_piece(value: ast.AST) -> str:
        """One f-string segment as text, an interpolation as ``{name}``."""
        if isinstance(value, ast.Constant) and isinstance(value.value, str):
            return value.value
        if not isinstance(value, ast.FormattedValue):
            return ""
        inner = value.value
        if isinstance(inner, ast.Name):
            return "{" + inner.id + "}"
        if isinstance(inner, ast.Attribute):
            return "{" + inner.attr + "}"
        return "{?}"

    @classmethod
    def _query_text(cls, node: ast.AST) -> str | None:
        """Literal text of a string node, f-string placeholders preserved.

        An interpolated label survives as ``{name}`` rather than being lost,
        so ``f"...ON (n:{label})..."`` and its DROP produce the same token.
        """
        if isinstance(node, ast.Constant) and isinstance(node.value, str):
            return node.value
        if isinstance(node, ast.JoinedStr):
            return "".join(map(cls._fstring_piece, node.values))
        return None

    @staticmethod
    def _ids_under(roots: Iterable[ast.AST]) -> frozenset[int]:
        """Identity of every node in the subtrees rooted at ``roots``."""
        return frozenset(id(sub) for root in roots for sub in ast.walk(root))

    @staticmethod
    def _docstring(node: ast.AST) -> ast.Constant | None:
        """The node's docstring constant, when it has one."""
        body = getattr(node, "body", None)
        if not isinstance(body, list) or not body:
            return None
        first = body[0]
        if (
            isinstance(first, ast.Expr)
            and isinstance(first.value, ast.Constant)
            and isinstance(first.value.value, str)
        ):
            return first.value
        return None

    def _check_ddl_teardown(self, path: Path, tree: ast.AST) -> None:
        """T19 — DDL a test CREATEs, the test must DROP.

        ``CREATE CONSTRAINT`` / ``CREATE INDEX`` on a per-test label is
        permanent: node sweeps delete NODES only, the schema survives, and
        Memgraph replays every entry at boot recovery. T17 covers exactly one
        shape of this — the ``chunk_label=`` keyword — and structurally cannot
        see raw Cypher, which is how every other site writes it.

        Measured 2026-08-06: 57 leaked constraints across 7 databases on the
        test cluster, from five fixtures issuing raw ``CREATE CONSTRAINT`` with
        no drop. ``sweep_test_pollution.py`` contains no ``DROP CONSTRAINT`` at
        all, so nothing in the project could ever have cleaned them.

        Pairing is by label TOKEN, so an interpolated label pairs through its
        placeholder. DDL that a test merely ASSERTS ABOUT — a generator's
        expected output compared against a literal — is not executed DDL and
        is excluded; counting it would repeat the artifact-vs-description
        confusion this project has now measured eight times.
        """
        # Strings that DESCRIBE DDL rather than run it. Two shapes:
        # operands of an assert/comparison (a generator's expected output),
        # and docstrings — a fixture that documents the DDL it measured
        # ("CREATE INDEX ON :<label>(id) (2,514ms) ...") is prose, not a
        # statement. Missing the docstring case produced a false positive on
        # tests/semantic_graph/test_memgraph_adapter.py the first time this
        # rule ran: the ninth instance in one day of a scanner with no model
        # of which input is the artifact and which is prose about it
        # (concepts/gates-must-separate-artifact-from-description).
        nodes = list(self._nodes)

        described = self._ids_under(
            n for n in nodes if isinstance(n, (ast.Assert, ast.Compare))
        ) | frozenset(map(id, filter(None, (self._docstring(n) for n in nodes))))

        # An f-string's own literal pieces are Constants too; scoring them
        # separately would split "CREATE CONSTRAINT ON (n:" from its label.
        joined = [n for n in nodes if isinstance(n, ast.JoinedStr)]
        inner = self._ids_under(joined) - frozenset(map(id, joined))

        # DDL only counts when it reaches an execution seam. A test-review
        # corpus is Python source stored as DATA — this rule's own test file
        # embeds "CREATE CONSTRAINT ..." in string constants it writes to a
        # temp file — and a string that is never handed to a call executes
        # nothing.
        executed = self._ids_under(n for n in nodes if isinstance(n, ast.Call))

        texts = [
            (n, text)
            for n, text in ((n, self._query_text(n)) for n in nodes)
            if text and id(n) not in inner and self._DDL_WORD.search(text)
        ]

        dropped = frozenset(
            self._token(m) for _, text in texts for m in self._DDL_DROP.finditer(text)
        )
        created: dict[str, ast.AST] = {}
        for node, text in texts:
            if id(node) in described or id(node) not in executed:
                continue
            created.update(
                {
                    self._token(m): node
                    for m in self._DDL_CREATE.finditer(text)
                    if self._token(m) not in created
                }
            )

        # A DROP whose label is COMPUTED can drop anything, so it satisfies
        # every create in the file. The idiom is a teardown loop —
        # ``for label in (A, B): run(f"DROP CONSTRAINT ON (n:{label}) ...")``
        # — which creates by literal and drops by loop variable, so literal
        # pairing would flag correct teardown as a leak.
        wildcard_drop = any("{" in token for token in dropped)
        leaked = (
            {}
            if wildcard_drop
            else {
                label: node for label, node in created.items() if label not in dropped
            }
        )

        for label, node in sorted(leaked.items()):
            self._add(
                path,
                node,
                "T19",
                f"creates DDL on ':{label}' and never drops it — the schema "
                "outlives the test forever (node sweeps delete nodes only) and "
                "Memgraph replays it at every boot recovery; drop it in teardown",
            )

    def _check_bare_driver_session(self, path: Path, tree: ast.AST) -> None:
        """T18 — a raw ``<x>.driver.session()`` reads the WRONG database.

        Since the 2026-07-26 orchgraph cutover the operational KG and the
        shared semantic/RAG graph are two DATABASES on one cluster, so the
        URI no longer picks between them — the ``database=`` argument does.
        Stores and repositories thread it on every call; reaching past them
        to ``.driver.session()`` does not, so the session lands on the
        cluster default.

        This is a FALSE-GREEN shape, not a style nit: five sync/promotion
        test modules were verifying an empty database and passing anyway
        (measured 2026-07-26). Use the object's own ``session()`` —
        ``store.session()``, ``repo.session()`` — which binds the database
        the code under test actually writes to.
        """
        kg_operational = any(
            isinstance(n, ast.Name) and n.id in ("NEO4J_URI", "resolve_kg_database")
            for n in self._nodes
        )
        for node in self._nodes:
            if not isinstance(node, ast.Call):
                continue
            func = node.func
            # Shape: <anything>.driver.session(...)
            if not isinstance(func, ast.Attribute) or func.attr != "session":
                continue
            # Two shapes reach a raw driver: `<obj>.driver.session()` and a
            # bare driver held directly (`neo4j_cleanup_driver.session()`) —
            # the second is how a fixture verifying the wrong database
            # slipped past the first sweep.
            #
            # The bare-name shape is only flagged in modules that target the
            # OPERATIONAL KG (`kg_operational` below). Semantic-graph tests
            # use the same driver class against the cluster DEFAULT database,
            # which is correct for them — flagging those turned 0 findings
            # into 352.
            receiver = func.value
            if isinstance(receiver, ast.Attribute):
                if receiver.attr != "driver":
                    continue
            elif isinstance(receiver, ast.Name):
                if not receiver.id.endswith("driver") or not kg_operational:
                    continue
            else:
                continue
            # `database=` supplied explicitly is the caller doing it right.
            if any(kw.arg == "database" for kw in node.keywords):
                continue
            # A session running only CLUSTER-level statements has no database
            # to bind: `SHOW DATABASES` enumerates the cluster, and there is
            # no store or repository whose session() could scope it. Measured
            # 2026-08-06 on tests/core/test_worker_database_mapping.py, where
            # the raw driver is deliberate and correct.
            if self._only_cluster_level(tree, node):
                continue
            self._add(
                path,
                node,
                "T18",
                "bare `.driver.session()` reads the cluster default database, "
                "not the one under test — use the store/repo's own session()",
            )

    # Statements that address the CLUSTER, not a database within it. A
    # session running only these has nothing to scope: `database=` on a
    # `SHOW DATABASES` call would name one of the very things being listed.
    _CLUSTER_LEVEL = re.compile(
        r"^\s*(SHOW\s+(DATABASES|VERSION|REPLICATION\s+ROLE|STORAGE\s+INFO)"
        r"|CREATE\s+DATABASE|DROP\s+DATABASE|USE\s+DATABASE)\b",
        re.IGNORECASE,
    )

    def _only_cluster_level(self, tree: ast.AST, call: ast.Call) -> bool:
        """True when the ``with``-block around ``call`` runs cluster-level DDL only.

        Conservative by construction: a block that runs even one ordinary
        query, or whose statements cannot be read statically, is NOT exempt —
        so the false-green shape T18 exists to catch stays caught.
        """
        blocks = [
            node
            for node in self._nodes
            if isinstance(node, (ast.With, ast.AsyncWith))
            and any(item.context_expr is call for item in node.items)
        ]
        if not blocks:
            return False

        runs = [
            sub
            for stmt in blocks[0].body
            for sub in ast.walk(stmt)
            if isinstance(sub, ast.Call)
            and isinstance(sub.func, ast.Attribute)
            and sub.func.attr == "run"
        ]
        # A query with no readable literal cannot be proven cluster-level.
        texts = [self._query_text(r.args[0]) if r.args else None for r in runs]
        return (
            bool(texts)
            and all(t is not None for t in texts)
            and all(self._CLUSTER_LEVEL.match(t) for t in texts if t)
        )

    def _has_asyncio_mode(self) -> bool:
        """Check if project has asyncio_mode = 'auto' in pytest config."""
        project_root = Path("/home/mohamed/workspace/orchestrata-ultimate-architecture")
        for config_file in ("pyproject.toml", "pytest.ini", "setup.cfg"):
            p = project_root / config_file
            if p.exists():
                text = p.read_text()
                if "asyncio_mode" in text:
                    return True
        return False

    # ------------------------------------------------------------------
    # T21 — env key drift/fabrication detection
    # ------------------------------------------------------------------

    def _build_env_registry(
        self, exclude_file: Path | None = None
    ) -> dict[str, tuple[str, str, int]]:
        """Build the registry of declared env key constants.

        Walks all .py files under registry_roots (default: src/orchestrata_ultimate + scripts),
        collecting module-level assignments where target ends with '_ENV' and value is a string Constant.

        Excludes the file being reviewed (if provided) to avoid self-tripping when the file
        being checked also declares constants.

        Returns: dict mapping constant_value -> (constant_name, declaring_module, lineno)
        Cache built once per checker instance per review run.
        """
        if self._env_registry is not None:
            return self._env_registry

        roots = (
            self._registry_roots
            if self._registry_roots
            else [
                Path(__file__).resolve().parents[1] / "src" / "orchestrata_ultimate",
                Path(__file__).resolve().parents[1] / "scripts",
            ]
        )

        registry: dict[str, tuple[str, str, int]] = {}
        module_paths: list[Path] = []

        for root in roots:
            if not root.exists():
                continue
            module_paths.extend(root.rglob("*.py"))

        for path in module_paths:
            # Skip the file being reviewed
            if exclude_file and path == exclude_file:
                continue

            try:
                source = path.read_text(encoding="utf-8")
                tree = ast.parse(source)
            except SyntaxError:
                continue

            # Only check module-level assignments (in the body of the Module node)
            for node in tree.body:
                if not isinstance(node, ast.Assign):
                    continue
                for target in node.targets:
                    if not isinstance(target, ast.Name):
                        continue
                    name = target.id
                    if not name.endswith("_ENV"):
                        continue
                    if not isinstance(node.value, ast.Constant) or not isinstance(
                        node.value.value, str
                    ):
                        continue
                    value = node.value.value
                    registry[value] = (name, str(path), node.lineno)

        self._env_registry = registry
        return registry

    def _build_general_registry(
        self, exclude_file: Path | None = None
    ) -> dict[str, tuple[str, str, int]]:
        """Build the GENERAL constants registry (T17-GEN): every module-level
        UPPER_CASE string-constant assignment EXCEPT the ``*_ENV`` names (which
        the landed T21 tier owns). The general tier is exact-match WARNING only —
        the measured >=2-token drift over all UPPER_CASE values is ~100% FP, so
        no drift ERROR is emitted here (T17 Stage A ruling)."""
        if self._general_registry is not None:
            return self._general_registry

        roots = (
            self._registry_roots
            if self._registry_roots
            else [
                Path(__file__).resolve().parents[1] / "src" / "orchestrata_ultimate",
                Path(__file__).resolve().parents[1] / "scripts",
            ]
        )

        registry: dict[str, tuple[str, str, int]] = {}
        for root in roots:
            if not root.exists():
                continue
            for path in root.rglob("*.py"):
                if exclude_file and path == exclude_file:
                    continue
                try:
                    tree = ast.parse(path.read_text(encoding="utf-8"))
                except SyntaxError:
                    continue
                for node in tree.body:
                    if not isinstance(node, ast.Assign):
                        continue
                    for target in node.targets:
                        if not isinstance(target, ast.Name):
                            continue
                        name = target.id
                        if name.endswith("_ENV") or not name.isupper():
                            continue
                        if not isinstance(node.value, ast.Constant) or not isinstance(
                            node.value.value, str
                        ):
                            continue
                        registry[node.value.value] = (name, str(path), node.lineno)

        self._general_registry = registry
        return registry

    def _token_similarity(self, literal: str, registry_value: str) -> tuple[bool, int]:
        """Check if literal shares >=2 consecutive tokens with registry_value.

        Tokenize by splitting on '_', drop empties, compare token runs.
        Returns (similar, max_consecutive_tokens) where similar means >=2 consecutive shared tokens.

        Case-insensitive comparison.
        """
        if literal == registry_value:
            return (False, 0)  # Exact match is not drift

        def tokenize(s: str) -> list[str]:
            return [t for t in s.lower().split("_") if t]

        literal_tokens = tokenize(literal)
        registry_tokens = tokenize(registry_value)

        if not literal_tokens or not registry_tokens:
            return (False, 0)

        # Find max consecutive matching run
        max_consecutive = 0
        for i, lt in enumerate(literal_tokens):
            for j, rt in enumerate(registry_tokens):
                consecutive = 0
                li, ri = i, j
                while (
                    li < len(literal_tokens)
                    and ri < len(registry_tokens)
                    and literal_tokens[li] == registry_tokens[ri]
                ):
                    consecutive += 1
                    li += 1
                    ri += 1
                max_consecutive = max(max_consecutive, consecutive)

        return (max_consecutive >= 2, max_consecutive)

    @staticmethod
    def _is_env_key_shaped(literal: str) -> bool:
        """True when literal is env-key-shaped: contains '_' and all cased chars are uppercase.

        Env keys follow UPPER_SNAKE_CASE by universal convention. This shape gate filters
        out lowercase/mixed-case literals (JSON field names, dict payload keys) that share
        tokens with env constants but are NOT env keys.

        Example: "head_commit" → False (lowercase), "ORCH_HEAD_COMMIT" → True (uppercase)
        """
        if "_" not in literal:
            return False
        # Every cased character must be uppercase
        for c in literal:
            if c.isalpha() and not c.isupper():
                return False
        return True

    def _check_env_key_drift(self, path: Path, tree: ast.AST, lines: list[str]) -> None:
        """T21 — detect env key drift/fabrication by comparing literals to declared constants.

        Two levels:
        - T21 ERROR: literal shares >=2 consecutive tokens with a registry value
          but is != it (e.g. "ORCH_TEST_DEADLINE" vs "ORCHESTRATA_TEST_DEADLINE")
        - T21 WARNING: literal exactly matches a registry value used outside its
          declaring module → should import the constant instead

        SHAPE GATE: a literal enters T21 comparison ONLY if it is env-key-shaped
        (contains '_' AND every cased character is uppercase). This filters out
        lowercase/mixed-case literals like JSON field names ("head_commit") that
        share tokens with env constants but are not env key literals.

        Non-goals (do NOT flag): infra keys with NO declared constant; comments/
        docstrings; f-strings' dynamic parts. Self-exemption via path exemption.

        The rule applies to ANY .py file passed on the CLI (T1-T20 keep their
        current test-named scoping unchanged).
        """
        # Build registry excluding the file being reviewed
        registry = self._build_env_registry(exclude_file=path)

        # Skip the rule's own test file to avoid false positives from fixture code
        # (concatenated fixture literals and the self-exemption pattern).
        # D6: self-exemption via documented path exemption - the test file
        # contains fixture string literals that would otherwise trigger T21.
        if "tests/scripts/test_test_review_tool.py" in str(path):
            return

        # D8: self-exemption for the review tool itself — it uses env key literals
        # in string checks (not os.environ accesses) and importing the constants
        # would be circular (the tool checks the constants it uses).
        if "scripts/test-review.py" in str(path):
            return

        # Track which module declared each constant for exact-match warnings
        # Maps literal value -> (constant_name, declaring_module)
        constant_to_module: dict[str, tuple[str, str]] = {}
        for value, (name, module, _) in registry.items():
            constant_to_module[value] = (name, module)

        # T17-GEN general tier: exact-match WARNING only (no drift — the >=2-token
        # drift over all UPPER_CASE values is ~100% FP). Separate map so the
        # landed *_ENV tier semantics stay byte-stable.
        general_registry = self._build_general_registry(exclude_file=path)
        general_constant_to_module: dict[str, tuple[str, str]] = {}
        for value, (name, module, _) in general_registry.items():
            general_constant_to_module[value] = (name, module)

        # Collect docstring node IDs to exclude
        docstring_ids = set()
        for node in ast.walk(tree):
            if isinstance(
                node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.Module)
            ):
                docstring = self._docstring(node)
                if docstring is not None:
                    docstring_ids.add(id(docstring))

        # Collect ALL string literals in code position (excluding docstrings)
        # This matches the brief's "ast.walk over ast.Constant where value is str"
        string_literals: list[tuple[str, int]] = []
        for node in ast.walk(tree):
            if isinstance(node, ast.Constant) and isinstance(node.value, str):
                # Skip empty strings and docstrings
                if node.value and id(node) not in docstring_ids:
                    string_literals.append((node.value, node.lineno))

        # Remove duplicates while preserving line order
        seen: set[str] = set()
        unique_literals: list[tuple[str, int]] = []
        for lit, lineno in string_literals:
            if lit not in seen:
                seen.add(lit)
                unique_literals.append((lit, lineno))

        # Check each literal
        for literal, lineno in unique_literals:
            # SHAPE GATE: skip literals that are not env-key-shaped
            if not self._is_env_key_shaped(literal):
                continue

            # Skip literals ending in '_ENV' (constant NAME-shaped strings, not key values)
            if literal.endswith("_ENV"):
                continue

            # Skip infra keys that have no declared constant (RUL-003 style inline literals)
            # A literal is "infra" if it doesn't match any registry value and doesn't
            # share >=2 tokens with any registry value
            # D11: exact match PRECEDES drift. If literal exactly equals ANY registry value,
            # it's an exact-match issue (WARNING), NOT a drift issue (ERROR).
            has_match = False
            exact_match_value = None  # The registry value that matches exactly, if any
            best_match = (
                None  # (reg_value, max_consecutive_tokens) for similar-but-not-equal
            )
            for reg_value in registry:
                if literal == reg_value:
                    has_match = True
                    exact_match_value = reg_value  # Record exact match
                    # Keep checking for drift against other entries? No - exact match wins.
                    # Per D11: exact match PRECEDES drift. Stop checking similarity.
                    break
                similar, max_consecutive = self._token_similarity(literal, reg_value)
                if similar:
                    has_match = True
                    if best_match is None or max_consecutive > best_match[1]:
                        best_match = (reg_value, max_consecutive)

            # Create a simple node-like object for reporting line numbers
            class LineNode:
                def __init__(self, lineno):
                    self.lineno = lineno

            line_node = LineNode(lineno)

            # T17-GEN general tier: exact-match WARNING (retyped constant value),
            # no drift — the >=2-token drift over all UPPER_CASE values is ~100% FP.
            # Runs for every shape-gated literal, independent of the *_ENV match
            # (a general constant's value is not an env key, so it has no *_ENV
            # entry to match against).
            if literal in general_constant_to_module:
                gen_name, gen_module = general_constant_to_module[literal]
                if gen_module != str(path):
                    self._add(
                        path,
                        line_node,
                        "T21",
                        f"T21 retyped constant '{literal}' has a declared "
                        f"constant NAME='{gen_name}' at {gen_module} — import it "
                        f"instead of retyping",
                        Severity.WARNING,
                    )

            if not has_match:
                # No registry match at all - this is an infra key, skip it
                continue

            # D11: exact match PRECEDES drift. If literal exactly equals ANY registry value,
            # it's an exact-match issue (WARNING), NOT a drift issue (ERROR).
            # Check exact match first.
            has_drift_issue = False
            if exact_match_value is not None:
                # Exact match found - issue WARNING (unless self-declaration)
                # (drift check skipped - exact match wins)
                pass
            elif best_match is not None:
                # Drift issue (similar but not equal)
                reg_value, _ = best_match
                const_name, const_module, _ = registry[reg_value]
                has_drift_issue = True
                self._add(
                    path,
                    line_node,
                    "T21",
                    f"T21 possible fabricated/drifted env key '{literal}' — "
                    f"declared constant NAME='{const_name}' at {const_module}:{_}",
                    Severity.ERROR,
                )
            else:
                # Fallback: iterate through registry (shouldn't happen given has_match=True)
                for reg_value, (const_name, const_module, _) in registry.items():
                    similar, _ = self._token_similarity(literal, reg_value)
                    if similar:
                        self._add(
                            path,
                            line_node,
                            "T21",
                            f"T21 possible fabricated/drifted env key '{literal}' — "
                            f"declared constant NAME='{const_name}' at {const_module}:{_}",
                            Severity.ERROR,
                        )
                        break  # One error per literal is enough

            # Check for exact match outside declaring module
            # Skip if the literal was already flagged as drift (same literal can't be both)
            # OR if the literal is declared in this file (self-declaration)
            if has_drift_issue:
                # Already flagged as drift, skip exact match warning
                pass
            else:
                # Check for self-declaration
                is_self_declaration = False
                for node in tree.body:
                    if isinstance(node, ast.Assign):
                        for target in node.targets:
                            if isinstance(target, ast.Name):
                                if target.id.endswith("_ENV"):
                                    if isinstance(
                                        node.value, ast.Constant
                                    ) and isinstance(node.value.value, str):
                                        if node.value.value == literal:
                                            is_self_declaration = True
                                            break
                        if is_self_declaration:
                            break

                if not is_self_declaration and literal in constant_to_module:
                    const_name, declaring_module = constant_to_module[literal]
                    if declaring_module != str(path):
                        self._add(
                            path,
                            line_node,
                            "T21",
                            f"T21 env key '{literal}' has a declared constant NAME='{const_name}' at {declaring_module} — "
                            f"import it instead of retyping",
                            Severity.WARNING,
                        )

    # ------------------------------------------------------------------
    # Helpers
    # ------------------------------------------------------------------

    def _add(
        self,
        path: Path,
        node: ast.AST,
        rule: str,
        message: str,
        severity: Severity = Severity.ERROR,
    ) -> None:
        self.report.add(str(path), getattr(node, "lineno", 1), rule, message, severity)


# Severity rank, most severe first — used to filter by --min-severity. Severity
# names do NOT sort correctly alphabetically (e.g. "INFO" < "WARNING" would
# wrongly include INFO issues when min-severity=WARNING), so filtering must use
# this explicit rank rather than comparing Severity.value strings.
_SEVERITY_RANK: dict[Severity, int] = {
    Severity.ERROR: 3,
    Severity.WARNING: 2,
    Severity.INFO: 1,
}


def output_text(
    report: TestReport, min_severity: Severity, output_file: str | None = None
) -> None:
    out = sys.stdout if output_file is None else open(output_file, "w")
    min_rank = _SEVERITY_RANK[min_severity]
    filtered = [i for i in report.issues if _SEVERITY_RANK[i.severity] >= min_rank]

    print("=" * 80, file=out)
    print("TEST REVIEW REPORT", file=out)
    print("=" * 80, file=out)
    print(f"\nFiles Analyzed: {report.files_analyzed}", file=out)
    print(f"Total Issues: {len(filtered)}\n", file=out)

    if not filtered:
        print("✅ All test files comply with standards.", file=out)
    else:
        by_severity: dict[Severity, list[TestIssue]] = {s: [] for s in Severity}
        for i in filtered:
            by_severity[i.severity].append(i)

        for sev in (Severity.ERROR, Severity.WARNING, Severity.INFO):
            items = by_severity[sev]
            if items:
                print(f"\n{sev.value} ({len(items)}):", file=out)
                for item in items:
                    print(
                        f"  {item.file}:{item.line} [{item.rule}] {item.message}",
                        file=out,
                    )

    if output_file:
        out.close()


def _apply_worklist_baseline(report: TestReport, baseline_path: str) -> None:
    """Downgrade baseline rows (T21 exact-key WARNINGs named at dispatch time) to
    INFO — named debt, non-blocking. New findings keep their severity. A
    missing/unparsable baseline is a no-op (the baseline is advisory, not a
    correctness gate)."""
    import json

    try:
        data = json.loads(Path(baseline_path).read_text(encoding="utf-8"))
    except OSError, json.JSONDecodeError:
        return
    baseline = {(Path(r["file"]).name, r["line"]) for r in data.get("rows", [])}
    for issue in report.issues:
        if issue.rule != "T21" or issue.severity != Severity.WARNING:
            continue
        if (Path(issue.file).name, issue.line) in baseline:
            issue.severity = Severity.INFO


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Test file reviewer for orchestrata-ultimate"
    )
    parser.add_argument("paths", nargs="+", help="Test files or directories to review")
    parser.add_argument("--output", "-o", help="Output file for report")
    parser.add_argument("--stdout", action="store_true", help="Print report to stdout")
    parser.add_argument(
        "--min-severity", choices=["ERROR", "WARNING", "INFO"], default="WARNING"
    )
    parser.add_argument(
        "--worklist-baseline",
        help="Worklist-baseline JSON (T21-ROUTING): baseline rows downgrade to "
        "INFO (named debt, non-blocking); new findings keep their severity.",
    )
    args = parser.parse_args()

    files = find_test_files(args.paths)
    if not files:
        print("No test files found.", file=sys.stderr)
        return 0

    report = TestReport()
    checker = TestFileChecker(report)
    for f in files:
        checker.check_file(f)

    if args.worklist_baseline:
        _apply_worklist_baseline(report, args.worklist_baseline)

    min_sev = Severity(args.min_severity)
    output_text(report, min_sev, None if args.stdout else args.output)

    if args.stdout and args.output:
        output_text(report, min_sev, args.output)

    error_count = sum(1 for i in report.issues if i.severity == Severity.ERROR)
    return 1 if error_count > 0 else 0


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