A dispatch exception after the hand-off is an unknown outcome, never "nobody reached"

Coordinator correction to the R1 batch (03 recovery §2 D5, probe P4). 5e5b85b26 kept the
dispatch_ask except branch writing status=failed and dispatched=False for every sent seat
and counted that ask as unroutable, telling the mind and the owner that nobody was reached
while the substrate had already received the seats — an invitation to a second paid ask
with the first outcome unknown. Missing state never proves no send or zero cost, and an
unknown is never cleaned: every seat handed to the substrate now stays pending and
unresolved with "dispatch outcome unknown: <exc>", its barrier dispatched fact stands (the
Swarm obligation is met; the gate reads awaiting, nothing holds), a reply that still
arrives settles it as answered through the custody hook, and unroutable_asks counts only
the recorded attempts and asks whose every seat carries a positive pre-dispatch refusal.
The read view, the ask summary, the progress line, the settled frame and the final-answer
disclosure say "outcome unknown — a new ask may pay twice"; a settled row clears the flag.

ouroboros/owner_hurry.py

@@ -650,7 +650,9 @@ def plan_review_disclosure(decision: Dict[str, Any], forced_reason: str = "") ->
         return f"\n\n⚠️ Plan review: {count} remain{'s' if len(objections) == 1 else ''} open — {listed}; {why}."
     if pending:
         return ("\n\n⚠️ Plan review: the answers of ask(s) " + ", ".join(pending)
-                + " had not arrived when the task ended; silence holds nothing and is not agreement.")
+                + " had not arrived when the task ended; silence holds nothing and is not agreement"
+                + (f"; the dispatch outcome of ask(s) {', '.join(decision['unknown_outcome'])} is unknown — a new ask may pay twice"
+                   if decision.get("unknown_outcome") else "") + ".")
     if unmet and decision.get("allow"):
         attempts = int(decision.get("asks_unroutable") or 0)
         reached = (f"{attempts} ask{'' if attempts == 1 else 's'} could reach no reviewer (every named route was unavailable)"

ouroboros/plan_record.py

@@ -353,14 +353,22 @@ def engaged(record: Mapping[str, Any]) -> bool:
 
 
 def unroutable_asks(record: Mapping[str, Any]) -> int:
-    """How many asks the mind made that reached nobody: the recorded attempts (refused before any
-    row was written) plus the recorded asks whose every participant was refused before dispatch
-    or lost to a dispatch failure — asked, but no reviewer reached."""
+    """How many asks the mind made that POSITIVELY reached nobody: the recorded attempts (refused
+    before any row was written) plus the recorded asks whose every participant carries a typed
+    pre-dispatch refusal (``not_dispatched``). An ask whose dispatch outcome is unknown (an
+    exception after the hand-off) is never counted here: missing state proves no "no send"."""
     refused = sum(1 for ask in record.get("asks") or []
-                  if ask.get("participants") and not any(p.get("dispatched") for p in ask["participants"]))
+                  if ask.get("participants") and all(p.get("status") == "not_dispatched" for p in ask["participants"]))
     return len(record.get("attempts") or []) + refused
 
 
+def unknown_outcome_asks(record: Mapping[str, Any]) -> List[str]:
+    """Asks with a seat whose outcome is unknown (pending and unresolved: custody lost, or the
+    dispatch lost after the hand-off) — a new ask to them may pay twice."""
+    return [str(a.get("ask_id") or "") for a in record.get("asks") or []
+            if any(p.get("status") == "pending" and p.get("unresolved") for p in a.get("participants") or [])]
+
+
 # ------------------------------------------------------------------ gate
 
 def gate_projection(record: Mapping[str, Any], *, enforcement: str, cyber_pro: bool, hurry: bool,
@@ -394,6 +402,7 @@ def gate_projection(record: Mapping[str, Any], *, enforcement: str, cyber_pro: b
                                                      "revision", "owner_marked", "legacy")} for o in objections],
         "holding": [f"{o['seat_id']}:{o['id']}" for o in holding],
         "pending_asks": pending, "asks_dispatched": dispatched, "asks_unroutable": unroutable_asks(record),
