fix(subscriber): the disk fallback takes an audit status only when it is at least as new as the record

The audit mirror is written after the registry commits, so a redispatched
record (attempt 2, pending) can sit next to attempt 1's completed
status.json, and a failed mirror leaves it there. The fallback validated the
token and attempt against the record, which say nothing about the audit
status's provenance, and closed attempt 2 with attempt 1's completion
(review P1, Issue #57). A versioned record (transition_seq) now takes the
audit status only when its transition_seq is at least as new; an audit
status without a sequence cannot prove which attempt it describes and is
ignored by a versioned record. An unversioned record (never transitioned,
or MAM_ENGINE_CAS=0) and a missing record keep the old rule.

Regressions: the redispatch/flush interval, the failed-mirror leftover, the
unversioned audit status, a newer audit-only terminal, and conflict
refusals and legacy claims next to a moved registry.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

.agents/skills/lib_py/delegation/job_subscriber.py

@@ -84,11 +84,13 @@ def _check_disk_fallback(
         disk_status = None
         rec_token = None
         rec_attempt = None
+        rec_seq = None
         try:
             job_rec = load_job(jid, registry_dir)
             disk_status = job_rec.get("status")
             rec_token = job_rec.get("dispatch_token")
             rec_attempt = job_rec.get("attempt_id", job_rec.get("iteration"))
+            rec_seq = job_rec.get("transition_seq")
         except Exception:
             job_rec = {}
         if not disk_status or disk_status not in ("completed", "error", "cancelled"):
@@ -99,7 +101,20 @@ def _check_disk_fallback(
             except Exception:
                 logged = None
             if logged:
-                disk_status = logged.get("status")
+                # The audit mirror is written after the registry commits, so a
+                # record that is not terminal may sit next to the previous
+                # attempt's terminal status. A versioned record (the engine's
+                # transition_seq) takes the audit status only when the audit is
+                # at least as new; an audit status without a sequence cannot
+                # prove which attempt it describes, so a versioned record
+                # ignores it. An unversioned record (never transitioned, or
+                # MAM_ENGINE_CAS=0) and a missing record keep the old rule.
+                aud_seq = logged.get("transition_seq")
+                if rec_seq is not None and (aud_seq is None or int(aud_seq) < int(rec_seq)):
+                    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")
 
         if disk_status in ("completed", "error", "cancelled"):
             if expected_dispatch_tokens:

tests/test_issue57_logs_follow_registry.py

@@ -35,7 +35,7 @@ if str(SKILLS) not in sys.path:
     sys.path.insert(0, str(SKILLS))
 
 from lib_py import control_plane  # noqa: E402
-from lib_py.delegation import job_subscriber, mqtt_common, registry  # noqa: E402
+from lib_py.delegation import engine, job_subscriber, mqtt_common, registry  # noqa: E402
 from test_issue21_phase2_loop_a import _mutate, _prepare, _rows, _skill, _wt  # noqa: E402
 from test_issue21_phase2_loop_a import ctl  # noqa: E402,F401
 
@@ -310,3 +310,67 @@ d['herdr_sessions'] = [{{'name': 's57-creator-claude', 'status': 'running', 'del
         (line,) = [l for l in res.stdout.splitlines() if l.startswith("s57-creator-claude")]
         assert jid in line and "completed" in line, line
     assert not (tmp / ".mam" / "delegate_job_logs").exists()
+
+
+# ---------------------------------------------------------------------------
+# review (codex P2): conflict refusals and legacy claims follow the registry too
+# ---------------------------------------------------------------------------
+
+@pytest.mark.parametrize("path", ["conflict", "legacy-claim"])
+def test_conflict_refusals_and_legacy_claims_follow_the_registry(moved, monkeypatch, path):
+    ws, reg, logs = moved
+    jid = registry.register_job("p2", registry_dir=str(reg), agent_session="herdr:t57")
+    if path == "conflict":
+        assert engine.transition(jid, "completed", registry_dir=str(reg), source="operator").committed
+        assert not engine.transition(jid, "error", registry_dir=str(reg), source="late").committed
+        expected = "conflict_refused"
+    else:
+        monkeypatch.setenv("MAM_ENGINE_CAS", "0")
+        assert registry.pick_pending("herdr:t57", registry_dir=str(reg)) == jid
+        expected = "status_changed"
+    assert expected in [e.get("event") for e in mqtt_common.iter_logged_events(jid, str(logs))]
+
+
+# ---------------------------------------------------------------------------
+# review (codex P1): an old audit terminal must not close a redispatched attempt
+# ---------------------------------------------------------------------------
+
+def test_old_audit_terminal_does_not_complete_the_redispatched_attempt(moved, monkeypatch):
+    ws, reg, logs = moved
+    jid = registry.register_job("p1", registry_dir=str(reg), agent_session="herdr:t57")
+    assert engine.transition(jid, "completed", registry_dir=str(reg), source="operator").committed
+    old_audit = mqtt_common.read_logged_status(jid, str(logs))
+    flush, observed = engine._flush, {}
+
+    def before_the_mirror(notes, registry_dir):
+        # the real gap: attempt 2 is committed, the audit still says attempt 1 completed
+        rec = mqtt_common.load_job(jid, registry_dir)
+        assert rec["status"] == "pending" and rec["attempt_id"] == 2
+        assert old_audit["transition_seq"] < rec["transition_seq"]
+        pending, terminal = {jid}, {}
+        job_subscriber._check_disk_fallback(pending, registry_dir, terminal, {jid: rec["dispatch_token"]}, {jid: 2})
+        observed.update(pending=pending, terminal=terminal)
+        flush(notes, registry_dir)
+
+    monkeypatch.setattr(engine, "_flush", before_the_mirror)
+    assert engine.transition(jid, "redispatch", registry_dir=str(reg), source="operator",
+                             fields={"attempt_id": 2}).committed
+    assert observed == {"pending": {jid}, "terminal": {}}, "the old completion was attributed to attempt 2"
+    rec = mqtt_common.load_job(jid, str(reg))
+    # a failed mirror leaves the old status.json behind (update_logged_status
+    # itself refuses to go backwards, so the file is written directly): still ignored
+    mqtt_common.job_log_path(jid, mqtt_common.STATUS_FILENAME, str(logs)).write_text(json.dumps(old_audit))
+    assert mqtt_common.read_logged_status(jid, str(logs))["status"] == "completed"
+    pending, terminal = {jid}, {}
+    job_subscriber._check_disk_fallback(pending, str(reg), terminal, {jid: rec["dispatch_token"]}, {jid: 2})
+    assert pending == {jid} and terminal == {}
+    # an audit status without a sequence cannot prove it is about this attempt
+    mqtt_common.update_logged_status(jid, "completed", logs_dir=str(logs))
+    pending, terminal = {jid}, {}
+    job_subscriber._check_disk_fallback(pending, str(reg), terminal, {jid: rec["dispatch_token"]}, {jid: 2})
+    assert pending == {jid} and terminal == {}
+    # a newer audit-only terminal is still taken
+    mqtt_common.update_logged_status(jid, "completed", logs_dir=str(logs), transition_seq=rec["transition_seq"] + 1)
+    pending, terminal = {jid}, {}
+    job_subscriber._check_disk_fallback(pending, str(reg), terminal, {jid: rec["dispatch_token"]}, {jid: 2})
+    assert pending == set() and terminal == {jid: "completed"}