test(issue-5): phase 2a writer characterization before the CAS boundary

Fixes what the job-status writers do today, so routing them through a single
transition boundary has an oracle instead of a promise. Same ordering that paid
off in phase 1.

X-02 states the claim path's real contract rather than the one the plan first
assumed: pick_pending filters on pending under the lock, so it is already a
correct swap. It duplicates the chokepoint, it does not bypass its safety. The
test pins the properties that must survive routing -- only pending is claimed,
only one session wins, and a session never sees another's work.

Every test is parametrized over MAM_ENGINE_CAS so the legacy body and the engine
body must agree.

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

tests/test_issue5_phase2a_loop.py

@@ -0,0 +1,154 @@
+#!/usr/bin/env python3
+"""Issue #5 Phase 2a — lib_py/loop core and shadow watcher.
+
+Written in the order the plan fixes: characterize the current writers BEFORE
+routing them through a CAS boundary, so the refactor has an oracle rather than a
+promise. Phase 1 proved the value of that ordering; it also proved that a fake
+which cannot express a failure hides it, so every double here models the real
+contract (herdr's exit/stream split, the registry lock, the record on disk).
+
+Sections
+  1. X-01..X-04  current writer behaviour (must hold before and after P2A-01).
+  2. engine      the single CAS transition boundary and its lock discipline.
+  3. watcher     herdr agent wait wrapping, stream split, reconnect backoff.
+  4. policy      decisions checked against the phase-0 scenario contracts.
+  5. shadow      the runner that cannot write, proven by hashing the registry.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+SKILLS = REPO_ROOT / ".agents" / "skills"
+WATCHER_FIXTURES = REPO_ROOT / "tests" / "fixtures" / "issue_5" / "watcher"
+
+CAS_MODES = ("0", "1")  # MAM_ENGINE_CAS: legacy unconditional merge / CAS engine
+
+
+# ==============================================================================
+# Helpers
+# ==============================================================================
+
+def _reg(tmp_path) -> str:
+    d = tmp_path / ".mam" / "jobs"
+    d.mkdir(parents=True, exist_ok=True)
+    return str(d)
+
+
+def _make_job(registry_dir, **over):
+    from lib_py.delegation import registry
+    job_id = registry.register_job(over.pop("prompt", "p"), registry_dir=registry_dir, **over)
+    return job_id
+
+
+def _load(registry_dir, job_id):
+    from lib_py.delegation import mqtt_common
+    return mqtt_common.load_job(job_id, registry_dir)
+
+
+def _dir_fingerprint(path: Path) -> dict:
+    """File name -> sha256 for every file under path. Detects any byte change."""
+    out = {}
+    for p in sorted(Path(path).rglob("*")):
+        if p.is_file():
+            out[str(p.relative_to(path))] = hashlib.sha256(p.read_bytes()).hexdigest()
+    return out
+
+
+@pytest.fixture(autouse=True)
+def _cas_on(monkeypatch):
+    """Default every test to the CAS engine unless it says otherwise."""
+    monkeypatch.setenv("MAM_ENGINE_CAS", "1")
+
+
+# ==============================================================================
+# 1. X-01..X-04 — the writers as they behave today
+# ==============================================================================
+
+@pytest.mark.parametrize("mode", CAS_MODES)
+def test_x01_update_job_status_merges_arbitrary_fields(mam_sandbox, tmp_path, monkeypatch, mode):
+    """X-01: update_job_status merges whatever it is given, in both modes.
+
+    Its signature, return value and side effects are load-bearing: phase 1's
+    G-1..G-4 assert through it, so routing it through the engine must not change
+    what a caller observes."""
+    monkeypatch.setenv("MAM_ENGINE_CAS", mode)
+    monkeypatch.chdir(mam_sandbox)
+    from lib_py.delegation import mqtt_common
+
+    reg = _reg(tmp_path)
+    job_id = _make_job(reg)
+    out = mqtt_common.update_job_status(job_id, reg, status="running", iteration=4)
+    assert out["status"] == "running"
+    assert out["iteration"] == 4
+    assert out["updated_at"], "updated_at must always be refreshed"
+    assert _load(reg, job_id)["status"] == "running"
+
+
+@pytest.mark.parametrize("mode", CAS_MODES)
+def test_x02_pick_pending_claims_only_pending_and_only_once(mam_sandbox, tmp_path, monkeypatch, mode):
+    """X-02: the claim path's real contract.
+
+    pick_pending is not broken today: it filters on pending under the lock, so
+    pending->running is already a correct swap. It is a *duplicate* of the
+    chokepoint, not a hole in it. Routing it through the engine must preserve
+    exactly this behaviour."""
+    monkeypatch.setenv("MAM_ENGINE_CAS", mode)
+    monkeypatch.chdir(mam_sandbox)
+    from lib_py.delegation import registry
+
+    reg = _reg(tmp_path)
+    job_id = _make_job(reg, agent_session="herdr:s1")
+    assert registry.pick_pending("herdr:s1", reg) == job_id
+    assert _load(reg, job_id)["status"] == "running"
+    # already claimed -> no second winner
+    assert registry.pick_pending("herdr:s1", reg) is None
+    # a different session never sees another session's work
+    other = _make_job(reg, agent_session="herdr:s2")
+    assert registry.pick_pending("herdr:s1", reg) is None
+    assert registry.pick_pending("herdr:s2", reg) == other
+
+
+@pytest.mark.parametrize("mode", CAS_MODES)
+def test_x03_status_write_mirrors_into_the_audit_log(mam_sandbox, tmp_path, monkeypatch, mode):
+    """X-03: a status change is mirrored to the audit log, in both modes.
+
+    P2A-01 moves that mirror out of the lock; it must not lose it."""
+    monkeypatch.setenv("MAM_ENGINE_CAS", mode)
+    monkeypatch.chdir(mam_sandbox)
+    from lib_py.delegation import mqtt_common
+
+    reg = _reg(tmp_path)
+    job_id = _make_job(reg)
+    mqtt_common.update_job_status(job_id, reg, status="completed")
+
+    logged = mqtt_common.read_logged_status(job_id, mqtt_common.get_logs_dir())
+    assert logged and logged.get("status") == "completed"
+    events = list(mqtt_common.iter_logged_events(job_id, mqtt_common.get_logs_dir()))
+    changes = [e for e in events if e.get("event") == "status_changed"]
+    assert changes, "a status change must appear in the audit log"
+    assert changes[-1]["to"] == "completed"
+
+
+@pytest.mark.parametrize("mode", CAS_MODES)
+def test_x04_next_seq_and_register_are_unaffected(mam_sandbox, tmp_path, monkeypatch, mode):
+    """X-04: the non-status writers keep their behaviour under either mode."""
+    monkeypatch.setenv("MAM_ENGINE_CAS", mode)
+    monkeypatch.chdir(mam_sandbox)
+    from lib_py.delegation import mqtt_common
+
+    reg = _reg(tmp_path)
+    job_id = _make_job(reg)
+    assert _load(reg, job_id)["status"] == "pending"
+    assert mqtt_common.next_seq(job_id, reg) == 1
+    assert mqtt_common.next_seq(job_id, reg) == 2
+    assert _load(reg, job_id)["last_seq"] == 2
+    assert _load(reg, job_id)["status"] == "pending", "next_seq must not touch status"