diff --git a/.agents/skills/lib_py/delegation/engine.py b/.agents/skills/lib_py/delegation/engine.py index 813533f..e2c738c 100644 --- a/.agents/skills/lib_py/delegation/engine.py +++ b/.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 diff --git a/.agents/skills/lib_py/delegation/job_subscriber.py b/.agents/skills/lib_py/delegation/job_subscriber.py index 6fa4784..c871cb0 100755 --- a/.agents/skills/lib_py/delegation/job_subscriber.py +++ b/.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)) diff --git a/.agents/skills/lib_py/loop/autocomplete.py b/.agents/skills/lib_py/loop/autocomplete.py index 54964b5..f6f805a 100644 --- a/.agents/skills/lib_py/loop/autocomplete.py +++ b/.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", ) diff --git a/.agents/skills/lib_py/loop/store.py b/.agents/skills/lib_py/loop/store.py index 3319f8e..24717b8 100644 --- a/.agents/skills/lib_py/loop/store.py +++ b/.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") diff --git a/tests/test_issue5_phase2b_autocomplete.py b/tests/test_issue5_phase2b_autocomplete.py index 8f36ea6..be26533 100644 --- a/tests/test_issue5_phase2b_autocomplete.py +++ b/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