runner.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/testing/runner.py
 
"""
The §20 stand-in orchestrator: serial in-process execution (Plan 184).
 
Purpose
-------
Drive one stand-in invocation end to end: argv parse → ADR-T-12
discovery → §10 undeclared-root scan → profile/marker/seed composition
→ serial IN-PROCESS ``pytest.main`` with a collector plugin → §20.2
process-level CWD/env restore checks → §11 report assembly → §4 exit
mapping. In-process execution is the pinned decision: only it lets the
post-run ``os.environ``/CWD diff observe anything.
 
Responsibilities
----------------
- ``run_standin`` — the runner API; returns report + exit code + the
  API-level execution-order witness (the report's ``tests`` array is
  nodeid-sorted by contract, so order lives here).
- ``_Collector`` — the pytest plugin: collection accounting, static
  scan before execution, dry-run/violation short-circuit, per-test
  watchdog, outcome aggregation, collection-error capture.
- ``_SignalState`` — the ADR-T-08 subset for a serial in-process run.
 
Signal-handling subset (ADR-T-08, stated per plan T3)
-----------------------------------------------------
First ``SIGINT``/``SIGTERM`` (plus ``SIGBREAK`` on Windows): there are
no worker processes to TERM and nothing to wait ten seconds for, so the
stand-in unwinds the in-process pytest session, writes the partial
report with ``status: "orchestrator_interrupted"`` and the
``ORCHESTRATOR_INTERRUPTED`` signal entry, and exits 3. A second signal
hard-exits 130 with no report (the report-less second-signal wording
carries a known erratum upstream, B_04 P2 — out of scope here).
Handlers install only when running on the main thread.
 
Per-test-timeout deviation (ADR-T-07/§16.1, stated per plan T3)
---------------------------------------------------------------
The spec pins ``pytest-timeout`` as the serial stand-in's mechanism,
but its only Windows-viable method (``thread``) terminates the ENTIRE
process on expiry — for an in-process stand-in that would kill the
orchestrator and make the §20 surviving-report contract (timeout
outcome + duration + exit 1) unsatisfiable. The stand-in therefore
implements the ADR's intent with a watchdog timer that soft-interrupts
the hung test via ``PyThreadState_SetAsyncExc`` (the run continues;
the nodeid reports ``outcome: "timeout"`` with its elapsed duration).
``pytest-timeout`` remains the installed, pinned dependency and the
v1.0 in-worker backstop. Residual: a test blocked inside a C call is
interrupted only when that call returns — the same class of residual
pytest-timeout's signal method carries.
 
Diagnostics
-----------
Domain: OC_TESTING
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- The restore checks consume :data:`TRACKED_ENV_PREFIXES` from its
  single export (§20.2) — object identity, no copy.
- The runner restores its own CWD and the recursion-guard variable
  before diffing, so only TEST leaks are visible.
- ``report["exit_code"]`` always equals the returned exit code.
 
Dependencies
------------
- pytest (in-process engine)
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.testing.cli (argv contract)
- oc_chat.testing.package_walk (discovery + root scan)
- oc_chat.testing.static_scan (static AST guard)
- oc_chat.testing.report (report constants + exit mapping)
- oc_chat.testing.env_inventory (tracked prefixes)
- oc_chat.testing.scheduler (the parallel path - worker plan, unit schedule, bounded child pool and crash reconciliation)
- _pytest.mark.expression (lazy; Expression - validates the composed marker expression before the run)
- pytest_randomly (not imported; the seed-pinning plugin loaded with -p randomly when --seed is given, disabled with -p no:randomly otherwise)
"""
from __future__ import annotations
 
from collections.abc import Generator, Iterator, Sequence
import ctypes
from contextlib import redirect_stdout
from dataclasses import dataclass, field
import io
import os
from pathlib import Path
import signal
import threading
import time
import uuid
 
import pytest
 
from oc_diagnostics import _dbg
 
from oc_chat.testing.cli import StandInConfig, parse_argv
from oc_chat.testing.env_inventory import TRACKED_ENV_PREFIXES
from oc_chat.testing.package_walk import (
    WalkResult,
    scan_undeclared_roots,
    walk_for_package,
)
from oc_chat.testing.report import (
    SCHEMA_STAMP,
    SUPPORTED_FEATURES,
    compute_exit_and_status,
    make_error,
    make_violation,
)
from oc_chat.testing.scheduler import (
    DIST_MODE_PARALLEL,
    DIST_MODE_SERIAL,
    SchedulerUnavailable,
    UnitOutcome,
    WHOLE_FILE_MARKER,
    WorkerPlan,
    build_schedule,
    clamp_error,
    crash_error,
    crash_violation,
    load_groups,
    manifest_path_for,
    resolve_worker_plan,
    run_units,
    unclassified_violation,
)
from oc_chat.testing.static_scan import scan_files
 
#: Recursion guard: set while the stand-in's in-process pytest session
#: runs, so live-suite acceptance tests can skip their own heavy
#: stand-in invocation instead of recursing.
STANDIN_ACTIVE_ENV = "OC_CHAT_STANDIN_ACTIVE"
 
_PROFILE_EXPRESSIONS = {
    "strict": "not requires_network",
    "smoke": "smoke and not slow and not requires_mcp and not requires_network",
    # The §5 network-exclusion lift needs package config
    # (allow_network=true) that no package declares yet; without it the
    # exclusion can never lift, so the stand-in composes the same base
    # exclusion as strict.
    "full": "not requires_network",
}
 
_LONGREPR_CAP = 2000
 
 
class StandInTimeoutError(Exception):
    """Raised asynchronously inside a hung test by the watchdog."""
 
 
