From 9389d7ebe27cad4b1e74b4176a563baa41f37d3b Mon Sep 17 00:00:00 2001 From: Godopu Date: Sun, 6 Sep 2026 18:46:48 +0900 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_01UrDNJkHmm18uu7n7JwV5Qd --- tests/test_issue5_phase2a_loop.py | 154 ++++++++++++++++++++++++++++++ 1 file changed, 154 insertions(+) create mode 100644 tests/test_issue5_phase2a_loop.py diff --git a/tests/test_issue5_phase2a_loop.py b/tests/test_issue5_phase2a_loop.py new file mode 100644 index 0000000..5efdbed --- /dev/null +++ b/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"