clanker-saddle/src/Services/TaskService.vala

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

namespace ClankerSaddle.Services {

    /** Valid task statuses (the `tasks.status` wire values). */
    public const string STATUS_DRAFT = "draft";
    public const string STATUS_QUEUED = "queued";
    public const string STATUS_ASSIGNED = "assigned";
    public const string STATUS_WORKING = "working";
    public const string STATUS_BLOCKED = "blocked";
    public const string STATUS_REVIEW_REQUESTED = "review_requested";
    public const string STATUS_SUSPENDED = "suspended";
    public const string STATUS_DONE = "done";
    public const string STATUS_FAILED = "failed";

    /** Valid {@link TaskBlock.block_type} values. */
    public const string BLOCK_QUESTION = "question";
    public const string BLOCK_RFC = "rfc";
    public const string BLOCK_REVIEW = "review";
    public const string BLOCK_PLAN = "plan";
    /**
     * The goal agent's completion report (plan 15 C1): raised on the
     * goal-run task by {@link GoalService.submit_completion} — the
     * goal's review kinds are plan and completion.
     */
    public const string BLOCK_COMPLETION = "completion";

    /**
     * The agent-celebration entry type (`saddle yeehaw <message>`): the
     * daemon pushes it as a round-results entry carrying the message in
     * the output field, so it stores like any other `session_commands`
     * row — nothing new is persisted for the badge or the card.
     */
    public const string ENTRY_YEEHAW = "yeehaw";

    /** The task page's yeehaw card cap (newest entries kept). */
    public const int YEEHAW_CARD_MAX = 3;

    /**
     * The stable verdict wordings (plan 11): they are stored as the
     * block answers AND ride the resuming allocation's
     * `resume_notice` verbatim, so agent and human always read the
     * same text.
     */
    public const string NOTICE_PLAN_APPROVED = "Plan approved. Proceed with implementation.";
    public const string NOTICE_PLAN_FEEDBACK = "Plan feedback received:";
    public const string NOTICE_REVIEW_APPROVED = "Review approved. Continue with the task.";
    public const string NOTICE_REVIEW_FEEDBACK = "Review feedback received:";

    /**
     * A task's computed presentation state (stored status + unanswered
     * blocks): `review requested` while a review block is unanswered
     * (even when the stored status is terminal — the post-mortem's
     * `done`-with-open-review case), `parked` for the suspended state
     * (a goal agent between events), else the stored status. Pure
     * presentation — the stored statuses and the dispatcher are
     * untouched.
     */
    public class TaskDisplay : Object {
        public string status { get; set; default = ""; }
        public bool terminal { get; set; }
        public bool failed { get; set; }

        public static TaskDisplay compute(TaskEntity task, bool review_pending) {
            var view = new TaskDisplay() {
                terminal = task.status == STATUS_DONE || task.status == STATUS_FAILED,
                failed = task.status == STATUS_FAILED
            };
            if (review_pending) {
                view.status = "review requested";
            } else if (task.status == STATUS_SUSPENDED) {
                view.status = "parked";
            } else {
                view.status = task.status;
            }
            return view;
        }
    }

    /**
     * The full task aggregate loaded by {@link TaskService.get_task}: the
     * task itself plus its comments, blocks, progress history and
     * session history.
     */
    public class TaskDetail : Object {
        public TaskEntity task { get; set; }
        public ImmutableLot<TaskComment> comments { get; set; }
        public ImmutableLot<TaskBlock> blocks { get; set; }
        public ImmutableLot<TaskProgress> progress { get; set; }
        public ImmutableLot<Session> sessions { get; set; }
    }

    /**
     * One agent celebration (`saddle yeehaw`) read back for display: the
     * message from the entry's output field and when it landed.
     */
    public class YeehawEntry : Object {
        public string message { get; set; default = ""; }
        public DateTime? at { get; set; }
    }

    /**
     * The task's celebration feed for the yeehaw card: the newest
     * {@link YEEHAW_CARD_MAX} entries plus the uncapped total (the
     * "+N more" hint derives from the two).
     */
    public class YeehawFeed : Object {
        public Vector<YeehawEntry> latest { get; set; default = new Vector<YeehawEntry>(); }
        public int64 total { get; set; }
    }

    /**
     * Task storage and lifecycle: creation, enqueue-after-aware
     * readiness, agent interaction primitives (comments, blocks,
     * progress, reviews) and completion. Saddle bookkeeping rows are
     * kept consistent on every terminal transition. The plan-12 task
     * graph is goal-shaped: no parents, no dependencies, no plans —
     * goals own their tasks.
     */
    public class TaskService : Object {

        private OrmSession db = inject<OrmSession>();
        private Connection connection = inject<Connection>();
        private StatePulse pulse = inject<StatePulse>();
        private PushService push = inject<PushService>();

        /**
         * The goal's task-terminal hook (plan 12): a worker task under
         * a goal records a supervision-narrative event — the parked
         * goal agent's resume machinery coalesces on these rows (see
         * {@link GoalService.resume_due}). Inline (not via
         * GoalService) because GoalService depends on this service.
         */
        private async void note_goal_task_event(TaskEntity task, string outcome_word) throws Error {
            if (task.goal_id <= 0 || task.kind != TASK_KIND_TASK) {
                return;
            }
            string lead = "";
            if (task.summary != null && ((!)task.summary).length > 0) {
                int cut = ((!)task.summary).index_of("\n");
                lead = cut >= 0 ? ((!)task.summary).substring(0, cut) : (!)task.summary;
            }
            string text = lead.length > 0
                ? @"Task #$(task.id) $(task.title) — $(outcome_word): $(lead)"
                : @"Task #$(task.id) $(task.title) — $(outcome_word)";
            yield db.insert_async<GoalEvent>(new GoalEvent() {
                goal_id = task.goal_id,
                kind = task.status == STATUS_DONE ? "task-done" : "task-failed",
                text = text,
                task_id = task.id,
                created_at = new DateTime.now_local()
            });
        }

