clanker-saddle/src/Services/TaskService.vala

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

namespace ClankerSaddle.Services {

    /** Valid task statuses (the `tasks.status` wire values). */
    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";

    /**
     * A task's computed presentation state (stored status + open
     * subtasks + 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), `delegating x/y`
     * while non-terminal subtasks exist, `waiting on subtasks` for the
     * parked suspended state, otherwise the stored status. Pure
     * presentation — the stored statuses and the dispatcher are
     * untouched.
     */
    public class TaskDisplay : Object {
        public string status { get; set; default = ""; }
        public bool terminal { get; set; }
        public bool failed { get; set; }
        public int64 subtask_total { get; set; }
        public int64 subtask_open { get; set; }

        public static TaskDisplay compute(TaskEntity task, bool review_pending,
                int64 subtask_total, int64 subtask_open) {
            var view = new TaskDisplay() {
                terminal = task.status == STATUS_DONE || task.status == STATUS_FAILED,
                failed = task.status == STATUS_FAILED,
                subtask_total = subtask_total,
                subtask_open = subtask_open
            };
            if (review_pending) {
                view.status = "review requested";
            } else if (subtask_open > 0) {
                view.status = @"delegating $(subtask_total - subtask_open)/$(subtask_total)";
            } else if (task.status == STATUS_SUSPENDED) {
                view.status = "waiting on subtasks";
            } 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, child
     * tasks, dependencies 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<TaskEntity> subtasks { get; set; }
        public ImmutableLot<TaskDependency> dependencies { get; set; }
        public ImmutableLot<Session> sessions { get; set; }
    }

    /**
     * Task storage and lifecycle: creation, dependency-aware readiness,
     * agent interaction primitives (comments, blocks, plans, progress,
     * reviews, subtasks) and completion. Saddle bookkeeping rows are kept
     * consistent on every terminal transition.
     */
    public class TaskService : Object {

        private OrmSession db = inject<OrmSession>();
        private StatePulse pulse = inject<StatePulse>();

        /**
         * Inserts a task (status 'queued'). `parent_task_id`,
         * `description`, `task_branch` and `branch_from` may be null;
         * `branch_from` prefills from the project's default branch and
         * `task_branch` is inherited from the parent task when unset;
         * priority defaults to 5 (1 is highest).
         */
        public async TaskEntity create_task(int64 project_id, string title, string? description,
                int64 priority, bool is_design_task, int64? parent_task_id,
                string? task_branch = null, string? branch_from = null) 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 (resolved_task_branch == null && parent_task_id != null) {
                var parent = yield db.query<TaskEntity>()
                    .where(expr("id == $0", elem<int64?>(parent_task_id)))
                    .first_async();
                if (parent != null && ((!)parent).task_branch != null) {
                    resolved_task_branch = ((!)parent).task_branch;
                }
            }

            var task = new TaskEntity() {
                parent_task_id = parent_task_id != null ? ((!)parent_task_id) : 0,
                project_id = project_id,
                title = title,
                description = description != null && description.length > 0 ? description : null,
                priority = priority.clamp(1, 9),
                is_design_task = is_design_task,
                status = STATUS_QUEUED,
                task_branch = resolved_task_branch,
                branch_from = resolved_branch_from.length > 0 ? resolved_branch_from : null,
                created_at = new DateTime.now_local()
            };
            yield db.insert_async<TaskEntity>(task);
            yield pulse.bump(task.id, "task created");
            return task;
        }

