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 == {}