        /**
         * Inserts a task (status 'draft' until its creator enqueues
         * it). `description`, `task_branch` and `branch_from` may be
         * null; `branch_from` prefills from the project's default
         * branch; priority clamps to 1–9 (5 is the default, 1 the
         * highest). The plan-12 shape: `kind` ("task"/"goal_run"),
         * `mode` (implement/investigate/review/test, orchestrate for
         * goal runs), `creator` ("human"/"goal" — enqueueing is
         * creator-only), `goal_id` (the owning goal) and
         * `enqueue_after_task_id` (human-created tasks only: the
         * auto-enqueue predecessor, cycle-checked by the caller).
         */
        public async TaskEntity create_task(int64 project_id, string title, string? description,
                int64 priority,
                string? task_branch = null, string? branch_from = null,
                string status = STATUS_DRAFT,
                string kind = TASK_KIND_TASK, string mode = "implement",
                string creator = CREATOR_HUMAN, int64 goal_id = 0,
                int64 enqueue_after_task_id = 0) throws Error {
            string resolved_branch_from = branch_from ?? "";
            if (resolved_branch_from.length == 0) {
                var project = yield db.query<Project>()
                    .where(expr("id == $0", elem<int64?>(project_id)))
                    .first_async();
                if (project != null) {
                    resolved_branch_from = ((!)project).default_branch;
                }
            }
            string? resolved_task_branch = task_branch != null && task_branch.length > 0 ? task_branch : null;
            if (goal_id > 0 && kind == TASK_KIND_TASK) {
                // Goal worker tasks exist only under a live goal.
                var goal = yield db.query<Goal>()
                    .where(expr("id == $0", elem<int64?>(goal_id)))
                    .first_async();
                if (goal == null) {
                    throw new Error(Quark.from_string("clanker-saddle"), 75,
                        @"No goal exists with id $(goal_id)");
                }
            }

            var task = new TaskEntity() {
                project_id = project_id,
                title = title,
                description = description != null && description.length > 0 ? description : null,
                priority = priority.clamp(1, 9),
                status = status,
                task_branch = resolved_task_branch,
                branch_from = resolved_branch_from.length > 0 ? resolved_branch_from : null,
                kind = kind,
                mode = mode,
                creator = creator,
                goal_id = goal_id,
                enqueue_after_task_id = enqueue_after_task_id,
                created_at = new DateTime.now_local()
            };
            yield db.insert_async<TaskEntity>(task);
            if (goal_id > 0 && kind == TASK_KIND_TASK) {
                // Goal worker tasks run on ephemeral auto-named
                // branches cut from the goal branch (the daemon builds
                // the worktrees; the stored name feeds compare links).
                var goal = yield db.query<Goal>()
                    .where(expr("id == $0", elem<int64?>(goal_id)))
                    .first_async();
                if (goal != null) {
                    task.task_branch = @"task/$(goal_id)-$(task.id)";
                    task.branch_from = ((!)goal).branch_name;
                    yield db.update_async<TaskEntity>(task);
                }
            }
            yield pulse.bump(task.id, "task created");
            return task;
        }

        /**
         * Loads the full aggregate for `id`: the task, its comments,
         * blocks, progress and sessions. Returns null when the task
         * does not exist.
         */
        public async TaskDetail? get_task(int64 id) throws Error {
            var task = yield db.query<TaskEntity>()
                .where(expr("id == $0", elem<int64?>(id)))
                .first_async();
            if (task == null) {
                return null;
            }
            return new TaskDetail() {
                task = (!)task,
                comments = yield list_comments(id),
                blocks = yield list_blocks(id),
                progress = yield list_progress(id),
                sessions = yield db.query<Session>()
                    .where(expr("task_id == $0", elem<int64?>(id)))
                    .order_by(expr("id"))
                    .materialise_async()
            };
        }

        /**
         * Lists tasks, optionally filtered by status and/or project, ordered
         * by priority (1 first) then creation time.
         */
        /**
         * Computed display states for a batch of tasks (all lists share
         * this), keyed by task id string. Unanswered review blocks load
         * once, not per task.
         */
        public async Dictionary<string, TaskDisplay> display_map_for(
                Lot<TaskEntity> tasks) throws Error {
            var wanted = new Invercargill.DataStructures.HashSet<string>();
            foreach (var task in tasks) {
                wanted.add(task.id.to_string());
            }

            var review_pending = new Invercargill.DataStructures.HashSet<string>();
            foreach (var block in yield db.query<TaskBlock>()
                    .where(expr("type == $0", elem(BLOCK_REVIEW)))
                    .materialise_async()) {
                if (block.answered_at == null && wanted.has(block.task_id.to_string())) {
                    review_pending.add(block.task_id.to_string());
                }
            }

            var map = new Dictionary<string, TaskDisplay>();
            foreach (var task in tasks) {
                map.set(task.id.to_string(),
                    TaskDisplay.compute(task, review_pending.has(task.id.to_string())));
            }
            return map;
        }

        /**
         * The display state of a single task — the detail page's path,
         * computed from its already-loaded aggregate (no extra queries).
         */
        public static TaskDisplay display_of_detail(TaskDetail detail) {
            bool review_pending = false;
            foreach (var block in detail.blocks) {
                if (block.block_type == BLOCK_REVIEW && block.answered_at == null) {
                    review_pending = true;
                }
            }
            return TaskDisplay.compute(detail.task, review_pending);
        }

        public async ImmutableLot<TaskEntity> list_tasks(string? status, int64? project_id) throws Error {
            var tasks = yield db.query<TaskEntity>().materialise_async();
            var rows = new Vector<TaskEntity>();
            foreach (var task in tasks) {
                if (status != null && status.length > 0 && task.status != status) {
                    continue;
                }
                if (project_id != null && task.project_id != project_id) {
                    continue;
                }
                rows.add(task);
            }
            var sorted = rows.sort((a, b) => {
                if (a.priority != b.priority) {
                    return (int)(a.priority - b.priority);
                }
                int64 a_created = a.created_at != null ? ((!)a.created_at).to_unix() : 0;
                int64 b_created = b.created_at != null ? ((!)b.created_at).to_unix() : 0;
                return (int)(a_created - b_created);
            });
            return sorted.to_immutable_buffer();
        }

        /**
         * Saves the human-editable fields of the task identified by `id`
         * under the draft-lifecycle edit matrix: in DRAFT all fields
         * except status are editable (title, description/mandate,
         * priority, branch fields, the plan-12 work mode and —
         * human-created tasks only — the auto-enqueue predecessor,
         * cycle-validated); in non-draft non-terminal
         * states only priority applies; in terminal states nothing
         * does. Fields not editable in the current state are silently
         * left unchanged (null arguments always mean "leave
         * unchanged"; an empty branch field clears it; an
         * `enqueue_after_task_id` of -1 CLEARS the predecessor).
         * Returns null when the task does not exist. The pulse only
         * fires when a value actually moved, so periodic status
         * refreshes do not spam live pages.
         */
        public async TaskEntity? edit_task(int64 id, string? title, string? description,
                int64? priority, string? task_branch, string? branch_from,
                string? mode = null, int64? enqueue_after_task_id = null) throws Error {
            var task = yield db.query<TaskEntity>()
                .where(expr("id == $0", elem<int64?>(id)))
                .first_async();
            if (task == null) {
                return null;
            }
            bool changed = false;
            bool draft = ((!)task).status == STATUS_DRAFT;
            if (draft) {
                if (title != null && title.length > 0 && ((!)task).title != title) {
                    ((!)task).title = title;
                    changed = true;
                }
                if (description != null) {
                    string? next_description = description.length > 0 ? description : null;
                    if (((!)task).description != next_description) {
                        ((!)task).description = next_description;
                        changed = true;
                    }
                }
                if (task_branch != null) {
                    string? next_branch = task_branch.length > 0 ? task_branch : null;
                    if (((!)task).task_branch != next_branch) {
                        ((!)task).task_branch = next_branch;
                        changed = true;
                    }
                }
                if (branch_from != null) {
                    string? next_from = branch_from.length > 0 ? branch_from : null;
                    if (((!)task).branch_from != next_from) {
                        ((!)task).branch_from = next_from;
                        changed = true;
                    }
                }
                if (mode != null && mode.length > 0) {
                    bool mode_valid = false;
                    foreach (string candidate in TASK_MODES) {
                        if (mode == candidate) {
                            mode_valid = true;
                            break;
                        }
                    }
                    if (!mode_valid) {
                        throw new Error(Quark.from_string("clanker-saddle"), 79,
                            @"Unknown mode '$(mode)' — valid modes: " + string.joinv(", ", TASK_MODES));
                    }
                    if (((!)task).mode != mode) {
                        ((!)task).mode = mode;
                        changed = true;
                    }
                }
                // Human-created tasks only: the auto-enqueue
                // predecessor is settable/clearable while draft, and
                // setting it validates the chain (no cycles).
                if (enqueue_after_task_id != null
                        && ((!)task).creator == CREATOR_HUMAN
                        && ((!)task).kind == TASK_KIND_TASK) {
                    int64 next = (!)enqueue_after_task_id;
                    if (next < 0) {
                        // The -1 sentinel clears the predecessor.
                        next = 0;
                    }
                    if (next > 0) {
                        yield validate_enqueue_after_cycle(id, next);
                    }
                    if (((!)task).enqueue_after_task_id != next) {
                        ((!)task).enqueue_after_task_id = next;
                        changed = true;
                    }
                }
            }
            // Priority stays editable in every non-terminal state.
            if ((((!)task).status != STATUS_DONE && ((!)task).status != STATUS_FAILED)
                    && priority != null) {
                int64 clamped = ((!)priority).clamp(1, 9);
                if (((!)task).priority != clamped) {
                    ((!)task).priority = clamped;
                    changed = true;
                }
            }
            yield db.update_async<TaskEntity>(task);
            if (changed) {
                yield pulse.bump(id, "task saved");
            }
            return task;
        }