        /**
         * Loads the full aggregate for `id`: the task, its comments, blocks,
         * progress, child tasks, dependencies 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),
                subtasks = yield db.query<TaskEntity>()
                    .where(expr("parent_task_id == $0", elem<int64?>(id)))
                    .order_by(expr("id"))
                    .materialise_async(),
                dependencies = yield db.query<TaskDependency>()
                    .where(expr("task_id == $0", elem<int64?>(id)))
                    .materialise_async(),
                sessions = yield db.query<Session>()
                    .where(expr("task_id == $0", elem<int64?>(id)))
                    .order_by(expr("id"))
                    .materialise_async()
            };
        }

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

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

            // Subtask counters per parent (stringly keyed: the Invercargill
            // collections cannot key on value types).
            var open_by_parent = new Dictionary<string, string>();
            var total_by_parent = new Dictionary<string, string>();
            foreach (var child in yield db.query<TaskEntity>()
                    .where(expr("parent_task_id > 0"))
                    .materialise_async()) {
                string parent = child.parent_task_id.to_string();
                if (!wanted.has(parent)) {
                    continue;
                }
                total_by_parent.set(parent,
                    (int64.parse(total_by_parent.get_or_default(parent) ?? "0") + 1).to_string());
                if (child.status != STATUS_DONE && child.status != STATUS_FAILED) {
                    open_by_parent.set(parent,
                        (int64.parse(open_by_parent.get_or_default(parent) ?? "0") + 1).to_string());
                }
            }

            var map = new Dictionary<string, TaskDisplay>();
            foreach (var task in tasks) {
                string key = task.id.to_string();
                map.set(key, TaskDisplay.compute(task,
                    review_pending.has(key),
                    int64.parse(total_by_parent.get_or_default(key) ?? "0"),
                    int64.parse(open_by_parent.get_or_default(key) ?? "0")));
            }
            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;
            int64 subtask_total = 0;
            int64 subtask_open = 0;
            foreach (var block in detail.blocks) {
                if (block.block_type == BLOCK_REVIEW && block.answered_at == null) {
                    review_pending = true;
                }
            }
            foreach (var child in detail.subtasks) {
                subtask_total++;
                if (child.status != STATUS_DONE && child.status != STATUS_FAILED) {
                    subtask_open++;
                }
            }
            return TaskDisplay.compute(detail.task, review_pending, subtask_total, subtask_open);
        }

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

        /**
         * Saves the editable fields of the task identified by `id`; null
         * arguments keep their current values. 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? update_task(int64 id, string? title, string? description,
                int64? priority, string? status) 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;
            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 (priority != null) {
                int64 clamped = ((!)priority).clamp(1, 9);
                if (((!)task).priority != clamped) {
                    ((!)task).priority = clamped;
                    changed = true;
                }
            }
            if (status != null && status.length > 0 && ((!)task).status != status) {
                ((!)task).status = status;
                if (status == STATUS_DONE && ((!)task).completed_at == null) {
                    ((!)task).completed_at = new DateTime.now_local();
                }
                changed = true;
            }
            yield db.update_async<TaskEntity>(task);
            if (changed) {
                yield pulse.bump(id, @"task $(status ?? "saved")");
            }
            return task;
        }

        /**
         * Saves the task's markdown plan (`saddle plan` or the web UI's
         * plan editor). An empty markdown clears the plan. Returns null
         * when the task does not exist.
         */
        public async TaskEntity? update_plan(int64 id, string markdown) throws Error {
            var task = yield db.query<TaskEntity>()
                .where(expr("id == $0", elem<int64?>(id)))
                .first_async();
            if (task == null) {
                return null;
            }
            ((!)task).plan = markdown.strip().length > 0 ? markdown.strip() : null;
            yield db.update_async<TaskEntity>((!)task);
            yield pulse.bump(id, "plan updated");
            return task;
        }

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

        /**
         * Saves the task's branch fields: `task_branch` is the branch the
         * agent works on, `branch_from` the base it is cut from. Null
         * arguments keep their current values. Returns null when the task
         * does not exist.
         */
        public async TaskEntity? set_branch_fields(int64 id, string? task_branch,
                string? branch_from) throws Error {
            var task = yield db.query<TaskEntity>()
                .where(expr("id == $0", elem<int64?>(id)))
                .first_async();
            if (task == null) {
                return null;
            }
            if (task_branch != null && task_branch.length > 0) {
                ((!)task).task_branch = task_branch;
            }
            if (branch_from != null && branch_from.length > 0) {
                ((!)task).branch_from = branch_from;
            }
            yield db.update_async<TaskEntity>((!)task);
            yield pulse.bump(id, "branch fields saved");
            return task;
        }

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

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

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

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

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

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

        /**
         * Records a human's answer to a block. With an open session the
         * task stays 'blocked'/'review_requested' — the dispatcher resumes
         * the paused workspace and hands the answer over; without one the
         * task flips straight back to 'working'.
         */
        public async void answer_block(int64 block_id, string answer) 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();
            yield db.update_async<TaskBlock>((!)block);

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

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

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

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

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

        /** Records that `task_id` must not start until `depends_on_task_id` is done. */
        public async void add_dependency(int64 task_id, int64 depends_on_task_id) throws Error {
            if (task_id == depends_on_task_id) {
                throw new Error(Quark.from_string("clanker-saddle"), 6,
                    "A task cannot depend on itself");
            }
            var existing = yield db.query<TaskDependency>()
                .where(expr("task_id == $0 && depends_on_task_id == $1",
                    elem<int64?>(task_id), elem<int64?>(depends_on_task_id)))
                .materialise_async();
            if (existing.length > 0) {
                return;
            }
            var dependency = new TaskDependency() {
                task_id = task_id,
                depends_on_task_id = depends_on_task_id
            };
            yield db.insert_async<TaskDependency>(dependency);
            yield pulse.bump(task_id, "dependency added");
        }

