fix(issue-5): address phase 2a round 2 findings (F-C, F-D, R1, R4)

Both reviewers sustained the round 1 fixes and found four more. Each was
reproduced here before being accepted; none were rebutted.

F-C (claude, blocking) -- my own round 1 fix caused this. Making refused writes
exit non-zero also classified an idempotent ACK as a refusal, and the wrapper
finalises an approved job with `status --set completed` on a record that is
already completed. It runs under `set -e`, so the *success* path aborted with
exit 4 before its validation hook and its final stdout line. Reproduced at the
shell. An idempotent result is now distinguishable from a refusal and exits 0; a
genuine conflict still exits 4.

F-D (claude) -- routing the operator write around update_status dropped the
status vocabulary check. Under the rollback flag an invalid status was written
to the record with exit 0, where before it was refused with exit 1 and the record
left alone. The check now runs first, in both modes.

R1 (codex, blocking) -- the previous fix compared identifiers only when they
happened to be present, so evidence carrying none at all was still approved on
the dispatch record's own flags. That is the self-assertion the check exists to
replace. Every identifier must now be supplied and agree, and a report manifest
must associate with the attempt being judged. Identifiers are read from a
collector envelope beside the payload rather than from inside it: a native Herdr
payload may not carry MAM fields, so one appearing there is either fabricated or
a contract violation, and either way it is not evidence.

R4 (codex) -- a sequence number orders nothing unless the compare and the write
are atomic. Two flushes could both read the old value, the newer land first, and
the slower overwrite it. Reproduced deterministically. The mirror now serialises
read-compare-write under a per-job audit lock, taken outside the registry
transaction so a slow mirror still cannot lengthen it, and each write uses its
own temp file instead of a shared name.

All eight new regressions fail against the pre-fix code.

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

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

@@ -81,10 +81,16 @@ def cas_enabled() -> bool:
 
 
 class TransitionResult(object):
-    __slots__ = ("committed", "from_status", "to_status", "reason", "record", "notifications")
+    __slots__ = ("committed", "from_status", "to_status", "reason", "record",
+                 "notifications", "idempotent")
 
-    def __init__(self, committed, from_status, to_status, reason, record, notifications):
+    def __init__(self, committed, from_status, to_status, reason, record, notifications,
+                 idempotent=False):
         self.committed = committed
+        # Not every uncommitted transition is a refusal. Re-receiving a terminal
+        # that already holds is a no-op ACK in the contract table, and a caller
+        # that treats it as failure turns a success path into an error.
+        self.idempotent = idempotent
         self.from_status = from_status
         self.to_status = to_status
         self.reason = reason
@@ -97,13 +103,14 @@ class TransitionResult(object):
 
 
 class _Decision(object):
-    __slots__ = ("commit", "to_status", "reason", "conflict")
+    __slots__ = ("commit", "to_status", "reason", "conflict", "idempotent")
 
-    def __init__(self, commit, to_status, reason, conflict=False):
+    def __init__(self, commit, to_status, reason, conflict=False, idempotent=False):
         self.commit = commit
         self.to_status = to_status
         self.reason = reason
         self.conflict = conflict
