From 803ba4388482c1add20ba2e016937a8b9ff4e19a Mon Sep 17 00:00:00 2001 From: Godopu Date: Sun, 6 Sep 2026 21:30:38 +0900 Subject: [PATCH] feat(issue-5): auto-completion gate and durable pending notification The gate does not re-decide anything. Phase 2a already decides, fail-closed and against contract-pinned vectors; this supplies real evidence to that decision and commits only if three independent gates agree: the global switch is on, the backend has observed evidence that it reaches a quiescent state, and the job itself opted in. Default is off. The backend allowlist is evidence-based rather than optimistic. Inferring "finished" from stillness requires stillness to have been observed, and only three of six backends have that. An operator may narrow the list; a name that was never observed is dropped rather than trusted, so it cannot be widened by configuration. Everything else holds. An unreadable manifest, a report rewritten after finalization, missing evidence for a declared side effect, a policy that raises: all refuse. "Could not decide" is never "done". A late explicit terminal wins, because the boundary refuses the inferred one -- that is the boundary working, not an error in the gate. Notification intent is committed in the same atomic write as the state, so a crash between committing and publishing leaves a record that still says the world has not been told. Recovery re-sends; duplicates are already dropped by sequence. A failed send deliberately leaves the intent pending, because an unsent notification that looks sent is worse than one that is retried. Two fixes this surfaced. Settling used a request that terminal records refuse, which made an owed notification impossible to ever settle; settling is now a status-free bookkeeping request, valid in any state. And the phase 2a guard against shell references was matching hashlib.sha256 as a substring, so it now requires a word boundary. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01UrDNJkHmm18uu7n7JwV5Qd --- .agents/skills/lib_py/delegation/engine.py | 24 +- .agents/skills/lib_py/delegation/notifier.py | 97 ++++++++ .agents/skills/lib_py/loop/autocomplete.py | 137 +++++++++++ tests/test_issue5_phase2a_loop.py | 13 +- tests/test_issue5_phase2b_autocomplete.py | 246 +++++++++++++++++++ 5 files changed, 513 insertions(+), 4 deletions(-) create mode 100644 .agents/skills/lib_py/delegation/notifier.py create mode 100644 .agents/skills/lib_py/loop/autocomplete.py diff --git a/.agents/skills/lib_py/delegation/engine.py b/.agents/skills/lib_py/delegation/engine.py index a1a9d6e..78603ab 100644 --- a/.agents/skills/lib_py/delegation/engine.py +++ b/.agents/skills/lib_py/delegation/engine.py @@ -55,8 +55,16 @@ REQUEST_TO_STATUS = { # separate requests gated on authority rather than an exception in the table. "redispatch": "pending", "finalize": "error", + # Bookkeeping: records that an owed notification was delivered. It cannot + # change status, and it is *about* a terminal, so refusing it in a terminal + # state would make an owed notification impossible to ever settle. + "notification_settled": None, } +# Requests that carry no status at all and are therefore safe in any state, +# including a terminal one. Keep this list short and obviously status-free. +BOOKKEEPING_REQUESTS = ("notification_settled",) + # 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") @@ -160,6 +168,9 @@ def _decide(record: Dict[str, Any], request: str, expected_status, expected_atte return _Decision(False, current, "{} requires operator authority (source={!r})".format(request, source)) + if request in BOOKKEEPING_REQUESTS: + return _Decision(True, None, "bookkeeping ({})".format(request)) + if current in TERMINAL_STATUSES: # Idempotent re-receipt of the same terminal is a no-op ack. if requested == current: @@ -207,7 +218,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") -> TransitionResult: + expected_attempt=None, fields=None, source="unknown", + notify=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 @@ -231,7 +243,15 @@ def transition(job_id, request="merge", *, registry_dir=None, expected_status=No record["updated_at"] = mqtt_common._utcnow() # 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)) + 1 + record["transition_seq"] = int(record.get("transition_seq", 0) or 0) + 1 + 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 + # 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", ""))) mqtt_common._atomic_write_record(job_id, registry_dir, record) to_status = record.get("status") diff --git a/.agents/skills/lib_py/delegation/notifier.py b/.agents/skills/lib_py/delegation/notifier.py new file mode 100644 index 0000000..59a41e9 --- /dev/null +++ b/.agents/skills/lib_py/delegation/notifier.py @@ -0,0 +1,97 @@ +"""Durable notification intent, carried in the job record. + +The contract forbids a separate outbox store, so the intent rides in the record +the transition already writes: one atomic commit records both the new state and +the fact that the world has not been told yet. Publishing then happens outside +the registry lock, and settling the intent is itself an ordinary transition. + +That ordering is the point. If the process dies between the commit and the +publish, the record still says a notification is owed, and startup recovery +re-sends it. Duplicates are safe: the subscriber already drops any event whose +sequence is not greater than the highest it has accepted. +""" +from __future__ import annotations + +import logging +from typing import Any, Dict, List, Optional + +from lib_py.delegation import mqtt_common + +logger = logging.getLogger("delegate_job.notifier") + +FIELD = "pending_notification" + + +def intent_for(event: str, seq: int, timestamp: str, detail: str = "") -> Dict[str, Any]: + return {"event": event, "seq": seq, "timestamp": timestamp, "detail": detail} + + +def pending(job_id: str, registry_dir: str) -> Optional[Dict[str, Any]]: + try: + return mqtt_common.load_job(job_id, registry_dir).get(FIELD) + except FileNotFoundError: + return None + + +def jobs_with_pending(registry_dir: str) -> List[str]: + """Records still owing a notification. The input to restart recovery.""" + from lib_py.delegation import registry # local: keeps module import order flat + + out = [] + for record in registry.list_jobs(registry_dir): + if record.get(FIELD): + out.append(record["job_id"]) + return out + + +def settle(job_id: str, registry_dir: str, source: str = "notifier") -> bool: + """Clear the intent once the event is on the wire. + + 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. + """ + from lib_py.delegation import engine + + result = engine.transition(job_id, "notification_settled", registry_dir=registry_dir, + fields={FIELD: None}, source=source) + return result.committed + + +def flush(job_id: str, registry_dir: str, publish=None, attempts: int = 3) -> bool: + """Publish an owed notification and settle it. Safe to call repeatedly. + + Returns True when nothing is owed afterwards. A failure leaves the intent in + place on purpose: an unsent notification that looks sent is worse than one + that is retried. + """ + intent = pending(job_id, registry_dir) + if not intent: + return True + + if publish is None: + from lib_py.delegation import publish_event as _pe + + def publish(job, intent_, registry_dir_): + return _pe.main(["--registry-dir", registry_dir_, "--job", job, + "--event", intent_["event"], "--detail", + intent_.get("detail", ""), "--attempts", str(attempts)]) + + try: + rc = publish(job_id, intent, registry_dir) + except Exception as exc: + logger.warning("notification publish failed for %s: %s", job_id, exc) + return False + if rc != 0: + logger.warning("notification for %s not accepted (rc=%s); leaving it pending", + job_id, rc) + return False + return settle(job_id, registry_dir) + + +def recover(registry_dir: str, publish=None) -> Dict[str, bool]: + """Re-send every owed notification. Called at startup after a crash.""" + results = {} + for job_id in jobs_with_pending(registry_dir): + results[job_id] = flush(job_id, registry_dir, publish=publish) + return results diff --git a/.agents/skills/lib_py/loop/autocomplete.py b/.agents/skills/lib_py/loop/autocomplete.py new file mode 100644 index 0000000..4aa88af --- /dev/null +++ b/.agents/skills/lib_py/loop/autocomplete.py @@ -0,0 +1,137 @@ +"""The auto-completion gate. + +Phase 2a already decides, correctly and fail-closed. This does not re-decide; +it supplies real evidence to that decision and, only if three independent gates +agree, asks the transition boundary to commit. + +Three gates, all of which must hold: + * the global switch is on (default off), + * the backend has observed evidence that it reaches a quiescent state, + * the job itself opted in. + +Everything else is refusal. A failure anywhere -- an unreadable manifest, an +exception in the policy, an unknown decision -- holds. "Could not decide" is +never "done". +""" +from __future__ import annotations + +import logging +import os +from typing import Any, Dict, List, Optional + +from lib_py.loop import finalize, policy, store + +logger = logging.getLogger("loop.autocomplete") + +FLAG = "MAM_AUTO_COMPLETE" +BACKENDS_FLAG = "MAM_AUTO_COMPLETE_BACKENDS" + +# Backends with recorded evidence that they reach a quiescent state. Auto +# completion infers "finished" from stillness, so a backend whose stillness has +# never been observed has no basis for the inference. agy has only been seen +# working; hermes and opencode have no observations at all. +EVIDENCE_BACKED_BACKENDS = ("claude", "grok", "codex") + +SOURCE = "auto_complete" + + +def enabled() -> bool: + """Default off. Read per call so an operator can flip it without a restart.""" + return os.environ.get(FLAG, "0") == "1" + + +def allowed_backends() -> tuple: + """The allowlist may be narrowed by an operator, never widened. + + Widening would need new observations, so an unknown name in the environment + is dropped rather than trusted. + """ + raw = os.environ.get(BACKENDS_FLAG) + if not raw: + return EVIDENCE_BACKED_BACKENDS + requested = [name.strip() for name in raw.split(",") if name.strip()] + return tuple(n for n in requested if n in EVIDENCE_BACKED_BACKENDS) + + +class GateResult(object): + __slots__ = ("committed", "decision", "reason") + + def __init__(self, committed, decision, reason): + self.committed = committed + self.decision = decision + self.reason = reason + + def __repr__(self): # pragma: no cover - debugging aid + return "GateResult(committed={!r}, {!r}, {!r})".format( + self.committed, self.decision, self.reason) + + +def _gates_open(record: Dict[str, Any]) -> Optional[str]: + """None when all three gates pass, else why they did not.""" + if not enabled(): + return "auto-completion is disabled ({}=0)".format(FLAG) + 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()): + 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 evaluate(job_id, scenario, registry_dir=None, workspace=None, + report_dir=None) -> GateResult: + """Decide, and commit only if every gate agrees. + + `scenario` is the composite evidence the watcher assembled, in the same shape + the phase 0 fixtures use, so the decision is made by exactly the code the + scenario contracts already pin. + """ + from lib_py.delegation import engine, mqtt_common + + registry_dir = registry_dir or mqtt_common.DEFAULT_REGISTRY_DIR + try: + record = mqtt_common.load_job(job_id, registry_dir) + except FileNotFoundError: + return GateResult(False, "hold", "no such job") + + closed = _gates_open(record) + 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) + + try: + decision = policy.decide(scenario) + except Exception as exc: # a policy that cannot decide has not approved + logger.warning("policy raised for %s: %s", job_id, exc) + return GateResult(False, "hold", "policy error: {}".format(exc)) + + if not decision.would_complete: + return GateResult(False, decision.decision, decision.reason) + + # The attempt guard lives here, in the evidence, not in the record: a record + # 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, + fields={"completed_by": SOURCE, "auto_complete_reason": decision.reason}, + notify="completed", + ) + if not result.committed: + # A late explicit terminal, or a job that moved on. The boundary refused; + # that is the boundary doing its job, not an error here. + logger.info("auto-completion declined for %s: %s", job_id, result.reason) + return GateResult(False, "hold", result.reason) + + logger.info("auto-completed %s (%s)", job_id, decision.reason) + return GateResult(True, decision.decision, decision.reason) diff --git a/tests/test_issue5_phase2a_loop.py b/tests/test_issue5_phase2a_loop.py index bd5164d..89b6289 100644 --- a/tests/test_issue5_phase2a_loop.py +++ b/tests/test_issue5_phase2a_loop.py @@ -691,10 +691,19 @@ def test_g9_loop_package_imports_only_delegation_among_siblings(): def test_g9_phase2a_spawns_no_daemon_and_calls_no_frozen_helper(): """The two phase 1 handoffs stay dormant: preexec_fn is unreachable because nothing here spawns, and no frozen lib.sh helper is invoked.""" + import re + + # `lib.sh` needs a word boundary: a bare substring search also matches + # hashlib.sha256, which is how this guard first fired on innocent code. + patterns = { + "ensure_subscriber_daemon": re.compile(r"ensure_subscriber_daemon"), + "preexec_fn": re.compile(r"preexec_fn"), + "lib.sh": re.compile(r"\blib\.sh\b"), + } for path in sorted((SKILLS / "lib_py" / "loop").glob("*.py")): src = path.read_text(encoding="utf-8") - for forbidden in ("ensure_subscriber_daemon", "preexec_fn", "lib.sh"): - assert forbidden not in src, "{} references {}".format(path.name, forbidden) + for name, pattern in patterns.items(): + assert not pattern.search(src), "{} references {}".format(path.name, name) # ============================================================================== diff --git a/tests/test_issue5_phase2b_autocomplete.py b/tests/test_issue5_phase2b_autocomplete.py index 818a5d3..1dbede7 100644 --- a/tests/test_issue5_phase2b_autocomplete.py +++ b/tests/test_issue5_phase2b_autocomplete.py @@ -271,3 +271,249 @@ def test_publish_event_reports_a_bad_side_effect_instead_of_publishing(mam_sandb "--finalize-report", str(path), "--side-effect", "malformed", ]) assert rc == 1 + + +# ============================================================================== +# 3. autocomplete — the gate, and everything that must stop it +# ============================================================================== + +def _ready_job(tmp_path, mam_sandbox, *, backend="claude-code", opt_in=True, + artifacts=None, body="report\n"): + """A job in exactly the state that should auto-complete, plus its evidence.""" + from lib_py.loop import finalize, store + + reg = _reg(tmp_path) + job_id = _make_job(reg, agent=backend, expected_artifacts=artifacts or []) + from lib_py.delegation import engine + engine.transition(job_id, "claim", registry_dir=reg) + if opt_in: + engine.transition(job_id, "progress", registry_dir=reg, + fields={"auto_complete": True}) + + scenario = _scenario() + dispatch = scenario["dispatch_context"] + dispatch["job_id"] = job_id + for item in scenario["timeline"]: + env = item.get("collector_attribution") + if env: + env["job_id"] = job_id + # A non-herdr payload is a legitimate identifier source, so it has to + # agree too. Leaving the fixture's original id here is a contradiction, + # and the policy correctly refuses it. + payload = item.get("payload") or {} + if item.get("source_kind") != "herdr" and "job_id" in payload: + payload["job_id"] = job_id + store.record_dispatch(job_id, dispatch["attempt_id"], dispatch["agent_session_uuid"], + dispatch["pane_process_generation"], workspace=str(mam_sandbox)) + store.mark_execution_anchor(job_id, workspace=str(mam_sandbox)) + + report = _report(tmp_path, body) + finalize.finalize_report(report, job_id, dispatch["attempt_id"], + side_effects=finalize.parse_side_effects( + ["{}=ref".format(a) for a in (artifacts or [])])) + return job_id, reg, scenario, report.parent + + +def test_gate_commits_when_every_condition_holds(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) + result = autocomplete.evaluate(job_id, scenario, registry_dir=reg, + workspace=str(mam_sandbox), report_dir=report_dir) + assert result.committed, result.reason + record = _load(reg, job_id) + assert record["status"] == "completed" + assert record["completed_by"] == "auto_complete", "an inferred terminal must be marked" + assert record["pending_notification"]["event"] == "completed" + + +def test_g4_default_off_commits_nothing_and_leaves_the_registry_untouched( + mam_sandbox, tmp_path, monkeypatch +): + """G4: with the flag unset, the registry must not change at all.""" + monkeypatch.chdir(mam_sandbox) + 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), report_dir=report_dir) + assert result.committed is False + assert "disabled" in result.reason + assert _fingerprint(Path(reg)) == before, "the default must write nothing" + + +@pytest.mark.parametrize("backend", ["agy", "hermes", "opencode", "unknown-agent"]) +def test_g9_a_backend_without_quiescence_evidence_never_auto_completes( + mam_sandbox, tmp_path, monkeypatch, backend +): + """Inferring 'finished' from stillness needs stillness to have been observed.""" + 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=backend) + 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 + assert _load(reg, job_id)["status"] == "running" + + +def test_the_allowlist_can_be_narrowed_but_never_widened(monkeypatch): + from lib_py.loop import autocomplete + + monkeypatch.setenv("MAM_AUTO_COMPLETE_BACKENDS", "claude") + assert autocomplete.allowed_backends() == ("claude",) + monkeypatch.setenv("MAM_AUTO_COMPLETE_BACKENDS", "claude,hermes,opencode") + assert autocomplete.allowed_backends() == ("claude",), "unobserved backends are dropped" + + +def test_a_job_that_did_not_opt_in_is_not_auto_completed(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, opt_in=False) + result = autocomplete.evaluate(job_id, scenario, registry_dir=reg, + workspace=str(mam_sandbox), report_dir=report_dir) + assert result.committed is False and "opt in" in result.reason + + +def test_g5_a_rewritten_report_stops_the_gate(mam_sandbox, tmp_path, monkeypatch): + """G5: the manifest is checked against the file, so a late edit disqualifies.""" + 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) + (report_dir / "report-final.md").write_text("rewritten after finalization\n", encoding="utf-8") + result = autocomplete.evaluate(job_id, scenario, registry_dir=reg, + workspace=str(mam_sandbox), report_dir=report_dir) + assert result.committed is False and "changed after finalization" in result.reason + assert _load(reg, job_id)["status"] == "running" + + +def test_g5_missing_side_effect_evidence_stops_the_gate(mam_sandbox, tmp_path, monkeypatch): + monkeypatch.chdir(mam_sandbox) + monkeypatch.setenv("MAM_AUTO_COMPLETE", "1") + from lib_py.loop import autocomplete, finalize + + job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox, + artifacts=["git_push"]) + # re-declare without the evidence the job said it would produce + finalize.finalize_report(report_dir / "report-final.md", job_id, 1, side_effects=[]) + result = autocomplete.evaluate(job_id, scenario, registry_dir=reg, + workspace=str(mam_sandbox), report_dir=report_dir) + assert result.committed is False and "git_push" in result.reason + + +def test_d7_a_policy_that_cannot_decide_holds(mam_sandbox, tmp_path, monkeypatch): + """'Could not decide' is never 'done'.""" + monkeypatch.chdir(mam_sandbox) + monkeypatch.setenv("MAM_AUTO_COMPLETE", "1") + from lib_py.loop import autocomplete, policy + + job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox) + + def explode(_scenario): + raise RuntimeError("evidence is contradictory") + + monkeypatch.setattr(policy, "decide", explode) + result = autocomplete.evaluate(job_id, scenario, registry_dir=reg, + workspace=str(mam_sandbox), report_dir=report_dir) + assert result.committed is False and "policy error" in result.reason + assert _load(reg, job_id)["status"] == "running" + + +def test_a_late_explicit_terminal_beats_the_gate(mam_sandbox, tmp_path, monkeypatch): + """If the agent already reported, the boundary refuses and that is correct.""" + monkeypatch.chdir(mam_sandbox) + monkeypatch.setenv("MAM_AUTO_COMPLETE", "1") + from lib_py.delegation import mqtt_common + from lib_py.loop import autocomplete + + job_id, reg, scenario, report_dir = _ready_job(tmp_path, mam_sandbox) + mqtt_common.update_job_status(job_id, reg, _source="publish_event", status="error") + result = autocomplete.evaluate(job_id, scenario, registry_dir=reg, + workspace=str(mam_sandbox), report_dir=report_dir) + assert result.committed is False + record = _load(reg, job_id) + assert record["status"] == "error", "the explicit report wins" + assert "completed_by" not in record + + +# ============================================================================== +# 4. notifier — the intent is durable, and recovery re-sends it +# ============================================================================== + +def test_the_notification_intent_is_committed_with_the_state(mam_sandbox, tmp_path, monkeypatch): + 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") + record = _load(reg, job_id) + assert record["status"] == "completed" + intent = notifier.pending(job_id, reg) + assert intent and intent["event"] == "completed" and intent["seq"] == record["transition_seq"] + + +def test_flush_publishes_then_settles(mam_sandbox, tmp_path, monkeypatch): + 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") + + sent = [] + assert notifier.flush(job_id, reg, publish=lambda j, i, r: sent.append(i) or 0) + assert len(sent) == 1 and sent[0]["event"] == "completed" + assert notifier.pending(job_id, reg) is None + # settling is idempotent: a second flush has nothing to do + assert notifier.flush(job_id, reg, publish=lambda *a: 1 / 0) + + +def test_a_failed_send_leaves_the_intent_pending(mam_sandbox, tmp_path, monkeypatch): + """An unsent notification that looks sent is worse than one that is retried.""" + 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") + + assert notifier.flush(job_id, reg, publish=lambda *a: 2) is False + assert notifier.pending(job_id, reg) is not None + assert notifier.flush(job_id, reg, publish=lambda *a: (_ for _ in ()).throw(OSError())) is False + assert notifier.pending(job_id, reg) is not None + assert _load(reg, job_id)["status"] == "completed", "the state stays committed regardless" + + +def test_g7_restart_recovery_resends_every_owed_notification(mam_sandbox, tmp_path, monkeypatch): + """G7: the crash window between committing and publishing.""" + monkeypatch.chdir(mam_sandbox) + from lib_py.delegation import engine, notifier + + reg = _reg(tmp_path) + owed = [] + for _ in range(3): + job_id = _make_job(reg) + engine.transition(job_id, "claim", registry_dir=reg) + engine.transition(job_id, "completed", registry_dir=reg, notify="completed") + owed.append(job_id) + settled = _make_job(reg) + engine.transition(settled, "claim", registry_dir=reg) + + assert sorted(notifier.jobs_with_pending(reg)) == sorted(owed) + + sent = [] + 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"