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 system-event comment author (plan 15 C2): harness-authored
     * bookkeeping rows (aborts, requeues, retries, session
     * restorations) — never agent content.
     */
    public const string COMMENT_AUTHOR_SYSTEM = "system";
    public const string COMMENT_AUTHOR_AGENT = "agent";

    /**
     * Plan 16 S5 (U2): MECHANICAL harness bookkeeping that every
     * user-facing timeline filters out — the session-restore footnote
     * and the park-until notice (both remain in the raw terminal log,
     * which keeps everything). Matched by substring so the daemon's
     * [harness]-prefixed wordings and the web's own phrasing all stay
     * covered; substantive system rows (aborts, requeues, retries)
     * never match.
     */
    public static bool comment_is_mechanical(string body) {
        return body.contains("session context restored")
            || body.contains("parked until");
    }

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

    /** The display label of a goal-run parked on a review-family block (U3). */
    public const string DISPLAY_AWAITING_REVIEW = "awaiting your review";

    /**
     * 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. Plan 16
     * S5 (U3): a GOAL-RUN task parked on a review-family block (plan
     * or completion review) reads "awaiting your review" — agreeing
     * with its goal's label, never a bare "blocked". 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,
                bool goal_review_pending = false) {
            var view = new TaskDisplay() {
                terminal = task.status == STATUS_DONE || task.status == STATUS_FAILED,
                failed = task.status == STATUS_FAILED
            };
            if (goal_review_pending && task.kind == TASK_KIND_GOAL_RUN) {
                view.status = DISPLAY_AWAITING_REVIEW;
            } else if (review_pending) {
                view.status = "review requested";
            } else if (task.status == STATUS_SUSPENDED) {
                view.status = "parked";
            } else {
                view.status = task.status;
            }
            return view;
        }
    }

    /**
     * One LLM-reported progress line on a task — the ledger's progress
     * events projected back into the feed shape the pages read (id is
     * the event id, summary the payload's text).
     */
    public class TaskProgress : Object {

        public int64 id { get; set; }
        public int64 task_id { get; set; }
        public string summary { get; set; default = ""; }
        public DateTime? created_at { get; set; default = null; }
    }

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

    /**
     * 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>();
        private EventService events = inject<EventService>();
        private SaddlePollState poll_state = inject<SaddlePollState>();
        // Goal 28 task F: answers whose owner is BUSY ride the
        // singleton delivery hub (this service is request-scoped and
        // deliberately holds no SaddleService dependency — the
        // dispatcher's tick drains the queue through one).
        private AnswerDeliveryHub answer_hub = inject<AnswerDeliveryHub>();

        /**
         * 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 {
            // Goal 8 task 4: project_id 0 is a STANDALONE task — its
            // own ephemeral workspace, no project row anywhere.
            if (project_id < 0) {
                throw new Error(Quark.from_string("clanker-saddle"), 76,
                    @"Invalid project id $(project_id)");
            }
            string resolved_branch_from = branch_from ?? "";
            if (resolved_branch_from.length == 0 && project_id > 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");
            yield append_state_change(task.id, "", status, "Task created",
                creator == CREATOR_HUMAN ? EVENT_ACTOR_USER : EVENT_ACTOR_MODEL);
            return task;
        }

        /**
         * Appends a task-scoped state_change event carrying `from`/`to`
         * plus a human-readable line the timeline and narrative read
         * back verbatim.
         */
        private async void append_state_change(int64 task_id, string from, string to,
                string text, string actor) throws Error {
            var payload = new Json.Object();
            payload.set_string_member("from", from);
            payload.set_string_member("to", to);
            payload.set_string_member("text", text);
            yield events.append(EVENT_STATE_CHANGE, actor, EVENT_SCOPE_TASK, task_id,
                payload);
        }

        /**
         * 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_desc(expr("id"))
                    .materialise_async()
            };
        }

        /**
         * Computed display states for a batch of tasks (all lists share
         * this), keyed by task id string. Unanswered review-family
         * blocks load in ONE batched query over the wanted tasks, not
         * per task. Plan 16 S5 (U3): a goal-run task with a pending
         * review-family block (plan/completion/review) reads "awaiting
         * your review" — the run-task card agrees with its goal.
         */
        public async Dictionary<string, TaskDisplay> display_map_for(
                Lot<TaskEntity> tasks) throws Error {
            var wanted = new Invercargill.DataStructures.HashSet<string>();
            var goal_runs = new Invercargill.DataStructures.HashSet<string>();
            foreach (var task in tasks) {
                wanted.add(task.id.to_string());
                if (task.kind == TASK_KIND_GOAL_RUN) {
                    goal_runs.add(task.id.to_string());
                }
            }

            var review_pending = new Invercargill.DataStructures.HashSet<string>();
            var goal_review_pending = new Invercargill.DataStructures.HashSet<string>();
            if (wanted.length > 0) {
                foreach (var block in yield db.query<TaskBlock>()
                        .where(id_in("task_id", wanted))
                        .materialise_async()) {
                    if (block.answered_at != null
                            || !wanted.has(block.task_id.to_string())) {
                        continue;
                    }
                    if (block.block_type == BLOCK_REVIEW) {
                        review_pending.add(block.task_id.to_string());
                    }
                    if (block.block_type == BLOCK_REVIEW
                            || block.block_type == BLOCK_COMPLETION
                            || block.block_type == BLOCK_PLAN) {
                        if (goal_runs.has(block.task_id.to_string())) {
                            goal_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()),
                        goal_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;
            bool goal_review_pending = false;
            foreach (var block in detail.blocks) {
                if (block.answered_at != null) {
                    continue;
                }
                if (block.block_type == BLOCK_REVIEW) {
                    review_pending = true;
                }
                if (detail.task.kind == TASK_KIND_GOAL_RUN
                        && (block.block_type == BLOCK_REVIEW
                            || block.block_type == BLOCK_COMPLETION
                            || block.block_type == BLOCK_PLAN)) {
                    goal_review_pending = true;
                }
            }
            return TaskDisplay.compute(detail.task, review_pending, goal_review_pending);
        }

        /**
         * Lists tasks, optionally filtered by status and/or project,
         * newest first (creation descending, id as the same-second
         * tiebreak) — the list-page consistency sweep: every table
         * leads with the most recent row. Priority stays a visible
         * column, no longer a sort key; the dispatcher's candidate
         * order ({@link get_ready_tasks}) keeps priority-first.
         */
        public async ImmutableLot<TaskEntity> list_tasks(string? status, int64? project_id) throws Error {
            var query = db.query<TaskEntity>();
            if (status != null && status.length > 0) {
                query.where(expr("status == $0", elem(status)));
            }
            if (project_id != null) {
                query.where(expr("project_id == $0", elem<int64?>(project_id)));
            }
            return yield query
                .order_by_desc(expr("created_at"))
                .order_by_desc(expr("id"))
                .materialise_async();
        }

        /**
         * Every standalone human-workable task row (kind task, no
         * goal), any status, ordered like {@link list_tasks}
         * (newest first) — the auto-enqueue-after dropdowns' option
         * list, queried for exactly the rows they show instead of
         * every task in the table.
         */
        public async ImmutableLot<TaskEntity> list_standalone_tasks() throws Error {
            return yield db.query<TaskEntity>()
                .where(expr("kind == $0 && goal_id == $1",
                    elem(TASK_KIND_TASK), elem<int64?>(0)))
                .order_by_desc(expr("created_at"))
                .order_by_desc(expr("id"))
                .materialise_async();
        }

        /**
         * 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) {
                    if (!is_valid_mode(mode)) {
                        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) {
                string from = ((!)task).status;
                if (status != STATUS_DONE && status != STATUS_FAILED) {
                    // Terminal states are sticky (plan 13 B2's
                    // invariant): a NON-terminal transition must never
                    // downgrade a done/failed task. The entity writes
                    // below interleave with completions on the main
                    // loop (the terminal checks in the callers have
                    // already yielded by then — a late 'working' push
                    // racing a 'done' push used to resurrect the task
                    // as working AFTER its session closed done), so
                    // the guard lives in the UPDATE's WHERE: atomic.
                    // The row is re-read afterwards so the caller sees
                    // the truth (terminal included), never a stale
                    // entity it might write back.
                    var guard = connection.create_command(
                        "UPDATE tasks SET status = :status WHERE id = :id"
                        + " AND status NOT IN ('done', 'failed')");
                    guard.with_parameter<Invercargill.Element>("status",
                        elem(status));
                    guard.with_parameter<Invercargill.Element>("id",
                        elem<int64?>(id));
                    yield guard.execute_non_query_async();
                    task = yield db.query<TaskEntity>()
                        .where(expr("id == $0", elem<int64?>(id)))
                        .first_async();
                    if (task == null) {
                        return null;
                    }
                } else {
                    ((!)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)");
                var moved = new Json.Object();
                moved.set_string_member("from", from);
                moved.set_string_member("to", status);
                moved.set_string_member("text", @"Status → $(status)");
                yield events.append(EVENT_STATE_CHANGE, EVENT_ACTOR_SYSTEM,
                    EVENT_SCOPE_TASK, id, moved);
            }
            return task;
        }

        /** Every task under `goal_id` (both kinds), newest 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_desc(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");
            yield append_state_change(id, STATUS_DRAFT, STATUS_QUEUED,
                "Task enqueued", actor == CREATOR_HUMAN ? EVENT_ACTOR_USER : EVENT_ACTOR_MODEL);
        }

        /**
         * 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 row teardown every task deletion runs (the shared core
         * of the human Delete action and the goal cascade's quiet
         * delete): attached rules, capabilities, progress, comments
         * and blocks die with the task, predecessor references clear,
         * the row goes and the pulse fires.
         */
        private async void delete_task_rows(TaskEntity task) throws Error {
            foreach (var link in yield db.query<TaskRule>()
                    .where(expr("task_id == $0", elem<int64?>(task.id)))
                    .materialise_async()) {
                yield db.delete_async<TaskRule>(link);
            }
            foreach (var link in yield db.query<TaskCapability>()
                    .where(expr("task_id == $0", elem<int64?>(task.id)))
                    .materialise_async()) {
                yield db.delete_async<TaskCapability>(link);
            }
            foreach (var row in yield db.query<TaskComment>()
                    .where(expr("task_id == $0", elem<int64?>(task.id)))
                    .materialise_async()) {
                yield db.delete_async<TaskComment>(row);
            }
            foreach (var row in yield db.query<TaskBlock>()
                    .where(expr("task_id == $0", elem<int64?>(task.id)))
                    .materialise_async()) {
                yield db.delete_async<TaskBlock>(row);
            }
            yield events.delete_for_task(task.id);
            yield clear_enqueue_after_references(task.id);
            yield db.delete_async<TaskEntity>(task);
            yield pulse.bump(task.id, "task deleted");
        }

        /**
         * 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;
            }
            yield delete_task_rows((!)task);
        }

        /**
         * 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)");
            }
            yield delete_task_rows((!)task);
        }

        /**
         * 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 to the task's ledger (the daemon's
         * progress events land in the same ledger through the chunks
         * push; the goal-era results-entry path that wrote
         * `task_progress` rows is gone).
         */
        public async TaskProgress add_progress(int64 task_id, string summary) throws Error {
            var payload = new Json.Object();
            payload.set_string_member("text", summary);
            var record = yield events.append(EVENT_PROGRESS, EVENT_ACTOR_MODEL,
                EVENT_SCOPE_TASK, task_id, payload);
            yield pulse.bump(task_id, "progress");
            return new TaskProgress() {
                id = record.id,
                task_id = task_id,
                summary = summary,
                created_at = record.created_at
            };
        }

        /** The task's progress history, newest first. */
        public async ImmutableLot<TaskProgress> list_progress(int64 task_id) throws Error {
            var rows = new Vector<TaskProgress>();
            foreach (var record in yield events.list_for_task(task_id,
                    { EVENT_PROGRESS })) {
                rows.add(new TaskProgress() {
                    id = record.id,
                    task_id = task_id,
                    summary = EventService.payload_string(record, "text"),
                    created_at = record.created_at
                });
            }
            return rows.to_immutable_buffer();
        }

        /**
         * Appends a comment; `author` is 'agent', 'user:<name>' or
         * 'system' (the harness-authored bookkeeping rows — plan 15
         * C2's distinct system-event class, never agent content). The
         * comment event references the row (the task_comments table
         * stays; the ledger carries the reference plus the body).
         */
        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);
            var payload = new Json.Object();
            payload.set_int_member("comment_id", comment.id);
            payload.set_string_member("body", body);
            payload.set_string_member("author", author);
            if (comment.quote != null) {
                payload.set_string_member("quote", (!)comment.quote);
            }
            yield events.append(EVENT_COMMENT,
                author == "agent" ? EVENT_ACTOR_MODEL
                : author.has_prefix("user:") ? EVENT_ACTOR_USER : EVENT_ACTOR_SYSTEM,
                EVENT_SCOPE_TASK, task_id, payload);
            yield pulse.bump(task_id, "comment");
            return comment;
        }

        /**
         * Records a harness/system event on the task's thread (plan
         * 15 C2): park/resume bookkeeping (aborts, requeues,
         * retries), the session-context restoration, unreachable
         * notices — authored 'system', rendered with the distinct
         * muted system styling, never an agent comment row.
         */
        public async TaskComment add_system_comment(int64 task_id, string body) throws Error {
            return yield add_comment(task_id, COMMENT_AUTHOR_SYSTEM, body);
        }

        /**
         * Materialises an AGENT comment row from its already-stored
         * ledger event: the plan-17 chunks push carries the remark as
         * a comment event, and this creates the task_comments row the
         * page's comment section renders. No second event is appended
         * (the ledger already carries the remark); only fresh
         * (session, seq) pairs reach here, so a redelivered push can
         * never double the row.
         */
        public async TaskComment adopt_agent_comment(int64 task_id, string body) throws Error {
            var comment = new TaskComment() {
                task_id = task_id,
                author = COMMENT_AUTHOR_AGENT,
                body = body,
                created_at = new DateTime.now_local()
            };
            yield db.insert_async<TaskComment>(comment);
            yield pulse.bump(task_id, "agent comment");
            return comment;
        }

        /**
         * 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();
            string previous_status = task != null ? ((!)task).status : "";
            if (task != null && (((!)task).status == STATUS_ASSIGNED || ((!)task).status == STATUS_WORKING
                    || ((!)task).status == STATUS_BLOCKED || ((!)task).status == STATUS_REVIEW_REQUESTED)) {
                // The park flip is a guarded update: a completion
                // interleaving the raise keeps its terminal state (the
                // read above already yielded — the entity is stale by
                // write time under load; plan 13 B2's sticky rule).
                var guard = connection.create_command(
                    "UPDATE tasks SET status = :status WHERE id = :id"
                    + " AND status IN ('assigned', 'working', 'blocked', 'review_requested')");
                guard.with_parameter<Invercargill.Element>("status",
                    elem(block.block_type == BLOCK_REVIEW
                        ? STATUS_REVIEW_REQUESTED : STATUS_BLOCKED));
                guard.with_parameter<Invercargill.Element>("id",
                    elem<int64?>(task_id));
                yield guard.execute_non_query_async();
                // Plan 16 D4: a raised block parks the round — the
                // session row CLOSES 'parked' right here (every raise
                // path funnels through this flip, entry-driven raises
                // included, so no parked task ever keeps a dangling
                // open session). The workspace awaits its resuming
                // allocation on a fresh row.
                yield park_open_sessions(task_id);
            }
            yield pulse.bump(task_id, "block raised");
            var raised_payload = new Json.Object();
            raised_payload.set_string_member("from", previous_status);
            raised_payload.set_string_member("to", block.block_type == BLOCK_REVIEW
                ? STATUS_REVIEW_REQUESTED : STATUS_BLOCKED);
            raised_payload.set_string_member("block_type", block.block_type);
            raised_payload.set_string_member("question", block.question);
            raised_payload.set_string_member("text", @"$(block.block_type) raised: $(block.question)");
            yield events.append(EVENT_STATE_CHANGE, EVENT_ACTOR_SYSTEM, EVENT_SCOPE_TASK,
                task_id, raised_payload);
            // 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. A parked round (its
         * session closed 'parked' at the raise — plan 16 D4) stays
         * 'blocked'/'review_requested': the dispatcher resumes the
         * paused workspace and hands the answer over. Only a block on
         * a task that NEVER RAN (no session rows at all) 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);

            bool review_family = ((!)block).block_type == BLOCK_REVIEW
                || ((!)block).block_type == BLOCK_PLAN
                || ((!)block).block_type == BLOCK_COMPLETION;
            var verdict_payload = new Json.Object();
            verdict_payload.set_int_member("block_id", block_id);
            verdict_payload.set_string_member("block_type", ((!)block).block_type);
            if (((!)block).answered_by != null) {
                verdict_payload.set_string_member("by", ((!)block).answered_by);
            }
            if (review_family) {
                bool approved = answer == NOTICE_PLAN_APPROVED
                    || answer == NOTICE_REVIEW_APPROVED;
                verdict_payload.set_string_member("verdict",
                    approved ? "approved" : "feedback");
                verdict_payload.set_string_member("text", @"$(approved ? "approved" : "feedback") — $(answer)");
                yield events.append(EVENT_VERDICT,
                    ((!)block).answered_by != null ? EVENT_ACTOR_USER : EVENT_ACTOR_SYSTEM,
                    EVENT_SCOPE_TASK, ((!)block).task_id, verdict_payload);
            } else {
                verdict_payload.set_string_member("answer", answer);
                verdict_payload.set_string_member("text", @"$(((!)block).block_type) answered: $(answer)");
                yield events.append(EVENT_ANSWER,
                    ((!)block).answered_by != null ? EVENT_ACTOR_USER : EVENT_ACTOR_SYSTEM,
                    EVENT_SCOPE_TASK, ((!)block).task_id, verdict_payload);
            }

            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
                        && (yield newest_session_for_task(((!)block).task_id)) == null) {
                    ((!)task).status = STATUS_WORKING;
                    yield db.update_async<TaskEntity>((!)task);
                }
            }
            // Goal 28 task F (busy-answer delivery): when the task's
            // affine saddle holds a RUNNING session right now (the
            // race-restarted round, or a crash-restore that adopted
            // the parked workspace), NO affinity resume can ever
            // deliver this verdict — the busy saddle cannot ask. Park
            // it on the singleton hub instead: the dispatcher's tick
            // injects it into the running session through the
            // daemon's task-update channel ({@link
            // SaddleService.deliver_pending_answers}). A task with no
            // open session keeps the resume path (the verdict rides
            // the next resuming assignment's resume_notice, as
            // before).
            var running = yield open_session_for_task(((!)block).task_id);
            if (running != null) {
                answer_hub.offer(new PendingAnswerDelivery() {
                    block_id = block_id,
                    task_id = ((!)block).task_id,
                    notice = verdict_notice_for((!)block)
                });
            }
            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 "";
            }
            if (((!)task).project_id <= 0) {
                // Standalone tasks own no repository — no compare URL.
                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.
         * Batched: the candidate tasks load once, their sessions in
         * one query, the saddles in another.
         */
        public async Invercargill.DataStructures.HashSet<string> unreachable_task_ids() throws Error {
            var ids = new Invercargill.DataStructures.HashSet<string>();
            var candidate_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()) {
                    candidate_ids.add(task.id.to_string());
                }
            }
            if (candidate_ids.length == 0) {
                return ids;
            }

            // The newest open session per task: sessions ordered by id
            // ascending, the LAST open row per task wins (the
            // pre-batch semantics) — walked backwards, first hit per
            // task keeps the slot.
            var row_tasks = new Vector<string>();
            var row_saddles = new Vector<string>();
            foreach (var session in yield db.query<Session>()
                    .where(id_in("task_id", candidate_ids))
                    .order_by(expr("id"))
                    .materialise_async()) {
                if (session.ended_at == null) {
                    row_tasks.add(session.task_id.to_string());
                    row_saddles.add(session.saddle_id.to_string());
                }
            }
            var open_task_ids = new Vector<string>();
            var open_saddle_ids = new Vector<string>();
            var taken_tasks = new Invercargill.DataStructures.HashSet<string>();
            for (int index = (int) row_tasks.length - 1; index >= 0; index--) {
                if (taken_tasks.add(row_tasks.get(index))) {
                    open_task_ids.add(row_tasks.get(index));
                    open_saddle_ids.add(row_saddles.get(index));
                }
            }
            if (open_task_ids.length == 0) {
                return ids;
            }

            var saddle_ids = new Invercargill.DataStructures.HashSet<string>();
            foreach (var saddle_id in open_saddle_ids) {
                saddle_ids.add(saddle_id);
            }
            var offline = new Invercargill.DataStructures.HashSet<string>();
            foreach (var saddle in yield db.query<Saddle>()
                    .where(id_in("id", saddle_ids))
                    .materialise_async()) {
                if (saddle.status == SADDLE_OFFLINE) {
                    offline.add(saddle.id.to_string());
                }
            }
            for (int index = 0; index < open_task_ids.length; index++) {
                if (offline.has(open_saddle_ids.get(index))) {
                    ids.add(open_task_ids.get(index));
                }
            }
            return ids;
        }

        /**
         * An OR-chain of `column == id` comparisons for a batch of ids
         * (the grammar has no IN operator), or null for an empty
         * batch — the WHERE-in the batched lookups build.
         */
        /**
         * The OR-chain batch convention (plan 16 item 23 note): id
         * batches compare through decimal STRINGS — Invercargill
         * collections corrupt boxed int64 vectors, and the expression
         * layer's string compare against integer columns reads fine
         * (SQLite's dynamic typing). Shared with SaddleService's
         * derived fleet lists.
         */
        internal static Expression? id_in(string column,
                Invercargill.DataStructures.HashSet<string> ids) throws Error {
            Expression? chain = null;
            foreach (var id in ids) {
                var comparison = expr(column + " == $0", elem(id));
                chain = chain == null
                    ? comparison : new BinaryExpression((!)chain, comparison,
                        BinaryOperator.OR);
            }
            return chain;
        }

        /**
         * 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 EntryJson.parse_string_array(block.options)) {
                    if (option == answer) {
                        return "Human selected: " + answer;
                    }
                }
                return "Human answered (other): " + answer;
            }
            return "Human answer: " + answer;
        }

        /**
         * The ONE terminal stamping path (complete/fail/cancel): owns
         * the status flip, the `completed_at` stamp and the summary
         * precedence — a non-empty `summary` replaces whatever the
         * row holds, else `empty_summary` applies only when the row
         * holds nothing. Callers keep their own idempotence guard
         * (terminal states are sticky, plan 13 B2).
         */
        private async void mark_terminal(TaskEntity task, string status, string? summary,
                string? empty_summary = null) throws Error {
            string from = task.status;
            task.status = status;
            task.completed_at = new DateTime.now_local();
            if (summary != null && summary.length > 0) {
                task.summary = summary;
            } else if ((task.summary ?? "").length == 0 && empty_summary != null) {
                task.summary = empty_summary;
            }
            yield db.update_async<TaskEntity>(task);
            var payload = new Json.Object();
            payload.set_string_member("from", from);
            payload.set_string_member("to", status);
            payload.set_string_member("text", status == STATUS_DONE
                ? "Task completed" : "Task failed");
            if ((task.summary ?? "").length > 0) {
                payload.set_string_member("summary", (!)task.summary);
            }
            yield events.append(EVENT_STATE_CHANGE, EVENT_ACTOR_SYSTEM,
                EVENT_SCOPE_TASK, task.id, payload);
        }

        /**
         * The ONE park stamping path: `suspended` plus the
         * `suspended_at` stamp, stamped once per parking window so the
         * resume trigger only sees goal events that landed after the
         * suspension. No-op when already parked (the stamp survives).
         * Plan 16 D4: parking closes the round's open sessions
         * ('parked') — the pause record the resume machinery keys on
         * is the newest session row, open or closed.
         */
        public async void set_parked(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 park_open_sessions(task_id);
            yield pulse.bump(task_id, "task suspended");
            yield append_state_change(task_id, STATUS_WORKING, STATUS_SUSPENDED,
                "Parked — awaiting the resume", EVENT_ACTOR_SYSTEM);
            // Plan 16 S5 (U2): the mechanical park notice persists in
            // the audit ledger (a system comment event) and never
            // renders on a user-facing timeline.
            yield add_system_comment(task_id,
                "goal agent parked until an owned task completes or fails");
        }

        /**
         * Marks a task done: records `summary`, 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 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;
            }

            yield mark_terminal((!)task, STATUS_DONE, summary);

            yield release_task(task_id, outcome, false);
            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 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");
            }

            yield mark_terminal((!)task, STATUS_FAILED, null, "Cancelled by user");
            {
                var payload = new Json.Object();
                payload.set_string_member("text", "Task cancelled by user");
                yield events.append(EVENT_CANCEL, EVENT_ACTOR_USER, EVENT_SCOPE_TASK,
                    task_id, payload);
            }

            yield release_task(task_id, OUTCOME_STOPPED, false);
            yield pulse.bump(task_id, "task cancelled");
        }

        /**
         * Records a saddle-reported failure: status 'failed' with the
         * daemon's message as summary, the open session closed with
         * outcome 'failed' and its saddle freed. Plan 16 S4
         * (terminal-text single representation): `tasks.summary` IS
         * the terminal cache — no duplicate agent comment is written
         * (the narrative renders `summary_lead`; the task page renders
         * the summary card). 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 summary write 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;
            }

            yield mark_terminal((!)task, STATUS_FAILED, summary,
                "Saddle reported a failure");

            yield release_task(task_id, OUTCOME_FAILED, false);
            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));
        }

        /** 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 record in yield events.list_for_task(task_id, new string[0])) {
                    if (record.created_at != null) {
                        park = (!)record.created_at;
                        break;
                    }
                }
            }
            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 record in yield events.list_for_task(sibling.id,
                        { EVENT_PROGRESS })) {
                    if (record.created_at != null
                            && ((!)record.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 — {@link release_task} with the requeue, the
         * single owner of the bookkeeping. 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 {
            yield release_task(task_id, OUTCOME_STOPPED, true);
            yield pulse.bump(task_id, "task requeued");
        }

        /**
         * Crash recovery at the ask: a daemon that asks for work is
         * provably idle — its own state machine parks the ask loop
         * while any session runs — so an open session on an
         * assigned/working task can only be a crash orphan (both
         * sides restart, nobody closes the row, and the busy record
         * then wedges the fleet: the saddle asks, the claim rolls
         * back, forever, while the UI shows it busy). Every orphan
         * older than the grace window requeues — the very ask that
         * found it can win the task straight back. The grace window
         * protects a manual /start whose session row lands while an
         * ask is in flight.
         */
        public const int64 ASK_RECOVERY_GRACE_US = 120 * 1000000L;

        /**
         * Goal 28 task F: how long a round's last push keeps its
         * session un-requeueable after the daemon asks — the window
         * in which a park's transition may still be in flight
         * (matches the daemon-side park settle).
         */
        public const int64 PARK_SETTLE_GRACE_S = 5;

        public async int requeue_interrupted_sessions(Saddle saddle) throws Error {
            var requeued = 0;
            foreach (var session in yield db.query<Session>()
                    .where(expr("saddle_id == $0", elem<int64?>(saddle.id)))
                    .order_by(expr("id"))
                    .materialise_async()) {
                if (session.ended_at != null) {
                    continue;
                }
                var started = session.started_at ?? new DateTime.now_local();
                if (new DateTime.now_local().difference(started)
                        < ASK_RECOVERY_GRACE_US) {
                    continue;
                }
                // Goal 28 task F (the allocation guard's requeue leg):
                // a round that pushed chunks/results moments ago was
                // LIVE — its closing transition (a park carrying the
                // block payload) may still be queued in the daemon's
                // outbox. Requeueing here handed the parked task back
                // as a crash restore with its question gone (the
                // task-187 race); hold instead — the ask 204s, the
                // transition lands, the next ask sees the parked row.
                if (poll_state.seconds_since_push(session.id)
                        < PARK_SETTLE_GRACE_S) {
                    continue;
                }
                var task = yield db.query<TaskEntity>()
                    .where(expr("id == $0", elem<int64?>(session.task_id)))
                    .first_async();
                if (task == null || (((!)task).status != STATUS_ASSIGNED
                        && ((!)task).status != STATUS_WORKING)) {
                    continue;
                }
                yield requeue_task(((!)task).id);
                yield add_system_comment(((!)task).id,
                    @"Session interrupted (saddle '$(saddle.name)' restarted mid-run); requeued");
                requeued++;
            }
            return requeued;
        }

        /**
         * Returns a failed task to the queue for another attempt.
         * Plan 16 S4: the failure summary is the terminal cache of the
         * attempt that earned it — retry clears it and copies it
         * NOWHERE (the narrative rendered its lead, the page rendered
         * its card; the next attempt writes a fresh one), so no
         * summary→comment duplicate joins the thread. `completed_at`
         * is withdrawn, any open session closes and its saddle frees —
         * the same bookkeeping as {@link requeue_task} (retry's flip
         * is unconditional: the attempt resets whatever it held),
         * 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)");
            }

            ((!)task).status = STATUS_QUEUED;
            ((!)task).summary = null;
            ((!)task).completed_at = null;
            yield db.update_async<TaskEntity>((!)task);

            yield release_task(task_id, OUTCOME_STOPPED, false);
            yield add_system_comment(task_id, "Task retried by user");
            yield append_state_change(task_id, STATUS_FAILED, STATUS_QUEUED,
                "Task retried by user", EVENT_ACTOR_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. Only the CANDIDATES'
         * predecessors are queried (not every task row).
         */
        public async ImmutableLot<TaskEntity> get_ready_tasks() throws Error {
            var queued = yield db.query<TaskEntity>()
                .where(expr("status == $0", elem(STATUS_QUEUED)))
                .order_by(expr("priority"))
                .order_by(expr("created_at"))
                .materialise_async();

            var predecessor_ids = new Invercargill.DataStructures.HashSet<string>();
            foreach (var task in queued) {
                if (task.enqueue_after_task_id > 0) {
                    predecessor_ids.add(task.enqueue_after_task_id.to_string());
                }
            }
            var predecessor_status = new Dictionary<string, string>();
            if (predecessor_ids.length > 0) {
                foreach (var predecessor in yield db.query<TaskEntity>()
                        .where(id_in("id", predecessor_ids))
                        .materialise_async()) {
                    predecessor_status.set(predecessor.id.to_string(), predecessor.status);
                }
            }

            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. A predecessor row that
                    // does not resolve keeps the dependent waiting.
                    string? status = predecessor_status.get_or_default(
                        task.enqueue_after_task_id.to_string());
                    ready = status != null && status == STATUS_DONE;
                }
                if (ready) {
                    rows.add(task);
                }
            }
            return rows.to_immutable_buffer();
        }

        /** Comments on a task, newest first — the thread leads with
         * the latest remark (the list-page consistency sweep). */
        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_desc(expr("id"))
                .materialise_async();
        }

        /** Blocks on a task, newest first — the blocking sections
         * lead with the most recent question, RFC or review (the
         * list-page consistency sweep). */
        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_desc(expr("id"))
                .materialise_async();
        }

        /** The task's newest block, or null when it has none — the
         * FIRST row (list_blocks is newest-first since the ordering
         * sweep; the walker keeps the semantics explicit). */
        public async TaskBlock? newest_block(int64 task_id) throws Error {
            foreach (var block in yield list_blocks(task_id)) {
                return block;
            }
            return null;
        }

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

        /**
         * Goal 28 task F (batched verdict composition): every block
         * ANSWERED since `since` (the resumed session's start, or the
         * suspension stamp), oldest first — the resume notice joins
         * their verdict texts so a newer block can never swallow an
         * older answer. Undelivered-only filtering is the caller's
         * (the poll-state ledger decides what the agent has already
         * seen mid-session).
         */
        public async ImmutableLot<TaskBlock> blocks_answered_since(int64 task_id,
                DateTime? since) throws Error {
            var rows = new Vector<TaskBlock>();
            // list_blocks is newest-first; walk it backwards so the
            // batch reads oldest-first.
            var newest_first = yield list_blocks(task_id);
            var reversed = new Vector<TaskBlock>();
            foreach (var block in newest_first) {
                reversed.add(block);
            }
            for (int i = (int) reversed.length - 1; i >= 0; i--) {
                var block = reversed[i];
                if (block.answer == null || block.answered_at == null) {
                    continue;
                }
                if (since != null && block.answered_at.compare(((!)since)) < 0) {
                    continue;
                }
                rows.add(block);
            }
            return rows.to_immutable_buffer();
        }

        /**
         * The joined verdict text for a batch of answered blocks
         * (oldest first — {@link blocks_answered_since}): each
         * block's {@link verdict_notice_for} wording on its own
         * paragraph, in answer order. An empty batch yields "".
         */
        public static string batched_verdict_notice(ImmutableLot<TaskBlock> blocks) {
            var parts = new Vector<string>();
            foreach (var block in blocks) {
                parts.add(verdict_notice_for(block));
            }
            return string.joinv("\n\n", parts.to_array());
        }

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

        /**
         * The task's newest session row, open or closed (plan 16 D4:
         * a parked round's row is closed 'parked' — it is still the
         * pause record the affinity resume, the session-scoped task
         * CLI and the goal page's run surfaces key on), or null when
         * the task never ran.
         */
        public async Session? newest_session_for_task(int64 task_id) throws Error {
            Session? newest = null;
            foreach (var session in yield db.query<Session>()
                    .where(expr("task_id == $0", elem<int64?>(task_id)))
                    .order_by(expr("id"))
                    .materialise_async()) {
                newest = session;
            }
            return newest;
        }

        /**
         * Closes the task's open sessions with outcome 'parked'
         * (plan 16 D4 — the ONE park close every raise and suspend
         * path runs; idempotent, nothing to do for rows already
         * closed). The workspace stays on its saddle awaiting the
         * resuming allocation that opens the successor row.
         */
        public async void park_open_sessions(int64 task_id) throws Error {
            foreach (var session in yield db.query<Session>()
                    .where(expr("task_id == $0", elem<int64?>(task_id)))
                    .materialise_async()) {
                if (session.ended_at != null) {
                    continue;
                }
                session.ended_at = new DateTime.now_local();
                session.outcome = OUTCOME_PARKED;
                yield db.update_async<Session>(session);
            }
        }

        /**
         * The ONE release path (plan 16 R2, arch item 13): every path
         * that takes a task off its saddle runs through here. Closes
         * the task's open sessions with `outcome` — the close IS the
         * free (plan 16 item 23: busy/idle derive from the open
         * sessions; no saddle-row write remains to race an ask
         * allocation). `requeue` first returns a non-terminal task to
         * the queue (assigned/working/blocked/review_requested/
         * suspended — the suspension stamp withdraws so the next
         * dispatch opens a fresh parking window).
         *
         * The terminal callers (complete/cancel/fail) pass their own
         * outcome with the status already stamped; requeue_task and
         * the unregister path ride the requeue; retry_task does its
         * own unconditional flip (summary and completed_at die with
         * the attempt) and releases without one. The offline variant
         * ({@link SaddleService.mark_saddle_unreachable}) calls this
         * for its derived held task before its own guarded OFFLINE
         * stamp; its keep-parked gate (a task that parked mid-flight)
         * skips the release entirely.
         */
        /**
         * Goal 8 task 4 (comment-resurrect): re-queues a TERMINAL
         * non-goal task (goal_id == 0, status done/failed) for another
         * round after a human comment. The completion summary and
         * `completed_at` STAY (the report of the earlier round remains
         * readable — the daemon's next finish overwrites the summary),
         * and a fresh parked session row is seeded on the task's last
         * saddle so the plain-task requeue path (which keys on the
         * newest session's affine saddle) has a workspace to resume;
         * no session row at all (a fixture-shaped task) requeues onto
         * the open queue instead. The resurrection is idempotent from
         * the caller's side: only a terminal task flips.
         */
        public async void requeue_terminal_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).status != STATUS_DONE && ((!)task).status != STATUS_FAILED) {
                return;
            }
            string from = ((!)task).status;
            ((!)task).status = STATUS_QUEUED;
            ((!)task).completed_at = null;
            yield db.update_async<TaskEntity>((!)task);
            yield append_state_change(task_id, from, STATUS_QUEUED,
                @"Task re-opened by comment ($(from))", EVENT_ACTOR_USER);
            yield pulse.bump(task_id, "task re-opened by comment");
        }

        /**
         * Seeds a CLOSED 'parked' session row for `task_id` on the
         * saddle of its newest prior session (the resurrect paths'
         * resume anchor — the allocation's parked-resume arm needs a
         * session row to key the affine saddle and the successor row
         * on). Returns true when a row was seeded; false when the task
         * has no session history at all.
         */
        public async bool seed_parked_session(int64 task_id) throws Error {
            var newest = yield newest_session_for_task(task_id);
            if (newest == null) {
                return false;
            }
            var seeded = new Session() {
                task_id = task_id,
                saddle_id = ((!)newest).saddle_id,
                started_at = ((!)newest).started_at ?? new DateTime.now_local(),
                ended_at = new DateTime.now_local(),
                outcome = OUTCOME_PARKED
            };
            yield db.insert_async<Session>(seeded);
            return true;
        }

        public async void release_task(int64 task_id, string outcome, bool requeue) throws Error {
            if (requeue) {
                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)) {
                    string from = ((!)task).status;
                    ((!)task).status = STATUS_QUEUED;
                    if (((!)task).suspended_at != null) {
                        ((!)task).suspended_at = null;
                    }
                    yield db.update_async<TaskEntity>((!)task);
                    yield append_state_change(task_id, from, STATUS_QUEUED,
                        "Task requeued", EVENT_ACTOR_SYSTEM);
                }
            }
            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);
            }
        }
    }
}