@dataclass
class RunResult:
    """The runner API result: report document, exit code, witnesses."""
 
    report: dict
    exit_code: int
    execution_order: list[str] = field(default_factory=list)
    captured_output: str = ""
 
 
@dataclass
class SessionOutcome:
    """One guarded in-process pytest session (Plan 240 extraction).
 
    The serial stand-in and every scheduler child run through the SAME
    function, so the §20.2 restore contract, the watchdog and the
    collector plugin cannot drift apart between the two paths.
    """
 
    rc: int = 0
    records: dict[str, dict] = field(default_factory=dict)
    timed_out: tuple[str, ...] = ()
    execution_order: list[str] = field(default_factory=list)
    collected_pre: list[tuple[str, Path]] = field(default_factory=list)
    selected: list[tuple[str, Path]] = field(default_factory=list)
    #: Root-relative paths whose tests carried ``whole_file`` at
    #: collection: the Plan 286 P2 opt-out the scheduler reads. A child
    #: is handed nodeids and never sees a mark, so the PARENT's collect
    #: pass is the only place this is observable.
    whole_file: tuple[str, ...] = ()
    collect_errors: list[str] = field(default_factory=list)
    static_violations: list[dict] = field(default_factory=list)
    blocked: bool = False
    violations: list[dict] = field(default_factory=list)
    output: str = ""
 
 
class _SignalState:
    """ADR-T-08 subset state machine for a serial in-process run."""
 
    def __init__(self) -> None:
        self.interrupted: str | None = None
        self._entered = False
        self.hard_exit = os._exit  # seam: tests monkeypatch this
 
    def on_signal(self, signum: int, frame: object) -> None:
        """First signal: record + unwind pytest. Second: exit 130."""
        func = "on_signal"
        if self._entered:
            _dbg("OC_TESTING", func, "second signal -> hard exit 130", 2)
            self.hard_exit(130)
            return
        self._entered = True
        try:
            self.interrupted = signal.Signals(signum).name
        except ValueError:
            self.interrupted = f"signal {signum}"
        _dbg("OC_TESTING", func, f"first signal {self.interrupted}", 2)
        raise KeyboardInterrupt
 
 
class _Collector:
    """In-process pytest plugin: accounting, static gate, watchdog."""
 
    def __init__(self, *, dry_run: bool, timeout: int | None,
                 thread_ident: int) -> None:
        self.dry_run = dry_run
        self.timeout = timeout
        self.thread_ident = thread_ident
        self.collected_pre: list[tuple[str, Path]] = []
        self.selected: list[tuple[str, Path]] = []
        self.whole_file: tuple[str, ...] = ()
        self.static_violations: list[dict] = []
        self.blocked = False
        self.records: dict[str, dict] = {}
        self.execution_order: list[str] = []
        self.timed_out: set[str] = set()
        self.collect_errors: list[str] = []
        self._current_nodeid: str | None = None
        self.session_start_cwd: str | None = None
        self.last_seen_cwd: str | None = None
 
    # -- session CWD observation (§20.2) ---------------------------------
    # The engine (pytest 8+/plugins) restores the invocation CWD during
    # session teardown, which masks any test's chdir leak from a diff
    # taken after pytest.main() returns — verified empirically: the
    # leaked CWD is visible at the last test's teardown report and
    # already restored by sessionfinish. The check therefore observes
    # the CWD at every test-phase report; the final observation is the
    # post-run state "around the process-level run", taken before the
    # engine's cosmetic restoration.
 
    def pytest_sessionstart(self, session: pytest.Session) -> None:
        self.session_start_cwd = os.getcwd()
 
    # -- collection ------------------------------------------------------
 
    @pytest.hookimpl(wrapper=True)
    def pytest_collection_modifyitems(
        self, config: pytest.Config, items: list[pytest.Item],
    ) -> Iterator[None]:
        self.collected_pre = [(i.nodeid, Path(i.path)) for i in items]
        yield
        # post-deselection view is taken at collection_finish
 
    def pytest_collection_finish(self, session: pytest.Session) -> None:
        self.selected = [(i.nodeid, Path(i.path)) for i in session.items]
        # Plan 286 P2: the opt-out is a MARK, and only a collection
        # sees marks. Recorded by nodeid prefix, which is the same
        # root-relative key the scheduler groups by.
        self.whole_file = tuple(sorted({
            item.nodeid.split("::")[0] for item in session.items
            if item.get_closest_marker(WHOLE_FILE_MARKER) is not None
        }))
 
    def pytest_collectreport(self, report: pytest.CollectReport) -> None:
        if report.failed:
            self.collect_errors.append(
                str(report.longrepr)[:_LONGREPR_CAP]
            )
 
    # -- execution gate --------------------------------------------------
 
    @pytest.hookimpl(tryfirst=True)
    def pytest_runtestloop(self, session: pytest.Session) -> bool | None:
        func = "pytest_runtestloop"
        files = sorted({path for _n, path in self.selected})
        self.static_violations = scan_files(files)
        has_errors = any(
            v["severity"] == "error" for v in self.static_violations
        )
        if has_errors:
            _dbg("OC_TESTING", func, "static violations -> skip execution", 2)
            self.blocked = True
            return True
        if self.dry_run:
            _dbg("OC_TESTING", func, "dry-run -> skip execution", 3)
            return True
        return None
 
    # -- per-test watchdog ----------------------------------------------
 
    def _fire(self, nodeid: str) -> None:
        func = "_fire"
        if self._current_nodeid != nodeid:
            return  # test already finished; never interrupt a neighbour
        self.timed_out.add(nodeid)
        _dbg("OC_TESTING", func, f"timeout -> soft interrupt {nodeid}", 2)
        ctypes.pythonapi.PyThreadState_SetAsyncExc(
            ctypes.c_ulong(self.thread_ident),
            ctypes.py_object(StandInTimeoutError),
        )
 
    @pytest.hookimpl(wrapper=True)
    def pytest_runtest_protocol(
        self, item: pytest.Item, nextitem: pytest.Item | None,
    ) -> Generator[None, object, object]:
        timer: threading.Timer | None = None
        self._current_nodeid = item.nodeid
        if self.timeout:
            timer = threading.Timer(
                self.timeout, self._fire, args=(item.nodeid,)
            )
            timer.daemon = True
            timer.start()
        try:
            result = yield
        finally:
            self._current_nodeid = None
            if timer is not None:
                timer.cancel()
        return result
 
    # -- outcome aggregation ---------------------------------------------
 
    def pytest_runtest_logreport(self, report: pytest.TestReport) -> None:
        self.last_seen_cwd = os.getcwd()
        rec = self.records.setdefault(report.nodeid, {
            "failed": False, "skipped": False,
            "duration": 0.0, "longrepr": None,
        })
        if report.nodeid not in self.execution_order:
            self.execution_order.append(report.nodeid)
        rec["duration"] += getattr(report, "duration", 0.0)
        if report.failed:
            rec["failed"] = True
            rec["longrepr"] = str(report.longrepr)[:_LONGREPR_CAP]
        elif report.skipped:
            rec["skipped"] = True
 
 
