feat: 409-aware dispatch — machine-readable conflicts recover zombies (abort, bounded idle wait, one retry, narrated), live-task conflicts fence and alert; failed pushes mark the session push_failed and retries reuse the row instead of churning new ones

src/Services/Dispatcher.vala

@@ -138,12 +138,18 @@ namespace ClankerSaddle.Services {
                     }
                     capability_wait_logged.remove(task.id.to_string());
 
-                    var session = new Session() {
-                        task_id = task.id,
-                        saddle_id = ((!)saddle).id,
-                        started_at = new DateTime.now_local()
-                    };
-                    yield orm.insert_async<Session>(session);
+                    // A failed push marks its session push_failed; the
+                    // next attempt reuses that row instead of churning a
+                    // fresh one per retry.
+                    var session = yield tasks.reuse_push_failed_session(task.id, ((!)saddle).id);
+                    if (session == null) {
+                        session = new Session() {
+                            task_id = task.id,
+                            saddle_id = ((!)saddle).id,
+                            started_at = new DateTime.now_local()
+                        };
+                        yield orm.insert_async<Session>((!)session);
+                    }
 
                     task.status = STATUS_ASSIGNED;
                     task.started_at = new DateTime.now_local();
@@ -152,13 +158,22 @@ namespace ClankerSaddle.Services {
                     subtask_events.record_dispatch(task);
 
                     try {
-                        yield saddles.seed_session_cursor(((!)saddle).id, session.id);
-                        yield saddles.push_task_to_saddle(((!)saddle).id, task, session.id, projects);
+                        yield saddles.seed_session_cursor(((!)saddle).id, ((!)session).id);
+                        bool pushed = yield saddles.push_with_zombie_recovery(((!)saddle),
+                            task, ((!)session).id, projects, tasks);
+                        if (!pushed) {
+                            // Double-booked: the saddle is fenced and both
+                            // tasks carry comments; requeue this one.
+                            yield tasks.mark_session_push_failed(((!)session).id);
+                            yield tasks.requeue_task(task.id);
+                            continue;
+                        }
                         yield saddles.mark_task_comments_delivered(task.id);
                         record_strike(((!)saddle).id, 0);
-                        print(@"[ClankerSaddle] Dispatched task $(task.id) to saddle '$(((!)saddle).name)' (session $(session.id))\n");
+                        print(@"[ClankerSaddle] Dispatched task $(task.id) to saddle '$(((!)saddle).name)' (session $(((!)session).id))\n");
                     } catch (Error e) {
                         warning(@"[ClankerSaddle] Pushing task $(task.id) to saddle $(((!)saddle).id) failed: $(e.message)");
+                        yield tasks.mark_session_push_failed(((!)session).id);
                         yield abandon_saddle(((!)saddle).id, saddles, tasks,
                             @"assignment push failed: $(e.message)");
                     }

src/Services/SaddleService.vala

@@ -277,6 +277,114 @@ namespace ClankerSaddle.Services {
                 .materialise_async();
         }
 
+        /**
+         * {@link push_task_to_saddle} with 409-aware recovery: a busy
+         * daemon answering `/start` with a conflict is either running a
+         * task the web has moved on from (a zombie — abort it, wait for
+         * idle within ~30s, retry the push once, and comment the
+         * recovery) or a genuine double-booking against a task the web
+         * still considers live (fence the saddle and alert; never abort
+         * live work). Returns false only for the fenced double-booking;
+         * everything unrecoverable throws for the caller's abandon path.
+         */
+        public async bool push_with_zombie_recovery(Saddle saddle, TaskEntity task,
+                int64 session_id, ProjectService projects, TaskService tasks) throws Error {
+            try {
+                yield push_task_to_saddle(saddle.id, task, session_id, projects);
+                return true;
+            } catch (Error e) {
+                if (e.code != 409) {
+                    throw e;
+                }
+                int64 running_task_id = 0;
+                string conflict_state = "";
+                try {
+                    var parser = new Json.Parser();
+                    parser.load_from_data(e.message.has_prefix("busy: ")
+                        ? e.message.substring("busy: ".length) : "{}");
+                    var body = parser.get_root().get_object();
+                    if (body.has_member("running_task_id")
+                            && body.get_member("running_task_id").get_value_type() == typeof(int64)) {
+                        running_task_id = body.get_int_member("running_task_id");
+                    }
+                    if (body.has_member("state")) {
+                        conflict_state = body.get_string_member("state");
+                    }
+                } catch (Error parse_error) {
+                    // Unparseable conflict: treat as a plain failure.
+                    throw e;
+                }
+
+                var running = running_task_id > 0
+                    ? yield tasks.get_task(running_task_id) : null;
+                bool live = running != null
+                    && ((!)running).task.status != STATUS_DONE
+                    && ((!)running).task.status != STATUS_FAILED;
+                if (live) {
+                    warning(@"[ClankerSaddle] Saddle '$(saddle.name)' double-booked: daemon runs task $(running_task_id) ($(conflict_state)) which the web still considers live; fencing");
+                    saddle.status = "recovering";
+                    saddle.current_task_id = 0;
+                    saddle.allocated_at = null;
+                    yield db.update_async<Saddle>(saddle);
+                    yield tasks.add_comment(running_task_id, "bot",
+                        @"saddle '$(saddle.name)' was double-booked (another dispatch attempted while it runs this task); saddle fenced");
+                    yield tasks.add_comment(task.id, "bot",
+                        @"dispatch refused: saddle '$(saddle.name)' runs task $(running_task_id); saddle fenced");
+                    yield pulse.bump(0, "saddle double-booked");
+                    return false;
+                }
+
+                // Zombie: the daemon runs a task the web has buried.
+                print(@"[ClankerSaddle] Recovering zombie session on saddle '$(saddle.name)' (daemon ran task $(running_task_id))\n");
+                var abort_payload = new Json.Builder();
+                abort_payload.begin_object();
+                abort_payload.end_object();
+                Json.Node? abort_root = null;
+                int abort_code = yield request_saddle(saddle.host, "POST", "/abort",
+                    (owned) abort_payload, out abort_root);
+                if (abort_code >= 400) {
+                    throw new Error(Quark.from_string("clanker-saddle"), 46,
+                        @"zombie abort returned HTTP $(abort_code)");
+                }
+                // Bounded idle wait (~30s), then one retry.
+                bool idle = false;
+                for (int attempt = 0; attempt < 15 && !idle; attempt++) {
+                    var status = yield request_saddle_json(saddle.host, "GET", "/status", null);
+                    idle = status.has_member("state")
+                        && status.get_string_member("state") == "idle";
+                    if (!idle) {
+                        GLib.Timeout.add(2000, push_with_zombie_recovery.callback);
+                        yield;
+                    }
+                }
+                if (!idle) {
+                    throw new Error(Quark.from_string("clanker-saddle"), 47,
+                        "daemon never went idle after the zombie abort");
+                }
+                try {
+                    yield push_task_to_saddle(saddle.id, task, session_id, projects);
+                } catch (Error retry_error) {
+                    if (retry_error.code != 409) {
+                        throw retry_error;
+                    }
+                    // The daemon picked something up between the abort and
+                    // the retry — treat like a double-booking.
+                    warning(@"[ClankerSaddle] Saddle '$(saddle.name)' busy again after zombie recovery; fencing");
+                    saddle.status = "recovering";
+                    saddle.current_task_id = 0;
+                    saddle.allocated_at = null;
+                    yield db.update_async<Saddle>(saddle);
+                    yield tasks.add_comment(task.id, "bot",
+                        @"saddle '$(saddle.name)' became busy again after the zombie recovery; saddle fenced");
+                    yield pulse.bump(0, "saddle fenced after retry");
+                    return false;
+                }
+                yield tasks.add_comment(task.id, "bot",
+                    @"recovered zombie session on saddle '$(saddle.name)' (daemon was still running task $(running_task_id))");
+                return true;
+            }
+        }
+
         /**
          * Cancelling / requeueing / stopping a task with a live session
          * now reaches its daemon: `POST /abort` (authoritative stop, the
@@ -1768,6 +1876,20 @@ namespace ClankerSaddle.Services {
                 owned Json.Builder? payload) throws Error {
             Json.Node? root = null;
             int status_code = yield request_saddle(host, method, path, (owned) payload, out root);
+            if (status_code == 409) {
+                // Machine-readable per the wire contract: the body (state,
+                // running_task_id) rides in the message so the dispatch
+                // path can distinguish zombie conflicts from transport
+                // failures and recover.
+                string body_text = "{}";
+                if (root != null && root.get_node_type() == Json.NodeType.OBJECT) {
+                    var generator = new Json.Generator();
+                    generator.set_root(root);
+                    body_text = generator.to_data(null);
+                }
+                throw new Error(Quark.from_string("clanker-saddle"), 409,
+                    @"busy: $(body_text)");
+            }
             if (status_code >= 400) {
                 throw new Error(Quark.from_string("clanker-saddle"), 15,
                     @"Saddle request returned HTTP $(status_code)");

src/Services/TaskService.vala

@@ -635,6 +635,50 @@ 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);
+        }
+
         /** Persists the auto-approve toggle for `task_id`. */
         public async void save_auto_approve(int64 task_id, bool enabled) throws Error {
             var task = yield db.query<TaskEntity>()

test/web_stack_test.py

@@ -219,6 +219,9 @@ class FakeSaddle:
         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
+        self.zombie_409 = 0              # /start 409s once per unit (then decrements)
+        self.conflict_running = 0        # while > 0, /start persistently 409s with this id
+        self.hold_state = False          # freeze /status lifecycle transitions
 
     def set_state(self, state, summary=None):
         """Advances the scripted lifecycle.
@@ -287,6 +290,21 @@ class FakeSaddle:
             length = int(self.headers.get("Content-Length") or 0)
             raw = self.rfile.read(length).decode() if length else "{}"
             if self.path == "/start":
+                if self.daemon.conflict_running > 0 or self.daemon.zombie_409 > 0:
+                    # Wire contract: a machine-readable conflict.
+                    running = self.daemon.conflict_running
+                    if running == 0:
+                        running = self.daemon.running_task
+                        self.daemon.zombie_409 -= 1
+                    body = json.dumps({"ok": False, "error": "busy",
+                                       "state": "working",
+                                       "running_task_id": running}).encode()
+                    self.send_response(409)
+                    self.send_header("Content-Type", "application/json")
+                    self.send_header("Content-Length", str(len(body)))
+                    self.end_headers()
+                    self.wfile.write(body)
+                    return
                 try:
                     payload = json.loads(raw)
                 except ValueError:
@@ -343,6 +361,9 @@ class FakeSaddle:
                 if self.daemon.resumed_pending:
                     response["resumed"] = True
                     self.daemon.resumed_pending = False
+                if self.daemon.hold_state:
+                    self._send(200, response)
+                    return
                 if self.daemon.state == "failing":
                     # Fails fast (a rejected /start script).
                     if time.monotonic() - self.daemon.state_entered >= 1:
@@ -1973,6 +1994,97 @@ def main():
         check(wait_task_done(auth, during_fence, timeout=60) is not None,
               "queued work dispatches once the fence clears")
 
+
+        print("== R7 B2: 409-aware dispatch ==")
+        # Zombie: the daemon (per the fake) still runs an old completed task;
+        # one 409, then abort + retry recovers without churning sessions.
+        fake.state = "idle"
+        fake.zombie_409 = 1
+        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
+            "project_id": "1",
+            "title": "Zombie recovery",
+            "priority": "2",
+        })
+        zombie_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
+                           if t["title"] == "Zombie recovery"][0])
+        check(wait_task_done(auth, zombie_task, timeout=90) is not None,
+              "zombie conflict recovers and the task completes")
+        detail = task_state(auth, zombie_task)
+        check(any("recovered zombie session" in c["body"] for c in detail.get("comments", [])),
+              "the zombie recovery is narrated")
+        conn = db()
+        zombie_sessions = conn.execute(
+            "select count(*) from sessions where task_id = ?", (zombie_task,)).fetchone()[0]
+        conn.close()
+        check(zombie_sessions == 1,
+              "no session churn: one row across the failed and retried push (%d)"
+              % zombie_sessions)
+
+        # Live conflict: the daemon claims to run a task the web still sees
+        # live — fence + alert, never abort live work.
+        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
+            "project_id": "1",
+            "title": "Conflict live task",
+            "priority": "2",
+        })
+        conflict_live = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
+                             if t["title"] == "Conflict live task"][0])
+        check(wait_for(lambda: task_state(auth, conflict_live).get("status") == "working"
+                       or False, timeout=60, what="conflict task working") is not None,
+              "live task reaches working")
+        # Hold the daemon mid-working and desynchronise the web's view:
+        # double-booking means the web believes the saddle free while the
+        # daemon still claims the live task.
+        fake.hold_state = True
+        fake.conflict_running = conflict_live
+        conn = db()
+        conn.execute("update saddles set status = 'idle', current_task_id = 0,"
+                     " allocated_at = null where name = 'fake-saddle'")
+        conn.commit()
+        conn.close()
+        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
+            "project_id": "1",
+            "title": "Conflict newcomer",
+            "priority": "3",
+        })
+        conflict_new = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
+                            if t["title"] == "Conflict newcomer"][0])
+        check(wait_for(lambda: saddle_card("fake-saddle")["status"] == "recovering" or False,
+                       timeout=30, what="double-booked saddle fenced") is not None,
+              "double-booked saddle is fenced")
+        check(any("double-booked" in c["body"]
+                  for c in task_state(auth, conflict_live).get("comments", [])),
+              "the live task gets the alert comment")
+        check(any("dispatch refused" in c["body"]
+                  for c in task_state(auth, conflict_new).get("comments", [])),
+              "the refused task gets its comment")
+        conn = db()
+        new_sessions = conn.execute(
+            "select outcome from sessions where task_id = ?", (conflict_new,)).fetchall()
+        conn.close()
+        check(len(new_sessions) == 1 and new_sessions[0][0] == "push_failed",
+              "failed push marks the session push_failed (no churn)")
+        check(conflict_live not in fake.abort_log,
+              "the live task is never aborted")
+        fake.conflict_running = 0
+        fake.hold_state = False
+        fake.state = "idle"
+        act(auth, task_state(auth, conflict_live)["complete"],
+            {"summary": "finished before the conflict"})
+        check(wait_for(lambda: task_state(auth, conflict_live).get("status") == "done"
+                       or False, timeout=30, what="live task completed") is not None,
+              "the live task completes without an abort")
+        check(wait_for(lambda: saddle_card("fake-saddle")["status"] == "idle" or False,
+                       timeout=40, what="fence clears after conflict") is not None,
+              "fence clears once the daemon goes idle")
+        conn = db()
+        after_clear = conn.execute(
+            "select count(*) from sessions where task_id = ?", (conflict_new,)).fetchone()[0]
+        conn.close()
+        check(after_clear == 1, "still one session row for the refused task")
+        check(wait_task_done(auth, conflict_new, timeout=90) is not None,
+              "the refused task dispatches after the fence clears")
+
     finally:
         # Exit-code diagnostic: distinguishes a crashed app (negative
         # returncode, e.g. -11 SIGSEGV) from a clean shutdown. When