+        "unknown_outcome": unknown_outcome_asks(record),
         "force_plan_met": not force_plan or dispatched > 0,
         "revision": int(latest.get("n") or 0) if latest else 0,
         "unreadable": bool(unreadable), "reason": str(hard_rail or ""),
@@ -411,8 +420,8 @@ def gate_projection(record: Mapping[str, Any], *, enforcement: str, cyber_pro: b
     elif unreadable:
         status, allow = "unreadable", False  # a gate that cannot read its authority is engaged, not absent
     elif force_plan and dispatched == 0:
-        # Blocking holds while the mind never asked; an ask that reached nobody (a recorded attempt,
-        # or an ask whose every seat was refused) is disclosed, not held — outages never hold. Advisory and the owner's hurry
+        # Blocking holds while the mind never asked; an ask that positively reached nobody (a recorded
+        # attempt, or every seat refused before dispatch) is disclosed, not held — outages never hold. Advisory and the owner's hurry
         # never hold a plan that exists; a task that wrote no plan at all is still held to write
         # one (the Swarm admitted it on that condition).
         tried = engaged(record) or legacy_engaged
@@ -573,7 +582,7 @@ def read_view(record: Mapping[str, Any], legacy: Optional[Mapping[str, Any]] = N
         "asks": [{"ask_id": a.get("ask_id"), "ts": a.get("ts"), "revision": a.get("revision"), "message": a.get("message"),
                   "participants": [{**{k: p.get(k) for k in ("seat_id", "model", "owner_marked", "status", "dispatched",
                                                                "not_dispatched_reason", "error", "parse_error",
-                                                               "failure_code", "reported_cause", "reset_at")},
+                                                               "failure_code", "reported_cause", "reset_at", "unresolved")},
                                     "ignored_items": len((p.get("reply") or {}).get("ignored") or []) if isinstance(p.get("reply"), dict) else 0}
                                    for p in a.get("participants") or []],
                   "record_id": a.get("record_id")} for a in record.get("asks") or []],

ouroboros/tools/plan_ask.py

@@ -507,6 +507,7 @@ def reply_facts_from_row(row: Mapping[str, Any]) -> Dict[str, Any]:
         **{k: row[k] for k in _PARTICIPANT_RECEIPT_KEYS if row.get(k)},
     }
     state = str(row.get("operation_state") or "settled")
+    facts["unresolved"] = False  # a settled row resolves an earlier unknown outcome; the pending branch below may re-raise it
     if state == "not_dispatched":
         # A typed $0 refusal before the paid boundary (a spent window, a dead route, the deadline before
         # the first send): the seat never reached a reviewer, so the dispatch flag the barrier wrote is
@@ -752,10 +753,17 @@ def dispatch_ask(ctx: Any, prepared: PreparedAsk, *, revision: Mapping[str, Any]
                                      "plan_revision": revision.get("n")}))
     except Exception as exc:
         log.error("plan ask %s dispatch failed: %s", prepared.ask_id, exc, exc_info=True)
+        # The substrate already held every sent seat: an exception after the hand-off is UNKNOWN —
+        # never proof that no send happened or that nothing was paid. The seats stay pending and
+        # unresolved (a reply that still arrives settles them through the custody hook), the barrier's
+        # dispatched fact stands, and the mind is told a new ask may pay twice. Missing state is never
+        # cleaned into "nobody reached".
+        unknown = {"status": "pending", "unresolved": True, "error": f"dispatch outcome unknown: {exc}"}
         for sid in sent:
