clanker-saddle/src/Services/SaddleService.vala

using Invercargill;
using Invercargill.DataStructures;
using Invercargill.Expressions;
using InvercargillSql;
using InvercargillSql.Orm;
using Inversion;
using Soup;
using ClankerSaddle.Data;

namespace ClankerSaddle.Services {

    /**
     * Valid saddle statuses (the `saddles.status` wire values, plan 16
     * R2 / arch item 26 — the raw literals died with these). The row
     * default and the migration seed keep their literals (the Data
     * layer does not reference the services).
     */
    public const string SADDLE_IDLE = "idle";
    public const string SADDLE_BUSY = "busy";
    public const string SADDLE_OFFLINE = "offline";

    /**
     * Session outcomes (the `sessions.outcome` values the park, the
     * ONE release path and the terminal pushes write): `parked` (a
     * round that parked on a block or a suspend — the row closes at
     * park time and the workspace awaits its resuming allocation;
     * plan 16 D4, aligned with the daemon's file-log brackets),
     * `stopped` (a cancelled/requeued/unreachable release),
     * `done`/`failed` (the terminal pushes).
     */
    public const string OUTCOME_STOPPED = "stopped";
    public const string OUTCOME_PARKED = "parked";
    public const string OUTCOME_DONE = "done";
    public const string OUTCOME_FAILED = "failed";

    /** Caps of the `project_notes` array in the /start payload. */
    public const int64 PROJECT_NOTES_MAX_ENTRIES = 30;
    public const int64 PROJECT_NOTES_MAX_CHARS = 20000;

    /** A block answer awaiting collection by one saddle's next poll. */
    public class PendingAnswer : Object {
        public int64 block_id { get; set; }
        public string answer { get; set; default = ""; }
        public bool branch_advanced { get; set; }
    }

    /** A saddle registration plus the API token generated for it. */
    public class SaddleRegistration : Object {
        public Saddle saddle { get; set; }
        public string token { get; set; default = ""; }
    }

    /**
     * A saddle's current work as the cards render it: the running
     * task's title, plus the GOAL it stands for when that task is a
     * goal agent's internal run.
     */
    public class CurrentWork : Object {
        public int64 task_id { get; set; }
        public string task_title { get; set; default = ""; }
        /** Set when the current task is a goal agent's run: the card links the goal. */
        public int64 goal_id { get; set; }
        public string goal_title { get; set; default = ""; }
        /** The running session the derivation keyed on (plan 16 item 23). */
        public int64 session_id { get; set; }
        /** When that session started — the "allocated" display stamp. */
        public DateTime? started_at { get; set; default = null; }
    }

    /**
     * The OpenAI-compatible endpoint every saddle assignment is sent to,
     * read from the environment: `SADDLE_LLM_BASE_URL` (required — dispatch
     * pauses without it), `SADDLE_LLM_API_KEY` and `SADDLE_LLM_MODEL`.
     */
    public class LlmEndpoint : Object {

        public string base_url { get; set; default = ""; }
        public string api_key { get; set; default = ""; }
        public string model { get; set; default = ""; }

        /** The configured endpoint, or null when the base URL is unset. */
        public static LlmEndpoint? from_environment() {
            var base_url = (Environment.get_variable("SADDLE_LLM_BASE_URL") ?? "").strip();
            if (base_url.length == 0) {
                return null;
            }
            return new LlmEndpoint() {
                base_url = base_url,
                api_key = (Environment.get_variable("SADDLE_LLM_API_KEY") ?? "").strip(),
                model = (Environment.get_variable("SADDLE_LLM_MODEL") ?? "").strip()
            };
        }
    }

    /**
     * Saddle registry and control-plane client (plan 10 phase 2): registers
     * containers, derives idle capacity for the dispatcher's wake
     * computation (plan 16 item 23: busy iff an open session on an
     * assigned/working task exists — {@link current_work_view} is the ONE
     * truth accessor; the saddle row keeps only the OFFLINE probe state),
     * wakes sleeping daemons, and ingests the telemetry the daemons PUSH —
     * raw PTY chunks (`POST /api/saddle/chunks`) and round results with
     * state transitions (`POST /api/saddle/results`) — reusing the storage
     * logic the deleted `/output` + `/log` pollers ran. GET /status
     * against a daemon is exceptional only: the staleness probe for a
     * silent runner and the manual refresh button. Block answers are
     * handed out exactly once (delivered ids are remembered in memory; a
     * restart re-delivers and the daemon deduplicates).
     */
    public class SaddleService : Object {

        private OrmSession db = inject<OrmSession>();
        private Connection connection = inject<Connection>();
        /**
         * Every daemon request shares this session and therefore its
         * timeouts: without them a single hung request (a daemon wedged
         * mid-response over the keep-alive pool) froze the dispatcher's
         * tick latch forever. `timeout` bounds each request
         * (SADDLE_HTTP_TIMEOUT_S, default 30s); `idle_timeout` reaps
         * pooled connections so a peer that vanished between requests is
         * not dialed again.
         */
        private Soup.Session http = new Soup.Session() {
            timeout = http_timeout_seconds(),
            idle_timeout = 10
        };

        private static uint http_timeout_seconds() {
            var raw = Environment.get_variable("SADDLE_HTTP_TIMEOUT_S") ?? "";
            int64 parsed = int64.parse(raw);
            return parsed > 0 ? (uint) parsed : 30;
        }
        private SaddlePollState poll_state = inject<SaddlePollState>();
        private StatePulse pulse = inject<StatePulse>();
        private PushService push = inject<PushService>();
        private RuleService rules = inject<RuleService>();
        private CapabilityService capability_service = inject<CapabilityService>();
        private TaskService task_service = inject<TaskService>();
        private GoalService goal_service = inject<GoalService>();
        private EventService event_service = inject<EventService>();
        private Astralis.WebConfig config = inject<Astralis.WebConfig>();

        /**
         * Registers a saddle container: inserts the row with a freshly
         * minted per-saddle token (plan 10) and hands the credential to
         * the daemon via `POST /register {saddle_token, web_base_url}`
         * (best effort — an older daemon without the route stays
         * registered; the operator copies the token from the saddles
         * page instead). `host` is host:port of its control server; the
         * returned {@link SaddleRegistration} carries the token the
         * daemon must present as `Authorization: Bearer …` on
         * `/api/saddle/*` calls.
         */
        public async SaddleRegistration register_saddle(string name, string host) throws Error {
            var saddle = new Saddle() {
                name = name,
                host = host,
                status = SADDLE_IDLE,
                token = mint_saddle_token()
            };
            yield db.insert_async<Saddle>(saddle);
            yield register_daemon(saddle);
            yield pulse.bump(0, "saddle registered");
            return new SaddleRegistration() { saddle = saddle, token = saddle.token ?? "" };
        }

        /** A fresh per-saddle bearer token (`saddle-<random>`). */
        private static string mint_saddle_token() {
            return @"saddle-$(Uuid.string_random().replace("-", ""))";
        }

        /**
         * Delivers the per-saddle token to the daemon's
         * `POST /register` endpoint (wire contract): the daemon
         * persists it for its calls home and answers
         * `{ok, ssh_public_key, version}` — the key is stored on the
         * row the moment the daemon reports one. A daemon without the
         * route (HTTP 404/405) or unreachable at registration time is
         * tolerated: the row keeps the token, the operator can rety
         * via a refresh or copy the token manually.
         */
        public async void register_daemon(Saddle saddle) throws Error {
            var payload = new Json.Builder();
            payload.begin_object();
            payload.set_member_name("saddle_token");
            payload.add_string_value(saddle.token ?? "");
            payload.set_member_name("web_base_url");
            payload.add_string_value(callback_base_url());
            payload.end_object();
            try {
                var response = yield request_saddle_json(saddle.host, "POST", "/register",
                    (owned) payload);
                string reported_key = EntryJson.string_member(response, "ssh_public_key");
                // goal 8 task 6: the daemon reports the workspace root it
                // mounts so assignment paths address the ASKING saddle's
                // filesystem; absent on older daemons (kept null → the
                // SADDLE_PROJECTS_ROOT fallback chain)
                string reported_root = EntryJson.string_member(response, "workspace_root");
                if ((reported_key.length > 0 && reported_key != (saddle.ssh_public_key ?? ""))
                        || reported_root != (saddle.workspace_root ?? "")) {
                    if (reported_key.length > 0) {
                        saddle.ssh_public_key = reported_key;
                    }
                    saddle.workspace_root = reported_root.length > 0 ? reported_root : null;
                    yield db.update_async<Saddle>(saddle);
                }
                print(@"[ClankerSaddle] Registered with saddle '$(saddle.name)'s daemon (token delivered)\n");
            } catch (Error e) {
                print(@"[ClankerSaddle] Saddle '$(saddle.name)'s daemon did not accept /register ($(e.message)); token stays on the row\n");
            }
        }

