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";

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

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

    /**
     * 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>();

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

        /**
         * 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, ordered
         * by priority (1 first) then creation time — filtered and sorted
         * in the query, one materialisation.
         */
        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(expr("priority"))
                .order_by(expr("created_at"))
                .materialise_async();
        }

        /**
         * Every standalone human-workable task row (kind task, no
         * goal), any status, ordered like {@link list_tasks} — 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(expr("priority"))
                .order_by(expr("created_at"))
                .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) {
                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)");
            }
            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 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<TaskProgress>()
                    .where(expr("task_id == $0", elem<int64?>(task.id)))
                    .materialise_async()) {
                yield db.delete_async<TaskProgress>(row);
            }
            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 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 (`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();
        }

        /**
         * 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', 'user:<name>' or
         * 'system' (the harness-authored bookkeeping rows — plan 15
         * C2's distinct system-event class, never agent content).
         */
        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;
        }

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

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

            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);
                }
            }
            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.
         * 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.
         */
        private 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 {
            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);
        }

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

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

            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 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 — {@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");
        }

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

        /**
         * 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` and frees whatever
         * saddle row still runs it — one conditional UPDATE, atomic
         * against a concurrent ask allocation claiming a new task for
         * the saddle between a read and a full-row write. `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}) keeps its
         * own guarded OFFLINE write for the saddle row and calls this
         * after it — the idle-free below no-ops against that row (its
         * current_task_id already cleared under the entry-seen guard)
         * while the session close applies; its keep-parked gate (an
         * answered block awaiting delivery) skips the release
         * entirely.
         */
        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)) {
                    ((!)task).status = STATUS_QUEUED;
                    if (((!)task).suspended_at != null) {
                        ((!)task).suspended_at = null;
                    }
                    yield db.update_async<TaskEntity>((!)task);
                }
            }
            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);
            }
            var command = connection.create_command(
                "UPDATE saddles SET status = '" + SADDLE_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();
        }
    }
}