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 {

    /** 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 human comment awaiting delivery to one saddle's conversation. */
    public class PendingComment : Object {
        public int64 comment_id { get; set; }
        public string author { get; set; default = ""; }
        public string body { get; set; default = ""; }
        public string quote { get; set; default = ""; }
    }

    /** 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 = ""; }
    }

    /**
     * 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, finds idle capacity for the dispatcher's wake
     * computation, 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 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 = "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 = string_member(response, "ssh_public_key");
                if (reported_key.length > 0 && reported_key != (saddle.ssh_public_key ?? "")) {
                    saddle.ssh_public_key = reported_key;
                    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");
            }
        }

        /**
         * Startup migration of the per-saddle token: every saddle row
         * gets its canonical token — the latest legacy `api_tokens`
         * value it already holds (daemons registered before plan 10
         * keep their credential), or a freshly minted one. Existing
         * rows with a token are left alone.
         */
        public async void ensure_saddle_tokens() throws Error {
            foreach (var saddle in yield list_saddles()) {
                if ((saddle.token ?? "").length > 0) {
                    continue;
                }
                string? legacy = yield token_for_saddle(saddle.id);
                saddle.token = legacy != null && ((!)legacy).length > 0
                    ? (!)legacy : mint_saddle_token();
                yield db.update_async<Saddle>(saddle);
                print(@"[ClankerSaddle] Saddle '$(saddle.name)' token ready ($(saddle.token.substring(0, 12))…)\n");
            }
        }

        /**
          * Generates and stores a fresh API token for `saddle_id`,
          * replacing any earlier one on the wire (older rows remain valid
          * until removed; in practice each saddle holds one).
          */
        public async string generate_token(int64 saddle_id) throws Error {
            var row = new ApiToken() {
                saddle_id = saddle_id,
                token = @"saddle-$(Uuid.string_random().replace("-", ""))",
                created_at = new DateTime.now_local()
            };
            yield db.insert_async<ApiToken>(row);
            return row.token;
        }

        /**
         * True when `token` matches the credential of `saddle_id` — its
         * per-saddle row token (plan 10), or any legacy `api_tokens`
         * row still held for it (tokens minted before the column
         * existed keep validating) — the credential check for the
         * `/api/saddle/*` endpoints.
         */
        public async bool validate_token(int64 saddle_id, string token) throws Error {
            if (token.length == 0) {
                return false;
            }
            var saddle = yield get_saddle(saddle_id);
            if (saddle != null && ((!)saddle).token == token) {
                return true;
            }
            var rows = yield db.query<ApiToken>()
                .where(expr("saddle_id == $0", elem<int64?>(saddle_id)))
                .materialise_async();
            foreach (var row in rows) {
                if (row.token == token) {
                    return true;
                }
            }
            return false;
        }

        /**
         * The saddle a bearer token belongs to, or null — the
         * credential check for the id-less `/api/saddle/*` endpoints
         * (`next-task`, `note`, `task`): the row token
         * first, then any legacy `api_tokens` row (mapped to its
         * saddle). Removing the saddle revokes both.
         */
        public async Saddle? saddle_for_token(string token) throws Error {
            if (token.length == 0) {
                return null;
            }
            var by_row = yield db.query<Saddle>()
                .where(expr("token == $0", elem(token)))
                .first_async();
            if (by_row != null) {
                return by_row;
            }
            var legacy = yield db.query<ApiToken>()
                .where(expr("token == $0", elem(token)))
                .first_async();
            if (legacy == null) {
                return null;
            }
            return yield get_saddle(((!)legacy).saddle_id);
        }

        /**
         * The most recent API token issued for `saddle_id`, or null: the
         * row token first, then the latest legacy `api_tokens` row.
         */
        public async string? token_for_saddle(int64 saddle_id) throws Error {
            var saddle = yield get_saddle(saddle_id);
            if (saddle != null && (((!)saddle).token ?? "").length > 0) {
                return ((!)saddle).token;
            }
            ApiToken? latest = null;
            var rows = yield db.query<ApiToken>()
                .where(expr("saddle_id == $0", elem<int64?>(saddle_id)))
                .order_by(expr("id"))
                .materialise_async();
            foreach (var row in rows) {
                latest = row;
            }
            return latest != null ? ((!)latest).token : null;
        }

        /**
         * 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; tasks it was running are
         * re-queued. Revocation: the row (and its per-saddle token)
         * dies here, and every legacy `api_tokens` row held for the
         * saddle is deleted — 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;
            }
            if (((!)saddle).current_task_id > 0) {
                var task = yield db.query<TaskEntity>()
                    .where(expr("id == $0", elem<int64?>(((!)saddle).current_task_id)))
                    .first_async();
                if (task != null && (((!)task).status == STATUS_ASSIGNED
                        || ((!)task).status == STATUS_WORKING || ((!)task).status == STATUS_BLOCKED
                        || ((!)task).status == STATUS_REVIEW_REQUESTED)) {
                    ((!)task).status = STATUS_QUEUED;
                    yield db.update_async<TaskEntity>((!)task);
                }
            }
            foreach (var row in yield db.query<ApiToken>()
                    .where(expr("saddle_id == $0", elem<int64?>(id)))
                    .materialise_async()) {
                yield db.delete_async<ApiToken>(row);
            }
            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();
        }

        /**
         * Lists saddles with status 'busy' (the dispatcher's poll
         * candidates).
         */
        public async ImmutableLot<Saddle> list_busy_saddles() throws Error {
            return yield db.query<Saddle>()
                .where(expr("status == $0", elem("busy")))
                .materialise_async();
        }

        /**
         * Records a /status observation into the health snapshot,
         * tolerating older daemons without the uptime/activity fields.
         */
        /**
         * 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 — because the daemon surfaces
         * `resumed` on the FIRST status after a workspace restore,
         * whichever probe happens to receive it (the dispatcher's poll,
         * a heal pass, a manual refresh) — the once-per-session resume
         * progress entry.
         */
        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 12: a daemon running (or parking — the workspace
            // stays while the saddle frees) a goal agent exposes the
            // push state of the goal branches on /status; cache it on
            // the goal row for the completion-review UI.
            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);
                    string raw = generator.to_data(null);
                    int64 push_goal_id = 0;
                    if (saddle.current_task_id > 0) {
                        push_goal_id = yield goal_id_of_run(saddle.current_task_id);
                    } else {
                        // Parked: the open session still names the
                        // goal agent's workspace.
                        var open_session = yield open_session_on_saddle(saddle.id);
                        if (open_session != null) {
                            push_goal_id = yield goal_id_of_run(((!)open_session).task_id);
                        }
                    }
                    if (push_goal_id > 0) {
                        yield goal_service.store_branch_push_state(push_goal_id, raw);
                    }
                }
            }

            // The "interrupted -> resumed" timeline arc. Absent on older
            // daemons; marked once per session in SaddlePollState.
            if (status.has_member("resumed")
                    && status.get_member("resumed").get_value_type() == typeof(bool)
                    && status.get_boolean_member("resumed")
                    && saddle.current_task_id > 0) {
                var session = yield task_service.open_session_for_task(saddle.current_task_id);
                if (session != null && poll_state.mark_resumed(((!)session).id)) {
                    yield task_service.add_progress(saddle.current_task_id,
                        "Saddle restored the session context; task resumed");
                }
            }
        }

        /**
         * 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
         * whose own open session was found (NOT re-resolved through the
         * 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 open_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_comment(session.task_id, "agent", @"$(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_comment(session.task_id, "agent",
                    @"$(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) keep their sessions, which the
         * resume machinery needs. Each move lands a agent 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 != "offline") {
                        continue;
                    }
                    print(@"[ClankerSaddle] Task $(task.id) requeued: its saddle went offline\n");
                    yield tasks.requeue_task(task.id);
                    yield tasks.add_comment(task.id, "agent",
                        "saddle went offline; task requeued");
                }
            }
        }

        /**
         * Parks a saddle (idle, no task) only when it still runs
         * `task_id` — a single conditional UPDATE, atomic against the
         * ask allocation claiming a NEW task for the saddle between the
         * poller's read and its write (the poller's "blocked"/"suspended"
         * parking races endpoint-side claims otherwise, leaving the
         * freshly-claimed task assigned to a busy@0 saddle nothing
         * polls).
         */
        public async void park_if_current(int64 saddle_id, int64 task_id) throws Error {
            var command = connection.create_command(
                "UPDATE saddles SET status = 'idle', current_task_id = 0, allocated_at = NULL"
                + " WHERE id = :id AND current_task_id = :task");
            command.with_parameter<Invercargill.Element>("id", elem<int64?>(saddle_id));
            command.with_parameter<Invercargill.Element>("task", elem<int64?>(task_id));
            yield command.execute_non_query_async();
            yield pulse.bump(0, "saddle parked");
        }

        /**
         * Every saddle with status 'idle', registration order — the
         * dispatcher's capability-matching candidate list.
         */
        public async ImmutableLot<Saddle> list_idle_saddles() throws Error {
            return yield db.query<Saddle>()
                .where(expr("status == $0", elem("idle")))
                .order_by(expr("id"))
                .materialise_async();
        }

        /**
         * Marks a saddle offline when its control server cannot be
         * reached, releasing the task it was holding back to 'queued'
         * and closing that task's open session (outcome 'stopped'). The
         * requeue lands as a agent comment on the task, so the timeline
         * shows the interruption; combined with the daemon's
         * crash-resume, the next dispatch restores the workspace and
         * reports `resumed` on /status.
         */
        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;
            }
            int64 held_task_at_entry = ((!)saddle).current_task_id;
            if (((!)saddle).current_task_id > 0) {
                int64 task_id = ((!)saddle).current_task_id;
                bool keep_parked = false;
                var task = yield db.query<TaskEntity>()
                    .where(expr("id == $0", elem<int64?>(task_id)))
                    .first_async();
                if (task != null && (((!)task).status == STATUS_ASSIGNED
                        || ((!)task).status == STATUS_WORKING || ((!)task).status == STATUS_BLOCKED
                        || ((!)task).status == STATUS_REVIEW_REQUESTED)) {
                    keep_parked = yield tasks.newest_block_answered(task_id);
                    if (keep_parked) {
                        // Parked-with-answered: the verdict must survive
                        // for the resume (answer delivery, or the
                        // affinity allocation when the daemon asks
                        // again) — the session stays open and the task
                        // stays parked.
                        print(@"[ClankerSaddle] Task $(task_id) keeps its parked session: its answered block awaits delivery\n");
                    } else {
                        ((!)task).status = STATUS_QUEUED;
                        yield db.update_async<TaskEntity>((!)task);
                        string unreachable_text = reason != null && reason.length > 0
                            ? @"Saddle became unreachable ($(reason)); task requeued"
                            : "Saddle became unreachable; task requeued";
                        yield tasks.add_comment(task_id, "agent", unreachable_text);
                    }
                }

                if (!keep_parked) {
                    var sessions = yield db.query<ClankerSaddle.Data.Session>()
                        .where(expr("task_id == $0", elem<int64?>(task_id)))
                        .materialise_async();
                    foreach (var session in sessions) {
                        if (session.ended_at != null) {
                            continue;
                        }
                        session.ended_at = new DateTime.now_local();
                        session.outcome = "stopped";
                        yield db.update_async<ClankerSaddle.Data.Session>(session);
                    }
                }
            }
            var offline_write = connection.create_command(
                "UPDATE saddles SET status = 'offline', current_task_id = 0, allocated_at = NULL"
                + " WHERE id = :id AND current_task_id = :seen");
            offline_write.with_parameter<Invercargill.Element>("id", elem<int64?>(id));
            offline_write.with_parameter<Invercargill.Element>("seen",
                elem<int64?>(held_task_at_entry));
            yield offline_write.execute_non_query_async();
            // 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");
        }

        /**
         * 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 = yield db.query<Project>()
                .where(expr("id == $0", elem<int64?>(task.project_id)))
                .first_async();
            if (project == null) {
                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.
            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 {
                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).name);
            payload.set_member_name("project_branch");
            payload.add_string_value(((!)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).default_branch);

            payload.set_member_name("co_developed");
            payload.begin_array();
            payload.add_value(project_node(((!)project), task.id));
            foreach (var linked in co_developed) {
                payload.add_value(project_node(linked, task.id));
            }
            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));
            }
            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).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>();
            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.
            string callback_token = saddle.token ?? "";
            if (callback_token.length == 0) {
                string? legacy = yield token_for_saddle(saddle.id);
                callback_token = legacy ?? "";
            }
            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(callback_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 refreshes its registered
         * status: 'idle' or 'busy' from the daemon state, 'offline' when
         * the daemon cannot be reached. The write is guarded on the
         * assignment seen at load — an ask allocation may claim the
         * saddle while the probe is in flight, 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)");
            }
            int64 seen_task = ((!)saddle).current_task_id;
            string next_status = "idle";
            int64 next_task = 0;
            DateTime? next_allocated = null;
            try {
                var status = yield request_saddle_json(((!)saddle).host, "GET", "/status", null);
                string state = status.has_member("state") ? status.get_string_member_with_default("state", "") : "";
                yield record_status_probe(((!)saddle), status);
                if (state == "working") {
                    // Reconcile with the web's own records: busy only when
                    // a live session truly holds this saddle.
                    var session = yield open_session_on_saddle(id);
                    next_status = "busy";
                    next_task = session != null ? ((!)session).task_id : 0;
                    next_allocated = session != null ? ((!)session).started_at : null;
                }
                // Idle AND the parked states (blocked/suspended, plus
                // terminal and unknown states): the workspace is
                // parked, so the saddle is logically free — marking it
                // busy here used to wedge assignment against a daemon
                // that would never poll.
            } catch (Error e) {
                next_status = "offline";
            }
            var command = connection.create_command(
                "UPDATE saddles SET status = :status, current_task_id = :task,"
                + " allocated_at = :allocated WHERE id = :id AND current_task_id = :seen");
            command.with_parameter<Invercargill.Element>("status", elem(next_status));
            command.with_parameter<Invercargill.Element>("task", elem<int64?>(next_task));
            command.with_parameter<Invercargill.Element>("allocated", elem<DateTime?>(next_allocated));
            command.with_parameter<Invercargill.Element>("id", elem<int64?>(id));
            command.with_parameter<Invercargill.Element>("seen", elem<int64?>(seen_task));
            yield command.execute_non_query_async();
            yield pulse.bump(0, "saddle status refreshed");
        }

        /**
         * The saddle's open (live or parked) session, or null — the
         * reconciliation source for whether a daemon is genuinely
         * running one of our tasks.
         */
        private async ClankerSaddle.Data.Session? open_session_on_saddle(int64 saddle_id) throws Error {
            var sessions = yield db.query<ClankerSaddle.Data.Session>()
                .where(expr("saddle_id == $0", elem<int64?>(saddle_id)))
                .order_by(expr("id"))
                .materialise_async();
            ClankerSaddle.Data.Session? newest_open = null;
            foreach (var session in sessions) {
                if (session.ended_at == null) {
                    newest_open = session;
                }
            }
            return newest_open;
        }

        /**
         * Answered blocks of the saddle's current task not yet handed over
         * the poll endpoint; marks them delivered.
         */
        public async ImmutableLot<PendingAnswer> take_pending_answers(int64 saddle_id) throws Error {
            var saddle = yield get_saddle(saddle_id);
            if (saddle == null || ((!)saddle).current_task_id == 0) {
                return new Vector<PendingAnswer>().to_immutable_buffer();
            }
            return yield take_pending_answers_for_task(((!)saddle).current_task_id);
        }

        /**
         * Answered blocks of `task_id` not yet handed over anywhere;
         * marks them delivered. Used both by the poll endpoint and by
         * {@link push_task_to_saddle} (a resumed assignment carries its
         * pending answers in the payload).
         */
        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();
        }

        /**
         * Human comments of the saddle's current task not yet delivered to
         * the daemon; marks them delivered (the poll response hands them
         * over exactly once).
         */
        public async ImmutableLot<PendingComment> take_pending_comments(int64 saddle_id) throws Error {
            var saddle = yield get_saddle(saddle_id);
            if (saddle == null || ((!)saddle).current_task_id == 0) {
                return new Vector<PendingComment>().to_immutable_buffer();
            }
            var comments = yield db.query<TaskComment>()
                .where(expr("task_id == $0", elem<int64?>(((!)saddle).current_task_id)))
                .order_by(expr("id"))
                .materialise_async();
            var rows = new Vector<PendingComment>();
            foreach (var comment in comments) {
                if (!comment.author.has_prefix("user:")
                        || poll_state.is_comment_delivered(comment.id)) {
                    continue;
                }
                poll_state.mark_comment_delivered(comment.id);
                rows.add(new PendingComment() {
                    comment_id = comment.id,
                    author = comment.author,
                    body = comment.body,
                    quote = comment.quote ?? ""
                });
            }
            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);
                }
            }
        }

        /**
         * Projects the agent's conversation-level events onto the task:
         * `saddle comment` lines become agent comments, `saddle plan`
         * lines update the plan, `saddle progress-update` lines append
         * progress, `saddle review` raises the review banner, the
         * `saddle done` entry completes the task with the summary the
         * agent gave, and the `saddle failure` entry fails it with the
         * explanation the agent piped in. `saddle yeehaw` entries are
         * handled explicitly but project nothing — the stored
         * `session_commands` row IS the celebration the card, badge
         * and timeline read back. Returns true once a done or failed
         * event finished the task.
         *
         * Plan 12: on a GOAL-RUN session the plan/done/failure verbs
         * are goal-context — a plan submission enters plan review, a
         * done submits the completion report (the goal-run task parks
         * instead of completing), a failure fails the goal — and
         * comments also land on the goal's thread.
         *
         * Plan 13 B2: the projections key on the DEDUPED entry stream
         * ({@link store_pushed_entries} claims every id once per
         * saddle, ever) and stop at terminal state — a task already
         * done/failed, a goal already finished, projects nothing
         * further. The live failure's 115-event/187-comment fan-out is
         * structurally impossible: replays never reach this function,
         * and the honest burst of DISTINCT failure ids dies at the
         * first one.
         */
        public async bool apply_entry_events(int64 task_id, Vector<Json.Object> entries,
                TaskService tasks) throws Error {
            var 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 false;
            }
            foreach (var entry in entries) {
                string command = string_member(entry, "command");
                string output = string_member(entry, "output");
                switch (string_member(entry, "type")) {
                    case "comment":
                        // Plan 13 B4: body-first — the entry's output IS
                        // the comment body; the command echo (the
                        // `saddle comment` invocation, heredoc shapes
                        // included) must never become the projected
                        // text. The parsed command stays as a legacy
                        // fallback for pushes without output.
                        string body = output;
                        if (body.length == 0) {
                            body = strip_quotes(strip_verb(command, "saddle comment"));
                        }
                        if (body.length > 0) {
                            yield tasks.add_comment(task_id, "agent", body);
                            if (goal_id > 0) {
                                yield goal_service.add_comment(goal_id, "agent", body);
                                yield goal_service.record_event(goal_id, "comment",
                                    snippet_text(body));
                            }
                        }
                        break;
                    case "plan":
                        // Plan 12: plans are goal-context only — a
                        // goal-run's `saddle plan` enters plan review;
                        // plain task runs carry no plan member.
                        if (goal_id > 0) {
                            string markdown = output.length > 0 ? output
                                : strip_quotes(strip_verb(command, "saddle plan"));
                            if (markdown.length > 0) {
                                yield goal_service.submit_plan(goal_id, markdown);
                            }
                        }
                        break;
                    case "progress":
                        string summary = strip_verb(command, "saddle progress-update");
                        if (summary.length == 0) {
                            summary = output;
                        }
                        if (summary.length > 0) {
                            yield tasks.add_progress(task_id, strip_quotes(summary));
                        }
                        break;
                    case "review":
                        yield tasks.request_review(task_id,
                            output.length > 0 ? output
                            : strip_quotes(strip_verb(command, "saddle review")));
                        break;
                    case "done":
                        if (goal_id > 0) {
                            string report = output.length > 0 ? output
                                : strip_quotes(strip_verb(command, "saddle done"));
                            yield goal_service.submit_completion(goal_id,
                                report.length > 0 ? report : "The goal agent submitted its completion report");
                            return true;
                        }
                        {
                            // the DONE entry's output IS the report
                            // (the command is just "saddle done") —
                            // falling back to it keeps the summary when
                            // this entry path beats the transition push
                            string report = output.length > 0 ? output
                                : strip_quotes(strip_verb(command, "saddle done"));
                            yield tasks.complete_task(task_id,
                                report.length > 0 ? report : null, null, "done");
                        }
                        return true;
                    case "failed":
                        string explanation = output.length > 0 ? output
                            : strip_quotes(strip_verb(command, "saddle failure"));
                        if (goal_id > 0) {
                            yield goal_service.fail_goal(goal_id,
                                explanation.length > 0 ? explanation : null);
                            return true;
                        }
                        yield tasks.fail_task(task_id, explanation);
                        return true;
                    case "yeehaw":
                        // The celebration needs no task-side projection:
                        // store_pushed_entries already persisted the row
                        // the yeehaw card, badge and timeline read from.
                        break;
                    default:
                        break;
                }
            }
            return false;
        }

        /**
         * 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;
        }

        /** A one-line snippet of a body (narrative event text). */
        private static string snippet_text(string body) {
            string flat = body.replace("\n", " ").strip();
            return flat.length > 160 ? flat.substring(0, 160) + "…" : flat;
        }

        /**
         * 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;
            }
            foreach (var existing in yield tasks.list_blocks(task_id)) {
                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"),
                string_array_member(block, block.has_member("options") ? "options" : "block_options"),
                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 = string_member(obj, primary);
            return value.length > 0 ? value : 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") == "rfc"
                ? "Request for comments" : "";
        }

        /**
         * Stores a pushed batch of raw PTY chunks (`POST
         * /api/saddle/chunks`, plan 10 phase 2) as the open session's
         * `session_raw_log` rows — the same storage logic the deleted
         * `GET /log` poller ran, driven by the daemon's flusher instead
         * of the web's cursor. Idempotent on the unique
         * (session_id, seq) pair: a chunk whose row already exists (an
         * outbox redelivery after a lost response) is consumed, not an
         * error; anything else stops the pass so the daemon's retry
         * re-sends the tail. A truly empty text stores no row (it
         * renders as nothing); whitespace-only chunks DO store. Returns
         * the count of rows newly stored.
         */
        public async int64 store_pushed_chunks(int64 session_id, Json.Array chunks) throws Error {
            int64 stored = 0;
            foreach (var element in chunks.get_elements()) {
                if (element.get_node_type() != Json.NodeType.OBJECT) {
                    continue;
                }
                var chunk = element.get_object();
                int64 seq = chunk.has_member("seq")
                    && chunk.get_member("seq").get_value_type() == typeof(int64)
                    ? chunk.get_int_member("seq") : 0;
                if (seq <= 0) {
                    continue;
                }
                string text = string_member(chunk, "text");
                if (text.length == 0) {
                    continue;
                }
                try {
                    yield db.insert_async<SessionRawLog>(new SessionRawLog() {
                        session_id = session_id,
                        seq = seq,
                        chunk = text
                    });
                    stored++;
                } catch (Error insert_error) {
                    // The unique (session_id, seq) pair is the idempotency
                    // guarantee: a row already there means the push is a
                    // redelivery — consume it. Any other failure is real.
                    var existing = yield db.query<SessionRawLog>()
                        .where(expr("session_id == $0 && seq == $1",
                            elem<int64?>(session_id), elem<int64?>(seq)))
                        .first_async();
                    if (existing == null) {
                        throw insert_error;
                    }
                }
            }
            return stored;
        }

        /**
         * Stores a pushed batch of round entries (`POST
         * /api/saddle/results`, plan 10 phase 2) as `session_commands`
         * rows — the mapping the deleted `GET /output` poller ran. Plan
         * 13 B1: the dedup key is the DAEMON-GLOBAL (saddle, entry id)
         * pair ({@link SaddleEntryDedup}), persisted in the database so
         * it survives web restarts and ignores session churn entirely —
         * a redelivered entry is consumed silently under whatever
         * session id the payload carries, and a push may only CREATE
         * rows for entries whose ids have not already landed on this
         * saddle. Entries without an id always store. Returns the
         * entries stored, oldest first, for {@link apply_entry_events}
         * — the deduped entry stream every projection keys on (B2).
         */
        public async Vector<Json.Object> store_pushed_entries(Saddle saddle, int64 session_id,
                Json.Array entries) throws Error {
            var stored = new Vector<Json.Object>();
            foreach (var element in entries.get_elements()) {
                if (element.get_node_type() != Json.NodeType.OBJECT) {
                    continue;
                }
                var entry = element.get_object();
                int64 entry_id = entry.has_member("id")
                    && entry.get_member("id").get_value_type() == typeof(int64)
                    ? entry.get_int_member("id") : 0;
                if (entry_id > 0 && !(yield claim_entry_id(saddle.id, entry_id))) {
                    // A redelivery under any session binding — consumed,
                    // never duplicated, never projected.
                    continue;
                }
                var command = new SessionCommand() {
                    session_id = session_id,
                    command_text = string_member(entry, "command"),
                    output_text = entry.has_member("output")
                        && entry.get_member("output").get_string() != null
                        ? entry.get_string_member("output") : null,
                    exit_code = entry.has_member("exit_code")
                        && entry.get_member("exit_code").get_value_type() == typeof(int64)
                        ? entry.get_int_member("exit_code") : 0,
                    entry_type = string_member(entry, "type").length > 0
                        ? string_member(entry, "type") : "command",
                    piped_input = optional_string_member(entry, "input"),
                    started_at = parse_iso(entry, "started_at") ?? parse_iso(entry, "timestamp"),
                    completed_at = parse_iso(entry, "completed_at"),
                    duration_ms = entry.has_member("duration_ms")
                        && entry.get_member("duration_ms").get_value_type() == typeof(int64)
                        ? entry.get_int_member("duration_ms") : 0
                };
                yield db.insert_async<SessionCommand>(command);
                stored.add(entry);
            }
            return stored;
        }

        /**
         * Claims `(saddle_id, entry_id)` for ingestion: true when the
         * id is fresh on this saddle, false when it already landed (an
         * outbox redelivery — consumed silently regardless of the
         * payload's session binding). The dedup row is inserted BEFORE
         * the entry stores, so a mid-pass failure can never re-project
         * an entry the daemon retries (exactly-once projections beat
         * at-least-once storage).
         */
        private async bool claim_entry_id(int64 saddle_id, int64 entry_id) throws Error {
            try {
                yield db.insert_async<SaddleEntryDedup>(new SaddleEntryDedup() {
                    saddle_id = saddle_id,
                    entry_id = entry_id
                });
                return true;
            } catch (Error insert_error) {
                var existing = yield db.query<SaddleEntryDedup>()
                    .where(expr("saddle_id == $0 && entry_id == $1",
                        elem<int64?>(saddle_id), elem<int64?>(entry_id)))
                    .first_async();
                if (existing == null) {
                    throw insert_error;
                }
                return false;
            }
        }

        /**
         * 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 (freeing the saddle and closing the session) —
         * 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 saddle; `suspended`
         * parks the same way — for a goal agent the resume is driven
         * by the goal's task-terminal events; and `working` flips the
         * task to working (never downgrading a pending review).
         *
         * 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(Saddle saddle, 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, it does not complete.
                        yield goal_service.submit_completion(goal_id,
                            summary != null && summary.length > 0 ? summary
                                : "The goal agent submitted its completion report");
                        yield park_if_current(saddle.id, task_id);
                        break;
                    }
                    yield tasks.complete_task(task_id, summary, null, "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 park_if_current(saddle.id, 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,
                                optional_string_member((!)block, "body") ?? question);
                        } else {
                            yield goal_service.raise_agent_block(goal_id, question,
                                block_string_or((!)block, "type", "block_type"),
                                string_array_member((!)block,
                                    (!)block.has_member("options") ? "options" : "block_options"),
                                optional_string_member((!)block, "body"));
                        }
                        yield tasks.set_status(task_id, STATUS_BLOCKED);
                        yield park_if_current(saddle.id, task_id);
                        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. The blocked task
                    // plus its open session is the pause record the
                    // affinity allocation resumes from once the block is
                    // answered.
                    yield park_if_current(saddle.id, task_id);
                    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.
                    yield tasks.suspend_task(task_id);
                    yield park_if_current(saddle.id, task_id);
                    break;
                case "working":
                    // A pending review keeps its banner: started_at (and
                    // the first-start push) only stamp for work that
                    // never allocated.
                    var current = yield tasks.set_status(task_id, STATUS_WORKING);
                    if (current != null && ((!)current).status != STATUS_REVIEW_REQUESTED
                            && ((!)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) 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 $(saddle.current_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.
         */
        private static Json.Node project_node(Project project, int64 task_id, bool is_reference = false) {
            var root_var = (Environment.get_variable("SADDLE_PROJECTS_ROOT") ?? "").strip();
            string projects_root = root_var.length > 0 ? root_var : "/workspace";
            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();
        }

        /** `command` minus a leading `verb` and its separating whitespace. */
        private static string strip_verb(string command, string verb) {
            var trimmed = command.strip();
            if (trimmed == verb) {
                return "";
            }
            return trimmed.has_prefix(verb + " ") ? trimmed.substring(verb.length + 1).strip() : trimmed;
        }

        /**
         * 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;
        }

        /** Strips one layer of matching double quotes, when present. */
        private static string strip_quotes(string text) {
            if (text.length >= 2 && text.has_prefix("\"") && text.has_suffix("\"")) {
                return text.substring(1, text.length - 2);
            }
            return text;
        }

        private static string string_member(Json.Object obj, string name) {
            return obj.has_member(name) && obj.get_member(name).get_string() != null
                ? obj.get_string_member(name) : "";
        }

        /** A nullable string member: null when absent or empty. */
        private static string? optional_string_member(Json.Object obj, string name) {
            string value = string_member(obj, name);
            return value.length > 0 ? value : null;
        }

        /**
         * A string-array member: native JSON arrays are read element-wise;
         * a JSON-encoded string is parsed; anything else is empty.
         */
        private static string[] string_array_member(Json.Object obj, string name) {
            if (!obj.has_member(name)) {
                return new string[0];
            }
            var member = obj.get_member(name);
            if (member.get_node_type() == Json.NodeType.ARRAY) {
                var values = new Vector<string>();
                foreach (var element in member.get_array().get_elements()) {
                    if (element.get_node_type() == Json.NodeType.VALUE
                            && element.get_string() != null) {
                        values.add(element.get_string());
                    }
                }
                return values.to_array();
            }
            string raw = string_member(obj, name);
            if (raw.length == 0) {
                return new string[0];
            }
            try {
                var parser = new Json.Parser();
                parser.load_from_data(raw);
                var root = parser.get_root();
                if (root != null && root.get_node_type() == Json.NodeType.ARRAY) {
                    var values = new Vector<string>();
                    foreach (var element in root.get_array().get_elements()) {
                        if (element.get_node_type() == Json.NodeType.VALUE
                                && element.get_string() != null) {
                            values.add(element.get_string());
                        }
                    }
                    return values.to_array();
                }
            } catch (Error e) {
                warning("[ClankerSaddle] Ignoring malformed options array: %s", e.message);
            }
            return new string[0];
        }

        private static DateTime? parse_iso(Json.Object obj, string name) {
            if (!obj.has_member(name)) {
                return null;
            }
            var value = obj.get_member(name);
            if (value.get_node_type() != Json.NodeType.VALUE || value.get_string() == null) {
                return null;
            }
            return new DateTime.from_iso8601(value.get_string(), new TimeZone.utc());
        }
    }
}