        /**
         * The saddle a bearer token belongs to, or null — the
         * credential check for the id-less `/api/saddle/*` endpoints
         * (`next-task`, `note`, `task`): the per-saddle row token,
         * the single credential store since plan 10 (arch 24 dropped
         * the legacy api_tokens fallback — nothing deployed predates
         * the row token). Removing the saddle revokes it.
         */
        public async Saddle? saddle_for_token(string token) throws Error {
            if (token.length == 0) {
                return null;
            }
            return yield db.query<Saddle>()
                .where(expr("token == $0", elem(token)))
                .first_async();
        }

        /**
         * This app's externally reachable base URL — what daemons use to
         * POST notes back — read from the `saddle_callback_base_url` key
         * of web-config.json, falling back to the
         * `SADDLE_CALLBACK_BASE_URL` environment variable and then to
         * `http://localhost:8080`. The daemon typically runs in a
         * container or on another host, so its view of this app differs
         * from the browser's; set the key to whatever reaches this
         * process from there.
         */
        public string callback_base_url() {
            string configured = config.get_string("saddle_callback_base_url", "").strip();
            if (configured.length > 0) {
                return configured;
            }
            var from_env = (Environment.get_variable("SADDLE_CALLBACK_BASE_URL") ?? "").strip();
            if (from_env.length > 0) {
                return from_env;
            }
            return "http://localhost:8080";
        }

        /**
         * Removes a saddle registration; the task it was running (the
         * DERIVED running work — plan 16 item 23: sessions are the
         * truth) releases through the ONE path — back to the queue,
         * its open sessions closed stopped (no orphan open session
         * outlives the row). Revocation: the row dies here, and its
         * per-saddle token with it — nothing authorises
         * `/api/saddle/*` for it afterwards.
         */
        public async void unregister_saddle(int64 id) throws Error {
            var saddle = yield get_saddle(id);
            if (saddle == null) {
                return;
            }
            var work = yield current_work_view((!)saddle);
            if (work != null) {
                yield task_service.release_task(((!)work).task_id, OUTCOME_STOPPED, true);
            }
            yield db.delete_async<Saddle>((!)saddle);
            yield pulse.bump(0, "saddle removed");
        }

        /** Loads the saddle identified by `id`, or null. */
        public async Saddle? get_saddle(int64 id) throws Error {
            return yield db.query<Saddle>()
                .where(expr("id == $0", elem<int64?>(id)))
                .first_async();
        }

        /** Lists every saddle ordered by name. */
        public async ImmutableLot<Saddle> list_saddles() throws Error {
            return yield db.query<Saddle>()
                .order_by(expr("name"))
                .materialise_async();
        }

        /**
         * The saddle's current work resolved from the SESSION records —
         * the ONE truth accessor every surface reads (plan 16 item 23:
         * sessions own "who runs what"; the saddle row's cache columns
         * are gone). The running session is the newest open session on
         * the saddle whose task is still assigned/working — busy iff it
         * exists, idle otherwise; a parked task's session closed
         * 'parked' at its park (D4), so parks read as free exactly
         * like the daemon's file-log brackets say. The task join also
         * makes the derivation drift-tolerant: a stale open row on a
         * task that has since moved on cannot wedge a saddle busy.
         * Null when the saddle holds no running work.
         */
        public async CurrentWork? current_work_view(Saddle saddle) throws Error {
            var session = yield running_session_on_saddle(saddle.id);
            if (session == null) {
                return null;
            }
            var work = new CurrentWork() {
                task_id = ((!)session).task_id,
                session_id = ((!)session).id,
                started_at = ((!)session).started_at
            };
            var task = yield db.query<TaskEntity>()
                .where(expr("id == $0", elem<int64?>(((!)session).task_id)))
                .first_async();
            if (task == null) {
                return work;
            }
            work.task_title = ((!)task).title;
            if (((!)task).kind == TASK_KIND_GOAL_RUN) {
                var goal = yield db.query<Goal>()
                    .where(expr("id == $0", elem<int64?>(((!)task).goal_id)))
                    .first_async();
                if (goal != null) {
                    work.goal_id = ((!)goal).id;
                    work.goal_title = ((!)goal).title;
                }
            }
            return work;
        }

        /**
         * The derived busy/idle/offline status label every saddle view
         * shares — 'busy' iff {@link current_work_view} holds running
         * work, else the row's stored OFFLINE probe state ('offline' or
         * 'idle').
         */
        public static string derived_status(Saddle saddle, CurrentWork? work) {
            if (work != null) {
                return SADDLE_BUSY;
            }
            return saddle.status == SADDLE_OFFLINE ? SADDLE_OFFLINE : SADDLE_IDLE;
        }

        /**
         * The saddle's RUNNING session — the newest open session whose
         * task is assigned/working (the null check runs in SQL: the
         * expression layer cannot compare columns against null). At
         * most one exists per saddle by construction: every park closes
         * its row (D4) and the claim's guard refuses a saddle that
         * already holds one.
         */
        private async ClankerSaddle.Data.Session? running_session_on_saddle(int64 saddle_id)
                 throws Error {
            var command = connection.create_command(
                "SELECT s.id AS session_id, s.task_id AS task_id, s.started_at AS started"
                + " FROM sessions s JOIN tasks t ON t.id = s.task_id"
                + " WHERE s.saddle_id = :saddle AND s.ended_at IS NULL"
                + " AND t.status IN ('assigned', 'working')"
                + " ORDER BY s.id DESC LIMIT 1");
            command.with_parameter<Invercargill.Element>("saddle",
                elem<int64?>(saddle_id));
            ClankerSaddle.Data.Session? found = null;
            foreach (var row in yield command.execute_query_async()) {
                int64? session_id = 0;
                int64? task_id = 0;
                ((!)row.get("session_id")).try_get_as<int64?>(out session_id);
                ((!)row.get("task_id")).try_get_as<int64?>(out task_id);
                found = new ClankerSaddle.Data.Session() {
                    id = session_id ?? 0,
                    task_id = task_id ?? 0,
                    saddle_id = saddle_id,
                    started_at = raw_date_time_of(row.get("started"))
                };
            }
            return found;
        }

        /** A raw query column back as a DateTime (null when null/unreadable). */
        private static DateTime? raw_date_time_of(Invercargill.Element? element) {
            if (element == null) {
                return null;
            }
            DateTime value;
            return ((!)element).try_get_as<DateTime>(out value) ? value : null;
        }

        /**
         * The ids (decimal strings — the OR-chain batch convention) of
         * every saddle holding a running session — the derived busy
         * set the fleet lists and the offline stamp share.
         */
        private async Invercargill.DataStructures.HashSet<string> busy_saddle_ids()
                 throws Error {
            var busy = new Invercargill.DataStructures.HashSet<string>();
            var command = connection.create_command(
                "SELECT DISTINCT s.saddle_id AS saddle_id FROM sessions s"
                + " JOIN tasks t ON t.id = s.task_id"
                + " WHERE s.ended_at IS NULL"
                + " AND t.status IN ('assigned', 'working')");
            foreach (var row in yield command.execute_query_async()) {
                int64? saddle_id = 0;
                ((!)row.get("saddle_id")).try_get_as<int64?>(out saddle_id);
                if (saddle_id != null) {
                    busy.add(((!)saddle_id).to_string());
                }
            }
            return busy;
        }

        /**
         * Saddles holding a running session (the dispatcher's poll
         * candidates) — derived from the session records, registration
         * order.
         */
        public async ImmutableLot<Saddle> list_busy_saddles() throws Error {
            var busy = yield busy_saddle_ids();
            if (busy.count() == 0) {
                return (new Vector<Saddle>()).to_immutable_buffer();
            }
            var rows = new Vector<Saddle>();
            foreach (var saddle in yield db.query<Saddle>()
                    .where(TaskService.id_in("id", busy))
                    .order_by(expr("id"))
                    .materialise_async()) {
                rows.add(saddle);
            }
            return rows.to_immutable_buffer();
        }