def _outcome_for(nodeid: str, rec: dict, timed_out: set[str]) -> str:
    if nodeid in timed_out:
        return "timeout"
    if rec["failed"]:
        return "failed"
    if rec["skipped"]:
        return "skipped"
    return "passed"
 
 
def run_session(
    *,
    pytest_args: Sequence[str],
    root: Path,
    timeout: int | None,
    dry_run: bool,
    sig_state: _SignalState | None = None,
    install_signal_handlers: bool = True,
    violation_file: str | None = None,
    violation_scope: str = "",
) -> SessionOutcome:
    """Run ONE guarded in-process pytest session (§20.2).
 
    Everything the §20.2 process-level contract needs happens here and
    only here: the recursion guard, the CWD move and its restoration,
    the tracked-environment snapshot taken around the run, the static
    gate, the per-test watchdog and the leak diff. The serial stand-in
    calls it once; the scheduler calls it once per child, in that
    child's own process, which is what makes the §20.2 findings
    attributable to a group at all.
 
    Parameters
    ----------
    violation_file
        The ``file`` field of any runtime finding; defaults to *root*.
    violation_scope
        Prefixed to a finding's message when non-empty, so a child's
        leak names the worker and group that produced it.
    """
    func = "run_session"
    collector = _Collector(
        dry_run=dry_run, timeout=timeout,
        thread_ident=threading.get_ident(),
    )
    prior_handlers: list[tuple[int, object]] = []
    on_main_thread = threading.current_thread() is threading.main_thread()
    install = (
        install_signal_handlers and on_main_thread and sig_state is not None
    )
    outer_cwd = os.getcwd()
    guard_prior = os.environ.get(STANDIN_ACTIVE_ENV)
    env_before = {
        k: v for k, v in os.environ.items()
        if k.startswith(TRACKED_ENV_PREFIXES)
    }
    buffer = io.StringIO()
    rc: int
    try:
        if install and sig_state is not None:
            for signame in ("SIGINT", "SIGTERM", "SIGBREAK"):
                signum = getattr(signal, signame, None)
                if signum is not None:
                    prior_handlers.append(
                        (signum, signal.signal(signum, sig_state.on_signal))
                    )
        os.environ[STANDIN_ACTIVE_ENV] = "1"
        os.chdir(root)
        ref_cwd = os.getcwd()
        _dbg("OC_TESTING", func, f"pytest args: {list(pytest_args)}", 3)
        with redirect_stdout(buffer):
            try:
                rc = int(pytest.main(list(pytest_args), plugins=[collector]))
            except KeyboardInterrupt:
                rc = 2
        post_cwd = os.getcwd()
    finally:
        for signum, handler in prior_handlers:
            signal.signal(signum, handler)
        os.chdir(outer_cwd)
        if guard_prior is None:
            os.environ.pop(STANDIN_ACTIVE_ENV, None)
        else:
            os.environ[STANDIN_ACTIVE_ENV] = guard_prior
 
    env_after = {
        k: v for k, v in os.environ.items()
        if k.startswith(TRACKED_ENV_PREFIXES)
    }
 
    # -- runtime restore findings (run-level; file = the package root) ----
    # Primary CWD observation: session start vs session finish (inside
    # the engine — pytest 8+ restores the invocation CWD after the
    # session, masking leaks from any later diff). The outer ref/post
    # diff stays as the fallback net for paths where no session ran.
    where = violation_file if violation_file is not None else str(root)
    tag = f"[{violation_scope}] " if violation_scope else ""
    violations: list[dict] = []
    cwd_leak: tuple[str, str] | None = None
    start_cwd = collector.session_start_cwd or ref_cwd
    if (
        collector.last_seen_cwd is not None
        and collector.last_seen_cwd != start_cwd
    ):
        cwd_leak = (start_cwd, collector.last_seen_cwd)
    elif post_cwd != ref_cwd:
        cwd_leak = (ref_cwd, post_cwd)
    if cwd_leak is not None:
        violations.append(make_violation(
            "cwd_chdir", where, None, "error",
            f"{tag}process CWD not restored around the run: "
            f"{cwd_leak[0]!r} -> {cwd_leak[1]!r} (section 20.2 "
            "process-level check)",
        ))
    leaked = sorted(
        set(env_after) - set(env_before)
        | {k for k in env_before if k in env_after
           and env_before[k] != env_after[k]}
    )
    removed = sorted(set(env_before) - set(env_after))
    if leaked or removed:
        details = []
        if leaked:
            details.append("set/changed: " + ", ".join(leaked))
        if removed:
            details.append("removed: " + ", ".join(removed))
        violations.append(make_violation(
            "env_mutation", where, None, "error",
            f"{tag}tracked environment not restored around the run "
            f"({'; '.join(details)}) — prefixes {TRACKED_ENV_PREFIXES} "
            "(section 20.2 process-level check)",
        ))
    for key in leaked:
        os.environ.pop(key, None)
    for key in removed:
        os.environ[key] = env_before[key]
 
    return SessionOutcome(
        rc=rc,
        records=collector.records,
        timed_out=tuple(sorted(collector.timed_out)),
        execution_order=list(collector.execution_order),
        collected_pre=list(collector.collected_pre),
        selected=list(collector.selected),
        whole_file=collector.whole_file,
        collect_errors=list(collector.collect_errors),
        static_violations=list(collector.static_violations),
        blocked=collector.blocked,
        violations=violations,
        output=buffer.getvalue(),
    )
 
 
