feat: cancel/requeue/stop reaches the daemon — POST /abort with narrated outcome; a failed abort fences the saddle as 'recovering' (not dispatchable) until an idle probe clears it or three failed probes escalate to offline

src/Services/Dispatcher.vala

@@ -115,6 +115,7 @@ namespace ClankerSaddle.Services {
                 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();
                 var ready = yield tasks.get_ready_tasks();
                 foreach (var task in ready) {
                     var saddle = yield pick_idle_capable_saddle(saddles, capabilities, task);

src/Services/SaddleService.vala

@@ -71,6 +71,9 @@ namespace ClankerSaddle.Services {
     public class SaddleService : Object {
 
         private OrmSession db = inject<OrmSession>();
+        /** Consecutive failed probes per fenced saddle (string-keyed). */
+        private Invercargill.DataStructures.Dictionary<string, string> fence_strikes =
+            new Invercargill.DataStructures.Dictionary<string, string>();
         /**
          * Every daemon request shares this session and therefore its
          * timeouts: without them a single hung request (a daemon wedged
@@ -274,6 +277,109 @@ namespace ClankerSaddle.Services {
                 .materialise_async();
         }
 
+        /**
+         * Cancelling / requeueing / stopping a task with a live session
+         * now reaches its daemon: `POST /abort` (authoritative stop, the
+         * workspace is kept on disk per the park contract). Success is
+         * narrated as "<verb>; daemon aborted"; a transport failure
+         * FENCES the saddle as 'recovering' — not dispatchable, reason
+         * visible — until a probe shows the daemon idle again (or the
+         * fence escalates to offline after three failed probes). The
+         * daemon must never keep running a task the web has moved on
+         * from as a zombie.
+         */
+        public async void abort_live_session(TaskService tasks, int64 task_id,
+                string verb) throws Error {
+            var session = yield tasks.open_session_for_task(task_id);
+            if (session == null) {
+                return;
+            }
+            var saddle = yield get_saddle(((!)session).saddle_id);
+            if (saddle == null) {
+                return;
+            }
+
+            var payload = new Json.Builder();
+            payload.begin_object();
+            payload.end_object();
+            Json.Node? root = null;
+            try {
+                int status_code = yield request_saddle(((!)saddle).host, "POST",
+                    "/abort", (owned) payload, out root);
+                if (status_code >= 400) {
+                    throw new Error(Quark.from_string("clanker-saddle"), 45,
+                        @"abort returned HTTP $(status_code)");
+                }
+                yield tasks.add_comment(task_id, "bot", @"$(verb); daemon aborted");
+            } catch (Error e) {
+                warning(@"[ClankerSaddle] Abort for task $(task_id) failed: $(e.message); fencing saddle '$(((!)saddle).name)'");
+                ((!)saddle).status = "recovering";
+                ((!)saddle).current_task_id = 0;
+                ((!)saddle).allocated_at = null;
+                yield db.update_async<Saddle>((!)saddle);
+                yield pulse.bump(0, "saddle fenced");
+                yield tasks.add_comment(task_id, "bot",
+                    @"$(verb); daemon unreachable — saddle fenced");
+            }
+        }
+
+        /**
+         * {@link abort_live_session} keyed by saddle instead of task —
+         * the unregister path (whatever task the saddle's open session
+         * holds).
+         */
+        public async void abort_saddle_session(TaskService tasks, int64 saddle_id,
+                string verb) throws Error {
+            var session = yield open_session_on_saddle(saddle_id);
+            if (session == null) {
+                return;
+            }
+            yield abort_live_session(tasks, ((!)session).task_id, verb);
+        }
+
+        /**
+         * Fenced ('recovering') saddles are probed each tick: an idle
+         * daemon clears the fence; three consecutive failed probes
+         * escalate to offline (the normal strike escalation).
+         */
+        public async void recover_fenced_saddles() throws Error {
+            foreach (var saddle in yield db.query<Saddle>()
+                    .where(expr("status == $0", elem("recovering")))
+                    .materialise_async()) {
+                try {
+                    var status = yield request_saddle_json(saddle.host, "GET", "/status", null);
+                    string state = status.has_member("state")
+                        ? status.get_string_member("state") : "";
+                    if (state == "idle") {
+                        saddle.status = "idle";
+                        saddle.current_task_id = 0;
+                        saddle.allocated_at = null;
+                        yield db.update_async<Saddle>(saddle);
+                        fence_strikes.set(saddle.id.to_string(), "0");
+                        print(@"[ClankerSaddle] Fence cleared on saddle '$(saddle.name)' (daemon idle)\n");
+                        yield pulse.bump(0, "saddle fence cleared");
+                    }
+                    // Still busy (abort racing a finishing round, or the
+                    // daemon wedged): the next tick re-probes.
+                } catch (Error e) {
+                    string stored = "";
+                    int count = fence_strikes.try_get(saddle.id.to_string(), out stored)
+                        ? int.parse(stored) : 0;
+                    count += 1;
+                    fence_strikes.set(saddle.id.to_string(), count.to_string());
+                    if (count >= 3) {
+                        saddle.status = "offline";
+                        saddle.current_task_id = 0;
+                        saddle.allocated_at = null;
+                        yield db.update_async<Saddle>(saddle);
+                        fence_strikes.set(saddle.id.to_string(), "0");
+                        warning(@"[ClankerSaddle] Fenced saddle '$(saddle.name)' escalated to offline after $(count) failed probes");
+                        yield pulse.bump(0, "saddle fence escalated");
+                    }
+                }
+            }
+        }
+
         /**
          * Heals stale-busy saddles (the round-3 stranding: `saddle-1`
          * left busy with `current_task_id = 0` forever, so nothing ever

src/actions/SaddleActions.vala

@@ -51,6 +51,7 @@ namespace ClankerSaddle {
     public class UnregisterSaddleAction : TypedStatumAction<SaddleActionPrivate> {
 
         private SaddleService saddles = inject<SaddleService>();
+        private TaskService tasks = inject<TaskService>();
         private UserService users = inject<UserService>();
 
         protected override Type private_type { get { return typeof(SaddleActionPrivate); } }
@@ -68,6 +69,7 @@ namespace ClankerSaddle {
             }
 
             try {
+                yield saddles.abort_saddle_session(tasks, ((!)ctx).saddle_id, "saddle removed");
                 yield saddles.unregister_saddle(((!)ctx).saddle_id);
                 return directives().notify("success", "Saddle removed").navigate("/saddles");
             } catch (Error e) {

src/actions/TaskActions.vala

@@ -364,6 +364,7 @@ namespace ClankerSaddle {
     public class CancelTaskAction : SealedTaskAction {
 
         private TaskService tasks = inject<TaskService>();
+        private SaddleService saddles = inject<SaddleService>();
         private UserService users = inject<UserService>();
 
         public override async DirectiveBuilder handle() throws GLib.Error {
@@ -378,6 +379,7 @@ namespace ClankerSaddle {
             }
 
             try {
+                yield saddles.abort_live_session(tasks, task_id, "cancelled");
                 yield tasks.cancel_task(task_id);
                 return directives().notify("success", "Task cancelled").navigate(@"/tasks/$(task_id)");
             } catch (Error e) {
@@ -395,6 +397,7 @@ namespace ClankerSaddle {
     public class RetryTaskAction : SealedTaskAction {
 
         private TaskService tasks = inject<TaskService>();
+        private SaddleService saddles = inject<SaddleService>();
         private UserService users = inject<UserService>();
 
         public override async DirectiveBuilder handle() throws GLib.Error {
@@ -418,6 +421,7 @@ namespace ClankerSaddle {
             }
 
             try {
+                yield saddles.abort_live_session(tasks, task_id, "retried");
                 yield tasks.retry_task(task_id);
                 return directives().notify("success", "Task requeued for another attempt")
                     .navigate(@"/tasks/$(task_id)");
@@ -454,6 +458,7 @@ namespace ClankerSaddle {
             try {
                 var session = yield task_service.open_session_for_task(task_id);
                 if (session != null) {
+                    yield saddle_service.abort_live_session(task_service, task_id, "stopped");
                     yield saddle_service.mark_saddle_unreachable(((!)session).saddle_id, task_service);
                 } else {
                     yield task_service.requeue_task(task_id);

test/web_stack_test.py

@@ -216,6 +216,9 @@ class FakeSaddle:
         self.fail_subtask_status = False  # exercise the web's retry path
         self.hang_next = {}              # path suffix -> hang count (never responds)
         self.fail_log_count = 0          # /log requests failing (500) while > 0
+        self.abort_failures = 0          # /abort requests failing (500) while > 0
+        self.abort_log = []              # task ids aborted via POST /abort
+        self.running_task = 0            # task id the daemon believes it runs
 
     def set_state(self, state, summary=None):
         """Advances the scripted lifecycle.
@@ -290,6 +293,7 @@ class FakeSaddle:
                     payload = {"raw": raw}
                 self.daemon.start_payloads.append(payload)
                 task_id = (payload.get("task") or {}).get("id")
+                self.daemon.running_task = int(task_id or 0)
                 count = self.daemon.start_counts.get(task_id, 0) + 1
                 self.daemon.start_counts[task_id] = count
                 if count > 1:
@@ -305,6 +309,17 @@ class FakeSaddle:
                 self._send(200, {})
             elif self.path == "/answer":
                 self._send(200, {})
+            elif self.path == "/abort":
+                if self.daemon.abort_failures > 0:
+                    self.daemon.abort_failures -= 1
+                    self._send(500, {"error": "abort failed (simulated)"})
+                    return
+                aborted = self.daemon.running_task
+                self.daemon.set_state("idle")
+                self.daemon.summary = ""
+                self.daemon.running_task = 0
+                self.daemon.abort_log.append(aborted)
+                self._send(200, {"ok": True, "aborted_task_id": aborted})
             elif self.path == "/subtask-status":
                 try:
                     payload = json.loads(raw)
@@ -1898,6 +1913,66 @@ def main():
         check(wait_task_done(auth, stranded, timeout=90) is not None,
               "recovered task dispatches and completes")
 
+
+        print("== R7 B1: cancel reaches the daemon ==")
+        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
+            "project_id": "1",
+            "title": "Cancel live",
+            "priority": "2",
+        })
+        cancel_live = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
+                           if t["title"] == "Cancel live"][0])
+        check(wait_for(lambda: task_state(auth, cancel_live).get("status") == "working"
+                       or False, timeout=60, what="cancel-live working") is not None,
+              "task reaches working before the cancel")
+        act(auth, task_state(auth, cancel_live)["cancel"], {})
+        check(wait_for(lambda: cancel_live in fake.abort_log or False,
+                       timeout=30, what="daemon abort recorded") is not None,
+              "cancelling POSTs /abort to the daemon")
+        check(any(c["body"] == "cancelled; daemon aborted"
+                  for c in task_state(auth, cancel_live).get("comments", [])),
+              "the abort outcome is narrated")
+        check(task_state(auth, cancel_live)["status"] == "failed",
+              "cancelled task lands failed as before")
+
+        # Transport/500 failure on abort fences the saddle; recovery probe
+        # clears the fence; nothing dispatches while fenced.
+        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
+            "project_id": "1",
+            "title": "Fence runner",
+            "priority": "2",
+        })
+        fence_runner = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
+                            if t["title"] == "Fence runner"][0])
+        check(wait_for(lambda: task_state(auth, fence_runner).get("status") == "working"
+                       or False, timeout=60, what="fence runner working") is not None,
+              "second task reaches working")
+        fake.abort_failures = 1
+        act(auth, task_state(auth, fence_runner)["cancel"], {})
+        check(any(c["body"] == "cancelled; daemon unreachable — saddle fenced"
+                  for c in task_state(auth, fence_runner).get("comments", [])),
+              "failed abort narrates the fence")
+        check(saddle_card("fake-saddle")["status"] == "recovering",
+              "saddle fenced as recovering")
+        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
+            "project_id": "1",
+            "title": "During fence",
+            "priority": "3",
+        })
+        during_fence = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
+                            if t["title"] == "During fence"][0])
+        time.sleep(7)
+        check(task_state(auth, during_fence)["status"] == "queued",
+              "fenced saddle is not dispatchable")
+        # A real daemon lands idle once its (unaborted) round ends; the
+        # fake's scripted lifecycle parks on 'done', so model that.
+        fake.state = "idle"
+        check(wait_for(lambda: saddle_card("fake-saddle")["status"] == "idle" or False,
+                       timeout=30, what="fence clears") is not None,
+              "an idle probe clears the fence")
+        check(wait_task_done(auth, during_fence, timeout=60) is not None,
+              "queued work dispatches once the fence clears")
+
     finally:
         # Exit-code diagnostic: distinguishes a crashed app (negative
         # returncode, e.g. -11 SIGSEGV) from a clean shutdown. When