        /**
         * Persists the saddle's git identity — what its daemon commits
         * as (`git_identity` in every /start). Empty/null clears the
         * fields so the daemon's fallback applies.
         */
        public async void update_git_identity(int64 id, string? git_name,
                string? git_email) throws Error {
            var saddle = yield get_saddle(id);
            if (saddle == null) {
                throw new Error(Quark.from_string("clanker-saddle"), 17,
                    @"No saddle exists with id $(id)");
            }
            ((!)saddle).git_name = git_name != null && git_name.length > 0 ? git_name : null;
            ((!)saddle).git_email = git_email != null && git_email.length > 0 ? git_email : null;
            yield db.update_async<Saddle>((!)saddle);
            yield pulse.bump(0, "saddle identity saved");
        }

        /**
         * Persists the daemon's reported `ssh_public_key` onto the saddle
         * row (the plan-9 contract: the key travels on /status, empty
         * until the daemon generates one; the saddles page shows it for
         * adding as a deploy key). A single-column UPDATE — the
         * full-row write could stomp a concurrent ask claim. Only writes
         * when it changed.
         */
        private async void store_reported_ssh_key(Saddle saddle, Json.Object status) throws Error {
            if (!status.has_member("ssh_public_key")) {
                return;
            }
            string reported = status.get_string_member_with_default("ssh_public_key", "");
            string stored = saddle.ssh_public_key ?? "";
            if (reported == stored) {
                return;
            }
            var command = connection.create_command(
                "UPDATE saddles SET ssh_public_key = :key WHERE id = :id");
            command.with_parameter<Invercargill.Element>("key", elem(reported.length > 0 ? reported : null));
            command.with_parameter<Invercargill.Element>("id", elem<int64?>(saddle.id));
            yield command.execute_non_query_async();
        }

        /**
         * Records a /status observation: the health snapshot, the
         * daemon-reported ssh key, and the goal-branch push flags
         * (plan 15 C4). The `resumed` flag needs no projection — the
         * restore's record is the daemon's own system entry in the
         * raw terminal log (plan 16 S5/U2).
         */
        private async void record_status_probe(Saddle saddle, Json.Object status) throws Error {
            string state = status.has_member("state")
                ? status.get_string_member_with_default("state", "") : "";
            int64 uptime = status.has_member("uptime_seconds")
                && status.get_member("uptime_seconds").get_value_type() == typeof(int64)
                ? status.get_int_member("uptime_seconds") : 0;
            int64 activity = status.has_member("last_activity_epoch")
                && status.get_member("last_activity_epoch").get_value_type() == typeof(int64)
                ? status.get_int_member("last_activity_epoch") : 0;
            poll_state.record_probe(saddle.id, state, uptime, activity);
            yield store_reported_ssh_key(saddle, status);

            // Plan 15 C4: a daemon running (or parking — the
            // workspace stays while the saddle frees) a goal agent
            // exposes the per-repo push flags of the goal branches on
            // /status; they are recorded as LIVE probe state on the
            // saddle (never a stored snapshot on the goal row — a
            // pre-push snapshot is exactly the F9 bug), and the goal
            // page + the completion-accept gate read them back at
            // render/answer time.
            if (status.has_member("goal_branches")) {
                var goal_branches = status.get_member("goal_branches");
                if (goal_branches.get_node_type() == Json.NodeType.ARRAY
                        || goal_branches.get_node_type() == Json.NodeType.OBJECT) {
                    var generator = new Json.Generator();
                    generator.set_root(goal_branches);
                    poll_state.record_goal_branches(saddle.id, generator.to_data(null));
                }
            }

            // Plan 16 S5 (U2): the daemon's own system entry (typed
            // 'system' in the results stream) is the raw-terminal
            // record of a workspace restore — the web no longer authors
            // a duplicate timeline row for it. The `resumed` flag
            // carries no further web-side projection.
        }

        /**
         * Wakes a sleeping daemon (plan 10 wire contract): `POST /wake`
         * `{}` — a capable task is ready; sleeping daemons ask for work,
         * busy ones ignore it, idle-awake ones re-ask. Idempotent. The
         * caller records the pending wake for the lost-wake guard.
         */
        public async void wake_saddle(Saddle saddle) throws Error {
            var payload = new Json.Builder();
            payload.begin_object();
            payload.end_object();
            Json.Node? root = null;
            int status_code = yield request_saddle(saddle.host, "POST", "/wake",
                (owned) payload, out root);
            if (status_code >= 400) {
                throw new Error(Quark.from_string("clanker-saddle"), 49,
                    @"wake returned HTTP $(status_code)");
            }
        }

        /**
         * Cancelling / requeueing / stopping a task with a live session
         * reaches its daemon (wire contract): `POST /task-update
         * {kind: "cancel"}` — the authoritative stop through the daemon's
         * abort machinery, the workspace kept on disk per the park
         * contract. Success is narrated as "<verb>; daemon aborted"; a
         * transport failure marks the saddle offline (the daemon's next
         * ask proves it alive again and flips the row back to idle). 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;
            }
            yield abort_session_on_its_saddle(tasks, (!)session, verb);
        }

        /**
         * SADDLE_STALE_S (default 90s): the silence window after which a
         * running daemon owes an exceptional GET /status probe.
         */
        public static uint stale_after_seconds() {
            var raw = Environment.get_variable("SADDLE_STALE_S") ?? "";
            int64 parsed = int64.parse(raw);
            return parsed > 0 ? (uint) parsed : 90;
        }

        /**
         * {@link abort_live_session} keyed by saddle instead of task —
         * the unregister path. The cancel is delivered to the saddle's
         * RUNNING session (the derived record; plan 16 item 23) — not
         * re-resolved through a task's lowest-id open session, which a
         * stale fixture row on another saddle would redirect to the
         * wrong daemon.
         */
        public async void abort_saddle_session(TaskService tasks, int64 saddle_id,
                string verb) throws Error {
            var session = yield running_session_on_saddle(saddle_id);
            if (session == null) {
                return;
            }
            yield abort_session_on_its_saddle(tasks, (!)session, verb);
        }

        /**
         * The cancel delivery core shared by the task-keyed and
         * saddle-keyed abort paths: `POST /task-update {kind: cancel}`
         * to the daemon holding THIS session.
         */
        private async void abort_session_on_its_saddle(TaskService tasks,
                ClankerSaddle.Data.Session session, string verb) throws Error {
            var saddle = yield get_saddle(session.saddle_id);
            if (saddle == null) {
                return;
            }

            var payload = new Json.Builder();
            payload.begin_object();
            payload.set_member_name("kind");
            payload.add_string_value("cancel");
            payload.end_object();
            Json.Node? root = null;
            try {
                int status_code = yield request_saddle(((!)saddle).host, "POST",
                    "/task-update", (owned) payload, out root);
                if (status_code >= 400) {
                    throw new Error(Quark.from_string("clanker-saddle"), 45,
                        @"task-update cancel returned HTTP $(status_code)");
                }
                yield tasks.add_system_comment(session.task_id, @"$(verb); daemon aborted");
            } catch (Error e) {
                warning(@"[ClankerSaddle] Abort for task $(session.task_id) failed: $(e.message); marking saddle '$(((!)saddle).name)' offline");
                yield mark_saddle_unreachable(((!)saddle).id, tasks,
                    @"abort delivery failed ($(e.message))");
                yield tasks.add_system_comment(session.task_id,
                    @"$(verb); daemon unreachable — saddle marked offline");
            }
        }

        /**
         * Requeues tasks caught running on saddles that have since gone
         * offline (a manual status refresh, a restart) — without this
         * pass they would wait forever for a daemon that will not answer.
         * Only `assigned`/`working` tasks move: parked tasks
         * (blocked/review/suspended, their sessions closed 'parked')
         * keep their records, which the resume machinery needs. Each
         * move is {@link TaskService.requeue_task} — the release owner
         * closes the session (the close IS the free, item 23) — plus a
         * system comment so the timeline shows why the task restarted.
         */
        public async void requeue_tasks_on_offline_saddles(TaskService tasks) throws Error {
            foreach (var status in new string[] { STATUS_ASSIGNED, STATUS_WORKING }) {
                foreach (var task in yield db.query<TaskEntity>()
                        .where(expr("status == $0", elem(status)))
                        .materialise_async()) {
                    var session = yield tasks.open_session_for_task(task.id);
                    if (session == null) {
                        continue;
                    }
                    if (yield tasks.newest_block_answered(task.id)) {
                        // Parked-with-answered: a verdict awaits the
                        // resume machinery — never requeue it.
                        continue;
                    }
                    var saddle = yield get_saddle(((!)session).saddle_id);
                    if (saddle == null || ((!)saddle).status != SADDLE_OFFLINE) {
                        continue;
                    }
                    print(@"[ClankerSaddle] Task $(task.id) requeued: its saddle went offline\n");
                    yield tasks.requeue_task(task.id);
                    yield tasks.add_system_comment(task.id,
                        "saddle went offline; task requeued");
                }
            }
        }