def _build_report(
    *,
    run_id: str,
    status: str,
    exit_code: int,
    workspace_root: str | None,
    workspace_mode: str | None,
    packages: list[str],
    profile: str | None,
    marker: str | None,
    seed: int | None,
    nodeids: list[str],
    workers_requested: int | str | None,
    workers_actual: int | None,
    dist_mode: str | None,
    collected: int,
    tests: list[dict],
    violations: list[dict],
    errors: list[dict],
    duration_seconds: float,
    scheduling: dict | None = None,
) -> dict:
    """Assemble the §11/§20.3 report document (every field, always)."""
    passed = sum(1 for t in tests if t["outcome"] == "passed")
    failed = sum(1 for t in tests if t["outcome"] == "failed")
    skipped = sum(1 for t in tests if t["outcome"] == "skipped")
    timeout = sum(1 for t in tests if t["outcome"] == "timeout")
    return {
        "schema": SCHEMA_STAMP,
        "run_id": run_id,
        "status": status,
        "exit_code": exit_code,
        "workspace": {"root": workspace_root, "mode": workspace_mode},
        "selection": {
            "packages": packages,
            "profile": profile,
            "marker": marker,
            "seed": seed,
            "nodeids": nodeids,
            "pytest_args": [],
        },
        "workers": {
            "requested": workers_requested,
            "actual": workers_actual,
            "dist_mode": dist_mode,
        },
        "summary": {
            "collected": collected,
            "passed": passed,
            "failed": failed,
            "skipped": skipped,
            "timeout": timeout,
            "duration_seconds": round(duration_seconds, 2),
            "tests_count": len(tests),
            "tests_truncated": False,
        },
        "tests": sorted(tests, key=lambda t: t["nodeid"]),
        "contract_violations": violations,
        "artifacts": {"junit_xml": None, "ndjson": None,
                      "ndjson_stream": None},
        "errors": errors,
        "supported_features": _features_for(dist_mode),
        "scheduling": scheduling if scheduling is not None
        else serial_scheduling("not_started"),
    }
 
 
def _features_for(dist_mode: str | None) -> dict[str, str]:
    """§20.3 ``supported_features`` for THIS run.
 
    ADR-T-11's v0.3 amendment (readiness audit A-19) makes exactly one
    value run-dependent: a serial stand-in never runs the ``auto``
    algorithm and reports ``workers_auto: "deferred"``, while a
    parallel-capable one reports ``"enforced"``. Plan 240 is what makes
    the second case reachable; the pinned constant stays the serial
    baseline, so a serial report is unchanged from plan 184's.
    """
    features = dict(SUPPORTED_FEATURES)
    if dist_mode == DIST_MODE_PARALLEL:
        features["workers_auto"] = "enforced"
    return features
 
 
def serial_scheduling(reason: str | None) -> dict:
    """The ``scheduling`` block of a run that distributed nothing.
 
    *reason* records WHY, so a run that could have parallelised and did
    not is never silent about it: ``workers_1``, ``dry_run``,
    ``no_manifest``, ``single_worker``, ``single_unit``,
    ``classifier_unavailable`` or ``worker_cap_unusable``.
    """
    return {
        "mode": "serial",
        "reason": reason,
        "manifest": None,
        "workers_cap": None,
        "workers_resolved": None,
        "groups": [],
        "unclassified": [],
    }
 
 
