A seat refused before dispatch never counts as a reached reviewer; an all-refused ask is attempted, not held

Scope R1, F1 (Astra) + Fable 1. dispatch_ask writes dispatched=True for every seat it is
about to send; a typed $0 refusal before the paid boundary (a spent subscription window
from route health, a dead route, the owner deadline before the first send) comes back as
operation_state "not_dispatched" and left that flag standing, so asks_dispatched counted a
refusal as an ask that reached a reviewer and a Swarm force_plan read as met.
reply_facts_from_row — the one conversion both the barrier rows and a late custody
settlement pass through — now withdraws the flag with the refusal. An ask whose every seat
was refused before dispatch, or lost to a dispatch exception, counts with the recorded
no-route attempts (plan_record.unroutable_asks): asked, nobody reached — a Blocking
force_plan is disclosed, not held; a plan the mind never asked about stays held.

ouroboros/plan_record.py

@@ -353,8 +353,12 @@ def engaged(record: Mapping[str, Any]) -> bool:
 
 
 def unroutable_asks(record: Mapping[str, Any]) -> int:
-    """How many asks the mind made that no route could take (recorded attempts, nobody called)."""
-    return len(record.get("attempts") or [])
+    """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."""
+    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"]))
+    return len(record.get("attempts") or []) + refused
 
 
 # ------------------------------------------------------------------ gate
@@ -407,8 +411,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 no route could take (a recorded
-        # attempt) is disclosed, not held — outages never hold. Advisory and the owner's hurry
+        # 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
         # 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

ouroboros/tools/plan_ask.py

@@ -547,7 +547,10 @@ def reply_facts_from_row(row: Mapping[str, Any]) -> Dict[str, Any]:
     }
     state = str(row.get("operation_state") or "settled")
     if state == "not_dispatched":
