using Invercargill; using Invercargill.DataStructures; using Invercargill.Expressions; using InvercargillSql; using InvercargillSql.Orm; using Inversion; using ClankerSaddle.Data; namespace ClankerSaddle.Services { /** Valid task statuses (the `tasks.status` wire values). */ public const string STATUS_DRAFT = "draft"; public const string STATUS_QUEUED = "queued"; public const string STATUS_ASSIGNED = "assigned"; public const string STATUS_WORKING = "working"; public const string STATUS_BLOCKED = "blocked"; public const string STATUS_REVIEW_REQUESTED = "review_requested"; public const string STATUS_SUSPENDED = "suspended"; public const string STATUS_DONE = "done"; public const string STATUS_FAILED = "failed"; /** Valid {@link TaskBlock.block_type} values. */ public const string BLOCK_QUESTION = "question"; public const string BLOCK_RFC = "rfc"; public const string BLOCK_REVIEW = "review"; public const string BLOCK_PLAN = "plan"; /** * The goal agent's completion report (plan 15 C1): raised on the * goal-run task by {@link GoalService.submit_completion} — the * goal's review kinds are plan and completion. */ public const string BLOCK_COMPLETION = "completion"; /** * The agent-celebration entry type (`saddle yeehaw `): the * daemon pushes it as a round-results entry carrying the message in * the output field, so it stores like any other `session_commands` * row — nothing new is persisted for the badge or the card. */ public const string ENTRY_YEEHAW = "yeehaw"; /** The task page's yeehaw card cap (newest entries kept). */ public const int YEEHAW_CARD_MAX = 3; /** * The stable verdict wordings (plan 11): they are stored as the * block answers AND ride the resuming allocation's * `resume_notice` verbatim, so agent and human always read the * same text. */ public const string NOTICE_PLAN_APPROVED = "Plan approved. Proceed with implementation."; public const string NOTICE_PLAN_FEEDBACK = "Plan feedback received:"; public const string NOTICE_REVIEW_APPROVED = "Review approved. Continue with the task."; public const string NOTICE_REVIEW_FEEDBACK = "Review feedback received:"; /** * A task's computed presentation state (stored status + unanswered * blocks): `review requested` while a review block is unanswered * (even when the stored status is terminal — the post-mortem's * `done`-with-open-review case), `parked` for the suspended state * (a goal agent between events), else the stored status. Pure * presentation — the stored statuses and the dispatcher are * untouched. */ public class TaskDisplay : Object { public string status { get; set; default = ""; } public bool terminal { get; set; } public bool failed { get; set; } public static TaskDisplay compute(TaskEntity task, bool review_pending) { var view = new TaskDisplay() { terminal = task.status == STATUS_DONE || task.status == STATUS_FAILED, failed = task.status == STATUS_FAILED }; if (review_pending) { view.status = "review requested"; } else if (task.status == STATUS_SUSPENDED) { view.status = "parked"; } else { view.status = task.status; } return view; } } /** * The full task aggregate loaded by {@link TaskService.get_task}: the * task itself plus its comments, blocks, progress history and * session history. */ public class TaskDetail : Object { public TaskEntity task { get; set; } public ImmutableLot comments { get; set; } public ImmutableLot blocks { get; set; } public ImmutableLot progress { get; set; } public ImmutableLot 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 latest { get; set; default = new Vector(); } 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(); private Connection connection = inject(); private StatePulse pulse = inject(); private PushService push = inject(); /** * The goal's task-terminal hook (plan 12): a worker task under * a goal records a supervision-narrative event — the parked * goal agent's resume machinery coalesces on these rows (see * {@link GoalService.resume_due}). Inline (not via * GoalService) because GoalService depends on this service. */ private async void note_goal_task_event(TaskEntity task, string outcome_word) throws Error { if (task.goal_id <= 0 || task.kind != TASK_KIND_TASK) { return; } string lead = ""; if (task.summary != null && ((!)task.summary).length > 0) { int cut = ((!)task.summary).index_of("\n"); lead = cut >= 0 ? ((!)task.summary).substring(0, cut) : (!)task.summary; } string text = lead.length > 0 ? @"Task #$(task.id) $(task.title) — $(outcome_word): $(lead)" : @"Task #$(task.id) $(task.title) — $(outcome_word)"; yield db.insert_async(new GoalEvent() { goal_id = task.goal_id, kind = task.status == STATUS_DONE ? "task-done" : "task-failed", text = text, task_id = task.id, created_at = new DateTime.now_local() }); } /** * Inserts a task (status 'draft' until its creator enqueues * it). `description`, `task_branch` and `branch_from` may be * null; `branch_from` prefills from the project's default * branch; priority clamps to 1–9 (5 is the default, 1 the * highest). The plan-12 shape: `kind` ("task"/"goal_run"), * `mode` (implement/investigate/review/test, orchestrate for * goal runs), `creator` ("human"/"goal" — enqueueing is * creator-only), `goal_id` (the owning goal) and * `enqueue_after_task_id` (human-created tasks only: the * auto-enqueue predecessor, cycle-checked by the caller). */ public async TaskEntity create_task(int64 project_id, string title, string? description, int64 priority, string? task_branch = null, string? branch_from = null, string status = STATUS_DRAFT, string kind = TASK_KIND_TASK, string mode = "implement", string creator = CREATOR_HUMAN, int64 goal_id = 0, int64 enqueue_after_task_id = 0) throws Error { string resolved_branch_from = branch_from ?? ""; if (resolved_branch_from.length == 0) { var project = yield db.query() .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 (goal_id > 0 && kind == TASK_KIND_TASK) { // Goal worker tasks exist only under a live goal. var goal = yield db.query() .where(expr("id == $0", elem(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(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() .where(expr("id == $0", elem(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(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() .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), 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 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()); } } var map = new Dictionary(); foreach (var task in tasks) { map.set(task.id.to_string(), TaskDisplay.compute(task, review_pending.has(task.id.to_string()))); } return map; } /** * The display state of a single task — the detail page's path, * computed from its already-loaded aggregate (no extra queries). */ public static TaskDisplay display_of_detail(TaskDetail detail) { bool review_pending = false; foreach (var block in detail.blocks) { if (block.block_type == BLOCK_REVIEW && block.answered_at == null) { review_pending = true; } } return TaskDisplay.compute(detail.task, review_pending); } public async ImmutableLot 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 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() .where(expr("id == $0", elem(id))) .first_async(); if (task == null) { return null; } bool changed = false; bool draft = ((!)task).status == STATUS_DRAFT; if (draft) { if (title != null && title.length > 0 && ((!)task).title != title) { ((!)task).title = title; changed = true; } if (description != null) { string? next_description = description.length > 0 ? description : null; if (((!)task).description != next_description) { ((!)task).description = next_description; changed = true; } } if (task_branch != null) { string? next_branch = task_branch.length > 0 ? task_branch : null; if (((!)task).task_branch != next_branch) { ((!)task).task_branch = next_branch; changed = true; } } if (branch_from != null) { string? next_from = branch_from.length > 0 ? branch_from : null; if (((!)task).branch_from != next_from) { ((!)task).branch_from = next_from; changed = true; } } if (mode != null && mode.length > 0) { bool mode_valid = false; foreach (string candidate in TASK_MODES) { if (mode == candidate) { mode_valid = true; break; } } if (!mode_valid) { throw new Error(Quark.from_string("clanker-saddle"), 79, @"Unknown mode '$(mode)' — valid modes: " + string.joinv(", ", TASK_MODES)); } if (((!)task).mode != mode) { ((!)task).mode = mode; changed = true; } } // Human-created tasks only: the auto-enqueue // predecessor is settable/clearable while draft, and // setting it validates the chain (no cycles). if (enqueue_after_task_id != null && ((!)task).creator == CREATOR_HUMAN && ((!)task).kind == TASK_KIND_TASK) { int64 next = (!)enqueue_after_task_id; if (next < 0) { // The -1 sentinel clears the predecessor. next = 0; } if (next > 0) { yield validate_enqueue_after_cycle(id, next); } if (((!)task).enqueue_after_task_id != next) { ((!)task).enqueue_after_task_id = next; changed = true; } } } // Priority stays editable in every non-terminal state. if ((((!)task).status != STATUS_DONE && ((!)task).status != STATUS_FAILED) && priority != null) { int64 clamped = ((!)priority).clamp(1, 9); if (((!)task).priority != clamped) { ((!)task).priority = clamped; changed = true; } } yield db.update_async(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() .where(expr("id == $0", elem(id))) .first_async(); if (task == null) { return null; } if (((!)task).status != status) { ((!)task).status = status; if (status == STATUS_DONE && ((!)task).completed_at == null) { ((!)task).completed_at = new DateTime.now_local(); } yield db.update_async(task); yield pulse.bump(id, @"task $(status)"); } return task; } /** Every task under `goal_id` (both kinds), oldest first. */ public async ImmutableLot list_tasks_for_goal(int64 goal_id) throws Error { return yield db.query() .where(expr("goal_id == $0", elem(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() .where(expr("id == $0", elem(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((!)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(); 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() .where(expr("id == $0", elem(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() .where(expr("enqueue_after_task_id == $0 && status == $1", elem(predecessor.id), elem(STATUS_DRAFT))) .materialise_async()) { try { yield enqueue_task(dependent.id, dependent.creator); } catch (Error e) { warning("[ClankerSaddle] Auto-enqueue of task %lld after %lld failed: %s", dependent.id, predecessor.id, e.message); } } } /** * The delete the goal cascade uses on its own draft tasks: * every attached row dies with the task and no dependents * guard applies (the whole goal is going away together). */ public async void delete_task_quietly(int64 id) throws Error { var task = yield db.query() .where(expr("id == $0", elem(id))) .first_async(); if (task == null) { return; } foreach (var link in yield db.query() .where(expr("task_id == $0", elem(id))) .materialise_async()) { yield db.delete_async(link); } foreach (var link in yield db.query() .where(expr("task_id == $0", elem(id))) .materialise_async()) { yield db.delete_async(link); } foreach (var row in yield db.query() .where(expr("task_id == $0", elem(id))) .materialise_async()) { yield db.delete_async(row); } foreach (var row in yield db.query() .where(expr("task_id == $0", elem(id))) .materialise_async()) { yield db.delete_async(row); } foreach (var row in yield db.query() .where(expr("task_id == $0", elem(id))) .materialise_async()) { yield db.delete_async(row); } yield clear_enqueue_after_references(id); yield db.delete_async((!)task); yield pulse.bump(id, "task deleted"); } /** * The human-only Delete action: removes a DRAFT task. Any * auto-enqueue-after predecessor references pointing at it are * cleared first (a dangling predecessor would strand its * dependents in not-ready forever). Attached rows (comments, * blocks, progress, rules, capabilities) die with the task. */ public async void delete_task(int64 id) throws Error { var task = yield db.query() .where(expr("id == $0", elem(id))) .first_async(); if (task == null) { return; } if (((!)task).status != STATUS_DRAFT) { throw new Error(Quark.from_string("clanker-saddle"), 53, "Only draft tasks can be deleted (cancel or retry handles the rest)"); } foreach (var link in yield db.query() .where(expr("task_id == $0", elem(id))) .materialise_async()) { yield db.delete_async(link); } foreach (var link in yield db.query() .where(expr("task_id == $0", elem(id))) .materialise_async()) { yield db.delete_async(link); } foreach (var row in yield db.query() .where(expr("task_id == $0", elem(id))) .materialise_async()) { yield db.delete_async(row); } foreach (var row in yield db.query() .where(expr("task_id == $0", elem(id))) .materialise_async()) { yield db.delete_async(row); } foreach (var row in yield db.query() .where(expr("task_id == $0", elem(id))) .materialise_async()) { yield db.delete_async(row); } yield clear_enqueue_after_references(id); yield db.delete_async((!)task); yield pulse.bump(id, "task deleted"); } /** * Clears `enqueue_after_task_id` references pointing at * `id`: a deleted predecessor must not strand its dependents * (their readiness check would wait forever on a task that no * longer exists). */ private async void clear_enqueue_after_references(int64 id) throws Error { foreach (var dependent in yield db.query() .where(expr("enqueue_after_task_id == $0", elem(id))) .materialise_async()) { dependent.enqueue_after_task_id = 0; yield db.update_async(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(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(); } /** * True when any of the task's sessions carries a `yeehaw` entry — * the badge's presence check, one cheap EXISTS over the existing * `session_commands` rows (nothing new is stored for the badge). */ public async bool has_yeehaw(int64 task_id) throws Error { var command = connection.create_command( "SELECT EXISTS(SELECT 1 FROM session_commands sc" + " JOIN sessions s ON s.id = sc.session_id" + " WHERE s.task_id = :task AND sc.type = :kind) AS hit"); command.with_parameter("task", elem(task_id)); command.with_parameter("kind", elem(ENTRY_YEEHAW)); var hit = yield command.execute_scalar_async(); int64? value = 0; return hit != null && hit.try_get_as(out value) && ((!)value) != 0; } /** * Every task id whose sessions carry a `yeehaw` entry (stringly * keyed) — the batch presence check the task list and the * dashboard's task references badge from, one DISTINCT query * instead of one EXISTS per row. */ public async Invercargill.DataStructures.HashSet yeehaw_task_ids() throws Error { var ids = new Invercargill.DataStructures.HashSet(); 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("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(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("task", elem(task_id)); total_command.with_parameter("kind", elem(ENTRY_YEEHAW)); var total = yield total_command.execute_scalar_async(); int64? total_value = 0; if (total != null) { total.try_get_as(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("task", elem(task_id)); rows_command.with_parameter("kind", elem(ENTRY_YEEHAW)); rows_command.with_parameter("cap", elem(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(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(out value) ? value : null; } /** Appends a comment; `author` is 'agent' 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, 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(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"); // Web Push (plan 11 task E): every raised block notifies all // push-subscribed users globally — the goal agent's mirrored // blocks stay quiet (the goal-level block owns the // notification with the goal's title and URL). if (!quiet && task != null) { push.notify_global(PushService.block_notification( ((!)task).title, task_id, block.block_type, block.question)); } return block; } /** * Records a review request (`saddle review`): a review-type block * carrying the summary markdown, with the task flipped to * 'review_requested' while it waits for the human verdict. */ public async TaskBlock request_review(int64 task_id, string summary) throws Error { return yield add_block(task_id, "Review requested", BLOCK_REVIEW, null, summary.length > 0 ? summary : null); } /** * Records a human's answer to a block. With an open session the * task stays 'blocked'/'review_requested' — the dispatcher resumes * the paused workspace and hands the answer over; without one the * task flips straight back to 'working'. */ public async void answer_block(int64 block_id, string answer, string? answered_by = null) throws Error { var block = yield db.query() .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(); ((!)block).answered_by = answered_by; 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; } /** * Task ids currently `assigned`/`working` whose open session sits * on an offline or fenced saddle — the queue marks them * "saddle unreachable — requeueing" while recovery runs. */ public async Invercargill.DataStructures.HashSet unreachable_task_ids() throws Error { var ids = new Invercargill.DataStructures.HashSet(); foreach (var status in new string[] { STATUS_ASSIGNED, STATUS_WORKING }) { foreach (var task in yield db.query() .where(expr("status == $0", elem(status))) .materialise_async()) { var session = yield open_session_for_task(task.id); if (session == null) { continue; } var saddle = yield db.query() .where(expr("id == $0", elem(((!)session).saddle_id))) .first_async(); if (saddle != null && (((!)saddle).status == "offline" || ((!)saddle).status == "recovering")) { ids.add(task.id.to_string()); } } } return ids; } /** * The resume_notice wording for an answered block (the plan-11 * stable verdicts): review and plan answers are already stored * in the stable wording and pass through verbatim; question * replies are formatted the way the daemon used to format * delivered answers ("Human selected: …" for option matches, * "Human answered (other): …" otherwise); RFC/plain answers * carry the "Human answer: …" prefix. */ public static string verdict_notice_for(TaskBlock block) { string answer = block.answer ?? ""; if (block.block_type == BLOCK_REVIEW || block.block_type == BLOCK_PLAN) { return answer; } if (block.block_type == BLOCK_QUESTION) { foreach (var option in parse_block_options(block.options)) { if (option == answer) { return "Human selected: " + answer; } } return "Human answered (other): " + answer; } return "Human answer: " + answer; } /** The stored options JSON (a string array) back into a Vala array. */ public static string[] parse_block_options(string? options_json) { if (options_json == null || ((!)options_json).length == 0) { return new string[0]; } try { var parser = new Json.Parser(); parser.load_from_data((!)options_json); var root = parser.get_root(); if (root == null || root.get_node_type() != Json.NodeType.ARRAY) { return new string[0]; } var values = new Vector(); foreach (var element in root.get_array().get_elements()) { if (element.get_node_type() == Json.NodeType.VALUE && element.get_string() != null) { values.add(element.get_string()); } } return values.to_array(); } catch (Error e) { return new string[0]; } } /** * Marks a task done: records `summary` and `working_branch`, stamps * `completed_at`, closes any open session (outcome `outcome`) and * frees the saddle that was running it. Terminal transitions * project ONCE (plan 13 B2): a task already done or failed keeps * its first outcome — a redelivered push or a second terminal * entry with a fresh id re-runs no bookkeeping, no comment, no * event (retry_task is the one path that un-terminals). */ public async void complete_task(int64 task_id, string? summary, string? working_branch, string outcome) throws Error { var task = yield db.query() .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).status == STATUS_DONE || ((!)task).status == STATUS_FAILED) { return; } ((!)task).status = STATUS_DONE; ((!)task).completed_at = new DateTime.now_local(); if (summary != null && summary.length > 0) { ((!)task).summary = summary; } if (working_branch != null && working_branch.length > 0) { ((!)task).working_branch = working_branch; } yield db.update_async((!)task); yield close_open_sessions(task_id, outcome); yield free_saddles_running(task_id); yield pulse.bump(task_id, "task completed"); // Web Push (plan 11 task E): completion notifies the users // holding a "Notify me" pref for this task. push.notify_opted_in(task_id, PushService.lifecycle_notification( ((!)task).title, task_id, "completed", summary)); // Plan 12: the goal's narrative hears the terminal event // (a parked goal agent resumes on it), and a done // predecessor auto-enqueues its dependents. yield note_goal_task_event(((!)task), "completed"); yield enqueue_dependents_of(((!)task)); } /** * Cancels a task: status 'failed' with a cancellation summary. Any * open session is closed with outcome 'stopped' and its saddle * freed. Draft tasks are refused — they are deleted, not * cancelled (nothing ever ran). Idempotent on terminal tasks * (plan 13 B2): the first outcome stands. */ public async void cancel_task(int64 task_id) throws Error { var task = yield db.query() .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)"); } if (((!)task).status == STATUS_DONE || ((!)task).status == STATUS_FAILED) { return; } if (((!)task).status == STATUS_DRAFT) { throw new Error(Quark.from_string("clanker-saddle"), 55, "Draft tasks are deleted, not cancelled"); } ((!)task).status = STATUS_FAILED; ((!)task).completed_at = new DateTime.now_local(); if (((!)task).summary == null || ((!)task).summary.length == 0) { ((!)task).summary = "Cancelled by user"; } yield db.update_async((!)task); yield close_open_sessions(task_id, "stopped"); yield free_saddles_running(task_id); yield pulse.bump(task_id, "task cancelled"); yield note_goal_task_event(((!)task), "cancelled"); } /** * Records a saddle-reported failure: status 'failed' with the * daemon's message as summary, the explanation stored again as a * agent comment (so the thread carries it), the open session * closed with outcome 'failed' and its saddle freed. Terminal * transitions project ONCE (plan 13 B2): the live failure's * honest burst — a dozen-plus distinct `saddle failure` entry * ids in one session — still fails the task exactly once, with * one comment and one goal event. */ public async void fail_task(int64 task_id, string? summary) throws Error { var task = yield db.query() .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)"); } if (((!)task).status == STATUS_DONE || ((!)task).status == STATUS_FAILED) { return; } ((!)task).status = STATUS_FAILED; ((!)task).completed_at = new DateTime.now_local(); if (summary != null && summary.length > 0) { ((!)task).summary = summary; yield add_comment(task_id, "agent", summary); } else if (((!)task).summary == null || ((!)task).summary.length == 0) { ((!)task).summary = "Saddle reported a failure"; } yield db.update_async((!)task); yield close_open_sessions(task_id, "failed"); yield free_saddles_running(task_id); yield pulse.bump(task_id, "task failed"); // Web Push (plan 11 task E): an agent-reported failure // notifies the opted-in users (a human cancellation does // not — the cancellator already knows). push.notify_opted_in(task_id, PushService.lifecycle_notification( ((!)task).title, task_id, "failed", summary)); yield note_goal_task_event(((!)task), "failed"); } /** * Parks a suspended task (`saddle suspend`): stores the status * and stamps `suspended_at` once per parking window, so the * resume trigger only sees goal events that landed after the * suspension (a parked goal agent between supervision rounds). */ public async void suspend_task(int64 task_id) throws Error { var task = yield db.query() .where(expr("id == $0", elem(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((!)task); yield pulse.bump(task_id, "task suspended"); } /** /** Withdraws the suspension stamp after a successful resume. */ public async void clear_suspension(int64 task_id) throws Error { var task = yield db.query() .where(expr("id == $0", elem(task_id))) .first_async(); if (task == null || ((!)task).suspended_at == null) { return; } ((!)task).suspended_at = null; yield db.update_async((!)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() .where(expr("id == $0", elem(task_id))) .first_async(); if (me == null) { return false; } DateTime? park = ((!)me).suspended_at; if (park == null) { foreach (var block in yield db.query() .where(expr("task_id == $0", elem(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() .where(expr("task_id == $0", elem(task_id))) .materialise_async()) { foreach (var entry in yield db.query() .where(expr("session_id == $0", elem(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() .where(expr("task_id == $0", elem(sibling.id))) .materialise_async()) { if (session.started_at != null && session.started_at.compare((!)park) > 0) { return true; } } foreach (var progress in yield db.query() .where(expr("task_id == $0", elem(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 sibling_tasks(TaskEntity task) throws Error { var rows = new Vector(); var seen = new Invercargill.DataStructures.HashSet(); if (task.goal_id > 0) { foreach (var sibling in yield db.query() .where(expr("goal_id == $0", elem(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() .where(expr("project_id == $0 && task_branch == $1", elem(task.project_id), elem(((!)task).task_branch))) .materialise_async()) { if (sibling.id != task.id && seen.add(sibling.id.to_string())) { rows.add(sibling); } } } return rows.to_immutable_buffer(); } /** * Returns a non-terminal task to the queue: closes its open * sessions (outcome 'stopped') and frees the saddle that was * running it. Plan 14 B1: a parked (suspended) task requeues * too — the goal page's requeue-to-resume reaches a goal * agent parked on a dead saddle; the suspension stamp clears * so the next dispatch starts a fresh parking window. */ public async void requeue_task(int64 task_id) throws Error { var task = yield db.query() .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_SUSPENDED)) { ((!)task).status = STATUS_QUEUED; if (((!)task).suspended_at != null) { ((!)task).suspended_at = null; } 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 an agent comment (so the history keeps * it) before being cleared from the task, `completed_at` is * withdrawn, any open session closes and its saddle frees — the * same bookkeeping as {@link requeue_task}, plus the audit * comment "Task retried by user". */ public async void retry_task(int64 task_id) throws Error { var task = yield db.query() .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, "agent", (!)(((!)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, "agent", "Task retried by user"); yield pulse.bump(task_id, "task retried"); } /** * Queued tasks whose auto-enqueue predecessor (if any) is * done, ordered by priority (1 first) then creation time — * the dispatcher's candidate list. */ public async ImmutableLot 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 rows = new Vector(); foreach (var task in queued) { bool ready = true; if (task.enqueue_after_task_id > 0) { // The auto-enqueue predecessor also gates dispatch: // a dependent that reached the queue some other way // (a predecessor set after a manual enqueue) still // waits for it to finish. ready = false; foreach (var other in all_tasks) { if (other.id == task.enqueue_after_task_id) { ready = other.status == STATUS_DONE; break; } } } if (ready) { rows.add(task); } } var sorted = rows.sort((a, b) => { if (a.priority != b.priority) { return (int)(a.priority - b.priority); } int64 a_created = a.created_at != null ? ((!)a.created_at).to_unix() : 0; int64 b_created = b.created_at != null ? ((!)b.created_at).to_unix() : 0; return (int)(a_created - b_created); }); return sorted.to_immutable_buffer(); } /** Comments on a task, oldest first. */ public async ImmutableLot 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(); } /** 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 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 { // One conditional UPDATE, atomic against a concurrent ask // allocation claiming a new task for the saddle between a // read and a full-row write. var command = connection.create_command( "UPDATE saddles SET status = 'idle', current_task_id = 0, allocated_at = NULL" + " WHERE current_task_id = :task"); command.with_parameter("task", elem(task_id)); yield command.execute_non_query_async(); } } }