fix: answered blocks resume flip daemons — verdict delivery + affine wake

The live regression (clanker-test.db): a task parked on a question sat
blocked forever after the human answered — no wake, no ask, no
re-allocation. Root cause chain, all three suspects interacting:

- the wake target check required poll_state's SLEEPING signature (last
  ask returned 204), but a daemon parked mid-round holds its LLM round
  open in the block's answer wait and its idle loop cannot ask while
  the session is live (session_busy parks it; wakes are swallowed), so
  its last ask stays the allocating 200 — the row reads idle while
  poll_state says awake, and the wake never fires. The fallback ("the
  affine saddle's own idle-awake ask") is unreachable for such a
  daemon. The parked branch now targets any idle saddle that has ever
  asked; the offerable set itself already included
  blocked-with-answered tasks.
- phase 2's poller teardown deleted phase 1's only working channel for
  those mid-session parks (a daemon state==blocked got its verdict via
  POST /answer) without a replacement: the real daemon cannot ask while
  blocked, so wake->ask->affinity-claim alone cannot resume it. The
  dispatcher gains an answer-delivery pass (before the wake pass):
  blocked/review-requested tasks whose newest block carries an
  undelivered verdict, with the open session on an idle-awake saddle
  (question park) or a busy row still running the task (review park —
  saddle review blocks the round without a parking state push), get
  POST /answer {answer, branch_advanced}; 200 marks the verdict
  delivered and conditionally flips task/saddle back to working/busy;
  409 stays undelivered for the affinity allocation (a restarted
  daemon asks on its own); transport failure is the normal offline
  bookkeeping. Sleeping saddles are left to the wake pass — their ask
  allocates the resume with the answers riding the assignment.
- answer_block correctly keeps the task parked (the resume must carry
  the verdict); the missing transition back is the delivery/allocation
  above.

The offline requeue paths (tick pass + mark_saddle_unreachable) no
longer touch parked-with-answered tasks: the verdict and the open
session survive for the resume instead of being requeued away.

Harness: the fake daemon serves POST /answer (200 into the parked
answer wait, 409 when nothing waits, resumed round continues after the
response — the real daemon's contract). New scenarios: the mandated
sleeping sequence (ask -> assignment -> blocked push -> ask -> 204 ->
sleeps -> answer -> WAKE -> ask -> resume carries the verdict -> same
session continues), the live parked-mid-round signature (no wake, no
re-ask; the verdict is delivered and the SAME session completes —
fails on the pre-fix tree), and offline-requeue preservation for both
park shapes.

src/Services/AllocationService.vala

@@ -116,7 +116,7 @@ namespace ClankerSaddle.Services {
             var rows = new Vector<OfferableTask>();
             foreach (var status in new string[] { STATUS_BLOCKED, STATUS_REVIEW_REQUESTED }) {
                 foreach (var task in yield tasks.list_tasks(status, null)) {
-                    if (yield newest_block_answered(task.id)) {
+                    if (yield tasks.newest_block_answered(task.id)) {
                         var parked = yield parked_offerable(task, false);
                         if (parked != null) {
                             rows.add((!)parked);
@@ -182,15 +182,6 @@ namespace ClankerSaddle.Services {
             };
         }
 
-        /** True when the task's newest block is answered (resume-ready). */
-        private async bool newest_block_answered(int64 task_id) throws Error {
-            TaskBlock? newest = null;
-            foreach (var block in yield tasks.list_blocks(task_id)) {
-                newest = block;
-            }
-            return newest != null && ((!)newest).answer != null;
-        }
-
         /**
          * The single-transaction claim. `BEGIN IMMEDIATE` takes the
          * SQLite write lock up front; the section is fully synchronous

src/Services/Dispatcher.vala

@@ -85,8 +85,11 @@ namespace ClankerSaddle.Services {
 
         /**
           * One dispatch cycle: the lost-wake guard offlines saddles whose
-          * wake went unanswered, then the wake computation offers every
-          * offerable task to a capable sleeping daemon (a parked task's
+          * wake went unanswered, then answered verdicts are delivered to
+          * daemons still holding their parked block open (POST /answer —
+          * those daemons cannot ask mid-session), then the wake
+          * computation offers every remaining offerable task to a
+          * capable sleeping or idle-awake ask daemon (a parked task's
           * answered block or terminal subtask event resumes it through
           * the affinity allocation — no push paths remain). Pending
           * subtask events and human comments are delivered to RUNNING
@@ -110,6 +113,7 @@ namespace ClankerSaddle.Services {
             } else {
                 warned_no_llm = false;
                 yield enforce_lost_wake_guard(saddles, tasks);
+                yield saddles.deliver_answered_verdicts(tasks);
                 yield wake_sleeping_saddles(saddles, capabilities, allocation);
             }
 
@@ -167,11 +171,17 @@ namespace ClankerSaddle.Services {
         /**
          * The wake computation: for every offerable task (see
          * {@link AllocationService.offerable_tasks}) find a capable
-         * SLEEPING saddle and POST /wake — the daemon asks and the
+         * ask-capable saddle and POST /wake — the daemon asks and the
          * allocation transaction hands the task over. Parked tasks
          * target their affine saddle first (only it can be allocated
-         * them). Saddles with a pending wake inside the guard window
-         * are skipped — the guard decides their fate.
+         * them): any idle saddle that has ever asked is a valid target —
+         * a SLEEPING one (its last ask returned 204) wakes to ask, and
+         * an idle-awake one re-asks; requiring the 204 signature here
+         * used to strand parked tasks whose saddle never asked again
+         * after claiming them (the answer-delivery pass above handles
+         * daemons still holding the block open, which cannot ask).
+         * Saddles with a pending wake inside the guard window are
+         * skipped — the guard decides their fate.
          */
         private async void wake_sleeping_saddles(SaddleService saddles,
                 CapabilityService capabilities, AllocationService allocation) throws Error {
@@ -181,7 +191,7 @@ namespace ClankerSaddle.Services {
                     var affine = offerable.affine_saddle_id > 0
                         ? yield saddles.get_saddle(offerable.affine_saddle_id) : null;
                     if (affine != null && ((!)affine).status == "idle"
-                            && poll_state.is_sleeping(((!)affine).id)
+                            && poll_state.has_ever_asked(((!)affine).id)
                             && !poll_state.wake_pending(((!)affine).id, lost_wake_guard_ms())) {
                         target = (!)affine;
                     }

src/Services/SaddleService.vala

@@ -447,6 +447,114 @@ namespace ClankerSaddle.Services {
             }
         }
 
+        /**
+         * The dispatcher's answer-delivery pass: a parked task whose
+         * newest block carries an UNDELIVERED human verdict gets the
+         * verdict pushed straight into the daemon's live answer loop —
+         * `POST /answer {answer, branch_advanced}` — and the session
+         * simply continues (phase 1's blocked-daemon routing, restored).
+         *
+         * Why this exists alongside the wake: a daemon parked on a
+         * question or review holds its LLM round open in the block's
+         * answer wait; its idle loop cannot ask while the session is
+         * live, so the saddle's last ask is still the one that claimed
+         * the task — it never reads as SLEEPING and no wake can serve
+         * it. Two affine row shapes are delivered to: an idle row whose
+         * saddle is awake (the question park), and a busy row still
+         * running this very task (a review park — `saddle review`
+         * blocks the round without a parking state push). A genuinely
+         * sleeping saddle (its last ask returned 204, so nothing is
+         * parked mid-session) is skipped: the wake pass owns that
+         * resume, and the daemon's ask allocates it with the answers
+         * riding the assignment. A 409 ("no block is waiting") leaves
+         * the answer undelivered on purpose — a restarted daemon asks
+         * on its own, and the affinity allocation then carries the
+         * verdict; a transport failure is the normal unreachable
+         * bookkeeping (the task keeps its parked session for the
+         * resume).
+         */
+        public async void deliver_answered_verdicts(TaskService tasks) throws Error {
+            foreach (var status in new string[] { STATUS_BLOCKED, STATUS_REVIEW_REQUESTED }) {
+                foreach (var task in yield tasks.list_tasks(status, null)) {
+                    var newest = yield tasks.newest_block(task.id);
+                    if (newest == null || ((!)newest).answer == null
+                            || poll_state.is_delivered(((!)newest).id)) {
+                        continue;
+                    }
+                    var session = yield tasks.open_session_for_task(task.id);
+                    if (session == null) {
+                        continue;
+                    }
+                    var saddle = yield get_saddle(((!)session).saddle_id);
+                    if (saddle == null) {
+                        continue;
+                    }
+                    bool idle_awake = ((!)saddle).status == "idle"
+                        && !poll_state.is_sleeping(((!)saddle).id);
+                    bool busy_on_this_task = ((!)saddle).status == "busy"
+                        && ((!)saddle).current_task_id == task.id;
+                    if (!idle_awake && !busy_on_this_task) {
+                        continue;
+                    }
+
+                    int64 task_id = task.id;
+                    int64 block_id = ((!)newest).id;
+                    try {
+                        bool advanced = yield tasks.branch_advanced_since(task_id);
+                        var payload = new Json.Builder();
+                        payload.begin_object();
+                        payload.set_member_name("answer");
+                        payload.add_string_value(raw_answer_for_saddle(((!)newest).answer ?? ""));
+                        payload.set_member_name("branch_advanced");
+                        payload.add_boolean_value(advanced);
+                        payload.end_object();
+                        Json.Node? root = null;
+                        int status_code = yield request_saddle(((!)saddle).host, "POST",
+                            "/answer", (owned) payload, out root);
+                        if (status_code == 200) {
+                            poll_state.mark_delivered(block_id);
+                            // The daemon is working the task again. Both
+                            // writes are conditional so pushes that
+                            // already landed (a fast resumed round, a
+                            // completion) are never stomped, and a
+                            // concurrent ask claim on the idle row is
+                            // never overwritten.
+                            var flip = connection.create_command(
+                                "UPDATE tasks SET status = 'working' WHERE id = :task"
+                                + " AND status IN ('blocked', 'review_requested')");
+                            flip.with_parameter<Invercargill.Element>("task", elem<int64?>(task_id));
+                            int flipped = yield flip.execute_non_query_async();
+                            var command = connection.create_command(
+                                "UPDATE saddles SET status = 'busy', current_task_id = :task,"
+                                + " allocated_at = :allocated WHERE id = :id AND current_task_id = 0"
+                                + " AND :task IN (SELECT id FROM tasks WHERE status IN"
+                                + " ('blocked', 'review_requested', 'working', 'assigned'))");
+                            command.with_parameter<Invercargill.Element>("task", elem<int64?>(task_id));
+                            command.with_parameter<Invercargill.Element>("allocated",
+                                elem<DateTime?>(new DateTime.now_local()));
+                            command.with_parameter<Invercargill.Element>("id",
+                                elem<int64?>(((!)saddle).id));
+                            yield command.execute_non_query_async();
+                            if (flipped > 0) {
+                                yield pulse.bump(task_id, "verdict delivered; task working");
+                                tasks.note_requeued_subtask(task);
+                            }
+                            print(@"[ClankerSaddle] Delivered verdict for task $(task_id) block $(block_id)"
+                                + @" to saddle '$(((!)saddle).name)' — the parked session continues\n");
+                        } else if (status_code != 409) {
+                            warning(@"[ClankerSaddle] Answer delivery to saddle $(((!)saddle).id)"
+                                + @" returned HTTP $(status_code); retrying next tick");
+                        }
+                    } catch (Error e) {
+                        warning(@"[ClankerSaddle] Answer delivery to saddle $(((!)saddle).id)"
+                            + @" failed: $(e.message); marking offline");
+                        yield mark_saddle_unreachable(((!)saddle).id, tasks,
+                            @"answer delivery failed ($(e.message))");
+                    }
+                }
+            }
+        }
+
         /**
          * Cancelling / requeueing / stopping a task with a live session
          * reaches its daemon (wire contract): `POST /task-update
@@ -546,6 +654,11 @@ namespace ClankerSaddle.Services {
                     if (session == null) {
                         continue;
                     }
+                    if (yield tasks.newest_block_answered(task.id)) {
+                        // Parked-with-answered: a verdict awaits the
+                        // resume machinery — never requeue it.
+                        continue;
+                    }
                     var saddle = yield get_saddle(((!)session).saddle_id);
                     if (saddle == null || ((!)saddle).status != "offline") {
                         continue;
@@ -606,31 +719,44 @@ namespace ClankerSaddle.Services {
             int64 held_task_at_entry = ((!)saddle).current_task_id;
             if (((!)saddle).current_task_id > 0) {
                 int64 task_id = ((!)saddle).current_task_id;
+                bool keep_parked = false;
                 var task = yield db.query<TaskEntity>()
                     .where(expr("id == $0", elem<int64?>(task_id)))
                     .first_async();
                 if (task != null && (((!)task).status == STATUS_ASSIGNED
                         || ((!)task).status == STATUS_WORKING || ((!)task).status == STATUS_BLOCKED
                         || ((!)task).status == STATUS_REVIEW_REQUESTED)) {
-                    ((!)task).status = STATUS_QUEUED;
-                    yield db.update_async<TaskEntity>((!)task);
-                    string unreachable_text = reason != null && reason.length > 0
-                        ? @"Saddle became unreachable ($(reason)); task requeued"
-                        : "Saddle became unreachable; task requeued";
-                    yield tasks.add_comment(task_id, "bot", unreachable_text);
-                    tasks.note_requeued_subtask(((!)task));
+                    keep_parked = yield tasks.newest_block_answered(task_id);
+                    if (keep_parked) {
+                        // Parked-with-answered: the verdict must survive
+                        // for the resume (answer delivery, or the
+                        // affinity allocation when the daemon asks
+                        // again) — the session stays open and the task
+                        // stays parked.
+                        print(@"[ClankerSaddle] Task $(task_id) keeps its parked session: its answered block awaits delivery\n");
+                    } else {
+                        ((!)task).status = STATUS_QUEUED;
+                        yield db.update_async<TaskEntity>((!)task);
+                        string unreachable_text = reason != null && reason.length > 0
+                            ? @"Saddle became unreachable ($(reason)); task requeued"
+                            : "Saddle became unreachable; task requeued";
+                        yield tasks.add_comment(task_id, "bot", unreachable_text);
+                        tasks.note_requeued_subtask(((!)task));
+                    }
                 }
 
-                var sessions = yield db.query<ClankerSaddle.Data.Session>()
-                    .where(expr("task_id == $0", elem<int64?>(task_id)))
-                    .materialise_async();
-                foreach (var session in sessions) {
-                    if (session.ended_at != null) {
-                        continue;
+                if (!keep_parked) {
+                    var sessions = yield db.query<ClankerSaddle.Data.Session>()
+                        .where(expr("task_id == $0", elem<int64?>(task_id)))
+                        .materialise_async();
+                    foreach (var session in sessions) {
+                        if (session.ended_at != null) {
+                            continue;
+                        }
+                        session.ended_at = new DateTime.now_local();
+                        session.outcome = "stopped";
+                        yield db.update_async<ClankerSaddle.Data.Session>(session);
                     }
-                    session.ended_at = new DateTime.now_local();
-                    session.outcome = "stopped";
-                    yield db.update_async<ClankerSaddle.Data.Session>(session);
                 }
             }
             var offline_write = connection.create_command(

src/Services/TaskService.vala

@@ -1224,6 +1224,27 @@ namespace ClankerSaddle.Services {
                 .materialise_async();
         }
 
+        /** The task's newest block, or null when it has none. */
+        public async TaskBlock? newest_block(int64 task_id) throws Error {
+            TaskBlock? newest = null;
+            foreach (var block in yield list_blocks(task_id)) {
+                newest = block;
+            }
+            return newest;
+        }
+
+        /**
+         * True when the task's newest block exists and carries an
+         * answer — a verdict is waiting to reach the parked workspace
+         * (the answer-delivery pass, the affinity resume, or a later
+         * assignment payload). Such a task must never be requeued: the
+         * requeue closes the session the resume machinery needs.
+         */
+        public async bool newest_block_answered(int64 task_id) throws Error {
+            var newest = yield newest_block(task_id);
+            return newest != null && ((!)newest).answer != null;
+        }
+
         /** Every dependency edge in the graph (for queue annotations). */
         public async ImmutableLot<TaskDependency> list_all_dependencies() throws Error {
             return yield db.query<TaskDependency>().materialise_async();

test/web_stack_test.py

@@ -46,6 +46,13 @@ Plan 10 phase 2 (telemetry flip):
   idempotent on the daemon entry ids), blocked raises the block row +
   parks + bumps the pulse (sealed refresh shows it), done/failed
   complete/fail the task, suspended parks for the affinity resume
+- answer resumes, both daemon shapes: a SLEEPING affine saddle (its
+  post-park ask returned 204) is woken, its ask allocating the resume
+  with the verdict riding the assignment; a daemon parked MID-ROUND
+  (its round holds the block's answer wait open, so it can never ask)
+  gets the verdict delivered into that wait via POST /answer — same
+  session, no wake. A daemon dying with a parked-with-answered task
+  never requeues it: the verdict and session survive for the resume
 - staleness: a silent runner gets ONE exceptional GET /status probe —
   alive keeps it running, dead requeues the task with a bot comment and
   marks the saddle offline; a recovered daemon's ask flips offline rows
@@ -269,11 +276,14 @@ class FakeSaddle:
     to, then PUSHES its telemetry home: raw PTY chunks via POST
     /api/saddle/chunks and round results with state transitions via
     POST /api/saddle/results (working at round start, terminal states
-    when the scripted lifecycle ends them). /output, /log and /start are
-    gone (404) like a post-phase-2 daemon; /task-update serves the
-    cancel/comment/subtask-status kinds. Every trigger can be disabled
-    per test for determinism; ask_now() performs a synchronous manual
-    ask.
+    when the scripted lifecycle ends them). /output, /log and /start
+    are gone (404) like a post-phase-2 daemon; /task-update serves the
+    cancel/comment/subtask-status kinds; POST /answer receives a
+    delivered verdict when the round is parked in a block's answer
+    wait (409 when nothing waits — the real daemon's contract; the
+    resumed round continues after the response). Every trigger can be
+    disabled per test for determinism; ask_now() performs a
+    synchronous manual ask.
     """
 
     PROMPT = "bbarrow@fedora:/workspace/task-1$ "
@@ -320,6 +330,10 @@ class FakeSaddle:
         self.serve_register = True         # False: 404 /register (pre-flip daemon)
         self.last_ask_trigger = 0.0
         self.port = None
+        # --- answer delivery (the parked-session resume) ---
+        self.pending_block = None        # the block payload this daemon waits on (state blocked)
+        self.answer_log = []             # recorded POST /answer bodies
+        self.answer_push_delay = 0.2     # the resumed round's first push, after the response
 
     def set_state(self, state, summary=None):
         """Advances the scripted lifecycle."""
@@ -391,6 +405,10 @@ class FakeSaddle:
             payload["entries"] = entries
         if block is not None:
             payload["block"] = block
+            if state == "blocked":
+                # The daemon parks its round on this block's answer wait
+                # (the real daemon's wait_for_answer).
+                self.pending_block = dict(block)
         if summary is not None:
             payload["summary"] = summary
         try:
@@ -460,6 +478,9 @@ class FakeSaddle:
         self.running_task = int(task_id or 0)
         self.running_title = task.get("title") or ""
         self.session_id = payload.get("session_id")
+        # The assignment's answers resolve any block this daemon was
+        # parked on (resume-carries-answer).
+        self.pending_block = None
         count = self.start_counts.get(task_id, 0) + 1
         self.start_counts[task_id] = count
         if count > 1:
@@ -539,6 +560,43 @@ class FakeSaddle:
                 if self.daemon.ask_on_wake:
                     self.daemon.maybe_trigger_ask("wake")
                 return
+            if self.path == "/answer":
+                # The web delivers a human verdict into the daemon's
+                # live answer wait (the pre-flip channel, and the only
+                # resume for a session parked mid-round — a blocked
+                # daemon cannot ask). 409 mirrors the real daemon when
+                # no block is pending. The resumed round continues
+                # AFTER the response goes out (the real daemon resumes
+                # its parked wait_for_answer via an idle callback).
+                try:
+                    payload = json.loads(raw)
+                except ValueError:
+                    self._send(400, {"error": "bad json"})
+                    return
+                parked = self.daemon.pending_block
+                if self.daemon.state != "blocked" or parked is None:
+                    self._send(409, {"error": "no block is waiting for an answer"})
+                    return
+                self.daemon.pending_block = None
+                self.daemon.answer_log.append({
+                    "answer": payload.get("answer", ""),
+                    "branch_advanced": payload.get("branch_advanced"),
+                    "block": dict(parked)})
+                self.daemon.set_state("working")
+                self._send(200, {"ok": True})
+                daemon_ref = self.daemon
+
+                def continue_round():
+                    time.sleep(daemon_ref.answer_push_delay)
+                    if daemon_ref.dead or daemon_ref.silent:
+                        return
+                    daemon_ref.push_results("working")
+                    daemon_ref.set_state("done", "resumed after the verdict")
+                    daemon_ref.push_results("done", summary="resumed after the verdict")
+                    if daemon_ref.ask_on_terminal:
+                        daemon_ref.maybe_trigger_ask("terminal")
+                threading.Thread(target=continue_round, daemon=True).start()
+                return
             if self.path == "/task-update":
                 # Plan 10: cancel (abort machinery), comment (injected
                 # turn) and subtask-status (live subtask set) — delivered
@@ -3083,6 +3141,217 @@ def main():
         check(wait_task_done(auth, live_task, timeout=60) is not None,
               "live-block task completes after answering")
 
+        print("== phase 2: answer resume — wake -> ask -> affinity claim ==")
+        # The full live sequence against a daemon that asks after its
+        # park: ask -> assignment -> blocked push (park) -> ask -> 204
+        # -> sleeps (row idle) -> human answers -> the dispatcher WAKES
+        # the sleeping affine saddle -> its ask allocates the resume,
+        # the verdict riding the assignment -> the session continues.
+        # A private capability tag keeps the fleet's other sleeping
+        # saddles from claiming these tasks.
+        saddles_state = slot_public(auth, "/saddles", "saddles")
+        act(auth, saddles_state["create_capability"], {"name": "answerflow"})
+        answerflow_cap = int([c["id"] for c in slot_public(auth, "/saddles", "saddles")["capabilities"]
+                              if c["name"] == "answerflow"][0])
+
+        def answerflow_daemon(**kwargs):
+            daemon = FakeSaddle()
+            daemon.ask_on_register = False   # manual, deterministic asks
+            daemon.hold_state = True         # the tests drive the lifecycle
+            for key, value in kwargs.items():
+                setattr(daemon, key, value)
+            daemon.serve()
+            return daemon
+
+        def register_answerflow(daemon, name):
+            act(auth, slot_public(auth, "/saddles", "saddles")["register"], {
+                "name": name, "host": "127.0.0.1:%d" % daemon.port})
+            act(auth, saddle_card(name)["attach_capability"],
+                {"capability_id": str(answerflow_cap)})
+
+        def create_answerflow_task(title):
+            act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
+                "project_id": "1", "title": title, "priority": "2",
+                "capability_id": [str(answerflow_cap)]})
+            return int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
+                        if t["title"] == title][0])
+
+        def retire_daemon(daemon, name):
+            # deterministic teardown: a dead daemon and an offline row
+            # never interfere with the sections that follow
+            daemon.kill()
+            act(auth, saddle_card(name)["refresh"], {})
+
+        sleeper = answerflow_daemon(ask_on_wake=True)
+        register_answerflow(sleeper, "resume-sleeper")
+        sleeping_task = create_answerflow_task("Sleeping resume task")
+        ask_status, assignment = sleeper.ask_now()
+        check(ask_status == 200 and int(assignment["task"]["id"]) == sleeping_task,
+              "the daemon's ask claims the queued task")
+        sleeping_session = assignment.get("session_id")
+        sleeper.state = "blocked"
+        check(sleeper.push_results("blocked", block={
+            "question": "Pick a side", "type": "question",
+            "options": ["Port", "Starboard"]}) == 200,
+              "the blocked push parks the session")
+        check(wait_for(lambda: task_state(auth, sleeping_task).get("status") == "blocked"
+                       or False, timeout=30, what="sleeping task parked") is not None,
+              "the task parks as blocked")
+        check(saddle_card("resume-sleeper")["status"] == "idle",
+              "the parked session's saddle row reads idle")
+        ask_status, _ = sleeper.ask_now()
+        check(ask_status == 204,
+              "the parked daemon's ask returns 204 — it sleeps until woken")
+        time.sleep(3)  # several dispatch ticks pass with the daemon asleep
+        check(len(sleeper.wake_log) == 0,
+              "the sleeping daemon is never woken while the block waits")
+        detail = task_state(auth, sleeping_task)
+        side_block = [b for b in detail["blocks"] if b["question"] == "Pick a side"][0]
+        act(auth, side_block["answer_action"], {"option": "Starboard"})
+        check(wait_for(lambda: len(sleeper.wake_log) > 0 or False,
+                       timeout=30, what="wake after the answer") is not None,
+              "answering the block wakes the sleeping affine saddle")
+        check(wait_for(lambda: any(
+            int(p.get("task", {}).get("id") or 0) == sleeping_task
+            and p.get("session_id") == sleeping_session
+            and any(a.get("answer") == "Starboard"
+                    for a in p.get("task", {}).get("answers", []))
+            for p in sleeper.start_payloads[1:]) or False,
+            timeout=30, what="resume allocation carries the verdict") is not None,
+              "the woken daemon's ask allocates the resume — same session, verdict carried")
+        # The session continues: the resumed round runs to completion on
+        # the SAME session id.
+        sleeper.push_results("done", summary="sleep-resumed round complete")
+        check(wait_task_done(auth, sleeping_task, timeout=30) is not None,
+              "the resumed session continues to completion")
+        conn = db()
+        session_rows = conn.execute(
+            "select count(*) from sessions where task_id = ?", (sleeping_task,)).fetchone()[0]
+        conn.close()
+        check(session_rows == 1,
+              "the sleeping resume kept one session throughout")
+        retire_daemon(sleeper, "resume-sleeper")
+
+        print("== phase 2: answer resume — verdict delivered to the parked session ==")
+        # The live regression's daemon shape (clanker-test.db): after the
+        # blocked push the daemon CANNOT ask again — its round is parked
+        # in the block's answer wait, so the saddle's last ask stays the
+        # one that claimed the task (never sleeping, never woken). The
+        # verdict must be DELIVERED into that answer wait.
+        parked = answerflow_daemon(ask_on_wake=False,      # a blocked daemon cannot ask on wake
+                                   ask_on_terminal=False,  # and not after the round either
+                                   ask_on_idle_status=False)
+        register_answerflow(parked, "resume-parked")
+        parked_task = create_answerflow_task("Parked resume task")
+        ask_status, assignment = parked.ask_now()
+        check(ask_status == 200 and int(assignment["task"]["id"]) == parked_task,
+              "the daemon's ask claims the task (its last ask)")
+        parked_session = assignment.get("session_id")
+        parked.state = "blocked"
+        check(parked.push_results("blocked", block={
+            "question": "Which database?", "type": "question",
+            "options": ["SQLite", "Postgres"]}) == 200,
+              "the blocked push parks the session mid-round")
+        check(wait_for(lambda: task_state(auth, parked_task).get("status") == "blocked"
+                       or False, timeout=30, what="parked task blocked") is not None,
+              "the task parks as blocked")
+        check(saddle_card("resume-parked")["status"] == "idle",
+              "the parked session's saddle row reads idle")
+        time.sleep(3)  # dispatch ticks pass: no wake, no ask ever reaches it
+        asks_at_answer = len(parked.asks)
+        check(len(parked.wake_log) == 0 and asks_at_answer == 1,
+              "the parked daemon is never woken and never asks again (the live signature)")
+        detail = task_state(auth, parked_task)
+        db_block = [b for b in detail["blocks"] if b["question"] == "Which database?"][0]
+        act(auth, db_block["answer_action"], {"option": "SQLite"})
+        check(wait_for(lambda: len(parked.answer_log) > 0 or False,
+                       timeout=30, what="verdict delivered") is not None,
+              "the verdict is delivered into the parked daemon's answer wait")
+        check(parked.answer_log[0]["answer"] == "SQLite",
+              "the delivered verdict is the human's choice")
+        check(parked.answer_log[0]["branch_advanced"] is False,
+              "the delivery carries branch_advanced=false with no siblings")
+        check(len(parked.wake_log) == 0 and len(parked.asks) == asks_at_answer,
+              "no wake, no ask — the delivery itself resumed the session")
+        check(wait_task_done(auth, parked_task, timeout=30) is not None,
+              "the parked session continues to completion")
+        conn = db()
+        row = conn.execute("select outcome from sessions where id = ?",
+                           (parked_session,)).fetchone()
+        session_rows = conn.execute(
+            "select count(*) from sessions where task_id = ?", (parked_task,)).fetchone()[0]
+        conn.close()
+        check(row is not None and row[0] == "done" and session_rows == 1,
+              "the delivered verdict resumed the SAME session to done")
+        retire_daemon(parked, "resume-parked")
+
+        print("== phase 2: offline requeue keeps parked-with-answered tasks ==")
+        # A daemon dying while its parked task carries an answered
+        # verdict: going offline must not requeue the task or close its
+        # session — the resume machinery (delivery once the daemon
+        # returns, or the affinity allocation on its next ask) needs
+        # both. The question park leaves the row idle; the review park
+        # keeps it busy@task (`saddle review` blocks the round without
+        # a parking state push), exercising the unreachable path's
+        # requeue decision both ways.
+        stranded = answerflow_daemon()
+        register_answerflow(stranded, "resume-stranded")
+        stranded_task = create_answerflow_task("Stranded answered task")
+        ask_status, _ = stranded.ask_now()
+        check(ask_status == 200, "the stranded daemon claims its task")
+        stranded.state = "blocked"
+        stranded.push_results("blocked", block={
+            "question": "Hold or fold?", "type": "question", "options": ["Hold", "Fold"]})
+        check(wait_for(lambda: task_state(auth, stranded_task).get("status") == "blocked"
+                       or False, timeout=30, what="stranded task parked") is not None,
+              "the stranded task parks as blocked")
+        detail = task_state(auth, stranded_task)
+        hold_block = [b for b in detail["blocks"] if b["question"] == "Hold or fold?"][0]
+        act(auth, hold_block["answer_action"], {"option": "Hold"})
+        stranded.kill()  # the daemon dies before the verdict is delivered
+        act(auth, saddle_card("resume-stranded")["refresh"], {})
+        check(wait_for(lambda: saddle_card("resume-stranded")["status"] == "offline"
+                       or False, timeout=30, what="stranded saddle offline") is not None,
+              "the dead daemon's saddle row goes offline")
+        time.sleep(7)  # dispatch ticks run their offline passes
+        check(task_state(auth, stranded_task)["status"] == "blocked",
+              "the parked-with-answered task is not requeued by the offline pass")
+        conn = db()
+        open_session = conn.execute(
+            "select count(*) from sessions where task_id = ? and ended_at is null",
+            (stranded_task,)).fetchone()[0]
+        conn.close()
+        check(open_session == 1,
+              "the parked-with-answered task keeps its open session for the resume")
+
+        reviewer = answerflow_daemon(ask_on_terminal=False)
+        register_answerflow(reviewer, "resume-reviewer")
+        review_task = create_answerflow_task("Review park task")
+        ask_status, _ = reviewer.ask_now()
+        check(ask_status == 200, "the reviewer claims its task")
+        reviewer.queue_output("review", "saddle review", "Ready for review: wired and tested.")
+        check(wait_for(lambda: task_state(auth, review_task).get("status") == "review_requested"
+                       or False, timeout=30, what="review park raised") is not None,
+              "the review entry parks the task as review_requested (row stays busy)")
+        reviewer.kill()  # the daemon dies before the verdict can be delivered
+        detail = task_state(auth, review_task)
+        verdict_block = [b for b in detail["blocks"]
+                         if b["block_type"] == "review"][0]
+        act(auth, verdict_block["answer_action"], {"answer": "Human approved the review"})
+        check(wait_for(lambda: saddle_card("resume-reviewer")["status"] == "offline"
+                       or False, timeout=30, what="reviewer offline") is not None,
+              "the failed verdict delivery drives the dead reviewer offline")
+        time.sleep(7)  # dispatch ticks run their offline passes
+        check(task_state(auth, review_task)["status"] == "review_requested",
+              "the review-parked task with an answered verdict is never requeued")
+        conn = db()
+        open_session = conn.execute(
+            "select count(*) from sessions where task_id = ? and ended_at is null",
+            (review_task,)).fetchone()[0]
+        conn.close()
+        check(open_session == 1,
+              "the review-parked task keeps its open session for the resume")
+
         print("== phase 2: staleness — one probe decides ==")
         # A silent-but-alive runner (long think, nothing to flush): after
         # SADDLE_STALE_S (3s here) of silence it owes ONE exceptional