def _pre_execution_result(
    *,
    run_id: str,
    t0: float,
    errors: list[dict],
    cfg: StandInConfig | None,
    walk: WalkResult | None,
    failed_step: str,
) -> RunResult:
    """Build a §11 pre-execution failure report along the pinned
    resolution chain argv → workspace → package → profile → seed →
    workers: every leaf at-or-after *failed_step* is null."""
    order = ("argv", "workspace", "package", "profile", "seed", "workers")
    reached = order.index(failed_step)
 
    def _resolved(owning_step: str) -> bool:
        """A field is non-null iff its owning step completed before the
        failed one."""
        return order.index(owning_step) < reached
 
    workspace_root = str(walk.root) if walk and _resolved("workspace") else None
    workspace_mode = (
        "single_package" if walk and _resolved("workspace") else None
    )
    packages = [walk.package_id] if walk and _resolved("package") else []
    profile = cfg.profile if cfg and _resolved("profile") else None
    seed = None
    workers_requested = cfg.workers if cfg and _resolved("workers") else None
    report = _build_report(
        run_id=run_id,
        status="configuration_error",
        exit_code=4,
        workspace_root=workspace_root,
        workspace_mode=workspace_mode,
        packages=packages,
        profile=profile,
        marker=cfg.marker if cfg else None,
        seed=seed,
        nodeids=list(cfg.nodeids) if cfg else [],
        workers_requested=workers_requested,
        workers_actual=None,
        dist_mode=None,
        collected=0,
        tests=[],
        violations=[],
        errors=errors,
        duration_seconds=time.monotonic() - t0,
    )
    return RunResult(report=report, exit_code=4)
 
 
def run_standin(argv: list[str], cwd: Path | None = None) -> RunResult:
    """Run one stand-in invocation; never raises for contract-covered
    failures (internal surprises become a 9002 orchestrator error)."""
    func = "run_standin"
    t0 = time.monotonic()
    run_id = str(uuid.uuid4())
    base_cwd = Path(cwd) if cwd is not None else Path.cwd()
    _dbg("OC_TESTING", func, f"argv={argv} cwd={base_cwd}", 2)
    try:
        return _run(argv, base_cwd, run_id, t0)
    except SystemExit:
        raise
    except BaseException as exc:  # 9002: never a bare traceback surface
        _dbg("OC_TESTING", func, f"internal error: {exc!r}", 1)
        errors = [make_error(
            "orchestrator", "9002", "fatal",
            f"stand-in internal error: {type(exc).__name__}: "
            f"{str(exc)[:_LONGREPR_CAP]}",
        )]
        report = _build_report(
            run_id=run_id, status="orchestrator_error", exit_code=3,
            workspace_root=None, workspace_mode=None, packages=[],
            profile=None, marker=None, seed=None, nodeids=[],
            workers_requested=None, workers_actual=None, dist_mode=None,
            collected=0, tests=[], violations=[], errors=errors,
            duration_seconds=time.monotonic() - t0,
        )
        return RunResult(report=report, exit_code=3)
 
 
