tmpl/multi-agent-mux
1
0 Fork 0
Code Issues 9 Pull Requests 1 Actions Packages Projects Releases Wiki Activity
Files
b49af56c3cc2ca63bc9a95b6c799a2376c01013c
multi-agent-mux/tests/test_issue5_phase2a_loop.py
T
GodopuandClaude Opus 5 b49af56c3c feat(issue-5): herdr agent wait watcher with reconnect backoff and a shadow store 
watcher wraps one `herdr agent wait` barrier and observes only; it never
concludes. The live contract is encoded rather than described: the timeout
parameter is named timeout_ms because herdr counts milliseconds, --until is
always passed explicitly since omitting it silently drops `working`, and a
non-zero exit is read from stderr because stdout is empty by contract.

A timeout counts as healthy. The socket answered and the state simply did not
change; treating that as a failure would make an idle agent look unreachable.
Only a matched snapshot becomes evidence, so a timeout can never be mistaken for
an observation of agent state.

Backoff doubles from 1s to a 16s ceiling and drops the cached snapshot on
failure. Reasoning from state captured before a gap is exactly what would make a
disconnect look like a completion, which is why the contract answer there is
hold_reconnect.

store keeps dispatch and attribution evidence outside the registry. The job
record carries none of attempt_id, agent_session_uuid or
pane_process_generation, phase 1 froze its schema, and this phase may not write
the registry at all. Keeping the evidence here lets shadow mode carry full
attribution while the registry stays untouched; promoting these fields is a 2b
decision. The execution anchor starts false because acknowledging a dispatch is
not evidence that the attempt ran.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UrDNJkHmm18uu7n7JwV5Qd
2026-09-06 18:54:27 +09:00

540 lines
22 KiB
Python
Raw Blame History

