feat: subtask status push — POST /subtask-status to the parent's live/parked-session daemon on creation and every status change, coalescing best-effort queue retried per tick (mirrors pending comments)

meson.build

@@ -121,6 +121,7 @@ app_sources = files(
     'src/Services/Dispatcher.vala',
     'src/Services/RuleService.vala',
     'src/Services/CapabilityService.vala',
+    'src/Services/SubtaskEventQueue.vala',
     'src/Api/SaddleApi.vala',
     'src/Api/TerminalSse.vala',
     'src/actions/Forms.vala',

src/Services/Dispatcher.vala

@@ -86,6 +86,7 @@ namespace ClankerSaddle.Services {
             var saddles = scope.resolve<SaddleService>();
             var projects = scope.resolve<ProjectService>();
             var capabilities = scope.resolve<CapabilityService>();
+            var subtask_events = scope.resolve<SubtaskEventQueue>();
 
             if (LlmEndpoint.from_environment() == null) {
                 if (!warned_no_llm) {
@@ -127,6 +128,7 @@ namespace ClankerSaddle.Services {
                     task.started_at = new DateTime.now_local();
                     yield orm.update_async<TaskEntity>(task);
                     yield saddles.update_saddle_status(((!)saddle).id, "busy", task.id);
+                    subtask_events.record_dispatch(task);
 
                     try {
                         yield saddles.seed_session_cursor(((!)saddle).id, session.id);
@@ -143,6 +145,7 @@ namespace ClankerSaddle.Services {
 
             yield resume_answered_tasks(tasks, saddles, projects);
             yield resume_suspended_parents(tasks, saddles, projects);
+            yield saddles.deliver_pending_subtask_events(tasks);
 
             foreach (var saddle in yield saddles.list_busy_saddles()) {
                 try {

src/Services/SaddleService.vala

@@ -75,6 +75,7 @@ namespace ClankerSaddle.Services {
         private SaddlePollState poll_state = inject<SaddlePollState>();
         private StatePulse pulse = inject<StatePulse>();
         private RuleService rules = inject<RuleService>();
+        private SubtaskEventQueue subtask_events = inject<SubtaskEventQueue>();
         private Astralis.WebConfig config = inject<Astralis.WebConfig>();
 
         /**
@@ -317,6 +318,7 @@ namespace ClankerSaddle.Services {
                     yield db.update_async<TaskEntity>((!)task);
                     yield tasks.add_comment(task_id, "bot",
                         "Saddle became unreachable; task requeued");
+                    tasks.note_requeued_subtask(((!)task));
                 }
 
                 var sessions = yield db.query<ClankerSaddle.Data.Session>()
@@ -856,6 +858,65 @@ namespace ClankerSaddle.Services {
             }
         }
 
+        /**
+         * Pushes pending subtask status events to the daemon hosting each
+         * parent's live or parked session — `POST /subtask-status`
+         * (wire contract, no auth like `/answer` and `/notify`), so the
+         * daemon's live subtask set stays current for its `done` and
+         * `suspend` guards. Best-effort mirror of
+         * {@link deliver_pending_comments}: a parent without an open
+         * session drops its events (its next `/start` carries the full
+         * `subtasks` array), a daemon without the route (404/405) does
+         * too, and transport failures stay queued for the next
+         * dispatcher tick. The queue coalesces to the latest state per
+         * subtask, so retries never duplicate stale events.
+         */
+        public async void deliver_pending_subtask_events(TaskService tasks) throws Error {
+            foreach (var event in subtask_events.snapshot()) {
+                var session = yield tasks.open_session_for_task(event.parent_task_id);
+                if (session == null) {
+                    subtask_events.discard(event.subtask_id);
+                    continue;
+                }
+                var saddle = yield get_saddle(((!)session).saddle_id);
+                if (saddle == null) {
+                    subtask_events.discard(event.subtask_id);
+                    continue;
+                }
+
+                var payload = new Json.Builder();
+                payload.begin_object();
+                payload.set_member_name("parent_task_id");
+                payload.add_int_value(event.parent_task_id);
+                payload.set_member_name("subtask_id");
+                payload.add_int_value(event.subtask_id);
+                payload.set_member_name("title");
+                payload.add_string_value(event.title);
+                payload.set_member_name("status");
+                payload.add_string_value(event.status);
+                if (event.summary.length > 0) {
+                    payload.set_member_name("summary");
+                    payload.add_string_value(event.summary);
+                }
+                payload.end_object();
+
+                Json.Node? root = null;
+                int status_code = yield request_saddle(((!)saddle).host, "POST",
+                    "/subtask-status", (owned) payload, out root);
+                if (status_code == 200 || status_code == 204) {
+                    subtask_events.discard(event.subtask_id);
+                } else if (status_code == 404 || status_code == 405) {
+                    // Older daemon without the route: its subtask set
+                    // comes only from /start payloads.
+                    subtask_events.discard(event.subtask_id);
+                    debug(@"[ClankerSaddle] Saddle $(((!)saddle).id) has no /subtask-status route");
+                } else {
+                    warning(@"[ClankerSaddle] Subtask status push for $(event.subtask_id)"
+                        + @" to saddle $(((!)saddle).id) returned HTTP $(status_code); retrying next tick");
+                }
+            }
+        }
+
         /**
          * Marks the task's human comments as delivered — they were handed
          * over inside an assignment payload, so neither the poll endpoint

src/Services/SubtaskEventQueue.vala

@@ -0,0 +1,101 @@
+using Invercargill;
+using Invercargill.DataStructures;
+using Inversion;
+using ClankerSaddle.Data;
+
+namespace ClankerSaddle.Services {
+
+    /** One subtask status event awaiting delivery to the parent's daemon. */
+    public class SubtaskStatusEvent : Object {
+        public int64 parent_task_id { get; set; }
+        public int64 subtask_id { get; set; }
+        public string title { get; set; default = ""; }
+        public string status { get; set; default = ""; }
+        public string summary { get; set; default = ""; }
+    }
+
+    /**
+     * 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 —
+     * {@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}.
+     *
+     * Coalescing: only the LATEST event per subtask stays queued — a
+     * burst of transitions collapses to what the parent's daemon needs
+     * (the current state), keeping the queue bounded by subtask count.
+     * Ids are stringified: the Invercargill collections cannot key on
+     * value types.
+     */
+    public class SubtaskEventQueue : Object {
+
+        private Vector<SubtaskStatusEvent> pending = new Vector<SubtaskStatusEvent>();
+        private HashSet<string> queued_ids = new HashSet<string>();
+        private Mutex queue_lock = Mutex();
+
+        /**
+         * Enqueues `event`, replacing any earlier event for the same
+         * subtask (the replacement moves to the back — newest last).
+         */
+        public void enqueue(SubtaskStatusEvent event) {
+            queue_lock.lock();
+            string key = event.subtask_id.to_string();
+            if (queued_ids.has(key)) {
+                pending.remove_first_where(e => e.subtask_id == event.subtask_id);
+            } else {
+                queued_ids.add(key);
+            }
+            pending.add(event);
+            queue_lock.unlock();
+        }
+
+        /** A copy of the pending events, oldest first. */
+        public ImmutableLot<SubtaskStatusEvent> snapshot() {
+            queue_lock.lock();
+            var copy = new Vector<SubtaskStatusEvent>();
+            foreach (var event in pending) {
+                copy.add(event);
+            }
+            queue_lock.unlock();
+            return copy.to_immutable_buffer();
+        }
+
+        /** Removes the (delivered or dropped) event for `subtask_id`. */
+        public void discard(int64 subtask_id) {
+            queue_lock.lock();
+            pending.remove_first_where(e => e.subtask_id == subtask_id);
+            queued_ids.remove(subtask_id.to_string());
+            queue_lock.unlock();
+        }
+
+        /** The number of distinct subtasks awaiting delivery. */
+        public uint length {
+            get {
+                queue_lock.lock();
+                uint count = pending.length;
+                queue_lock.unlock();
+                return count;
+            }
+        }
+
+        /**
+         * Records a dispatched subtask (the dispatcher writes the
+         * assigned transition directly, bypassing
+         * {@link TaskService.update_task} — this is its hook). The wire
+         * value is `working`.
+         */
+        public void record_dispatch(TaskEntity subtask) {
+            if (subtask.parent_task_id <= 0) {
+                return;
+            }
+            enqueue(new SubtaskStatusEvent() {
+                parent_task_id = subtask.parent_task_id,
+                subtask_id = subtask.id,
+                title = subtask.title,
+                status = "working",
+                summary = subtask.summary ?? ""
+            });
+        }
+    }
+}

src/Services/TaskService.vala

@@ -85,6 +85,33 @@ namespace ClankerSaddle.Services {
 
         private OrmSession db = inject<OrmSession>();
         private StatePulse pulse = inject<StatePulse>();
+        private SubtaskEventQueue subtask_events = inject<SubtaskEventQueue>();
+
+        /**
+         * 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;
+         * 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).
+         */
+        private void note_subtask_event(TaskEntity task) {
+            if (task.parent_task_id <= 0) {
+                return;
+            }
+            string wire_status = STATUS_WORKING;
+            if (task.status == STATUS_QUEUED || task.status == STATUS_DONE
+                    || task.status == STATUS_FAILED) {
+                wire_status = task.status;
+            }
+            subtask_events.enqueue(new SubtaskStatusEvent() {
+                parent_task_id = task.parent_task_id,
+                subtask_id = task.id,
+                title = task.title,
+                status = wire_status,
+                summary = task.summary ?? ""
+            });
+        }
 
         /**
          * Inserts a task (status 'queued'). `parent_task_id`,
@@ -129,9 +156,19 @@ namespace ClankerSaddle.Services {
             };
             yield db.insert_async<TaskEntity>(task);
             yield pulse.bump(task.id, "task created");
+            note_subtask_event(task);
             return task;
         }
 
+        /**
+         * Public requeue note for callers that write the queued status
+         * directly ({@link SaddleService.mark_saddle_unreachable}): the
+         * subtask's parent daemon hears about the state change.
+         */
+        public void note_requeued_subtask(TaskEntity task) {
+            note_subtask_event(task);
+        }
+
         /**
          * Loads the full aggregate for `id`: the task, its comments, blocks,
          * progress, child tasks, dependencies and sessions. Returns null
@@ -305,6 +342,9 @@ namespace ClankerSaddle.Services {
             yield db.update_async<TaskEntity>(task);
             if (changed) {
                 yield pulse.bump(id, @"task $(status ?? "saved")");
+                if (status != null && status.length > 0) {
+                    note_subtask_event(((!)task));
+                }
             }
             return task;
         }
@@ -635,6 +675,7 @@ namespace ClankerSaddle.Services {
             yield close_open_sessions(task_id, outcome);
             yield free_saddles_running(task_id);
             yield pulse.bump(task_id, "task completed");
+            note_subtask_event(((!)task));
         }
 
         /**
@@ -661,6 +702,7 @@ namespace ClankerSaddle.Services {
             yield close_open_sessions(task_id, "stopped");
             yield free_saddles_running(task_id);
             yield pulse.bump(task_id, "task cancelled");
+            note_subtask_event(((!)task));
         }
 
         /**
@@ -691,6 +733,7 @@ namespace ClankerSaddle.Services {
             yield close_open_sessions(task_id, "failed");
             yield free_saddles_running(task_id);
             yield pulse.bump(task_id, "task failed");
+            note_subtask_event(((!)task));
         }
 
         /**
@@ -792,6 +835,9 @@ namespace ClankerSaddle.Services {
             yield close_open_sessions(task_id, "stopped");
             yield free_saddles_running(task_id);
             yield pulse.bump(task_id, "task requeued");
+            if (task != null) {
+                note_subtask_event(((!)task));
+            }
         }
 
         /**
@@ -823,6 +869,7 @@ namespace ClankerSaddle.Services {
             yield free_saddles_running(task_id);
             yield add_comment(task_id, "bot", "Task retried by user");
             yield pulse.bump(task_id, "task retried");
+            note_subtask_event(task);
         }
 
         /**

src/main.vala

@@ -59,6 +59,7 @@ int main(string[] args) {
         application.add_scoped<SaddleService>();
         application.add_scoped<RuleService>();
         application.add_scoped<CapabilityService>();
+        application.add_singleton<SubtaskEventQueue>();
         application.add_singleton<SaddlePollState>();
         application.add_singleton<StatePulse>();
         application.add_singleton<Dispatcher>();

test/web_stack_test.py

@@ -212,6 +212,8 @@ class FakeSaddle:
         self.suspend_after_start = False  # next task parks as suspended
         self.pending_output = []        # entries drained by the next /output poll
         self.output_seq = 100
+        self.subtask_status_events = []  # recorded POST /subtask-status payloads
+        self.fail_subtask_status = False  # exercise the web's retry path
 
     def queue_output(self, entry_type, command, output):
         """Queues a typed entry for the dispatcher's next /output poll."""
@@ -280,6 +282,17 @@ class FakeSaddle:
                 self._send(200, {})
             elif self.path == "/answer":
                 self._send(200, {})
+            elif self.path == "/subtask-status":
+                try:
+                    payload = json.loads(raw)
+                except ValueError:
+                    self._send(400, {"error": "bad json"})
+                    return
+                if self.daemon.fail_subtask_status:
+                    self._send(500, {"error": "simulated failure"})
+                    return
+                self.daemon.subtask_status_events.append(payload)
+                self._send(200, {"ok": True})
             else:
                 self._send(404, {"error": "no such route"})
 
@@ -1201,6 +1214,70 @@ def main():
 
 
 
+
+        print("== R3 B1: subtask status push ==")
+        # A parked (suspended) parent gives delivery a stable open session.
+        fake.suspend_after_start = True
+        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
+            "project_id": "1",
+            "title": "Push parent",
+            "priority": "2",
+        })
+        push_parent = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
+                           if t["title"] == "Push parent"][0])
+        check(wait_for(lambda: task_state(auth, push_parent).get("status") == "suspended"
+                       or False, timeout=60, what="push parent parks") is not None,
+              "push parent parks with an open session")
+
+        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
+            "project_id": "1",
+            "title": "Push child",
+            "parent_task_id": str(push_parent),
+        })
+        push_child = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
+                          if t["title"] == "Push child"][0])
+
+        def events_for(subtask_id):
+            return [e for e in fake.subtask_status_events
+                    if int(e.get("subtask_id") or 0) == subtask_id]
+
+        check(wait_for(lambda: len(events_for(push_child)) > 0 or False,
+                       timeout=30, what="queued subtask event") is not None,
+              "subtask creation pushes a queued status event")
+        first_event = events_for(push_child)[0]
+        check(int(first_event.get("parent_task_id") or 0) == push_parent
+              and first_event.get("title") == "Push child"
+              and first_event.get("status") in ("queued", "working"),
+              "status event carries parent id, title and a wire status (%s)"
+              % first_event.get("status"))
+
+        # Real lifecycle: the child dispatches (working) and completes (done
+        # with summary) — the parent's daemon hears every transition.
+        check(wait_for(lambda: any(e.get("status") == "done" and "finished" in (e.get("summary") or "")
+                                   for e in events_for(push_child)) or False,
+                       timeout=90, what="done subtask event") is not None,
+              "subtask completion pushes done with the summary")
+        check(any(e.get("status") == "working" for e in events_for(push_child)),
+              "dispatch pushes a working status event")
+
+        # Best-effort retry: while the daemon 500s, events stay queued and
+        # land once it recovers.
+        fake.fail_subtask_status = True
+        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
+            "project_id": "1",
+            "title": "Push child two",
+            "parent_task_id": str(push_parent),
+        })
+        push_child_two = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
+                              if t["title"] == "Push child two"][0])
+        time.sleep(7)
+        check(len(events_for(push_child_two)) == 0,
+              "failed pushes stay queued (no event delivered)")
+        fake.fail_subtask_status = False
+        check(wait_for(lambda: len(events_for(push_child_two)) > 0 or False,
+                       timeout=30, what="retried subtask event") is not None,
+              "recovered pushes deliver on a later tick")
+
     finally:
         app.terminate()
         try: