phase 2: telemetry flip — pushed chunks/results ingestion, staleness probe, pollers retired

Web side of the plan-10 phase-2 wire contract:

- POST /api/saddle/chunks: batched raw PTY pushes store into
  session_raw_log (the store_new_raw_log logic, push-driven),
  idempotent on the (session_id, seq) uniqueness — outbox
  redeliveries are consumed; the SSE terminal is unchanged.
- POST /api/saddle/results: round entries store like the /output
  path did (the daemon's global entry ids drive the per-session
  cursor, so at-least-once pushes store exactly once) and the state
  transitions drive the old /status-switch bookkeeping: blocked
  raises the block payload (ensure_block) and parks the saddle with
  the pulse bump (live questions without reload), suspended parks
  for the affinity resume, done/failed complete/fail with the pushed
  summary, working flips (never downgrading a review or
  resurrecting terminal work).
- Staleness: SADDLE_STALE_S (default 90) of chunk/results silence
  owes ONE exceptional GET /status; failure requeues the task with a
  bot comment and marks the saddle offline. A recovered daemon's
  ask flips an offline row back to idle (proof of life — no fence/
  heal passes exist anymore).
- Cancellation is POST /task-update {kind: cancel} to the RUNNING
  daemon; comment and subtask-status delivery moved onto the same
  channel per the wire contract.
- Deleted the superseded layers: /output + /log pollers, 3-strike
  unreachable logic, heal/wedge/fence/stranded-assignment recovery,
  push_with_zombie_recovery, the /start push machinery (rollback
  path retired — dispatch is ask/allocation only) and the seed-time
  cursor probes. GET /status manual refresh (the UI button) stays.

Harness: the fake daemon pushes telemetry (opening flush +
heartbeats + terminal states, queue_output as entries) and serves
/task-update; /output, /log and /start 404. 281 checks green
including: chunk idempotency/order, blocked push -> block row +
pulse + sealed refresh + resume-with-answer, done/failed pushes,
entry redelivery dedup, the one-probe staleness policy (alive keeps
running, dead requeues + offlines), token auth + session ownership,
and the deleted paths.

README.md

@@ -10,10 +10,10 @@ during the day; the agent works while you sleep.
 
 ```
 Browser ←→ clanker-saddle (Spry app: SQLite queue, Statum UI, dispatcher)
-                          ↕ HTTP (poll-based, 5s ticks)
-              saddle daemons (Vala: bash session + LLM loop, port 7100)
+                          ↕ HTTP (daemons push; asks/wakes only go out)
+               saddle daemons (Vala: bash session + LLM loop, port 7100)
                           ↕ curl
-              any OpenAI-compatible chat-completions endpoint
+               any OpenAI-compatible chat-completions endpoint
 ```
 
 The web app owns projects, tasks, dependencies and the saddle registry.
@@ -24,11 +24,16 @@ offerable task — workspace affinity first (a task parked on this saddle
 is offered only to it), then capability-tag matching. The dispatcher
 wakes sleeping daemons when work becomes ready (`POST /wake`, the affine
 saddle first for parked tasks) and declares a saddle offline when no ask
-arrives within the lost-wake guard window. During the phase-1 transition
-the poll-based telemetry stays: busy saddles are still polled for
-`GET /output`, `GET /log` and `GET /status` every five seconds (with the
-heal/fence layers), and pre-flip daemons that never ask keep their
-`/answer` and `/start` resume paths.
+arrives within the lost-wake guard window. Telemetry is push-based
+(phase 2): daemons flush raw PTY batches to `POST /api/saddle/chunks`
+and round results with state transitions to `POST /api/saddle/results` —
+`blocked` raises the question as an answerable block and parks the
+saddle, `suspended` parks for the subtask-event resume, `done`/`failed`
+complete or fail the task. Nothing polls for output; `GET /status`
+against a daemon is exceptional only — the staleness probe (a silent
+runner owes one probe after `SADDLE_STALE_S`; a failed probe requeues
+its task and marks the saddle offline) and the manual refresh button.
+A recovered daemon's ask flips an offline row back to idle.
 
 | Component | Lives | Role |
 |---|---|---|
@@ -43,6 +48,7 @@ Orchestration behaviour to know:
 | `SADDLE_LLM_BASE_URL` | LLM endpoint sent with every assignment; unset pauses dispatch |
 | `SADDLE_LLM_API_KEY` / `SADDLE_LLM_MODEL` | Optional credentials/model for the above |
 | `SADDLE_LOST_WAKE_S` | Lost-wake guard window (default 15s): no ask after a wake → the saddle goes offline and the next capable one is woken |
+| `SADDLE_STALE_S` | Running-liveness window (default 90s): no chunks or results for this long → one exceptional `GET /status` probe; failure requeues the task and marks the saddle offline |
 | `SADDLE_PROJECTS_ROOT` | Checkouts are addressed as `root/<project-name>` inside the saddle container (default `/workspace`) |
 
 ## Quick start (local)
@@ -132,32 +138,36 @@ with ephemeral keys.
    the token from the saddle card instead.
 
 Tasks can also arrive from the agent itself: `saddle subtask` lines
-create child tasks, and push-style daemons may report through
-`POST /api/saddle/{id}/update` (`output`, `comment`, `block`, `subtask`,
-`done`, `status`) and collect answers from
-`GET /api/saddle/{id}/poll` — every `/api/saddle/*` call is authenticated
-by the per-saddle bearer token (removing the saddle revokes it).
+create child tasks (through the synchronous `POST /api/saddle/subtask`
+callback), and push-style daemons report through `POST
+/api/saddle/results` (round entries plus state transitions) and `POST
+/api/saddle/chunks` (raw PTY batches) — every `/api/saddle/*` call is
+authenticated by the per-saddle bearer token (removing the saddle
+revokes it).
 
 ## Monitoring agent progress
 
 - **`/tasks`** — live queue: status chips (`queued → assigned → working →
   blocked → done/failed`), dependency states and ready flags.
