fix(S2): enforce the no-import-time-exit contract as a real preflight

Two unsoundnesses in the harness guard, both fixed:

(1) The guard ran too late. test_harness_contract.py was itself an auto-discovered module,
so an earlier-sorting test_*.py that sys.exit()s at import ended the process before the
scan ran. The check is now a PREFLIGHT (tests/_preflight.py) that run_tests.run() calls
FIRST, over the source, before importing any test module — so an offender cannot sort
ahead of it. A violation prints PREFLIGHT FAIL and the runner returns 1.

(2) Pruning function/lambda/class bodies was unsound: a body is reached at import when a
module-scope call or an immediately-invoked lambda runs it. The invariant is now strict
and purely LOCATION-based (no call-graph analysis): sys.exit / raise SystemExit is allowed
ONLY inside a standalone  guard body — nowhere else, function
bodies included.

Regressions: MUST_CATCH now includes a helper called at module scope, an IIFE, an uncalled
function body and a class method body; the MUST_PASS cases that permitted function-body
exits are removed. A load-bearing regression builds a throwaway tests dir whose
earliest-sorting file exits at import and confirms the preflight reports it (plus a
helper-invoked offender and a clean guarded module). Verified end to end: injecting an
earlier-sorting offender into the real tests dir makes run_tests return 1, not a silent 0.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JxKjqdGEFzq4UzZupw379G

Author: PhysShell

tests/_preflight.py

@@ -0,0 +1,93 @@
+"""Preflight for the aggregate test runner.
+
+run_tests.py discovers every `test_*.py` and imports it with importlib. A single
+import-time `sys.exit(...)` / `raise SystemExit(...)` in ANY of them ends the whole process
+before the aggregate return code is collected — and the offender can sort BEFORE whatever
+module is meant to police it, so a policing test module cannot catch it. The check must
+therefore run as a PREFLIGHT, before the first test import (run_tests.run() calls
+check_test_files() first and aborts on any violation).
+
+The invariant is deliberately strict and purely LOCATION-based — no call-graph analysis,
+which an immediately-invoked helper or lambda defeats:
+
+    a test_*.py may use sys.exit / raise SystemExit ONLY inside the body of a standalone
+    top-level `if __name__ == "__main__":` guard — nowhere else (not module scope, not a
+    function/lambda/class body, not a decorator or default).
+"""
+
+from __future__ import annotations
+
+import ast
+import os
+
+
+def _is_main_guard(test: ast.expr) -> bool:
+    """Structurally EXACTLY `__name__ == "__main__"`."""
+    return (isinstance(test, ast.Compare)
+            and isinstance(test.left, ast.Name) and test.left.id == "__name__"
+            and len(test.ops) == 1 and isinstance(test.ops[0], ast.Eq)
+            and len(test.comparators) == 1
+            and isinstance(test.comparators[0], ast.Constant)
+            and test.comparators[0].value == "__main__")
+
+
+def _is_systemexit(exc: ast.expr | None) -> bool:
+    if isinstance(exc, ast.Call):
+        exc = exc.func
+    return isinstance(exc, ast.Name) and exc.id == "SystemExit"
+
+
+def _is_sys_exit(func: ast.expr) -> bool:
+    return (isinstance(func, ast.Attribute) and func.attr == "exit"
+            and isinstance(func.value, ast.Name) and func.value.id == "sys")
+
+
+def _guard_body_ids(module: ast.Module) -> set[int]:
+    """Every AST node lexically inside a standalone top-level `if __name__ == "__main__":`
+    BODY (its `else`/`elif` are NOT included — they run on import)."""
+    allowed: set[int] = set()
+    for stmt in module.body:
+        if isinstance(stmt, ast.If) and _is_main_guard(stmt.test):
+            for s in stmt.body:
+                for node in ast.walk(s):
+                    allowed.add(id(node))
+    return allowed
+
+
+def exit_violations(tree: ast.Module) -> list[ast.AST]:
+    """Every sys.exit call / raise SystemExit that sits OUTSIDE a main-guard body — which,
+    for a module imported by the runner, is any that could run at import (directly or via a
+    helper called at module scope). No exemption for function/lambda/class bodies."""
+    allowed = _guard_body_ids(tree)
+    bad: list[ast.AST] = []
+    for node in ast.walk(tree):
+        if id(node) in allowed:
+            continue
+        if isinstance(node, ast.Raise) and _is_systemexit(node.exc):
+            bad.append(node)
+        elif isinstance(node, ast.Call) and _is_sys_exit(node.func):
+            bad.append(node)
+    return bad
+
+
+def check_test_files(tests_dir: str) -> list[str]:
+    """Return a list of human-readable violations across every test_*.py in `tests_dir`.
+    Empty means every test module is safe for the runner to import."""
+    problems: list[str] = []
+    for fname in sorted(os.listdir(tests_dir)):
+        if not (fname.startswith("test_") and fname.endswith(".py")):
+            continue
+        path = os.path.join(tests_dir, fname)
+        try:
+            with open(path, encoding="utf-8") as fh:
+                tree = ast.parse(fh.read(), filename=fname)
+        except (OSError, SyntaxError) as exc:
+            problems.append(f"{fname}: cannot parse ({exc})")
+            continue
+        for node in exit_violations(tree):
+            problems.append(
+                f"{fname}:{node.lineno}: sys.exit / raise SystemExit outside the "
+                "`if __name__ == \"__main__\"` guard body — it would end the aggregate "
+                "runner at import time"
+            )
+    return problems