        /**
         * Every saddle free for work — derived (plan 16 item 23): not
         * offline (the probe state) and holding no running session —
         * registration order, the dispatcher's capability-matching
         * candidate list. A parked task's closed session reads free
         * exactly like the freed daemon it brackets.
         */
        public async ImmutableLot<Saddle> list_idle_saddles() throws Error {
            var busy = yield busy_saddle_ids();
            var rows = new Vector<Saddle>();
            foreach (var saddle in yield db.query<Saddle>()
                    .order_by(expr("id"))
                    .materialise_async()) {
                if (saddle.status == SADDLE_OFFLINE
                        || busy.contains(((int64) saddle.id).to_string())) {
                    continue;
                }
                rows.add(saddle);
            }
            return rows.to_immutable_buffer();
        }

        /**
         * Marks a saddle offline when its control server cannot be
         * reached. The task it was RUNNING (the derived running
         * session — plan 16 item 23) releases back to 'queued' through
         * {@link TaskService.release_task} (the ONE owner; it closes
         * the session 'stopped'), with a system comment so the timeline
         * shows the interruption; combined with the daemon's
         * crash-resume, the next dispatch restores the workspace and
         * reports `resumed` on /status. A task that parked mid-flight
         * (its session already closed 'parked', the verdict awaiting
         * the resume) keeps its park. The offline stamp itself is the
         * ONE surviving guarded write — see {@link stamp_offline}.
         */
        public async void mark_saddle_unreachable(int64 id, TaskService tasks,
                string? reason = null) throws Error {
            var saddle = yield get_saddle(id);
            if (saddle == null) {
                return;
            }
            var session = yield running_session_on_saddle(id);
            int64 seen_session = session != null ? ((!)session).id : 0;
            int64 held_task = 0;
            if (session != null) {
                held_task = ((!)session).task_id;
                // The requeue flip is guarded: a park (or terminal)
                // interleaving this path keeps its state — the parked
                // verdict must survive for the resume.
                var flip = connection.create_command(
                    "UPDATE tasks SET status = 'queued' WHERE id = :id"
                    + " AND status IN ('assigned', 'working')");
                flip.with_parameter<Invercargill.Element>("id", elem<int64?>(held_task));
                if ((yield flip.execute_non_query_async()) > 0) {
                    string unreachable_text = reason != null && reason.length > 0
                        ? @"Saddle became unreachable ($(reason)); task requeued"
                        : "Saddle became unreachable; task requeued";
                    yield tasks.add_system_comment(held_task, unreachable_text);
                    // Closes the held session FIRST — the offline
                    // stamp's guard refuses to land over a running
                    // session, and by then there is none.
                    yield tasks.release_task(held_task, OUTCOME_STOPPED, false);
                } else {
                    // Parked (or finished) mid-flight — the park closed
                    // its session and the verdict must survive for the
                    // resume; nothing to release.
                    held_task = 0;
                }
            }
            yield stamp_offline(id, seen_session);
            // An offline saddle is neither sleeping nor wake-awaiting;
            // a restarted daemon re-establishes both by asking.
            poll_state.clear_ask(id);
            yield pulse.bump(0, "saddle unreachable");
        }

        /**
         * The ONE surviving guarded saddle write (plan 16 item 23):
         * the OFFLINE probe stamp. The guard is the sessions-truth
         * analogue of the old seen-assignment guard — the stamp lands
         * unless a running session the probe did NOT see at entry
         * appeared meanwhile (an ask claim is proof of life and must
         * not be stomped offline). The session observed at entry
         * (`seen_session`) never blocks its own stamp.
         */
        private async void stamp_offline(int64 saddle_id, int64 seen_session) throws Error {
            var command = connection.create_command(
                "UPDATE saddles SET status = '" + SADDLE_OFFLINE + "' WHERE id = :id"
                + " AND NOT EXISTS (SELECT 1 FROM sessions s"
                + " JOIN tasks t ON t.id = s.task_id"
                + " WHERE s.saddle_id = :id AND s.ended_at IS NULL"
                + " AND t.status IN ('assigned', 'working')"
                + " AND s.id != :seen)");
            command.with_parameter<Invercargill.Element>("id", elem<int64?>(saddle_id));
            command.with_parameter<Invercargill.Element>("seen", elem<int64?>(seen_session));
            yield command.execute_non_query_async();
        }

        /**
         * Assembles the assignment payload shared by the /start push and
         * the `POST /api/saddle/next-task` response — `{task, kind,
         * mode, creator, goal?, rules, project_notes, callback,
         * llm_config, session_id}` — with the project, links (the
         * goal's own links for goal work, else the project's), branch
         * fields and recent comments inside the task object, plus any
         * block answers not yet handed over (a resumed workspace picks
         * its pending answers up from the assignment itself), the
         * effective rule set ({@link RuleService.effective_rules}) as
         * the top-level `rules` array, the capped `project_notes` of
         * the primary and co-developed projects, the note-creation
         * `callback` (base URL + the saddle's own bearer token), the
         * saddle's `git_identity`, a `capabilities` context (plan 10
         * phase 3: `saddle_tags` — the allocated saddle's own tags —
         * and `fleet_tags` — every tag name known across the fleet),
         * the plan-12 goal context (`goal` for goal work; the
         * worker's `ephemeral_branch`) and — for resumed workspaces —
         * the `resume_notice`/`branch_advanced` context. `session_id`
         * (the web-side run row the daemon reports against) rides
         * along so pushed telemetry can key on it. Project checkouts
         * are addressed by the `SADDLE_PROJECTS_ROOT` convention
         * (`root/task-<task id>/name`) and cloned from each project's
         * `git_url`; the LLM endpoint by the {@link LlmEndpoint}
         * environment.
         */
        public async Json.Object build_assignment_payload(Saddle saddle, TaskEntity task,
                int64 session_id, ProjectService projects, string? resume_notice = null,
                bool? branch_advanced = null,
                string? context_update = null) throws Error {
            var endpoint = LlmEndpoint.from_environment();
            if (endpoint == null) {
                throw new Error(Quark.from_string("clanker-saddle"), 24,
                    "No LLM endpoint configured (set SADDLE_LLM_BASE_URL)");
            }

            var project = task.project_id > 0
                ? yield db.query<Project>()
                    .where(expr("id == $0", elem<int64?>(task.project_id)))
                    .first_async()
                : null;
            // Goal 8 task 4: project_id 0 is a STANDALONE task — no
            // project row exists (and none is looked up), the
            // co-developed/reference arrays ship EMPTY and the daemon
            // renders the standalone prompt section. A task with a
            // non-zero project_id whose row is missing is still the
            // data-integrity error it always was.
            if (project == null && task.project_id > 0) {
                throw new Error(Quark.from_string("clanker-saddle"), 12,
                    @"Task $(task.id) has no project $(task.project_id)");
            }

            // Plan 12: goal work (the agent's own run and its worker
            // tasks) scopes its co-developed/reference set through the
            // GOAL's links; standalone tasks through the project's.
            // Standalone tasks (project 0) own no links at all — the
            // simplest consistent shape: both arrays stay empty.
            Goal? goal = null;
            if (task.goal_id > 0) {
                goal = yield db.query<Goal>()
                    .where(expr("id == $0", elem<int64?>(task.goal_id)))
                    .first_async();
            }
            var co_developed = new Vector<Project>();
            var references = new Vector<Project>();
            if (goal != null) {
                foreach (var link in yield goal_service.list_links(task.goal_id)) {
                    var linked = yield db.query<Project>()
                        .where(expr("id == $0", elem<int64?>(link.project_id)))
                        .first_async();
                    if (linked == null) {
                        continue;
                    }
                    if (link.relationship == LINK_CO_DEVELOPED) {
                        co_developed.add((!)linked);
                    } else if (link.relationship == LINK_REFERENCE) {
                        references.add((!)linked);
                    }
                }
            } else if (project != null) {
                foreach (var link in yield projects.list_links(task.project_id)) {
                    var linked = yield db.query<Project>()
                        .where(expr("id == $0", elem<int64?>(link.linked_project_id)))
                        .first_async();
                    if (linked == null) {
                        continue;
                    }
                    if (link.relationship == LINK_CO_DEVELOPED) {
                        co_developed.add((!)linked);
                    } else if (link.relationship == LINK_REFERENCE) {
                        references.add((!)linked);
                    }
                }
            }

            var payload = new Json.Builder();
            payload.begin_object();
            payload.set_member_name("task");
            payload.begin_object();
            payload.set_member_name("id");
            payload.add_int_value(task.id);
            payload.set_member_name("title");
            payload.add_string_value(task.title);
            payload.set_member_name("description");
            payload.add_string_value(task.description ?? "");
            payload.set_member_name("project_name");
            payload.add_string_value(project != null ? ((!)project).name : "standalone");
            payload.set_member_name("project_branch");
            payload.add_string_value(project != null ? ((!)project).default_branch : "");
            payload.set_member_name("task_branch");
            payload.add_string_value(task.task_branch ?? "");
            payload.set_member_name("branch_from");
            payload.add_string_value(task.branch_from
                ?? (project != null ? ((!)project).default_branch : ""));

            payload.set_member_name("co_developed");
            payload.begin_array();
            if (project != null) {
                payload.add_value(project_node(((!)project), task.id, false, saddle));
            }
            foreach (var linked in co_developed) {
                payload.add_value(project_node(linked, task.id, false, saddle));
            }
            payload.end_array();
            payload.set_member_name("reference");
            payload.begin_array();
            foreach (var linked in references) {
                payload.add_value(project_node(linked, task.id, true, saddle));
            }
            payload.end_array();

            payload.set_member_name("comments");
            payload.begin_array();
            var comments = yield db.query<TaskComment>()
                .where(expr("task_id == $0", elem<int64?>(task.id)))
                .order_by(expr("id"))
                .materialise_async();
            foreach (var comment in comments) {
                payload.begin_object();
                payload.set_member_name("author");
                payload.add_string_value(comment.author);
                payload.set_member_name("body");
                payload.add_string_value(comment.body);
                if (comment.quote != null) {
                    payload.set_member_name("quote");
                    payload.add_string_value(((!)comment.quote));
                }
                payload.end_object();
            }
            payload.end_array();

            payload.set_member_name("answers");
            payload.begin_array();
            foreach (var pending in yield take_pending_answers_for_task(task.id)) {
                payload.begin_object();
                payload.set_member_name("block_id");
                payload.add_int_value(pending.block_id);
                payload.set_member_name("answer");
                payload.add_string_value(raw_answer_for_saddle(pending.answer));
                if (branch_advanced != null) {
                    payload.set_member_name("branch_advanced");
                    payload.add_boolean_value((!)branch_advanced);
                }
                payload.end_object();
            }
            payload.end_array();
            payload.end_object();

            // The plan-12 wire additions: kind ("task"|"goal_run"),
            // mode and creator ride the top level beside the task.
            payload.set_member_name("kind");
            payload.add_string_value(task.kind);
            payload.set_member_name("mode");
            payload.add_string_value(task.mode);
            payload.set_member_name("creator");
            payload.add_string_value(task.creator);
            // The goal context (goal work only): the goal agent gets
            // the full brief + approved plan; a goal worker gets the
            // same object as read-only reference (the daemon adds the
            // stick-to-mandate shaping) plus its ephemeral branch.
            if (goal != null) {
                payload.set_member_name("goal");
                payload.begin_object();
                payload.set_member_name("id");
                payload.add_int_value(((!)goal).id);
                payload.set_member_name("branch");
                payload.add_string_value(((!)goal).branch_name);
                payload.set_member_name("base_branch");
                payload.add_string_value(((!)goal).base_branch
                    ?? (project != null ? ((!)project).default_branch : ""));
                payload.set_member_name("brief");
                payload.add_string_value(((!)goal).brief);
                payload.set_member_name("plan");
                payload.add_string_value(((!)goal).plan ?? "");
                payload.end_object();
                if (task.kind == TASK_KIND_TASK) {
                    payload.set_member_name("ephemeral_branch");
                    payload.add_string_value(@"task/$(task.goal_id)-$(task.id)");
                }
            }

            // The wire contract's top-level `rules` addition: the
            // effective rule set (task ∪ primary ∪ co-developed, deduped,
            // stable order — see RuleService.effective_rules). Goal work
            // expands co-developed through the goal's links.
            payload.set_member_name("rules");
            payload.begin_array();
            foreach (var rule in yield rules.effective_rules_for(task, projects,
                    goal != null ? task.goal_id : 0)) {
                payload.begin_object();
                payload.set_member_name("name");
                payload.add_string_value(rule.name);
                payload.set_member_name("body");
                payload.add_string_value(rule.body);
                payload.end_object();
            }
            payload.end_array();

            // The wire contract's `git_identity` addition (plan 9): the
            // name/email the daemon commits as, from the saddle row.
            // Omitted entirely when unset so the daemon's fallback
            // identity applies.
            string identity_name = saddle.git_name ?? "";
            string identity_email = saddle.git_email ?? "";
            if (identity_name.length > 0 || identity_email.length > 0) {
                payload.set_member_name("git_identity");
                payload.begin_object();
                payload.set_member_name("name");
                payload.add_string_value(identity_name);
                payload.set_member_name("email");
                payload.add_string_value(identity_email);
                payload.end_object();
            }

            // The wire contract's `capabilities` context (plan 10
            // phase 3): the allocated saddle's own tags plus every tag
            // name known across the fleet (the operator-managed
            // catalogue) — the daemon renders both into its system
            // prompt and validates `task create --capability`.
            payload.set_member_name("capabilities");
            payload.begin_object();
            payload.set_member_name("saddle_tags");
            payload.begin_array();
            foreach (var carried in yield capability_service.list_for_saddle(saddle.id)) {
                payload.add_string_value(carried.name);
            }
            payload.end_array();
            payload.set_member_name("fleet_tags");
            payload.begin_array();
            foreach (var known in yield capability_service.list_capabilities()) {
                payload.add_string_value(known.name);
            }
            payload.end_array();
            payload.end_object();

            // The wire contract's `resume_notice` addition: on a
            // workspace-adopting start (a suspended parent's resume), the
            // daemon injects it as the next user turn instead of the
            // opening prompt.
            if (resume_notice != null && resume_notice.length > 0) {
                payload.set_member_name("resume_notice");
                payload.add_string_value(resume_notice);
            }

            // The wire's `branch_advanced` (resume pushes): whether the
            // shared branch moved while this task was parked — false or
            // absent means the daemon omits the pull instruction entirely.
            if (branch_advanced != null) {
                payload.set_member_name("branch_advanced");
                payload.add_boolean_value((!)branch_advanced);
            }

            // The wire's `context_update` (resume pushes): a short
            // markdown delta of the instruction context that changed
            // while the workspace was parked — the goal's plan
            // approved since the run's last round, rules created,
            // edited or attached, or new project notes. The daemon
            // appends it as a TURN (prompt-cache friendly) instead of
            // refreshing the system prompt; absent when nothing
            // changed.
            if (context_update != null && context_update.length > 0) {
                payload.set_member_name("context_update");
                payload.add_string_value(context_update);
            }

            // The wire contract's `project_notes` addition: notes of the
            // primary and co-developed projects, oldest first, capped at
            // 30 entries / 20 000 chars (head kept).
            var note_project_names = new Dictionary<string, string>();
            var project_notes_rows = new Vector<ClankerSaddle.Data.ProjectNote>();
            if (project != null) {
                note_project_names.set(((!)project).id.to_string(), ((!)project).name);
                foreach (var note in yield projects.list_notes(((!)project).id)) {
                    project_notes_rows.add(note);
                }
            }
            foreach (var linked in co_developed) {
                note_project_names.set(linked.id.to_string(), linked.name);
                foreach (var note in yield projects.list_notes(linked.id)) {
                    project_notes_rows.add(note);
                }
            }
            var project_notes = project_notes_rows.sort((a, b) => (int)(a.id - b.id));
            payload.set_member_name("project_notes");
            payload.begin_array();
            int64 note_entries = 0;
            int64 note_chars = 0;
            foreach (var note in project_notes) {
                if (note_entries >= PROJECT_NOTES_MAX_ENTRIES
                        || note_chars >= PROJECT_NOTES_MAX_CHARS) {
                    break;
                }
                string note_body = note.body;
                if (note_chars + note_body.length > PROJECT_NOTES_MAX_CHARS) {
                    note_body = note_body.substring(0,
                        (int)(PROJECT_NOTES_MAX_CHARS - note_chars));
                }
                payload.begin_object();
                payload.set_member_name("project");
                payload.add_string_value(note_project_names.get_or_default(note.project_id.to_string()) ?? "");
                payload.set_member_name("body");
                payload.add_string_value(note_body);
                payload.end_object();
                note_entries++;
                note_chars += note_body.length;
            }
            payload.end_array();

            // The wire contract's `callback` addition: this app's
            // externally reachable base URL plus the bearer token the
            // daemon presents when posting back — its own per-saddle
            // credential (plan 10), which every `/api/saddle/*` call
            // carries.
            payload.set_member_name("callback");
            payload.begin_object();
            payload.set_member_name("base_url");
            payload.add_string_value(callback_base_url());
            payload.set_member_name("token");
            payload.add_string_value(saddle.token ?? "");
            payload.end_object();

            payload.set_member_name("llm_config");
            payload.begin_object();
            payload.set_member_name("base_url");
            payload.add_string_value(((!)endpoint).base_url);
            payload.set_member_name("api_key");
            payload.add_string_value(((!)endpoint).api_key);
            payload.set_member_name("model");
            payload.add_string_value(((!)endpoint).model);
            payload.end_object();

            // The web-side run row the daemon's pushed telemetry keys on
            // (plan-10 phase-1 wire addition: it rides every assignment
            // so /start consumers can ignore it harmlessly).
            payload.set_member_name("session_id");
            payload.add_int_value(session_id);
            payload.end_object();

            var root = payload.get_root().copy();
            return root.get_object();
        }

