fix(issue-5): bind the auto-completion decision atomically to its attempt

Two findings from round 2, both reproduced before being accepted.

R2 -- attribution was compared before the terminal write and nothing tied the
two together. A job could complete, be re-dispatched and claimed again in that
window, and evidence from the first attempt then satisfied a status check and
finished the second one. Reproduced with real transitions: committed, with the
registry already on iteration 2.

This corrects my own earlier reasoning, and I want to be plain about it. Last
round I argued that the attempt guard could live entirely in the evidence layer
because that is where attempt identity lives. That holds for comparing evidence
to a dispatch, and it does not hold for atomicity: a precondition can only be
atomic with the commit if the transaction can read it under its own lock. So the
identity now lives in the record.

It is an opaque dispatch token, not the iteration counter. That distinction is
the part of the earlier rebuttal that still stands: iteration already means the
loop round, and comparing a dispatch attempt against it produces a guard that is
vacuous when they agree and wrong when they diverge. A token has no competing
meaning. It is minted when an attempt begins -- claim and redispatch -- and the
gate passes the token it decided against, so any intervening re-dispatch refuses
the commit.

R4 -- the intent identity was being emitted but nothing consumed it. A replayed
notification is re-published with a fresh wire sequence, so the monotonic guard
cannot recognise it; to that guard a retry looks new. The subscriber now
deduplicates on the stable intent identity, leaving the sequence guard as the
rule for ordinary events. Across a restart the disk fallback resolves terminal
status and a duplicate terminal is refused as an idempotent ack, so a replay
after restart cannot double-apply either.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UrDNJkHmm18uu7n7JwV5Qd

.agents/skills/lib_py/delegation/engine.py

@@ -28,6 +28,7 @@ from __future__ import annotations
 
 import logging
 import os
+import uuid
 from typing import Any, Dict, List, Optional
 
 from lib_py.delegation import mqtt_common, notifier
@@ -71,6 +72,20 @@ BOOKKEEPING_FIELDS = ("pending_notification",)
 # Distinguishes "no expectation" from "expect the field to be None".
 _UNSET = object()
 
+# An opaque identity for one dispatch attempt, minted when an attempt begins and
+# stored in the record itself. It lives here, not only in the shadow store,
+# because a precondition can only be atomic with the commit if the transaction
+# can read it under its own lock. Comparing the shadow store before the call
+# leaves a window in which the job is completed, re-dispatched and claimed again,
+# after which stale evidence satisfies a status check and completes a new
+# attempt. An opaque token is used rather than the iteration counter because
+# iteration already means the loop round, and conflating the two produces a guard
+# that is either vacuous or wrong.
+DISPATCH_TOKEN_FIELD = "dispatch_token"
+
+# The requests that begin a new attempt, and therefore mint a new token.
+ATTEMPT_STARTING_REQUESTS = ("claim", "redispatch")
+
 # Only these sources may move a record out of a terminal state. An agent can
 # never walk back its own terminal; that is the whole point of the CAS table.
 OPERATOR_SOURCES = ("registry_cli", "operator")
@@ -128,7 +143,8 @@ class _Decision(object):
 
 
 def _decide(record: Dict[str, Any], request: str, expected_status, expected_attempt,
-            source="unknown", fields=None, expected_notification=_UNSET) -> _Decision:
+            source="unknown", fields=None, expected_notification=_UNSET,
+            expected_dispatch_token=None) -> _Decision:
     """Pure CAS decision. No I/O, so it is safe to call under the lock.
 
     Implements the target table in docs/contracts/issue-5/state-transitions.md §3.
@@ -162,6 +178,13 @@ def _decide(record: Dict[str, Any], request: str, expected_status, expected_atte
                              "stale attempt: expected {!r}, record has {!r}".format(
                                  expected_attempt, actual))
 
+    if expected_dispatch_token is not None:
+        actual_token = record.get(DISPATCH_TOKEN_FIELD)
+        if actual_token != expected_dispatch_token:
+            return _Decision(False, current,
+                             "dispatch token changed: the attempt this decision was "
+                             "made for is no longer the current one")
+
     if expected_status is not None and current != expected_status:
         return _Decision(False, current,
                          "precondition failed: expected status {!r}, found {!r}".format(
@@ -243,7 +266,8 @@ def _notification(job_id, kind, **fields) -> Dict[str, Any]:
 
 def transition(job_id, request="merge", *, registry_dir=None, expected_status=None,
                expected_attempt=None, fields=None, source="unknown",
-               notify=None, expected_notification=_UNSET) -> TransitionResult:
+               notify=None, expected_notification=_UNSET,
+               expected_dispatch_token=None) -> TransitionResult:
     """Decide and commit one state transition. The only writer of job status.
 
     Returns a result whose `notifications` have ALREADY been flushed; the field
