fix(subscriber): the disk fallback re-reads the record before taking an audit status (Issue #57, Mac R1)

The record (status, token, attempt, transition_seq) was read before the
audit status.json; if A error -> redispatch (attempt 2) -> B completed
happened in between, the audit terminal was B's while the token and attempt
validated were A's, and A's caller was told "completed". The audit status
is now taken only when a fresh read of the record still shows the same
token, attempt and transition_seq; otherwise the job stays pending and the
next round reads both afresh. The no-record case is unchanged.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
GodopuandClaude Fable 5.1 committed 2026-10-15 00:04:52 +00:00
1 parent 151af99765
commit 1c598bb546
2 files changed
+60 -1

No files matched your search

@@ -557,3 +557,45 @@ def test_logs_without_registry_dir_follows_the_workspace_from_any_cwd(tmp_path,
res = _wrapper(ws, "logs", jid, cwd=elsewhere)
assert res.returncode == 0 and "registered" in res.stdout, res.stdout + res.stderr
assert not (elsewhere / ".mam").exists()
# ---------------------------------------------------------------------------
# Mac review R1: an audit terminal of a later dispatch is not the earlier one's success
# ---------------------------------------------------------------------------
def test_disk_fallback_rejects_a_terminal_that_belongs_to_a_later_dispatch(moved, monkeypatch):
"""Between the registry read and the audit read: A error -> redispatch
(attempt 2) -> B completed. The caller waiting on A must not be told
"completed"; the caller of B must."""
ws, reg, logs = moved
jid = registry.register_job("r1", registry_dir=str(reg), agent_session="herdr:t57")
assert registry.pick_pending("herdr:t57", registry_dir=str(reg)) == jid # claim: running, token A
rec = mqtt_common.load_job(jid, str(reg))
tok_a = rec["dispatch_token"]
assert rec["status"] == "running" and rec["attempt_id"] == 1
read, fired = mqtt_common.read_logged_status, []
def interleave(job, logs_dir=None):
if not fired: # once, right between the two reads, with real transitions
fired.append(1)
assert engine.transition(jid, "error", registry_dir=str(reg), source="operator").committed
assert engine.transition(jid, "redispatch", registry_dir=str(reg), source="operator",
fields={"attempt_id": 2}).committed
assert engine.transition(jid, "completed", registry_dir=str(reg), source="operator").committed
return read(job, logs_dir)
monkeypatch.setattr(mqtt_common, "read_logged_status", interleave)
pending, terminal = {jid}, {}
job_subscriber._check_disk_fallback(pending, str(reg), terminal, {jid: tok_a}, {jid: 1})
assert fired, "the interleaving did not run"
assert pending == {jid} and terminal == {}, "B's completion was attributed to A"
rec = mqtt_common.load_job(jid, str(reg))
tok_b = rec["dispatch_token"]
assert tok_b != tok_a and rec["attempt_id"] == 2 and rec["status"] == "completed"
# the next round, and the caller of B, see the truth
pending, terminal = {jid}, {}
job_subscriber._check_disk_fallback(pending, str(reg), terminal, {jid: tok_b}, {jid: 2})
assert pending == set() and terminal == {jid: "completed"}
pending, terminal = {jid}, {}
job_subscriber._check_disk_fallback(pending, str(reg), terminal, {jid: tok_a}, {jid: 1})
assert pending == {jid} and terminal == {}