tests/run_tests.py

@@ -994,6 +994,18 @@ def helper_and_report_smoke() -> list[str]:
 
 
 def run() -> int:
+    # PREFLIGHT — BEFORE importing any test module. A test_*.py that ends the process at
+    # import (sys.exit / raise SystemExit outside its `__main__` guard) would silently
+    # truncate this run, and it can sort before whatever module is meant to police it, so
+    # the check must happen here, first, over the source rather than by importing.
+    from _preflight import check_test_files
+    _here = os.path.dirname(os.path.abspath(__file__))
+    preflight = check_test_files(_here)
+    for problem in preflight:
+        print(f"PREFLIGHT FAIL: {problem}")
+    if preflight:
+        return 1
+
     passed = 0
     failed = 0
     for name, body, expected in CASES:

tests/test_harness_contract.py

@@ -1,163 +1,115 @@
-"""Guard the aggregate test-runner contract itself.
-
-The bug this pins down: a `test_*.py` that executes its checks at import time and ends with
-`sys.exit(...)` ends the WHOLE process when run_tests.py imports it, silently dropping every
-module discovered after it and the aggregate return code. This module statically proves that
-CANNOT happen — every sibling `test_*.py` exposes `run()` and never calls sys.exit /
-raise SystemExit at module scope (only under an `if __name__ == "__main__"` guard).
+"""Guard the aggregate test-runner contract — the ENFORCEABLE version.
+
+The real enforcement is a PREFLIGHT in run_tests.py: before importing any test_*.py it
+calls _preflight.check_test_files(), which refuses any sys.exit / raise SystemExit that
+is not inside a standalone `if __name__ == "__main__":` guard body. That runs first, so an
+offender cannot sort ahead of the check and end the process during import.
+
+This module proves the scanner behind that preflight: it exercises the location invariant
+(no exemption for function/lambda/class bodies — an immediately-invoked helper defeats any
+call-graph exemption), checks the live tests directory, and — the load-bearing regression —
+builds a throwaway tests directory whose EARLIEST-sorting file exits at import and confirms
+the preflight reports it (so the runner would return failure, not silent success).
 """
 
 from __future__ import annotations
 
 import ast
 import os
+import tempfile
+
+from _preflight import check_test_files, exit_violations
 
 failures: list[str] = []
 checks = 0
 
 _HERE = os.path.dirname(os.path.abspath(__file__))
 
-
-def _is_main_guard(test: ast.expr) -> bool:
-    """Structurally recognise EXACTLY `__name__ == "__main__"` — not any top-level `if`,
-    so `if True: sys.exit(0)` is NOT waved through."""
-    return (isinstance(test, ast.Compare)
-            and isinstance(test.left, ast.Name) and test.left.id == "__name__"
-            and len(test.ops) == 1 and isinstance(test.ops[0], ast.Eq)
-            and len(test.comparators) == 1
-            and isinstance(test.comparators[0], ast.Constant)
-            and test.comparators[0].value == "__main__")
-
-
-def _future_annotations(module: ast.AST) -> bool:
-    body = getattr(module, "body", [])
-    for stmt in body:
-        if isinstance(stmt, ast.ImportFrom) and stmt.module == "__future__":
-            if any(alias.name == "annotations" for alias in stmt.names):
-                return True
-    return False
-
-
-def _annotation_exprs(func: ast.FunctionDef | ast.AsyncFunctionDef) -> list[ast.expr]:
-    a = func.args
-    exprs = [arg.annotation for arg in
-             (*a.posonlyargs, *a.args, *a.kwonlyargs) if arg.annotation]
-    for extra in (a.vararg, a.kwarg):
-        if extra is not None and extra.annotation is not None:
-            exprs.append(extra.annotation)
-    if func.returns is not None:
-        exprs.append(func.returns)
-    return exprs
-
-
-def _module_scope_exit(node: ast.AST, ann_eager: bool | None = None) -> bool:
-    """True if a sys.exit(...) / raise SystemExit(...) would fire at IMPORT time. Only the
-    genuinely DEFERRED subtrees are pruned: a function / async-function / lambda BODY, and
-    the `if __name__ == "__main__"` body. Everything else runs on import and is scanned —
-    class bodies + bases + keywords, decorators, argument defaults, lambda defaults, and
-    (unless `from __future__ import annotations` is in force) annotations. The node itself
-    is checked BEFORE its children, so a decorator / default that IS `sys.exit(...)` is
-    caught, not just its arguments."""
-    if ann_eager is None:  # decided once, at the module root, then threaded down
-        ann_eager = not _future_annotations(node)
-
-    if isinstance(node, ast.Raise) and _is_systemexit(node.exc):
-        return True
-    if isinstance(node, ast.Call) and _is_sys_exit(node.func):
-        return True
-    if isinstance(node, ast.If) and _is_main_guard(node.test):
-        # ONLY the guard's body is exempt — its `else:` (and `elif`) still run on import.
-        return any(_module_scope_exit(s, ann_eager) for s in node.orelse)
-    if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
-        eager = [*node.decorator_list, *node.args.defaults,
-                 *[d for d in node.args.kw_defaults if d is not None]]
-        if ann_eager:
-            eager += _annotation_exprs(node)
-        return any(_module_scope_exit(e, ann_eager) for e in eager)
-    if isinstance(node, ast.Lambda):
-        eager = [*node.args.defaults, *[d for d in node.args.kw_defaults if d is not None]]
-        return any(_module_scope_exit(e, ann_eager) for e in eager)
-    # Everything else — module body, class body/bases/keywords/decorators, module-scope
-    # control flow — executes on import; scan every child.
-    return any(_module_scope_exit(c, ann_eager) for c in ast.iter_child_nodes(node))
-
-
-def _is_systemexit(exc: ast.expr | None) -> bool:
-    if isinstance(exc, ast.Call):
-        exc = exc.func
-    return isinstance(exc, ast.Name) and exc.id == "SystemExit"
-
-
-def _is_sys_exit(func: ast.expr) -> bool:
-    return (isinstance(func, ast.Attribute) and func.attr == "exit"
-            and isinstance(func.value, ast.Name) and func.value.id == "sys")
-
-
-_SELFTEST_MUST_CATCH = (
-    ("bare sys.exit at module scope", "import sys\nsys.exit(0)\n"),
-    ("exit inside `if True:`", "import sys\nif True:\n    sys.exit(0)\n"),
-    ("exit inside a top-level try",
+# MUST be reported as violations — including exits buried in a function / lambda / class
+# body, because a module-scope call (or an IIFE) runs them at import.
+_MUST_CATCH = (
+    ("bare module-scope exit", "import sys\nsys.exit(0)\n"),
+    ("exit in `if True:`", "import sys\nif True:\n    sys.exit(0)\n"),
+    ("exit in a top-level try",
      "import sys\ntry:\n    sys.exit(0)\nexcept Exception:\n    pass\n"),
     ("raise SystemExit at module scope", "raise SystemExit(1)\n"),
-    ("exit inside a for loop", "import sys\nfor _ in range(1):\n    sys.exit(0)\n"),
-    ("sys.exit in a main-guard else",
+    ("exit in a for loop", "import sys\nfor _ in range(1):\n    sys.exit(0)\n"),
+    ("exit in a main-guard else",
      "import sys\nif __name__ == '__main__':\n    pass\nelse:\n    sys.exit(0)\n"),
     ("raise SystemExit in a main-guard else",
      "if __name__ == '__main__':\n    pass\nelse:\n    raise SystemExit(1)\n"),
     ("exit in a main-guard elif",
      "import sys\nif __name__ == '__main__':\n    pass\nelif True:\n    sys.exit(0)\n"),
-    # Declarations that run code AT IMPORT — the trapdoors that skipping a whole
-    # FunctionDef / ClassDef / Lambda node would miss.
     ("exit in a class body", "import sys\nclass C:\n    sys.exit(7)\n"),
     ("exit in a class base", "import sys\nclass C(sys.exit(7)):\n    pass\n"),
     ("exit in a class decorator", "import sys\n@sys.exit(7)\nclass C:\n    pass\n"),
     ("exit in a function default", "import sys\ndef f(value=sys.exit(7)):\n    pass\n"),
     ("exit in a function decorator", "import sys\n@sys.exit(7)\ndef f():\n    pass\n"),
-    ("exit in a keyword-only default",
-     "import sys\ndef f(*, value=sys.exit(7)):\n    pass\n"),
     ("exit in a lambda default", "import sys\nf = lambda value=sys.exit(7): None\n"),
-    ("exit in an eager annotation", "import sys\ndef f(x: sys.exit(7)):\n    pass\n"),
+    # The blocker-2 forms: a body that IS reached at import.
+    ("exit in a helper called at module scope",
+     "import sys\ndef abort():\n    sys.exit(7)\nabort()\n"),
+    ("exit in an immediately-invoked lambda", "import sys\nv = (lambda: sys.exit(7))()\n"),
+    # And even an UNCALLED body — the strict location rule refuses it regardless.
+    ("exit in an uncalled function body", "import sys\ndef f():\n    sys.exit(7)\n"),
+    ("exit in a class method body",
+     "import sys\nclass C:\n    def m(self):\n        sys.exit(7)\n"),
 )
-_SELFTEST_MUST_PASS = (
+# MUST be accepted: only the `__main__` guard body, or no exit at all, or a string literal.
+_MUST_PASS = (
     ("guarded entrypoint", "import sys\nif __name__ == '__main__':\n    sys.exit(0)\n"),
-    ("exit inside a function", "import sys\ndef run():\n    sys.exit(0)\n"),
-    ("exit inside a lambda", "f = lambda: __import__('sys').exit(0)\n"),
-    ("exit inside a class method body",
-     "import sys\nclass C:\n    def m(self):\n        sys.exit(0)\n"),
-    ("stringified annotation under future-annotations",
-     "from __future__ import annotations\nimport sys\ndef f(x: sys.exit(7)):\n    pass\n"),
+    ("guarded raise SystemExit(run())",
+     "def run():\n    return 0\nif __name__ == '__main__':\n    raise SystemExit(run())\n"),
+    ("guarded exit nested under an inner if",
+     "import sys\nif __name__ == '__main__':\n    if '--x' in sys.argv:\n"
+     "        raise SystemExit(0)\n"),
+    ("no exit at all", "def run():\n    return 0\n"),
+    ("exit only as a string literal", "MSG = 'call sys.exit(0) to quit'\n"),
 )
 
 
+def _violates(src: str) -> bool:
+    return bool(exit_violations(ast.parse(src)))
+
+
 def run() -> int:
     global checks
-    # Self-test the guard first: it must catch the ways the original bug could recur, and
-    # must NOT flag the sanctioned entrypoint or an exit that only lives inside a scope.
-    for label, src in _SELFTEST_MUST_CATCH:
-        checks += 1
-        if not _module_scope_exit(ast.parse(src)):
-            failures.append(f"guard self-test: failed to catch {label}")
-    for label, src in _SELFTEST_MUST_PASS:
+    for label, src in _MUST_CATCH:
         checks += 1
-        if _module_scope_exit(ast.parse(src)):
-            failures.append(f"guard self-test: wrongly flagged {label}")
-
-    for fname in sorted(os.listdir(_HERE)):
-        if not (fname.startswith("test_") and fname.endswith(".py")):
-            continue
-        if fname == os.path.basename(__file__):
-            continue
+        if not _violates(src):
+            failures.append(f"scanner self-test: failed to catch {label}")
+    for label, src in _MUST_PASS:
         checks += 1
-        src = open(os.path.join(_HERE, fname), encoding="utf-8").read()
-        tree = ast.parse(src)
-        has_run = any(isinstance(n, ast.FunctionDef) and n.name == "run"
-                      for n in ast.iter_child_nodes(tree))
-        if not has_run:
-            failures.append(f"{fname}: has no module-level run()")
-        if _module_scope_exit(tree):
-            failures.append(f"{fname}: calls sys.exit/raise SystemExit at import scope "
-                            "(would short-circuit the aggregate runner)")
-    print(f"harness contract: {checks - len(failures)}/{checks} test modules honour run()")
+        if _violates(src):
+            failures.append(f"scanner self-test: wrongly flagged {label}")
+
+    # The live tests directory must itself be clean (this is what the runner enforces).
+    checks += 1
+    live = check_test_files(_HERE)
+    if live:
+        failures.append("live tests directory has import-time-exit violations: "
+                        + "; ".join(live))
+
+    # The load-bearing regression: an offender that sorts BEFORE any policing module must be
+    # caught by the preflight, so the runner returns failure instead of a silent green. Also
+    # a helper invoked at module scope — the blocker-2 case call-graph pruning would miss.
+    checks += 1
+    with tempfile.TemporaryDirectory() as tmp:
+        with open(os.path.join(tmp, "test_aaa_offender.py"), "w", encoding="utf-8") as fh:
+            fh.write("import sys\nsys.exit(0)\n")
+        with open(os.path.join(tmp, "test_zzz_helper.py"), "w", encoding="utf-8") as fh:
+            fh.write("import sys\n\n\ndef _boom():\n    sys.exit(0)\n\n\n_boom()\n")
+        with open(os.path.join(tmp, "test_mmm_clean.py"), "w", encoding="utf-8") as fh:
+            fh.write("def run():\n    return 0\n\n\nif __name__ == '__main__':\n"
+                     "    raise SystemExit(run())\n")
+        found = check_test_files(tmp)
+        offenders = {p.split(":", 1)[0] for p in found}
+        if offenders != {"test_aaa_offender.py", "test_zzz_helper.py"}:
+            failures.append(f"preflight regression: flagged {sorted(offenders)}, expected "
+                            "the bare-exit and helper-invoked offenders only "
+                            "(the guarded module must pass)")
+
+    print(f"harness contract: {checks - len(failures)}/{checks} checks pass")
     for f in failures:
         print(f"  FAIL: {f}")
     return 1 if failures else 0