        /**
         * The internal status-transition path (daemon pushes, park and
         * result machinery) — never reachable from a human form.
         * Stamps `completed_at` on done. Returns null when the task
         * does not exist.
         */
        public async TaskEntity? set_status(int64 id, string status) throws Error {
            var task = yield db.query<TaskEntity>()
                .where(expr("id == $0", elem<int64?>(id)))
                .first_async();
            if (task == null) {
                return null;
            }
            if (((!)task).status != status) {
                ((!)task).status = status;
                if (status == STATUS_DONE && ((!)task).completed_at == null) {
                    ((!)task).completed_at = new DateTime.now_local();
                }
                yield db.update_async<TaskEntity>(task);
                yield pulse.bump(id, @"task $(status)");
            }
            return task;
        }

        /** Every task under `goal_id` (both kinds), oldest first. */
        public async ImmutableLot<TaskEntity> list_tasks_for_goal(int64 goal_id) throws Error {
            return yield db.query<TaskEntity>()
                .where(expr("goal_id == $0", elem<int64?>(goal_id)))
                .order_by(expr("id"))
                .materialise_async();
        }

        /**
         * The Enqueue action (creator-only, plan 12): moves a draft
         * task to the queue. Humans enqueue theirs, goal agents theirs
         * — an actor that is not the task's creator is refused (the
         * goal agent sequences its own tasks). A pending
         * auto-enqueue predecessor does not refuse the enqueue — the
         * dispatcher's readiness check holds dispatch back until it
         * finishes. The internal auto-enqueue-after pass calls this as
         * the dependent's creator (the creator's pre-authorisation).
         */
        public async void enqueue_task(int64 id, string actor = CREATOR_HUMAN) throws Error {
            var task = yield db.query<TaskEntity>()
                .where(expr("id == $0", elem<int64?>(id)))
                .first_async();
            if (task == null) {
                throw new Error(Quark.from_string("clanker-saddle"), 50,
                    @"No task exists with id $(id)");
            }
            if (((!)task).status != STATUS_DRAFT) {
                throw new Error(Quark.from_string("clanker-saddle"), 51,
                    "Only draft tasks can be enqueued");
            }
            if (((!)task).creator != actor) {
                throw new Error(Quark.from_string("clanker-saddle"), 76,
                    @"Only the task's creator may enqueue it (task created by '$(((!)task).creator)', actor '$(actor)')");
            }

            ((!)task).status = STATUS_QUEUED;
            yield db.update_async<TaskEntity>((!)task);
            yield pulse.bump(id, "task enqueued");
        }

        /**
         * The cycle check for `enqueue_after_task_id` (human-created
         * tasks only, settable/clearable while the dependent is
         * draft): following the predecessor chain from
         * `predecessor_id` must never reach `task_id` — a cycle would
         * strand every task on it in draft forever.
         */
        public async void validate_enqueue_after_cycle(int64 task_id,
                int64 predecessor_id) throws Error {
            if (predecessor_id <= 0) {
                return;
            }
            if (predecessor_id == task_id) {
                throw new Error(Quark.from_string("clanker-saddle"), 77,
                    "A task cannot enqueue after itself");
            }
            var seen = new Invercargill.DataStructures.HashSet<string>();
            int64 current = predecessor_id;
            while (current > 0 && seen.add(current.to_string())) {
                if (current == task_id) {
                    throw new Error(Quark.from_string("clanker-saddle"), 78,
                        @"enqueue-after would form a cycle through task $(predecessor_id)");
                }
                var row = yield db.query<TaskEntity>()
                    .where(expr("id == $0", elem<int64?>(current)))
                    .first_async();
                if (row == null) {
                    return;
                }
                current = ((!)row).enqueue_after_task_id;
            }
            if (!seen.has(task_id.to_string())) {
                return;
            }
            throw new Error(Quark.from_string("clanker-saddle"), 78,
                @"enqueue-after would form a cycle through task $(predecessor_id)");
        }

        /**
         * The auto-enqueue-after pass (plan 12): when a predecessor
         * reaches `done`, its draft dependents enqueue as their
         * creator (the pre-authorised enqueue). A failed or cancelled
         * predecessor leaves them in draft — the creator decides.
         */
        private async void enqueue_dependents_of(TaskEntity predecessor) throws Error {
            if (predecessor.status != STATUS_DONE) {
                return;
            }
            foreach (var dependent in yield db.query<TaskEntity>()
                    .where(expr("enqueue_after_task_id == $0 && status == $1",
                        elem<int64?>(predecessor.id), elem(STATUS_DRAFT)))
                    .materialise_async()) {
                try {
                    yield enqueue_task(dependent.id, dependent.creator);
                } catch (Error e) {
                    warning("[ClankerSaddle] Auto-enqueue of task %lld after %lld failed: %s",
                        dependent.id, predecessor.id, e.message);
                }
            }
        }

        /**
         * The delete the goal cascade uses on its own draft tasks:
         * every attached row dies with the task and no dependents
         * guard applies (the whole goal is going away together).
         */
        public async void delete_task_quietly(int64 id) throws Error {
            var task = yield db.query<TaskEntity>()
                .where(expr("id == $0", elem<int64?>(id)))
                .first_async();
            if (task == null) {
                return;
            }
            foreach (var link in yield db.query<TaskRule>()
                    .where(expr("task_id == $0", elem<int64?>(id)))
                    .materialise_async()) {
                yield db.delete_async<TaskRule>(link);
            }
            foreach (var link in yield db.query<TaskCapability>()
                    .where(expr("task_id == $0", elem<int64?>(id)))
                    .materialise_async()) {
                yield db.delete_async<TaskCapability>(link);
            }
            foreach (var row in yield db.query<TaskProgress>()
                    .where(expr("task_id == $0", elem<int64?>(id)))
                    .materialise_async()) {
                yield db.delete_async<TaskProgress>(row);
            }
            foreach (var row in yield db.query<TaskComment>()
                    .where(expr("task_id == $0", elem<int64?>(id)))
                    .materialise_async()) {
                yield db.delete_async<TaskComment>(row);
            }
            foreach (var row in yield db.query<TaskBlock>()
                    .where(expr("task_id == $0", elem<int64?>(id)))
                    .materialise_async()) {
                yield db.delete_async<TaskBlock>(row);
            }
            yield clear_enqueue_after_references(id);
            yield db.delete_async<TaskEntity>((!)task);
            yield pulse.bump(id, "task deleted");
        }

        /**
         * The human-only Delete action: removes a DRAFT task. Any
         * auto-enqueue-after predecessor references pointing at it are
         * cleared first (a dangling predecessor would strand its
         * dependents in not-ready forever). Attached rows (comments,
         * blocks, progress, rules, capabilities) die with the task.
         */
        public async void delete_task(int64 id) throws Error {
            var task = yield db.query<TaskEntity>()
                .where(expr("id == $0", elem<int64?>(id)))
                .first_async();
            if (task == null) {
                return;
            }
            if (((!)task).status != STATUS_DRAFT) {
                throw new Error(Quark.from_string("clanker-saddle"), 53,
                    "Only draft tasks can be deleted (cancel or retry handles the rest)");
            }

            foreach (var link in yield db.query<TaskRule>()
                    .where(expr("task_id == $0", elem<int64?>(id)))
                    .materialise_async()) {
                yield db.delete_async<TaskRule>(link);
            }
            foreach (var link in yield db.query<TaskCapability>()
                    .where(expr("task_id == $0", elem<int64?>(id)))
                    .materialise_async()) {
                yield db.delete_async<TaskCapability>(link);
            }
            foreach (var row in yield db.query<TaskProgress>()
                    .where(expr("task_id == $0", elem<int64?>(id)))
                    .materialise_async()) {
                yield db.delete_async<TaskProgress>(row);
            }
            foreach (var row in yield db.query<TaskComment>()
                    .where(expr("task_id == $0", elem<int64?>(id)))
                    .materialise_async()) {
                yield db.delete_async<TaskComment>(row);
            }
            foreach (var row in yield db.query<TaskBlock>()
                    .where(expr("task_id == $0", elem<int64?>(id)))
                    .materialise_async()) {
                yield db.delete_async<TaskBlock>(row);
            }
            yield clear_enqueue_after_references(id);
            yield db.delete_async<TaskEntity>((!)task);
            yield pulse.bump(id, "task deleted");
        }

        /**
         * Clears `enqueue_after_task_id` references pointing at
         * `id`: a deleted predecessor must not strand its dependents
         * (their readiness check would wait forever on a task that no
         * longer exists).
         */
        private async void clear_enqueue_after_references(int64 id) throws Error {
            foreach (var dependent in yield db.query<TaskEntity>()
                    .where(expr("enqueue_after_task_id == $0", elem<int64?>(id)))
                    .materialise_async()) {
                dependent.enqueue_after_task_id = 0;
                yield db.update_async<TaskEntity>(dependent);
            }
        }

        /**
         * Appends a progress line (`saddle progress-update`) to the task's
         * history and returns the new row.
         */
        public async TaskProgress add_progress(int64 task_id, string summary) throws Error {
            var row = new TaskProgress() {
                task_id = task_id,
                summary = summary,
                created_at = new DateTime.now_local()
            };
            yield db.insert_async<TaskProgress>(row);
            yield pulse.bump(task_id, "progress");
            return row;
        }

        /** The task's progress history, newest first. */
        public async ImmutableLot<TaskProgress> list_progress(int64 task_id) throws Error {
            return yield db.query<TaskProgress>()
                .where(expr("task_id == $0", elem<int64?>(task_id)))
                .order_by_desc(expr("id"))
                .materialise_async();
        }

        /**
         * True when any of the task's sessions carries a `yeehaw` entry —
         * the badge's presence check, one cheap EXISTS over the existing
         * `session_commands` rows (nothing new is stored for the badge).
         */
        public async bool has_yeehaw(int64 task_id) throws Error {
            var command = connection.create_command(
                "SELECT EXISTS(SELECT 1 FROM session_commands sc"
                + " JOIN sessions s ON s.id = sc.session_id"
                + " WHERE s.task_id = :task AND sc.type = :kind) AS hit");
            command.with_parameter<Invercargill.Element>("task", elem<int64?>(task_id));
            command.with_parameter<Invercargill.Element>("kind", elem(ENTRY_YEEHAW));
            var hit = yield command.execute_scalar_async();
            int64? value = 0;
            return hit != null && hit.try_get_as<int64?>(out value) && ((!)value) != 0;
        }

        /**
         * Every task id whose sessions carry a `yeehaw` entry (stringly
         * keyed) — the batch presence check the task list and the
         * dashboard's task references badge from, one DISTINCT query
         * instead of one EXISTS per row.
         */
        public async Invercargill.DataStructures.HashSet<string> yeehaw_task_ids() throws Error {
            var ids = new Invercargill.DataStructures.HashSet<string>();
            var command = connection.create_command(
                "SELECT DISTINCT s.task_id AS task_id FROM sessions s"
                + " JOIN session_commands sc ON sc.session_id = s.id"
                + " WHERE sc.type = :kind");
            command.with_parameter<Invercargill.Element>("kind", elem(ENTRY_YEEHAW));
            foreach (var row in yield command.execute_query_async()) {
                var element = row.get("task_id");
                int64? value = 0;
                if (element != null && element.try_get_as<int64?>(out value) && value != null) {
                    ids.add(((!)value).to_string());
                }
            }
            return ids;
        }