-        facts.update(status="not_dispatched", not_dispatched_reason=str(row.get("error") or "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
+        # withdrawn here — the one conversion both the barrier rows and a late settlement pass through.
+        facts.update(status="not_dispatched", not_dispatched_reason=str(row.get("error") or "not dispatched"), dispatched=False)
     elif row.get("late_result_pending") or state in {"pending_dispatch", "in_flight", "custody_lost"}:
         facts.update(status="pending", unresolved=state == "custody_lost")
     elif row.get("error"):

tests/test_owner_hurry_s3.py

@@ -509,7 +509,7 @@ def _plan_record(kind: str) -> dict:
     ("kind", "hurry_allows"),
     [
         ("held", True), ("awaiting", True),        # locally advisory / proceed
-        ("unsent", True),                          # a plan that reached nobody: unmet, disclosed, released by the hurry
+        ("unsent", True),                          # an ask that reached nobody: attempted, disclosed, never held (scope R1 F1)
         ("absent", False),                         # no plan at all: the Swarm admitted the task to write one — held
     ],
 )
@@ -534,12 +534,13 @@ def test_enforce_swarm_actions_task_local_advisory_matrix(tmp_path, monkeypatch,
         # Attribution rides the decision for the task detail.
         assert decision["owner_hurry_local_advisory"] is True
         assert decision["configured_enforcement"] == "blocking"
-    # WITHOUT the latch the same blocking install holds the held and unmet states; an
-    # ask whose answers are merely awaited never held (silence holds nothing).
+    # WITHOUT the latch the same blocking install holds the held state and the plan never
+    # written; an ask whose answers are merely awaited never held (silence holds nothing), and
+    # an ask every seat refused before dispatch is attempted — disclosed, not held (outages never hold).
     unlatched_ctx = _acceptance_ctx(tmp_path, latched=False)
     unlatched_ctx.task_metadata = {"force_plan": True}
     held_unlatched, _t = _swarm_held(unlatched_ctx)
-    assert held_unlatched is (kind != "awaiting")
+    assert held_unlatched is (kind in ("held", "absent"))
     # Durable review state was never touched (the monkeypatched loader is the
     # only read; nothing wrote a plan record).
     assert not (tmp_path / "task_results").exists() or "plan_record" not in json.dumps(

tests/test_plan_organ_scope_r1.py

@@ -6,9 +6,10 @@ Pure record rules live in ``tests/test_plan_record.py``; the web panel in
 from __future__ import annotations
 
 from ouroboros import owner_hurry as oh
+from ouroboros import plan_record
 from ouroboros.review_records import ReviewActorRecord
-from ouroboros.tools import plan_ask, plan_review as pr
-from tests.test_plan_task_tool import CLEAN, _call, _record, _reply, _slots, _user_text, _write, harness as _harness
+from ouroboros.tools import plan_ask
+from tests.test_plan_task_tool import CLEAN, _record, _reply, _slots, _user_text, _write, harness as _harness
 
 harness = _harness  # noqa: F811 - pytest fixture re-export
 
@@ -40,3 +41,62 @@ def test_a_reply_behind_a_quoted_done_when_row_keeps_its_objection_and_holds(har
     s1 = next(p for p in _record(harness)["asks"][0]["participants"] if p["seat_id"] == "s1")
     assert s1["status"] == "answered" and s1["reply"]["prose"] == "Too thin." and s1["reply"]["objections"] == [{"id": "o1", "text": "no story arc"}]
     assert any("s1 answered" in line or "m/a answered — 1 objection" in line for line in harness.progress)
+
+
+# ---------------------------------------------------------------- F1: a $0 pre-dispatch refusal never counts as a reached reviewer
+
+def test_a_seat_refused_before_dispatch_is_not_reached_and_an_all_refused_ask_is_attempted_not_held(harness):
+    harness.state["slots"] = _slots(("s1", "m/a"), ("s2", "m/b"))
+    harness.install({"s1": dict(REFUSED), "s2": dict(REFUSED)})
+    ctx = harness.make_ctx(force_plan=True)
+    text, _result = _write(ctx, ask="council")
+    rows = {p["seat_id"]: p for p in _record(harness)["asks"][0]["participants"]}
+    assert all(p["status"] == "not_dispatched" and p["dispatched"] is False for p in rows.values())
+    assert rows["s1"]["failure_code"] == "subscription_window_exhausted" and "not sent" in text
+    decision = oh.force_plan_decision(ctx, {}, enforcement="blocking")
+    assert decision["asks_dispatched"] == 0 and decision["asks_unroutable"] == 1 and decision["force_plan_met"] is False
+    assert decision["status"] == "force_plan_unmet" and decision["allow"] is True and decision["holding"] == []
+    assert oh.unmet_force_plan_obligation(ctx)["unmet"] is True, "unmet (nobody reached), so a promote still carries it"
+    # The other direction: one seat refused and one reached is a reached ask; the refused seat never saw the plan.
+    other = harness.make_ctx(task_id="task-2", force_plan=True)
+    harness.install({"s1": dict(REFUSED), "s2": CLEAN})
+    _write(other, ask="council")
+    met = oh.force_plan_decision(other, {}, enforcement="blocking")
+    assert met["asks_dispatched"] == 1 and met["force_plan_met"] is True and met["asks_unroutable"] == 0
+    views = plan_record.participant_views(_record(harness, "task-2"))
+    assert views["s1"]["last_seen_revision"] is None and views["s2"]["last_seen_revision"] == 1
+
+
+def test_a_late_pre_dispatch_refusal_through_the_custody_hook_withdraws_the_dispatch_flag_too(harness):
+    harness.state["slots"] = _slots(("s1", "m/a"), ("s2", "m/b"))
+    sub = harness.install({"s1": dict(PENDING), "s2": dict(PENDING)})
+    ctx = harness.make_ctx(force_plan=True)
+    _write(ctx, ask="council")
+    assert oh.force_plan_decision(ctx, {}, enforcement="blocking")["asks_dispatched"] == 1, "pending seats were sent"
+    request = sub.calls[0]["request"]
+    for seat in ("s1", "s2"):
+        slot = next(s for s in harness.state["slots"] if s.slot_id == seat)
+        actor = ReviewActorRecord(slot_id=seat, model=slot.model, status="not_dispatched", raw_text="", error=REFUSED["error"],
+                                  operation_id=f"op-{seat}-refused", operation_state="not_dispatched",
+                                  failure_code="subscription_window_exhausted", reset_at=REFUSED["reset_at"])
+        plan_ask.settle_released_reply(ctx, request=request, task_id="task-1", actor=actor, slot=slot, settled_wave={}, roster_size=2)
+        decision = oh.force_plan_decision(ctx, {}, enforcement="blocking")
+        row = next(p for p in _record(harness)["asks"][0]["participants"] if p["seat_id"] == seat)
+        assert row["status"] == "not_dispatched" and row["dispatched"] is False and row["reset_at"] == REFUSED["reset_at"]
+    assert decision["asks_dispatched"] == 0 and decision["asks_unroutable"] == 1 and decision["allow"] is True
+    assert decision["status"] == "force_plan_unmet"
+
+
+def test_an_exception_during_dispatch_leaves_an_attempted_ask_not_a_hold(harness):
+    import ouroboros.review_substrate as rs
+
+    def boom(request, **_kw):
+        raise RuntimeError("substrate down")
+
+    harness.monkeypatch.setattr(rs, "run_review_request", boom)
+    ctx = harness.make_ctx(force_plan=True)
+    _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)
+    decision = oh.force_plan_decision(ctx, {}, enforcement="blocking")
+    assert decision["asks_unroutable"] == 1 and decision["allow"] is True and decision["status"] == "force_plan_unmet"