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 comments { get; set; } public ImmutableLot blocks { get; set; } public ImmutableLot progress { get; set; } public ImmutableLot subtasks { get; set; } public ImmutableLot dependencies { get; set; } public ImmutableLot 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(); private StatePulse pulse = inject(); /** * 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() .where(expr("id == $0", elem(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() .where(expr("id == $0", elem(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(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() .where(expr("id == $0", elem(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() .where(expr("parent_task_id == $0", elem(id))) .order_by(expr("id")) .materialise_async(), dependencies = yield db.query() .where(expr("task_id == $0", elem(id))) .materialise_async(), sessions = yield db.query() .where(expr("task_id == $0", elem(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 display_map_for( Lot tasks) throws Error { var wanted = new Invercargill.DataStructures.HashSet(); foreach (var task in tasks) { wanted.add(task.id.to_string()); } var review_pending = new Invercargill.DataStructures.HashSet(); foreach (var block in yield db.query() .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(); var total_by_parent = new Dictionary(); foreach (var child in yield db.query() .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(); 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 list_tasks(string? status, int64? project_id) throws Error { var tasks = yield db.query().materialise_async(); var rows = new Vector(); 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() .where(expr("id == $0", elem(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(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() .where(expr("id == $0", elem(id))) .first_async(); if (task == null) { return null; } ((!)task).plan = markdown.strip().length > 0 ? markdown.strip() : null; yield db.update_async((!)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(row); yield pulse.bump(task_id, "progress"); return row; } /** The task's progress history, newest first. */ public async ImmutableLot list_progress(int64 task_id) throws Error { return yield db.query() .where(expr("task_id == $0", elem(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() .where(expr("id == $0", elem(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((!)task); yield pulse.bump(id, "branch fields saved"); return task; } /** Appends a comment; `author` is 'bot' or 'user:'. */ 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(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(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(block); var task = yield db.query() .where(expr("id == $0", elem(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((!)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() .where(expr("id == $0", elem(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((!)block); var task = yield db.query() .where(expr("id == $0", elem(((!)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((!)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() .where(expr("id == $0", elem(task_id))) .first_async(); if (task == null || ((!)task).task_branch == null || ((!)task).task_branch.length == 0) { return ""; } var project = yield db.query() .where(expr("id == $0", elem(((!)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() .where(expr("task_id == $0 && depends_on_task_id == $1", elem(task_id), elem(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(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() .where(expr("id == $0", elem(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() .where(expr("id == $0", elem(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((!)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() .where(expr("id == $0", elem(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((!)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() .where(expr("id == $0", elem(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((!)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() .where(expr("id == $0", elem(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((!)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() .where(expr("id == $0", elem(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((!)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 get_ready_tasks() throws Error { var queued = yield db.query() .where(expr("status == $0", elem(STATUS_QUEUED))) .materialise_async(); var all_tasks = yield db.query().materialise_async(); var dependencies = yield db.query().materialise_async(); var rows = new Vector(); 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 list_comments(int64 task_id) throws Error { return yield db.query() .where(expr("task_id == $0", elem(task_id))) .order_by(expr("id")) .materialise_async(); } /** Blocks on a task, oldest first. */ public async ImmutableLot list_blocks(int64 task_id) throws Error { return yield db.query() .where(expr("task_id == $0", elem(task_id))) .order_by(expr("id")) .materialise_async(); } /** Every dependency edge in the graph (for queue annotations). */ public async ImmutableLot list_all_dependencies() throws Error { return yield db.query().materialise_async(); } /** All commands recorded against a session, oldest first. */ public async ImmutableLot list_commands(int64 session_id) throws Error { return yield db.query() .where(expr("session_id == $0", elem(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() .where(expr("session_id == $0", elem(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() .where(expr("task_id == $0", elem(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() .where(expr("task_id == $0", elem(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); } } private async void free_saddles_running(int64 task_id) throws Error { var saddles = yield db.query() .where(expr("current_task_id == $0", elem(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); } } } }