-- **`/tasks/{id}`** — the control room; it refreshes every three seconds
-  while a saddle works: comments thread (yours and the agent's), blocks,
-  subtasks, session history, and the agent's final summary.
-- **`/tasks/{id}/terminal`** — the full bash session log: every command
-  the agent ran with its combined output and exit code, plus its
-  commentary lines (`# …`) and saddle events.
+- **`/tasks/{id}`** — the control room; it refreshes on every state
+  pulse while a saddle works: comments thread (yours and the agent's),
+  blocks, subtasks, session history, and the agent's final summary.
+- **`/tasks/{id}/terminal`** — the full bash session log: the raw PTY
+  stream the daemon pushes (chunk batches land in `session_raw_log` and
+  stream over SSE byte-exact), with the typed entry view as the fallback
+  for legacy sessions.
 - **`/`** — dashboard: saddle cards, queue summary, recent activity.
 
 ## Answering block questions
 
-When an agent needs a human decision it runs
-`saddle block "<question>"`; the task flips to **blocked** and the
-question appears on its page with an answer form. Type the answer, submit
-— the dispatcher delivers it to the daemon within five seconds, the agent
-resumes with the answer in its conversation, and the task returns to
-`working`. Answers are delivered exactly once per question.
+When an agent needs a human decision it runs `saddle block "<question>"`;
+the daemon pushes the `blocked` state with the block payload, the task
+flips to **blocked** (the pulse bump lands the question on live pages
+without a reload) and the saddle parks. Type the answer, submit — the
+task becomes resume-offerable, the dispatcher wakes the affine saddle,
+and the daemon's ask returns the assignment with the answer in its
+conversation; the task returns to `working`. Answers ride the resume
+assignment exactly once per question.
 
 ## Rules and project notes
 
@@ -225,9 +235,15 @@ earlier run via `/stop` before it starts.
 self-contained suite (no daemon needed) that spawns the app in a scratch
 directory plus an in-process fake saddle daemon speaking the wire
 contract, and asserts the web-stack features: plan-review body hiding,
-the effective `rules` array of `/start`, project notes with the
+the effective `rules` array of assignments, project notes with the
 `/api/saddle/note` callback and the payload's `project_notes`/`callback`
-caps, and raw-log polling into the terminal view.
+caps, the ask/allocation flip (concurrent asks race one winner, affinity
+pinning, the lost-wake guard), and the phase-2 telemetry flip (pushed
+chunks idempotent and ordered into the terminal view, results pushes
+with blocked/suspended/done/failed state transitions, entry-redelivery
+idempotency, the one-probe staleness policy with its dead-daemon
+requeue, and the retired `/output`//`/log`//`/start` daemon routes 404ing
+while the web side never calls them).
 
 ## Realtime channels and the browser's SSE budget
 

src/Api/SaddleApi.vala

@@ -651,6 +651,8 @@ namespace ClankerSaddle.Api {
         private AllocationService allocation = inject<AllocationService>();
         private ProjectService projects = inject<ProjectService>();
         private SaddlePollState poll_state = inject<SaddlePollState>();
+        private InvercargillSql.Orm.OrmSession db = inject<InvercargillSql.Orm.OrmSession>();
+        private StatePulse pulse = inject<StatePulse>();
 
         public virtual async HttpResult handle_request(HttpContext http_context,
                 RouteContext route_context) throws Error {
@@ -659,6 +661,17 @@ namespace ClankerSaddle.Api {
                 return SaddleUpdateEndpoint.json_error(401, "Missing or invalid saddle token");
             }
 
+            // An ask from an offline saddle is proof of life: the daemon
+            // recovered (its 15-minute safety self-ask, or it came back
+            // after the probe/guard marked it offline) and no recovery
+            // pass exists to flip the row — the ask itself does.
+            if (((!)saddle).status == "offline") {
+                ((!)saddle).status = "idle";
+                yield db.update_async<Saddle>(((!)saddle));
+                yield pulse.bump(0, "saddle asked (offline -> idle)");
+                print(@"[ClankerSaddle] Saddle '$(((!)saddle).name)' asked while offline; marking idle\n");
+            }
+
             // The ask's arrival answers any pending wake (the lost-wake
             // guard must not fire for a saddle that just asked); the
             // outcome is recorded once the allocation resolves.
@@ -688,6 +701,164 @@ namespace ClankerSaddle.Api {
         }
     }
 
+    /**
+     * `POST /api/saddle/chunks` (plan 10 phase 2) — the raw-PTY push
+     * channel, authorised by the pushing saddle's bearer token:
+     *
+     * ```json
+     * {"session_id": 12, "chunks": [{"seq": 41, "text": "…"}]}
+     * ```
+     *
+     * The daemon's flusher batches chunks (at least twice per round and
+     * on a 10 s timer while output flows) and pushes them here; the
+     * batch is stored into `session_raw_log` (so the SSE terminal keeps
+     * working unchanged), idempotent on the existing (session_id, seq)
+     * uniqueness — an outbox redelivery after a lost response is
+     * consumed, not duplicated. Any push implies the daemon is alive
+     * (the staleness clock resets).
+     */
+    public class SaddleChunksEndpoint : Object, Endpoint {
+
+        private SaddleService saddles = inject<SaddleService>();
+        private SaddlePollState poll_state = inject<SaddlePollState>();
+        private InvercargillSql.Orm.OrmSession db = inject<InvercargillSql.Orm.OrmSession>();
+
+        public virtual async HttpResult handle_request(HttpContext http_context,
+                RouteContext route_context) throws Error {
+            var saddle = yield saddles.saddle_for_token(bearer_token(http_context));
+            if (saddle == null) {
+                return SaddleUpdateEndpoint.json_error(401, "Missing or invalid saddle token");
+            }
+
+            Json.Object body;
+            try {
+                body = yield SaddleUpdateEndpoint.read_json_body(http_context);
+            } catch (Error e) {
+                return SaddleUpdateEndpoint.json_error(400, e.message);
+            }
+
+            int64 session_id = body.has_member("session_id")
+                && body.get_member("session_id").get_value_type() == typeof(int64)
+                ? body.get_int_member("session_id") : 0;
+            var session = yield session_owned_by(db, session_id, (!)saddle);
+            if (session == null) {
+                return SaddleUpdateEndpoint.json_error(400,
+                    @"No session $(session_id) exists on saddle $(((!)saddle).id)");
+            }
+
+            if (!(body.has_member("chunks")
+                    && body.get_member("chunks").get_node_type() == Json.NodeType.ARRAY)) {
+                return SaddleUpdateEndpoint.json_error(400,
+                    "A chunks array is required");
+            }
+
+            poll_state.note_push(session_id);
+            try {
+                int64 stored = yield saddles.store_pushed_chunks(session_id,
+                    body.get_array_member("chunks"));
+                var response = new Json.Object();
+                response.set_int_member("stored", stored);
+                return SaddleUpdateEndpoint.json_ok(response);
+            } catch (Error e) {
+                return SaddleUpdateEndpoint.json_error(400, e.message);
+            }
+        }
+
+        internal static async ClankerSaddle.Data.Session? session_owned_by(
+                InvercargillSql.Orm.OrmSession db, int64 session_id, Saddle saddle) throws Error {
+            if (session_id <= 0) {
+                return null;
+            }
+            var session = yield db.query<ClankerSaddle.Data.Session>()
+                .where(Invercargill.Expressions.expr("id == $0",
+                    Invercargill.Expressions.elem<int64?>(session_id)))
+                .first_async();
+            if (session == null || ((!)session).saddle_id != saddle.id) {
+                return null;
+            }
+            return session;
+        }
+    }
+
+    /**
+     * `POST /api/saddle/results` (plan 10 phase 2) — the round-results
+     * push channel, authorised by the pushing saddle's bearer token:
+     *
+     * ```json
+     * {"session_id": 12, "state": "blocked",
+     *  "entries": [{"id": 9, "command": "…", "output": "…", "exit_code": 0,
+     *               "type": "command", "timestamp": "…"}],
+     *  "block": {"question": "…", "type": "question", "options": […],
+     *            "body": "…"},
+     *  "summary": "…"}
+     * ```
+     *
+     * Entries keep today's OutputEntry shape and store like the deleted
+     * /output poll path (the daemon's global entry ids make redeliveries
+     * idempotent). The state transitions drive the same bookkeeping the
+     * /status poll switch ran: `blocked` raises the block payload as an
+     * answerable `task_blocks` row (the pulse bump makes the question
+     * appear on live pages) and parks the saddle; `suspended` parks for
+     * the subtask-event resume; `done`/`failed` complete or fail the
+     * task with the pushed summary; `working` flips the task to working.
+     */
+    public class SaddleResultsEndpoint : Object, Endpoint {
+
+        private SaddleService saddles = inject<SaddleService>();
+        private TaskService tasks = inject<TaskService>();
+        private SaddlePollState poll_state = inject<SaddlePollState>();
+        private InvercargillSql.Orm.OrmSession db = inject<InvercargillSql.Orm.OrmSession>();
+
+        public virtual async HttpResult handle_request(HttpContext http_context,
+                RouteContext route_context) throws Error {
+            var saddle = yield saddles.saddle_for_token(bearer_token(http_context));
+            if (saddle == null) {
+                return SaddleUpdateEndpoint.json_error(401, "Missing or invalid saddle token");
+            }
+
+            Json.Object body;
+            try {
+                body = yield SaddleUpdateEndpoint.read_json_body(http_context);
+            } catch (Error e) {
+                return SaddleUpdateEndpoint.json_error(400, e.message);
+            }
+
+            int64 session_id = body.has_member("session_id")
+                && body.get_member("session_id").get_value_type() == typeof(int64)
+                ? body.get_int_member("session_id") : 0;
+            var session = yield SaddleChunksEndpoint.session_owned_by(db, session_id, (!)saddle);
+            if (session == null) {
+                return SaddleUpdateEndpoint.json_error(400,
+                    @"No session $(session_id) exists on saddle $(((!)saddle).id)");
+            }
+
+            string state = SaddleUpdateEndpoint.get_string_or(body, "state", "");
+            if (state.length == 0) {
+                return SaddleUpdateEndpoint.json_error(400, "A state is required");
+            }
+            Json.Object? block = body.has_member("block")
+                && body.get_member("block").get_node_type() == Json.NodeType.OBJECT
+                ? body.get_object_member("block") : null;
+            string? summary = SaddleUpdateEndpoint.get_string_or(body, "summary", null);
+
+            poll_state.note_push(session_id);
+            try {
+                var response = new Json.Object();
+                if (body.has_member("entries")
+                        && body.get_member("entries").get_node_type() == Json.NodeType.ARRAY) {
+                    var stored = yield saddles.store_pushed_entries(session_id,
+                        body.get_array_member("entries"));
+                    yield saddles.apply_entry_events(((!)session).task_id, stored, tasks);
+                }
+                yield saddles.apply_result_state((!)saddle, (!)session, state, block,
+                    summary, tasks);
+                return SaddleUpdateEndpoint.json_ok(response);
+            } catch (Error e) {
+                return SaddleUpdateEndpoint.json_error(400, e.message);
+            }
+        }
+    }
+
     /**
      * The request's bearer token: the `Authorization` header minus its
      * `Bearer ` prefix, or "" when the header is missing or shaped

src/Services/AllocationService.vala

@@ -73,6 +73,7 @@ namespace ClankerSaddle.Services {
         private CapabilityService capabilities = inject<CapabilityService>();
         private SubtaskEventQueue subtask_events = inject<SubtaskEventQueue>();
         private StatePulse pulse = inject<StatePulse>();
+        private SaddlePollState poll_state = inject<SaddlePollState>();
 
         /**
          * Allocates the best offerable task to `saddle`, or null when
@@ -241,34 +242,13 @@ namespace ClankerSaddle.Services {
                     }
                     session_id = ((!)newest_open).id;
                 } else {
-                    Session? reusable = null;
-                    var earlier = db.query<Session>()
-                        .where(expr("task_id == $0 && saddle_id == $1",
-                            elem<int64?>(candidate.task.id), elem<int64?>(saddle.id)))
-                        .order_by(expr("id"))
-                        .materialise();
-                    foreach (var open in earlier) {
-                        if (open.outcome == "push_failed") {
-                            reusable = open;
-                        }
-                    }
-                    if (reusable != null) {
-                        // Reuse the failed attempt's row instead of
-                        // churning a fresh one per retry.
-                        ((!)reusable).outcome = null;
-                        ((!)reusable).ended_at = null;
-                        ((!)reusable).started_at = new DateTime.now_local();
-                        db.update<Session>((!)reusable);
-                        session_id = ((!)reusable).id;
-                    } else {
-                        var session = new Session() {
-                            task_id = candidate.task.id,
-                            saddle_id = saddle.id,
-                            started_at = new DateTime.now_local()
-                        };
-                        db.insert<Session>(session);
-                        session_id = session.id;
-                    }
+                    var session = new Session() {
+                        task_id = candidate.task.id,
+                        saddle_id = saddle.id,
+                        started_at = new DateTime.now_local()
+                    };
+                    db.insert<Session>(session);
+                    session_id = session.id;
                 }
 
                 ((!)task).status = STATUS_ASSIGNED;
@@ -330,10 +310,10 @@ namespace ClankerSaddle.Services {
          * Assembles the claimed task's assignment payload (today's
          * /start shape): pending block answers ride the task object,
          * a suspended parent's resume carries the coalesced subtask
-         * report as `resume_notice` plus the `branch_advanced` verdict,
-         * and the output cursor is seeded from the daemon before the
-         * payload is handed over (the daemon cannot start before the
-         * ask is answered, so the baseline predates its first entry).
+         * report as `resume_notice` plus the `branch_advanced` verdict.
+         * The hand-over itself starts the session's push-liveness clock
+         * (the daemon cannot push before it owns the assignment), so
+         * the staleness probe's window opens here.
          */
         private async Json.Object assemble_assignment(SaddleService saddles,
                 ProjectService projects, Saddle saddle, AllocationClaim claim) throws Error {
@@ -356,7 +336,7 @@ namespace ClankerSaddle.Services {
                 }
             }
 
-            yield saddles.seed_session_cursor(saddle.id, claim.session_id);
+            poll_state.note_push(claim.session_id);
             var payload = yield saddles.build_assignment_payload(saddle, task,
                 claim.session_id, projects, resume_notice, branch_advanced);
             payload.set_int_member("session_id", claim.session_id);

src/Services/Dispatcher.vala

@@ -6,20 +6,20 @@ using ClankerSaddle.Data;
 namespace ClankerSaddle.Services {
 
     /**
-     * The task dispatcher (plan 10, phase 1): a five-second timer that
+     * The task dispatcher (plan 10, phase 2): a five-second timer that
      * computes wakes — when a task becomes offerable and a capable
      * SLEEPING saddle exists (its last ask returned 204), the daemon is
      * `POST /wake`d and fetches the work itself via
      * `POST /api/saddle/next-task`; parked tasks target their affine
      * saddle first. The lost-wake guard declares a saddle offline when
      * no ask arrives within the guard window after a wake, freeing the
-     * work for the next capable sleeping saddle.
-     *
-     * The pre-flip machinery stays fully functional during phase 1
-     * (rollback path): parked-task resumes on non-sleeping (pre-flip)
-     * daemons still go out as /answer deliveries and /start pushes,
-     * and busy saddles keep being polled for /output, /log and /status
-     * with the 3-strike, heal and fence layers.
+     * work for the next capable sleeping saddle. The staleness pass
+     * probes a RUNNING daemon that has pushed neither chunks nor
+     * results for the SADDLE_STALE_S window — one exceptional
+     * `GET /status`; a failed probe requeues the task with a bot
+     * comment and marks the saddle offline. Round telemetry arrives as
+     * pushes (`/api/saddle/chunks`, `/api/saddle/results`); the
+     * dispatcher never polls for output.
      *
      * Each tick resolves fresh request-scoped services (so no database
      * scope is held between ticks) and never overlaps with itself: a slow
@@ -41,14 +41,6 @@ namespace ClankerSaddle.Services {
         private Invercargill.DataStructures.HashSet<string> capability_wait_logged =
             new Invercargill.DataStructures.HashSet<string>();
 
-        /**
-         * Consecutive communication failures per saddle id. Transient
-         * errors (a pooled connection dropped mid-reuse, a momentary
-         * SQLite lock) must not tear down a running session, so a saddle
-         * is only declared unreachable after three strikes in a row.
-         */
-        private Dictionary<string, string> strikes = new Dictionary<string, string>();
-
         /** Starts the repeating timer. */
         public void start() {
             Timeout.add_seconds(5, () => {
@@ -84,15 +76,6 @@ namespace ClankerSaddle.Services {
             }
         }
 
-        private int strike_count(int64 saddle_id) {
-            string stored = "";
-            return strikes.try_get(saddle_id.to_string(), out stored) ? int.parse(stored) : 0;
-        }
-
-        private void record_strike(int64 saddle_id, int count) {
-            strikes.set(saddle_id.to_string(), count.to_string());
-        }
-
         /** SADDLE_LOST_WAKE_S (default 15s), in milliseconds. */
         public static uint lost_wake_guard_ms() {
             var raw = Environment.get_variable("SADDLE_LOST_WAKE_S") ?? "";
@@ -101,16 +84,16 @@ namespace ClankerSaddle.Services {
         }
 
         /**
-          * One dispatch cycle: recovery passes first (stale-busy healing
-          * frees saddles within the same tick they strand), then the
-          * lost-wake guard offlines saddles whose wake went unanswered,
-          * then the wake computation offers every offerable task to a
-          * capable sleeping daemon. Parked-task resumes against
-          * non-sleeping (pre-flip) daemons still run the /answer + /start
-          * resume paths, subtask events are pushed, and busy saddles are
-          * polled for output and status changes. Wake computation pauses
-          * while {@link LlmEndpoint} is unconfigured (polling and
-          * resumes continue).
+          * 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
+          * 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
+          * daemons via `POST /task-update`, and silent runners get their
+          * one exceptional staleness probe. Wake computation pauses while
+          * {@link LlmEndpoint} is unconfigured (deliveries and probes
+          * continue).
           */
         public async void tick() throws Error {
             var scope = container.create_scope();
@@ -126,119 +109,38 @@ namespace ClankerSaddle.Services {
                 }
             } else {
                 warned_no_llm = false;
-                // Stale-busy healing comes first so a freed saddle is
-                // dispatchable within the same tick; tasks stranded on
-                // offline saddles requeue right after.
-                yield saddles.heal_stale_busy_saddles(tasks);
-                yield saddles.requeue_tasks_on_offline_saddles(tasks);
-                yield saddles.recover_stranded_assignments(tasks);
-                yield saddles.recover_fenced_saddles();
                 yield enforce_lost_wake_guard(saddles, tasks);
                 yield wake_sleeping_saddles(saddles, capabilities, allocation);
             }
 
-            yield resume_answered_tasks(tasks, saddles, scope.resolve<ProjectService>());
-            yield resume_suspended_parents(tasks, saddles, scope.resolve<ProjectService>());
+            yield saddles.requeue_tasks_on_offline_saddles(tasks);
             yield saddles.deliver_pending_subtask_events(tasks);
 
             foreach (var saddle in yield saddles.list_busy_saddles()) {
+                // Human comments ride POST /task-update to the RUNNING
+                // daemon (best effort — a transport failure stays
+                // undelivered for the next tick; the staleness probe
+                // owns the daemon's fate, not a strike counter).
                 try {
-                    yield saddles.poll_saddle_updates(saddle.id, tasks);
                     yield saddles.deliver_pending_comments(saddle.id, saddle.current_task_id);
-                    record_strike(saddle.id, 0);
-                } catch (Error e) {
-                    int count = strike_count(saddle.id) + 1;
-                    record_strike(saddle.id, count);
-                    warning(@"[ClankerSaddle] Polling saddle $(saddle.id) failed (attempt $(count)/3): $(e.message)");
-                    if (count >= 3) {
-                        yield abandon_saddle(saddle.id, saddles, tasks,
-                            @"daemon unreachable after $(count) failed polls: $(e.message)");
-                    }
-                }
-            }
-        }
-
-        /**
-         * Resumes blocked/review-requested tasks whose newest block has
-         * been answered, on the saddle holding their open (parked)
-         * session — the workspace-pause counterpart of the dispatch loop.
-         * Capabilities do not apply here: the parked workspace pins the
-         * task to the saddle that holds it.
-         */
-        private async void resume_answered_tasks(TaskService tasks, SaddleService saddles,
-                ProjectService projects) throws Error {
-            var candidates = new Vector<TaskEntity>();
-            foreach (var status in new string[] { STATUS_BLOCKED, STATUS_REVIEW_REQUESTED }) {
-                foreach (var task in yield tasks.list_tasks(status, null)) {
-                    candidates.add(task);
-                }
-            }
-
-            foreach (var task in candidates) {
-                TaskBlock? newest = null;
-                foreach (var block in yield tasks.list_blocks(task.id)) {
-                    newest = block;
-                }
-                if (newest == null || ((!)newest).answer == null) {
-                    continue;
-                }
-
-                var session = yield tasks.open_session_for_task(task.id);
-                if (session == null) {
-                    continue;
-                }
-                try {
-                    yield saddles.resume_answered_task(task, (!)session, tasks, projects);
                 } catch (Error e) {
-                    warning(@"[ClankerSaddle] Resuming task $(task.id) failed: $(e.message)");
+                    warning(@"[ClankerSaddle] Comment delivery to saddle $(saddle.id) failed: $(e.message)");
                 }
-            }
-        }
 
-        /**
-         * Resumes suspended parents whose subtasks produced events since
-         * they parked (a subtask completed or failed): a fresh /start on
-         * the saddle holding the parked workspace, carrying the coalesced
-         * markdown status report of every subtask as the resume notice.
-         * Resumes on EACH such event — not only when all subtasks finish.
-         */
-        private async void resume_suspended_parents(TaskService tasks, SaddleService saddles,
-                ProjectService projects) throws Error {
-            foreach (var parent in yield tasks.list_tasks(STATUS_SUSPENDED, null)) {
-                if (!(yield tasks.has_terminal_subtask_since(parent))) {
+                if (saddle.current_task_id == 0) {
                     continue;
                 }
-                var session = yield tasks.open_session_for_task(parent.id);
+                var session = yield tasks.open_session_for_task(saddle.current_task_id);
                 if (session == null) {
                     continue;
                 }
-                try {
-                    string notice = yield tasks.subtask_status_report(parent.id);
-                    bool resumed = yield saddles.resume_answered_task(parent, (!)session,
-                        tasks, projects, notice);
-                    if (resumed) {
-                        // The parking window is over: a later suspension
-                        // starts a fresh event window.
-                        yield tasks.clear_suspension(parent.id);
-                        print(@"[ClankerSaddle] Resumed suspended task $(parent.id) with a subtask report\n");
-                    }
-                } catch (Error e) {
-                    warning(@"[ClankerSaddle] Resuming suspended task $(parent.id) failed: $(e.message)");
+                if (poll_state.take_stale_probe(((!)session).id,
+                        SaddleService.stale_after_seconds() * 1000)) {
+                    yield saddles.probe_stale_saddle(saddle, tasks);
                 }
             }
         }
 
-        /**
-         * Declares a saddle unreachable once its failures accumulated:
-         * the task it held returns to the queue (with a bot comment) and
-         * its session closes.
-         */
-        private async void abandon_saddle(int64 saddle_id, SaddleService saddles,
-                TaskService tasks, string? reason = null) throws Error {
-            record_strike(saddle_id, 0);
-            yield saddles.mark_saddle_unreachable(saddle_id, tasks, reason);
-        }
-
         /**
          * The lost-wake guard: a wake was sent, the guard window has
          * passed, and still no ask arrived ({@link
@@ -268,10 +170,8 @@ namespace ClankerSaddle.Services {
          * SLEEPING 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); a non-sleeping affine saddle (a pre-flip daemon, or
-         * one blocked in its answer loop) is left to the /answer +
-         * /start resume paths. Saddles with a pending wake inside the
-         * guard window are skipped — the guard decides their fate.
+         * them). 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 {
@@ -285,7 +185,8 @@ namespace ClankerSaddle.Services {
                             && !poll_state.wake_pending(((!)affine).id, lost_wake_guard_ms())) {
                         target = (!)affine;
                     }
-                    // Otherwise: the resume paths own the task this tick.
+                    // Otherwise: the task stays offerable for the affine
+                    // saddle's own idle-awake ask.
                 } else {
                     foreach (var candidate in yield saddles.list_idle_saddles()) {
                         if (!(yield capabilities.saddle_covers_task(candidate.id,
@@ -293,9 +194,6 @@ namespace ClankerSaddle.Services {
                             continue;
                         }
                         if (!poll_state.is_sleeping(candidate.id)) {
-                            // A pre-flip daemon never asks; waking it is
-                            // pointless. It stays eligible for the
-                            // rollback /start paths only.
                             continue;
                         }
                         if (poll_state.wake_pending(candidate.id, lost_wake_guard_ms())) {

src/Services/SaddlePollState.vala

@@ -5,15 +5,17 @@ namespace ClankerSaddle.Services {
 
     /**
      * Process-wide saddle poll bookkeeping: the last daemon-side output
-     * entry id stored per session (so repeated polls never duplicate
-     * `session_commands` rows), the last daemon-side raw-log `seq`
-     * stored per session (`session_raw_log`), the block answers and
-     * human comments already handed to a saddle (each delivered exactly
-     * once; a restart re-delivers and the daemon deduplicates), plus
-     * the plan-10 ask/wake liveness bookkeeping: the outcome of a
-     * saddle's last `POST /api/saddle/next-task` ask (a saddle counts
-     * as SLEEPING when its last ask returned 204) and the pending wake
-     * each sleeping saddle was sent (the lost-wake guard's timer).
+     * entry id stored per session (so redelivered results pushes never
+     * duplicate `session_commands` rows), the block answers and human
+     * comments already handed to a saddle (each delivered exactly once;
+     * a restart re-delivers and the daemon deduplicates), plus the
+     * plan-10 ask/wake liveness bookkeeping: the outcome of a saddle's
+     * last `POST /api/saddle/next-task` ask (a saddle counts as SLEEPING
+     * when its last ask returned 204) and the pending wake each sleeping
+     * saddle was sent (the lost-wake guard's timer), and the phase-2
+     * push-liveness clock: the last chunks/results arrival per session
+     * (the staleness probe's timer — one exceptional probe per silent
+     * episode, reset by the next push).
      *
      * A singleton by necessity: {@link SaddleService} is request-scoped,
      * so this object is where the cross-request cursors live. Ids are
@@ -26,9 +28,6 @@ namespace ClankerSaddle.Services {
         private Dictionary<string, string> probe_activity = new Dictionary<string, string>();
         private Dictionary<string, string> probe_contact = new Dictionary<string, string>();
         private Dictionary<string, string> last_stored_output_id = new Dictionary<string, string>();
-        private Dictionary<string, string> last_stored_raw_seq = new Dictionary<string, string>();
-        private Dictionary<string, string> last_raw_probe_ms = new Dictionary<string, string>();
-        private HashSet<string> raw_polling_started = new HashSet<string>();
         private HashSet<string> delivered_answers = new HashSet<string>();
         private HashSet<string> delivered_comments = new HashSet<string>();
         private HashSet<string> resumed_notified = new HashSet<string>();
@@ -36,6 +35,9 @@ namespace ClankerSaddle.Services {
         private HashSet<string> ask_seen = new HashSet<string>();
         private Dictionary<string, string> wake_at_ms = new Dictionary<string, string>();
         private Dictionary<string, string> wake_task = new Dictionary<string, string>();
+        private Dictionary<string, string> push_at_ms = new Dictionary<string, string>();
+        private HashSet<string> push_seen = new HashSet<string>();
+        private HashSet<string> stale_probed = new HashSet<string>();
         private Mutex state_lock = Mutex();
 
         /** The last output entry id stored for `session_id` (0 when none). */
@@ -54,46 +56,6 @@ namespace ClankerSaddle.Services {
             state_lock.unlock();
         }
 
-        /**
-         * Starts raw-log polling for `session_id` from the baseline seq
-         * (the daemon's current `next`, read before the assignment is
-         * pushed). Later calls for an already-started session are
-         * ignored — a resumed session continues from its stored cursor.
-         */
-        public void start_raw_polling(int64 session_id, int64 baseline_seq) {
-            state_lock.lock();
-            string key = session_id.to_string();
-            if (raw_polling_started.add(key)) {
-                last_stored_raw_seq.set(key, baseline_seq.to_string());
-            }
-            state_lock.unlock();
-        }
-
-        /**
-         * The last raw-log seq stored for `session_id`, or null when
-         * raw polling never started (a daemon without GET /log).
-         */
-        public int64? raw_since_for(int64 session_id) {
-            state_lock.lock();
-            string stored = "";
-            bool found = last_stored_raw_seq.try_get(session_id.to_string(), out stored)
-                && raw_polling_started.has(session_id.to_string());
-            state_lock.unlock();
-            // A ternary with a boxed null crashes valac's codegen here;
-            // branch explicitly instead.
-            if (found) {
-                return int64.parse(stored);
-            }
-            return null;
-        }
-
-        /** Records the new high-water raw-log seq for `session_id`. */
-        public void store_raw_since(int64 session_id, int64 seq) {
-            state_lock.lock();
-            last_stored_raw_seq.set(session_id.to_string(), seq.to_string());
-            state_lock.unlock();
-        }
-
         /**
          * Records a daemon /status observation for the saddles-page
          * health panel: reported state, uptime, last-activity epoch and
@@ -123,29 +85,6 @@ namespace ClankerSaddle.Services {
             state_lock.unlock();
         }
 
-        /**
-         * True when a fresh raw-log capability probe for `session_id` is
-         * due: never probed, or the last probe is older than
-         * `interval_ms` (a single transport blip at seed time must not
-         * permanently strip the session's raw terminal view).
-         */
-        public bool raw_probe_due(int64 session_id, uint interval_ms) {
-            state_lock.lock();
-            string stored = "";
-            bool due = !last_raw_probe_ms.try_get(session_id.to_string(), out stored)
-                || (get_monotonic_time() / 1000 - int64.parse(stored)) >= interval_ms;
-            state_lock.unlock();
-            return due;
-        }
-
-        /** Records a raw-log capability probe attempt for `session_id`. */
-        public void mark_raw_probe(int64 session_id) {
-            state_lock.lock();
-            last_raw_probe_ms.set(session_id.to_string(),
-                (get_monotonic_time() / 1000).to_string());
-            state_lock.unlock();
-        }
-
         /**
          * Marks `session_id`'s daemon-side resume as notified; returns
          * false when the progress entry already went out.
@@ -249,9 +188,7 @@ namespace ClankerSaddle.Services {
         /**
          * True once `saddle_id` has ever asked (post-flip daemon): it
          * asks whenever idle-awake, so its work — fresh or resumed —
-         * always flows through the allocation, and the /answer and
-         * /start push paths leave it alone (except a daemon blocked in
-         * its answer loop, which cannot ask and still takes /answer).
+         * always flows through the allocation.
          */
         public bool has_ever_asked(int64 saddle_id) {
             state_lock.lock();
@@ -285,8 +222,8 @@ namespace ClankerSaddle.Services {
         /**
          * True when `saddle_id` has a pending wake at least
          * `guard_ms` old — an ask should have arrived by now; the
-         * guard declares the saddle offline and frees the work for the
-         * next capable one.
+         * guard declares the saddle offline and frees the work for
+         * the next capable one.
          */
         public bool wake_overdue(int64 saddle_id, uint guard_ms) {
             state_lock.lock();
@@ -319,5 +256,54 @@ namespace ClankerSaddle.Services {
             state_lock.unlock();
             return found ? int64.parse(stored) : 0;
         }
+
+        /**
+         * Records a chunks/results push arriving for `session_id`
+         * (plan 10 phase 2): the running daemon is alive, and any
+         * running staleness episode ends — a later silent episode
+         * earns its own exceptional probe.
+         */
+        public void note_push(int64 session_id) {
+            state_lock.lock();
+            string key = session_id.to_string();
+            push_at_ms.set(key, (get_monotonic_time() / 1000).to_string());
+            push_seen.add(key);
+            stale_probed.remove(key);
+            state_lock.unlock();
+        }
+
+        /**
+         * Atomically decides the one exceptional staleness probe for
+         * `session_id`: true exactly when pushes have been seen, the
+         * silence has reached `stale_ms` and no probe fired for this
+         * episode yet (the flag flips here, so a slow probe is never
+         * double-fired by the next tick). A session with no recorded
+         * push (a web restart wiped the clock) is seeded now and
+         * starts its window instead of firing immediately — a fresh
+         * baseline must not masquerade as 90 s of silence.
+         */
+        public bool take_stale_probe(int64 session_id, uint stale_ms) {
+            state_lock.lock();
+            string key = session_id.to_string();
+            int64 now_ms = get_monotonic_time() / 1000;
+            if (!push_seen.has(key)) {
+                push_seen.add(key);
+                push_at_ms.set(key, now_ms.to_string());
+                state_lock.unlock();
+                return false;
+            }
+            if (stale_probed.has(key)) {
+                state_lock.unlock();
+                return false;
+            }
+            string stored = "";
+            bool known = push_at_ms.try_get(key, out stored);
+            bool due = known && (now_ms - int64.parse(stored)) >= stale_ms;
+            if (due) {
+                stale_probed.add(key);
+            }
+            state_lock.unlock();
+            return due;
+        }
     }
 }

src/Services/SubtaskEventQueue.vala

@@ -17,7 +17,8 @@ namespace ClankerSaddle.Services {
     /**
      * Process-wide queue of subtask status events for daemons hosting
      * the parent's live or parked session (wire contract:
-     * `POST /subtask-status`). A singleton by necessity —
+     * `POST /task-update {kind: "subtask-status"}`). A singleton by
+     * necessity —
      * {@link TaskService} records events from any request or poll while
      * {@link SaddleService} drains them on the dispatcher tick, mirroring
      * the pending-comment bookkeeping in {@link SaddlePollState}.

src/Services/TaskService.vala

@@ -91,8 +91,9 @@ namespace ClankerSaddle.Services {
 
         /**
          * Records a subtask's current state for the daemon hosting its
-         * parent's live or parked session (`POST /subtask-status`, wire
-         * contract). The wire carries the four guard-relevant values;
+         * parent's live or parked session (`POST /task-update
+         * {kind: "subtask-status"}`, wire contract). The wire carries
+         * the four guard-relevant values;
          * intermediate web states (assigned, blocked, review-requested,
          * suspended) all mean "still in flight" and map to `working`.
          * Only the latest state per subtask is kept (queue coalescing).
@@ -637,50 +638,6 @@ namespace ClankerSaddle.Services {
             yield pulse.bump(task_id, "dependency added");
         }
 
-        /**
-         * Reuses the task's previous push-failed session row on
-         * `saddle_id` (resetting it for the new attempt) instead of
-         * creating a fresh row per retry — the sessions 2/3 churn of the
-         * cancel-then-hang run. Null when no reusable row exists.
-         */
-        public async Session? reuse_push_failed_session(int64 task_id, int64 saddle_id) throws Error {
-            Session? candidate = null;
-            foreach (var row in yield db.query<Session>()
-                    .where(expr("task_id == $0 && saddle_id == $1",
-                        elem<int64?>(task_id), elem<int64?>(saddle_id)))
-                    .order_by(expr("id"))
-                    .materialise_async()) {
-                if (row.outcome == "push_failed") {
-                    candidate = row;
-                }
-            }
-            if (candidate == null) {
-                return null;
-            }
-            ((!)candidate).outcome = null;
-            ((!)candidate).ended_at = null;
-            ((!)candidate).started_at = new DateTime.now_local();
-            yield db.update_async<Session>((!)candidate);
-            return candidate;
-        }
-
-        /**
-         * Marks a session row `push_failed` (ended): the dispatch attempt
-         * produced no daemon session, and the next attempt reuses this
-         * row rather than churning a new one.
-         */
-        public async void mark_session_push_failed(int64 session_id) throws Error {
-            var row = yield db.query<Session>()
-                .where(expr("id == $0", elem<int64?>(session_id)))
-                .first_async();
-            if (row == null || ((!)row).ended_at != null) {
-                return;
-            }
-            ((!)row).outcome = "push_failed";
-            ((!)row).ended_at = new DateTime.now_local();
-            yield db.update_async<Session>((!)row);
-        }
-
         /**
          * Task ids currently `assigned`/`working` whose open session sits
          * on an offline or fenced saddle — the queue marks them

src/entrypoints/SaddlesEntrypoint.vala

@@ -142,13 +142,11 @@ namespace ClankerSaddle {
                 string health = "unknown";
                 if (saddle.status == "offline") {
                     health = "offline";
-                } else if (saddle.status == "recovering") {
-                    health = "fenced";
                 } else if (daemon_state.length > 0) {
                     health = "healthy";
                     if (daemon_state == "working" && activity > 0
                             && now_ms / 1000 - activity
-                                >= SaddleService.wedge_threshold_seconds()) {
+                                >= SaddleService.stale_after_seconds()) {
                         health = "stale";
                     }
                 }

src/main.vala

@@ -138,6 +138,10 @@ int main(string[] args) {
             new EndpointRoute("/api/saddle/{id}/poll", Method.GET));
         application.add_endpoint<SaddleNextTaskEndpoint>(
             new EndpointRoute("/api/saddle/next-task", Method.POST));
+        application.add_endpoint<SaddleChunksEndpoint>(
+            new EndpointRoute("/api/saddle/chunks", Method.POST));
+        application.add_endpoint<SaddleResultsEndpoint>(
+            new EndpointRoute("/api/saddle/results", Method.POST));
         application.add_endpoint<SaddleNoteEndpoint>(
             new EndpointRoute("/api/saddle/note", Method.POST));
         application.add_endpoint<SaddleSubtaskEndpoint>(