        /**
         * The task's yeehaw card data: the newest
         * {@link YEEHAW_CARD_MAX} celebrations (message + when, newest
         * first) plus the uncapped total. Reads the existing
         * `session_commands` rows through the sessions join — the card
         * stores nothing of its own.
         */
        public async YeehawFeed yeehaw_feed(int64 task_id) throws Error {
            var total_command = connection.create_command(
                "SELECT COUNT(*) AS total FROM session_commands sc"
                + " JOIN sessions s ON s.id = sc.session_id"
                + " WHERE s.task_id = :task AND sc.type = :kind");
            total_command.with_parameter<Invercargill.Element>("task", elem<int64?>(task_id));
            total_command.with_parameter<Invercargill.Element>("kind", elem(ENTRY_YEEHAW));
            var total = yield total_command.execute_scalar_async();
            int64? total_value = 0;
            if (total != null) {
                total.try_get_as<int64?>(out total_value);
            }

            var feed = new YeehawFeed() { total = total_value };
            var rows_command = connection.create_command(
                "SELECT sc.output_text AS message, sc.command_text AS command,"
                + " sc.started_at AS started, sc.completed_at AS completed"
                + " FROM session_commands sc JOIN sessions s ON s.id = sc.session_id"
                + " WHERE s.task_id = :task AND sc.type = :kind"
                + " ORDER BY sc.id DESC LIMIT :cap");
            rows_command.with_parameter<Invercargill.Element>("task", elem<int64?>(task_id));
            rows_command.with_parameter<Invercargill.Element>("kind", elem(ENTRY_YEEHAW));
            rows_command.with_parameter<Invercargill.Element>("cap", elem<int64?>(YEEHAW_CARD_MAX));
            foreach (var row in yield rows_command.execute_query_async()) {
                string message = string_of(row.get("message"));
                if (message.length == 0) {
                    // The message rides the output field; the command
                    // line is the fallback for empty-output pushes.
                    message = string_of(row.get("command"));
                }
                DateTime? at = date_time_of(row.get("started"));
                if (at == null) {
                    at = date_time_of(row.get("completed"));
                }
                feed.latest.add(new YeehawEntry() { message = message, at = at });
            }
            return feed;
        }

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

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

        /** Appends a comment; `author` is 'agent' or 'user:<name>'. */
        public async TaskComment add_comment(int64 task_id, string author, string body,
                string? quote = null) throws Error {
            var comment = new TaskComment() {
                task_id = task_id,
                author = author,
                body = body,
                quote = quote != null && quote.strip().length > 0 ? quote.strip() : null,
                created_at = new DateTime.now_local()
            };
            yield db.insert_async<TaskComment>(comment);
            yield pulse.bump(task_id, "comment");
            return comment;
        }

        /**
         * Appends a terminal entry to the task's open session so
         * conversation-level events (blocks, plans, progress, reviews,
         * human comments) appear in the streamed log. A no-op when the
         * task has no open session.
         */
        public async void log_session_entry(int64 task_id, string entry_type,
                string command, string? output) throws Error {
            var session = yield open_session_for_task(task_id);
            if (session == null) {
                return;
            }
            var entry = new SessionCommand() {
                session_id = ((!)session).id,
                command_text = command,
                output_text = output,
                entry_type = entry_type,
                started_at = new DateTime.now_local()
            };
            yield db.insert_async<SessionCommand>(entry);
        }