def _run(argv: list[str], base_cwd: Path, run_id: str, t0: float) -> RunResult:
    func = "_run"
    cfg, parse_errors = parse_argv(argv)
 
    if parse_errors:
        code = parse_errors[0]["code"]
        walk: WalkResult | None = None
        if code == "PROFILE_INVALID":
            walk, _werr = walk_for_package(base_cwd)
            failed_step = "profile"
        elif code == "9003":
            failed_step = "workspace"
        else:
            failed_step = "argv"
        return _pre_execution_result(
            run_id=run_id, t0=t0, errors=parse_errors, cfg=cfg,
            walk=walk, failed_step=failed_step,
        )
 
    assert cfg is not None
    walk, walk_error = walk_for_package(base_cwd)
    if walk_error is not None:
        return _pre_execution_result(
            run_id=run_id, t0=t0, errors=[walk_error], cfg=cfg,
            walk=None, failed_step="workspace",
        )
    assert walk is not None
 
    if cfg.package is not None and cfg.package != walk.package_id:
        error = make_error(
            "configuration", "PACKAGE_UNKNOWN", "fatal",
            f"--package {cfg.package!r} does not match the discovered "
            f"package {walk.package_id!r} (single-package fallback; no "
            "workspace file in scope)",
        )
        return _pre_execution_result(
            run_id=run_id, t0=t0, errors=[error], cfg=cfg,
            walk=walk, failed_step="package",
        )
 
    profile = cfg.profile if cfg.profile_explicit else (
        walk.profile_pin if walk.profile_pin in _PROFILE_EXPRESSIONS
        else "strict"
    )
    seed = cfg.seed if cfg.seed is not None else 0
 
    errors: list[dict] = []
    violations: list[dict] = []
 
    def _resolved_report(status: str, exit_code: int, *, collected: int = 0,
                         tests: list[dict] | None = None) -> RunResult:
        report = _build_report(
            run_id=run_id, status=status, exit_code=exit_code,
            workspace_root=str(walk.root), workspace_mode="single_package",
            packages=[walk.package_id], profile=profile, marker=cfg.marker,
            seed=seed, nodeids=list(cfg.nodeids),
            workers_requested=cfg.workers, workers_actual=1,
            dist_mode="serial", collected=collected,
            tests=tests or [], violations=violations, errors=errors,
            duration_seconds=time.monotonic() - t0,
        )
        return RunResult(report=report, exit_code=exit_code)
 
    # -- §10 undeclared-root scan (fatal in strict, incl. --dry-run) -----
    undeclared = scan_undeclared_roots(
        walk.root, walk.roots, walk.python_files
    )
    if undeclared:
        severity = "fatal" if profile == "strict" else "warning"
        for directory in undeclared:
            try:
                shown = directory.relative_to(walk.root)
            except ValueError:
                shown = directory
            errors.append(make_error(
                "configuration", "UNDECLARED_TEST_ROOT", severity,
                f"undeclared test root: {shown} contains test files but "
                "is absent from the effective test roots (testpaths; the "
                "A-18 class)",
            ))
        if profile == "strict":
            return _resolved_report("configuration_error", 4)
 
    # -- marker composition + validation ---------------------------------
    profile_expr = _PROFILE_EXPRESSIONS[profile]
    effective_expr = (
        f"({profile_expr}) and ({cfg.marker})" if cfg.marker else profile_expr
    )
    try:
        from _pytest.mark.expression import Expression
        try:
            Expression.compile(effective_expr)
        except SyntaxError as exc:
            # pytest 9 raises SyntaxError here (its pre-9 ParseError is
            # gone).
            errors.append(make_error(
                "configuration", "MARKER_PARSE_ERROR", "fatal",
                f"marker expression {cfg.marker!r} failed to parse when "
                f"composed with the {profile} profile: {exc}",
            ))
            return _resolved_report("configuration_error", 4)
    except ImportError:  # pragma: no cover - pytest internals moved
        _dbg("OC_TESTING", func, "Expression API unavailable; deferring "
             "marker validation to pytest", 2)
 
    # -- root existence ---------------------------------------------------
    root_pairs = list(zip(walk.testpaths or ["."], walk.roots))
    existing_pairs = [(tp, rp) for tp, rp in root_pairs if rp.is_dir()]
    if not cfg.nodeids:
        for tp, rp in root_pairs:
            if not rp.is_dir():
                errors.append(make_error(
                    "configuration", "EMPTY_TEST_ROOT", "warning",
                    f"declared test root {tp!r} does not exist (stale "
                    "declaration)",
                ))
        if not existing_pairs:
            errors.append(make_error(
                "configuration", "NO_TESTS_COLLECTED", "fatal",
                "the union of the effective test roots contains no "
                "existing directories; the configuration points at no "
                "tests",
            ))
            return _resolved_report("configuration_error", 4)
 
    # -- compose pytest args ----------------------------------------------
    # The BASE args carry everything but the selection: a scheduler child
    # appends its own unit's nodeids instead of the run's roots, so the
    # two paths compose one identical prefix.
    base_args = ["-c", str(walk.config_file), "--rootdir", str(walk.root)]
    if cfg.seed is not None:
        base_args += ["-p", "randomly", f"--randomly-seed={cfg.seed}"]
    else:
        base_args += ["-p", "no:randomly"]
    base_args += ["-m", effective_expr]
    if cfg.nodeids:
        selection = list(cfg.nodeids)
    else:
        selection = [tp for tp, _rp in existing_pairs]
    pytest_args = base_args + selection
 
    # -- decide the execution path (Plan 240 P2/P3) -----------------------
    sig_state = _SignalState()
    workspace_root = walk.root.parent
    worker_plan: WorkerPlan | None = None
    manifest_path: Path | None = None
    serial_reason: str | None = None
 
    if cfg.dry_run:
        serial_reason = "dry_run"
    elif cfg.workers == 1:
        serial_reason = "workers_1"
    else:
        try:
            manifest_path = manifest_path_for(
                walk.root, walk.testpaths, workspace_root,
            )
        except SchedulerUnavailable as exc:
            _dbg("OC_TESTING", func, f"classifier unavailable: {exc}", 2)
            manifest_path = None
            serial_reason = "classifier_unavailable"
        if manifest_path is not None and not manifest_path.is_file():
            _dbg("OC_TESTING", func,
                 f"no manifest at {manifest_path}; serial", 2)
            manifest_path = None
            serial_reason = "no_manifest"
        if manifest_path is not None:
            try:
                worker_plan = resolve_worker_plan(
                    cfg.workers, workspace_root, walk.package_id,
                )
            except SchedulerUnavailable as exc:
                _dbg("OC_TESTING", func, f"worker cap unusable: {exc}", 1)
                manifest_path = None
                serial_reason = "worker_cap_unusable"
    if worker_plan is not None:
        if worker_plan.clamped:
            errors.append(clamp_error(worker_plan))
        if worker_plan.resolved <= 1:
            serial_reason = "single_worker"
 
    # -- run: the scheduler when it can, the plan-184 serial path else ----
    scheduling = serial_scheduling(serial_reason)
    workers_actual = 1
    dist_mode = DIST_MODE_SERIAL
    session: SessionOutcome | None = None
    unit_outcomes: list[UnitOutcome] = []
    pool_interrupted = False
    scheduled = False
    units: tuple = ()
 
    if serial_reason is None and worker_plan is not None:
        assert manifest_path is not None
        collect = run_session(
            pytest_args=pytest_args, root=walk.root,
            timeout=cfg.per_test_timeout, dry_run=True,
            sig_state=sig_state, violation_file=str(walk.root),
        )
        nodeids_by_file: dict[str, list[str]] = {}
        for nodeid, _path in collect.selected:
            nodeids_by_file.setdefault(nodeid.split("::")[0], []).append(
                nodeid,
            )
        missing: tuple[str, ...] = ()
        try:
            groups, missing = load_groups(
                manifest_path, nodeids_by_file, workspace_root,
            )
        except SchedulerUnavailable as exc:
            groups = {}
            serial_reason = "classifier_unavailable"
            _dbg("OC_TESTING", func, f"classifier vanished: {exc}", 1)
        except Exception as exc:  # EXTERNAL_ERROR(L2) - unreadable manifest
            groups = {}
            serial_reason = "manifest_unreadable"
            errors.append(make_error(
                "configuration", "MANIFEST_UNREADABLE", "warning",
                f"the parallel manifest {manifest_path.name} could not be "
                f"read ({exc}); the run fell back to serial execution",
            ))
        if serial_reason is None:
            units = build_schedule(
                groups, nodeids_by_file, whole_file=collect.whole_file,
            )
            if collect.blocked:
                serial_reason = "static_gate"
            elif sig_state.interrupted is not None:
                serial_reason = "interrupted_during_collection"
            elif len(units) <= 1:
                serial_reason = "single_unit"
        if serial_reason is None:
            if missing:
                violations.append(unclassified_violation(walk.root, missing))
            workers_actual = min(worker_plan.resolved, len(units))
            dist_mode = (
                DIST_MODE_PARALLEL if workers_actual > 1 else DIST_MODE_SERIAL
            )
            guard_prior = os.environ.get(STANDIN_ACTIVE_ENV)
            prior_handlers: list[tuple[int, object]] = []
            on_main = threading.current_thread() is threading.main_thread()
            try:
                if on_main:
                    for signame in ("SIGINT", "SIGTERM", "SIGBREAK"):
                        signum = getattr(signal, signame, None)
                        if signum is not None:
                            prior_handlers.append((
                                signum,
                                signal.signal(signum, sig_state.on_signal),
                            ))
                # P7: every child inherits the recursion guard, so the
                # nested pytest session inside it stays serial and the
                # plan-239 root conftest never re-parallelises it.
                os.environ[STANDIN_ACTIVE_ENV] = "1"
                scheduled = True
                unit_outcomes, pool_interrupted = run_units(
                    units, workers=workers_actual, root=walk.root,
                    pytest_args=base_args, timeout=cfg.per_test_timeout,
                    origin=t0,
                )
            finally:
                for signum, handler in prior_handlers:
                    signal.signal(signum, handler)
                if guard_prior is None:
                    os.environ.pop(STANDIN_ACTIVE_ENV, None)
                else:
                    os.environ[STANDIN_ACTIVE_ENV] = guard_prior
            session = collect
            scheduling = _parallel_scheduling(
                manifest_path=manifest_path, root=walk.root,
                plan=worker_plan, outcomes=unit_outcomes, missing=missing,
            )
        else:
            scheduling = serial_scheduling(serial_reason)
 
    if not scheduled:
        if session is not None:
            # A collect pass ran and then declined to parallelise; the
            # serial path re-runs so its report is the plan-184 one,
            # unchanged in every field but ``scheduling``.
            _dbg("OC_TESTING", func,
                 f"falling back to serial ({serial_reason})", 2)
        session = run_session(
            pytest_args=pytest_args, root=walk.root,
            timeout=cfg.per_test_timeout, dry_run=cfg.dry_run,
            sig_state=sig_state, violation_file=str(walk.root),
        )
        unit_outcomes = []
        workers_actual = 1
        dist_mode = DIST_MODE_SERIAL
        scheduling = serial_scheduling(serial_reason)
 
    assert session is not None
    violations = session.static_violations + session.violations + violations
    for outcome in unit_outcomes:
        violations.extend(outcome.violations)
        if outcome.crashed:
            violations.append(crash_violation(outcome))
            errors.append(crash_error(outcome))
 
    # -- classify ----------------------------------------------------------
    captured = session.output
    execution_order = list(session.execution_order)
    if unit_outcomes:
        captured = "".join(
            [session.output]
            + [f"\n--- {o.unit.worker_id} ({o.unit.group}) ---\n{o.output}"
               for o in unit_outcomes]
        )
        execution_order = [
            nodeid for outcome in unit_outcomes
            for nodeid in outcome.unit.nodeids
            if nodeid in outcome.records
        ]
 
    def _finish(status: str, exit_code: int, tests: list[dict],
                collected: int) -> RunResult:
        report = _build_report(
            run_id=run_id, status=status, exit_code=exit_code,
            workspace_root=str(walk.root), workspace_mode="single_package",
            packages=[walk.package_id], profile=profile, marker=cfg.marker,
            seed=seed, nodeids=list(cfg.nodeids),
            workers_requested=cfg.workers, workers_actual=workers_actual,
            dist_mode=dist_mode, collected=collected, tests=tests,
            violations=violations, errors=errors,
            duration_seconds=time.monotonic() - t0,
            scheduling=scheduling,
        )
        return RunResult(
            report=report, exit_code=exit_code,
            execution_order=execution_order,
            captured_output=captured,
        )
 
    if unit_outcomes:
        tests = _tests_from_units(unit_outcomes)
        collect_errors = [
            message for outcome in unit_outcomes
            for message in outcome.collect_errors
        ]
        rcs = [outcome.rc for outcome in unit_outcomes
               if not outcome.crashed]
    else:
        tests = [
            {
                "nodeid": nodeid,
                "outcome": _outcome_for(nodeid, rec, set(session.timed_out)),
                "worker_id": "gw0",
                "duration_seconds": round(rec["duration"], 2),
                "failure_summary": (
                    rec["longrepr"]
                    if _outcome_for(nodeid, rec, set(session.timed_out))
                    in ("failed", "timeout")
                    else None
                ),
            }
            for nodeid, rec in session.records.items()
        ]
        collect_errors = list(session.collect_errors)
        rcs = [session.rc]
 
    collected_pre = len(session.collected_pre)
    selected_post = len(session.selected)
    interrupted = sig_state.interrupted is not None or pool_interrupted
 
    if interrupted:
        errors.append(make_error(
            "signal", "ORCHESTRATOR_INTERRUPTED", "fatal",
            f"received {sig_state.interrupted}", "orchestrator",
        ))
        return _finish("orchestrator_interrupted", 3, tests, collected_pre)
 
    if collect_errors:
        errors.append(make_error(
            "configuration", "COLLECTION_ERROR", "fatal",
            "pytest collection raised for the selected package: "
            + collect_errors[0],
        ))
        return _finish("configuration_error", 4, tests, collected_pre)
 
    unexpected = [rc for rc in rcs if rc in (2, 3)]
    if unexpected:
        errors.append(make_error(
            "orchestrator", "9008", "fatal",
            f"in-process pytest returned unexpected exit code "
            f"{unexpected[0]}",
        ))
        return _finish("orchestrator_error", 3, tests, collected_pre)
    if any(rc == 4 for rc in rcs):
        errors.append(make_error(
            "configuration", "9008", "fatal",
            "in-process pytest reported a usage error (exit 4); see the "
            "captured output",
        ))
        return _finish("configuration_error", 4, tests, collected_pre)
 
    if collected_pre == 0:
        errors.append(make_error(
            "configuration", "NO_TESTS_COLLECTED", "fatal",
            "collection completed but the union of the effective roots "
            "yielded zero collectable items",
        ))
        return _finish("configuration_error", 4, tests, 0)
 
    # per-root accounting (roots mode only): EMPTY_TEST_ROOT warnings
    if not cfg.nodeids:
        item_paths = [p.resolve() for _n, p in session.collected_pre]
        for tp, rp in existing_pairs:
            resolved_root = rp.resolve()
            count = sum(
                1 for p in item_paths
                if p == resolved_root or resolved_root in p.parents
            )
            if count == 0:
                errors.append(make_error(
                    "configuration", "EMPTY_TEST_ROOT", "warning",
                    f"declared test root {tp!r} collected zero items "
                    "while the run's union is non-empty",
                ))
 
    if selected_post == 0 and not session.blocked:
        if profile == "smoke":
            _dbg("OC_TESTING", func,
                 "smoke fallback: empty effective selection; package "
                 "import/collection checks passed", 2)
        else:
            errors.append(make_error(
                "configuration", "MARKER_PROFILE_CONFLICT", "warning",
                "the effective selection (profile + --marker + --nodeid) "
                "evaluated to an empty set; nothing ran",
            ))
 
    error_violations = any(v["severity"] == "error" for v in violations)
    failed_or_timeout = any(
        t["outcome"] in ("failed", "timeout") for t in tests
    )
    exit_code, status = compute_exit_and_status(
        fatal_config=False, interrupted=False, orchestrator_error=False,
        error_violations=error_violations,
        failed_or_timeout=failed_or_timeout,
    )
    return _finish(status, exit_code, tests, collected_pre)
 
 