        /**
         * Probes a saddle's `GET /status` and records what a probe can
         * tell: the health snapshot, the daemon-reported ssh key and
         * the goal-branch push flags. Busy/idle need NO write anymore —
         * they are derived from the session records (plan 16 item 23).
         * The row's only remaining write is the OFFLINE probe state: a
         * successful probe is proof of life and clears it; a failed
         * probe stamps it through the guarded write (an ask claim that
         * landed while the probe was in flight is a NEWER proof of
         * life and must not be stomped).
         */
        public async void refresh_saddle_status(int64 id) throws Error {
            var saddle = yield get_saddle(id);
            if (saddle == null) {
                throw new Error(Quark.from_string("clanker-saddle"), 17,
                    @"No saddle exists with id $(id)");
            }
            var session = yield running_session_on_saddle(id);
            int64 seen_session = session != null ? ((!)session).id : 0;
            try {
                var status = yield request_saddle_json(((!)saddle).host, "GET", "/status", null);
                yield record_status_probe(((!)saddle), status);
                var clear = connection.create_command(
                    "UPDATE saddles SET status = '" + SADDLE_IDLE
                    + "' WHERE id = :id AND status = '" + SADDLE_OFFLINE + "'");
                clear.with_parameter<Invercargill.Element>("id", elem<int64?>(id));
                yield clear.execute_non_query_async();
            } catch (Error e) {
                yield stamp_offline(id, seen_session);
            }
            yield pulse.bump(0, "saddle status refreshed");
        }

        /**
         * Answered blocks of `task_id` not yet handed over anywhere;
         * marks them delivered — a resumed assignment carries its
         * pending answers in the payload ({@link push_task_to_saddle}).
         */
        public async ImmutableLot<PendingAnswer> take_pending_answers_for_task(int64 task_id) throws Error {
            var blocks = yield db.query<TaskBlock>()
                .where(expr("task_id == $0", elem<int64?>(task_id)))
                .materialise_async();
            var rows = new Vector<PendingAnswer>();
            foreach (var block in blocks) {
                if (block.answer == null) {
                    continue;
                }
                if (!poll_state.mark_delivered(block.id)) {
                    continue;
                }
                rows.add(new PendingAnswer() {
                    block_id = block.id,
                    answer = raw_answer_for_saddle(((!)block).answer ?? "")
                });
            }
            return rows.to_immutable_buffer();
        }

        /**
         * Pushes the task's undelivered human comments to the RUNNING
         * daemon's `POST /task-update {kind: "comment"}` endpoint so
         * they land in the live conversation. Best effort: a daemon
         * without the route (HTTP 404/405) keeps the comments pending,
         * and other failures are logged and retried on the next
         * dispatcher tick.
         */
        public async void deliver_pending_comments(int64 saddle_id, int64 task_id) throws Error {
            var saddle = yield get_saddle(saddle_id);
            if (saddle == null) {
                return;
            }
            var comments = yield db.query<TaskComment>()
                .where(expr("task_id == $0", elem<int64?>(task_id)))
                .order_by(expr("id"))
                .materialise_async();
            foreach (var comment in comments) {
                if (!comment.author.has_prefix("user:")
                        || poll_state.is_comment_delivered(comment.id)) {
                    continue;
                }
                var payload = new Json.Builder();
                payload.begin_object();
                payload.set_member_name("kind");
                payload.add_string_value("comment");
                payload.set_member_name("text");
                payload.add_string_value(comment.body);
                payload.set_member_name("quote");
                payload.add_string_value(comment.quote ?? "");
                payload.end_object();

                Json.Node? root = null;
                int status_code = yield request_saddle(((!)saddle).host, "POST",
                    "/task-update", (owned) payload, out root);
                if (status_code == 200 || status_code == 204) {
                    poll_state.mark_comment_delivered(comment.id);
                } else if (status_code != 404 && status_code != 405) {
                    warning(@"[ClankerSaddle] Comment task-update to saddle $(saddle_id) returned HTTP $(status_code)");
                }
            }
        }

        /**
         * Marks the task's human comments as delivered — they were handed
         * over inside an assignment payload, so neither the poll endpoint
         * nor `/notify` should repeat them.
         */
        public async void mark_task_comments_delivered(int64 task_id) throws Error {
            var comments = yield db.query<TaskComment>()
                .where(expr("task_id == $0", elem<int64?>(task_id)))
                .materialise_async();
            foreach (var comment in comments) {
                if (comment.author.has_prefix("user:")) {
                    poll_state.mark_comment_delivered(comment.id);
                }
            }
        }

        /**
         * The goal id when `task_id` is a goal agent's run task (kind
         * 'goal_run'), else 0 — the discriminator for the goal-context
         * verb handling above.
         */
        private async int64 goal_id_of_run(int64 task_id) throws Error {
            var task = yield db.query<TaskEntity>()
                .where(expr("id == $0", elem<int64?>(task_id)))
                .first_async();
            return task != null && ((!)task).kind == TASK_KIND_GOAL_RUN
                ? ((!)task).goal_id : 0;
        }

        /**
         * Raises the daemon's pushed blocking prompt as a `task_blocks`
         * row exactly once: a not-yet-answered row, or an answered row
         * whose answer has not been delivered to the daemon yet, means the
         * prompt is already represented. The results push's `block`
         * object carries the question plus the block type, options and
         * markdown body (the `/status` field names — `block_question`
         * &c — are accepted as a fallback). `saddle block rfc` carries
         * no question text, so rfc prompts surface under a stable
         * fallback title.
         */
        private async void ensure_block(TaskService tasks, int64 task_id, Json.Object block) throws Error {
            string question = block_question_from_payload(block);
            if (question.length == 0) {
                return;
            }
            string incoming_type = block_string_or(block, "type", "block_type");
            foreach (var existing in yield tasks.list_blocks(task_id)) {
                if (existing.answer == null || !poll_state.is_delivered(existing.id)) {
                    // A pending review-family block of EITHER kind
                    // (review/completion) swallows the other's push: the
                    // daemon's done park and the web's completion raise
                    // are the SAME review raised from two paths — one
                    // row, whichever landed first.
                    if ((incoming_type == BLOCK_REVIEW || incoming_type == BLOCK_COMPLETION)
                            && (existing.block_type == BLOCK_REVIEW
                                || existing.block_type == BLOCK_COMPLETION)) {
                        return;
                    }
                }
                if (existing.question != question) {
                    continue;
                }
                if (existing.answer == null || !poll_state.is_delivered(existing.id)) {
                    return;
                }
            }
            yield tasks.add_block(task_id, question,
                block_string_or(block, "type", "block_type"),
                EntryJson.string_array_member(block, block.has_member("options") ? "options" : "block_options"),
                EntryJson.optional_string_member(block, block.has_member("body") ? "body" : "block_body"));
        }

