Ghost-working sweep: requeue stale open sessions from the dispatcher tick

Goal 28 closeout. A saddle reboot can kill a round without a transition push; on a single-saddle fleet the two recovery passes that could clear the row never fire: requeue_tasks_on_offline_saddles needs the saddle row offline (the rebooted saddle reads online while running something else) and requeue_interrupted_sessions rides an /next-task ask (the one saddle is busy or parked on the goal run, so no ask ever comes). The dispatcher's busy-saddle tick now runs requeue_stale_working_sessions: an open session on an assigned/working task whose last push is older than STALE_WORKING_S (default 20 min, safely beyond a WAIT-chained MAX_TIMEOUT_MS round; SADDLE_STALE_WORKING_S overrides for tests) requeues through the same path as the ask-gated recovery, with the park-settle hold intact. SaddlePollState.seconds_since_push_seeded reads the push clock web-restart-safe (an unseen session seeds now and reads 0, never fake-ancient). Tests: the ghostworking bundle (9 checks) — a ghost 'working' row requeues from the tick while a live still-pushing round is untouched and the requeued task is claimable again.

src/Services/Dispatcher.vala

@@ -144,6 +144,21 @@ namespace ClankerSaddle.Services {
                 yield push.flush();
 
                 foreach (var saddle in yield saddles.list_busy_saddles()) {
+                    // Goal 28 (the ghost-working sweep): a busy record
+                    // whose session has been push-silent past
+                    // STALE_WORKING_S is a round that died without a
+                    // transition (saddle reboot on a single-saddle
+                    // fleet — no ask can ever fire to recover it).
+                    // Requeue it here; the sweep's own gates (park
+                    // settle, seeded clock) protect live rounds.
+                    try {
+                        var swept = yield tasks.requeue_stale_working_sessions(saddle);
+                        if (swept > 0) {
+                            print(@"[ClankerSaddle] Ghost-working sweep: requeued $swept stale session(s) on saddle $(saddle.name)\n");
+                        }
+                    } catch (Error e) {
+                        warning(@"[ClankerSaddle] Ghost-working sweep on saddle $(saddle.id) failed: $(e.message)");
+                    }
                     // The derived busy pass (plan 16 item 23): each busy
                     // saddle's running work IS its open session on an
                     // assigned/working task — the ONE truth accessor

src/Services/SaddlePollState.vala

@@ -301,6 +301,43 @@ namespace ClankerSaddle.Services {
             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.
+         */
+        /**
+         * A seeded read of the push clock: unlike {@link
+         * seconds_since_push} (MAX for an unseen session — a web
+         * restart wiped the baseline), an unseen session is seeded
+         * NOW and reads 0, so a fresh baseline can never masquerade
+         * as a long silence. Goal 28: the stale-working sweep's
+         * clock.
+         */
+        public int64 seconds_since_push_seeded(int64 session_id) {
+            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 0;
+            }
+            string stored = "";
+            bool known = push_at_ms.try_get(key, out stored);
+            state_lock.unlock();
+            if (!known) {
+                return 0;
+            }
+            return (now_ms - int64.parse(stored)) / 1000;
+        }
+
         /**
          * Atomically decides the one exceptional staleness probe for
          * `session_id`: true exactly when pushes have been seen, the

src/Services/TaskService.vala

@@ -1562,6 +1562,87 @@ namespace ClankerSaddle.Services {
             return requeued;
         }
 
+        /**
+         * Goal 28: how long an open session on an assigned/working
+         * task may stay push-silent before the dispatcher's
+         * stale-working sweep requeues it. Must comfortably exceed
+         * the LONGEST legitimate quiet stretch inside one round: a
+         * WAIT-chained run_command can legally run to MAX_TIMEOUT_MS
+         * (900 s) with consecutive WAITs and no push between asks —
+         * plus round-trip slack. SADDLE_STALE_WORKING_S overrides
+         * (seconds) for tests.
+         */
+        public const int64 STALE_WORKING_S_DEFAULT = 20 * 60;
+
+        public static int64 stale_working_s() {
+            var configured = Environment.get_variable("SADDLE_STALE_WORKING_S");
+            if (configured != null && configured.length > 0) {
+                var parsed = int64.parse(configured);
+                if (parsed > 0) {
+                    return parsed;
+                }
+            }
+            return STALE_WORKING_S_DEFAULT;
+        }
+
+        /**
+         * Goal 28 (the ghost-working sweep): requeues STALE open
+         * sessions on assigned/working tasks — driven by the
+         * dispatcher's tick, NOT by an /next-task ask. On a
+         * single-saddle fleet the ask-gated {@link
+         * requeue_interrupted_sessions} can never fire for a dead
+         * round: the one saddle is busy (or parked on the goal run)
+         * and asks only when idle, so a saddle reboot leaves the
+         * task row 'working' forever with nothing to clear it —
+         * the row wedges the fleet exactly like the crash orphan it
+         * is. This pass needs no offline saddle (the rebooted
+         * saddle reads online while running something else) and no
+         * ask: a session whose LAST push is older than {@link
+         * STALE_WORKING_S} is provably not a live round (live
+         * rounds push telemetry at least every few minutes; the
+         * clock seeds on first sight so a web restart cannot fake
+         * age) and requeues through the same path as the ask-gated
+         * recovery — the park-settle hold inside applies (a fresh
+         * push means the round is alive, park in flight).
+         */
+        public async int requeue_stale_working_sessions(Saddle saddle) throws Error {
+            var requeued = 0;
+            foreach (var session in yield db.query<Session>()
+                    .where(expr("saddle_id == $0", elem<int64?>(saddle.id)))
+                    .order_by(expr("id"))
+                    .materialise_async()) {
+                if (session.ended_at != null) {
+                    continue;
+                }
+                var task = yield db.query<TaskEntity>()
+                    .where(expr("id == $0", elem<int64?>(session.task_id)))
+                    .first_async();
+                if (task == null || (((!)task).status != STATUS_ASSIGNED
+                        && ((!)task).status != STATUS_WORKING)) {
+                    continue;
+                }
+                // the park-in-flight hold (task F): a round that
+                // pushed moments ago is live — its closing transition
+                // may still be queued; never requeue fresh pushes
+                if (poll_state.seconds_since_push(session.id)
+                        < PARK_SETTLE_GRACE_S) {
+                    continue;
+                }
+                // the staleness gate itself: seeded-on-first-sight,
+                // so a web restart starts the clock rather than
+                // faking an ancient silence
+                if (poll_state.seconds_since_push_seeded(session.id)
+                        < stale_working_s()) {
+                    continue;
+                }
+                yield requeue_task(((!)task).id);
+                yield add_system_comment(((!)task).id,
+                    @"Session stale (no telemetry for $(stale_working_s() / 60) minutes on saddle '$(saddle.name)' — the round died without a transition, e.g. a saddle reboot); requeued");
+                requeued++;
+            }
+            return requeued;
+        }
+
         /**
          * Returns a failed task to the queue for another attempt.
          * Plan 16 S4: the failure summary is the terminal cache of the

test/bundles/ghostworking.py

@@ -0,0 +1,106 @@
+"""The ghostworking bundle (goal 28): the single-saddle ghost 'working' row.
+
+A saddle reboot can kill a round without any transition push; on a
+single-saddle fleet the recovery passes that COULD clear the row never
+fire — requeue_tasks_on_offline_saddles needs the saddle row offline
+(the rebooted saddle reads online while running something else), and
+requeue_interrupted_sessions rides an /next-task ask (the one saddle
+is busy running another session, so no ask ever comes). The dispatcher
+tick's ghost-working sweep requeues a stale open session on an
+assigned/working task after STALE_WORKING_S without either crutch.
+"""
+from harness import *  # noqa: F401,F403 — the shared wire helpers
+
+# TaskService.STALE_WORKING_S is 20 minutes; the sweep reads it from
+# SADDLE_STALE_WORKING_S so the test can compress the window.
+STALE_ENV = {"SADDLE_STALE_WORKING_S": "6"}
+
+
+def run(ctx):
+    auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token
+
+    print("== goal 28: a ghost 'working' row requeues from the dispatcher tick ==")
+    saddles_state = slot_public(auth, "/saddles", "saddles")
+    act(auth, saddles_state["create_capability"], {"name": "ghostwake"})
+    cap_id = int([c["id"] for c in slot_public(auth, "/saddles", "saddles")["capabilities"]
+                  if c["name"] == "ghostwake"][0])
+
+    # The ghost: a daemon that claims a task, pushes working, then
+    # DIES silently (no further pushes, no transition — a reboot).
+    ghost = FakeSaddle()
+    ghost.ask_on_register = False
+    ghost.hold_state = True
+    ghost.serve()
+    act(auth, slot_public(auth, "/saddles", "saddles")["register"],
+        {"name": "ghost-daemon", "host": "127.0.0.1:%d" % ghost.port})
+    act(auth, saddle_card(auth, "ghost-daemon")["attach_capability"],
+        {"capability_id": str(cap_id)})
+
+    task_id = create_queued(auth, {
+        "project_id": "1", "title": "Ghost: dies without a transition",
+        "priority": "2", "capability_id": [str(cap_id)]})
+    ask_status, assignment = ghost.ask_now()
+    check(ask_status == 200 and int(assignment["task"]["id"]) == task_id,
+          "the ask claims the queued task")
+    ghost.state = "working"
+    check(ghost.push_results("working") == 200, "the working push opens the round")
+
+    conn = db()
+    row = conn.execute(
+        "select status from tasks where id = ?", (task_id,)).fetchone()
+    check(row is not None and row[0] == "working", "the task row reads working")
+
+    # The LIVE peer on another saddle: pushes keep coming, so the sweep
+    # must never touch it (the seeded clock + fresh pushes protect it).
+    live = FakeSaddle()
+    live.ask_on_register = False
+    live.hold_state = True
+    live.serve()
+    act(auth, slot_public(auth, "/saddles", "saddles")["register"],
+        {"name": "live-daemon", "host": "127.0.0.1:%d" % live.port})
+    act(auth, saddle_card(auth, "live-daemon")["attach_capability"],
+        {"capability_id": str(cap_id)})
+    live_task = create_queued(auth, {
+        "project_id": "1", "title": "Live: keeps pushing", "priority": "1",
+        "capability_id": [str(cap_id)]})
+    ask_status, live_assignment = live.ask_now()
+    check(ask_status == 200 and int(live_assignment["task"]["id"]) == live_task,
+          "the live peer claims its task")
+    live.state = "working"
+    check(live.push_results("working") == 200, "the live round pushes working")
+
+    # Silence the ghost: no pushes for the (compressed) stale window
+    # while the live daemon keeps pushing. The sweep must requeue ONLY
+    # the ghost's row, and the requeued task must be claimable again.
+    deadline = time.time() + 40
+    requeued = False
+    while time.time() < deadline:
+        live.state = "working"
+        live.push_results("working")
+        row = conn.execute(
+            "select status from tasks where id = ?", (task_id,)).fetchone()
+        if row is not None and row[0] == "queued":
+            requeued = True
+            break
+        time.sleep(2.0)
+    check(requeued,
+          "the ghost 'working' row requeues once its silence passes the stale window")
+
+    row = conn.execute(
+        "select status from tasks where id = ?", (live_task,)).fetchone()
+    check(row is not None and row[0] == "working",
+          "the live, still-pushing round is NOT touched by the sweep")
+
+    # The requeued task is claimable again — the fleet is unwedged.
+    claim_status, reclaim = ghost.ask_now()
+    check(claim_status == 200 and int(reclaim["task"]["id"]) == task_id,
+          "the requeued ghost task is claimable again")
+
+    ghost.state = "done"
+    ghost.push_results("done", summary="# done")
+    live.state = "done"
+    live.push_results("done", summary="# done")
+    wait_for(lambda: task_state(auth, task_id).get("status") == "done"
+             or False, timeout=30, what="ghost task done")
+    wait_for(lambda: task_state(auth, live_task).get("status") == "done"
+             or False, timeout=30, what="live task done")

test/harness.py

@@ -1055,6 +1055,9 @@ def start_app():
         # Plan 12: the goal-agent resume coalescing window shrinks to
         # one second so the event-driven resume tests settle quickly.
         "SADDLE_GOAL_RESUME_S": os.environ.get("SADDLE_GOAL_RESUME_S", "1"),
+        # Goal 28 (ghost-working sweep): compress the stale window in
+        # tests — the ghostworking bundle sets its own default here.
+        "SADDLE_STALE_WORKING_S": os.environ.get("SADDLE_STALE_WORKING_S", "6"),
         # Never inherit a caller-poisoned LD_LIBRARY_PATH (e.g. one that
         # resolves the ancient /lib64 copies): empty makes the loader use
         # the binary's RUNPATH, which is the dev stack it was built
@@ -1331,7 +1334,8 @@ def start_bundle(name):
 
 
 BUNDLES_ORDER = ("pages", "dispatch", "subtasks", "lineage",
-                 "allocation", "lifecycle", "answers", "answerwake", "push",
+                 "allocation", "lifecycle", "answers", "answerwake", "ghostworking",
+                 "push",
                  "goals", "replay", "goalui", "forcewake", "users", "stacks",
                  "notes", "resurrect", "origins", "parity", "understanding",
                  "discoveries", "knowledge_cycle", "advisor", "chatui",