settled-reach/tooling/test_job_exit_codes.py

#!/usr/bin/env python3
"""A detached job's exit code must survive (T-1279).

This is the non-negotiable in D-263, pointed at its worst hiding place. A
foreground command that swallows a failure at least does it in front of someone.
A background runner that reports "started" and then loses the failure does it
where nobody is looking, and whatever gated on the run carries on as if it had
passed.

Four properties, and the ordering of the cases matters:

  1. `jobs wait` exits with the JOB's code, so a Makefile or hook can gate on a
     detached run exactly as on a foreground one.
  2. It works whether the job fails almost immediately or long after the parent
     has exited. Those are different halves of the recording path: the fast case
     can finish while the parent is still alive, the slow case certainly cannot,
     and only the second proves the child records its own ending.
  3. `--detach` exits 0 for STARTING, which is a different claim from the job
     succeeding — so it has to say so in words rather than leave a reader to
     infer it from a 0.
  4. A failed job nobody waited on is visible as failed, not merely absent.

The failing cases come first deliberately. A job runner that has only ever run
successful jobs has never been tested.

Run: python3 tooling/test_job_exit_codes.py
"""

import json
import re
import shutil
import subprocess
import sys
import time
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent

JOB_ID = re.compile(r"^\d{8}T\d{6}-[0-9a-f]{4}$")


def _reach(*args: str) -> subprocess.CompletedProcess[str]:
    """Invoke by BARE NAME — never a path or an interpreter (T-1261)."""
    return subprocess.run(
        ["reach", *args], capture_output=True, text=True, cwd=REPO_ROOT
    )


def _detach(*args: str) -> tuple[str, subprocess.CompletedProcess[str], float]:
    started = time.monotonic()
    result = _reach("--detach", *args)
    elapsed = time.monotonic() - started
    job_id = result.stdout.strip().splitlines()[0] if result.stdout.strip() else ""
    return job_id, result, elapsed


def test_failure_survives(failures: list[str], seconds: str, code: str, label: str) -> None:
    """A job that fails is waited on and relays its own code."""
    job_id, spawn, elapsed = _detach(
        "dev", "selftest", "--seconds", seconds, "--fail", "--exit-code", code
    )

    if not JOB_ID.match(job_id):
        failures.append(f"[{label}] --detach did not print a job id, got {job_id!r}")
        return
    if spawn.returncode != 0:
        failures.append(
            f"[{label}] --detach exited {spawn.returncode}; starting a job succeeded, "
            "so the launch itself must report 0"
        )

    # Property 3: the 0 above must not be mistakable for "the job succeeded".
    if "not succeeded" not in spawn.stderr:
        failures.append(
            f"[{label}] --detach's output does not distinguish STARTED from SUCCEEDED "
            "in words; a bare 0 invites exactly the wrong reading"
        )

    # Property 2, slow case: the parent must be long gone before the child ends.
    if label == "slow" and elapsed > float(seconds) / 2:
        failures.append(
            f"[{label}] --detach took {elapsed:.2f}s for a {seconds}s job — it waited, "
            "which means this case is not testing what it claims to"
        )

    waited = _reach("jobs", "wait", job_id)
    if waited.returncode != int(code):
        failures.append(
            f"[{label}] jobs wait exited {waited.returncode}, expected {code} — "
            "the failure was lost between the job and its caller, which is the "
            "exit-0 trap in the place nobody watches"
        )

    # Property 4: visible as failed without anyone having waited.
    listed = _reach("jobs", "list", "--limit", "20")
    row = [line for line in listed.stdout.splitlines() if job_id in line]
    if not row:
        failures.append(f"[{label}] the job is absent from jobs list")
    elif "failed" not in row[0]:
        failures.append(
            f"[{label}] jobs list does not show the job as failed: {row[0]!r}"
        )


def test_success_relays_zero(failures: list[str]) -> None:
    job_id, spawn, _ = _detach("dev", "selftest", "--seconds", "0")
    if spawn.returncode != 0:
        failures.append(f"[success] --detach exited {spawn.returncode}")
        return
    waited = _reach("jobs", "wait", job_id)
    if waited.returncode != 0:
        failures.append(
            f"[success] jobs wait exited {waited.returncode} for a job that succeeded"
        )


def test_events_are_correlated(failures: list[str]) -> None:
    """Every event a detached job emits carries its job id."""
    job_id, _, _ = _detach("dev", "selftest", "--seconds", "0")
    _reach("jobs", "wait", job_id)

    log = REPO_ROOT / ".cache" / "reach" / "jobs" / f"{job_id}.jsonl"
    if not log.is_file():
        failures.append("[correlation] no log file was written for a detached job")
        return
    for line in log.read_text(encoding="utf-8").splitlines():
        if not line.strip():
            continue
        event = json.loads(line)
        if event.get("job") != job_id:
            failures.append(
                f"[correlation] an event carries job={event.get('job')!r}, expected "
                f"{job_id!r} — an untagged line cannot be attributed when logs interleave"
            )
            return


def main() -> int:
    if shutil.which("reach") is None:
        print(
            "test_job_exit_codes: `reach` is not on PATH.\n  Fix: make install-reach",
            file=sys.stderr,
        )
        return 1

    failures: list[str] = []
    # Failing paths first — see the module docstring.
    test_failure_survives(failures, seconds="0", code="3", label="fast")
    test_failure_survives(failures, seconds="2", code="7", label="slow")
    test_success_relays_zero(failures)
    test_events_are_correlated(failures)

    if failures:
        print("test_job_exit_codes: FAIL", file=sys.stderr)
        for failure in failures:
            print(f"  - {failure}", file=sys.stderr)
        return 1
    print(
        "test_job_exit_codes: OK — codes relay through detach for fast and slow "
        "failures, start is distinguished from success, events stay correlated"
    )
    return 0


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