        /** The first non-empty of two string members (empty when both are). */
        private static string block_string_or(Json.Object obj, string primary, string fallback) {
            string value = EntryJson.string_member(obj, primary);
            return value.length > 0 ? value : EntryJson.string_member(obj, fallback);
        }

        /**
         * The question a daemon block is recognised by: `question`
         * (or the `/status`-style `block_question`) verbatim, or the
         * rfc fallback when the daemon raised an RFC (the
         * `saddle block rfc` shape carries its document as the body and
         * no question operand). Both the block-raising and answer-
         * delivery paths key blocks by this string.
         */
        private static string block_question_from_payload(Json.Object block) {
            string question = block_string_or(block, "question", "block_question");
            if (question.length > 0) {
                return question;
            }
            return block_string_or(block, "type", "block_type") == BLOCK_RFC
                ? "Request for comments" : "";
        }

        /**
         * Applies a pushed state transition (`POST /api/saddle/results`,
         * plan 10 phase 2) — the same task/saddle bookkeeping the
         * deleted /status poll switch ran, driven by the daemon's push:
         * `done`/`failed` complete or fail the task with the pushed
         * summary (closing the session through the release owner) —
         * except on a goal-run session, where they submit the
         * completion report or fail the goal (plan 12); `blocked`
         * raises the block payload as an answerable block on the
         * goal run's OWN task row (plan 15 C1: the goal agent's
         * blocks ride the run task — no mirrored second row exists)
         * and parks the round (the raise closes its session 'parked',
         * D4); `suspended`
         * parks the same way — for a goal agent the resume is driven
         * by the goal's task-terminal events; and `working` flips an
         * ASSIGNED task to working (a parked task keeps its park —
         * the verdict arrives as a resuming allocation, never as a
         * state push; plan 16 D4). No saddle-row write remains
         * (plan 16 item 23): busy/idle derive from the sessions.
         *
         * Plan 13 B2: a push against a task that is already terminal —
         * or a goal-context push against a finished goal — is an
         * outbox redelivery and is consumed silently, whatever state
         * it carries; terminal transitions and their narrative fire
         * once, ever.
         */
        public async void apply_result_state(ClankerSaddle.Data.Session session,
                string state, Json.Object? block, string? summary, TaskService tasks) throws Error {
            int64 task_id = session.task_id;
            int64 goal_id = yield goal_id_of_run(task_id);
            bool terminal = yield task_is_terminal(task_id);
            if (!terminal && goal_id > 0) {
                terminal = yield goal_is_terminal(goal_id);
            }
            if (terminal) {
                return;
            }

            switch (state) {
                case "done":
                    if (goal_id > 0) {
                        // Goal context: `saddle done` submits the
                        // completion report and enters completion
                        // review — the run parks on the mirrored
                        // block (submit_completion's raise already
                        // closed the session 'parked'), it does not
                        // complete.
                        yield goal_service.submit_completion(goal_id,
                            summary != null && summary.length > 0 ? summary
                                : "The goal agent submitted its completion report");
                        break;
                    }
                    yield tasks.complete_task(task_id, summary, OUTCOME_DONE);
                    break;
                case "failed":
                    string explanation = summary ?? "";
                    if (goal_id > 0) {
                        yield goal_service.fail_goal(goal_id,
                            explanation.length > 0 ? explanation : null);
                        break;
                    }
                    yield tasks.fail_task(task_id, explanation.length > 0 ? explanation : null);
                    break;
                case "blocked":
                    if (goal_id > 0) {
                        // Goal context: the block lands on the run
                        // task itself (plan 15 C1) — a plan payload
                        // enters plan review, anything else raises a
                        // goal question/RFC; the run parks on that
                        // very row for the verdict resume. A BLOCKED
                        // push WITHOUT a
                        // block payload (the park-unwinding round-end
                        // push, its trailing entries) raises nothing —
                        // the payload-carrying transition already did,
                        // and a synthesised fallback row would dupe it.
                        if (block == null) {
                            yield tasks.set_status(task_id, STATUS_BLOCKED);
                            yield tasks.park_open_sessions(task_id);
                            break;
                        }
                        string question = block_question_from_payload((!)block);
                        if (question.length == 0) {
                            question = "The goal agent raised a question";
                        }
                        if (block_string_or((!)block, "type", "block_type") == BLOCK_PLAN) {
                            yield goal_service.submit_plan(goal_id,
                                EntryJson.optional_string_member((!)block, "body") ?? question);
                        } else if (block_string_or((!)block, "type", "block_type") == BLOCK_REVIEW) {
                            // The daemon's submit_report done-park: a
                            // review-kind block on the goal run IS the
                            // completion submission (submit_completion
                            // adopts the raised row — one row, whichever
                            // path landed first — and drives the
                            // completion_review transition).
                            yield goal_service.submit_completion(goal_id,
                                EntryJson.optional_string_member((!)block, "body") ?? question);
                        } else {
                            yield goal_service.raise_agent_block(goal_id, question,
                                block_string_or((!)block, "type", "block_type"),
                                EntryJson.string_array_member((!)block,
                                    (!)block.has_member("options") ? "options" : "block_options"),
                                EntryJson.optional_string_member((!)block, "body"));
                        }
                        yield tasks.set_status(task_id, STATUS_BLOCKED);
                        break;
                    }
                    yield ensure_block(tasks, task_id, block != null ? (!)block : new Json.Object());
                    {
                        // A review-kind block parks as review_requested
                        // (the banner state); every other kind as
                        // blocked.
                        bool review = block != null
                            && block_string_or((!)block, "type", "block_type") == BLOCK_REVIEW;
                        yield tasks.set_status(task_id,
                            review ? STATUS_REVIEW_REQUESTED : STATUS_BLOCKED);
                    }
                    // The daemon parks the task's workspace while
                    // blocked, freeing the saddle for other work
                    // (plan 16 item 23: the raise already closed the
                    // round's session 'parked' and the task's blocked
                    // status IS the pause — no saddle-row write
                    // remains; busy/idle derive from the sessions).
                    break;
                case "suspended":
                    // `saddle suspend`: the same parking as a block, but
                    // the web drives resumption — a wake of the affine
                    // saddle once a goal event lands, the resume
                    // assignment carrying the coalesced report.
                    // (set_parked closed the round's session 'parked';
                    // the freed saddle is the derived view's reading.)
                    yield tasks.set_parked(task_id);
                    break;
                case "working":
                    // A working push only advances a task that is
                    // actually running — assigned (the claim's flip)
                    // or a legacy never-allocated queued row. A
                    // PARKED task (blocked / review / suspended, its
                    // session closed at the raise — plan 16 D4) keeps
                    // its park: the verdict arrives as a resuming
                    // allocation, never as a state push, so a
                    // park-unwinding or late round push can never
                    // resurrect the workspace as "working" out from
                    // under its review banner.
                    var advance = connection.create_command(
                        "UPDATE tasks SET status = 'working' WHERE id = :id"
                        + " AND status IN ('assigned', 'queued')"
                        + " AND status != 'working'");
                    advance.with_parameter<Invercargill.Element>("id",
                        elem<int64?>(task_id));
                    bool flipped = (yield advance.execute_non_query_async()) > 0;
                    var current = yield db.query<TaskEntity>()
                        .where(expr("id == $0", elem<int64?>(task_id)))
                        .first_async();
                    if (flipped) {
                        var moved = new Json.Object();
                        moved.set_string_member("from", STATUS_ASSIGNED);
                        moved.set_string_member("to", STATUS_WORKING);
                        moved.set_string_member("text", "Status → working");
                        yield event_service.append(EVENT_STATE_CHANGE,
                            EVENT_ACTOR_SYSTEM, EVENT_SCOPE_TASK, task_id, moved);
                    }
                    // started_at (and the first-start push) only stamp
                    // for work that never allocated; the guarded
                    // advance above means the re-read task is working
                    // exactly when the flip landed or it already ran.
                    if (current != null && ((!)current).status == STATUS_WORKING
                            && ((!)current).started_at == null) {
                        ((!)current).started_at = new DateTime.now_local();
                        yield db.update_async<TaskEntity>((!)current);
                        // The first-start push lives at the allocation
                        // (where started_at is stamped for dispatched
                        // tasks); this path only covers legacy status
                        // pushes for never-allocated work.
                        push.notify_opted_in(task_id, PushService.lifecycle_notification(
                            ((!)current).title, task_id, "started", null));
                    }
                    break;
                default:
                    break;
            }
        }