+        self.idempotent = idempotent
 
 
 def _decide(record: Dict[str, Any], request: str, expected_status, expected_attempt,
@@ -156,7 +163,8 @@ def _decide(record: Dict[str, Any], request: str, expected_status, expected_atte
     if current in TERMINAL_STATUSES:
         # Idempotent re-receipt of the same terminal is a no-op ack.
         if requested == current:
-            return _Decision(False, current, "idempotent: already {}".format(current))
+            return _Decision(False, current, "idempotent: already {}".format(current),
+                             idempotent=True)
 
         operator = source in OPERATOR_SOURCES
         if operator and request == "redispatch":
@@ -249,7 +257,8 @@ def transition(job_id, request="merge", *, registry_dir=None, expected_status=No
         logger.warning("refused %s for job %s: %s", request, job_id, decision.reason)
 
     return TransitionResult(decision.commit, from_status, to_status,
-                            decision.reason, record, notifications)
+                            decision.reason, record, notifications,
+                            idempotent=decision.idempotent)
 
 
 def _flush(notifications, registry_dir) -> None:

.agents/skills/lib_py/delegation/mqtt_common.py

@@ -577,25 +577,44 @@ def update_logged_status(job_id: str, status: str, logs_dir: Optional[str] = Non
         path = job_log_path(job_id, STATUS_FILENAME, logs_dir)
         path.parent.mkdir(parents=True, exist_ok=True)
 
-        incoming_seq = extras.get("transition_seq")
-        if incoming_seq is not None and path.exists():
-            try:
-                existing = json.loads(path.read_text(encoding="utf-8"))
-                existing_seq = existing.get("transition_seq")
-                if existing_seq is not None and int(existing_seq) >= int(incoming_seq):
-                    logger.debug("skipping stale status mirror for %s (seq %s <= %s)",
-                                 job_id, incoming_seq, existing_seq)
-                    return
-            except Exception:
-                pass  # unreadable existing file: fall through and rewrite it
+        # Compare and write under one per-job lock. A sequence number only orders
+        # writers that cannot interleave: with a bare check-then-write, two
+        # flushes can both read the old value, the newer one lands first, and the
+        # slower one then overwrites it. The lock is deliberately its own file so
+        # it is never confused with the registry lock, and it is taken here --
+        # outside the registry transaction -- so a slow mirror still cannot
+        # lengthen that transaction.
+        lock_path = path.with_name("." + path.name + ".lock")
+        with open(lock_path, "a+") as lock_fh:
+            with _file_lock(lock_fh):
+                incoming_seq = extras.get("transition_seq")
+                if incoming_seq is not None and path.exists():
+                    try:
+                        existing = json.loads(path.read_text(encoding="utf-8"))
+                        existing_seq = existing.get("transition_seq")
+                        if existing_seq is not None and int(existing_seq) >= int(incoming_seq):
+                            logger.debug("skipping stale status mirror for %s (seq %s <= %s)",
+                                         job_id, incoming_seq, existing_seq)
+                            return
+                    except Exception:
+                        pass  # unreadable existing file: fall through and rewrite it
 
-        record: Dict[str, Any] = {"job_id": job_id, "status": status, "updated_at": _utcnow()}
-        record.update(extras)
-        tmp = path.with_name(path.name + ".tmp")
-        with open(tmp, "w", encoding="utf-8") as fh:
-            json.dump(record, fh, ensure_ascii=False, indent=2)
-            fh.write("\n")
-        os.replace(tmp, path)
+                record: Dict[str, Any] = {"job_id": job_id, "status": status,
+                                          "updated_at": _utcnow()}
+                record.update(extras)
+                # A per-write temp name: a shared one lets concurrent writers
+                # clobber each other's half-written file before the rename.
+                fd, tmp = tempfile.mkstemp(dir=str(path.parent),
+                                           prefix="." + path.name + ".", suffix=".tmp")
+                try:
+                    with os.fdopen(fd, "w", encoding="utf-8") as fh:
+                        json.dump(record, fh, ensure_ascii=False, indent=2)
+                        fh.write("\n")
+                    os.replace(tmp, path)
+                except BaseException:
+                    if os.path.exists(tmp):
+                        os.unlink(tmp)
+                    raise
     except Exception as exc:  # pragma: no cover - best effort
         logger.warning("update_logged_status failed for job %s: %s", job_id, exc)
 

.agents/skills/lib_py/delegation/registry.py

@@ -327,23 +327,43 @@ def _build_parser() -> argparse.ArgumentParser:
 
 
 def _operator_write(job_id: str, registry_dir: str, **fields) -> int:
-    """Apply an operator-authority write and report refusal on the exit status.
+    """Apply an operator-authority write and report a refusal on the exit status.
 
     Before phase 2a this path could not be refused, so it always exited 0. Once
     the CAS boundary can decline a write, exiting 0 anyway would hide it -- which
-    is exactly how a refused loop re-dispatch looked like success to the wrapper.
+    is how a refused loop re-dispatch looked like success to the wrapper.
+
+    Two things this must NOT do, both learned the hard way:
+
+      * treat an idempotent ACK as a refusal. The wrapper finalises an approved
+        job with `status --set completed` on a record that is already completed,
+        and it runs under `set -e`, so reporting failure there aborts the success
+        path before the validation hook and the final stdout line.
+      * skip the status vocabulary check. That validation used to live in
+        `update_status`; routing around it let an invalid status through, and in
+        the rollback mode it was actually written to the record.
     """
     from lib_py.delegation import engine  # same package: no sibling hop (see __init__)
 
+    status = fields.get("status")
+    if status is not None and status not in VALID_STATUSES:
+        print("invalid status {!r}; expected one of {}".format(status, VALID_STATUSES),
+              file=sys.stderr)
+        return 1
+
     if not engine.cas_enabled():
         mqtt_common.update_job_status(job_id, registry_dir, **fields)
         return 0
+
     result = mqtt_common.transition_job_status(
         job_id, registry_dir, _source="registry_cli", **fields)
-    if not result.committed:
-        print("refused: {}".format(result.reason), file=sys.stderr)
-        return 4
-    return 0
+    if result.committed:
+        return 0
+    if result.idempotent:
+        # The requested state already holds. Nothing to do, and nothing wrong.
+        return 0
+    print("refused: {}".format(result.reason), file=sys.stderr)
+    return 4
 
 
 def main(argv: Optional[List[str]] = None) -> int:

.agents/skills/lib_py/loop/policy.py

@@ -17,6 +17,11 @@ from lib_py.loop.evidence import Decision
 # nothing about completion; it only makes completion *possible* to consider.
 QUIESCENT = ("idle", "done")
 
+# The four identifiers an external collector must supply before a completion can
+# be attributed to this attempt.
+ATTRIBUTION_FIELDS = ("job_id", "attempt_id", "agent_session_uuid",
+                      "pane_process_generation")
+
 
 def _timeline_by_kind(scenario):
     out = {}
@@ -28,22 +33,32 @@ def _timeline_by_kind(scenario):
 def collect_observed_identity(scenario) -> dict:
     """Gather identifiers the *evidence* carries, for comparison with dispatch.
 
-    Without this, attribution only proves the dispatch record is internally
-    well-formed -- it never confronts what was actually seen. Association of a
-    job or attempt to an observation is the external collector's business, so
-    these are read from the timeline items and the report manifest, never
-    invented as native Herdr fields.
+    Two sources, and the distinction matters. Association of a job or attempt to
+    an observation is the external collector's business, so it lives in a
+    ``collector_attribution`` envelope beside the payload -- never inside it,
+    because a native Herdr payload must stay native (phase 0 forbids MAM fields
+    there, and a fabricated one would be indistinguishable from a real reading).
+    The report manifest is a filesystem fact and stays in that payload.
+
+    Every occurrence is kept. Taking only the first would drop a later,
+    contradicting observation, which is precisely what this exists to catch.
     """
     observed = {}
     for item in scenario.get("timeline") or []:
+        envelope = item.get("collector_attribution") or {}
+        for field in ATTRIBUTION_FIELDS:
+            if envelope.get(field) is not None:
+                observed.setdefault(field, []).append(envelope[field])
         payload = item.get("payload") or {}
-        for field in ("job_id", "attempt_id", "agent_session_uuid",
-                      "pane_process_generation"):
-            if payload.get(field) is not None:
-                # Every occurrence is kept. Taking only the first would let a
-                # later, contradicting observation pass unnoticed -- which is the
-                # opposite of what this collection is for.
-                observed.setdefault(field, []).append(payload[field])
+        if item.get("source_kind") != "herdr":
+            # A MAM-side source (an event, a filesystem fact) legitimately names
+            # the job or attempt in its own payload. A native Herdr payload never
+            # may: phase 0 forbids MAM fields there, so an identifier appearing
+            # inside one is either fabricated or a contract violation, and either
+            # way it is not evidence.
+            for field in ATTRIBUTION_FIELDS:
+                if payload.get(field) is not None:
+                    observed.setdefault(field, []).append(payload[field])
         manifest = payload.get("report_manifest")
         if isinstance(manifest, dict) and manifest.get("attempt_id") is not None:
             observed.setdefault("manifest_attempt_id", []).append(manifest["attempt_id"])
@@ -97,15 +112,24 @@ def decide(scenario) -> Decision:
     if not attributed.ok:
         return Decision("hold", attributed.reason)
 
-    for field, values in observed.items():
-        for value in values:
+    # Absence of proof is not proof. Comparing only the identifiers that happen
+    # to be present would let evidence carrying none at all be approved on the
+    # dispatch record's own say-so, which is the self-assertion this check exists
+    # to replace. Every identifier must be supplied, then must agree.
+    for field in ATTRIBUTION_FIELDS:
+        if not observed.get(field):
+            return Decision("hold",
+                            "no observed {} to attribute this completion to".format(field))
+        for value in observed[field]:
             if value != dispatch.get(field):
                 return Decision("hold", "attribution mismatch on {}: observed {!r}, "
                                         "dispatched {!r}".format(field, value,
                                                                  dispatch.get(field)))
 
-    # The finalized report must belong to the attempt being judged; a manifest
-    # from another attempt is evidence of a different run finishing.
+    # The finalized report must belong to the attempt being judged. A missing
+    # manifest attempt holds for the same reason: it proves nothing.
+    if not manifest_attempts:
+        return Decision("hold", "no report manifest attempt to associate with this attempt")
     for manifest_attempt in manifest_attempts:
         if manifest_attempt != dispatch.get("attempt_id"):
             return Decision("hold",

tests/fixtures/issue_5/watcher/finalized_candidate.json

@@ -3,7 +3,7 @@
   "scenario_id": "finalized_candidate",
   "backend": "backend_agnostic",
   "provenance": "simulated",
-  "description": "Execution anchor verified, native status idle/done, atomic rename complete with valid SHA-256 commit manifest. Inferred candidate approved for CAS.",
+  "description": "Execution anchor verified, native status idle/done, atomic rename complete with valid SHA-256 commit manifest. Inferred candidate approved for CAS. The collector supplies the four identifiers alongside the native snapshot, which is what allows the completion to be attributed.",
   "dispatch_context": {
     "job_id": "test-job-finalized-01",
     "attempt_id": 1,
@@ -44,6 +44,12 @@
       "payload": {
         "agent_status": "idle",
         "interactive_ready": true
+      },
+      "collector_attribution": {
+        "job_id": "test-job-finalized-01",
+        "attempt_id": 1,
+        "agent_session_uuid": "7d5700d0-bab1-4f65-9665-6b887396a9a6",
+        "pane_process_generation": "w2X:p1:1234:2026-09-06T00:00:00Z"
       }
     }
   ],

tests/test_issue5_phase2a_loop.py

@@ -817,7 +817,9 @@ def test_r1_policy_rejects_evidence_that_contradicts_the_dispatch(field, value):
     scenario = json.loads((WATCHER_FIXTURES / "finalized_candidate.json").read_text(encoding="utf-8"))
     assert policy.decide(scenario).decision == "candidate_approved"
     mutated = copy.deepcopy(scenario)
-    mutated["timeline"][2]["payload"][field] = value
+    # The collector envelope is where an identifier legitimately lives; a native
+    # payload is not an identifier source at all.
+    mutated["timeline"][2]["collector_attribution"][field] = value
     decision = policy.decide(mutated)
     assert decision.decision == "hold", "{}={!r} must not be approved".format(field, value)
     assert field in decision.reason
@@ -887,3 +889,173 @@ def test_r3_an_unknown_request_is_refused(mam_sandbox, tmp_path, monkeypatch):
     res = engine.transition(job_id, "not_a_request", registry_dir=reg, fields={"x": 1})
     assert res.committed is False and "unknown request" in res.reason
     assert "x" not in _load(reg, job_id)
+
+
+# ==============================================================================
+# 9. Cross-review round 2 regressions (job 57166ae2)
+#
+# F-C (claude): an idempotent ACK was reported as a refusal, so the wrapper's
+#               approved path aborted under set -e before its validation hook.
+# F-D (claude): routing around update_status lost the status vocabulary check.
+# R1  (codex):  absence of evidence still approved; only present identifiers
+#               were compared.
+# R4  (codex):  the mirror's ordering guard was a check followed by a write with
+#               no lock between them.
+# ==============================================================================
+
+@pytest.mark.parametrize("mode", CAS_MODES)
+def test_fc_idempotent_finalize_is_success_not_refusal(mam_sandbox, tmp_path, monkeypatch, mode):
+    """F-C: the wrapper finalises an approved job that is already completed.
+
+    It runs under `set -e`, so reporting failure there aborts the success path
+    before the validation hook and the final stdout line."""
+    monkeypatch.setenv("MAM_ENGINE_CAS", mode)
+    monkeypatch.chdir(mam_sandbox)
+    from lib_py.delegation import mqtt_common, registry
+
+    reg = _reg(tmp_path)
+    job_id = _make_job(reg)
+    mqtt_common.update_job_status(job_id, reg, _source="publish_event", status="completed")
+
+    rc = registry.main(["--registry-dir", reg, "status", "--job", job_id, "--set", "completed"])
+    assert rc == 0, "re-affirming a state that already holds is an ack, not a failure"
+    assert _load(reg, job_id)["status"] == "completed"
+
+
+def test_fc_a_genuine_conflict_is_still_reported(mam_sandbox, tmp_path, monkeypatch):
+    """The F-C fix must not soften a real refusal back into silence."""
+    monkeypatch.chdir(mam_sandbox)
+    from lib_py.delegation import mqtt_common, registry
+    reg = _reg(tmp_path)
+    job_id = _make_job(reg)
+    mqtt_common.update_job_status(job_id, reg, _source="publish_event", status="completed")
+    rc = registry.main(["--registry-dir", reg, "status", "--job", job_id, "--set", "cancelled"])
+    assert rc == 4
+    assert _load(reg, job_id)["status"] == "completed"
+
+
+def test_fc_engine_marks_an_idempotent_result(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)
+    engine.transition(job_id, "completed", registry_dir=reg)
+    again = engine.transition(job_id, "completed", registry_dir=reg)
+    assert again.committed is False and again.idempotent is True
+    conflict = engine.transition(job_id, "cancelled", registry_dir=reg)
+    assert conflict.committed is False and conflict.idempotent is False
+
+
+@pytest.mark.parametrize("mode", CAS_MODES)
+def test_fd_invalid_status_is_rejected_in_both_modes(mam_sandbox, tmp_path, monkeypatch, mode):
+    """F-D: the vocabulary check must survive the routing, especially on the
+    rollback path where the engine is not there to catch it."""
+    monkeypatch.setenv("MAM_ENGINE_CAS", mode)
+    monkeypatch.chdir(mam_sandbox)
+    from lib_py.delegation import registry
+
+    reg = _reg(tmp_path)
+    job_id = _make_job(reg)
+    before = _dir_fingerprint(Path(reg))
+    rc = registry.main(["--registry-dir", reg, "status", "--job", job_id, "--set", "bogus"])
+    assert rc == 1, "an invalid status is a usage error, not a CAS refusal"
+    assert _dir_fingerprint(Path(reg)) == before, "an invalid status must not be written"
+
+
+@pytest.mark.parametrize("field", [
+    "job_id", "attempt_id", "agent_session_uuid", "pane_process_generation",
+])
+def test_r1_missing_attribution_evidence_holds(field):
+    """R1: absence of proof is not proof. Every identifier must be supplied."""
+    import copy
+    from lib_py.loop import policy
+    scenario = json.loads((WATCHER_FIXTURES / "finalized_candidate.json").read_text(encoding="utf-8"))
+    assert policy.decide(scenario).decision == "candidate_approved"
+
+    mutated = copy.deepcopy(scenario)
+    for item in mutated["timeline"]:
+        (item.get("collector_attribution") or {}).pop(field, None)
+        (item.get("payload") or {}).pop(field, None)
+    decision = policy.decide(mutated)
+    assert decision.decision == "hold", "missing {} must not be approved".format(field)
+    assert field in decision.reason
+
+
+def test_r1_missing_report_manifest_holds():
+    import copy
+    from lib_py.loop import policy
+    scenario = json.loads((WATCHER_FIXTURES / "finalized_candidate.json").read_text(encoding="utf-8"))
+    mutated = copy.deepcopy(scenario)
+    for item in mutated["timeline"]:
+        (item.get("payload") or {}).pop("report_manifest", None)
+    decision = policy.decide(mutated)
+    assert decision.decision == "hold" and "manifest" in decision.reason
+
+
+def test_r1_identifiers_inside_a_native_payload_are_not_evidence():
+    """A MAM field cannot appear in a native Herdr payload, so one found there is
+    either fabricated or a contract violation. Either way it proves nothing."""
+    import copy
+    from lib_py.loop import policy
+    scenario = json.loads((WATCHER_FIXTURES / "finalized_candidate.json").read_text(encoding="utf-8"))
+    mutated = copy.deepcopy(scenario)
+    dispatch = mutated["dispatch_context"]
+    herdr_item = [i for i in mutated["timeline"] if i["source_kind"] == "herdr"][-1]
+    herdr_item.pop("collector_attribution", None)
+    for field in ("job_id", "attempt_id", "agent_session_uuid"):
+        herdr_item["payload"][field] = dispatch[field]
+    assert policy.decide(mutated).decision != "candidate_approved"
+
+
+def test_r4_overlapping_mirror_flushes_cannot_lose_the_newer_state(mam_sandbox, monkeypatch):
+    """R4: a sequence number orders nothing unless compare and write are atomic.
+
+    Interrupts a flush after it has read, lets a newer one complete, and requires
+    the newer state to survive."""
+    import threading
+    from lib_py.delegation import mqtt_common
+
+    monkeypatch.chdir(mam_sandbox)
+    logs = mqtt_common.get_logs_dir()
+    job_id = "r4probe"
+    mqtt_common.update_logged_status(job_id, "pending", logs, transition_seq=0)
+    path = mqtt_common.job_log_path(job_id, mqtt_common.STATUS_FILENAME, logs)
+
+    real_read = Path.read_text
+    state = {}
+
+    def intercept(self, *a, **k):
+        text = real_read(self, *a, **k)
+        if str(self) == str(path) and "fired" not in state:
+            state["fired"] = True
+            thread = threading.Thread(
+                target=mqtt_common.update_logged_status,
+                args=(job_id, "completed", logs), kwargs={"transition_seq": 2})
+            thread.start()
+            state["thread"] = thread
+            thread.join(timeout=0.5)
+            state["blocked"] = thread.is_alive()
+        return text
+
+    monkeypatch.setattr(Path, "read_text", intercept)
+    mqtt_common.update_logged_status(job_id, "running", logs, transition_seq=1)
+    monkeypatch.undo()
+    state["thread"].join(timeout=5)
+
+    assert state.get("blocked") is True, "the competing flush must serialise, not race"
+    final = json.loads(path.read_text(encoding="utf-8"))
+    assert final["status"] == "completed" and final["transition_seq"] == 2
+
+
+def test_r4_each_mirror_write_uses_its_own_temp_file(mam_sandbox, monkeypatch):
+    """A shared temp name lets concurrent writers clobber each other pre-rename."""
+    monkeypatch.chdir(mam_sandbox)
+    from lib_py.delegation import mqtt_common
+    source = Path(SKILLS / "lib_py" / "delegation" / "mqtt_common.py").read_text(encoding="utf-8")
+    body = source[source.index("def update_logged_status"):source.index("def init_job_log")]
+    assert "mkstemp" in body, "the mirror must not reuse one fixed .tmp path"
+    logs = mqtt_common.get_logs_dir()
+    mqtt_common.update_logged_status("tmpprobe", "running", logs, transition_seq=1)
+    leftovers = list((Path(logs) / "tmpprobe").glob("*.tmp"))
+    assert not leftovers, "temp files must be renamed away, not left behind: {}".format(leftovers)