def _tests_from_units(outcomes: Sequence[UnitOutcome]) -> list[dict]:
    """Reconcile every unit's records into one §11 ``tests`` array.
 
    A crashed unit contributes EVERY nodeid it was given, reported
    ``failed`` with the spec-10 §8.4 crash summary: the child died
    without reporting, so nothing it might have finished can be
    believed, and a nodeid that vanished silently would be worse than
    one reported failed.
    """
    tests: list[dict] = []
    for outcome in outcomes:
        worker_id = outcome.unit.worker_id
        if outcome.crashed:
            summary = f"worker crashed ({outcome.crash_detail})"
            tests.extend({
                "nodeid": nodeid,
                "outcome": "failed",
                "worker_id": worker_id,
                "duration_seconds": 0.0,
                "failure_summary": summary,
            } for nodeid in outcome.unit.nodeids)
            continue
        timed_out = set(outcome.timed_out)
        for nodeid, rec in outcome.records.items():
            result = _outcome_for(nodeid, rec, timed_out)
            tests.append({
                "nodeid": nodeid,
                "outcome": result,
                "worker_id": worker_id,
                "duration_seconds": round(rec["duration"], 2),
                "failure_summary": (
                    rec["longrepr"]
                    if result in ("failed", "timeout") else None
                ),
            })
    return tests
 
 
