diff --git a/.agents/skills/lib_py/delegation/engine.py b/.agents/skills/lib_py/delegation/engine.py index 9952f60..a1a9d6e 100644 --- a/.agents/skills/lib_py/delegation/engine.py +++ b/.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: diff --git a/.agents/skills/lib_py/delegation/mqtt_common.py b/.agents/skills/lib_py/delegation/mqtt_common.py index 61830e9..6996eff 100644 --- a/.agents/skills/lib_py/delegation/mqtt_common.py +++ b/.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) diff --git a/.agents/skills/lib_py/delegation/registry.py b/.agents/skills/lib_py/delegation/registry.py index cb69f8e..2656263 100755 --- a/.agents/skills/lib_py/delegation/registry.py +++ b/.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: diff --git a/.agents/skills/lib_py/loop/policy.py b/.agents/skills/lib_py/loop/policy.py index 62bbe98..005ae65 100644 --- a/.agents/skills/lib_py/loop/policy.py +++ b/.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", diff --git a/tests/fixtures/issue_5/watcher/finalized_candidate.json b/tests/fixtures/issue_5/watcher/finalized_candidate.json index d8e5340..08a8a84 100644 --- a/tests/fixtures/issue_5/watcher/finalized_candidate.json +++ b/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" } } ], diff --git a/tests/test_issue5_phase2a_loop.py b/tests/test_issue5_phase2a_loop.py index 7da8dca..bd5164d 100644 --- a/tests/test_issue5_phase2a_loop.py +++ b/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)