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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UrDNJkHmm18uu7n7JwV5Qd

.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")

.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

.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)

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"):
+        for name, pattern in patterns.items():
-            assert forbidden not in src, "{} references {}".format(path.name, forbidden)
+            assert not pattern.search(src), "{} references {}".format(path.name, name)
 
 
 # ==============================================================================

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"