GodopuandClaude Fable 5.1
087eacaf21
Review round 3 (planner-reviewer-codex-01, reviewer-opencode-01): R1 blocking, R2 and the "ONE builder" wording non-blocking. All three reproduced; none rebutted. R1. _wait_started still returned 0 as soon as the record left `pending`, ahead of the shared predicate, so `error`, `cancelled` and another dispatch's `running` all acknowledged an injection that create's delegate_wait_started rejects - the two contracts the ack-finding brief said must not diverge. Both reviewers reproduced it with the real producer and no record manipulation; my S7n had hidden it by resetting the record to `pending`. The bypass and the console grep are gone. The wrapper's evidence is now delegate_started_recorded alone, the same call create makes. For a remote agent - no shared disk, so no line in events.log - the predicate reads the subscriber's audit record of the `started` it received. That record is attributable by construction: the subscriber verifies the HMAC and drops any event whose data.dispatch_token differs from the job's expected token before it writes anything. The token itself cannot go into the record (the audit redactor masks token-named keys), so the subscriber now stamps a boolean dispatch_verified, and with a token demanded only a verified record counts; an untokened legacy record fails closed. The registry status is never evidence; a terminal job is preserved by the terminal logic, not turned into an acknowledgement. Tests: W1/W2/W3 are the reviewers' three negatives and the positive, real producer, record left exactly as it produced it; W4 runs create's wait and the wrapper's wait side by side on all four and requires the same answer; W5 pins the remote route (a verified record counts, an unverified or non-started one does not); W6 drives the real _Watcher and shows it stamps the flag on a matching started and records nothing for a mismatch; S7n no longer resets the record. test_issue14's "registry ack over unconfirmed inject" produced its ack with a bare status flip - the very bypass - and now produces it the way an agent does, publish_event with the dispatch token; its scenario and every other assertion are unchanged. test_issue18's remote e2e accepted the status messages and now expects the shared one. R2. _code_only blanked a docstring's whole line range, so code sharing that line - `"""doc"""; call()` - vanished from the scan. It now removes only the docstring node from the AST and unparses the rest (comments drop out with it, code strings stay); test_g9b pins the four same-line shapes. The builder comment claimed one builder for create and the wrapper; the wrapper keeps _pub_cmd and S7l pins the two equal argument-for-argument. The comment now says that. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
1143 lines
49 KiB
Python
1143 lines
49 KiB
Python
#!/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"),
|
|
# Issue #12 migration: drops broker.password only, never status
|
|
("registry.py", "_cmd_scrub_secrets"),
|
|
}
|
|
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")
|
|
|
|
|
|
# ==============================================================================
|
|
# 6. policy — checked against the phase-0 scenario contracts, not fresh rules
|
|
# ==============================================================================
|
|
|
|
def _scenarios():
|
|
out = {}
|
|
for path in sorted(WATCHER_FIXTURES.glob("*.json")):
|
|
if path.name == "manifest.json":
|
|
continue
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
out[data["scenario_id"]] = data
|
|
return out
|
|
|
|
|
|
@pytest.mark.parametrize("scenario_id", [
|
|
# the phase-0 seven ...
|
|
"initial_idle", "tool_running", "report_pause_15s", "finalized_candidate",
|
|
"blocked_permission", "transport_disconnected", "restarted_generation",
|
|
# ... and the five P2A-04 requires, held to the same contract
|
|
"race_double_dispatch", "attempt_reuse", "partial_report",
|
|
"unknown_status", "restart_midflight",
|
|
# ... and the three phase 2b adds, held to the same contract
|
|
"side_effect_unproven", "no_verdict_fail_closed", "notification_crash_window",
|
|
])
|
|
def test_g7_policy_matches_the_phase0_expected_decision(scenario_id):
|
|
"""G7: the seven scenarios that survived three-reviewer scrutiny are the
|
|
contract. The policy must reproduce their decisions exactly."""
|
|
from lib_py.loop import policy
|
|
scenario = _scenarios()[scenario_id]
|
|
expected = scenario["expected_policy"]["decision"]
|
|
got = policy.decide(scenario)
|
|
assert got.decision == expected, (
|
|
"{}: expected {!r}, got {!r} ({})".format(scenario_id, expected, got.decision, got.reason)
|
|
)
|
|
|
|
|
|
def test_policy_is_pure_no_io_no_clock(monkeypatch):
|
|
"""Purity is what makes the fixtures usable as vectors; guard it."""
|
|
import socket
|
|
from lib_py.loop import policy
|
|
|
|
scenarios = list(_scenarios().values()) # load before the guards go up
|
|
|
|
def boom(*_a, **_k):
|
|
raise AssertionError("policy.decide must not perform I/O")
|
|
|
|
monkeypatch.setattr(subprocess, "run", boom)
|
|
monkeypatch.setattr(socket, "socket", boom)
|
|
monkeypatch.setattr(Path, "read_text", boom)
|
|
monkeypatch.setattr(time, "time", boom)
|
|
for scenario in scenarios:
|
|
policy.decide(scenario)
|
|
|
|
|
|
def test_policy_debounce_alone_never_authorises_completion():
|
|
"""A stable mtime is a candidate filter, not an authority to complete."""
|
|
from lib_py.loop import policy
|
|
scenario = json.loads((WATCHER_FIXTURES / "finalized_candidate.json").read_text(encoding="utf-8"))
|
|
scenario["filesystem"]["manifest_committed"] = False
|
|
assert policy.decide(scenario).decision == "hold"
|
|
scenario["filesystem"]["manifest_committed"] = True
|
|
scenario["filesystem"]["hash_verified"] = False
|
|
assert policy.decide(scenario).decision == "hold"
|
|
|
|
|
|
def test_policy_holds_when_the_execution_anchor_is_missing():
|
|
"""Idle is not evidence of finishing; a reused session looks identical."""
|
|
from lib_py.loop import policy
|
|
scenario = json.loads((WATCHER_FIXTURES / "finalized_candidate.json").read_text(encoding="utf-8"))
|
|
scenario["dispatch_context"]["execution_anchor_observed"] = False
|
|
decision = policy.decide(scenario)
|
|
assert decision.decision == "hold" and "anchor" in decision.reason
|
|
|
|
|
|
# ==============================================================================
|
|
# 7. shadow — cannot write, proven structurally and by hashing the registry
|
|
# ==============================================================================
|
|
|
|
def test_g4_shadow_run_leaves_the_registry_byte_identical(mam_sandbox, tmp_path, monkeypatch):
|
|
"""G4: stronger than 'status was not changed' -- nothing under the registry
|
|
directory changes at all."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
from lib_py.loop import shadow
|
|
|
|
reg = Path(_reg(tmp_path))
|
|
job_ids = [_make_job(str(reg)) for _ in range(3)]
|
|
from lib_py.delegation import engine
|
|
engine.transition(job_ids[0], "claim", registry_dir=str(reg))
|
|
before = _dir_fingerprint(reg)
|
|
assert before, "fixture precondition: the registry has content to protect"
|
|
|
|
runner = shadow.ShadowRunner(workspace=str(mam_sandbox))
|
|
results = runner.evaluate_all(list(_scenarios().values()))
|
|
|
|
assert _dir_fingerprint(reg) == before, "shadow mode wrote to the registry"
|
|
assert any(r["would_complete"] for r in results), "the approved scenario must be reported"
|
|
assert all(r["shadow"] is True for r in results)
|
|
|
|
|
|
def test_g4_shadow_runner_holds_no_writer_at_all():
|
|
"""Structural, not promised: the runner has no route to a write."""
|
|
import ast
|
|
|
|
src = (SKILLS / "lib_py" / "loop" / "shadow.py").read_text(encoding="utf-8")
|
|
tree = ast.parse(src)
|
|
imported = set()
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.ImportFrom) and node.module:
|
|
for alias in node.names:
|
|
imported.add("{}.{}".format(node.module, alias.name))
|
|
elif isinstance(node, ast.Import):
|
|
for alias in node.names:
|
|
imported.add(alias.name)
|
|
assert not any("engine" in name for name in imported), imported
|
|
for forbidden in ("update_job_status", "_atomic_write_record", "transition("):
|
|
assert forbidden not in src, "shadow must not reference {}".format(forbidden)
|
|
|
|
|
|
def test_shadow_records_its_would_complete_decisions(mam_sandbox, monkeypatch):
|
|
monkeypatch.chdir(mam_sandbox)
|
|
from lib_py.loop import shadow, store
|
|
|
|
scenario = _scenarios()["finalized_candidate"]
|
|
runner = shadow.ShadowRunner(workspace=str(mam_sandbox))
|
|
entry = runner.evaluate(scenario)
|
|
assert entry["would_complete"] is True
|
|
logged = store.read_decisions(entry["job_id"], workspace=str(mam_sandbox))
|
|
assert logged and logged[-1]["decision"] == "candidate_approved"
|
|
|
|
|
|
def test_g9_loop_package_imports_only_delegation_among_siblings():
|
|
"""G9: the rule that caught the engine's original placement."""
|
|
import ast
|
|
|
|
offenders = []
|
|
for path in sorted((SKILLS / "lib_py" / "loop").glob("*.py")):
|
|
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
|
|
for node in ast.walk(tree):
|
|
names = []
|
|
if isinstance(node, ast.Import):
|
|
names = [a.name for a in node.names]
|
|
elif isinstance(node, ast.ImportFrom) and node.module:
|
|
names = [node.module]
|
|
for name in names:
|
|
if name.startswith("lib_py.") and \
|
|
not name.startswith("lib_py.loop") and \
|
|
not name.startswith("lib_py.delegation"):
|
|
offenders.append("{}:{} {}".format(path.name, node.lineno, name))
|
|
assert not offenders, offenders
|
|
|
|
|
|
# The G9 guard reads CODE, not prose. Its claim is that no frozen lib.sh
|
|
# helper is *invoked* from lib_py/loop; a comment or docstring that names
|
|
# lib.sh (say, to point at the shell twin of a function so the two stay in
|
|
# sync) invokes nothing. Every way Python could actually reach a lib.sh helper
|
|
# - a path or `source` inside a subprocess argv, an identifier - lives in code
|
|
# text, which is still scanned in full, so the guard loses none of its reach.
|
|
# History: it first fired on `hashlib.sha256` (a word boundary fixed that),
|
|
# then on a docstring cross-reference in watcher.py; both were mentions.
|
|
_G9_PATTERNS = {
|
|
"ensure_subscriber_daemon": r"ensure_subscriber_daemon",
|
|
"preexec_fn": r"preexec_fn",
|
|
"lib.sh": r"\blib\.sh\b", # word boundary: `hashlib.sha256` is not lib.sh
|
|
}
|
|
|
|
|
|
def _code_only(src):
|
|
"""`src` as code only: the docstring expression statements are removed from
|
|
the AST and the rest is unparsed, which also drops comments (they are not
|
|
in the AST). Only the docstring *node* goes, so a statement that shares
|
|
its line - a docstring followed by `; call()` - is kept, as are string literals in code
|
|
(a path or `source` in a subprocess argv is exactly what must be caught).
|
|
Review R2: blanking the docstring's whole line range hid same-line code."""
|
|
import ast
|
|
|
|
tree = ast.parse(src)
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, (ast.Module, ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)):
|
|
body = getattr(node, "body", None) or []
|
|
first = body[0] if body else None
|
|
if (isinstance(first, ast.Expr) and isinstance(first.value, ast.Constant)
|
|
and isinstance(first.value.value, str)):
|
|
del body[0]
|
|
if not body:
|
|
body.append(ast.Pass())
|
|
return ast.unparse(tree)
|
|
|
|
|
|
def _g9_offenders(src):
|
|
import re
|
|
code = _code_only(src)
|
|
return [name for name, pat in _G9_PATTERNS.items() if re.search(pat, code)]
|
|
|
|
|
|
def test_g9_phase2a_spawns_no_daemon_and_calls_no_frozen_helper():
|
|
"""The two phase 1 handoffs stay dormant: preexec_fn is unreachable because
|
|
nothing here spawns, and no frozen lib.sh helper is invoked."""
|
|
for path in sorted((SKILLS / "lib_py" / "loop").glob("*.py")):
|
|
src = path.read_text(encoding="utf-8")
|
|
hits = _g9_offenders(src)
|
|
assert not hits, "{} references {}".format(path.name, ", ".join(hits))
|
|
|
|
|
|
def test_g9b_the_guard_still_catches_an_invocation_and_ignores_prose():
|
|
"""The narrowing must not cost the guard its purpose: an invocation-shaped
|
|
reference in code text still trips it; the same words in a docstring or a
|
|
comment do not."""
|
|
invoked = (
|
|
'import subprocess\n'
|
|
'def go():\n'
|
|
' return subprocess.run(["bash", "-c", "source .agents/skills/lib.sh; ensure_subscriber_daemon x"])\n'
|
|
)
|
|
assert set(_g9_offenders(invoked)) == {"lib.sh", "ensure_subscriber_daemon"}
|
|
path_only = 'CMD = ["bash", "/x/.agents/skills/lib.sh"]\n'
|
|
assert _g9_offenders(path_only) == ["lib.sh"]
|
|
prose = (
|
|
'"""Module doc: same convention as lib.sh\'s _herdr_caller_server."""\n'
|
|
'import hashlib\n'
|
|
'def f():\n'
|
|
' """See lib.sh; never call ensure_subscriber_daemon from here."""\n'
|
|
' # preexec_fn is documented in lib.sh\n'
|
|
' return hashlib.sha256(b"x").hexdigest()\n'
|
|
)
|
|
assert _g9_offenders(prose) == []
|
|
# Review R2: code sharing a line with a docstring must still be seen -
|
|
# module, one-line body, one-line def, and the last line of a multi-line
|
|
# docstring. Only the docstring node is removed, never its line.
|
|
same_line = [
|
|
'"""module doc"""; ensure_subscriber_daemon()\n',
|
|
'def f():\n """function doc"""; ensure_subscriber_daemon()\n',
|
|
'def f(): """function doc"""; ensure_subscriber_daemon()\n',
|
|
'def f():\n """multi\n line"""; ensure_subscriber_daemon()\n',
|
|
]
|
|
for snippet in same_line:
|
|
assert _g9_offenders(snippet) == ["ensure_subscriber_daemon"], snippet
|
|
|
|
|
|
# ==============================================================================
|
|
# 8. Cross-review regressions (job 9d1e0077)
|
|
#
|
|
# F-A (claude): the CAS table omitted two legitimate wrapper transitions, so
|
|
# `submit --type loop|discuss` silently stopped working by default
|
|
# and the CLI still exited 0.
|
|
# F-B (claude): the audit status mirror moved outside the lock and became
|
|
# last-writer-wins, so a slow flush could leave it behind.
|
|
# R1 (codex): the policy trusted the dispatch record's own flags and never
|
|
# confronted the identifiers the evidence carried.
|
|
# R2 (codex): a supplied expected_attempt was ignored when the record had none.
|
|
# R3 (codex): a raw status field rode in on a metadata merge.
|
|
# ==============================================================================
|
|
|
|
@pytest.mark.parametrize("mode", CAS_MODES)
|
|
def test_fa_wrapper_loop_redispatch_and_finalize_work(mam_sandbox, tmp_path, monkeypatch, mode):
|
|
"""F-A: the exact CLI calls the wrapper makes for --type loop|discuss.
|
|
|
|
Iteration reset hands a finished job to the next session; exhausting the
|
|
budget finalises it as an error. Both must work in either mode."""
|
|
monkeypatch.setenv("MAM_ENGINE_CAS", mode)
|
|
monkeypatch.chdir(mam_sandbox)
|
|
from lib_py.delegation import mqtt_common, registry
|
|
|
|
reg = _reg(tmp_path)
|
|
job_id = _make_job(reg, agent_session="herdr:claude", job_type="loop")
|
|
mqtt_common.update_job_status(job_id, reg, _source="publish_event", status="completed")
|
|
|
|
rc = registry.main(["--registry-dir", reg, "update", "--job", job_id,
|
|
"--agent-session", "herdr:rev", "--prompt", "review",
|
|
"--iteration", "2", "--role", "Reviewer", "--status", "pending"])
|
|
record = _load(reg, job_id)
|
|
assert rc == 0, "the wrapper's iteration reset must succeed"
|
|
assert record["status"] == "pending"
|
|
assert record["iteration"] == 2, "metadata must not be discarded"
|
|
assert record["agent_session"] == "herdr:rev"
|
|
|
|
mqtt_common.update_job_status(job_id, reg, _source="publish_event", status="completed")
|
|
rc2 = registry.main(["--registry-dir", reg, "status", "--job", job_id, "--set", "error"])
|
|
assert rc2 == 0
|
|
assert _load(reg, job_id)["status"] == "error"
|
|
|
|
|
|
def test_fa_redispatch_requires_operator_authority_and_a_forward_attempt(
|
|
mam_sandbox, tmp_path, monkeypatch
|
|
):
|
|
"""The new authority must not become a way for an agent to undo a 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)
|
|
engine.transition(job_id, "completed", registry_dir=reg)
|
|
|
|
agent = engine.transition(job_id, "redispatch", registry_dir=reg,
|
|
fields={"iteration": 2}, source="publish_event")
|
|
assert agent.committed is False and "operator authority" in agent.reason
|
|
|
|
stale = engine.transition(job_id, "redispatch", registry_dir=reg,
|
|
fields={"iteration": 1}, source="registry_cli")
|
|
assert stale.committed is False and "does not advance" in stale.reason
|
|
|
|
ok = engine.transition(job_id, "redispatch", registry_dir=reg,
|
|
fields={"iteration": 2}, source="registry_cli")
|
|
assert ok.committed and ok.to_status == "pending"
|
|
|
|
|
|
def test_fa_agent_still_cannot_move_a_terminal(mam_sandbox, tmp_path, monkeypatch):
|
|
"""The protection the CAS table exists for must survive the new authority."""
|
|
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, _source="publish_event", status="completed")
|
|
mqtt_common.update_job_status(job_id, reg, _source="publish_event", status="error")
|
|
assert _load(reg, job_id)["status"] == "completed"
|
|
|
|
|
|
def test_fa_a_refused_operator_write_exits_non_zero(mam_sandbox, tmp_path, monkeypatch):
|
|
"""The silence is what hid F-A: a refused write used to exit 0."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
from lib_py.delegation import engine, registry
|
|
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)
|
|
# cancelled after completed is a genuine conflict, and must be reported
|
|
rc = registry.main(["--registry-dir", reg, "status", "--job", job_id, "--set", "cancelled"])
|
|
assert rc != 0, "a refused operator write must not look like success"
|
|
assert _load(reg, job_id)["status"] == "completed"
|
|
|
|
|
|
def test_fb_a_late_status_mirror_cannot_overwrite_a_newer_one(mam_sandbox, tmp_path, monkeypatch):
|
|
"""F-B: the mirror runs outside the lock, so ordering rides on transition_seq."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
from lib_py.delegation import mqtt_common
|
|
reg = _reg(tmp_path)
|
|
job_id = _make_job(reg)
|
|
logs = mqtt_common.get_logs_dir()
|
|
|
|
mqtt_common.update_logged_status(job_id, "completed", logs, transition_seq=2)
|
|
mqtt_common.update_logged_status(job_id, "running", logs, transition_seq=1) # late flush
|
|
assert mqtt_common.read_logged_status(job_id, logs)["status"] == "completed"
|
|
# a genuinely newer transition still lands
|
|
mqtt_common.update_logged_status(job_id, "error", logs, transition_seq=3)
|
|
assert mqtt_common.read_logged_status(job_id, logs)["status"] == "error"
|
|
|
|
|
|
@pytest.mark.parametrize("field,value", [
|
|
("agent_session_uuid", "someone-else"),
|
|
("attempt_id", 999),
|
|
("job_id", "another-job"),
|
|
])
|
|
def test_r1_policy_rejects_evidence_that_contradicts_the_dispatch(field, value):
|
|
"""R1: attribution must confront what was observed, not just the record."""
|
|
from lib_py.loop import policy
|
|
import copy
|
|
scenario = json.loads((WATCHER_FIXTURES / "finalized_candidate.json").read_text(encoding="utf-8"))
|
|
assert policy.decide(scenario).decision == "candidate_approved"
|
|
mutated = copy.deepcopy(scenario)
|
|
# The collector envelope is where an identifier legitimately lives; a native
|
|
# payload is not an identifier source at all.
|
|
mutated["timeline"][2]["collector_attribution"][field] = value
|
|
decision = policy.decide(mutated)
|
|
assert decision.decision == "hold", "{}={!r} must not be approved".format(field, value)
|
|
assert field in decision.reason
|
|
|
|
|
|
def test_r1_policy_rejects_a_manifest_from_another_attempt():
|
|
"""A finalized report proves *an* attempt finished, not necessarily this one."""
|
|
from lib_py.loop import policy
|
|
import copy
|
|
scenario = json.loads((WATCHER_FIXTURES / "finalized_candidate.json").read_text(encoding="utf-8"))
|
|
mutated = copy.deepcopy(scenario)
|
|
mutated["timeline"][1]["payload"]["report_manifest"]["attempt_id"] = 999
|
|
decision = policy.decide(mutated)
|
|
assert decision.decision == "hold" and "manifest" in decision.reason
|
|
|
|
|
|
def test_r2_expected_attempt_must_be_proved_not_merely_uncontradicted(
|
|
mam_sandbox, tmp_path, monkeypatch
|
|
):
|
|
"""R2: a precondition the record cannot answer must fail closed."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
from lib_py.delegation import engine, mqtt_common
|
|
reg = _reg(tmp_path)
|
|
job_id = _make_job(reg)
|
|
engine.transition(job_id, "claim", registry_dir=reg)
|
|
|
|
record = _load(reg, job_id)
|
|
record.pop("iteration", None)
|
|
record.pop("attempt_id", None)
|
|
mqtt_common._atomic_write_record(job_id, reg, record)
|
|
before = _dir_fingerprint(Path(reg))
|
|
|
|
res = engine.transition(job_id, "completed", registry_dir=reg, expected_attempt=999)
|
|
assert res.committed is False and "carries none" in res.reason
|
|
assert _dir_fingerprint(Path(reg)) == before, "a rejection must not write"
|
|
|
|
# with an attempt present the precondition behaves normally
|
|
engine.transition(job_id, "progress", registry_dir=reg, fields={"attempt_id": 7})
|
|
assert engine.transition(job_id, "completed", registry_dir=reg,
|
|
expected_attempt=8).committed is False
|
|
assert engine.transition(job_id, "completed", registry_dir=reg,
|
|
expected_attempt=7).committed is True
|
|
|
|
|
|
@pytest.mark.parametrize("request_", ["progress", "permission_required", "merge"])
|
|
def test_r3_a_status_cannot_ride_in_as_a_metadata_field(
|
|
mam_sandbox, tmp_path, monkeypatch, request_
|
|
):
|
|
"""R3: the boundary itself must refuse it, not just one wrapper."""
|
|
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={"status": "completed"})
|
|
assert res.committed is False and "must be requested" in res.reason
|
|
assert _load(reg, job_id)["status"] == "running"
|
|
|
|
|
|
def test_r3_an_unknown_request_is_refused(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, "not_a_request", registry_dir=reg, fields={"x": 1})
|
|
assert res.committed is False and "unknown request" in res.reason
|
|
assert "x" not in _load(reg, job_id)
|
|
|
|
|
|
# ==============================================================================
|
|
# 9. Cross-review round 2 regressions (job 57166ae2)
|
|
#
|
|
# F-C (claude): an idempotent ACK was reported as a refusal, so the wrapper's
|
|
# approved path aborted under set -e before its validation hook.
|
|
# F-D (claude): routing around update_status lost the status vocabulary check.
|
|
# R1 (codex): absence of evidence still approved; only present identifiers
|
|
# were compared.
|
|
# R4 (codex): the mirror's ordering guard was a check followed by a write with
|
|
# no lock between them.
|
|
# ==============================================================================
|
|
|
|
@pytest.mark.parametrize("mode", CAS_MODES)
|
|
def test_fc_idempotent_finalize_is_success_not_refusal(mam_sandbox, tmp_path, monkeypatch, mode):
|
|
"""F-C: the wrapper finalises an approved job that is already completed.
|
|
|
|
It runs under `set -e`, so reporting failure there aborts the success path
|
|
before the validation hook and the final stdout line."""
|
|
monkeypatch.setenv("MAM_ENGINE_CAS", mode)
|
|
monkeypatch.chdir(mam_sandbox)
|
|
from lib_py.delegation import mqtt_common, registry
|
|
|
|
reg = _reg(tmp_path)
|
|
job_id = _make_job(reg)
|
|
mqtt_common.update_job_status(job_id, reg, _source="publish_event", status="completed")
|
|
|
|
rc = registry.main(["--registry-dir", reg, "status", "--job", job_id, "--set", "completed"])
|
|
assert rc == 0, "re-affirming a state that already holds is an ack, not a failure"
|
|
assert _load(reg, job_id)["status"] == "completed"
|
|
|
|
|
|
def test_fc_a_genuine_conflict_is_still_reported(mam_sandbox, tmp_path, monkeypatch):
|
|
"""The F-C fix must not soften a real refusal back into silence."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
from lib_py.delegation import mqtt_common, registry
|
|
reg = _reg(tmp_path)
|
|
job_id = _make_job(reg)
|
|
mqtt_common.update_job_status(job_id, reg, _source="publish_event", status="completed")
|
|
rc = registry.main(["--registry-dir", reg, "status", "--job", job_id, "--set", "cancelled"])
|
|
assert rc == 4
|
|
assert _load(reg, job_id)["status"] == "completed"
|
|
|
|
|
|
def test_fc_engine_marks_an_idempotent_result(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)
|
|
again = engine.transition(job_id, "completed", registry_dir=reg)
|
|
assert again.committed is False and again.idempotent is True
|
|
conflict = engine.transition(job_id, "cancelled", registry_dir=reg)
|
|
assert conflict.committed is False and conflict.idempotent is False
|
|
|
|
|
|
@pytest.mark.parametrize("mode", CAS_MODES)
|
|
def test_fd_invalid_status_is_rejected_in_both_modes(mam_sandbox, tmp_path, monkeypatch, mode):
|
|
"""F-D: the vocabulary check must survive the routing, especially on the
|
|
rollback path where the engine is not there to catch it."""
|
|
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)
|
|
before = _dir_fingerprint(Path(reg))
|
|
rc = registry.main(["--registry-dir", reg, "status", "--job", job_id, "--set", "bogus"])
|
|
assert rc == 1, "an invalid status is a usage error, not a CAS refusal"
|
|
assert _dir_fingerprint(Path(reg)) == before, "an invalid status must not be written"
|
|
|
|
|
|
@pytest.mark.parametrize("field", [
|
|
"job_id", "attempt_id", "agent_session_uuid", "pane_process_generation",
|
|
])
|
|
def test_r1_missing_attribution_evidence_holds(field):
|
|
"""R1: absence of proof is not proof. Every identifier must be supplied."""
|
|
import copy
|
|
from lib_py.loop import policy
|
|
scenario = json.loads((WATCHER_FIXTURES / "finalized_candidate.json").read_text(encoding="utf-8"))
|
|
assert policy.decide(scenario).decision == "candidate_approved"
|
|
|
|
mutated = copy.deepcopy(scenario)
|
|
for item in mutated["timeline"]:
|
|
(item.get("collector_attribution") or {}).pop(field, None)
|
|
(item.get("payload") or {}).pop(field, None)
|
|
decision = policy.decide(mutated)
|
|
assert decision.decision == "hold", "missing {} must not be approved".format(field)
|
|
assert field in decision.reason
|
|
|
|
|
|
def test_r1_missing_report_manifest_holds():
|
|
import copy
|
|
from lib_py.loop import policy
|
|
scenario = json.loads((WATCHER_FIXTURES / "finalized_candidate.json").read_text(encoding="utf-8"))
|
|
mutated = copy.deepcopy(scenario)
|
|
for item in mutated["timeline"]:
|
|
(item.get("payload") or {}).pop("report_manifest", None)
|
|
decision = policy.decide(mutated)
|
|
assert decision.decision == "hold" and "manifest" in decision.reason
|
|
|
|
|
|
def test_r1_identifiers_inside_a_native_payload_are_not_evidence():
|
|
"""A MAM field cannot appear in a native Herdr payload, so one found there is
|
|
either fabricated or a contract violation. Either way it proves nothing."""
|
|
import copy
|
|
from lib_py.loop import policy
|
|
scenario = json.loads((WATCHER_FIXTURES / "finalized_candidate.json").read_text(encoding="utf-8"))
|
|
mutated = copy.deepcopy(scenario)
|
|
dispatch = mutated["dispatch_context"]
|
|
herdr_item = [i for i in mutated["timeline"] if i["source_kind"] == "herdr"][-1]
|
|
herdr_item.pop("collector_attribution", None)
|
|
for field in ("job_id", "attempt_id", "agent_session_uuid"):
|
|
herdr_item["payload"][field] = dispatch[field]
|
|
assert policy.decide(mutated).decision != "candidate_approved"
|
|
|
|
|
|
def test_r4_overlapping_mirror_flushes_cannot_lose_the_newer_state(mam_sandbox, monkeypatch):
|
|
"""R4: a sequence number orders nothing unless compare and write are atomic.
|
|
|
|
Interrupts a flush after it has read, lets a newer one complete, and requires
|
|
the newer state to survive."""
|
|
import threading
|
|
from lib_py.delegation import mqtt_common
|
|
|
|
monkeypatch.chdir(mam_sandbox)
|
|
logs = mqtt_common.get_logs_dir()
|
|
job_id = "r4probe"
|
|
mqtt_common.update_logged_status(job_id, "pending", logs, transition_seq=0)
|
|
path = mqtt_common.job_log_path(job_id, mqtt_common.STATUS_FILENAME, logs)
|
|
|
|
real_read = Path.read_text
|
|
state = {}
|
|
|
|
def intercept(self, *a, **k):
|
|
text = real_read(self, *a, **k)
|
|
if str(self) == str(path) and "fired" not in state:
|
|
state["fired"] = True
|
|
thread = threading.Thread(
|
|
target=mqtt_common.update_logged_status,
|
|
args=(job_id, "completed", logs), kwargs={"transition_seq": 2})
|
|
thread.start()
|
|
state["thread"] = thread
|
|
thread.join(timeout=0.5)
|
|
state["blocked"] = thread.is_alive()
|
|
return text
|
|
|
|
monkeypatch.setattr(Path, "read_text", intercept)
|
|
mqtt_common.update_logged_status(job_id, "running", logs, transition_seq=1)
|
|
monkeypatch.undo()
|
|
state["thread"].join(timeout=5)
|
|
|
|
assert state.get("blocked") is True, "the competing flush must serialise, not race"
|
|
final = json.loads(path.read_text(encoding="utf-8"))
|
|
assert final["status"] == "completed" and final["transition_seq"] == 2
|
|
|
|
|
|
def test_r4_each_mirror_write_uses_its_own_temp_file(mam_sandbox, monkeypatch):
|
|
"""A shared temp name lets concurrent writers clobber each other pre-rename."""
|
|
monkeypatch.chdir(mam_sandbox)
|
|
from lib_py.delegation import mqtt_common
|
|
source = Path(SKILLS / "lib_py" / "delegation" / "mqtt_common.py").read_text(encoding="utf-8")
|
|
body = source[source.index("def update_logged_status"):source.index("def init_job_log")]
|
|
assert "mkstemp" in body, "the mirror must not reuse one fixed .tmp path"
|
|
logs = mqtt_common.get_logs_dir()
|
|
mqtt_common.update_logged_status("tmpprobe", "running", logs, transition_seq=1)
|
|
leftovers = list((Path(logs) / "tmpprobe").glob("*.tmp"))
|
|
assert not leftovers, "temp files must be renamed away, not left behind: {}".format(leftovers)
|