-            record = plan_record.update_participant(state_root, task_id, ask_id=prepared.ask_id, seat_id=sid,
-                                                    facts={"status": "failed", "error": f"dispatch failed: {exc}", "dispatched": False})
-            settle_ledger_seat(state_root, task_id, record_id, sid, {"status": "failed", "error": f"dispatch failed: {exc}"})
+            record = plan_record.update_participant(state_root, task_id, ask_id=prepared.ask_id, seat_id=sid, facts=unknown)
+            settle_ledger_seat(state_root, task_id, record_id, sid, unknown)
+        _progress(ctx, f"📐 Plan: ask {prepared.ask_id} — dispatch outcome unknown ({type(exc).__name__}); the {len(sent)} sent "
+                       f"participant{'' if len(sent) == 1 else 's'} stay pending, a reply may still arrive; a new ask may pay twice.")
         _reference(ctx, task_id, record, state_root)
         return record
     for row in rows:
@@ -878,8 +886,9 @@ def settle_released_reply(usage_ctx: Any, *, request: Any, task_id: str, actor:
 
 def settled_frame(ask_id: str, participants: List[Mapping[str, Any]], objections: List[Mapping[str, Any]]) -> str:
     """The mailbox frame when custody has nothing more to deliver for an ask: who answered,
-    who could not be read, who failed — and a seat whose custody was LOST is named as
-    unresolved (neither a failure nor a reply), never counted as settled."""
+    who could not be read, who failed — and a seat whose outcome is UNKNOWN (custody lost, or
+    the dispatch lost after the hand-off) is named as unresolved (neither a failure nor a
+    reply), never counted as settled; a new ask to it may pay twice."""
     answered = sum(1 for p in participants if p.get("status") == "answered")
     unread = sum(1 for p in participants if p.get("status") == "unparseable")
     failed = sum(1 for p in participants if p.get("status") == "failed")
@@ -889,8 +898,8 @@ def settled_frame(ask_id: str, participants: List[Mapping[str, Any]], objections
     head = (f"Plan ask {ask_id}: every participant settled" if not unresolved and not waiting
             else f"Plan ask {ask_id}: custody has nothing more to deliver")
     return (f"{head} — {answered} answered" + (f", {unread} unreadable" if unread else "") + (f", {failed} failed" if failed else "")
-            + (f", {len(unresolved)} unresolved ({', '.join(unresolved)}: custody lost — neither a failure nor a reply; nothing waits on it)"
-               if unresolved else "")
+            + (f", {len(unresolved)} unresolved ({', '.join(unresolved)}: outcome unknown — neither a failure nor a reply; nothing waits on it, "
+               "a new ask may pay twice)" if unresolved else "")
             + (f", {len(waiting)} still owed ({', '.join(waiting)})" if waiting else "")
             + (f"; {len(own)} open objection{'' if len(own) == 1 else 's'} from this ask" if own else "; no open objection from this ask")
             + ". plan_task() reads the answers.")

ouroboros/tools/plan_review.py

@@ -300,7 +300,11 @@ def _ask_summary(prepared: Any, record: Mapping[str, Any]) -> str:
     if ask.get("host_added"):
         lines.append("This is the first ask on a body plan, so every owner-marked reviewer was included: the host added "
                      + ", ".join(ask["host_added"]) + " beside the participants you named.")
-    pending = [p["seat_id"] for p in sent if p.get("status") == "pending"]
+    unknown = [p["seat_id"] for p in sent if p.get("status") == "pending" and p.get("unresolved")]
+    pending = [p["seat_id"] for p in sent if p.get("status") == "pending" and not p.get("unresolved")]
+    if unknown:
+        lines.append(f"Dispatch outcome unknown for {', '.join(unknown)} ({next(p.get('error') for p in sent if p['seat_id'] == unknown[0])}): "
+                     "the substrate held them when the host lost track, so a reply may still arrive through custody; a new ask to them may pay twice.")
     if pending:
         lines.append("Answers settle in the background and arrive in your mailbox (await_messages / wait_task); plan_task() reads them. "
                      f"Still working: {', '.join(pending)}.")
@@ -361,6 +365,9 @@ def render_record(record: Mapping[str, Any], decision: Mapping[str, Any], legacy
                 tag = plan_packet.failure_tag(p)
             elif p["status"] == "unparseable":
                 tag = "unparseable" + (f" — {p.get('parse_error')}" if p.get("parse_error") else "")
+            elif p["status"] == "pending" and p.get("unresolved"):
+                tag = ("pending — outcome unknown" + (f" ({p['error']})" if p.get("error") else "")
+                       + "; a reply may still arrive, and a new ask to it may pay twice")
             parts.append(f"{p['seat_id']} ({p.get('model') or '?'}{'' if p.get('owner_marked') else ', added'}): {tag}")
         lines.append(f"- ask {ask['ask_id']} on revision {ask['revision']}" + (f" — «{ask['message']}»" if ask.get("message") else "")
                      + ": " + "; ".join(parts))
@@ -414,7 +421,10 @@ def render_record(record: Mapping[str, Any], decision: Mapping[str, Any], legacy
     elif status == "objections_open":
         lines.append("Open objections are disclosed in your final answer; nothing holds it.")
     elif status == "awaiting":
-        lines.append(f"Answers still on their way for ask(s) {', '.join(decision.get('pending_asks') or view['pending_asks'])}; silence never holds.")
+        lines.append(f"Answers still on their way for ask(s) {', '.join(decision.get('pending_asks') or view['pending_asks'])}; silence never holds."
+                     + (f" The dispatch outcome of ask(s) {', '.join(decision['unknown_outcome'])} is unknown: the substrate held the "
+                        "seats when the host lost track, so a reply may still arrive and a new ask to them may pay twice."
+                        if decision.get("unknown_outcome") else ""))
     elif status == "force_plan_unmet" and decision.get("allow") and decision.get("asks_unroutable"):
         lines.append("This task was admitted with a planning obligation; your ask reached nobody (no route could take it, "
                      f"{decision['asks_unroutable']} attempt{'' if decision['asks_unroutable'] == 1 else 's'} recorded) — disclosed in "

tests/test_plan_organ_gaps.py

@@ -106,8 +106,9 @@ def test_the_settled_frame_names_an_unresolved_seat_instead_of_calling_everyone_
     rows = [{"seat_id": "s1", "status": "answered"}, {"seat_id": "s2", "status": "failed"},
             {"seat_id": "s3", "status": "pending", "unresolved": True}]
     text = plan_ask.settled_frame("a1", rows, [])
-    assert text == ("Plan ask a1: custody has nothing more to deliver — 1 answered, 1 failed, 1 unresolved (s3: custody lost — "
-                    "neither a failure nor a reply; nothing waits on it); no open objection from this ask. plan_task() reads the answers.")
+    assert text == ("Plan ask a1: custody has nothing more to deliver — 1 answered, 1 failed, 1 unresolved (s3: outcome unknown — "
+                    "neither a failure nor a reply; nothing waits on it, a new ask may pay twice); no open objection from this ask. "
+                    "plan_task() reads the answers.")
     clean = plan_ask.settled_frame("a1", [{"seat_id": "s1", "status": "answered"}, {"seat_id": "s2", "status": "answered"}],
                                    [{"ask_id": "a1", "id": "o1"}, {"ask_id": "a0", "id": "old"}])
     assert clean == "Plan ask a1: every participant settled — 2 answered; 1 open objection from this ask. plan_task() reads the answers."
@@ -124,7 +125,7 @@ def test_the_settled_frame_names_an_unresolved_seat_instead_of_calling_everyone_
     seat = _record(harness)["asks"][0]["participants"][1]
     assert seat["status"] == "pending" and seat["unresolved"] is True
     frames = _frames(harness, "a1")
-    assert len(frames) == 1 and "custody has nothing more to deliver — 2 answered, 1 unresolved (s2: custody lost" in frames[0]
+    assert len(frames) == 1 and "custody has nothing more to deliver — 2 answered, 1 unresolved (s2: outcome unknown" in frames[0]
     assert "every participant settled" not in frames[0]
 
 

tests/test_plan_organ_scope_r1.py

@@ -89,19 +89,44 @@ def test_a_late_pre_dispatch_refusal_through_the_custody_hook_withdraws_the_disp
     assert decision["status"] == "force_plan_unmet"
 
 
-def test_an_exception_during_dispatch_leaves_an_attempted_ask_not_a_hold(harness):
+def test_an_exception_after_the_hand_off_is_unknown_never_nobody_reached(harness):
+    """Coordinator correction (03 recovery §2 D5, probe P4): the substrate already held the seats, so an
+    exception after the fan-out proves neither "no send" nor "$0". The seats stay pending and unresolved,
+    the barrier's dispatched fact stands (force_plan met, nothing holds under Blocking), nothing is counted
+    unroutable, the read view / progress line / disclosure say "outcome unknown — a new ask may pay twice",
+    and a late reply still settles the seat as answered through the custody hook."""
     import ouroboros.review_substrate as rs
 
+    captured = {}
+
     def boom(request, **_kw):
-        raise RuntimeError("substrate down")
+        captured["request"] = request
+        raise RuntimeError("substrate down after hand-off")
 
     harness.monkeypatch.setattr(rs, "run_review_request", boom)
     ctx = harness.make_ctx(force_plan=True)
-    _write(ctx, ask="council")
+    text, _result = _write(ctx, ask="council")
     rows = _record(harness)["asks"][0]["participants"]
-    assert rows and all(p["status"] == "failed" and p["dispatched"] is False and "substrate down" in p["error"] for p in rows)
+    assert rows and all(p["status"] == "pending" and p["unresolved"] is True and p["dispatched"] is True
+                        and p["error"] == "dispatch outcome unknown: substrate down after hand-off" for p in rows)
     decision = oh.force_plan_decision(ctx, {}, enforcement="blocking")
-    assert decision["asks_unroutable"] == 1 and decision["allow"] is True and decision["status"] == "force_plan_unmet"
+    assert decision["asks_unroutable"] == 0 and decision["asks_dispatched"] == 1 and decision["force_plan_met"] is True
+    assert decision["allow"] is True and decision["status"] == "awaiting" and decision["unknown_outcome"] == ["a1"]
+    assert "Dispatch outcome unknown for s1, s2, s3" in text and "pay twice" in text and "reached nobody" not in text
+    assert "pending — outcome unknown (dispatch outcome unknown: substrate down after hand-off)" in text
+    assert any("dispatch outcome unknown (RuntimeError)" in line and "pay twice" in line for line in harness.progress)
+    assert "the dispatch outcome of ask(s) a1 is unknown — a new ask may pay twice" in oh.plan_review_disclosure(decision)
+    # A late reply still settles the unknown seat as answered through the custody hook — and holds.
+    _settle(ctx, harness, captured["request"], "s1", _reply("Late, but here.", objections=[("o1", "too thin")]), roster=3)
+    s1 = next(p for p in _record(harness)["asks"][0]["participants"] if p["seat_id"] == "s1")
+    assert s1["status"] == "answered" and s1["reply"]["objections"][0]["id"] == "o1" and s1["unresolved"] is False
+    assert pr._decision(ctx, _record(harness))["holding"] == ["s1:o1"]
+    # The other direction stays: every seat POSITIVELY refused before dispatch is attempted, disclosed, not held.
+    other = harness.make_ctx(task_id="task-2", force_plan=True)
+    harness.install({"s1": dict(REFUSED), "s2": dict(REFUSED), "s3": dict(REFUSED)})
+    _write(other, ask="council")
+    refused = oh.force_plan_decision(other, {}, enforcement="blocking")
+    assert refused["asks_unroutable"] == 1 and refused["status"] == "force_plan_unmet" and refused["allow"] is True
 
 
 # ---------------------------------------------------------------- F2: a late peer reply to an earlier ask reaches the next delivery