Park/answer-wake fix: allocation guard, busy-answer delivery, batched verdicts

Goal 28 task F. The task-187 race: a park's transition push had not landed when the freed daemon asked again, so allocation handed the parked task back as a crash restore with its question gone; a later answer found the affine saddle busy and had no delivery path. F1: the claim refuses a working/assigned row whose round session just closed parked (park in flight), the parked arm requires the newest block answered, and the crash-recovery sweep holds a round whose last push is fresh (PARK_SETTLE_GRACE_S). F2: an answered block whose task holds a RUNNING session rides the new AnswerDeliveryHub singleton; the dispatcher tick injects it into the running round through the daemon's task-update channel (kind answer), delivery-once via the poll-state ledger; sessions.answers_injected counts injections. F3: the resume notice composes EVERY block answered since the batched-verdict watermark (sessions.resume_since, M0008), oldest first; an already-injected verdict is summarised, never repeated. F6: an assignment over an unanswered newest block flags pending_question. Migrations: M0008_AnswerWake adds sessions.resume_since and answers_injected. Tests: the answerwake bundle (35 checks) replays the full task-187 arc end-to-end against the real allocation.

meson.build

@@ -207,6 +207,7 @@ app_sources = files(
     'src/Data/Migrations/M0005_Discoveries.vala',
     'src/Data/Migrations/M0006_Advisor.vala',
     'src/Data/Migrations/M0007_ParkingGuard.vala',
+    'src/Data/Migrations/M0008_AnswerWake.vala',
     'src/Services/ProjectService.vala',
     'src/Services/TaskService.vala',
     'src/Services/GoalService.vala',
@@ -214,6 +215,7 @@ app_sources = files(
     'src/Services/EventService.vala',
     'src/Services/LifecycleService.vala',
     'src/Services/SaddlePollState.vala',
+    'src/Services/AnswerDeliveryHub.vala',
     'src/Services/AllocationService.vala',
     'src/Services/StatePulse.vala',
     'src/Services/Dispatcher.vala',

src/Data/Entities.vala

@@ -333,9 +333,17 @@ namespace ClankerSaddle.Data {
         /**
          * What a resumed session's agent first received (plan 14 B4):
          * the resume_notice the allocation built, stored on the row at
-         * resume-allocation; null for fresh sessions.
+         * resume-allocation; null for fresh sessions. Goal 28 task F:
+         * `resume_since` records the batched-verdict watermark (the
+         * prior round's session start) so the FINAL resume can
+         * re-compose everything still undelivered, and
+         * `answers_injected` counts verdicts the busy-answer channel
+         * delivered mid-session (the follow-up park's notice
+         * summarises instead of repeating them).
          */
         public string? resume_notice { get; set; default = null; }
+        public DateTime? resume_since { get; set; default = null; }
+        public int64 answers_injected { get; set; default = 0; }
 
         public static void entity_mapping(EntityMapperBuilder<Session> cfg) {
             cfg.table("sessions")
@@ -345,7 +353,9 @@ namespace ClankerSaddle.Data {
                 .column<DateTime?>("started_at", o => o.started_at, (o, v) => o.started_at = v)
                 .column<DateTime?>("ended_at", o => o.ended_at, (o, v) => o.ended_at = v)
                 .column<string>("outcome", o => o.outcome, (o, v) => o.outcome = v)
-                .column<string>("resume_notice", o => o.resume_notice, (o, v) => o.resume_notice = v);
+                .column<string>("resume_notice", o => o.resume_notice, (o, v) => o.resume_notice = v)
+                .column<DateTime?>("resume_since", o => o.resume_since, (o, v) => o.resume_since = v)
+                .column<int64?>("answers_injected", o => o.answers_injected, (o, v) => o.answers_injected = v ?? 0);
         }
     }
 

src/Data/Migrations/M0008_AnswerWake.vala

@@ -0,0 +1,76 @@
+using InvercargillSql.Migrations;
+using Sqlite;
+
+namespace ClankerSaddle.Data.Migrations {
+
+    /**
+     * Goal 28 task F — the park/answer-wake fix. Two session-row
+     * columns for the answer-delivery bookkeeping:
+     *
+     * - `sessions.resume_since` — the batched-verdict watermark: the
+     *   started_at of the round that PARKED (the row a resume
+     *   succeeds). The resumed round's notice is composed from every
+     *   block answered at or after it, so a newer block can never
+     *   swallow an older verdict (the task-187 loss), and the FINAL
+     *   resume can re-compose whatever is still undelivered.
+     * - `sessions.answers_injected` — how many verdicts the
+     *   busy-answer channel delivered INTO this running session (the
+     *   mid-session task-update injections). A later park's resume
+     *   notice summarises them instead of repeating them.
+     *
+     * IDEMPOTENCE: same contract as M0002–M0007 — up() runs before
+     * the apply transaction opens (the builder only queues), so it
+     * introspects sqlite_master through a separate read-only handle
+     * and adds each column only when the schema still lacks it.
+     */
+    public class M0008_AnswerWake : Migration {
+
+        public override string name { get { return "Answer wake"; } }
+        public override uint64 serial { get { return 8; } }
+        public override string migration_namespace { get { return "clanker-saddle"; } }
+
+        private string db_path() {
+            var path = Environment.get_variable("SPRY_DB_PATH") ?? "";
+            if (path.strip().length == 0) {
+                return "./clanker-saddle.sqlite";
+            }
+            return path;
+        }
+
+        private bool has_column(string column) {
+            Database db;
+            int rc = Database.open_v2(db_path(), out db, Sqlite.OPEN_READONLY);
+            if (rc != Sqlite.OK) {
+                return false;
+            }
+            Statement statement;
+            rc = db.prepare_v2(
+                @"select 1 from pragma_table_info('sessions') where name = '$column'",
+                -1, out statement);
+            if (rc != Sqlite.OK) {
+                return false;
+            }
+            return statement.step() == Sqlite.ROW;
+        }
+
+        public override void up(MigrationBuilder b) {
+            if (!has_column("resume_since")) {
+                b.alter_table("sessions", (t) => {
+                    t.add_column<DateTime?>("resume_since");
+                });
+            }
+            if (!has_column("answers_injected")) {
+                b.alter_table("sessions", (t) => {
+                    t.add_column<int64?>("answers_injected");
+                });
+            }
+        }
+
+        public override void down(MigrationBuilder b) {
+            // SQLite cannot drop a column through the builder's
+            // alter_table path; the raw statements are the honest form.
+            b.execute_sql("alter table sessions drop column answers_injected");
+            b.execute_sql("alter table sessions drop column resume_since");
+        }
+    }
+}

src/Services/AllocationService.vala

@@ -239,6 +239,47 @@ namespace ClankerSaddle.Services {
                     return null;
                 }
 
+                // Goal 28 task F (the allocation guard): the task-187
+                // race — the park's transition push had not landed when
+                // the freed daemon asked again, so the row still read
+                // working/assigned with no block row in place — the
+                // claim handed the parked task back and the resumed
+                // session woke as a bare crash restore, its question
+                // gone. Two legs, both NARROW (a legitimately
+                // crash-recovered queued task MUST stay claimable —
+                // F6's whole point is that its unanswered question
+                // rides the re-allocation):
+                //
+                // 1. A row still reading working/assigned while its
+                //    round's session JUST closed 'parked' is a park
+                //    transition in flight (the block payload rides the
+                //    same push): the block row does not exist YET.
+                //    Hold — answer nothing (the ask 204s and the
+                //    daemon re-asks after its settle); the transition
+                //    lands within one push round-trip and the next ask
+                //    sees the parked row.
+                // 2. The PARKED resume arm (still_claimable's blocked/
+                //    review/suspended candidates) additionally requires
+                //    the newest block to carry an answer — an open
+                //    question must never be resumed onto its affine
+                //    saddle, the round is waiting on a human (that arm
+                //    has its own guard inside still_claimable).
+                if (((!)task).status == STATUS_ASSIGNED || ((!)task).status == STATUS_WORKING) {
+                    // The guarded claimable statuses never reach here
+                    // except the park-in-flight race; a straggler open
+                    // session that never parked would have been
+                    // requeued by the ask's crash-recovery pass.
+                    var parked_round = db.query<Session>()
+                        .where(expr("task_id == $0 && outcome == $1",
+                            elem<int64?>(candidate.task.id), elem(OUTCOME_PARKED)))
+                        .order_by_desc(expr("id"))
+                        .first();
+                    if (parked_round != null) {
+                        connection.execute("ROLLBACK");
+                        return null;
+                    }
+                }
+
                 var row = db.query<Saddle>()
                     .where(expr("id == $0", elem<int64?>(saddle.id)))
                     .first();
@@ -358,9 +399,28 @@ namespace ClankerSaddle.Services {
          */
         private bool still_claimable(TaskEntity task, OfferableTask candidate) throws Error {
             if (candidate.parked) {
-                return task.status == STATUS_BLOCKED
-                    || task.status == STATUS_REVIEW_REQUESTED
-                    || (candidate.suspended_resume && task.status == STATUS_SUSPENDED);
+                if (task.status != STATUS_BLOCKED
+                        && task.status != STATUS_REVIEW_REQUESTED
+                        && !(candidate.suspended_resume && task.status == STATUS_SUSPENDED)) {
+                    return false;
+                }
+                // Goal 28 task F: the parked arm only resumes a task
+                // whose questions are settled — the newest block must
+                // carry an answer. An unanswered newest block on a
+                // blocked row is an OPEN question (a human has not
+                // spoken yet); re-allocating it would wake the parked
+                // conversation with nothing to say, and its verdict
+                // would then have no delivery path (the busy saddle
+                // cannot ask). The in-transaction re-read races the
+                // raise the same way the candidate list did.
+                var newest = db.query<TaskBlock>()
+                    .where(expr("task_id == $0", elem<int64?>(task.id)))
+                    .order_by_desc(expr("id"))
+                    .first();
+                if (newest != null && ((!)newest).answer == null) {
+                    return false;
+                }
+                return true;
             }
             if (task.status != STATUS_QUEUED) {
                 return false;
@@ -453,9 +513,47 @@ namespace ClankerSaddle.Services {
                     // C1: the goal-run task's block IS the goal's block
                     // — its own answer is the whole verdict, no mirror
                     // sweep exists.
-                    var newest = yield tasks.newest_block(task.id);
-                    if (newest != null && ((!)newest).answer != null) {
-                        resume_notice = TaskService.verdict_notice_for((!)newest);
+                    // Goal 28 task F (batched verdict composition):
+                    // EVERY block answered since the parked round's
+                    // session start, oldest first — a newer block must
+                    // never swallow an older answer (the task-187 loss:
+                    // block 98's verdict died when block 99 superseded
+                    // it in the newest-only read). Answers already
+                    // delivered mid-session (the busy-answer channel)
+                    // are skipped: the running round has seen them, and
+                    // repeating a verdict as an opening turn would
+                    // confuse the restore.
+                    var parked_row = yield db.query<Session>()
+                        .where(expr("id == $0", elem<int64?>(claim.session_id)))
+                        .first_async();
+                    DateTime? since = null;
+                    Session? prior = null;
+                    foreach (var older in yield db.query<Session>()
+                            .where(expr("task_id == $0 && id < $1",
+                                elem<int64?>(task.id), elem<int64?>(claim.session_id)))
+                            .order_by_desc(expr("id"))
+                            .materialise_async()) {
+                        prior = older;
+                        break;
+                    }
+                    if (prior != null && ((!)prior).started_at != null) {
+                        since = ((!)prior).started_at;
+                    }
+                    if (parked_row != null) {
+                        ((!)parked_row).resume_since = since;
+                        yield db.update_async<Session>((!)parked_row);
+                    }
+                    var fresh = new Vector<TaskBlock>();
+                    foreach (var answered in yield tasks.blocks_answered_since(task.id, since)) {
+                        if (poll_state.is_delivered(answered.id)) {
+                            continue;
+                        }
+                        fresh.add(answered);
+                    }
+                    resume_notice = TaskService.batched_verdict_notice(
+                        fresh.to_immutable_buffer());
+                    if (resume_notice.length == 0) {
+                        resume_notice = null;
                     }
                 }
                 // Plan 14 B4: persist what the resumed agent first

src/Services/AnswerDeliveryHub.vala

@@ -0,0 +1,99 @@
+using Invercargill;
+using Invercargill.DataStructures;
+using Inversion;
+
+namespace ClankerSaddle.Services {
+
+    /**
+     * One block answer awaiting delivery to a RUNNING daemon (goal 28
+     * task F, the park/answer-wake fix): the question's round parked
+     * long ago, but a race-restarted session on the same task is
+     * working RIGHT NOW — the answer must reach it mid-session through
+     * the `POST /task-update {kind: "answer"}` injection channel, not
+     * wait for a resume that cannot happen (its session never parks
+     * on the block; the block predates it).
+     *
+     * {@link TaskService.answer_block} (request-scoped, no
+     * SaddleService dependency by design) parks these on the singleton
+     * hub; the dispatcher's tick and the answer action itself both
+     * drain them through a request-scoped SaddleService.
+     */
+    public class PendingAnswerDelivery : Object {
+        /** The answered block's id (delivery-once key). */
+        public int64 block_id { get; set; }
+        /** The block's task. */
+        public int64 task_id { get; set; }
+        /** The verdict text ({@link TaskService.verdict_notice_for}). */
+        public string notice { get; set; default = ""; }
+    }
+
+    /**
+     * Process-wide answer-delivery queue (goal 28 task F): answers
+     * whose owning saddle holds a RUNNING session on the task are
+     * injected into that session via the daemon's task-update channel
+     * — the same path human comments ride — instead of waiting for an
+     * affinity resume the busy saddle can never perform. Delivery is
+     * exactly-once per answer: the SaddlePollState delivered-answers
+     * ledger (the poll endpoint's memory) is the cross-path guard, and
+     * this queue only holds what has not been handed over yet (a
+     * restart or a failed POST re-drains; the daemon tolerates a
+     * duplicate as one extra user turn).
+     *
+     * A singleton by necessity: the answering side is request-scoped
+     * (web actions) while the draining side is the dispatcher tick.
+     */
+    public class AnswerDeliveryHub : Object {
+
+        private Vector<PendingAnswerDelivery> pending = new Vector<PendingAnswerDelivery>();
+        private Mutex lock = Mutex();
+
+        /**
+         * Parks an answer for mid-session delivery. Duplicate block
+         * ids are collapsed (an answer lands once; a re-answer of the
+         * same row replaces the queued notice — last write wins, the
+         * row is the truth).
+         */
+        public void offer(PendingAnswerDelivery delivery) {
+            @lock.lock();
+            bool replaced = false;
+            var kept = new Vector<PendingAnswerDelivery>();
+            foreach (var item in pending) {
+                if (item.block_id == delivery.block_id) {
+                    kept.add(delivery);
+                    replaced = true;
+                } else {
+                    kept.add(item);
+                }
+            }
+            if (!replaced) {
+                kept.add(delivery);
+            }
+            pending = kept;
+            @lock.unlock();
+        }
+
+        /**
+         * Takes every queued delivery (oldest first — arrival order,
+         * so a batch reads chronologically), leaving the queue empty.
+         * The drainer re-offers whatever could not be delivered.
+         */
+        public ImmutableLot<PendingAnswerDelivery> drain() {
+            @lock.lock();
+            var taken = new Vector<PendingAnswerDelivery>();
+            foreach (var item in pending) {
+                taken.add(item);
+            }
+            pending = new Vector<PendingAnswerDelivery>();
+            @lock.unlock();
+            return taken.to_immutable_buffer();
+        }
+
+        /** True when nothing is queued. */
+        public bool is_empty() {
+            @lock.lock();
+            bool empty = pending.length == 0;
+            @lock.unlock();
+            return empty;
+        }
+    }
+}

src/Services/Dispatcher.vala

@@ -30,6 +30,7 @@ namespace ClankerSaddle.Services {
         private Container container = inject<Container>();
         private SaddlePollState poll_state = inject<SaddlePollState>();
         private PushService push = inject<PushService>();
+        private AnswerDeliveryHub answer_hub = inject<AnswerDeliveryHub>();
         private bool ticking = false;
         private int64 tick_started_ms = 0;
         private bool warned_no_llm = false;
@@ -127,6 +128,17 @@ namespace ClankerSaddle.Services {
 
                 yield saddles.requeue_tasks_on_offline_saddles(tasks);
 
+                // Goal 28 task F (busy-answer delivery): answers whose
+                // owner is BUSY drain into the running session here —
+                // the injected verdict arrives as a user turn within
+                // seconds of the human's click, not at a resume that
+                // cannot happen. Failures re-queue on the hub.
+                try {
+                    yield saddles.deliver_pending_answers(tasks, answer_hub);
+                } catch (Error e) {
+                    warning("[ClankerSaddle] Answer delivery pass failed: %s", e.message);
+                }
+
                 // Web Push retry tick (plan 11 task E): anything a failed
                 // send left queued retries here (best-effort delivery).
                 yield push.flush();

src/Services/SaddlePollState.vala

@@ -122,6 +122,26 @@ namespace ClankerSaddle.Services {
             return !already;
         }
 
+        /**
+         * Goal 28 task F: seconds since `session_id`'s last
+         * chunks/results push, or int64.MAX when none was ever seen —
+         * the crash-recovery requeue's park-in-flight test (a round
+         * that pushed moments before its ask closed WAS live; its
+         * closing transition may still be queued in the daemon's
+         * outbox, so requeueing it would hand the parked task back —
+         * the task-187 race).
+         */
+        public int64 seconds_since_push(int64 session_id) {
+            state_lock.lock();
+            string stored = "";
+            bool known = push_at_ms.try_get(session_id.to_string(), out stored);
+            state_lock.unlock();
+            if (!known) {
+                return int64.MAX;
+            }
+            return (get_monotonic_time() / 1000 - int64.parse(stored)) / 1000;
+        }
+
         /** True when `block_id`'s answer was already handed over. */
         public bool is_delivered(int64 block_id) {
             state_lock.lock();

src/Services/SaddleService.vala

@@ -1075,6 +1075,17 @@ namespace ClankerSaddle.Services {
                 payload.add_string_value(resume_notice);
             }
 
+            // Goal 28 task F: `pending_question` — the task's newest
+            // block is UNANSWERED (a crashed parked task resumed
+            // through the allocation). The daemon appends the
+            // still-open notice to the restored conversation's opening
+            // turn so the agent does not barrel past its own question.
+            var newest_open = yield task_service.newest_block(task.id);
+            if (newest_open != null && ((!)newest_open).answer == null) {
+                payload.set_member_name("pending_question");
+                payload.add_boolean_value(true);
+            }
+
             // The wire's `branch_advanced` (resume pushes): whether the
             // shared branch moved while this task was parked — false or
             // absent means the daemon omits the pull instruction entirely.
@@ -1351,6 +1362,82 @@ namespace ClankerSaddle.Services {
             }
         }
 
+        /**
+         * Goal 28 task F (busy-answer delivery): the hub drainer. An
+         * answer whose task holds a RUNNING session reaches that
+         * session THROUGH the daemon's task-update channel (the same
+         * injection path human comments ride) — `POST /task-update
+         * {kind: "answer", text, question}` — because no affinity
+         * resume can deliver it: the busy saddle cannot ask, and the
+         * running round's conversation predates the block.
+         *
+         * Delivery-once: the poll-state answers ledger (the poll
+         * endpoint's memory) is the guard, so the resume notice never
+         * repeats a verdict the running round already saw. The
+         * session's `answers_injected` count increments per verdict —
+         * a later park's resume notice summarises them instead of
+         * repeating the texts. Failures (transport, HTTP) re-offer
+         * the delivery for the next tick: an answer must never be
+         * dropped, and the daemon tolerates a duplicate as one extra
+         * user turn.
+         */
+        public async void deliver_pending_answers(TaskService tasks,
+                AnswerDeliveryHub hub) throws Error {
+            foreach (var delivery in hub.drain()) {
+                var session = yield tasks.open_session_for_task(delivery.task_id);
+                if (session == null) {
+                    // No running round: the resume path owns delivery
+                    // (the verdict rides the next resuming
+                    // assignment's notice). Drop the queue entry —
+                    // the answer is safe in its block row.
+                    continue;
+                }
+                var saddle = yield get_saddle(((!)session).saddle_id);
+                if (saddle == null) {
+                    hub.offer(delivery);
+                    continue;
+                }
+                var payload = new Json.Builder();
+                payload.begin_object();
+                payload.set_member_name("kind");
+                payload.add_string_value("answer");
+                payload.set_member_name("text");
+                payload.add_string_value(delivery.notice);
+                payload.set_member_name("task_id");
+                payload.add_int_value(delivery.task_id);
+                payload.end_object();
+                Json.Node? root = null;
+                bool handed = false;
+                try {
+                    int status_code = yield request_saddle(((!)saddle).host, "POST",
+                        "/task-update", (owned) payload, out root);
+                    // 404/405: a daemon without the answer kind keeps
+                    // the delivery pending — re-offer for the next
+                    // tick (an upgrade may land under us). Everything
+                    // else that fails re-offers too: never drop an
+                    // answer.
+                    handed = status_code == 200 || status_code == 204;
+                    if (!handed) {
+                        warning("[ClankerSaddle] Answer task-update to saddle $(((!)saddle).id) returned HTTP %d",
+                            status_code);
+                        hub.offer(delivery);
+                        continue;
+                    }
+                } catch (Error e) {
+                    warning("[ClankerSaddle] Answer task-update to saddle $(((!)saddle).id) failed: %s",
+                        e.message);
+                    hub.offer(delivery);
+                    continue;
+                }
+                poll_state.mark_delivered(delivery.block_id);
+                ((!)session).answers_injected += 1;
+                yield db.update_async<ClankerSaddle.Data.Session>((!)session);
+                yield task_service.add_system_comment(delivery.task_id,
+                    @"verdict delivered into the running session (block #$(delivery.block_id))");
+                yield pulse.bump(delivery.task_id, "answer delivered mid-session");
+            }
+        }
+
         /**
          * The goal id when `task_id` is a goal agent's run task (kind
          * 'goal_run'), else 0 — the discriminator for the goal-context

src/Services/TaskService.vala

@@ -144,6 +144,12 @@ namespace ClankerSaddle.Services {
         private StatePulse pulse = inject<StatePulse>();
         private PushService push = inject<PushService>();
         private EventService events = inject<EventService>();
+        private SaddlePollState poll_state = inject<SaddlePollState>();
+        // Goal 28 task F: answers whose owner is BUSY ride the
+        // singleton delivery hub (this service is request-scoped and
+        // deliberately holds no SaddleService dependency — the
+        // dispatcher's tick drains the queue through one).
+        private AnswerDeliveryHub answer_hub = inject<AnswerDeliveryHub>();
 
         /**
          * Inserts a task (status 'draft' until its creator enqueues
@@ -971,6 +977,26 @@ namespace ClankerSaddle.Services {
                     yield db.update_async<TaskEntity>((!)task);
                 }
             }
+            // Goal 28 task F (busy-answer delivery): when the task's
+            // affine saddle holds a RUNNING session right now (the
+            // race-restarted round, or a crash-restore that adopted
+            // the parked workspace), NO affinity resume can ever
+            // deliver this verdict — the busy saddle cannot ask. Park
+            // it on the singleton hub instead: the dispatcher's tick
+            // injects it into the running session through the
+            // daemon's task-update channel ({@link
+            // SaddleService.deliver_pending_answers}). A task with no
+            // open session keeps the resume path (the verdict rides
+            // the next resuming assignment's resume_notice, as
+            // before).
+            var running = yield open_session_for_task(((!)block).task_id);
+            if (running != null) {
+                answer_hub.offer(new PendingAnswerDelivery() {
+                    block_id = block_id,
+                    task_id = ((!)block).task_id,
+                    notice = verdict_notice_for((!)block)
+                });
+            }
             yield pulse.bump(((!)block).task_id, "block answered");
         }
 
@@ -1487,6 +1513,14 @@ namespace ClankerSaddle.Services {
          */
         public const int64 ASK_RECOVERY_GRACE_US = 120 * 1000000L;
 
+        /**
+         * Goal 28 task F: how long a round's last push keeps its
+         * session un-requeueable after the daemon asks — the window
+         * in which a park's transition may still be in flight
+         * (matches the daemon-side park settle).
+         */
+        public const int64 PARK_SETTLE_GRACE_S = 5;
+
         public async int requeue_interrupted_sessions(Saddle saddle) throws Error {
             var requeued = 0;
             foreach (var session in yield db.query<Session>()
@@ -1501,6 +1535,18 @@ namespace ClankerSaddle.Services {
                         < ASK_RECOVERY_GRACE_US) {
                     continue;
                 }
+                // Goal 28 task F (the allocation guard's requeue leg):
+                // a round that pushed chunks/results moments ago was
+                // LIVE — its closing transition (a park carrying the
+                // block payload) may still be queued in the daemon's
+                // outbox. Requeueing here handed the parked task back
+                // as a crash restore with its question gone (the
+                // task-187 race); hold instead — the ask 204s, the
+                // transition lands, the next ask sees the parked row.
+                if (poll_state.seconds_since_push(session.id)
+                        < PARK_SETTLE_GRACE_S) {
+                    continue;
+                }
                 var task = yield db.query<TaskEntity>()
                     .where(expr("id == $0", elem<int64?>(session.task_id)))
                     .first_async();
@@ -1641,6 +1687,52 @@ namespace ClankerSaddle.Services {
             return newest != null && ((!)newest).answer != null;
         }
 
+        /**
+         * Goal 28 task F (batched verdict composition): every block
+         * ANSWERED since `since` (the resumed session's start, or the
+         * suspension stamp), oldest first — the resume notice joins
+         * their verdict texts so a newer block can never swallow an
+         * older answer. Undelivered-only filtering is the caller's
+         * (the poll-state ledger decides what the agent has already
+         * seen mid-session).
+         */
+        public async ImmutableLot<TaskBlock> blocks_answered_since(int64 task_id,
+                DateTime? since) throws Error {
+            var rows = new Vector<TaskBlock>();
+            // list_blocks is newest-first; walk it backwards so the
+            // batch reads oldest-first.
+            var newest_first = yield list_blocks(task_id);
+            var reversed = new Vector<TaskBlock>();
+            foreach (var block in newest_first) {
+                reversed.add(block);
+            }
+            for (int i = (int) reversed.length - 1; i >= 0; i--) {
+                var block = reversed[i];
+                if (block.answer == null || block.answered_at == null) {
+                    continue;
+                }
+                if (since != null && block.answered_at.compare(((!)since)) < 0) {
+                    continue;
+                }
+                rows.add(block);
+            }
+            return rows.to_immutable_buffer();
+        }
+
+        /**
+         * The joined verdict text for a batch of answered blocks
+         * (oldest first — {@link blocks_answered_since}): each
+         * block's {@link verdict_notice_for} wording on its own
+         * paragraph, in answer order. An empty batch yields "".
+         */
+        public static string batched_verdict_notice(ImmutableLot<TaskBlock> blocks) {
+            var parts = new Vector<string>();
+            foreach (var block in blocks) {
+                parts.add(verdict_notice_for(block));
+            }
+            return string.joinv("\n\n", parts.to_array());
+        }
+
         /**
          * The session still open for `task_id` (no `ended_at`), or null.
          * The null check runs in Vala: the expression layer cannot compare

src/main.vala

@@ -49,6 +49,7 @@ int main(string[] args) {
         db.add_migration<ClankerSaddle.Data.Migrations.M0005_Discoveries>();
         db.add_migration<ClankerSaddle.Data.Migrations.M0006_Advisor>();
         db.add_migration<ClankerSaddle.Data.Migrations.M0007_ParkingGuard>();
+        db.add_migration<ClankerSaddle.Data.Migrations.M0008_AnswerWake>();
         db.add_entity<Project>(Project.entity_mapping);
         db.add_entity<ProjectLink>(ProjectLink.entity_mapping);
         db.add_entity<TaskEntity>(TaskEntity.entity_mapping);
@@ -94,6 +95,7 @@ int main(string[] args) {
         application.add_scoped<TimelineService>();
         application.add_scoped<ChatService>();
         application.add_singleton<SaddlePollState>();
+        application.add_singleton<AnswerDeliveryHub>();
         application.add_singleton<StatePulse>();
         application.add_singleton<PushService>();
         application.add_singleton<Dispatcher>();

test/bundles/answerwake.py

@@ -0,0 +1,313 @@
+"""The answerwake bundle (goal 28 task F): the park/answer-wake race.
+
+The task-187 arc, replayed end-to-end against the real allocation:
+
+- an agent's ask_question parks the session (the blocked push);
+- the FREED daemon asks again IMMEDIATELY — the very race that used
+  to hand the same task back as a bare crash-restore — and the ask
+  must 204 while the block is unanswered (the allocation guard);
+- a CRASHED parked task resumes with its question still open: the
+  requeued task re-allocates and the assignment flags
+  `pending_question` so the restored conversation's opening turn says
+  the question is pending (F6) instead of barreling on;
+- a human answers while that session RUNS: the verdict is injected
+  into the running round through the daemon's task-update channel
+  (kind "answer"), delivery-once (F2) — no affinity resume can
+  deliver it, the busy saddle cannot ask;
+- the running round parks on a further question; answering it resumes
+  through the allocation and the resume notice carries the FRESH
+  verdict while the already-injected one is never repeated (F3's
+  batched composition with the delivery ledger).
+"""
+from harness import *  # noqa: F401,F403 — the shared wire helpers
+
+
+# The web-side park-in-flight hold (TaskService.PARK_SETTLE_GRACE_S is
+# 5s) plus a slack for the dispatcher tick — the crash-recovery sweep
+# must not fire until the crashed round's last push has aged past it.
+PARK_SETTLE_WAIT = 7.0
+
+
+def run(ctx):
+    auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token
+
+    print("== goal 28 F1: the park race — an immediate re-ask is held ==")
+    # A private capability tag keeps the shard's other sleeping
+    # saddles out of these tasks.
+    saddles_state = slot_public(auth, "/saddles", "saddles")
+    act(auth, saddles_state["create_capability"], {"name": "answerwake"})
+    cap_id = int([c["id"] for c in slot_public(auth, "/saddles", "saddles")["capabilities"]
+                  if c["name"] == "answerwake"][0])
+
+    def race_daemon(name, **kwargs):
+        daemon = FakeSaddle()
+        daemon.ask_on_register = False   # manual, deterministic asks
+        daemon.hold_state = True         # the test drives the lifecycle
+        for key, value in kwargs.items():
+            setattr(daemon, key, value)
+        daemon.serve()
+        act(auth, slot_public(auth, "/saddles", "saddles")["register"],
+            {"name": name, "host": "127.0.0.1:%d" % daemon.port})
+        act(auth, saddle_card(auth, name)["attach_capability"],
+            {"capability_id": str(cap_id)})
+        return daemon
+
+    def race_task(title):
+        return create_queued(auth, {
+            "project_id": "1", "title": title, "priority": "2",
+            "capability_id": [str(cap_id)]})
+
+    runner = race_daemon("race-runner")
+    task_id = race_task("Race: parked then re-asked")
+    ask_status, assignment = runner.ask_now()
+    check(ask_status == 200 and int(assignment["task"]["id"]) == task_id,
+          "the ask claims the queued task")
+    first_session = assignment.get("session_id")
+
+    # The park: the round blocks on a question. The web raises the
+    # block row and closes the session 'parked' synchronously inside
+    # the /results push.
+    runner.state = "blocked"
+    check(runner.push_results("blocked", block={
+        "question": "First: which side?", "type": "question",
+        "options": ["Port", "Starboard"]}) == 200,
+          "the blocked push parks the session")
+    check(wait_for(lambda: task_state(auth, task_id).get("status") == "blocked"
+                   or False, timeout=30, what="raced task parked") is not None,
+          "the task parks as blocked")
+    conn = db()
+    row = conn.execute(
+        "select outcome from sessions where id = ?", (first_session,)).fetchone()
+    conn.close()
+    check(row is not None and row[0] == "parked",
+          "the park closed the round's session row 'parked'")
+
+    # THE RACE: the freed daemon asks again IMMEDIATELY — the block is
+    # unanswered, so the allocation guard must hold the task (204),
+    # never hand it back. The pre-fix web re-allocated it here and the
+    # resumed round woke as a crash restore with its question gone.
+    ask_status, payload = runner.ask_now()
+    check(ask_status == 204,
+          "F1: the immediate re-ask returns 204 while the block is unanswered")
+    check(len([p for p in runner.start_payloads
+               if int(p.get("task", {}).get("id") or 0) == task_id]) == 1,
+          "F1: the parked task was never re-allocated")
+
+    # The guard holds for the pre-block race too: a task whose row
+    # still reads working/assigned while its round's session just
+    # closed 'parked' (the transition in flight) is not claimable.
+    conn = db()
+    conn.execute("update tasks set status = 'working' where id = ?", (task_id,))
+    conn.commit()
+    conn.close()
+    ask_status, _ = runner.ask_now()
+    check(ask_status == 204,
+          "F1: a working row with a fresh parked session is held (transition in flight)")
+    conn = db()
+    conn.execute("update tasks set status = 'blocked' where id = ?", (task_id,))
+    conn.commit()
+    conn.close()
+
+    print("== goal 28 F6: a crashed parked task resumes with its question open ==")
+    # The LEGIT unanswered-block resume (mandate F6): the parked task's
+    # daemon crashed and the crash-recovery sweep requeued it — the
+    # block row stays unanswered while the task re-enters the queue.
+    # The sweep's post-state is applied directly (it needs a stale
+    # OPEN session; this round's row is closed 'parked').
+    second_task = race_task("Race: crashed parked task")
+    ask_status, assignment = runner.ask_now()
+    check(ask_status == 200 and int(assignment["task"]["id"]) == second_task,
+          "the ask claims the second task")
+    second_first_session = assignment.get("session_id")
+    runner.state = "blocked"
+    check(runner.push_results("blocked", block={
+        "question": "Which side first?", "type": "question",
+        "options": ["Port", "Starboard"]}) == 200,
+          "the second task parks on a question")
+    check(wait_for(lambda: task_state(auth, second_task).get("status") == "blocked"
+                   or False, timeout=30, what="second task parked") is not None,
+          "the second task parks as blocked")
+
+    # Answering is the resume: the wake -> ask -> allocation carries
+    # the verdict on a fresh session row.
+    detail = task_state(auth, second_task)
+    side_block = [b for b in detail["blocks"]
+                  if b["question"] == "Which side first?"][0]
+    act(auth, side_block["answer_action"], {"option": "Port"})
+    check(wait_for(lambda: any(
+        int(p.get("task", {}).get("id") or 0) == second_task
+        and p.get("session_id") != second_first_session
+        and p.get("resume_notice") == "Human selected: Port"
+        for p in runner.start_payloads[1:]) or False,
+        timeout=30, what="second task resume allocation") is not None,
+          "the answer resumes the parked task through the allocation")
+    resumed = [p for p in runner.start_payloads
+               if int(p.get("task", {}).get("id") or 0) == second_task][-1]
+    running_session = resumed.get("session_id")
+    check(wait_for(lambda: task_state(auth, second_task).get("status") == "working"
+                   or False, timeout=15, what="second round working") is not None,
+          "the resumed round is running")
+
+    print("== goal 28 F2: an answer while a session runs is injected ==")
+    # The running round parks on a SECOND question, then the round is
+    # lost to a crash — the task-187 shape: the newest block is
+    # UNANSWERED while a fresh round is re-allocated (the crash
+    # sweep's post-state, applied directly for determinism).
+    runner.state = "blocked"
+    check(runner.push_results("blocked", block={
+        "question": "Which database?", "type": "question",
+        "options": ["SQLite", "Postgres"]}) == 200,
+          "the running round parks on a second question")
+    check(wait_for(lambda: task_state(auth, second_task).get("status") == "blocked"
+                   or False, timeout=30, what="second question raised") is not None,
+          "the second question parks the round")
+    # The crash: the round's session row is left OPEN and the task
+    # row reads working (the daemon died between its park push and
+    # its unwind). The next ask's crash-recovery sweep requeues it —
+    # but NOT while the round's last push is fresh (the park-in-flight
+    # hold); after the settle the sweep fires and the requeued task
+    # re-allocates through the QUEUE with its question still open.
+    conn = db()
+    # Aged past the sweep's 120s grace (the round ran for a while
+    # before it crashed) yet its LAST PUSH is the park transition,
+    # seconds old — the exact task-187 shape.
+    conn.execute(
+        "update sessions set ended_at = null, outcome = null,"
+        " started_at = strftime('%s','now','-10 minutes') where id = ?",
+        (running_session,))
+    conn.execute("update tasks set status = 'working' where id = ?", (second_task,))
+    conn.commit()
+    conn.close()
+    # Park-in-flight hold: an immediate ask requeues nothing (the
+    # round pushed moments ago — its transition may be queued).
+    ask_status, _ = runner.ask_now()
+    check(ask_status == 204,
+          "F1: the crash-recovery sweep holds a round whose last push is"
+          " fresh (park in flight)")
+    conn = db()
+    still_open = conn.execute(
+        "select count(*) from sessions where id = ? and ended_at is null",
+        (running_session,)).fetchone()[0]
+    task_row = conn.execute(
+        "select status from tasks where id = ?", (second_task,)).fetchone()[0]
+    conn.close()
+    check(still_open == 1 and task_row == "working",
+          "F1: nothing was requeued by the early ask")
+    # After the settle window the sweep fires on the very next ask:
+    # requeue (the crashed row closes 'stopped') and the freed task is
+    # claimed straight back by the same ask — one round-trip.
+    time.sleep(PARK_SETTLE_WAIT)
+    ask_status, assignment = runner.ask_now()
+    check(ask_status == 200 and assignment.get("session_id") != running_session,
+          "the settled sweep requeues the crashed round and the ask claims it"
+          " (fresh session row)")
+    # The answer below must find the round genuinely RUNNING: wait for
+    # the daemon to key on the new session and push its opening batch
+    # (the busy-owner state the delivery path keys on).
+    live_session = assignment.get("session_id")
+    check(wait_for(lambda: runner.session_id == live_session or False,
+                   timeout=15, what="daemon keyed on the live session") is not None,
+          "the daemon keyed its pushes on the live session")
+    runner.push_results("working")
+    check(wait_for(lambda: task_state(auth, second_task).get("status") == "working"
+                   or False, timeout=15, what="live round working") is not None,
+          "the re-allocated round is running")
+    check(assignment.get("pending_question") is True,
+          "F6: the assignment flags its unanswered newest block (pending_question)")
+
+    # The human answers the OPEN question while this round runs: no
+    # affinity resume can deliver it (the saddle is busy), so the
+    # verdict must ride the task-update injection channel. The
+    # dispatcher drains the hub every tick (TICK_S=1 in the suite).
+    # (The barrier above guarantees the open session exists BEFORE the
+    # answer fires — an answer against a closed task takes the resume
+    # path instead, which is F3's business.)
+    detail = task_state(auth, second_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: any(e.get("injected") and e.get("text") == "Human selected: SQLite"
+                               for e in runner.answer_log) or False,
+                   timeout=30, what="answer injected into the running session") is not None,
+          "F2: the answer reaches the RUNNING session via task-update kind answer")
+    conn = db()
+    check(wait_for(lambda: conn.execute(
+        "select count(*) from task_comments where task_id = ? and body = ?",
+        (second_task,
+         "verdict delivered into the running session (block #%s)" % db_block["id"])
+        ).fetchone()[0] > 0 or False,
+        timeout=15, what="delivery system comment") is not None,
+          "F2: the delivery is recorded on the task timeline")
+    conn.close()
+    # Delivery is exactly-once: no second injection may land.
+    check(quiet_window(quiet_seconds(),
+                       lambda: len([e for e in runner.answer_log if e.get("injected")]) == 1,
+                       "no duplicate answer injection"),
+          "F2: the answer is injected exactly once")
+    conn = db()
+    injected = conn.execute(
+        "select answers_injected from sessions where id = ?",
+        (live_session,)).fetchone()
+    conn.close()
+    check(injected is not None and injected[0] >= 1,
+          "F2: the injection is counted on the session row (answers_injected)")
+
+    print("== goal 28 F3: the resume notice composes the batch ==")
+    # The live round parks on a THIRD question; answering it resumes
+    # through the allocation. The notice carries the FRESH verdict and
+    # never repeats the already-injected one (the delivery ledger is
+    # the filter) — a newer block can no longer swallow an older
+    # answer silently.
+    runner.state = "blocked"
+    check(runner.push_results("blocked", block={
+        "question": "Final check?", "type": "question",
+        "options": ["Ship", "Hold"]}) == 200,
+          "the live round parks on a third question")
+    check(wait_for(lambda: task_state(auth, second_task).get("status") == "blocked"
+                   or False, timeout=30, what="third park") is not None,
+          "the task parks on its third question")
+    ask_status, _ = runner.ask_now()
+    check(ask_status == 204,
+          "the third park also holds the immediate re-ask (unanswered block)")
+
+    detail = task_state(auth, second_task)
+    final_block = [b for b in detail["blocks"]
+                   if b["question"] == "Final check?"][0]
+    act(auth, final_block["answer_action"], {"option": "Ship"})
+    check(wait_for(lambda: any(
+        int(p.get("task", {}).get("id") or 0) == second_task
+        and p.get("session_id") != live_session
+        and "Human selected: Ship" in (p.get("resume_notice") or "")
+        for p in runner.start_payloads[1:]) or False,
+        timeout=30, what="third resume allocation") is not None,
+          "F3: the answer resumes the parked task with its verdict")
+    final = [p for p in runner.start_payloads
+             if int(p.get("task", {}).get("id") or 0) == second_task][-1]
+    notice = final.get("resume_notice") or ""
+    check("Human selected: Ship" in notice,
+          "F3: the fresh verdict rides the resume notice")
+    check("Human selected: SQLite" not in notice,
+          "F3: the already-injected verdict is NOT repeated as a resume verdict")
+    check(final.get("pending_question") is not True,
+          "the resumed assignment has no open question left")
+
+    # The resume_notice persists on the session row (the terminal
+    # page's resume-info panel reads it) and the watermark was stored.
+    conn = db()
+    stored = conn.execute(
+        "select resume_notice from sessions where id = ?",
+        (final.get("session_id"),)).fetchone()
+    watermark = conn.execute(
+        "select resume_since is not null from sessions where id = ?",
+        (final.get("session_id"),)).fetchone()
+    conn.close()
+    check(stored is not None and "Human selected: Ship" in (stored[0] or ""),
+          "the resume notice persists on the resumed session row")
+    check(watermark is not None and watermark[0] == 1,
+          "the batched-verdict watermark persists (resume_since)")
+
+    # The final round completes cleanly.
+    runner.push_results("done", summary="race rounds complete")
+    check(wait_task_done(auth, second_task, timeout=30) is not None,
+          "the resumed round continues to completion")
+    runner.kill()

test/harness.py

@@ -868,6 +868,16 @@ class FakeSaddle:
                 elif kind == "comment":
                     self.daemon.comment_log.append(payload)
                     self._send(200, {"ok": True})
+                elif kind == "answer":
+                    # Goal 28 task F (busy-answer delivery): the web
+                    # injects a block verdict into the RUNNING round —
+                    # a human answered while this session was already
+                    # running, so no allocation resume can deliver it.
+                    self.daemon.answer_log.append({
+                        "injected": True,
+                        "text": payload.get("text", ""),
+                        "task_id": payload.get("task_id")})
+                    self._send(200, {"ok": True})
                 else:
                     self._send(400, {"error": "unknown task-update kind"})
                 return
@@ -1321,9 +1331,9 @@ def start_bundle(name):
 
 
 BUNDLES_ORDER = ("pages", "dispatch", "subtasks", "lineage",
-                 "allocation", "lifecycle", "answers", "push", "goals",
-                 "replay", "goalui", "forcewake", "users", "stacks", "notes",
-                 "resurrect", "origins", "parity", "understanding",
+                 "allocation", "lifecycle", "answers", "answerwake", "push",
+                 "goals", "replay", "goalui", "forcewake", "users", "stacks",
+                 "notes", "resurrect", "origins", "parity", "understanding",
                  "discoveries", "knowledge_cycle", "advisor", "chatui",
                  "advisor_arc", "advisor_fix", "advisor_refine", "canon",
                  "canon_deploy")