#!/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"
# ==============================================================================
# 2. engine — the CAS table (state-transitions.md §3) and lock discipline
# ==============================================================================
def test_engine_pending_to_running_on_claim(mam_sandbox, tmp_path, monkeypatch):
monkeypatch.chdir(mam_sandbox)
from lib_py.delegation import engine
reg = _reg(tmp_path)
job_id = _make_job(reg)
res = engine.transition(job_id, "claim", registry_dir=reg, expected_status="pending")
assert res.committed and res.from_status == "pending" and res.to_status == "running"
@pytest.mark.parametrize("request_", ["progress", "permission_required"])
def test_engine_non_terminal_requests_keep_running(mam_sandbox, tmp_path, monkeypatch, request_):
monkeypatch.chdir(mam_sandbox)
from lib_py.delegation import engine
reg = _reg(tmp_path)
job_id = _make_job(reg)
engine.transition(job_id, "claim", registry_dir=reg)
res = engine.transition(job_id, request_, registry_dir=reg,
fields={"blocked_reason": "tool"})
assert res.committed
assert _load(reg, job_id)["status"] == "running", "must stay non-terminal"
assert _load(reg, job_id)["blocked_reason"] == "tool"
@pytest.mark.parametrize("terminal", ["completed", "error", "cancelled"])
def test_engine_explicit_terminal_commits(mam_sandbox, tmp_path, monkeypatch, terminal):
monkeypatch.chdir(mam_sandbox)
from lib_py.delegation import engine
reg = _reg(tmp_path)
job_id = _make_job(reg)
engine.transition(job_id, "claim", registry_dir=reg)
res = engine.transition(job_id, terminal, registry_dir=reg)
assert res.committed and res.to_status == terminal
def test_engine_repeated_terminal_is_an_idempotent_noop(mam_sandbox, tmp_path, monkeypatch):
monkeypatch.chdir(mam_sandbox)
from lib_py.delegation import engine
reg = _reg(tmp_path)
job_id = _make_job(reg)
engine.transition(job_id, "claim", registry_dir=reg)
engine.transition(job_id, "completed", registry_dir=reg)
before = _load(reg, job_id)
res = engine.transition(job_id, "completed", registry_dir=reg)
assert res.committed is False
assert "idempotent" in res.reason
assert _load(reg, job_id)["transition_seq"] == before["transition_seq"], "no second write"
@pytest.mark.parametrize("late", ["error", "cancelled", "progress"])
def test_engine_refuses_to_reverse_a_terminal_and_audits_the_conflict(
mam_sandbox, tmp_path, monkeypatch, late
):
"""A late or conflicting signal must never walk a terminal record backwards."""
monkeypatch.chdir(mam_sandbox)
from lib_py.delegation import mqtt_common
from lib_py.delegation import engine
reg = _reg(tmp_path)
job_id = _make_job(reg)
engine.transition(job_id, "claim", registry_dir=reg)
engine.transition(job_id, "completed", registry_dir=reg)
res = engine.transition(job_id, late, registry_dir=reg, source="late-arrival")
assert res.committed is False
assert _load(reg, job_id)["status"] == "completed", "terminal must be preserved"
events = list(mqtt_common.iter_logged_events(job_id, mqtt_common.get_logs_dir()))
refusals = [e for e in events if e.get("event") == "conflict_refused"]
assert refusals, "a refused conflict must be auditable"
assert refusals[-1]["current"] == "completed"
def test_engine_rejects_a_stale_attempt(mam_sandbox, tmp_path, monkeypatch):
monkeypatch.chdir(mam_sandbox)
from lib_py.delegation import engine
reg = _reg(tmp_path)
job_id = _make_job(reg)
engine.transition(job_id, "claim", registry_dir=reg, fields={"attempt_id": 2})
res = engine.transition(job_id, "completed", registry_dir=reg, expected_attempt=1)
assert res.committed is False
assert "stale attempt" in res.reason
assert _load(reg, job_id)["status"] == "running"
def test_engine_deadline_timeout_becomes_error(mam_sandbox, tmp_path, monkeypatch):
monkeypatch.chdir(mam_sandbox)
from lib_py.delegation import engine
reg = _reg(tmp_path)
job_id = _make_job(reg)
engine.transition(job_id, "claim", registry_dir=reg)
res = engine.transition(job_id, "deadline_timeout", registry_dir=reg)
assert res.committed and res.to_status == "error"
def test_engine_precondition_failure_does_not_write(mam_sandbox, tmp_path, monkeypatch):
monkeypatch.chdir(mam_sandbox)
from lib_py.delegation import engine
reg = _reg(tmp_path)
job_id = _make_job(reg)
engine.transition(job_id, "claim", registry_dir=reg) # now running
res = engine.transition(job_id, "claim", registry_dir=reg, expected_status="pending")
assert res.committed is False and "precondition failed" in res.reason
def test_engine_transition_seq_is_monotonic(mam_sandbox, tmp_path, monkeypatch):
"""The audit mirror runs outside the lock, so ordering rides on this counter."""
monkeypatch.chdir(mam_sandbox)
from lib_py.delegation import engine
reg = _reg(tmp_path)
job_id = _make_job(reg)
seqs = []
for req in ("claim", "progress", "progress", "completed"):
engine.transition(job_id, req, registry_dir=reg)
seqs.append(_load(reg, job_id)["transition_seq"])
assert seqs == sorted(seqs) and len(set(seqs)) == len(seqs), seqs
def test_g5_no_blocking_io_while_the_registry_lock_is_held(mam_sandbox, tmp_path, monkeypatch):
"""G5: the lock may cover a decision and one record write, nothing else.
Watches the calls that can block on another process or the network and
asserts none of them happen between acquiring and releasing the lock."""
monkeypatch.chdir(mam_sandbox)
import socket
from lib_py.delegation import mqtt_common
from lib_py.delegation import engine
reg = _reg(tmp_path)
job_id = _make_job(reg)
held = {"depth": 0}
violations = []
real_lock = mqtt_common.registry_lock
import contextlib
@contextlib.contextmanager
def watched_lock(registry_dir):
held["depth"] += 1
try:
with real_lock(registry_dir):
yield
finally:
held["depth"] -= 1
def guard(name):
def spy(*args, **kwargs):
if held["depth"] > 0:
violations.append(name)
raise AssertionError("{} must not run under the registry lock".format(name))
return spy
monkeypatch.setattr(mqtt_common, "registry_lock", watched_lock)
monkeypatch.setattr(engine.mqtt_common, "registry_lock", watched_lock, raising=False)
monkeypatch.setattr(subprocess, "run", guard("subprocess.run"))
monkeypatch.setattr(subprocess, "Popen", guard("subprocess.Popen"))
monkeypatch.setattr(socket, "socket", guard("socket.socket"))
real_sleep = time.sleep
def sleep_spy(seconds):
if held["depth"] > 0:
violations.append("time.sleep")
return real_sleep(0)
monkeypatch.setattr(time, "sleep", sleep_spy)
for req in ("claim", "progress", "completed", "completed", "error"):
engine.transition(job_id, req, registry_dir=reg)
assert not violations, "blocking calls under the lock: {}".format(violations)
def test_g6_every_record_writer_is_the_engine_or_a_declared_exception():
"""G6: no new bypass of the transition boundary can be added unnoticed."""
import ast
allowed = {
# creation, and the sequence counter, which never touches status
("registry.py", "register_job"),
("mqtt_common.py", "next_seq"),
("mqtt_common.py", "_update_job_status_legacy"),
("registry.py", "_pick_pending_legacy"),
("engine.py", "transition"),
}
found = set()
for path in list((SKILLS / "lib_py" / "delegation").glob("*.py")) + \
list((SKILLS / "lib_py" / "loop").glob("*.py")):
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
for node in ast.walk(tree):
if not isinstance(node, ast.FunctionDef):
continue
for inner in ast.walk(node):
if isinstance(inner, ast.Call):
fn = inner.func
name = getattr(fn, "attr", getattr(fn, "id", None))
if name == "_atomic_write_record":
found.add((path.name, node.name))
assert found <= allowed, (
"job records are written outside the transition boundary: {}".format(found - allowed)
)
# ==============================================================================
# 3. watcher — herdr agent wait wrapping. No test ever runs a real herdr.
# ==============================================================================
class _Proc(object):
def __init__(self, returncode, stdout="", stderr=""):
self.returncode = returncode
self.stdout = stdout
self.stderr = stderr
def _runner(proc, seen=None):
def run(argv, **_kw):
if seen is not None:
seen.append(argv)
if isinstance(proc, Exception):
raise proc
return proc
return run
AGENT_SNAPSHOT = {
"id": "cli:agent:get",
"result": {"agent": {"agent_status": "idle", "interactive_ready": True,
"pane_id": "w2X:p1", "state_change_seq": 18},
"type": "agent_info"},
}
def test_watcher_passes_timeout_in_milliseconds_and_always_names_until():
"""The unit is the contract: --timeout is MS, and omitting --until drops `working`."""
from lib_py.loop import watcher
seen = []
watcher.wait_once("sess", until=("idle", "working"), timeout_ms=15000,
run=_runner(_Proc(0, json.dumps(AGENT_SNAPSHOT)), seen))
argv = seen[0]
assert argv[:4] == ["herdr", "agent", "wait", "sess"]
assert argv[argv.index("--timeout") + 1] == "15000", argv
assert argv.count("--until") == 2 and "working" in argv, argv
def test_watcher_rejects_a_nonpositive_timeout():
from lib_py.loop import watcher
with pytest.raises(ValueError):
watcher.build_argv("sess", timeout_ms=0)
def test_watcher_matched_snapshot_comes_from_stdout():
from lib_py.loop import watcher
out = watcher.wait_once("sess", run=_runner(_Proc(0, json.dumps(AGENT_SNAPSHOT))))
assert out.kind == watcher.MATCHED
assert out.agent["agent_status"] == "idle"
assert out.healthy
@pytest.mark.parametrize("code,expected", [
("timeout", "timeout"),
("agent_not_found", "agent_not_found"),
])
def test_watcher_reads_errors_from_stderr_not_stdout(code, expected):
"""exit 1 puts the payload on stderr and leaves stdout empty."""
from lib_py.loop import watcher
payload = json.dumps({"error": {"code": code, "message": "x"}, "id": "cli:agent:wait"})
out = watcher.wait_once("sess", run=_runner(_Proc(1, "", payload)))
assert out.kind == expected
assert out.error_code == code
assert out.healthy, "a socket that answered is healthy even when it says no"
def test_watcher_unparseable_or_unexpected_output_is_a_transport_error():
from lib_py.loop import watcher
assert watcher.wait_once("s", run=_runner(_Proc(1, "", "not json"))).kind == watcher.TRANSPORT_ERROR
assert watcher.wait_once("s", run=_runner(_Proc(0, "{}"))).kind == watcher.TRANSPORT_ERROR
assert watcher.wait_once("s", run=_runner(_Proc(7, "", "{}"))).kind == watcher.TRANSPORT_ERROR
def test_watcher_reports_a_failed_launch_rather_than_raising():
from lib_py.loop import watcher
out = watcher.wait_once("s", run=_runner(OSError("herdr missing")))
assert out.kind == watcher.TRANSPORT_ERROR and not out.healthy
def test_watcher_only_matched_outcomes_become_evidence():
"""A timeout proves nothing about agent state, so it must not become an observation."""
from lib_py.loop import watcher
matched = watcher.wait_once("s", run=_runner(_Proc(0, json.dumps(AGENT_SNAPSHOT))))
assert watcher.observation_from(matched).payload["agent_status"] == "idle"
timed_out = watcher.wait_once(
"s", run=_runner(_Proc(1, "", json.dumps({"error": {"code": "timeout"}}))))
assert watcher.observation_from(timed_out) is None
def test_watcher_backoff_doubles_to_a_ceiling_and_resets():
from lib_py.loop import watcher
b = watcher.ReconnectBackoff()
assert b.sequence(7) == [1.0, 2.0, 4.0, 8.0, 16.0, 16.0, 16.0]
b.record_failure(); b.record_failure()
assert b.consecutive_failures == 2
b.record_success({"agent_status": "idle"})
assert b.consecutive_failures == 0 and b.current == 0.0
def test_watcher_drops_its_cached_snapshot_when_the_socket_fails():
"""After a gap the watcher must re-observe; stale state is what makes a
disconnect look like a completion."""
from lib_py.loop import watcher
b = watcher.ReconnectBackoff()
b.record_success({"agent_status": "idle"})
assert b.cached_snapshot is not None
b.record_failure()
assert b.cached_snapshot is None
# ==============================================================================
# 4. store — attribution lives outside the registry
# ==============================================================================
def test_store_keeps_attribution_out_of_the_job_record(mam_sandbox, tmp_path, monkeypatch):
monkeypatch.chdir(mam_sandbox)
from lib_py.loop import store
reg = _reg(tmp_path)
job_id = _make_job(reg)
before = _dir_fingerprint(Path(reg))
store.record_dispatch(job_id, 1, "uuid-1", "w2X:p1:1234:2026-09-06T00:00:00Z",
workspace=str(mam_sandbox))
ctx = store.load_dispatch(job_id, workspace=str(mam_sandbox))
assert ctx["attempt_id"] == 1
assert ctx["execution_anchor_observed"] is False, "dispatch ack is not execution"
assert _dir_fingerprint(Path(reg)) == before, "the registry must be untouched"
def test_store_execution_anchor_is_recorded_separately(mam_sandbox, tmp_path, monkeypatch):
monkeypatch.chdir(mam_sandbox)
from lib_py.loop import store
reg = _reg(tmp_path)
job_id = _make_job(reg)
store.record_dispatch(job_id, 1, "u", "w2X:p1:1:t", workspace=str(mam_sandbox))
ctx = store.mark_execution_anchor(job_id, workspace=str(mam_sandbox))
assert ctx["execution_anchor_observed"] is True
# ==============================================================================
# 5. attribution — the four-fold predicate
# ==============================================================================
def _ctx(**over):
base = {"job_id": "j", "attempt_id": 1, "agent_session_uuid": "u",
"pane_process_generation": "w2X:p1:1234:t",
"dispatch_ack": True, "execution_anchor_observed": True}
base.update(over)
return base
def test_attribution_requires_all_four_plus_the_anchor():
from lib_py.loop import attribution
assert attribution.check(_ctx()).ok
assert not attribution.check(_ctx(execution_anchor_observed=False)).ok
assert not attribution.check(_ctx(dispatch_ack=False)).ok
for field in attribution.REQUIRED_FIELDS:
broken = _ctx()
del broken[field]
assert not attribution.check(broken).ok, field
empty = _ctx(**{field: None})
assert not attribution.check(empty).ok, field
def test_attribution_rejects_a_mismatch_on_any_identifier():
from lib_py.loop import attribution
assert not attribution.check(_ctx(), {"attempt_id": 2}).ok
assert not attribution.check(_ctx(), {"agent_session_uuid": "other"}).ok
assert attribution.check(_ctx(), {"attempt_id": 1, "job_id": "j"}).ok
def test_attribution_pane_generation_prefix_is_not_a_substring_match():
from lib_py.loop import attribution
assert attribution.pane_matches_generation("w2X:p1", "w2X:p1:1234:t")
assert not attribution.pane_matches_generation("w2X:p1", "w2X:p10:1234:t")
assert not attribution.pane_matches_generation("", "w2X:p1:1:t")
Reference in new issue View Git Blame Copy Permalink