def _parallel_scheduling(
    *,
    manifest_path: Path,
    root: Path,
    plan: WorkerPlan,
    outcomes: Sequence[UnitOutcome],
    missing: Sequence[str],
) -> dict:
    """The ``scheduling`` block of a run that distributed its groups.
 
    Sorted by group name so the document is byte-stable regardless of
    which child finished first (P3's determinism clause).
    """
    try:
        manifest = Path(manifest_path).resolve().relative_to(
            Path(root).resolve(),
        ).as_posix()
    except ValueError:  # pragma: no cover - manifest outside the package
        manifest = Path(manifest_path).as_posix()
    groups = [
        {
            "group": outcome.unit.group,
            "worker_id": outcome.unit.worker_id,
            "files": len(outcome.unit.files),
            "paths": list(outcome.unit.files),
            "tests": len(outcome.unit.nodeids),
            "duration_seconds": round(outcome.duration_seconds, 2),
            "started_seconds": round(outcome.started_seconds, 2),
            "finished_seconds": round(outcome.finished_seconds, 2),
            "pid": outcome.pid,
            "status": "crashed" if outcome.crashed else "completed",
            "violations": len(outcome.violations),
        }
        for outcome in outcomes
    ]
    groups.sort(key=lambda entry: (entry["group"], entry["worker_id"]))
    return {
        "mode": "parallel",
        "reason": None,
        "manifest": manifest,
        "workers_cap": plan.cap,
        "workers_resolved": plan.resolved,
        "groups": groups,
        "unclassified": sorted(missing),
    }
 
 
_dbg("OC_TESTING", "module", "loaded", 4)
 
__all__ = [
    "RunResult",
    "STANDIN_ACTIVE_ENV",
    "SessionOutcome",
    "StandInTimeoutError",
    "TRACKED_ENV_PREFIXES",
    "run_session",
    "run_standin",
    "serial_scheduling",
]