diff --git a/.agents/skills/lib_py/delegation/engine.py b/.agents/skills/lib_py/delegation/engine.py index 78603ab..813533f 100644 --- a/.agents/skills/lib_py/delegation/engine.py +++ b/.agents/skills/lib_py/delegation/engine.py @@ -30,7 +30,7 @@ import logging import os from typing import Any, Dict, List, Optional -from lib_py.delegation import mqtt_common +from lib_py.delegation import mqtt_common, notifier logger = logging.getLogger("loop.engine") @@ -65,6 +65,12 @@ REQUEST_TO_STATUS = { # including a terminal one. Keep this list short and obviously status-free. BOOKKEEPING_REQUESTS = ("notification_settled",) +# The only fields a bookkeeping request may write. +BOOKKEEPING_FIELDS = ("pending_notification",) + +# Distinguishes "no expectation" from "expect the field to be None". +_UNSET = object() + # 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") @@ -122,7 +128,7 @@ class _Decision(object): def _decide(record: Dict[str, Any], request: str, expected_status, expected_attempt, - source="unknown", fields=None) -> _Decision: + source="unknown", fields=None, expected_notification=_UNSET) -> _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. @@ -169,6 +175,24 @@ def _decide(record: Dict[str, Any], request: str, expected_status, expected_atte "{} requires operator authority (source={!r})".format(request, source)) if request in BOOKKEEPING_REQUESTS: + # Bookkeeping is allowed in a terminal state precisely because it cannot + # change anything of consequence. That has to be enforced here, at the + # boundary, and not merely by what today's one caller happens to pass: + # otherwise a status-free request becomes an unrestricted merge on a + # record nothing else is allowed to touch. + extra = sorted(set(fields) - set(BOOKKEEPING_FIELDS)) + if extra: + return _Decision(False, current, + "bookkeeping may not merge {}".format(extra), + conflict=True) + expected = expected_notification + if expected is not _UNSET: + current_intent = record.get("pending_notification") + if current_intent != expected: + # A newer intent replaced the one being settled while it was in + # flight. Clearing it now would drop a notification nobody sent. + return _Decision(False, current, + "notification intent changed while it was being sent") return _Decision(True, None, "bookkeeping ({})".format(request)) if current in TERMINAL_STATUSES: @@ -219,7 +243,7 @@ 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) -> TransitionResult: + notify=None, expected_notification=_UNSET) -> TransitionResult: """Decide and commit one state transition. The only writer of job status. Returns a result whose `notifications` have ALREADY been flushed; the field @@ -233,7 +257,8 @@ def transition(job_id, request="merge", *, registry_dir=None, expected_status=No record = mqtt_common.load_job(job_id, registry_dir) from_status = record.get("status") decision = _decide(record, request, expected_status, expected_attempt, - source=source, fields=fields) + source=source, fields=fields, + expected_notification=expected_notification) notifications: List[Dict[str, Any]] = [] if decision.commit: @@ -248,7 +273,6 @@ def transition(job_id, request="merge", *, registry_dir=None, expected_status=No # 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 # publish, the record still says a notification is owed. - from lib_py.delegation import notifier record[notifier.FIELD] = notifier.intent_for( notify, record["transition_seq"], record["updated_at"], detail=str(fields.get("detail", ""))) diff --git a/.agents/skills/lib_py/delegation/finalize.py b/.agents/skills/lib_py/delegation/finalize.py index aa03fc1..0b2004f 100644 --- a/.agents/skills/lib_py/delegation/finalize.py +++ b/.agents/skills/lib_py/delegation/finalize.py @@ -68,6 +68,25 @@ def _atomic_write(path: Path, text: str) -> None: raise +def validate_side_effect(entry) -> Optional[str]: + """None when the entry is usable evidence, else why it is not. + + A kind with no reference names something without evidencing it, which is the + shape "declared but unproven" takes. The CLI parser cannot be the only place + this is checked: finalize_report is a public API and a manifest can also be + written by hand. + """ + if not isinstance(entry, dict): + return "side effect entry is not an object" + kind = str(entry.get("kind") or "").strip() + reference = str(entry.get("reference") or "").strip() + if not kind: + return "side effect entry has no kind" + if not reference: + return "side effect {!r} has no reference".format(kind) + return None + + def parse_side_effects(raw: Optional[List[str]]) -> List[Dict[str, str]]: """`kind=reference` pairs from the command line into evidence entries.""" out = [] @@ -89,7 +108,14 @@ def finalize_report(report_path, job_id, attempt_id, side_effects=None, never observe a half-written report as finalized. The rename is what makes the transition atomic; the manifest is what makes it attributable. """ - report_path = Path(report_path) + for entry in side_effects or []: + problem = validate_side_effect(entry) + if problem: + raise ValueError(problem) + + # Absolute, so a later reader in a different working directory resolves the + # same file rather than a same-named one beside itself. + report_path = Path(report_path).resolve() staged = report_path.with_name(report_path.name + ".tmp") if staged.exists(): os.replace(str(staged), str(report_path)) @@ -157,8 +183,13 @@ def verify_manifest(report_dir, job_id, expected_attempt=None, # A job that declared side effects must show evidence for every one of them. # Partial evidence is no evidence: the work is not known to be complete. - declared = {e.get("kind") for e in manifest.get("side_effects") or []} - missing = [a for a in (expected_artifacts or []) if a not in declared] + evidenced = set() + for entry in manifest.get("side_effects") or []: + problem = validate_side_effect(entry) + if problem: + return ManifestVerdict(False, problem) + evidenced.add(str(entry["kind"]).strip()) + missing = [a for a in (expected_artifacts or []) if a not in evidenced] if missing: return ManifestVerdict(False, "no side-effect evidence for {}".format(missing)) diff --git a/.agents/skills/lib_py/delegation/notifier.py b/.agents/skills/lib_py/delegation/notifier.py index 59a41e9..20912e8 100644 --- a/.agents/skills/lib_py/delegation/notifier.py +++ b/.agents/skills/lib_py/delegation/notifier.py @@ -44,17 +44,25 @@ def jobs_with_pending(registry_dir: str) -> List[str]: return out -def settle(job_id: str, registry_dir: str, source: str = "notifier") -> bool: - """Clear the intent once the event is on the wire. +def settle(job_id: str, registry_dir: str, sent_intent=None, + source: str = "notifier") -> bool: + """Clear the intent that was actually sent, and only that one. Goes through the transition boundary like every other record write, using a request that carries no status: settling is about a terminal, so a request the terminal state would refuse could never be used to settle one. + + `sent_intent` is compared under the lock. Publishing happens outside it, so a + newer intent can legitimately replace the old one mid-flight; clearing + unconditionally would drop a notification nobody has sent. """ from lib_py.delegation import engine result = engine.transition(job_id, "notification_settled", registry_dir=registry_dir, - fields={FIELD: None}, source=source) + fields={FIELD: None}, source=source, + expected_notification=sent_intent) + if not result.committed and "changed while it was being sent" in (result.reason or ""): + logger.info("intent for %s was replaced during send; leaving the newer one", job_id) return result.committed @@ -70,12 +78,22 @@ def flush(job_id: str, registry_dir: str, publish=None, attempts: int = 3) -> bo return True if publish is None: + import json as _json + from lib_py.delegation import publish_event as _pe def publish(job, intent_, registry_dir_): + # The wire event gets a fresh registry sequence, so the intent's own + # identity is carried in the payload instead. A consumer dedupes on + # `notification_intent`, not on the transport sequence -- the + # transport number cannot identify a retry of the same intent. + identity = _json.dumps({"notification_intent": { + "seq": intent_.get("seq"), "event": intent_.get("event"), + "timestamp": intent_.get("timestamp")}}) return _pe.main(["--registry-dir", registry_dir_, "--job", job, "--event", intent_["event"], "--detail", - intent_.get("detail", ""), "--attempts", str(attempts)]) + intent_.get("detail", ""), "--data", identity, + "--attempts", str(attempts)]) try: rc = publish(job_id, intent, registry_dir) @@ -86,7 +104,7 @@ def flush(job_id: str, registry_dir: str, publish=None, attempts: int = 3) -> bo logger.warning("notification for %s not accepted (rc=%s); leaving it pending", job_id, rc) return False - return settle(job_id, registry_dir) + return settle(job_id, registry_dir, sent_intent=intent) def recover(registry_dir: str, publish=None) -> Dict[str, bool]: diff --git a/.agents/skills/lib_py/loop/autocomplete.py b/.agents/skills/lib_py/loop/autocomplete.py index a93a19a..54964b5 100644 --- a/.agents/skills/lib_py/loop/autocomplete.py +++ b/.agents/skills/lib_py/loop/autocomplete.py @@ -17,6 +17,8 @@ from __future__ import annotations import logging import os +import re +from pathlib import Path from typing import Any, Dict, List, Optional from lib_py.delegation import finalize @@ -33,6 +35,16 @@ BACKENDS_FLAG = "MAM_AUTO_COMPLETE_BACKENDS" # working; hermes and opencode have no observations at all. EVIDENCE_BACKED_BACKENDS = ("claude", "grok", "codex") +# The `agent` recorded on a job is a delegate key. Matching these exactly rather +# than by prefix keeps a future `claude-something-unverified` from inheriting +# evidence that was never collected for it. +EVIDENCE_BACKED_AGENT_KEYS = ( + "claude", "claude-code", "grok", "grok-build", "codex", "codex-cli", +) + +# A reviewer's verdict token, matched the way run_loop.sh matches it. +VERDICT_LINE = re.compile(r"^\[VERDICT: (PASS|NOT PASS)\][ \t]*\r?$") + SOURCE = "auto_complete" @@ -67,21 +79,74 @@ class GateResult(object): self.committed, self.decision, self.reason) +def _allowed_agent_keys() -> tuple: + allowed = allowed_backends() + return tuple(k for k in EVIDENCE_BACKED_AGENT_KEYS + if any(k == b or k.startswith(b + "-") for b in allowed)) + + def _gates_open(record: Dict[str, Any]) -> Optional[str]: - """None when all three gates pass, else why they did not.""" + """None when every gate passes, else why one did not.""" + from lib_py.delegation import engine + if not enabled(): return "auto-completion is disabled ({}=0)".format(FLAG) + if not engine.cas_enabled(): + # The operations note promises this switch disables auto-completion too, + # and it must: inferred completion is built on the CAS boundary, so + # committing while that boundary is rolled back would bypass it. + return "the CAS engine is rolled back (MAM_ENGINE_CAS=0)" backend = record.get("agent") or "" - # `agent` may be a delegate key such as `claude-code`; match on the prefix. - if not any(backend == b or backend.startswith(b) for b in allowed_backends()): + if backend not in _allowed_agent_keys(): return "backend {!r} has no observed quiescent-state evidence".format(backend) if not record.get("auto_complete"): return "job did not opt in to auto-completion" return None +def _verdict_present(report_path) -> bool: + """A reviewer's report must end with its verdict token. + + Phase 0 made this a safety contract: without a verdict, a reviewer job may + never be promoted to completed. run_loop.sh reads a missing token as NOT + PASS, so auto-completing before the token exists would start a correction + round against a review that was never finished. + """ + try: + lines = Path(report_path).read_text(encoding="utf-8").splitlines() + except OSError: + return False + for line in reversed(lines): + if line.strip(): + return bool(VERDICT_LINE.match(line)) + return False + + +def _scenario_matches(job_id, scenario, dispatch) -> Optional[str]: + """The evidence must be about this job and this dispatch. + + Without this the gate will happily complete one job using another job's + evidence: the policy only checks a scenario against the dispatch context + carried inside that same scenario, which is self-consistent by construction. + """ + context = scenario.get("dispatch_context") or {} + if context.get("job_id") != job_id: + return "evidence is for job {!r}, not {!r}".format(context.get("job_id"), job_id) + for field in ("attempt_id", "agent_session_uuid", "pane_process_generation"): + expected = dispatch.get(field) + if expected is None: + return "no recorded dispatch {} to compare the evidence against".format(field) + if context.get(field) != expected: + return "evidence {} {!r} does not match the recorded dispatch {!r}".format( + field, context.get(field), expected) + return None + + def evaluate(job_id, scenario, registry_dir=None, workspace=None, report_dir=None) -> GateResult: + # `report_dir` keeps a default only so the signature stays callable from a + # supervisor that has not resolved it yet; omitting it is a refusal, not a + # shortcut. See the check at the top of the body. """Decide, and commit only if every gate agrees. `scenario` is the composite evidence the watcher assembled, in the same shape @@ -100,16 +165,33 @@ def evaluate(job_id, scenario, registry_dir=None, workspace=None, if closed: return GateResult(False, "hold", closed) - # The manifest is verified against the file on disk, not against itself, and - # against any side effects this job declared it would perform. - dispatch = store.load_dispatch(job_id, workspace) or {} - attempt = dispatch.get("attempt_id") - if report_dir is not None: - verdict = finalize.verify_manifest( - report_dir, job_id, expected_attempt=attempt, - expected_artifacts=record.get("expected_artifacts") or []) - if not verdict.ok: - return GateResult(False, "hold", verdict.reason) + # Independent verification is mandatory, never conditional on the caller + # remembering an argument. Skipping it would put the collector's own flags + # back in charge of the decision, which is the self-certification this whole + # design exists to avoid. + if report_dir is None: + return GateResult(False, "hold", + "no report directory to independently verify the manifest against") + + dispatch = store.load_dispatch(job_id, workspace) + if not dispatch: + return GateResult(False, "hold", "no recorded dispatch to attribute this completion to") + + mismatch = _scenario_matches(job_id, scenario, dispatch) + if mismatch: + return GateResult(False, "hold", mismatch) + + verdict = finalize.verify_manifest( + report_dir, job_id, expected_attempt=dispatch.get("attempt_id"), + expected_artifacts=record.get("expected_artifacts") or []) + if not verdict.ok: + return GateResult(False, "hold", verdict.reason) + + # A reviewer that has not written its verdict has not finished reviewing. + if str(record.get("role") or "").lower() == "reviewer": + if not _verdict_present(verdict.manifest["report_path"]): + return GateResult(False, "hold", + "reviewer report carries no verdict token on its last line") try: decision = policy.decide(scenario) diff --git a/tests/test_issue5_phase2b_autocomplete.py b/tests/test_issue5_phase2b_autocomplete.py index d92419e..8f36ea6 100644 --- a/tests/test_issue5_phase2b_autocomplete.py +++ b/tests/test_issue5_phase2b_autocomplete.py @@ -519,3 +519,266 @@ def test_g7_restart_recovery_resends_every_owed_notification(mam_sandbox, tmp_pa results = notifier.recover(reg, publish=lambda j, i, r: sent.append(j) or 0) assert all(results.values()) and sorted(sent) == sorted(owed) assert notifier.jobs_with_pending(reg) == [], "nothing is owed after recovery" + + +# ============================================================================== +# 6. Cross-review regressions (job f63d8279) +# +# Three reviewers, seven blocking items, all reproduced before being accepted. +# The theme: a gate whose strictness depended on the caller, and a boundary that +# trusted its only current caller instead of enforcing its own rule. +# ============================================================================== + +def test_f1_an_omitted_report_directory_is_a_refusal_not_a_shortcut( + mam_sandbox, tmp_path, monkeypatch +): + """Independent verification cannot be optional: skipping it puts the + collector's own flags back in charge of the decision.""" + monkeypatch.chdir(mam_sandbox) + monkeypatch.setenv("MAM_AUTO_COMPLETE", "1") + from lib_py.loop import autocomplete + + job_id, reg, scenario, _report_dir = _ready_job(tmp_path, mam_sandbox) + before = _fingerprint(Path(reg)) + result = autocomplete.evaluate(job_id, scenario, registry_dir=reg, + workspace=str(mam_sandbox)) # no report_dir + assert result.committed is False and "independently verify" in result.reason + assert _fingerprint(Path(reg)) == before, "a refusal must not write" + + +def test_f1_an_empty_report_directory_is_also_a_refusal(mam_sandbox, tmp_path, monkeypatch): + monkeypatch.chdir(mam_sandbox) + monkeypatch.setenv("MAM_AUTO_COMPLETE", "1") + from lib_py.loop import autocomplete + + job_id, reg, scenario, _rd = _ready_job(tmp_path, mam_sandbox) + empty = tmp_path / "empty-reports" + empty.mkdir() + result = autocomplete.evaluate(job_id, scenario, registry_dir=reg, + workspace=str(mam_sandbox), report_dir=empty) + assert result.committed is False and "no completion manifest" in result.reason + + +def test_r2_evidence_for_another_job_cannot_complete_this_one( + mam_sandbox, tmp_path, monkeypatch +): + """The policy checks a scenario against the dispatch context inside that same + scenario, which is self-consistent by construction. The gate must bind the + evidence to the target.""" + monkeypatch.chdir(mam_sandbox) + monkeypatch.setenv("MAM_AUTO_COMPLETE", "1") + from lib_py.loop import autocomplete + + job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox) + scenario["dispatch_context"]["job_id"] = "some-other-job" + result = autocomplete.evaluate(job_id, scenario, registry_dir=reg, + workspace=str(mam_sandbox), report_dir=report_dir) + assert result.committed is False and "not" in result.reason + assert _load(reg, job_id)["status"] == "running" + + +@pytest.mark.parametrize("field", ["attempt_id", "agent_session_uuid", "pane_process_generation"]) +def test_r2_evidence_must_match_the_recorded_dispatch(mam_sandbox, tmp_path, monkeypatch, field): + monkeypatch.chdir(mam_sandbox) + monkeypatch.setenv("MAM_AUTO_COMPLETE", "1") + from lib_py.loop import autocomplete + + job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox) + scenario["dispatch_context"][field] = "not-what-was-dispatched" + result = autocomplete.evaluate(job_id, scenario, registry_dir=reg, + workspace=str(mam_sandbox), report_dir=report_dir) + assert result.committed is False and field in result.reason + + +def test_r2_a_job_with_no_recorded_dispatch_is_refused(mam_sandbox, tmp_path, monkeypatch): + monkeypatch.chdir(mam_sandbox) + monkeypatch.setenv("MAM_AUTO_COMPLETE", "1") + from lib_py.loop import autocomplete + + job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox) + (Path(mam_sandbox) / ".mam" / "shadow" / "attribution" / (job_id + ".json")).unlink() + result = autocomplete.evaluate(job_id, scenario, registry_dir=reg, + workspace=str(mam_sandbox), report_dir=report_dir) + assert result.committed is False and "no recorded dispatch" in result.reason + + +def test_the_cas_rollback_switch_also_disables_auto_completion(mam_sandbox, tmp_path, monkeypatch): + """Inferred completion is built on the CAS boundary, so committing while + that boundary is rolled back would bypass it.""" + monkeypatch.chdir(mam_sandbox) + monkeypatch.setenv("MAM_AUTO_COMPLETE", "1") + from lib_py.loop import autocomplete + + job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox) + monkeypatch.setenv("MAM_ENGINE_CAS", "0") + result = autocomplete.evaluate(job_id, scenario, registry_dir=reg, + workspace=str(mam_sandbox), report_dir=report_dir) + assert result.committed is False and "rolled back" in result.reason + assert _load(reg, job_id)["status"] == "running" + + +@pytest.mark.parametrize("agent", ["claude-experimental", "codexish", "grok2"]) +def test_o4_a_backend_key_must_match_exactly_not_by_prefix( + mam_sandbox, tmp_path, monkeypatch, agent +): + """A future name must not inherit evidence that was never collected for it.""" + monkeypatch.chdir(mam_sandbox) + monkeypatch.setenv("MAM_AUTO_COMPLETE", "1") + from lib_py.loop import autocomplete + + job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox, backend=agent) + result = autocomplete.evaluate(job_id, scenario, registry_dir=reg, + workspace=str(mam_sandbox), report_dir=report_dir) + assert result.committed is False and "quiescent-state evidence" in result.reason + + +def test_a_reviewer_without_a_verdict_token_is_not_completed(mam_sandbox, tmp_path, monkeypatch): + """Phase 0 safety contract: a reviewer that has not written its verdict has + not finished reviewing, and run_loop reads a missing token as NOT PASS.""" + monkeypatch.chdir(mam_sandbox) + monkeypatch.setenv("MAM_AUTO_COMPLETE", "1") + from lib_py.delegation import finalize + from lib_py.loop import autocomplete + + job_id, reg, scenario, report_dir = _ready_job( + tmp_path, mam_sandbox, body="a thorough review with no verdict line\n") + from lib_py.delegation import engine + engine.transition(job_id, "progress", registry_dir=reg, fields={"role": "Reviewer"}) + + result = autocomplete.evaluate(job_id, scenario, registry_dir=reg, + workspace=str(mam_sandbox), report_dir=report_dir) + assert result.committed is False and "verdict token" in result.reason + assert _load(reg, job_id)["status"] == "running" + + # writing the verdict requires a fresh finalization, and then it passes + report = report_dir / "report-final.md" + report.write_text("a thorough review\n\n[VERDICT: PASS]\n", encoding="utf-8") + finalize.finalize_report(report, job_id, scenario["dispatch_context"]["attempt_id"]) + ok = autocomplete.evaluate(job_id, scenario, registry_dir=reg, + workspace=str(mam_sandbox), report_dir=report_dir) + assert ok.committed, ok.reason + + +def test_a_worker_role_needs_no_verdict_token(mam_sandbox, tmp_path, monkeypatch): + """The token requirement is a reviewer contract, not a general one.""" + monkeypatch.chdir(mam_sandbox) + monkeypatch.setenv("MAM_AUTO_COMPLETE", "1") + from lib_py.loop import autocomplete + + job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox, + body="worker output, no verdict\n") + assert _load(reg, job_id)["role"] == "Worker" + result = autocomplete.evaluate(job_id, scenario, registry_dir=reg, + workspace=str(mam_sandbox), report_dir=report_dir) + assert result.committed, result.reason + + +def test_grok_f2_bookkeeping_cannot_merge_arbitrary_fields(mam_sandbox, tmp_path, monkeypatch): + """A status-free request must not become an unrestricted merge on a record + nothing else is allowed to touch.""" + 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, notify="completed") + before = _load(reg, job_id) + + result = engine.transition(job_id, "notification_settled", registry_dir=reg, + fields={"pending_notification": None, "prompt": "HACKED", + "agent_session": "herdr:attacker"}, + source="not-an-operator") + after = _load(reg, job_id) + assert result.committed is False and "may not merge" in result.reason + assert after["prompt"] == before["prompt"] + assert after["agent_session"] == before["agent_session"] + assert after["pending_notification"], "the intent survives a refused settle" + + # the legitimate settle still works + from lib_py.delegation import notifier + assert notifier.settle(job_id, reg, sent_intent=after["pending_notification"]) + assert notifier.pending(job_id, reg) is None + + +def test_r4_a_newer_intent_is_not_erased_by_a_slow_send(mam_sandbox, tmp_path, monkeypatch): + """Publishing happens outside the lock, so a replacement can land mid-flight. + Settling unconditionally would drop a notification nobody sent.""" + monkeypatch.chdir(mam_sandbox) + from lib_py.delegation import engine, notifier + + 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, notify="completed") + + def racing_publisher(job, intent, registry_dir): + engine.transition(job, "notification_settled", registry_dir=registry_dir, + fields={"pending_notification": + notifier.intent_for("error", 999, "later")}) + return 0 + + notifier.flush(job_id, reg, publish=racing_publisher) + still_owed = notifier.pending(job_id, reg) + assert still_owed and still_owed["seq"] == 999, "the replacement must survive" + + +def test_r4_the_published_payload_carries_the_intent_identity(mam_sandbox, tmp_path, monkeypatch): + """The transport sequence cannot identify a retry of the same intent, so the + intent's own identity travels in the payload for consumers to dedupe on.""" + monkeypatch.chdir(mam_sandbox) + from lib_py.delegation import engine, notifier + + reg = _reg(tmp_path) + job_id = _make_job(reg) + record = _load(reg, job_id) + record["broker"] = {"host": "127.0.0.1", "port": 65432, "tls": False} + Path(reg, job_id + ".json").write_text(json.dumps(record), encoding="utf-8") + engine.transition(job_id, "claim", registry_dir=reg) + engine.transition(job_id, "completed", registry_dir=reg, notify="completed") + + intent = notifier.pending(job_id, reg) + notifier.flush(job_id, reg, attempts=1) # real path; broker is unreachable + from lib_py.delegation import mqtt_common + events = list(mqtt_common.iter_logged_events(job_id, mqtt_common.get_logs_dir())) + published = [e for e in events if e.get("event") == "published"] + assert published, "the attempt is recorded even though the broker refused" + carried = published[-1]["payload"]["data"].get("notification_intent") + assert carried and carried["seq"] == intent["seq"] + + +@pytest.mark.parametrize("entry", [ + {"kind": "deploy"}, + {"kind": "deploy", "reference": " "}, + {"reference": "abc"}, + "not-an-object", +]) +def test_r3_a_side_effect_entry_must_carry_real_evidence(tmp_path, entry): + """A kind with no reference names something without evidencing it.""" + from lib_py.delegation import finalize + + path = _report(tmp_path) + with pytest.raises(ValueError): + finalize.finalize_report(path, "job1", 1, side_effects=[entry]) + + # and a hand-written manifest is rejected at verification too + finalize.finalize_report(path, "job1", 1, + side_effects=[{"kind": "deploy", "reference": "sha"}]) + manifest_file = finalize.manifest_in(path.parent) + forged = json.loads(manifest_file.read_text(encoding="utf-8")) + forged["side_effects"] = [entry] + manifest_file.write_text(json.dumps(forged), encoding="utf-8") + verdict = finalize.verify_manifest(path.parent, "job1", expected_artifacts=["deploy"]) + assert not verdict.ok + + +def test_o2_the_manifest_records_an_absolute_report_path(tmp_path, monkeypatch): + """A relative path would resolve against whatever directory the reader is in.""" + from lib_py.delegation import finalize + + path = _report(tmp_path) + monkeypatch.chdir(tmp_path) + manifest = finalize.finalize_report(Path("claude-reports/report-final.md"), "job1", 1) + assert Path(manifest["report_path"]).is_absolute() + monkeypatch.chdir(REPO_ROOT) + assert finalize.verify_manifest(path.parent, "job1").ok, "verification survives a chdir"