From 1c598bb54651ebacb73f917e4cd4b754ebfe7ab8 Mon Sep 17 00:00:00 2001 From: Godopu Date: Thu, 15 Oct 2026 00:04:52 +0000 Subject: [PATCH] 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 --- .../lib_py/delegation/job_subscriber.py | 19 ++++++++- tests/test_issue57_logs_follow_registry.py | 42 +++++++++++++++++++ 2 files changed, 60 insertions(+), 1 deletion(-) diff --git a/.agents/skills/lib_py/delegation/job_subscriber.py b/.agents/skills/lib_py/delegation/job_subscriber.py index dc5e424..12948fa 100755 --- a/.agents/skills/lib_py/delegation/job_subscriber.py +++ b/.agents/skills/lib_py/delegation/job_subscriber.py @@ -118,7 +118,24 @@ def _check_disk_fallback( logger.warning("disk fallback ignored audit status %r for %s: transition_seq %r is older " "than the record's %r", logged.get("status"), jid, aud_seq, rec_seq) else: - disk_status = logged.get("status") + # The record above was read before the audit status. If + # another dispatch began in between (A error -> redispatch + # -> B completed), the audit terminal is B's while the token + # and attempt checked below are A's. Take it only when the + # record still describes the same observation; otherwise + # leave the job pending for the next round, which reads + # both afresh (Issue #57, Mac review R1). + try: + now = load_job(jid, registry_dir) + except Exception: + now = {} + same = ((now.get("dispatch_token"), now.get("attempt_id", now.get("iteration")), + now.get("transition_seq")) == (rec_token, rec_attempt, rec_seq)) + if same: + disk_status = logged.get("status") + else: + logger.warning("disk fallback deferred %s: the record changed while the audit status " + "was being read", jid) if disk_status in ("completed", "error", "cancelled"): if expected_dispatch_tokens: diff --git a/tests/test_issue57_logs_follow_registry.py b/tests/test_issue57_logs_follow_registry.py index 94c544c..28c3673 100644 --- a/tests/test_issue57_logs_follow_registry.py +++ b/tests/test_issue57_logs_follow_registry.py @@ -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 == {}