@@ -258,7 +282,8 @@ def transition(job_id, request="merge", *, registry_dir=None, expected_status=No
         from_status = record.get("status")
         decision = _decide(record, request, expected_status, expected_attempt,
                            source=source, fields=fields,
-                           expected_notification=expected_notification)
+                           expected_notification=expected_notification,
+                           expected_dispatch_token=expected_dispatch_token)
 
         notifications: List[Dict[str, Any]] = []
         if decision.commit:
@@ -269,6 +294,10 @@ def transition(job_id, request="merge", *, registry_dir=None, expected_status=No
             # A monotonic counter so a late flush can still be ordered correctly
             # once the audit mirror happens outside the lock.
             record["transition_seq"] = int(record.get("transition_seq", 0) or 0) + 1
+            if request in ATTEMPT_STARTING_REQUESTS:
+                # A new attempt begins here, so anything decided about the
+                # previous one stops being committable from this moment on.
+                record[DISPATCH_TOKEN_FIELD] = uuid.uuid4().hex
             if notify is not None:
                 # One atomic commit records both the new state and the fact that
                 # the world has not been told yet. If the process dies before the

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

@@ -100,6 +100,17 @@ class _Watcher:
         self.expected = set(expected_job_ids)
         self.tokens = expected_tokens  # job_id -> expected auth_token (or None)
         self.last_seq = dict(expected_seqs)
+        # Intent identities already accepted, per job. A replayed notification is
+        # re-published with a *fresh* wire sequence, so the monotonic sequence
+        # guard below cannot recognise it: to that guard a retry looks like a new
+        # event. The intent identity is what stays stable across a retry, so
+        # deduplication has to happen on that.
+        #
+        # This is per-process state. Across a restart the disk fallback resolves
+        # terminal status from the registry, and a duplicate terminal is refused
+        # by the transition boundary as an idempotent ack, so a replay after
+        # restart cannot double-apply either.
+        self.seen_intents = {}
 
     def on_message(self, _client, _userdata, msg) -> None:
         # --- defensive parsing -------------------------------------------
@@ -129,6 +140,15 @@ class _Watcher:
         if seq is None or not isinstance(seq, int):
             logger.warning("drop event for job %s: missing or invalid seq", jid)
             return
+        intent = (payload.get("data") or {}).get("notification_intent")
+        if isinstance(intent, dict):
+            identity = (intent.get("seq"), intent.get("event"), intent.get("timestamp"))
+            if identity in self.seen_intents.setdefault(jid, set()):
+                logger.info("drop event for job %s: notification intent %s already accepted",
+                            jid, identity)
+                return
+            self.seen_intents[jid].add(identity)
+
         if seq <= self.last_seq.get(jid, 0):
             logger.warning("drop event for job %s: seq %d is not monotonically increasing (last %d)",
                            jid, seq, self.last_seq.get(jid, 0))

.agents/skills/lib_py/loop/autocomplete.py

@@ -181,6 +181,21 @@ def evaluate(job_id, scenario, registry_dir=None, workspace=None,
     if mismatch:
         return GateResult(False, "hold", mismatch)
 
+    # The evidence must belong to the attempt that is currently running, and the
+    # commit must be refused if that stops being true while this runs. Comparing
+    # here alone is not enough: the job can complete, be re-dispatched and
+    # claimed again before the terminal write, after which a status check alone
+    # would let stale evidence finish a new attempt.
+    from lib_py.delegation import engine as _engine
+
+    token = record.get(_engine.DISPATCH_TOKEN_FIELD)
+    if not token:
+        return GateResult(False, "hold",
+                          "job has no dispatch token to bind this decision to")
+    if dispatch.get("dispatch_token") != token:
+        return GateResult(False, "hold",
+                          "evidence belongs to a superseded dispatch of this job")
+
     verdict = finalize.verify_manifest(
         report_dir, job_id, expected_attempt=dispatch.get("attempt_id"),
         expected_artifacts=record.get("expected_artifacts") or [])
@@ -206,7 +221,7 @@ def evaluate(job_id, scenario, registry_dir=None, workspace=None,
     # carries a loop-iteration counter, which is a different thing.
     result = engine.transition(
         job_id, "completed", registry_dir=registry_dir,
-        expected_status="running", source=SOURCE,
+        expected_status="running", expected_dispatch_token=token, source=SOURCE,
         fields={"completed_by": SOURCE, "auto_complete_reason": decision.reason},
         notify="completed",
     )

.agents/skills/lib_py/loop/store.py

@@ -50,7 +50,7 @@ def _atomic_write(path: Path, text: str) -> None:
 
 
 def record_dispatch(job_id, attempt_id, agent_session_uuid, pane_process_generation,
-                    dispatch_ack=True, workspace=None) -> Dict[str, Any]:
+                    dispatch_ack=True, workspace=None, dispatch_token=None) -> Dict[str, Any]:
     """Persist the four identifiers at dispatch time.
 
     The execution anchor starts false on purpose: dispatch being acknowledged is
@@ -63,6 +63,10 @@ def record_dispatch(job_id, attempt_id, agent_session_uuid, pane_process_generat
         "pane_process_generation": pane_process_generation,
         "dispatch_ack": bool(dispatch_ack),
         "execution_anchor_observed": False,
+        # The record's token for the attempt this evidence belongs to. Carrying
+        # it here is what lets the gate prove the evidence and the commit are
+        # about the same attempt.
+        "dispatch_token": dispatch_token,
     }
     _atomic_write(attribution_path(job_id, workspace),
                   json.dumps(context, ensure_ascii=False, indent=2) + "\n")

tests/test_issue5_phase2b_autocomplete.py

@@ -304,8 +304,13 @@ def _ready_job(tmp_path, mam_sandbox, *, backend="claude-code", opt_in=True,
         payload = item.get("payload") or {}
         if item.get("source_kind") != "herdr" and "job_id" in payload:
             payload["job_id"] = job_id
+    # The dispatch record carries the record's own token for this attempt, which
+    # is what ties the evidence to the attempt that is currently running.
+    from lib_py.delegation import engine as _engine
+    token = _load(reg, job_id).get(_engine.DISPATCH_TOKEN_FIELD)
     store.record_dispatch(job_id, dispatch["attempt_id"], dispatch["agent_session_uuid"],
-                          dispatch["pane_process_generation"], workspace=str(mam_sandbox))
+                          dispatch["pane_process_generation"], workspace=str(mam_sandbox),
+                          dispatch_token=token)
     store.mark_execution_anchor(job_id, workspace=str(mam_sandbox))
 
     report = _report(tmp_path, body)
@@ -782,3 +787,170 @@ def test_o2_the_manifest_records_an_absolute_report_path(tmp_path, monkeypatch):
     assert Path(manifest["report_path"]).is_absolute()
     monkeypatch.chdir(REPO_ROOT)
     assert finalize.verify_manifest(path.parent, "job1").ok, "verification survives a chdir"
+
+
+# ==============================================================================
+# 7. Cross-review round 2 regressions (job 4cfc1649)
+#
+# R2 (codex): attribution was checked before the terminal commit but nothing
+#             bound the two, so a job could complete, be re-dispatched and
+#             claimed again in between, and stale evidence finished the new
+#             attempt.
+# R4 (codex): the intent identity was emitted but the real subscriber ignored
+#             it, so a replay with a fresh wire sequence was accepted twice.
+# ==============================================================================
+
+def test_r2_a_redispatch_between_the_check_and_the_commit_is_refused(
+    mam_sandbox, tmp_path, monkeypatch
+):
+    """The exact interleaving: complete, re-dispatch, claim again, then let the
+    original commit proceed. Evidence for attempt 1 must not finish attempt 2."""
+    monkeypatch.chdir(mam_sandbox)
+    monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
+    from lib_py.delegation import engine
+    from lib_py.loop import autocomplete
+
+    job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox)
+
+    real_transition = engine.transition
+    fired = {}
+
+    def racing(job, request="merge", **kwargs):
+        if request == "completed" and kwargs.get("source") == "auto_complete" and not fired:
+            fired["yes"] = True
+            engine.transition = real_transition
+            real_transition(job, "completed", registry_dir=reg)
+            real_transition(job, "redispatch", registry_dir=reg,
+                            source="registry_cli", fields={"iteration": 2})
+            real_transition(job, "claim", registry_dir=reg)
+            engine.transition = racing
+        return real_transition(job, request, **kwargs)
+
+    monkeypatch.setattr(engine, "transition", racing)
+    result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
+                                   workspace=str(mam_sandbox), report_dir=report_dir)
+    monkeypatch.setattr(engine, "transition", real_transition)
+
+    record = _load(reg, job_id)
+    assert result.committed is False and "dispatch token" in result.reason
+    assert record["status"] == "running", "the new attempt must still be running"
+    assert record["iteration"] == 2
+    assert "completed_by" not in record
+
+
+def test_r2_evidence_from_a_superseded_dispatch_is_refused(mam_sandbox, tmp_path, monkeypatch):
+    """Even without a race: evidence recorded against an older token is stale."""
+    monkeypatch.chdir(mam_sandbox)
+    monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
+    from lib_py.delegation import engine
+    from lib_py.loop import autocomplete
+
+    job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox)
+    # a full re-dispatch mints a new token; the recorded evidence keeps the old
+    engine.transition(job_id, "completed", registry_dir=reg)
+    engine.transition(job_id, "redispatch", registry_dir=reg,
+                      source="registry_cli", fields={"iteration": 2})
+    engine.transition(job_id, "claim", registry_dir=reg)
+
+    result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
+                                   workspace=str(mam_sandbox), report_dir=report_dir)
+    assert result.committed is False and "superseded dispatch" in result.reason
+
+
+def test_r2_a_job_without_a_dispatch_token_is_refused(mam_sandbox, tmp_path, monkeypatch):
+    """Nothing to bind the decision to is a refusal, not a pass."""
+    monkeypatch.chdir(mam_sandbox)
+    monkeypatch.setenv("MAM_AUTO_COMPLETE", "1")
+    from lib_py.delegation import engine, mqtt_common
+    from lib_py.loop import autocomplete
+
+    job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox)
+    record = _load(reg, job_id)
+    record.pop(engine.DISPATCH_TOKEN_FIELD, None)
+    mqtt_common._atomic_write_record(job_id, reg, record)
+
+    result = autocomplete.evaluate(job_id, scenario, registry_dir=reg,
+                                   workspace=str(mam_sandbox), report_dir=report_dir)
+    assert result.committed is False and "no dispatch token" in result.reason
+
+
+def test_r2_a_new_attempt_mints_a_new_token(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)
+    first = _load(reg, job_id)[engine.DISPATCH_TOKEN_FIELD]
+    engine.transition(job_id, "progress", registry_dir=reg, fields={"note": "x"})
+    assert _load(reg, job_id)[engine.DISPATCH_TOKEN_FIELD] == first, (
+        "an ordinary transition must not mint a new attempt"
+    )
+    engine.transition(job_id, "completed", registry_dir=reg)
+    engine.transition(job_id, "redispatch", registry_dir=reg,
+                      source="registry_cli", fields={"iteration": 2})
+    assert _load(reg, job_id)[engine.DISPATCH_TOKEN_FIELD] != first
+
+
+def test_r2_the_token_precondition_is_enforced_by_the_engine(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)
+    before = _fingerprint(Path(reg))
+
+    stale = engine.transition(job_id, "completed", registry_dir=reg,
+                              expected_dispatch_token="not-the-current-token")
+    assert stale.committed is False and "dispatch token changed" in stale.reason
+    assert _fingerprint(Path(reg)) == before, "a refused precondition must not write"
+
+    token = _load(reg, job_id)[engine.DISPATCH_TOKEN_FIELD]
+    assert engine.transition(job_id, "completed", registry_dir=reg,
+                             expected_dispatch_token=token).committed
+
+
+def test_r4_the_subscriber_drops_a_replayed_notification_intent():
+    """A replay is re-published with a fresh wire sequence, so the monotonic
+    guard cannot see it; the stable identity is what must be deduplicated."""
+    import types
+    from lib_py.delegation import job_subscriber
+
+    watcher = job_subscriber._Watcher({"j1"}, {"j1": None}, {"j1": 0})
+
+    def deliver(wire_seq, intent_seq):
+        payload = {"schema_version": 1, "seq": wire_seq, "job_id": "j1",
+                   "event": "completed", "timestamp": "t", "detail": "d",
+                   "data": {"notification_intent": {"seq": intent_seq,
+                                                    "event": "completed",
+                                                    "timestamp": "T"}}}
+        watcher.on_message(None, None, types.SimpleNamespace(
+            topic="python/mqtt/jobs/j1/events",
+            payload=json.dumps(payload).encode()))
+
+    deliver(1, 7)
+    deliver(2, 7)          # the same intent, retried
+    assert watcher.events.qsize() == 1, "a replayed intent must not be acted on twice"
+    deliver(3, 8)          # a genuinely different notification
+    assert watcher.events.qsize() == 2
+
+
+def test_r4_events_without_an_intent_are_unaffected():
+    """The existing sequence guard remains the rule for ordinary events."""
+    import types
+    from lib_py.delegation import job_subscriber
+
+    watcher = job_subscriber._Watcher({"j1"}, {"j1": None}, {"j1": 0})
+
+    def deliver(wire_seq):
+        payload = {"schema_version": 1, "seq": wire_seq, "job_id": "j1",
+                   "event": "progress", "timestamp": "t", "detail": "d", "data": {}}
+        watcher.on_message(None, None, types.SimpleNamespace(
+            topic="python/mqtt/jobs/j1/events",
+            payload=json.dumps(payload).encode()))
+
+    deliver(1)
+    deliver(2)
+    assert watcher.events.qsize() == 2
+    deliver(2)             # non-increasing: still dropped by the sequence guard
+    assert watcher.events.qsize() == 2