settled-reach/tooling/test_jobs.py

#!/usr/bin/env python3
"""Units for the jobs service (T-1278), called DIRECTLY — no CLI round trip.

That is half the point of these tests. D-263 says a service must be
transport-agnostic so it can be called by a test, by another service, or by a
future second front end. If nothing ever exercises that, the layering is
unverified decoration — a claim in a decision record with no evidence behind it.
Every test here imports `service` and calls a function.

The other half is the two behaviours that are easy to get wrong and impossible
to notice when they are:

  1. A partial trailing line. The log is appended to by a live process, so a
     reader can arrive mid-write. Parsing greedily would either crash or, worse,
     silently discard the event and advance past it — losing exactly one line,
     the one being written when someone looked.
  2. Reconciliation. A job killed outright cannot record its ending, so its file
     says `running` forever. Trusting the file leaves a corpse looking busy and
     hangs anything waiting on it.

Run: python3 tooling/test_jobs.py
"""

import json
import sys
import tempfile
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from tooling.domains.jobs import service  # noqa: E402
from tooling.domains.jobs.schemas import Job, Status  # noqa: E402


def _job(**overrides) -> Job:
    base = {
        "job": "20260101T000000-test",
        "command": "reach check client-version",
        "argv": ["check", "client-version"],
        "pid": 1,
        "started_at": "2026-01-01T00:00:00",
        "status": Status.RUNNING,
    }
    return Job(**{**base, **overrides})


def test_partial_trailing_line(failures: list[str]) -> None:
    """A half-written final line is left for the next read, not dropped."""
    with tempfile.TemporaryDirectory() as tmp:
        log = Path(tmp) / "j.jsonl"
        complete = json.dumps({"message": "one"}) + "\n"
        partial = '{"message": "tw'
        log.write_text(complete + partial, encoding="utf-8")

        events, offset = service.read_events_from(log, 0)
        if len(events) != 1:
            failures.append(f"partial line: expected 1 complete event, got {len(events)}")
        if offset != len(complete):
            failures.append(
                f"partial line: offset {offset} should stop at {len(complete)}, "
                "before the incomplete line — otherwise that event is lost forever"
            )

        # Now the writer finishes the line. The event must appear.
        log.write_text(complete + json.dumps({"message": "two"}) + "\n", encoding="utf-8")
        events, _ = service.read_events_from(log, offset)
        if [e.get("message") for e in events] != ["two"]:
            failures.append(f"partial line: resumed read lost the completed event: {events}")


def test_offset_resume_is_stable(failures: list[str]) -> None:
    """Reading from the end returns nothing and does not move the offset."""
    with tempfile.TemporaryDirectory() as tmp:
        log = Path(tmp) / "j.jsonl"
        log.write_text(json.dumps({"message": "one"}) + "\n", encoding="utf-8")
        _, first = service.read_events_from(log, 0)
        events, second = service.read_events_from(log, first)
        if events or first != second:
            failures.append(
                f"offset resume: re-reading returned {len(events)} events and moved "
                f"{first}->{second}; a follow loop would replay forever"
            )


def test_dead_pid_is_reconciled(failures: list[str]) -> None:
    """running + a pid that is gone == died, never running."""
    # PID 1 exists; a very high pid almost certainly does not.
    alive = service._reconcile(_job(pid=1))
    if alive.status is not Status.RUNNING:
        failures.append("reconcile: a live pid was reported as not running")

    dead = service._reconcile(_job(pid=4_000_000))
    if dead.status is not Status.DIED:
        failures.append(
            f"reconcile: a dead pid stayed {dead.status.value} — a corpse that looks "
            "busy hangs every caller waiting on it"
        )


def test_died_never_relays_success(failures: list[str]) -> None:
    """A job that died has no exit code of its own, and must not borrow 0."""
    died = _job(status=Status.DIED)
    if died.effective_exit_code == 0:
        failures.append(
            "died job relayed exit 0 — that is the exit-0 trap: a killed job "
            "reported as success to whatever gated on it"
        )
    finished = _job(status=Status.FAILED, exit_code=2)
    if finished.effective_exit_code != 2:
        failures.append("failed job did not relay its own exit code")


def _fixture_jobs(root: Path, specs: list[tuple[str, str, int]]) -> None:
    """Write job metadata into a throwaway repo root. specs: (id, status, pid)."""
    directory = root / ".cache" / "reach" / "jobs"
    directory.mkdir(parents=True, exist_ok=True)
    for job_id, status, pid in specs:
        (directory / f"{job_id}.json").write_text(
            json.dumps(
                {
                    "job": job_id,
                    "command": "reach dev selftest",
                    "argv": ["dev", "selftest"],
                    "pid": pid,
                    "started_at": "2026-01-01T00:00:00",
                    "status": status,
                }
            ),
            encoding="utf-8",
        )
        (directory / f"{job_id}.jsonl").write_text("", encoding="utf-8")


def test_prune_spares_running_but_not_corpses(failures: list[str]) -> None:
    """The retention rule, and the half of it that is easy to get wrong.

    A job that is genuinely running survives pruning however old it is — a
    generator can outlive the window. But a job whose FILE says running while
    its process is gone must be prunable, or every crashed job becomes immortal
    and those are exactly what accumulates. The check has to consult the process
    table, not the record.
    """
    import os

    from tooling.core import process

    with tempfile.TemporaryDirectory() as tmp:
        root = Path(tmp)
        (root / "project.yaml").write_text("version: 0.0.0\n", encoding="utf-8")
        _fixture_jobs(
            root,
            [
                ("20260101T000001-aaaa", "done", 1),
                ("20260101T000002-bbbb", "running", 4_000_000),  # corpse: dead pid
                ("20260101T000003-cccc", "running", os.getpid()),  # genuinely alive
                ("20260101T000004-dddd", "done", 1),
            ],
        )

        previous = os.environ.get("SR_REPO_ROOT")
        os.environ["SR_REPO_ROOT"] = str(root)
        try:
            removed = set(process.prune(keep=1))
        finally:
            if previous is None:
                del os.environ["SR_REPO_ROOT"]
            else:
                os.environ["SR_REPO_ROOT"] = previous

        if "20260101T000003-cccc" in removed:
            failures.append(
                "prune: removed a job whose process is alive — a long generator "
                "would lose its own log while still writing to it"
            )
        if "20260101T000002-bbbb" not in removed:
            failures.append(
                "prune: kept a job whose file says running but whose process is "
                "gone — a crashed job must not be immortal, and those are exactly "
                "the ones that accumulate"
            )
        if "20260101T000001-aaaa" not in removed:
            failures.append("prune: kept a finished job beyond the keep window")


def main() -> int:
    failures: list[str] = []
    test_partial_trailing_line(failures)
    test_offset_resume_is_stable(failures)
    test_dead_pid_is_reconciled(failures)
    test_died_never_relays_success(failures)
    test_prune_spares_running_but_not_corpses(failures)

    if failures:
        print("test_jobs: FAIL", file=sys.stderr)
        for failure in failures:
            print(f"  - {failure}", file=sys.stderr)
        return 1
    print("test_jobs: OK — offsets resume cleanly, dead pids reconcile, died never relays 0")
    return 0


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