        /** True when the task's status is done or failed. */
        private async bool task_is_terminal(int64 task_id) throws Error {
            var task = yield db.query<TaskEntity>()
                .where(expr("id == $0", elem<int64?>(task_id)))
                .first_async();
            return task != null && (((!)task).status == STATUS_DONE
                || ((!)task).status == STATUS_FAILED);
        }

        /** True when the goal's status is complete, failed or cancelled. */
        private async bool goal_is_terminal(int64 goal_id) throws Error {
            var goal = yield db.query<Goal>()
                .where(expr("id == $0", elem<int64?>(goal_id)))
                .first_async();
            return goal != null && (((!)goal).status == GOAL_COMPLETE
                || ((!)goal).status == GOAL_FAILED || ((!)goal).status == GOAL_CANCELLED);
        }

        /**
         * The running-liveness probe (plan 10 phase 2): a busy saddle's
         * session has seen neither chunks nor results for the staleness
         * window, so ONE exceptional `GET /status` decides its fate —
         * an answer means the daemon is alive (the health snapshot is
         * refreshed and the task keeps running; the silence episode
         * gets no second probe), a failure requeues the task with an agent
         * comment and marks the saddle offline (the normal unreachable
         * bookkeeping). The probe flag was taken by the caller; the
         * next push resets it.
         */
        public async void probe_stale_saddle(Saddle saddle, TaskService tasks,
                int64 running_task_id) throws Error {
            try {
                var status = yield request_saddle_json(saddle.host, "GET", "/status", null);
                yield record_status_probe(saddle, status);
                print(@"[ClankerSaddle] Stale saddle '$(saddle.name)' answered its exceptional status probe; task $(running_task_id) keeps running\n");
            } catch (Error e) {
                warning(@"[ClankerSaddle] Saddle '$(saddle.name)' sent neither chunks nor results for $(stale_after_seconds())s and its status probe failed ($(e.message)); marking offline");
                yield mark_saddle_unreachable(saddle.id, tasks,
                    @"no chunks or results for $(stale_after_seconds())s; status probe failed ($(e.message))");
            }
        }

        /**
         * Performs an HTTP request against a saddle daemon and parses the
         * response body as JSON; transport failures propagate to the
         * caller (the dispatcher marks the saddle unreachable). Transport
         * blips — typically a pooled keep-alive connection closed by the
         * daemon at the same instant the request reused it — are retried
         * once before giving up. Unlike {@link request_saddle_json} the
         * HTTP status is returned rather than thrown, and array bodies
         * parse too.
         */
        private async int request_saddle(string host, string method, string path,
                owned Json.Builder? payload, out Json.Node? root) throws Error {
            root = null;
            GLib.Error? failure = null;
            for (var attempt = 0; attempt < 2; attempt++) {
                try {
                    root = null;
                    int status = yield request_saddle_once(host, method, path, payload, out root);
                    return status;
                } catch (Error e) {
                    failure = e;
                }
                if (((!)failure).code != 13 && ((!)failure).code != 14) {
                    break;
                }
                GLib.Timeout.add(250, request_saddle.callback);
                yield;
            }
            throw (!)failure;
        }

        private async int request_saddle_once(string host, string method, string path,
                Json.Builder? payload, out Json.Node? root) throws Error {
            root = null;
            string base_url = host.has_prefix("http") ? host : @"http://$(host)";
            var message = new Soup.Message(method, @"$(base_url)$(path)");
            if (payload != null) {
                var generator = new Json.Generator();
                generator.set_root(((!)payload).get_root());
                message.set_request_body_from_bytes("application/json",
                    new Bytes(generator.to_data(null).data));
            }

            InputStream input;
            try {
                input = yield http.send_async(message, Priority.DEFAULT, null);
            } catch (Error e) {
                throw new Error(Quark.from_string("clanker-saddle"), 13,
                    @"Saddle request to $(base_url)$(path) failed: $(e.message)");
            }

            var output = new MemoryOutputStream.resizable();
            try {
                yield output.splice_async(input, OutputStreamSpliceFlags.CLOSE_SOURCE,
                    Priority.DEFAULT, null);
                yield output.close_async(Priority.DEFAULT, null);
            } catch (Error e) {
                throw new Error(Quark.from_string("clanker-saddle"), 14,
                    @"Failed reading saddle response: $(e.message)");
            }

            var written = output.get_data_size();
            uint8[] data = output.steal_data();
            // steal_data's array length is -1 and the buffer is sized to
            // the data — there is no NUL terminator to rely on — so slice
            // by the written byte count before reading it as a string.
            string response_text = written > 0 && data != null && data[0] != '\0'
                ? ((string) data).substring(0, (long) written) : "{}";

            var parser = new Json.Parser();
            try {
                parser.load_from_data(response_text);
            } catch (Error e) {
                throw new Error(Quark.from_string("clanker-saddle"), 16,
                    @"Saddle returned invalid JSON: $(e.message)");
            }
            root = parser.get_root();
            return (int) message.status_code;
        }

        /**
         * Performs an HTTP request against a saddle daemon and returns the
         * parsed JSON object body (an empty object for non-object bodies);
         * HTTP statuses of 400 and above throw.
         */
        private async Json.Object request_saddle_json(string host, string method, string path,
                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)");
            }
            if (root == null || root.get_node_type() != Json.NodeType.OBJECT) {
                return new Json.Object();
            }
            return root.get_object();
        }

        /**
         * A `{"name", "git_url", "path", "branch"}` project reference in
         * the saddle daemon's wire shape: `git_url` is the clone source,
         * and the checkout lives at
         * `SADDLE_PROJECTS_ROOT/task-<task_id>/<name>` inside the
         * saddle's container — one workspace per task, so concurrent
         * tasks never share a checkout.
         */
        /**
         * The projects root assignment paths are addressed against: the
         * ASKING saddle's registered `workspace_root` (the daemon's own
         * SADDLE_WORKSPACE_ROOT, reported at /register — goal 8 task 6,
         * so a fleet no longer assumes one shared /workspace mount),
         * else the `SADDLE_PROJECTS_ROOT` environment override, else
         * the historical "/workspace" default. Paths are ADVISORY: the
         * daemon owns its filesystem and rewrites them onto its own
         * workspace root.
         */
        private static string projects_root_for_saddle(Saddle? saddle) {
            var registered = saddle != null ? ((!)saddle).workspace_root : null;
            if (registered != null && ((!)registered).strip().length > 0) {
                return ((!)registered).strip();
            }
            var root_var = (Environment.get_variable("SADDLE_PROJECTS_ROOT") ?? "").strip();
            return root_var.length > 0 ? root_var : "/workspace";
        }

        private static Json.Node project_node(Project project, int64 task_id, bool is_reference = false,
                Saddle? saddle = null) {
            string projects_root = projects_root_for_saddle(saddle);
            var builder = new Json.Builder();
            builder.begin_object();
            builder.set_member_name("name");
            builder.add_string_value(project.name);
            builder.set_member_name("git_url");
            builder.add_string_value(project.git_url);
            builder.set_member_name("path");
            builder.add_string_value(is_reference
                ? @"$(projects_root)/task-$(task_id)/reference/$(project.name)"
                : @"$(projects_root)/task-$(task_id)/$(project.name)");
            builder.set_member_name("branch");
            builder.add_string_value(project.default_branch);
            builder.end_object();
            return builder.get_root();
        }

        /**
         * The answer as the saddle daemon expects it over /answer (or in a
         * resume payload): verbatim. The daemon applies its own
         * BlockRequest formatting, so this only strips the "Human
         * selected: …"/"Human answered (other): …" prefixes still stored
         * by answers recorded before the daemon took over formatting.
         */
        private static string raw_answer_for_saddle(string answer) {
            if (answer.has_prefix("Human selected: ")) {
                return answer.substring("Human selected: ".length);
            }
            if (answer.has_prefix("Human answered (other): ")) {
                return answer.substring("Human answered (other): ".length);
            }
            return answer;
        }
    }
}