        /**
         * Raises a blocking prompt on a task — `question` (optionally with
         * multiple-choice `options` and a markdown `body`), `rfc` or
         * `review` — and flips it to 'blocked' (or 'review_requested')
         * while it waits for a human answer. Returns the new block.
         */
        public async TaskBlock add_block(int64 task_id, string question, string block_type = BLOCK_QUESTION,
                string[]? options = null, string? body = null,
                bool quiet = false) throws Error {
            string options_json = null;
            if (options != null && options.length > 0) {
                var array = new Json.Array();
                foreach (var option in options) {
                    array.add_string_element(option);
                }
                options_json = Json.to_string(new Json.Node(Json.NodeType.ARRAY).init_array(array), false);
            }

            var block = new TaskBlock() {
                task_id = task_id,
                question = question,
                block_type = block_type.length > 0 ? block_type : BLOCK_QUESTION,
                options = options_json,
                body = body != null && body.length > 0 ? body : null,
                created_at = new DateTime.now_local()
            };
            yield db.insert_async<TaskBlock>(block);

            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)) {
                ((!)task).status = block.block_type == BLOCK_REVIEW
                    ? STATUS_REVIEW_REQUESTED : STATUS_BLOCKED;
                yield db.update_async<TaskEntity>((!)task);
            }
            yield pulse.bump(task_id, "block raised");
            // Web Push (plan 11 task E): every raised block notifies all
            // push-subscribed users globally — the goal agent's mirrored
            // blocks stay quiet (the goal-level block owns the
            // notification with the goal's title and URL).
            if (!quiet && task != null) {
                push.notify_global(PushService.block_notification(
                    ((!)task).title, task_id, block.block_type, block.question));
            }
            return block;
        }

        /**
         * Records a review request (`saddle review`): a review-type block
         * carrying the summary markdown, with the task flipped to
         * 'review_requested' while it waits for the human verdict.
         */
        public async TaskBlock request_review(int64 task_id, string summary) throws Error {
            return yield add_block(task_id, "Review requested", BLOCK_REVIEW, null,
                summary.length > 0 ? summary : null);
        }

        /**
         * Records a human's answer to a block. With an open session the
         * task stays 'blocked'/'review_requested' — the dispatcher resumes
         * the paused workspace and hands the answer over; without one the
         * task flips straight back to 'working'.
         */
        public async void answer_block(int64 block_id, string answer,
                string? answered_by = null) throws Error {
            var block = yield db.query<TaskBlock>()
                .where(expr("id == $0", elem<int64?>(block_id)))
                .first_async();
            if (block == null) {
                throw new Error(Quark.from_string("clanker-saddle"), 5,
                    @"No block exists with id $(block_id)");
            }

            ((!)block).answer = answer;
            ((!)block).answered_at = new DateTime.now_local();
            ((!)block).answered_by = answered_by;
            yield db.update_async<TaskBlock>((!)block);

            var task = yield db.query<TaskEntity>()
                .where(expr("id == $0", elem<int64?>(((!)block).task_id)))
                .first_async();
            if (task != null && (((!)task).status == STATUS_BLOCKED
                    || ((!)task).status == STATUS_REVIEW_REQUESTED)) {
                var session = yield open_session_for_task(((!)block).task_id);
                if (session == null) {
                    ((!)task).status = STATUS_WORKING;
                    yield db.update_async<TaskEntity>((!)task);
                }
            }
            yield pulse.bump(((!)block).task_id, "block answered");
        }

        /**
         * The git-host compare URL for the task's branches —
         * `{git_url}/compare/{branch_from}...{task_branch}` — or "" when
         * the task has no working branch.
         */
        public async string diff_url_for_task(int64 task_id) throws Error {
            var task = yield db.query<TaskEntity>()
                .where(expr("id == $0", elem<int64?>(task_id)))
                .first_async();
            if (task == null || ((!)task).task_branch == null
                    || ((!)task).task_branch.length == 0) {
                return "";
            }
            var project = yield db.query<Project>()
                .where(expr("id == $0", elem<int64?>(((!)task).project_id)))
                .first_async();
            if (project == null) {
                return "";
            }
            string base_branch = ((!)task).branch_from ?? ((!)project).default_branch;
            return @"$(https_git_url(((!)project).git_url))/compare/$(base_branch)...$(((!)task).task_branch)";
        }

        /**
         * Translates a project's git remote into its browsable HTTPS form
         * so compare links open in the web UI instead of an SSH checkout:
         * `ssh://git@host:port/owner/repo.git` and `git@host:owner/repo(.git)`
         * both become `https://host/owner/repo` (the port is dropped —
         * SSH-only ports like fabrica.unitatem.net:222 have no HTTPS
         * meaning). Anything already http(s) (or unrecognised) is kept
         * as-is, minus a trailing `.git`.
         */
        public static string https_git_url(string git_url) {
            string url = git_url.strip();
            if (url.has_prefix("http://") || url.has_prefix("https://")) {
                return url.has_suffix(".git") ? url.substring(0, url.length - 4) : url;
            }

            // ssh://git@host[:port]/owner/repo.git
            if (url.has_prefix("ssh://")) {
                string rest = url.substring("ssh://".length);
                if (rest.has_prefix("git@")) {
                    rest = rest.substring("git@".length);
                }
                int slash = rest.index_of("/");
                if (slash < 0) {
                    return url;
                }
                string host_port = rest.substring(0, slash);
                string path = rest.substring(slash);
                if (host_port.contains(":")) {
                    // SSH-only ports (fabrica.unitatem.net:222) carry no
                    // HTTPS meaning — the browse host is plain.
                    host_port = host_port.split(":")[0];
                }
                if (path.has_suffix(".git")) {
                    path = path.substring(0, path.length - 4);
                }
                return @"https://$(host_port)$(path)";
            }

            // git@host:owner/repo(.git) — scp-style (the colon sits before
            // any path slash)
            int at = url.index_of("@");
            int colon = url.index_of(":");
            int slash = url.index_of("/");
            if (at >= 0 && colon > at && (slash < 0 || colon < slash)) {
                string host = url.substring(at + 1, colon - at - 1);
                string path = url.substring(colon + 1);
                if (path.has_suffix(".git")) {
                    path = path.substring(0, path.length - 4);
                }
                return @"https://$(host)/$(path)";
            }
            return url;
        }

        /**
         * Task ids currently `assigned`/`working` whose open session sits
         * on an offline or fenced saddle — the queue marks them
         * "saddle unreachable — requeueing" while recovery runs.
         */
        public async Invercargill.DataStructures.HashSet<string> unreachable_task_ids() throws Error {
            var ids = new Invercargill.DataStructures.HashSet<string>();
            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 open_session_for_task(task.id);
                    if (session == null) {
                        continue;
                    }
                    var saddle = yield db.query<Saddle>()
                        .where(expr("id == $0", elem<int64?>(((!)session).saddle_id)))
                        .first_async();
                    if (saddle != null && (((!)saddle).status == "offline"
                            || ((!)saddle).status == "recovering")) {
                        ids.add(task.id.to_string());
                    }
                }
            }
            return ids;
        }

        /**
         * The resume_notice wording for an answered block (the plan-11
         * stable verdicts): review and plan answers are already stored
         * in the stable wording and pass through verbatim; question
         * replies are formatted the way the daemon used to format
         * delivered answers ("Human selected: …" for option matches,
         * "Human answered (other): …" otherwise); RFC/plain answers
         * carry the "Human answer: …" prefix.
         */
        public static string verdict_notice_for(TaskBlock block) {
            string answer = block.answer ?? "";
            if (block.block_type == BLOCK_REVIEW || block.block_type == BLOCK_PLAN) {
                return answer;
            }
            if (block.block_type == BLOCK_QUESTION) {
                foreach (var option in parse_block_options(block.options)) {
                    if (option == answer) {
                        return "Human selected: " + answer;
                    }
                }
                return "Human answered (other): " + answer;
            }
            return "Human answer: " + answer;
        }

        /** The stored options JSON (a string array) back into a Vala array. */
        public static string[] parse_block_options(string? options_json) {
            if (options_json == null || ((!)options_json).length == 0) {
                return new string[0];
            }
            try {
                var parser = new Json.Parser();
                parser.load_from_data((!)options_json);
                var root = parser.get_root();
                if (root == null || root.get_node_type() != Json.NodeType.ARRAY) {
                    return new string[0];
                }
                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) {
                return new string[0];
            }
        }

        /**
         * Marks a task done: records `summary` and `working_branch`, stamps
         * `completed_at`, closes any open session (outcome `outcome`) and
         * frees the saddle that was running it. Terminal transitions
         * project ONCE (plan 13 B2): a task already done or failed keeps
         * its first outcome — a redelivered push or a second terminal
         * entry with a fresh id re-runs no bookkeeping, no comment, no
         * event (retry_task is the one path that un-terminals).
         */
        public async void complete_task(int64 task_id, string? summary,
                string? working_branch, string outcome) throws Error {
            var task = yield db.query<TaskEntity>()
                .where(expr("id == $0", elem<int64?>(task_id)))
                .first_async();
            if (task == null) {
                throw new Error(Quark.from_string("clanker-saddle"), 8,
                    @"No task exists with id $(task_id)");
            }
            if (((!)task).status == STATUS_DONE || ((!)task).status == STATUS_FAILED) {
                return;
            }

            ((!)task).status = STATUS_DONE;
            ((!)task).completed_at = new DateTime.now_local();
            if (summary != null && summary.length > 0) {
                ((!)task).summary = summary;
            }
            if (working_branch != null && working_branch.length > 0) {
                ((!)task).working_branch = working_branch;
            }
            yield db.update_async<TaskEntity>((!)task);

            yield close_open_sessions(task_id, outcome);
            yield free_saddles_running(task_id);
            yield pulse.bump(task_id, "task completed");
            // Web Push (plan 11 task E): completion notifies the users
            // holding a "Notify me" pref for this task.
            push.notify_opted_in(task_id, PushService.lifecycle_notification(
                ((!)task).title, task_id, "completed", summary));
            // Plan 12: the goal's narrative hears the terminal event
            // (a parked goal agent resumes on it), and a done
            // predecessor auto-enqueues its dependents.
            yield note_goal_task_event(((!)task), "completed");
            yield enqueue_dependents_of(((!)task));
        }

        /**
         * Cancels a task: status 'failed' with a cancellation summary. Any
         * open session is closed with outcome 'stopped' and its saddle
         * freed. Draft tasks are refused — they are deleted, not
         * cancelled (nothing ever ran). Idempotent on terminal tasks
         * (plan 13 B2): the first outcome stands.
         */
        public async void cancel_task(int64 task_id) throws Error {
            var task = yield db.query<TaskEntity>()
                .where(expr("id == $0", elem<int64?>(task_id)))
                .first_async();
            if (task == null) {
                throw new Error(Quark.from_string("clanker-saddle"), 9,
                    @"No task exists with id $(task_id)");
            }
            if (((!)task).status == STATUS_DONE || ((!)task).status == STATUS_FAILED) {
                return;
            }
            if (((!)task).status == STATUS_DRAFT) {
                throw new Error(Quark.from_string("clanker-saddle"), 55,
                    "Draft tasks are deleted, not cancelled");
            }

            ((!)task).status = STATUS_FAILED;
            ((!)task).completed_at = new DateTime.now_local();
            if (((!)task).summary == null || ((!)task).summary.length == 0) {
                ((!)task).summary = "Cancelled by user";
            }
            yield db.update_async<TaskEntity>((!)task);

            yield close_open_sessions(task_id, "stopped");
            yield free_saddles_running(task_id);
            yield pulse.bump(task_id, "task cancelled");
            yield note_goal_task_event(((!)task), "cancelled");
        }

        /**
         * Records a saddle-reported failure: status 'failed' with the
         * daemon's message as summary, the explanation stored again as a
         * agent comment (so the thread carries it), the open session
         * closed with outcome 'failed' and its saddle freed. Terminal
         * transitions project ONCE (plan 13 B2): the live failure's
         * honest burst — a dozen-plus distinct `saddle failure` entry
         * ids in one session — still fails the task exactly once, with
         * one comment and one goal event.
         */
        public async void fail_task(int64 task_id, string? summary) throws Error {
            var task = yield db.query<TaskEntity>()
                .where(expr("id == $0", elem<int64?>(task_id)))
                .first_async();
            if (task == null) {
                throw new Error(Quark.from_string("clanker-saddle"), 25,
                    @"No task exists with id $(task_id)");
            }
            if (((!)task).status == STATUS_DONE || ((!)task).status == STATUS_FAILED) {
                return;
            }

            ((!)task).status = STATUS_FAILED;
            ((!)task).completed_at = new DateTime.now_local();
            if (summary != null && summary.length > 0) {
                ((!)task).summary = summary;
                yield add_comment(task_id, "agent", summary);
            } else if (((!)task).summary == null || ((!)task).summary.length == 0) {
                ((!)task).summary = "Saddle reported a failure";
            }
            yield db.update_async<TaskEntity>((!)task);

            yield close_open_sessions(task_id, "failed");
            yield free_saddles_running(task_id);
            yield pulse.bump(task_id, "task failed");
            // Web Push (plan 11 task E): an agent-reported failure
            // notifies the opted-in users (a human cancellation does
            // not — the cancellator already knows).
            push.notify_opted_in(task_id, PushService.lifecycle_notification(
                ((!)task).title, task_id, "failed", summary));
            yield note_goal_task_event(((!)task), "failed");
        }

        /**
         * Parks a suspended task (`saddle suspend`): stores the status
         * and stamps `suspended_at` once per parking window, so the
         * resume trigger only sees goal events that landed after the
         * suspension (a parked goal agent between supervision rounds).
         */
        public async void suspend_task(int64 task_id) throws Error {
            var task = yield db.query<TaskEntity>()
                .where(expr("id == $0", elem<int64?>(task_id)))
                .first_async();
            if (task == null || ((!)task).status == STATUS_SUSPENDED) {
                return;
            }
            ((!)task).status = STATUS_SUSPENDED;
            ((!)task).suspended_at = new DateTime.now_local();
            yield db.update_async<TaskEntity>((!)task);
            yield pulse.bump(task_id, "task suspended");
        }

        /**
        /** Withdraws the suspension stamp after a successful resume. */
        public async void clear_suspension(int64 task_id) throws Error {
            var task = yield db.query<TaskEntity>()
                .where(expr("id == $0", elem<int64?>(task_id)))
                .first_async();
            if (task == null || ((!)task).suspended_at == null) {
                return;
            }
            ((!)task).suspended_at = null;
            yield db.update_async<TaskEntity>((!)task);
        }

        /**
         * Whether any OTHER task sharing this task's working context
         * moved while it was parked — a sibling under the same parent,
         * or any task of the same project on the same shared working
         * branch. "Moved" means any recorded transition after the park
         * timestamp: a completion/failure (`completed_at`), a dispatch
         * (session start), or a progress entry. The park marker is the
         * suspension stamp, else the newest block's creation (the
         * blocked-at marker), else the session's last recorded
         * activity. False with no siblings or nothing moved — the wire's
         * `branch_advanced`, so the daemon can skip the pointless
         * `git pull --rebase` when the shared branch did not advance.
         */
        public async bool branch_advanced_since(int64 task_id) throws Error {
            var me = yield db.query<TaskEntity>()
                .where(expr("id == $0", elem<int64?>(task_id)))
                .first_async();
            if (me == null) {
                return false;
            }

            DateTime? park = ((!)me).suspended_at;
            if (park == null) {
                foreach (var block in yield db.query<TaskBlock>()
                        .where(expr("task_id == $0", elem<int64?>(task_id)))
                        .order_by(expr("id"))
                        .materialise_async()) {
                    if (block.created_at != null) {
                        park = ((!)block).created_at;
                    }
                }
            }
            if (park == null) {
                foreach (var session in yield db.query<Session>()
                        .where(expr("task_id == $0", elem<int64?>(task_id)))
                        .materialise_async()) {
                    foreach (var entry in yield db.query<SessionCommand>()
                            .where(expr("session_id == $0", elem<int64?>(session.id)))
                            .materialise_async()) {
                        if (entry.started_at != null
                                && (park == null
                                    || ((!)entry.started_at).compare((!)park) > 0)) {
                            park = (!)entry.started_at;
                        }
                    }
                }
            }
            if (park == null) {
                // Never parked (or nothing recorded): nothing could have
                // advanced past a marker that does not exist.
                return false;
            }

            foreach (var sibling in yield sibling_tasks((!)me)) {
                if (sibling.completed_at != null
                        && ((!)sibling.completed_at).compare((!)park) > 0) {
                    return true;
                }
                foreach (var session in yield db.query<Session>()
                        .where(expr("task_id == $0", elem<int64?>(sibling.id)))
                        .materialise_async()) {
                    if (session.started_at != null
                            && session.started_at.compare((!)park) > 0) {
                        return true;
                    }
                }
                foreach (var progress in yield db.query<TaskProgress>()
                        .where(expr("task_id == $0", elem<int64?>(sibling.id)))
                        .materialise_async()) {
                    if (progress.created_at != null
                            && progress.created_at.compare((!)park) > 0) {
                        return true;
                    }
                }
            }
            return false;
        }

        /**
         * The tasks sharing `task`'s working context (excluding it):
         * same-project tasks on the same non-empty working branch, and
         * — for a goal agent — every other task of its goal (their
         * completions merge into the goal branch, so the goal agent's
         * post-park pull must hear them).
         */
        private async ImmutableLot<TaskEntity> sibling_tasks(TaskEntity task) throws Error {
            var rows = new Vector<TaskEntity>();
            var seen = new Invercargill.DataStructures.HashSet<string>();
            if (task.goal_id > 0) {
                foreach (var sibling in yield db.query<TaskEntity>()
                        .where(expr("goal_id == $0", elem<int64?>(task.goal_id)))
                        .materialise_async()) {
                    if (sibling.id != task.id && seen.add(sibling.id.to_string())) {
                        rows.add(sibling);
                    }
                }
            }
            if (task.task_branch != null && ((!)task.task_branch).length > 0) {
                foreach (var sibling in yield db.query<TaskEntity>()
                        .where(expr("project_id == $0 && task_branch == $1",
                            elem<int64?>(task.project_id), elem(((!)task).task_branch)))
                        .materialise_async()) {
                    if (sibling.id != task.id && seen.add(sibling.id.to_string())) {
                        rows.add(sibling);
                    }
                }
            }
            return rows.to_immutable_buffer();
        }

        /**
         * Returns a non-terminal task to the queue: closes its open
         * sessions (outcome 'stopped') and frees the saddle that was
         * running it. Plan 14 B1: a parked (suspended) task requeues
         * too — the goal page's requeue-to-resume reaches a goal
         * agent parked on a dead saddle; the suspension stamp clears
         * so the next dispatch starts a fresh parking window.
         */
        public async void requeue_task(int64 task_id) throws Error {
            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
                    || ((!)task).status == STATUS_SUSPENDED)) {
                ((!)task).status = STATUS_QUEUED;
                if (((!)task).suspended_at != null) {
                    ((!)task).suspended_at = null;
                }
                yield db.update_async<TaskEntity>((!)task);
            }
            yield close_open_sessions(task_id, "stopped");
            yield free_saddles_running(task_id);
            yield pulse.bump(task_id, "task requeued");
        }

        /**
         * Returns a failed task to the queue for another attempt. The
         * failure summary moves into an agent comment (so the history keeps
         * it) before being cleared from the task, `completed_at` is
         * withdrawn, any open session closes and its saddle frees — the
         * same bookkeeping as {@link requeue_task}, plus the audit
         * comment "Task retried by user".
         */
        public async void retry_task(int64 task_id) throws Error {
            var task = yield db.query<TaskEntity>()
                .where(expr("id == $0", elem<int64?>(task_id)))
                .first_async();
            if (task == null) {
                throw new Error(Quark.from_string("clanker-saddle"), 8,
                    @"No task exists with id $(task_id)");
            }

            if (((!)task).summary != null && ((!)task).summary.length > 0) {
                yield add_comment(task_id, "agent", (!)(((!)task).summary));
            }
            ((!)task).status = STATUS_QUEUED;
            ((!)task).summary = null;
            ((!)task).completed_at = null;
            yield db.update_async<TaskEntity>((!)task);

            yield close_open_sessions(task_id, "stopped");
            yield free_saddles_running(task_id);
            yield add_comment(task_id, "agent", "Task retried by user");
            yield pulse.bump(task_id, "task retried");
        }

        /**
         * Queued tasks whose auto-enqueue predecessor (if any) is
         * done, ordered by priority (1 first) then creation time —
         * the dispatcher's candidate list.
         */
        public async ImmutableLot<TaskEntity> get_ready_tasks() throws Error {
            var queued = yield db.query<TaskEntity>()
                .where(expr("status == $0", elem(STATUS_QUEUED)))
                .materialise_async();
            var all_tasks = yield db.query<TaskEntity>().materialise_async();

            var rows = new Vector<TaskEntity>();
            foreach (var task in queued) {
                bool ready = true;
                if (task.enqueue_after_task_id > 0) {
                    // The auto-enqueue predecessor also gates dispatch:
                    // a dependent that reached the queue some other way
                    // (a predecessor set after a manual enqueue) still
                    // waits for it to finish.
                    ready = false;
                    foreach (var other in all_tasks) {
                        if (other.id == task.enqueue_after_task_id) {
                            ready = other.status == STATUS_DONE;
                            break;
                        }
                    }
                }
                if (ready) {
                    rows.add(task);
                }
            }

            var sorted = rows.sort((a, b) => {
                if (a.priority != b.priority) {
                    return (int)(a.priority - b.priority);
                }
                int64 a_created = a.created_at != null ? ((!)a.created_at).to_unix() : 0;
                int64 b_created = b.created_at != null ? ((!)b.created_at).to_unix() : 0;
                return (int)(a_created - b_created);
            });
            return sorted.to_immutable_buffer();
        }

        /** Comments on a task, oldest first. */
        public async ImmutableLot<TaskComment> list_comments(int64 task_id) throws Error {
            return yield db.query<TaskComment>()
                .where(expr("task_id == $0", elem<int64?>(task_id)))
                .order_by(expr("id"))
                .materialise_async();
        }

        /** Blocks on a task, oldest first. */
        public async ImmutableLot<TaskBlock> list_blocks(int64 task_id) throws Error {
            return yield db.query<TaskBlock>()
                .where(expr("task_id == $0", elem<int64?>(task_id)))
                .order_by(expr("id"))
                .materialise_async();
        }

        /** The task's newest block, or null when it has none. */
        public async TaskBlock? newest_block(int64 task_id) throws Error {
            TaskBlock? newest = null;
            foreach (var block in yield list_blocks(task_id)) {
                newest = block;
            }
            return newest;
        }

        /**
         * True when the task's newest block exists and carries an
         * answer — a verdict is waiting to reach the parked workspace
         * (the answer-delivery pass, the affinity resume, or a later
         * assignment payload). Such a task must never be requeued: the
         * requeue closes the session the resume machinery needs.
         * Plan 15 C1: goal blocks live on the run task as single rows
         * — no mirror pair exists to fall out of sync, so the newest
         * row's own answer is the whole truth.
         */
        public async bool newest_block_answered(int64 task_id) throws Error {
            var newest = yield newest_block(task_id);
            return newest != null && ((!)newest).answer != null;
        }

        /** All commands recorded against a session, oldest first. */
        public async ImmutableLot<SessionCommand> list_commands(int64 session_id) throws Error {
            return yield db.query<SessionCommand>()
                .where(expr("session_id == $0", elem<int64?>(session_id)))
                .order_by(expr("id"))
                .materialise_async();
        }

        /**
         * True when the session carries raw PTY log rows — the terminal
         * view then renders the concatenated raw log instead of the
         * typed entry stream.
         */
        public async bool has_raw_log(int64 session_id) throws Error {
            var probe = yield db.query<SessionRawLog>()
                .where(expr("session_id == $0", elem<int64?>(session_id)))
                .first_async();
            return probe != null;
        }

        /**
         * The session still open for `task_id` (no `ended_at`), or null.
         * The null check runs in Vala: the expression layer cannot compare
         * columns against null.
         */
        public async Session? open_session_for_task(int64 task_id) throws Error {
            var sessions = yield db.query<Session>()
                .where(expr("task_id == $0", elem<int64?>(task_id)))
                .order_by(expr("id"))
                .materialise_async();
            foreach (var session in sessions) {
                if (session.ended_at == null) {
                    return session;
                }
            }
            return null;
        }

        private async void close_open_sessions(int64 task_id, string outcome) throws Error {
            var sessions = yield db.query<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 = outcome;
                yield db.update_async<Session>(session);
            }
        }

        private async void free_saddles_running(int64 task_id) throws Error {
            // One conditional UPDATE, atomic against a concurrent ask
            // allocation claiming a new task for the saddle between a
            // read and a full-row write.
            var command = connection.create_command(
                "UPDATE saddles SET status = 'idle', current_task_id = 0, allocated_at = NULL"
                + " WHERE current_task_id = :task");
            command.with_parameter<Invercargill.Element>("task", elem<int64?>(task_id));
            yield command.execute_non_query_async();
        }
    }
}