        /** Creates a child task under `parent_task_id` (status 'queued'). */
        public async TaskEntity create_subtask(int64 parent_task_id, string title,
                string? description) throws Error {
            var parent = yield db.query<TaskEntity>()
                .where(expr("id == $0", elem<int64?>(parent_task_id)))
                .first_async();
            if (parent == null) {
                throw new Error(Quark.from_string("clanker-saddle"), 7,
                    @"No task exists with id $(parent_task_id)");
            }
            return yield create_task(((!)parent).project_id, title, description,
                ((!)parent).priority, false, parent_task_id);
        }

        /**
         * Marks a task done: records `summary` and `working_branch`, stamps
         * `completed_at`, closes any open session (outcome `outcome`) and
         * frees the saddle that was running it.
         */
        public async void complete_task(int64 task_id, string? summary,
                string? working_branch, string outcome) throws Error {
            var task = yield db.query<TaskEntity>()
                .where(expr("id == $0", elem<int64?>(task_id)))
                .first_async();
            if (task == null) {
                throw new Error(Quark.from_string("clanker-saddle"), 8,
                    @"No task exists with id $(task_id)");
            }

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

            yield close_open_sessions(task_id, outcome);
            yield free_saddles_running(task_id);
            yield pulse.bump(task_id, "task completed");
        }

        /**
         * Cancels a task: status 'failed' with a cancellation summary. Any
         * open session is closed with outcome 'stopped' and its saddle
         * freed.
         */
        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)");
            }

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

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

        /**
         * Records a saddle-reported failure: status 'failed' with the
         * daemon's message as summary, the explanation stored again as a
         * bot comment (so the thread carries it), the open session
         * closed with outcome 'failed' and its saddle freed.
         */
        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)");
            }

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

            yield close_open_sessions(task_id, "failed");
            yield free_saddles_running(task_id);
            yield pulse.bump(task_id, "task failed");
        }

        /**
         * Returns a non-terminal task to the queue: closes its open
         * sessions (outcome 'stopped') and frees the saddle that was
         * running it.
         */
        public async void requeue_task(int64 task_id) throws Error {
            var task = yield db.query<TaskEntity>()
                .where(expr("id == $0", elem<int64?>(task_id)))
                .first_async();
            if (task != null && (((!)task).status == STATUS_ASSIGNED
                    || ((!)task).status == STATUS_WORKING || ((!)task).status == STATUS_BLOCKED
                    || ((!)task).status == STATUS_REVIEW_REQUESTED)) {
                ((!)task).status = STATUS_QUEUED;
                yield db.update_async<TaskEntity>((!)task);
            }
            yield close_open_sessions(task_id, "stopped");
            yield free_saddles_running(task_id);
            yield pulse.bump(task_id, "task requeued");
        }

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

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

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

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

            var rows = new Vector<TaskEntity>();
            foreach (var task in queued) {
                bool ready = true;
                foreach (var dep in dependencies) {
                    if (dep.task_id != task.id) {
                        continue;
                    }
                    bool dep_done = false;
                    foreach (var other in all_tasks) {
                        if (other.id == dep.depends_on_task_id && other.status == STATUS_DONE) {
                            dep_done = true;
                            break;
                        }
                    }
                    if (!dep_done) {
                        ready = false;
                        break;
                    }
                }
                if (ready) {
                    rows.add(task);
                }
            }

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

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

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

        /** Every dependency edge in the graph (for queue annotations). */
        public async ImmutableLot<TaskDependency> list_all_dependencies() throws Error {
            return yield db.query<TaskDependency>().materialise_async();
        }

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

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

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

        private async void close_open_sessions(int64 task_id, string outcome) throws Error {
            var sessions = yield db.query<Session>()
                .where(expr("task_id == $0", elem<int64?>(task_id)))
                .materialise_async();
            foreach (var session in sessions) {
                if (session.ended_at != null) {
                    continue;
                }
                session.ended_at = new DateTime.now_local();
                session.outcome = outcome;
                yield db.update_async<Session>(session);
            }
        }

        private async void free_saddles_running(int64 task_id) throws Error {
            var saddles = yield db.query<Saddle>()
                .where(expr("current_task_id == $0", elem<int64?>(task_id)))
                .materialise_async();
            foreach (var saddle in saddles) {
                saddle.status = "idle";
                saddle.current_task_id = 0;
                saddle.allocated_at = null;
                yield db.update_async<Saddle>(saddle);
            }
        }
    }
}