scripts.run_tests
Reference › scripts › run_tests
scripts/run_tests.py is the unified, authoritative test runner for orchestrata-ultimate. One command checks infrastructure, runs the static gates, then runs the pytest suite in parallel batches and writes a report. A human runs it, and promote_env (as its default test command) and project_status (which reads its JSON report) refer to it; tests/scripts has several modules for it, including test_run_tests, test_run_tests_cap and test_run_tests_container_cpu. The parallel execution, scheduling and live observation live in orchestrata_ultimate.core.threading (the test executor, the live observer and reader); this script is the orchestration around them. See the pytest live observer page and observe_pytest for watching a run.
Invocation: positional test paths (default tests/) and these options. -j/--workers (default 8) sets parallel pytest processes; -t/--timeout (default 420) is the per-batch budget; -b/--batch-multiplier (default 6) is batches per worker, for work-stealing granularity; --backfill holds back a fraction of the estimated suite for tail-filling batches; -a/--pytest-arg passes extra pytest arguments; --solo-retry-cap (default 25) bounds how many batch-failed modules are re-run alone, and recoveries are reported as flakes and appended to a flake log; --skip-infra, --skip-services (a comma-separated list such as kernel,gateway,redis,graph), --skip-gates and --skip-freshness skip stages; --force ignores the proven-run reuse check; --json; --debug; --affected [REV] runs only the tests that affected_tests resolves for changes since a revision, as a non-authoritative subset run, with --affected-decision as an explicit human decision that lets a refused resolution proceed (RUL-040); and --include-rehearsal includes the rehearsal-class modules that gate runs exclude (RUL-039).
The flow in _main is: reject a test path that does not exist before anything starts; refresh stale inputs (the freshness guard); _check_infra starts or checks services (see start_test_services); _run_gates runs the gate scripts, such as gate_parallel_safety, as subprocesses; _try_reuse_report reuses a previous proven run for the same commit and dirty-tree signature (_run_identity); _run_suite runs the batches; packages under packages/ are partitioned and run per package (_partition_package_paths, _run_package_suites, ADR-0105). Afterwards it records slow-test candidates, container CPU use (see container_cpu_accounting) and failure excerpts, and _write_canonical_report writes the report under reports/. main wraps _main so that under --json standard output carries only the JSON document and all progress goes to standard error. A run cap and deadline (_test_cap_seconds, _run_deadline) bound the whole run and are passed to children through the environment. The exit code is 0 on success and 1 on failure or abort, and under --json an abort still emits a document.
At a glance
| Source | scripts/run_tests.py |
| Lines of code | 2153 |
| Classes | 0 |
| Functions | 28 |
| Imports (internal) | orchestrata_ultimate.core.threading, orchestrata_ultimate.core.threading.pytest_live_observer, orchestrata_ultimate.core.threading.pytest_live_reader, orchestrata_ultimate.core.threading.test_executor, scripts, scripts.container_cpu_accounting, scripts.test_green_ledger |
| Imported by | none |
| Tests that import it | tests/scripts/conftest.py, tests/scripts/test_affected_tests.py, tests/scripts/test_run_tests.py, tests/scripts/test_run_tests_cap.py, tests/scripts/test_run_tests_container_cpu.py, tests/scripts/test_run_tests_main_infra.py, tests/scripts/test_run_tests_rehearsal_exclusion.py |
Class diagram
No classes; see Functions.
Dependency diagram
API
Functions
| Function | Signature | Summary |
|---|---|---|
prune_stale_basetemps |
prune_stale_basetemps(root: Path | None = None, max_age_hours: float = 24.0) -> tuple[int, int] |
Reclaim stale pytest-of-<user> basetemp dirs. |
_check_infra |
_check_infra(skip_services: str = '', quiet_stdout: bool = False, worker_count: int = 0) -> bool |
Start services if needed. |
_run_one_gate |
_run_one_gate(gate_name: str, gate_script: Path, extra_args: tuple[str, ...] = ()) -> tuple[str, bool, str] |
Run a single gate subprocess. |
_run_gates |
_run_gates(target_dir: Path | None = None) -> bool |
Run blocking static gates. |
_run_identity |
_run_identity() -> tuple[str, str] |
(HEAD commit, dirty-tree signature). |
_try_reuse_report |
_try_reuse_report(scope_tests: list[str] | None, *, head: str, dirty_sig: str) -> dict | None |
The canonical report when it PROVES this run already, else None. |
_resolve_affected |
_resolve_affected(rev: str) -> dict | None |
Resolve rev's changed files to an affected-test subset via scripts/affected_tests.py --json (Phase 3.1). |
_plan_affected_run |
_plan_affected_run(resolved: dict, decision: str | None) -> tuple[str | None, dict, list[str] | None] |
Translate an affected_tests resolution into a run plan (pure). |
_affected_all_inert |
_affected_all_inert(resolved: dict) -> bool |
True when every changed file in a resolution is certified inert (docs / config / generated) — a pure non-code push that needs no test beyond the smoke floor (GATE-AFFECTED, 2026-10-21). |
_partition_package_paths |
_partition_package_paths(paths: list[str]) -> dict[str, list[str]] |
ADR-0105: group packages/<pkg>/… test paths by package. |
_run_package_suites |
_run_package_suites(grouped: dict[str, list[str]], *, timeout: int) -> dict |
ADR-0105: run each package's parity tests in its OWN venv. |
_slow_test_candidates |
_slow_test_candidates(started_after: float, threshold: float = 60.0) -> tuple[list[tuple[float, str]], list[tuple[float, str, str]]] |
Tests from THIS run (ts >= started_after) at/above threshold seconds, per the Phase 3.3 duration sink (reports/test-durations.ndjson, written automatically by pytest_live_observer at… |
_failure_excerpt |
_failure_excerpt(result: Mapping[str, Any] | None, max_lines: int = 40) -> str |
The assertion/error lines from a batch failure, for the flake log. |
_log_flake_recoveries |
_log_flake_recoveries(recovered: list[str], batch_results: Mapping[str, Any] | None = None, batch_plan: list[list[str]] | None = None) -> None |
Append solo-recovered modules to the flake-hunting log (NDJSON). |
_strip_cov_args |
_strip_cov_args(pytest_args: list[str]) -> list[str] |
Drop coverage flags for the solo-retry round. |
_run_suite |
_run_suite(paths: list[str], workers: int, timeout: int, pytest_args: list[str], json_output: bool, batch_multiplier: int = 3, solo_retry_cap: int = 5, backfill: float = 0.0, deadline: float | None = None) -> dict |
Run tests via the batched threadpool executor. |
_rehearsal_modules |
_rehearsal_modules() -> tuple[set[str] | None, str] |
The rehearsal-class module paths from the committed derivation artifact (RUL-039). |
_expand_paths |
_expand_paths(paths: list[str]) -> list[str] |
Expand directory paths into sorted test_*.py module paths. |
_supersede_module_totals |
_supersede_module_totals(test_totals: dict[str, int], old_counts: dict[str, int] | None, new_counts: dict[str, int] | None) -> None |
Replace one module's contribution in the suite-level test totals. |
_should_keep_full_record |
_should_keep_full_record(doc: dict[str, Any], existing: dict[str, Any] | None) -> bool |
True when a scoped run must yield to the current full-scope record. |
_write_canonical_report |
_write_canonical_report(doc: dict[str, Any], reports_dir: Path | None = None) -> None |
Atomically persist the canonical machine-readable report. |
_snapshot_container_cpu_before |
_snapshot_container_cpu_before(env: Mapping[str, str] | None = None) -> Any |
Best-effort pre-executor cgroup snapshot. |
_measure_container_cpu |
_measure_container_cpu(before: Any, *, env: Mapping[str, str] | None = None) -> dict[str, Any] | None |
Diff before against a fresh snapshot and persist the delta. |
main |
main() -> int |
Entry point. |
_test_cap_seconds |
_test_cap_seconds() -> int |
The test-phase hard cap in seconds (RUL-038). |
_run_deadline |
_run_deadline(cap_seconds: int) -> float |
The wall-clock deadline for the test phase. |
_child_run_env |
_child_run_env(deadline: float | None) -> dict[str, str] |
Environment variables to pass to child pytest processes. |
_main |
_main(json_stdout: TextIO) -> int |
Source
Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/run_tests.py
#!/usr/bin/env python3
"""Unified test runner for orchestrata-ultimate.
Single entry point that:
1. Verifies infrastructure (Redis, Graph) — starts if needed
2. Runs parallel safety gates (static analysis, no test execution)
3. Runs the full suite via the batched threadpool executor
4. Reports timing and results
RUL-032 (Mohamed, 2026-09-22, ``rulings/RUL-032-tests-run-once-per-code-
state.md``): before batching, every resolved module is partitioned against
``scripts/test_green_ledger.py`` — a module whose current content key
matches its last GREEN run is reused, not re-run (``--force`` bypasses).
Every module that passes this run is recorded green at its current key.
Architecture (per docs/development-and-testing.md):
The ThreadPoolExecutor dispatches 4 subprocesses. Each subprocess runs
a batch of ~244 test modules as a single pytest session — paying the
subprocess-startup tax 4 times instead of 978 times. ORCHESTRATA_WORKER_SLOT
is injected per subprocess for Redis DB isolation.
Usage:
python scripts/run_tests.py # Full suite
python scripts/run_tests.py tests/kernel/ # Subset
python scripts/run_tests.py --workers 8 # 8 parallel batches
python scripts/run_tests.py --skip-gates # Skip static checks
python scripts/run_tests.py --json # JSON output for CI
python scripts/run_tests.py --timeout 1800 # 30-min batch timeout
python scripts/run_tests.py --affected [REV] # Subset via affected_tests.py
# (default REV: HEAD) — NON-
# AUTHORITATIVE, see Phase 3.2.
"""
from __future__ import annotations
import argparse
import contextlib
import getpass
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import time
from collections.abc import Mapping
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from typing import Any, TextIO
# D7: Import env key constants to avoid hardcoded literals (T21 rule)
# The declared constants are:
# _RUN_ID_ENV = "ORCH_RUN_ID" (pytest_live_observer.py:282)
# _HEAD_COMMIT_ENV = "ORCH_HEAD_COMMIT" (pytest_live_observer.py:283)
# TEST_DEADLINE_ENV = "ORCHESTRATA_TEST_DEADLINE" (pytest_live_reader.py:305)
from orchestrata_ultimate.core.threading.pytest_live_observer import (
_HEAD_COMMIT_ENV,
_RUN_ID_ENV,
)
REPO_ROOT = Path(__file__).resolve().parent.parent
# Invoked directly (``python scripts/run_tests.py``, this file's own
# documented usage), Python puts THIS FILE'S OWN directory
# (REPO_ROOT/scripts) on sys.path[0] -- not REPO_ROOT -- so a sibling import
# like ``scripts.container_cpu_accounting`` (recommendation #5's lazy
# import, below) fails with "No module named 'scripts'" even though
# ``scripts/__init__.py`` makes it a real package. Harmless when already
# present (pytest's rootdir insertion, ``-m``, or a prior interpreter run
# already did this).
if str(REPO_ROOT) not in sys.path:
sys.path.insert(0, str(REPO_ROOT))
# Self-bootstrap into the project venv BEFORE importing anything from the
# package or spawning any child. This runner launches its infra/gate/test
# subprocesses via ``sys.executable``, so it MUST itself run under
# ``.venv/bin/python`` (the only interpreter where ``orchestrata_ultimate`` is
# importable). Under a bare ``python scripts/run_tests.py`` sys.executable is
# the SYSTEM interpreter, and the run dies deep inside start_test_services.py
# with ``ModuleNotFoundError: No module named 'orchestrata_ultimate'`` — which
# masquerades as "infra not ready". Re-exec under the venv so bare ``python``,
# ``.venv/bin/python``, and ``uv run python`` all behave identically. Idempotent:
# a no-op once already under the venv (the re-exec'd process passes the check).
_VENV_PYTHON = REPO_ROOT / ".venv" / "bin" / "python"
if _VENV_PYTHON.exists() and Path(sys.executable).resolve() != _VENV_PYTHON.resolve():
os.execv(
str(_VENV_PYTHON),
[str(_VENV_PYTHON), str(Path(__file__).resolve()), *sys.argv[1:]],
)
# The gate must be hermetic w.r.t. the caller's shell: several conftests and
# tools under test resolve dev binaries (basedpyright-langserver, mypy, ...)
# via PATH, which only works when the venv is activated. Prepend this
# interpreter's bin dir so an un-activated invocation behaves identically.
# No .resolve(): sys.executable in a venv is a SYMLINK into the uv python
# install — resolving it escapes the venv and prepends the wrong bin dir
# (live 2026-07-29: basedpyright-langserver 'not found' across 6 modules
# in a worktree whose venv had it, because the prepended dir was the
# interpreter install's bin, not .venv/bin).
_VENV_BIN = str(Path(sys.executable).parent)
if _VENV_BIN not in os.environ.get("PATH", "").split(os.pathsep):
# This IS the entrypoint, and the PATH it prepends is exactly what its own
# gate-ok: process-bootstrap env routing, inherited by the child pytest runs
os.environ["PATH"] = _VENV_BIN + os.pathsep + os.environ.get("PATH", "")
# ── Temp housekeeping ────────────────────────────────────────────────────────
def prune_stale_basetemps(
root: Path | None = None, max_age_hours: float = 24.0
) -> tuple[int, int]:
"""Reclaim stale ``pytest-of-<user>`` basetemp dirs. Returns (dirs, bytes).
A full-gate run spawns 1541 pytest subprocesses, each minting a numbered
basetemp dir under ``/tmp/pytest-of-<user>/``. pytest's own keep-last-3
cleanup does not contain this across many runs (stale lock files + many
roots) — measured 2026-08-05: 3,129 stale dirs, 16 GB, tmpfs 100% full,
which cascaded into ~135 ENOSPC test failures mid-gate. This prunes
anything older than ``max_age_hours`` at gate start, before the test
phase, so ``/tmp`` never accumulates across runs.
``root`` is the injection seam for tests (a real, disposable directory —
never mutate the live ``/tmp/pytest-of-<user>`` from a test). Live
concurrent runs are never touched: only children whose mtime is already
older than the cutoff are removed.
"""
if root is None:
root = Path(tempfile.gettempdir()) / f"pytest-of-{getpass.getuser()}"
if not root.exists():
return 0, 0
cutoff = time.time() - (max_age_hours * 3600)
dirs_removed = 0
bytes_reclaimed = 0
for child in root.iterdir():
if not child.is_dir():
continue
try:
mtime = child.stat().st_mtime
except OSError:
continue
if mtime >= cutoff:
continue
size = 0
try:
for f in child.rglob("*"):
with contextlib.suppress(OSError):
size += f.stat().st_size
except OSError:
pass
shutil.rmtree(child, ignore_errors=True)
dirs_removed += 1
bytes_reclaimed += size
return dirs_removed, bytes_reclaimed
# ── Infrastructure ───────────────────────────────────────────────────────────
def _check_infra(
skip_services: str = "", quiet_stdout: bool = False, worker_count: int = 0
) -> bool:
"""Start services if needed. Returns True if infra is ready.
Args:
skip_services: Comma-separated service names to skip (e.g. "kernel,gateway").
quiet_stdout: Capture the child's stdout and re-emit it through this
process's ``print`` instead of letting it inherit fd 1. Set under
``--json``, where fd 1 carries the JSON document and the child's
infrastructure report would corrupt it (see ``main``).
"""
services_script = REPO_ROOT / "scripts" / "start_test_services.py"
if not services_script.exists():
print(
"[run_tests] ⚠ start_test_services.py not found — skipping infra check",
flush=True,
)
return True
cmd = [sys.executable, str(services_script)]
if skip_services:
cmd.append(f"--skip={skip_services}")
if worker_count > 0:
# Provision every worker band this run will use (orchkg_w0..w<N-1>):
# gate children run with PYTEST_SKIP_SERVICE_START=1 and never create
# their own band, so without this a purged/fresh cluster fails every
# slot whose band no historical run happened to leave behind
# (iteration-13, 2026-08-05).
cmd.append(f"--ensure-worker-dbs={worker_count}")
result: subprocess.CompletedProcess[Any]
if quiet_stdout:
result = subprocess.run(cmd, cwd=str(REPO_ROOT), capture_output=True, text=True)
print(result.stdout, end="", flush=True)
print(result.stderr, end="", file=sys.stderr, flush=True)
else:
result = subprocess.run(cmd, cwd=str(REPO_ROOT))
if result.returncode != 0:
print("[run_tests] ❌ Infrastructure not ready — aborting", flush=True)
return False
# RUL-010 bans skipping verification, not performing it once in the parent:
# the children would otherwise redo the whole preflight per module.
# gate-ok: set only AFTER this runner verified the infrastructure itself
os.environ["PYTEST_SKIP_SERVICE_START"] = "1"
return True
# ── Safety Gates ─────────────────────────────────────────────────────────────
def _run_one_gate(
gate_name: str, gate_script: Path, extra_args: tuple[str, ...] = ()
) -> tuple[str, bool, str]:
"""Run a single gate subprocess. Returns (gate_name, passed, stdout)."""
print(f"[run_tests] 🔒 Running {gate_name} gate...", flush=True)
result = subprocess.run(
[sys.executable, str(gate_script), *extra_args],
cwd=str(REPO_ROOT),
capture_output=True,
text=True,
)
return gate_name, result.returncode == 0, result.stdout
def _run_gates(target_dir: Path | None = None) -> bool:
"""Run blocking static gates. Returns True if all gates pass.
Four gates, all BLOCKING (user ruling 2026-07-10: no violations
allowed):
- parallel-safety static analysis (flakiness patterns)
- test-purity audit (mocks/fakes/stubs/monkeypatch/skips) — previously
only ran inside the manual health_check, so its findings could not
block a suite run.
- prolog-boundary gate (reports/proposal-wire-prolog-governance.md §7
Phase 1): every Python Prolog query site must target a predicate/
arity that exists in the consulted rule base.
- rulings lint (plans/workflow-improvements-plan.md Phase 2, RUL-014):
every ruling in the git-backed wiki rulings canon must declare a real
mechanism (hook/gate/banned-deps/constitution) or an explicit
judgment-only mark. SKIPS (does not fail) when RULINGS_DIR /
~/workspace/knowledge-wiki/rulings is absent, so a machine without the
wiki checkout never goes false-red on this gate.
All four gates are independent, read-only static analyses with
disjoint outputs, so they run CONCURRENTLY (a ThreadPoolExecutor around
the subprocess.run calls — same pattern start_test_services.py already
uses for its 4 infra checks) instead of serially. Saves ~max(a, b, c, d)
vs a+b+c+d on the critical path (2026-07-18 speed profile, Lever 2).
Still fully blocking: any gate failing fails the run.
``target_dir`` is the integration-test seam (session 260): when set,
the three tree-walking gates (parallel safety ``--tests-dir``, test
purity positional path, prolog boundary ``--source``) analyze THAT tree
instead of the whole repo, so the wiring test exercises real gate
subprocesses against a small fixture in milliseconds instead of
~25-47s of full-tree analysis under gate load. Production (``main()``)
never passes it — the suite gate always analyzes the full tree.
"""
gates = [
# First, because every other gate's verdict is meaningless if the suite
# is about to import a different checkout's source than the one it is
# testing (2026-07-27: a leaked PYTHONPATH made a full run report 30
# phantom failures against another branch's src).
("source provenance", REPO_ROOT / "scripts" / "check_source_provenance.py"),
("parallel safety", REPO_ROOT / "scripts" / "gate_parallel_safety.py"),
("test purity", REPO_ROOT / "scripts" / "audit_test_purity.py"),
("prolog boundary", REPO_ROOT / "scripts" / "check_prolog_boundary.py"),
("rulings lint", REPO_ROOT / "scripts" / "rulings_lint.py"),
]
# Per-gate extra CLI args when a target tree is injected. Provenance is
# process-global by nature (it verifies THIS interpreter's import path)
# and rulings lint reads the wiki store, not the repo tree — both are
# sub-second and stay untargeted.
target_args: dict[str, tuple[str, ...]] = {}
if target_dir is not None:
target_args = {
"parallel safety": ("--tests-dir", str(target_dir)),
"test purity": (str(target_dir),),
"prolog boundary": ("--source", str(target_dir)),
}
to_run: list[tuple[str, Path]] = []
for gate_name, gate_script in gates:
if not gate_script.exists():
print(
f"[run_tests] ⚠ {gate_script.name} not found — skipping gate",
flush=True,
)
continue
to_run.append((gate_name, gate_script))
results: dict[str, tuple[bool, str]] = {}
if to_run:
with ThreadPoolExecutor(max_workers=len(to_run)) as executor:
futures = [
executor.submit(
_run_one_gate,
gate_name,
gate_script,
target_args.get(gate_name, ()),
)
for gate_name, gate_script in to_run
]
for future in as_completed(futures):
gate_name, passed, stdout = future.result()
results[gate_name] = (passed, stdout)
all_passed = True
for gate_name, _ in to_run:
passed, stdout = results[gate_name]
print(stdout, flush=True)
if not passed:
print(
f"[run_tests] ❌ {gate_name} gate failed — "
"fix violations before running tests",
flush=True,
)
all_passed = False
if all_passed:
print("[run_tests] ✅ All safety gates pass", flush=True)
return all_passed
# ── Proven-run reuse (gate dedup) ────────────────────────────────────────────
#
# The canonical report (reports/run-tests-latest.json) now carries the run's
# IDENTITY: HEAD commit, dirty-tree signature, and scope. A later invocation
# that would cover only tests an identical, already-passing run covered exits
# 0 in seconds instead of re-running — the 2026-08-11 sequence (targeted
# green runs at commit X, then a push of X re-running them) made this the
# gate's biggest single waste.
#
# Fail-closed, always: reuse requires ALL of — last run passed; HEAD commit
# identical; dirty-tree signature identical (names + content sha1, reports/
# excluded: the report itself rewrites every run); and scope coverage (a FULL
# report covers anything; an affected report covers only subsets of itself).
# Any mismatch, missing/malformed report, or --force: run normally.
def _run_identity() -> tuple[str, str]:
"""(HEAD commit, dirty-tree signature). The signature is sha1 over each
dirty/untracked file's name+content, reports/ excluded — it exists so an
interactive edit silently invalidates reuse rather than laundering it."""
try:
head = subprocess.run(
["git", "rev-parse", "HEAD"],
cwd=str(REPO_ROOT),
capture_output=True,
text=True,
check=True,
).stdout.strip()
status = subprocess.run(
["git", "status", "--porcelain"],
cwd=str(REPO_ROOT),
capture_output=True,
text=True,
check=True,
).stdout
except subprocess.CalledProcessError, OSError:
# Not a git checkout (run_tests' own synthetic test repos): a
# never-matching identity means reuse simply never fires —
# fail-closed, no behavior change.
return "", f"no-git-{os.getpid()}"
import hashlib as _hl
h = _hl.sha1()
for line in sorted(status.splitlines()):
rel = line[3:]
if rel.startswith("reports/"):
continue
path = REPO_ROOT / rel
h.update(rel.encode())
if path.is_file():
h.update(path.read_bytes())
return head, h.hexdigest()
def _try_reuse_report(
scope_tests: list[str] | None, *, head: str, dirty_sig: str
) -> dict | None:
"""The canonical report when it PROVES this run already, else None.
``scope_tests=None`` means a full-suite run — only a full-scope report
covers it. An affected run is covered by a full report, or by an affected
report whose test set is a superset of this run's.
"""
path = REPO_ROOT / "reports" / "run-tests-latest.json"
try:
report = json.loads(path.read_text(encoding="utf-8"))
except OSError, json.JSONDecodeError:
return None
totals = report.get("test_totals") or {}
if int(totals.get("failed") or 0) or int(totals.get("errors") or 0):
return None
if not int(totals.get("passed") or 0):
return None
if report.get("head_commit") != head or report.get("dirty_sig") != dirty_sig:
return None
scope = report.get("scope")
if scope == "full":
return report
if scope_tests is not None and isinstance(scope, list):
if set(scope_tests) <= set(scope):
return report
return None
# ── Affected-test selection (Phase 3.2) ─────────────────────────────────────
def _resolve_affected(rev: str) -> dict | None:
"""Resolve ``rev``'s changed files to an affected-test subset via
``scripts/affected_tests.py --json`` (Phase 3.1). Returns ``None`` on
any failure (bad git rev, script missing) so the caller can fall back
rather than silently running an empty subset.
"""
script = REPO_ROOT / "scripts" / "affected_tests.py"
if not script.exists():
print("[run_tests] ⚠ scripts/affected_tests.py not found", flush=True)
return None
result = subprocess.run(
[sys.executable, str(script), "--since", rev, "--json"],
cwd=str(REPO_ROOT),
capture_output=True,
text=True,
)
if result.returncode != 0:
print(
f"[run_tests] ⚠ affected_tests.py failed (rc={result.returncode}): "
f"{result.stderr.strip()[-500:]}",
flush=True,
)
return None
if not result.stdout.strip():
# rc=0 but stdout is empty/whitespace-only — NOT a JSON syntax error
# (json.loads("\n") raises "Expecting value: line 2 column 1", which
# misleads anyone debugging this toward a syntax bug that isn't
# there). Report what actually happened: no output, plus whatever
# landed on stderr, so an intermittent recurrence is diagnosable.
print(
f"[run_tests] ⚠ affected_tests.py produced NO OUTPUT "
f"(rc={result.returncode}, stdout={len(result.stdout)} bytes): "
f"{result.stderr.strip()[-500:]}",
flush=True,
)
return None
try:
# json.loads is typed Any; the declared return type is the contract
# this function actually offers its callers, so narrow it here rather
# than leaking Any (mypy no-any-return).
parsed = json.loads(result.stdout)
return parsed if isinstance(parsed, dict) else None
except json.JSONDecodeError as exc:
excerpt = repr(result.stdout[:200])
print(
f"[run_tests] ⚠ affected_tests.py produced invalid JSON: {exc} "
f"(stdout excerpt: {excerpt}) stderr: {result.stderr.strip()[-500:]}",
flush=True,
)
return None
#: The affected-verdict vocabulary (RUL-040). Kept as literals here — run_tests
#: consumes affected_tests.py's JSON over a subprocess, not by import — so the
#: two files' strings must agree; a drift is caught by the arm tests below.
_AFFECTED_BOUNDED = "bounded"
_AFFECTED_NAMED_FULL = "named_full"
_AFFECTED_REFUSED = "refused"
def _plan_affected_run(
resolved: dict,
decision: str | None,
) -> tuple[str | None, dict, list[str] | None]:
"""Translate an affected_tests resolution into a run plan (pure).
Returns ``(note, report_detail, abort_messages)``:
* ``bounded`` -> ``note`` = the subset note, ``report_detail`` carries the
verdict, ``abort_messages=None`` (run the resolved selection).
* ``named_full`` -> same, but ``report_detail["named_full"]`` lists each
named whole-suite verdict (path + reason) for the gate operator. The
whole suite is NEVER auto-run — the name is the verdict.
* ``refused`` -> with a ``decision`` citation: proceed with the bounded
remainder and record the refusal rows + decision. Without one:
``abort_messages`` names each missing fact (the refused reasons) so the
caller aborts — never a silent full run, in any arm.
"""
verdict = resolved.get("verdict")
rows = resolved.get("rows") or []
named_full = [r for r in rows if r.get("verdict") == _AFFECTED_NAMED_FULL]
refused = [r for r in rows if r.get("verdict") == _AFFECTED_REFUSED]
report: dict = {"verdict": verdict}
if named_full:
# RUL-040: surface named_full verdicts in EVERY arm — a refused
# resolution with a decision still shows the whole-suite-named changes
# (dependency/env/conftest). The operator sees every name, never a
# silently-dropped verdict.
report["named_full"] = [
{"path": r.get("path", ""), "reason": r.get("reason", "")}
for r in named_full
]
if verdict == _AFFECTED_REFUSED:
report["refused"] = [
{"path": r.get("path", ""), "reason": r.get("reason", "")} for r in refused
]
if not decision:
return None, report, [r.get("reason", "refused") for r in refused]
report["affected_decision"] = decision
note = f"refused; proceeded by explicit decision: {decision}"
return note, report, None
if verdict == _AFFECTED_NAMED_FULL:
note = (
"named_full verdict(s) recorded — whole-suite change named, "
"run explicitly if the blast radius requires it"
)
return note, report, None
note = "subset run — non-authoritative, full suite still owed"
return note, report, None
def _affected_all_inert(resolved: dict) -> bool:
"""True when every changed file in a resolution is certified inert
(docs / config / generated) — a pure non-code push that needs no test
beyond the smoke floor (GATE-AFFECTED, 2026-10-21)."""
rows = resolved.get("rows") or []
return bool(rows) and all(r.get("verdict") == "inert" for r in rows)
# ── Test Execution ───────────────────────────────────────────────────────────
def _partition_package_paths(paths: list[str]) -> dict[str, list[str]]:
"""ADR-0105: group ``packages/<pkg>/…`` test paths by package.
Only files/dirs under ``packages/<pkg>/`` reach the package runner; the
root batched executor keeps everything else. A ``tests/``-less package
contributes nothing here — its paths never resolve to package paths.
"""
grouped: dict[str, list[str]] = {}
for path in paths:
parts = path.split("/")
if path.startswith("packages/") and len(parts) >= 3:
grouped.setdefault(parts[1], []).append(path)
return grouped
def _run_package_suites(grouped: dict[str, list[str]], *, timeout: int) -> dict:
"""ADR-0105: run each package's parity tests in its OWN venv.
The parity suites import sibling packages resolved only by the
package's ``pyproject.toml`` path sources — the root venv declares none
of them, so running them under the root interpreter cannot even
collect. Each package group runs as ONE ``python -m pytest`` session
per (up to) 64 files, cwd = the package directory, interpreter = the
package's own ``.venv/bin/python``. A missing venv is provisioned via
``uv sync`` in the package dir (fail-closed: provisioning failure is a
run failure). Returns the same-shaped result dict as the root executor.
"""
totals = {"passed": 0, "failed": 0, "errors": 0, "skipped": 0}
modules: dict[str, dict[str, Any]] = {}
counts: dict[str, int] = {}
summary_re = re.compile(
r"(\d+) passed.*?(\d+) failed.*?(\d+) error|"
r"(\d+) passed(?:,| in)|"
r"no tests ran"
)
for pkg, paths in sorted(grouped.items()):
pkg_dir = REPO_ROOT / "packages" / pkg
py = pkg_dir / ".venv" / "bin" / "python"
if not py.exists():
print(
f"[run_tests] ⚡ packages/{pkg}: no venv — provisioning with "
f"uv sync --no-editable (ADR-0105)",
flush=True,
)
provision = subprocess.run(
["uv", "sync", "--no-editable"],
cwd=str(pkg_dir),
capture_output=True,
text=True,
timeout=timeout,
)
if provision.returncode != 0:
print(
f"[run_tests] ✗ packages/{pkg}: uv sync failed "
f"(rc={provision.returncode}): "
f"{provision.stderr.strip()[-400:]}",
flush=True,
)
totals["errors"] += len(paths)
for path in paths:
modules[path] = {"returncode": 1}
continue
for chunk_start in range(0, len(paths), 64):
chunk = paths[chunk_start : chunk_start + 64]
print(
f"[run_tests] 🧪 packages/{pkg}: {len(chunk)} parity test "
f"module(s) in the package venv ({', '.join(chunk[:4])}"
f"{'…' if len(chunk) > 4 else ''})",
flush=True,
)
# Resolver paths are repo-relative; pytest runs with cwd = the
# package dir, so pass ABSOLUTE paths — a repo-relative
# "packages/session/tests/…" resolved against the package dir
# would collect nothing (measured, ADR-0105 seam test).
absolute_chunk = [str(REPO_ROOT / p) for p in chunk]
proc = subprocess.run(
[str(py), "-m", "pytest", "-q", "--no-header", *absolute_chunk],
cwd=str(pkg_dir),
capture_output=True,
text=True,
timeout=timeout,
)
tail = (proc.stdout or "").strip().splitlines()
match = None
for line in reversed(tail):
match = summary_re.search(line)
if match:
break
passed = failed = errored = 0
if match and match.group(4):
passed = int(match.group(4))
elif match and not match.group(4):
passed = int(match.group(1) or 0)
failed = int(match.group(2) or 0)
errored = int(match.group(3) or 0)
rc = proc.returncode
if rc == 0:
# pytest rc 0 with a parsed "N passed" line: all green.
totals["passed"] += passed
for path in chunk:
modules[path] = {"returncode": 0}
counts[path] = 1
else:
print(
f"[run_tests] ✗ packages/{pkg} parity chunk failed "
f"(rc={rc}, passed={passed}, failed={failed}, "
f"errors={errored}):\n{(proc.stdout or '')[-1500:]}"
f"\n{(proc.stderr or '')[-800:]}",
flush=True,
)
totals["passed"] += passed
totals["failed"] += failed
totals["errors"] += errored
for path in chunk:
modules[path] = {"returncode": rc}
counts[path] = 0
return {
"test_totals": totals,
"passed": totals["passed"],
"failed": totals["failed"] + totals["errors"],
"total": sum(totals.values()),
"modules": modules,
"module_test_counts": counts,
}
def _slow_test_candidates(
started_after: float, threshold: float = 60.0
) -> tuple[list[tuple[float, str]], list[tuple[float, str, str]]]:
"""Tests from THIS run (``ts >= started_after``) at/above ``threshold``
seconds, per the Phase 3.3 duration sink (``reports/test-durations.ndjson``,
written automatically by ``pytest_live_observer`` at session-finish).
Returns ``(hits, acknowledged)``: ``hits`` are UNEXPLAINED slow tests —
the defect candidates. ``acknowledged`` are tests carrying a
``pytest.mark.known_slow`` marker whose reason the observer recorded into
the durations sink: slowness that has been reviewed and named (a spec'd
load measurement, contention-armored retry), reported separately so the
🐢 line stays a signal instead of training everyone to ignore it
(2026-08-11: two false-positive "defect candidates" per full run).
Reads the file directly rather than importing ``scripts/test_durations.py``
-- this module is invoked BOTH as a bare script (``python
scripts/run_tests.py``, no repo root on ``sys.path``) and as
``scripts.run_tests`` (tests), and a cross-``scripts``-package import
would only work in the latter mode. ``scripts/test_durations.py --slow``
remains the full historical-report CLI; this is just the just-ran-tests
filter for the runner's own summary line.
"""
empty: tuple[list[tuple[float, str]], list[tuple[float, str, str]]] = ([], [])
path = REPO_ROOT / "reports" / "test-durations.ndjson"
if not path.exists():
return empty
latest: dict[str, tuple[float, float, str | None]] = {}
try:
with path.open(encoding="utf-8") as fh:
for line in fh:
line = line.strip()
if not line:
continue
try:
obj = json.loads(line)
except json.JSONDecodeError:
continue
test = obj.get("test")
duration = obj.get("duration")
ts = obj.get("ts")
if not isinstance(test, str) or not isinstance(duration, (int, float)):
continue
ts_f = float(ts) if isinstance(ts, (int, float)) else 0.0
known = obj.get("known_slow")
prior = latest.get(test)
if prior is None or ts_f >= prior[0]:
latest[test] = (
ts_f,
float(duration),
known if isinstance(known, str) else None,
)
except OSError:
return empty
hits = [
(dur, test)
for test, (ts_f, dur, known) in latest.items()
if ts_f >= started_after and dur >= threshold and known is None
]
acknowledged = [
(dur, test, known)
for test, (ts_f, dur, known) in latest.items()
if ts_f >= started_after and dur >= threshold and known is not None
]
hits.sort(reverse=True)
acknowledged.sort(reverse=True)
return hits, acknowledged
# pytest traceback frame headers ("tests/x.py:33: in test_y") carry the
# WHERE of a failure; a timeout raised by SIGALRM has no assertion line, so
# without these the record names the symptom and discards the location.
_TB_FRAME_RE = re.compile(r"\S+\.py:\d+: in \S")
def _failure_excerpt(result: Mapping[str, Any] | None, max_lines: int = 40) -> str:
"""The assertion/error lines from a batch failure, for the flake log.
Solo-recovery OVERWRITES the batch result with the passing one, so
without capturing this the only surviving evidence of a load/order flake
is its module name. That made a recurring flake undiagnosable: on
2026-07-26 the same module recovered three times across the day and each
record read ``{ts, module}``, so the second occurrence could not be
compared against the first. Failing lines only, capped — the point is to
make recurrence comparable, not to archive whole logs.
Also keeps pytest's ``Captured stdout``/``Captured stderr`` section
bodies: an in-process check (e.g. test_kg_schema's canonical-schema
conformance gate, intermittent under batch load through 2026-08-05)
prints its drift report there, and the bare ``assert 1 == 0`` line the
old filter kept made the failure undiagnosable ("Unknown pattern").
Also keeps pytest-timeout's all-threads stack dump (the region between
its ``+++ Timeout +++`` banners): through 2026-08-14 a recurring 120s
hang in test_volk_extended left records reading only ``E Failed:
Timeout`` because none of the dump's lines matched the kept prefixes —
the one section that says WHERE the hang sat was the one discarded.
Tail-preferred (main thread prints last), with its own line budget so a
many-threaded dump cannot crowd out the assertion lines or vice versa.
"""
if not result:
return ""
text = f"{result.get('stdout', '')}\n{result.get('stderr', '')}"
keep: list[str] = []
timeout_dump: list[str] = []
in_captured_section = False
in_timeout_dump = False
for line in text.splitlines():
stripped = line.strip()
# pytest section banners are dash/equals-run lines; a "Captured
# stdout/stderr" banner OPENS a kept region, any other banner
# (warnings summary, short test summary, the next failure's own
# header) CLOSES it. pytest-timeout wraps its stack dump in
# plus-run "Timeout" banners, which toggle their own kept region.
is_banner = len(stripped) > 10 and (
(stripped.startswith("-") and stripped.endswith("-"))
or (stripped.startswith("=") and stripped.endswith("="))
)
is_plus_banner = (
len(stripped) > 10 and stripped.startswith("+") and stripped.endswith("+")
)
if is_plus_banner:
in_timeout_dump = "Timeout" in stripped and not in_timeout_dump
continue
if in_timeout_dump:
if stripped:
timeout_dump.append(stripped[:300])
continue
if is_banner:
in_captured_section = "Captured stdout" in stripped or (
"Captured stderr" in stripped
)
continue
if (
stripped.startswith(("E ", "assert ", "FAILED ", "> "))
or ("Error:" in stripped and stripped)
or _TB_FRAME_RE.match(stripped)
or (in_captured_section and stripped)
):
if len(keep) < max_lines:
keep.append(stripped[:300])
return "\n".join(keep + timeout_dump[-60:])
def _log_flake_recoveries(
recovered: list[str],
batch_results: Mapping[str, Any] | None = None,
batch_plan: list[list[str]] | None = None,
) -> None:
"""Append solo-recovered modules to the flake-hunting log (NDJSON).
A module that fails in a shared batch process but passes in isolation is
a load/order flake — exactly what flaky_test_hunter exists to chase. The
log is append-only so recurrence across runs is visible. Never lets a
logging failure break a test run.
``batch_results`` carries the pre-supersede BATCH result so the record
keeps the failure that actually happened (see :func:`_failure_excerpt`);
omitting it degrades to the old name-only record rather than failing.
"""
log_path = REPO_ROOT / ".flake_recoveries.ndjson"
try:
with log_path.open("a", encoding="utf-8") as fh:
for module in recovered:
record: dict[str, Any] = {
"ts": round(time.time(), 1),
"module": module,
}
excerpt = _failure_excerpt((batch_results or {}).get(module))
if excerpt:
record["batch_failure"] = excerpt
# The peers are the diagnosis. A module that passes alone and
# fails in a batch was broken by something ELSE in its
# process, and without this the record names only the victim.
# Recovering it afterwards is not possible: the junit temp
# dirs are deleted when the run ends (measured 2026-08-07 --
# gate 10's composition was gone before it could be read).
for batch in batch_plan or []:
if module in batch:
record["batch_peers"] = [m for m in batch if m != module]
break
fh.write(json.dumps(record) + "\n")
except OSError as exc:
print(f"[run_tests] ⚠ could not write {log_path.name}: {exc}", flush=True)
def _strip_cov_args(pytest_args: list[str]) -> list[str]:
"""Drop coverage flags for the solo-retry round.
The retry exists only to decide pass/fail; the module's coverage was
already recorded by the batch round (a failing test still executes its
lines). Passing ``--cov`` through would also make the executor's
stale-coverage purge delete the batch round's not-yet-combined
``.coverage.batch*`` files mid-run — the 2026-07-10 sweep lost 47 of 48
batch files exactly this way.
"""
return [a for a in pytest_args if not a.startswith("--cov")]
def _run_suite(
paths: list[str],
workers: int,
timeout: int,
pytest_args: list[str],
json_output: bool,
batch_multiplier: int = 3,
solo_retry_cap: int = 5,
backfill: float = 0.0,
deadline: float | None = None,
) -> dict:
"""Run tests via the batched threadpool executor. Returns result dict.
After the batch round, modules that FAILED get one solo-retry round
(each alone in its own pytest process, up to ``solo_retry_cap`` modules;
0 disables). Rationale (2026-07-10): four consecutive full runs each
failed exactly ONE module — a different one every run, every one green
solo — i.e. load/order tail flakes, not code defects. The retry mirrors
the documented pre-push practice ("re-run solo → if green, push") but
makes it systemic and TRANSPARENT: both rounds are reported, recovered
modules are logged to ``.flake_recoveries.ndjson`` for flake-hunting,
and a failure volume above the cap is never retried — that volume is
breakage, not flakes.
``deadline`` (RUL-038, test-phase cap): if set, injects the deadline into
the child pytest processes' environment so they can self-terminate when
the cap is reached. The child processes compare elapsed time against this
and kill themselves if exceeded.
"""
# Expand directories to test modules
modules = _expand_paths(paths)
if not modules:
print("[run_tests] ❌ No test modules found", flush=True)
# Carry the SAME keys a real run returns. This branch used to return
# a dict with "time" instead of "time_seconds", and the report step
# reads result["time_seconds"] unconditionally — so a target that
# matched no modules did not report "no modules", it died with
# `KeyError: 'time_seconds'` and a traceback, which reads as a broken
# runner rather than an empty target. Found 2026-08-07 while adding
# the zero-executed-tests guard below.
return {
"passed": 0,
"failed": 0,
"total": 0,
"time_seconds": 0.0,
"test_totals": {"passed": 0, "failed": 0, "errors": 0, "skipped": 0},
"modules": {},
"error": "no modules",
}
print(
f"[run_tests] 🧪 {len(modules)} modules · {workers} workers × "
f"{batch_multiplier} batches/worker · {timeout}s timeout",
flush=True,
)
# Use the project's own executor
from orchestrata_ultimate.core.threading.test_executor import (
BATCH_PLAN_KEY,
RERUN_SUMMARY_KEY,
TEST_TOTALS_KEY,
pop_aggregates,
run_tests_in_parallel,
)
# Stamp this run's identity so every duration record can be joined back to
# the run that produced it. Each batch is a separate pytest subprocess and
# `_run_subprocess` builds its env from `dict(os.environ)`, so the
# environment is the channel that reaches them; `run_tests_in_parallel`
# exposes no env seam here.
#
# setdefault, not assignment: an outer caller (CI, a wrapper, a test) that
# already established a run id keeps it, so its batches stay joinable to the
# same run rather than being renamed underneath it.
#
# Why at all: reports/test-durations.ndjson held 2.1M records of exactly
# {test, duration, ts}. Every performance question so far has had to
# reconstruct run boundaries by segmenting on timestamp gaps in a throwaway
# script — written, used, and deleted, repeatedly. See recommendation #1 of
# plans/prolog-learning-subsystem-research-2026-08-12.md, which measured the
# LEARNING half of that idea as not worth building and this half as the part
# that was.
os.environ.setdefault(_RUN_ID_ENV, f"{int(time.time())}-{os.getpid()}")
_head, _ = _run_identity()
if _head:
# Empty outside a git checkout (run_tests' own synthetic repos). Omitted
# rather than written blank: absent means "not recorded", where "" would
# claim a commit was looked up and found to be nothing.
os.environ.setdefault(_HEAD_COMMIT_ENV, _head)
# Build env with deadline for child processes (RUL-038)
child_env = {**os.environ, **_child_run_env(deadline)}
start = time.monotonic()
results = run_tests_in_parallel(
test_paths=modules,
pytest_args=pytest_args,
max_workers=workers,
timeout_per_test=float(timeout),
batch_multiplier=batch_multiplier,
backfill_fraction=backfill,
env=child_env,
)
# Strip every reserved bookkeeping key before the pass/fail tally: none
# carries a "returncode", so any left behind reads as an extra FAILED
# module. Popped via the executor's own helper so this call site cannot
# fall behind the key set (naming them by hand is what broke
# run_tests_threadpool.py when the batch plan was added, 2026-08-07).
_aggregates = pop_aggregates(results)
test_totals = _aggregates.get(TEST_TOTALS_KEY)
rerun_summary = _aggregates.get(RERUN_SUMMARY_KEY)
# Batch composition, for the flake log's `batch_peers`.
batch_plan = _aggregates.get(BATCH_PLAN_KEY)
# Solo-retry round (see the docstring): batch-failed modules get one
# isolated re-run each — max_workers=len(failed) with batch_multiplier=1
# gives every module its own bin, i.e. its own pytest process.
failed_first = sorted(p for p, r in results.items() if r.get("returncode", -1) != 0)
solo_summary: dict | None = None
if failed_first and solo_retry_cap > 0:
if len(failed_first) <= solo_retry_cap:
print(
f"[run_tests] 🔁 {len(failed_first)} module(s) failed in batch "
f"— re-running each in an isolated process "
f"(solo-retry cap {solo_retry_cap})",
flush=True,
)
# Timed, and the re-executed tests counted, because this round is
# not free and used to be reported as if it were. Measured
# 2026-08-19 on a full run: 428 tests across 19 modules executed in
# BOTH rounds, duplicating 2602 of 5632 test-seconds — 46% of all
# test time — while the summary said only "N recovered (load/order
# flakes)". A recovery mechanism that does not price its own
# recovery lets a contention regression hide indefinitely: every
# re-run module reports green, so nothing ever fails, and the only
# symptom is a suite that quietly takes twice as long.
_solo_started = time.monotonic()
solo_results = run_tests_in_parallel(
test_paths=failed_first,
pytest_args=_strip_cov_args(pytest_args),
max_workers=len(failed_first),
timeout_per_test=float(timeout),
batch_multiplier=1,
env={**os.environ, **_child_run_env(deadline)},
)
_solo_seconds = time.monotonic() - _solo_started
# The batch round's counts, read BEFORE the solo verdict
# supersedes them below — these are the tests paid for twice.
_duplicated_tests = sum(
sum((results.get(p, {}).get("testcase_counts") or {}).values())
for p in failed_first
)
pop_aggregates(solo_results)
recovered = sorted(
p
for p in failed_first
if solo_results.get(p, {}).get("returncode", -1) == 0
)
still_failed = [p for p in failed_first if p not in recovered]
solo_summary = {
"attempted": failed_first,
"recovered": recovered,
"still_failed": still_failed,
"duplicated_tests": _duplicated_tests,
"seconds": round(_solo_seconds, 1),
}
# Snapshot the BATCH failures before the solo verdict overwrites
# them — that output is the only record of what the flake
# actually looked like, and the passing solo run has none.
batch_failures = {p: results.get(p) for p in failed_first}
# The solo verdict supersedes the batch verdict either way: a
# recovery flips the module green; a repeat failure carries the
# fresher (isolated, load-free) failure detail. The suite-level
# per-TEST totals must follow the same supersede rule — without
# this, a recovered module's tests stay counted from its failed
# attempt (live 2026-07-29: a lone "errors: 1" in an otherwise
# all-green 30k-test run).
for p, r in solo_results.items():
if p in results:
if test_totals:
_supersede_module_totals(
test_totals,
results[p].get("testcase_counts"),
r.get("testcase_counts"),
)
results[p] = r
if recovered:
_log_flake_recoveries(recovered, batch_failures, batch_plan)
else:
# State what was not done — do NOT draw a conclusion the run
# declined to test. The old wording asserted "that volume is
# breakage, not load flakes", which the gate itself disproved on
# 2026-08-07: three separate runs tripped the cap, and every
# capped module passed in isolation. A cap bounds cost; it is not
# evidence about the failures it skipped.
print(
f"[run_tests] ⚠ {len(failed_first)} failed modules exceed "
f"the solo-retry cap ({solo_retry_cap}) — NOT retried, so "
f"whether they are breakage or load flakes is UNVERIFIED. "
f"Re-run them in isolation, or raise --solo-retry-cap.",
flush=True,
)
elapsed = time.monotonic() - start
passed = sum(1 for r in results.values() if r.get("returncode", -1) == 0)
failed = len(results) - passed
output = {
"test_totals": test_totals,
"rerun": rerun_summary,
"solo_retry": solo_summary,
"passed": passed,
"failed": failed,
"total": len(results),
"time_seconds": round(elapsed, 1),
# Always carry the FAILING modules (with their per-test JUnit detail) so
# the human summary can name them; passing modules are omitted off --json
# to keep the payload small.
"modules": (
results
if json_output
else {p: r for p, r in results.items() if r.get("returncode", 0) != 0}
),
# Compact per-module executed-test counts (one int each), persisted on
# EVERY run including all-green ones: suite totals were stable at
# 30663 across four consecutive gates, then stepped -236 on 2026-08-05
# with zero failures — and no artifact recorded WHICH modules ran
# fewer tests (junit dirs are deleted at run end, "modules" above
# omits green modules). This map makes run-over-run count deltas
# diffable forever.
"module_test_counts": {
p: sum((r.get("testcase_counts") or {}).values())
for p, r in results.items()
},
}
return output
def _rehearsal_modules() -> tuple[set[str] | None, str]:
"""The rehearsal-class module paths from the committed derivation artifact
(RUL-039). Returns ``(modules, error)``; on failure ``modules`` is None and
``error`` names the state + the regeneration command — the caller refuses
fail-closed rather than silently excluding nothing (or everything)."""
artifact = REPO_ROOT / "reports" / "rehearsal-class-modules.json"
if not artifact.is_file():
return None, (
"rehearsal-class artifact missing — regenerate with "
"`python scripts/derive_test_categories.py`"
)
try:
data = json.loads(artifact.read_text(encoding="utf-8"))
except OSError, json.JSONDecodeError:
return None, (
"rehearsal-class artifact unparsable — regenerate with "
"`python scripts/derive_test_categories.py`"
)
if data.get("derivation_version") != 1:
return None, (
"rehearsal-class artifact derivation_version mismatch — regenerate "
"with `python scripts/derive_test_categories.py`"
)
return {row["module"] for row in data.get("rows", [])}, ""
def _expand_paths(paths: list[str]) -> list[str]:
"""Expand directory paths into sorted test_*.py module paths."""
modules: list[str] = []
for p in paths:
path = Path(p)
if path.is_file() and path.suffix == ".py":
modules.append(str(path))
elif path.is_dir():
modules.extend(sorted(str(f) for f in path.rglob("test_*.py")))
else:
# Keep as-is — pytest will report the error
modules.append(p)
# De-dupe while preserving order
seen: set[str] = set()
out: list[str] = []
for m in modules:
if m not in seen:
seen.add(m)
out.append(m)
return out
def _supersede_module_totals(
test_totals: dict[str, int],
old_counts: dict[str, int] | None,
new_counts: dict[str, int] | None,
) -> None:
"""Replace one module's contribution in the suite-level test totals.
When a retry attempt supersedes a batch attempt, the superseded
attempt's per-test counts leave the totals and the fresh attempt's
enter — the same rule the per-module verdict already follows. A missing
counts dict (module produced no JUnit entry, e.g. a timeout-killed
batch) contributes zero in that direction.
"""
for key, n in (old_counts or {}).items():
if key in test_totals:
test_totals[key] -= n
for key, n in (new_counts or {}).items():
if key in test_totals:
test_totals[key] += n
def _should_keep_full_record(
doc: dict[str, Any], existing: dict[str, Any] | None
) -> bool:
"""True when a scoped run must yield to the current full-scope record.
A subset run proves less than a full run; overwriting the full record
with one destroys the verified-ancestor evidence ship reads (measured
2026-09-10: two 1-module-scope writes clobbered the authoritative
full-suite verdict in a single day). "Current" means the full record
names the same HEAD and dirty-tree signature as this run — a stale full
record proves nothing about this tree and is replaced normally.
"""
if doc.get("scope") == "full":
return False
if not isinstance(existing, dict) or existing.get("scope") != "full":
return False
return existing.get("head_commit") == doc.get("head_commit") and existing.get(
"dirty_sig"
) == doc.get("dirty_sig")
def _write_canonical_report(
doc: dict[str, Any], reports_dir: Path | None = None
) -> None:
"""Atomically persist the canonical machine-readable report.
``reports/run-tests-latest.json`` is written by the runner itself (temp
file + ``os.replace``) so that no invocation style — plain stdout,
``--json``, shell-redirected, or interrupted mid-suite — can leave
anything but valid JSON at the canonical path. Twice (2026-07-26 and
2026-07-29) that path held the human-readable stdout of a plain run
because the report only existed as a caller-side ``>`` redirect.
A scoped run or an abort document defers to a current full-scope record
(``_should_keep_full_record``). ``reports_dir`` defaults to
``REPO_ROOT / "reports"``; tests pass a ``tmp_path`` directory.
"""
if reports_dir is None:
reports_dir = REPO_ROOT / "reports"
reports_dir.mkdir(exist_ok=True)
target = reports_dir / "run-tests-latest.json"
try:
existing = json.loads(target.read_text(encoding="utf-8"))
except OSError, json.JSONDecodeError:
existing = None
if _should_keep_full_record(doc, existing):
what = "abort document" if "aborted" in doc else "scoped run"
print(
f"[run_tests] {what}: keeping the current full-scope record at "
"reports/run-tests-latest.json (it proves more)"
)
return
tmp = reports_dir / ".run-tests-latest.json.tmp"
tmp.write_text(json.dumps(doc, indent=2) + "\n")
os.replace(tmp, target)
# ── Container CPU accounting (recommendation #5) ────────────────────────────
#
# plans/prolog-learning-subsystem-research-2026-08-12.md: Memgraph and Redis
# run on the SAME physical cores as this suite and are EXCLUDED from its own
# 2,567.7 core-second accounting, so every "the suite is at its floor" claim
# -- including that document's own -- is conditioned on this unmeasured
# amount. Wired ALWAYS-ON as pure reporting: it can never fail the run (any
# error -- incus absent, a container down, an unreadable cgroup path, an
# unknown env profile -- logs one line and the suite proceeds exactly as it
# would have without this code), and it PERSISTS what it measures rather
# than only printing it, joined by run_id to the durations sink so a later
# question can answer itself instead of the research being redone and
# deleted again (the same failure recommendation #1 fixed for test
# durations).
def _snapshot_container_cpu_before(env: Mapping[str, str] | None = None) -> Any:
"""Best-effort pre-executor cgroup snapshot. Returns ``None`` (never
raises) on any failure -- this is pure observation with no correctness
role, and a measurement that can break the suite is worse than no
measurement. Respects the ACTIVE env profile (``ORCHESTRATA_ENV_PROFILE``,
default ``test``) via ``container_cpu_accounting.resolve_project`` --
never hardcoded.
``env`` is a dependency-injection seam (default: the real process
environment): every reader below takes it explicitly rather than
reading ``os.environ`` directly, so callers -- including tests -- can
vary it without process-global mutation (RUL-002; the same shape
``ingest_orchestrata_ultimate.py`` uses for its ``--backend`` override).
"""
e = env if env is not None else os.environ
try:
from scripts.container_cpu_accounting import resolve_project, take_snapshot
return take_snapshot(resolve_project(env=e), env=e)
except Exception as exc:
print(
f"[run_tests] ⚠ container CPU pre-snapshot failed (non-fatal): {exc}",
flush=True,
)
return None
def _measure_container_cpu(
before: Any, *, env: Mapping[str, str] | None = None
) -> dict[str, Any] | None:
"""Diff ``before`` against a fresh snapshot and persist the delta.
Same never-fail contract as ``_snapshot_container_cpu_before`` (see its
docstring) -- any error here is equally swallowed to one log line.
``before`` is ``None`` when the pre-snapshot already failed; there is
nothing to diff, so this is a no-op. Returns the report dict (for the
caller's own summary line) or ``None``. ``env`` is the same
dependency-injection seam as ``_snapshot_container_cpu_before``.
"""
if before is None:
return None
e = env if env is not None else os.environ
try:
from scripts.container_cpu_accounting import (
diff_report,
persist_diff_report,
take_snapshot,
)
after = take_snapshot(before.project, list(before.usage_seconds), env=e)
report = diff_report(before, after)
persist_diff_report(report, env=e)
return report
except Exception as exc:
print(
f"[run_tests] ⚠ container CPU measurement failed (non-fatal): {exc}",
flush=True,
)
return None
# ── CLI ──────────────────────────────────────────────────────────────────────
def main() -> int:
"""Entry point.
Under ``--json`` stdout carries the JSON document and NOTHING else: the
infrastructure report, the gate output and the ``[run_tests]`` progress
lines are diverted to stderr for the whole run. Without this,
``run_tests.py --json > reports/run-tests-latest.json`` — the command
``scripts/project_status.py`` tells you to run — writes a file that starts
with the infrastructure report and that no JSON parser accepts (exactly
what the committed report contained until 2026-07-26).
"""
if "--json" not in sys.argv[1:]:
return _main(sys.stdout)
real_stdout = sys.stdout
with contextlib.redirect_stdout(sys.stderr):
return _main(real_stdout)
# ── Test-phase cap and deadline (RUL-038) ────────────────────────────────────
def _test_cap_seconds() -> int:
"""The test-phase hard cap in seconds (RUL-038).
The test phase (after preflight and gates, before the executor dispatch)
must not exceed this cap. Child pytest processes receive this deadline and
kill themselves when it's reached.
The default is 300 seconds (5 minutes) to catch runaway tests without
blocking legitimate long-running suites. Adjust via ORCH_TEST_CAP_SECONDS.
"""
return int(os.environ.get("ORCH_TEST_CAP_SECONDS", "300"))
def _run_deadline(cap_seconds: int) -> float:
"""The wall-clock deadline for the test phase.
Current time + cap_seconds. Child processes compare their elapsed time
against this to decide when to self-terminate.
"""
return time.time() + cap_seconds
def _child_run_env(deadline: float | None) -> dict[str, str]:
"""Environment variables to pass to child pytest processes.
Injects the test-phase deadline so children can self-terminate when
the cap is reached. Also propagates the ORCH_RUN_ID so all children
share the same run identity.
deadline: float | None — if None, TEST_DEADLINE_ENV is omitted (children
will not enforce a cap but still get the SANCTIONED_RUNNER_ENV stamp).
"""
from orchestrata_ultimate.core.threading.pytest_live_reader import (
SANCTIONED_RUNNER_ENV,
TEST_DEADLINE_ENV,
)
env: dict[str, str] = {
SANCTIONED_RUNNER_ENV: "run_tests.py",
_RUN_ID_ENV: os.environ.get(_RUN_ID_ENV, ""),
}
if deadline is not None:
env[TEST_DEADLINE_ENV] = str(deadline)
return env
def _main(json_stdout: TextIO) -> int:
parser = argparse.ArgumentParser(
description="Unified test runner — infra + gates + batched executor",
)
parser.add_argument(
"paths",
nargs="*",
default=["tests/"],
help="Test paths (directories or files, default: tests/)",
)
parser.add_argument(
"-j",
"--workers",
type=int,
default=8,
help="Parallel pytest processes (default: 8). Raised from 4 on 2026-07-06: "
"per-worker Neo4j instances (NEO4J_URI_GW<N> in .env — 4 standalone incus "
"clones, 2 slots each) removed the shared-instance ceiling that previously "
"made 10 workers produce 111 timeout failures. Redis: per-slot DB 0..N. "
"Remaining shared services at higher -j: Gateway + CXP/KG gRPC.",
)
parser.add_argument(
"-t",
"--timeout",
type=int,
default=420,
help="Timeout per batch in seconds (default: 420 — with work-stealing "
"batches no healthy batch approaches 7 minutes; a batch that does IS "
"the defect)",
)
parser.add_argument(
"-b",
"--batch-multiplier",
type=int,
default=6,
help="Batches per worker (default: 6, raised from 3 on 2026-07-08: at "
"-j 8 the 273s per-batch session budget starved the tail of coarse "
"batches; 6/worker plus the test_orchestrata_tools AST-cache fix gave "
"a clean 7.0-min full run). >1 = work-stealing granularity: smaller "
"batches keep all slots busy to the end instead of letting heavy "
"modules pile into one long-tail bin.",
)
parser.add_argument(
"-a",
"--pytest-arg",
action="append",
default=[],
help="Extra pytest argument (repeatable)",
)
parser.add_argument(
"--backfill",
type=float,
default=0.0,
help="Fraction of estimated suite mass held back from static LPT "
"bins and drained as small tail-filling batches by whichever worker "
"frees first (test_executor._hold_back_backfill). 0 disables "
"(default). Designed for -b 1: one fat static bin per worker plus a "
"dynamic tail — the fat bins amortize per-subprocess overhead, the "
"tail flattens the end-of-run imbalance that made plain -b 1 regress "
"(369s vs 251s, 2026-08-05).",
)
parser.add_argument(
"--solo-retry-cap",
type=int,
default=25,
help="Max batch-failed modules to re-run in isolation (default: 25; "
"0 disables). Recoveries are load/order flakes — reported explicitly "
"and appended to .flake_recoveries.ndjson. Over the cap the modules "
"are left UNVERIFIED, not judged: the cap bounds cost, it does not "
"diagnose. Raised 5 -> 25 on 2026-08-07 — the cost premise was wrong "
"(9 modules re-ran solo in 5s against a 300-500s batch phase) and the "
"cap ran backwards: the more modules failed, the less the run checked.",
)
parser.add_argument(
"--skip-infra",
action="store_true",
help="Skip infrastructure check (assume services are running)",
)
parser.add_argument(
"--skip-services",
default="",
help="Comma-separated services to skip: kernel,gateway,redis,graph "
"('neo4j' accepted as a deprecated alias for 'graph')",
)
parser.add_argument(
"--skip-gates",
action="store_true",
help="Skip parallel safety static analysis",
)
parser.add_argument(
"--skip-freshness",
action="store_true",
help="Skip the freshness guard (stale-input detection + refresh)",
)
parser.add_argument(
"--force",
action="store_true",
help="Ignore the proven-run reuse check and run anyway",
)
parser.add_argument(
"--json",
action="store_true",
help="Output results as JSON",
)
parser.add_argument(
"--debug",
action="store_true",
help="Verbose failure diagnostics: capture DEBUG logs into each failing "
"test's report (junit_logging=all), long tracebacks, LOG_LEVEL=DEBUG.",
)
parser.add_argument(
"--affected",
nargs="?",
const="HEAD",
default=None,
metavar="REV",
help="Run only the tests scripts/affected_tests.py resolves for changes "
"since REV (default: HEAD). Overrides the positional `paths`. SUBSET "
"RUN — non-authoritative, the full suite is still owed (Phase 3.2). "
"The resolution is a VERDICT (RUL-040): `bounded` runs the resolved "
"subset; `named_full` runs the resolved subset and records each "
"named whole-suite change in the report (never an implicit full run — "
"the operator sees the name and decides); `refused` aborts naming the "
"missing fact and the changed set unless an explicit decision is "
"supplied via --affected-decision.",
)
parser.add_argument(
"--affected-decision",
metavar="CITATION",
default=None,
help="Explicit human decision that lets a `refused` affected resolution "
"proceed with the bounded remainder anyway. Recorded in the report and "
"on the refusal row (RUL-040: a refusal is never a silent full run).",
)
parser.add_argument(
"--include-rehearsal",
action="store_true",
help="Include rehearsal-class modules (real deploy/incus/tofu surfaces) "
"that the RUL-039 derivation excludes from gate runs by default. Use for "
"a deliberate rehearsal, never for a gate (RUL-039).",
)
args = parser.parse_args()
# Reject a test target that does not exist, before any infra spins up.
# `paths` is nargs="*", so ANY stray argument lands here — including a
# value handed to a store_true flag (`--json out.json` puts "out.json"
# in this list, measured 2026-08-07). Without this check the run
# narrowed itself to one nonexistent module and reported a green gate
# that had executed nothing. Name the offender and say why it is here.
# Resolved CWD-relative, exactly as _expand_paths does — not against
# REPO_ROOT, which the runner's own tests reassign to a tmp sandbox while
# still naming real repo-relative fixture modules.
missing_targets = [p for p in args.paths if not Path(p).exists()]
if missing_targets:
print(
"[run_tests] ❌ test target(s) do not exist: " + ", ".join(missing_targets),
flush=True,
)
print(
"[run_tests] If you did not mean to name a target: --json, "
"--debug, --skip-infra, --skip-gates and --skip-freshness take "
"NO value; anything written after one is consumed as a test "
"path. (--skip-services DOES take a value — a comma-separated "
"service list.)",
flush=True,
)
return 1
def _abort(stage: str, extra: dict | None = None) -> int:
"""Fail before any test ran — still emit a document under --json.
``extra`` is merged into the abort document, so a refusal can record
its named-refusal rows (RUL-040: a refusal is never a silent nothing).
"""
# Stamp the run identity: without head_commit/dirty_sig the guard in
# _should_keep_full_record cannot tell whether a full-scope record is
# CURRENT, and an abort document overwrote an in-flight full-scope
# report (2026-10-11 16:21Z, wt-GREENLET-FT: a nested test's
# infrastructure abort replaced the prod-gate run's report).
abort_head, abort_dirty = _run_identity()
doc = {
"aborted": stage,
"passed": 0,
"failed": 0,
"head_commit": abort_head,
"dirty_sig": abort_dirty,
}
if extra:
doc.update(extra)
_write_canonical_report(doc)
if args.json:
print(json.dumps(doc, indent=2), file=json_stdout)
json_stdout.flush()
return 1
# 0. Freshness guard — no run consumes stale inputs (ruling 2026-08-04:
# "no staleness, every run is fresh"). Detects+refreshes: AST snapshot
# runtime-merge loss, stale Prolog KBs, duration-sink bloat, observer
# ghost sessions, dead packer-cache entries; reports capture age.
guard_script = REPO_ROOT / "scripts" / "freshness_guard.py"
if not args.skip_freshness and not guard_script.exists():
# Sandboxed/synthetic REPO_ROOTs (tests) have no guard script — same
# not-found posture as the start_test_services.py check above.
print(
"[run_tests] ⚠ freshness_guard.py not found — skipping freshness guard",
flush=True,
)
elif not args.skip_freshness:
guard = subprocess.run(
[sys.executable, str(guard_script)],
cwd=REPO_ROOT,
timeout=900,
check=False,
)
if guard.returncode != 0:
return _abort("freshness")
# 0.5. Basetemp housekeeping — prune stale pytest-of-<user> dirs before
# the test phase so a long run never hits the tmpfs-full ENOSPC cascade
# (2026-08-05).
pruned, reclaimed = prune_stale_basetemps()
if pruned:
print(
f"[run_tests] 🧹 pruned {pruned} stale basetemp dir(s), "
f"{reclaimed / (1024 * 1024):.0f} MB reclaimed",
flush=True,
)
# 0.75. Affected-test resolution (Phase 3.2) — moved BEFORE infra so a
# pure non-code push short-circuits without booting the service stack
# (GATE-AFFECTED, 2026-10-21). The classification is pure git + static
# analysis and needs no services.
affected_note: str | None = None
affected_report: dict = {}
if args.affected is not None:
resolved = _resolve_affected(args.affected)
if resolved is None:
print(
"[run_tests] ⚠ --affected resolution failed — running the "
"requested paths unmodified",
flush=True,
)
else:
note, report_detail, abort_msgs = _plan_affected_run(
resolved, args.affected_decision
)
if abort_msgs:
# REFUSED (RUL-040): at least one change has no resolution and
# no explicit decision was supplied. Abort NAMING the missing
# facts + the changed set. Never a silent full run, never a
# silent nothing — the refusal rows are recorded via _abort.
print(
"[run_tests] ✗ --affected: REFUSED — cannot bound the "
"blast radius:",
flush=True,
)
for msg in abort_msgs:
print(f"[run_tests] {msg}", flush=True)
changed = resolved.get("changed") or []
if changed:
print(
f"[run_tests] changed: {', '.join(changed[:20])}"
f"{'…' if len(changed) > 20 else ''}",
flush=True,
)
print(
"[run_tests] re-run with --affected-decision "
"'<citation>' to proceed with the bounded remainder "
"explicitly",
flush=True,
)
return _abort("affected-refusal", extra={"affected": report_detail})
affected_note = note
affected_report = report_detail
args.paths = resolved.get("tests") or args.paths
# GATE-AFFECTED: every changed file is certified inert (docs /
# config / generated) — nothing executes or asserts on them, so
# the smoke floor would only re-prove already-verified services.
# Skip infra + gates + the smoke floor and pass.
if _affected_all_inert(resolved):
print(
"[run_tests] ⚪ --affected: all changes inert (non-code) — "
"no services or tests required; passing.",
flush=True,
)
head_commit, dirty_sig = _run_identity()
doc = {
"passed": 0,
"failed": 0,
"head_commit": head_commit,
"dirty_sig": dirty_sig,
"affected": affected_report,
"non_code": True,
"note": (
"all changed files inert (non-code) — skipped infra + "
"gates + smoke floor"
),
}
_write_canonical_report(doc)
if args.json:
print(json.dumps(doc, indent=2), file=json_stdout)
json_stdout.flush()
return 0
# 1. Infrastructure
if not args.skip_infra:
if not _check_infra(
skip_services=args.skip_services,
quiet_stdout=args.json,
# The solo-retry round runs failed modules with
# max_workers=len(failed) — up to solo_retry_cap — so its worker
# slots (and their per-slot orchkg_w<N> databases) can exceed
# args.workers when -j is small. Ensure bands for both phases.
worker_count=max(args.workers, args.solo_retry_cap),
):
return _abort("infrastructure")
else:
# A run that takes this path is NON-AUTHORITATIVE and says so
# gate-ok: the explicit --skip-infra opt-out (docs/kernel-is-required.md)
os.environ["PYTEST_SKIP_SERVICE_START"] = "1"
# gate-ok: same explicit opt-out
os.environ["PYTEST_SKIP_KERNEL"] = "1"
# 2. Safety gates
if not args.skip_gates:
if not _run_gates():
return _abort("gates")
# ADR-0105: partition package parity-suite paths (packages/<pkg>/tests/…)
# out of the root run. They import sibling packages that only the
# package's own venv resolves (path sources in packages/<pkg>/pyproject),
# so the root interpreter cannot collect them; _run_package_suites below
# executes each package's paths in its own venv and merges the totals.
# requested_paths keeps the FULL pre-partition scope: the reuse gate must
# identify runs by everything they were asked to run (a root-only report
# must never stand in for a run that also carries package suites).
requested_paths = list(args.paths)
package_paths = _partition_package_paths(args.paths)
args.paths = [p for p in args.paths if not p.startswith("packages/")]
# Proven-run reuse (gate dedup): a passing run at this EXACT tree
# identity (HEAD commit + dirty signature) whose scope covers this run's
# scope already proved it — exit 0 instead of re-running. --force bypasses.
head_commit, dirty_sig = _run_identity()
# An --affected run is NEVER full scope now: RUL-040 removed the silent
# full-run fallback, so even a named_full verdict runs the bounded
# remainder and records the name rather than auto-running the suite.
is_full_scope = args.affected is None and requested_paths == ["tests/"]
# The scope identity is the FULL requested set (root + package paths):
# _try_reuse_report matches by subset, so an empty post-partition root
# scope would wrongly reuse a root-only report and skip the package
# suites (measured: packages/session/tests reused a 1-test run and ran
# nothing). A report whose scope covered package paths carries their
# merged totals, so matching against the combined scope is exact.
scope_tests: list[str] | None = None if is_full_scope else sorted(requested_paths)
if not args.force:
proven = _try_reuse_report(scope_tests, head=head_commit, dirty_sig=dirty_sig)
if proven is not None:
print(
f"[run_tests] ♻ reusing the passing run at "
f"{str(proven.get('head_commit', ''))[:8]} "
f"(scope={proven.get('scope') if proven.get('scope') == 'full' else str(len(proven.get('scope') or [])) + ' tests'}; "
"identical HEAD + dirty tree) — nothing to re-run. --force to override.",
flush=True,
)
return 0
# 2c. RUL-032 (Mohamed, 2026-09-22): a test module whose content key --
# itself, its src import closure, its conftest chain, pyproject.toml --
# exactly matches its LAST GREEN run needs no re-run, whatever the git
# identity the report-level reuse check above used. Finer than
# _try_reuse_report: module-level and content-keyed, not
# HEAD-commit-and-dirty-tree-scoped, so a subset (or an escalated) run
# still reuses everything it proves it can instead of all-or-nothing.
# Lazy import: same reason as scripts.container_cpu_accounting above --
# a bare ``python scripts/run_tests.py`` puts scripts/ itself, not
# REPO_ROOT, on sys.path[0] until the sys.path fix near the top of this
# file runs, which happens before _main is ever called.
from scripts import test_green_ledger as ledger
# Computed from THIS module's REPO_ROOT (not test_green_ledger's own
# default) so a test that sandboxes ``rt.REPO_ROOT`` -- the established
# pattern for keeping reports/run-tests-latest.json real -- sandboxes
# the ledger the same way, automatically.
ledger_path = REPO_ROOT / "reports" / "test-green-ledger.json"
all_modules = _expand_paths(args.paths)
# RUL-039: exclude rehearsal-class modules (real deploy/incus/tofu surfaces)
# from gate runs. Explicitly-named files still run (a deliberate rehearsal);
# a missing/stale artifact fails closed — never silently exclude nothing.
rehearsal_excluded: list[str] = []
# RUL-039: the rehearsal exclusion + fail-closed staleness apply to GATE
# contexts only (--affected, the promotion plain suite). A deliberate
# subset/directory/file run names its modules explicitly and must not be
# blocked by a missing artifact (the brief: "naming a rehearsal module
# explicitly ... runs it").
if not args.include_rehearsal and (
args.affected is not None or requested_paths == ["tests/"]
):
rehearsal_set, rehearsal_err = _rehearsal_modules()
if rehearsal_set is None:
print(f"[run_tests] ✗ {rehearsal_err}", flush=True)
return _abort(
"rehearsal-artifact", extra={"rehearsal_error": rehearsal_err}
)
explicitly_named = {
str(Path(p))
for p in requested_paths
if Path(p).is_file() and p.endswith(".py")
}
rehearsal_excluded = [
m for m in all_modules if m in rehearsal_set and m not in explicitly_named
]
all_modules = [
m for m in all_modules if m not in rehearsal_set or m in explicitly_named
]
if rehearsal_excluded:
print(
f"[run_tests] 🎭 rehearsal-class excluded: "
f"{len(rehearsal_excluded)} module(s) (named at "
f"reports/rehearsal-class-modules.json)",
flush=True,
)
modules_to_run = all_modules
reused: dict[str, dict[str, Any]] = {}
if not args.force:
modules_to_run, reused = ledger.partition(
all_modules, repo_root=REPO_ROOT, path=ledger_path
)
print(
f"[run_tests] ♻ RUL-032: {len(reused)} module(s) unchanged since "
f"their last green run — reused, not re-run; "
f"{len(modules_to_run)} to run",
flush=True,
)
reused_test_count = sum(int(e.get("passed") or 0) for e in reused.values())
# 3. Run tests
pytest_defaults = ["-q", "--no-header", "--timeout=120"]
# run_tests.py is the full-suite runner: override pyproject.toml's default
# `-m "not slow and not llm_integration and not incus_live"` so slow tests
# are included, while llm_integration/incus_live remain opt-in special lanes.
pytest_args = (
pytest_defaults
+ ["-m", "not llm_integration and not incus_live"]
+ args.pytest_arg
)
if args.debug:
# Surface the DEBUG logs leading up to each failure through the batched
# runner: junit_logging=all writes captured logs into the JUnit report
# that _attribute_from_junit reads back per failing test.
os.environ.setdefault("LOG_LEVEL", "DEBUG")
pytest_args += ["--tb=long", "-o", "junit_logging=all"]
run_started_wall = time.time()
# RUL-038 (Mohamed, 2026-10-19): the test phase is SELECTED, COUNTED and
# CAPPED. The clock starts here -- after preflight and gates, before the
# selection count, which is test-phase work too. Every child pytest gets
# the sanctioned-runner stamp and the deadline; at the deadline each one
# kills itself (pytest_live_observer.arm_deadline), so the cap needs no
# cooperation from the executor's dispatch loop.
from orchestrata_ultimate.core.threading import pytest_live_reader as live
os.environ.setdefault(_RUN_ID_ENV, f"{int(time.time())}-{os.getpid()}")
cap_seconds = _test_cap_seconds()
deadline = _run_deadline(cap_seconds)
selected: int | None = 0
if modules_to_run:
selected = live.count_selected(
modules_to_run,
pytest_args,
env={**os.environ, **_child_run_env(deadline)},
chunks=args.workers,
timeout=max(1.0, deadline - time.time()),
)
print(
f"[run_tests] Selected: {'?' if selected is None else selected} tests "
f"in {len(modules_to_run)} modules ({len(reused)} modules reused from "
f"ledger) · hard cap {cap_seconds:.0f}s",
flush=True,
)
ticker = live.ProgressTicker(
os.environ.get(_RUN_ID_ENV, ""),
selected,
emit=lambda line: print(line, flush=True),
prefix="[run_tests] ",
).start()
# Recommendation #5 (plans/prolog-learning-subsystem-research-2026-08-12
# .md): bracket the EXECUTOR, not the whole process, so the core-seconds
# this produces describe the test phase, not infra preflight/gates. See
# _snapshot_container_cpu_before's docstring for the never-fail contract.
cpu_before = _snapshot_container_cpu_before()
capped = False
if modules_to_run:
try:
result = _run_suite(
paths=modules_to_run,
workers=args.workers,
timeout=args.timeout,
pytest_args=pytest_args,
json_output=args.json,
batch_multiplier=args.batch_multiplier,
solo_retry_cap=args.solo_retry_cap,
backfill=args.backfill,
deadline=deadline,
)
finally:
ticker.stop()
capped = bool(result.get("capped"))
# RUL-032: record every module that just went green (batch round or
# solo-recovered -- _run_suite already superseded a recovered
# module's result before returning) so a later run of THIS EXACT
# content never re-runs it.
failing_now = (
{
p
for p, r in result.get("modules", {}).items()
if r.get("returncode", 0) != 0
}
if args.json
# Non-JSON output: "modules" already holds ONLY the failures.
else set(result.get("modules", {}).keys())
)
passed_now = [p for p in modules_to_run if p not in failing_now]
if passed_now:
module_test_counts = result.get("module_test_counts") or {}
ledger.record_green(
{p: int(module_test_counts.get(p) or 0) for p in passed_now},
run_id=os.environ.get(_RUN_ID_ENV, f"{int(time.time())}-{os.getpid()}"),
head=head_commit,
repo_root=REPO_ROOT,
path=ledger_path,
)
if capped:
# The ledger write above IS the resume point: the next run reuses
# every module this one proved, and starts on what it never reached.
print(
f"[run_tests] ⛔ CAPPED at the {cap_seconds:.0f}s test-phase cap "
f"(RUL-038): {ticker.tick().removeprefix('[run_tests] ')} — "
f"{len(passed_now)} module(s) that passed are recorded in the "
"green ledger; re-run to continue from there.",
flush=True,
)
else:
ticker.stop()
# Every requested module was reused -- nothing to execute. Emit the
# same-shaped result the executor would, so the canonical report,
# --json, and ship's gate all see a normal run.
result = {
"test_totals": {"passed": 0, "failed": 0, "errors": 0, "skipped": 0},
"rerun": None,
"solo_retry": None,
"passed": 0,
"failed": 0,
"total": 0,
"time_seconds": 0.0,
"modules": {},
"module_test_counts": {},
}
# Fold the reused modules' recorded counts into the suite-level test
# totals -- so a run that reused everything still reports
# test_totals.passed > 0 (the false-green guard below, and ship's own
# gate, both depend on this) -- and name the reuse explicitly.
test_totals = result.get("test_totals") or {}
test_totals["passed"] = int(test_totals.get("passed") or 0) + reused_test_count
result["test_totals"] = test_totals
result["reused"] = {"modules": len(reused), "tests": reused_test_count}
# ADR-0105: execute the partitioned package parity suites in their own
# venvs and merge them into the same totals — a package failure fails
# the run exactly like a root-suite failure (the pre-push hook and
# ship's gate read these numbers).
if package_paths:
pkg_result = _run_package_suites(package_paths, timeout=args.timeout)
pkg_totals = pkg_result.get("test_totals") or {}
for key in ("passed", "failed", "errors", "skipped"):
test_totals[key] = int(test_totals.get(key) or 0) + int(
pkg_totals.get(key) or 0
)
result["test_totals"] = test_totals
result["modules"].update(pkg_result.get("modules") or {})
for key, value in (pkg_result.get("module_test_counts") or {}).items():
result["module_test_counts"][key] = value
result["failed"] = int(result.get("failed") or 0) + int(
pkg_result.get("failed") or 0
)
result["total"] = int(result.get("total") or 0) + int(
pkg_result.get("total") or 0
)
result["passed"] = int(result.get("passed") or 0) + int(
pkg_result.get("passed") or 0
)
cpu_report = _measure_container_cpu(cpu_before)
if cpu_report is not None:
core_seconds_by_container = ", ".join(
f"{name}={seconds:.3f}"
for name, seconds in cpu_report["core_seconds"].items()
)
print(
f"[run_tests] 🔥 container CPU (Memgraph/Redis -- excluded from "
f"the suite's own core-second accounting): "
f"{cpu_report['total_core_seconds']:.3f} core-s over "
f"{cpu_report['wall_seconds']:.1f}s wall ({core_seconds_by_container})",
flush=True,
)
# Phase 3.3: any test from THIS run over the 60s budget is a defect
# candidate, surfaced in the summary below (never a separate command).
# Tests marked known_slow are reported separately — reviewed, named
# slowness, not candidates.
slow_tests, acknowledged_slow = _slow_test_candidates(run_started_wall)
# 4. Report
result["slow_tests"] = [{"test": test, "duration": dur} for dur, test in slow_tests]
result["acknowledged_slow"] = [
{"test": test, "duration": dur, "reason": reason}
for dur, test, reason in acknowledged_slow
]
result["affected"] = affected_note
result["rehearsal_excluded"] = rehearsal_excluded
if affected_report:
# RUL-040: record the named_full verdicts (their paths) and any
# refused rows + the explicit decision, so the gate operator sees the
# NAMES — never a silent full run, never a silent nothing.
result["affected_details"] = affected_report
# Identity for the proven-run reuse check (gate dedup): a later
# invocation at this exact tree state with a covered scope reuses this
# report instead of re-running.
result["head_commit"] = head_commit
result["dirty_sig"] = dirty_sig
result["scope"] = "full" if scope_tests is None else scope_tests
# Computed BEFORE the summary line so its ✅/❌ agrees with the exit code.
# Reporting "✅ 0 tests passed ... 0/0 modules" and then failing on the
# guard below is two verdicts for one run.
_totals = result.get("test_totals") or {}
_executed = sum(
int(_totals.get(k) or 0) for k in ("passed", "failed", "errors", "skipped")
)
_write_canonical_report(result)
if args.json:
print(json.dumps(result, indent=2), file=json_stdout)
json_stdout.flush()
else:
elapsed = result["time_seconds"]
p, f = result["passed"], result["failed"]
tt = result.get("test_totals")
tests_part = (
f"{tt['passed']} tests passed, {tt['failed']} failed, "
f"{tt['errors']} errors, {tt['skipped']} skipped · "
if tt
else ""
)
solo = result.get("solo_retry")
solo_note = (
" · all batch failures recovered solo"
if solo and solo["recovered"] and not solo["still_failed"]
else ""
)
print(
f"\n[run_tests] {'✅' if f == 0 and _executed else '❌'} "
f"{tests_part}"
f"{p}/{p + f} modules "
f"in {elapsed:.0f}s ({elapsed/60:.1f}m)"
f"{solo_note}",
flush=True,
)
if affected_note:
print(f"[run_tests] ⚠ --affected: {affected_note}", flush=True)
if slow_tests:
print(
f"[run_tests] 🐢 {len(slow_tests)} test(s) over 60s — "
"defect candidate(s):",
flush=True,
)
for dur, test in slow_tests:
print(f"[run_tests] {dur:.1f}s {test}", flush=True)
if acknowledged_slow:
print(
f"[run_tests] 🐢 {len(acknowledged_slow)} known-slow "
"(acknowledged via pytest.mark.known_slow):",
flush=True,
)
for dur, test, reason in acknowledged_slow:
print(f"[run_tests] {dur:.1f}s {test} — {reason}", flush=True)
# Two-round transparency: name what failed in batch, what recovered
# in isolation (= load/order flake, logged for hunting), and what
# failed BOTH ways (= a real defect). Test-count totals above are
# the batch round's; module pass/fail counts are post-retry.
if solo:
print(
f"[run_tests] 🔁 solo retry: {len(solo['attempted'])} "
f"batch-failed module(s) re-run in isolation — "
f"{len(solo['recovered'])} recovered (load/order flakes), "
f"{len(solo['still_failed'])} still failed "
f"· COST {solo.get('duplicated_tests', 0)} test(s) executed "
f"twice, {solo.get('seconds', 0)}s serial tail",
flush=True,
)
for m in solo["recovered"]:
print(
f"[run_tests] recovered solo "
f"(logged to .flake_recoveries.ndjson): {m}",
flush=True,
)
# Phase-2B safety net: modules a first-pass batch dropped ("did not
# run" — a session-timeout/SIGKILL overrun, not a real failure) get
# ONE automatic rerun round. Surface it explicitly so a rerun never
# silently launders a genuine packing/imbalance regression into a
# quiet green run — a growing rerun count across runs IS the signal
# that LPT packing needs retuning (see test_executor.py's module
# docstring for the 2026-07-08 diagnosis this closes the loop on).
rerun = result.get("rerun")
if rerun and rerun.get("count"):
still_failed = rerun.get("still_failed") or []
print(
f"[run_tests] 🔁 rerun round: {rerun['count']} module(s) had "
f"no result after the first pass — "
f"{len(rerun.get('passed') or [])} passed on rerun, "
f"{len(still_failed)} still failed"
+ (f": {', '.join(still_failed)}" if still_failed else ""),
flush=True,
)
# Name each failing module and show its real per-test failure (message +
# traceback lifted from JUnit) so failures are self-explanatory without a
# re-run. Falls back to the batch tail for modules that never ran.
for mod, r in result.get("modules", {}).items():
if r.get("returncode", 0) != 0:
detail = (r.get("stdout") or "").strip()
print(f"\n[run_tests] ❌ FAILED {mod}", flush=True)
# Surface the executor's error verdict FIRST — a killed batch
# ("Timeout after Ns") otherwise shows only a truncated dot
# stream, and a whole-batch mass-failure reads like 250 real
# test failures (the 2026-07-06 forensics cost hours).
if r.get("error"):
print(f"[run_tests] batch error: {r['error']}", flush=True)
# A batch killed by a signal (e.g. the 2026-07-06 SIGSEGV in
# 3.14t's match_dealloc) reports its cause ONLY on stderr —
# pytest's faulthandler dump naming the running test lands
# there. Never swallow it.
stderr_tail = (r.get("stderr") or "").strip()
if stderr_tail:
print(
f"[run_tests] batch stderr:\n{stderr_tail[-2000:]}",
flush=True,
)
if detail:
print(detail[-4000:], flush=True)
# 5. Cleanup: services are managed by start_test_services.py atexit
# (CLEANUP_KERNEL=1 triggers cleanup for kernel + gateway)
# A run that executed NO tests is never a pass, whatever the module
# tallies say. Measured 2026-08-07: `--json <path>` was invoked with a
# path argument, but --json is store_true, so argparse handed the path to
# the POSITIONAL test-target list. The run targeted one nonexistent
# module, collected 0 tests, and exited 0 — a green gate that had tested
# nothing. Module accounting cannot catch that class on its own: zero
# failing modules and zero executed tests produce the same tally.
if _executed == 0:
print(
"[run_tests] ❌ FALSE GREEN GUARD: the run executed 0 tests. "
"A suite run that collects nothing is INVALID, never a pass. "
"Check the test targets — an unknown path (or a value handed to "
"a store_true flag like --json) is consumed as a positional "
"target and silently narrows the run to nothing.",
flush=True,
)
return 1
# A run that was stopped by the test-phase cap (RUL-038) exits non-zero
# even if no tests failed, because modules were never reached.
# The 'capped' flag is set in the result when _run_suite encounters a
# timed-out batch and records "capped": True in the result.
if capped:
return 1
return 0 if result["failed"] == 0 else 1
if __name__ == "__main__":
sys.exit(main())