using Invercargill; using Invercargill.DataStructures; using Invercargill.Expressions; using InvercargillSql; using InvercargillSql.Orm; using Inversion; using ClankerSaddle.Data; namespace ClankerSaddle.Services { /** Valid task statuses (the `tasks.status` wire values). */ public const string STATUS_DRAFT = "draft"; public const string STATUS_QUEUED = "queued"; public const string STATUS_ASSIGNED = "assigned"; public const string STATUS_WORKING = "working"; public const string STATUS_BLOCKED = "blocked"; public const string STATUS_REVIEW_REQUESTED = "review_requested"; public const string STATUS_SUSPENDED = "suspended"; public const string STATUS_DONE = "done"; public const string STATUS_FAILED = "failed"; /** Valid {@link TaskBlock.block_type} values. */ public const string BLOCK_QUESTION = "question"; public const string BLOCK_RFC = "rfc"; public const string BLOCK_REVIEW = "review"; public const string BLOCK_PLAN = "plan"; /** * The goal agent's completion report (plan 15 C1): raised on the * goal-run task by {@link GoalService.submit_completion} — the * goal's review kinds are plan and completion. */ public const string BLOCK_COMPLETION = "completion"; /** * The system-event comment author (plan 15 C2): harness-authored * bookkeeping rows (aborts, requeues, retries, session * restorations) — never agent content. */ public const string COMMENT_AUTHOR_SYSTEM = "system"; public const string COMMENT_AUTHOR_AGENT = "agent"; /** * Plan 16 S5 (U2): MECHANICAL harness bookkeeping that every * user-facing timeline filters out — the session-restore footnote * and the park-until notice (both remain in the raw terminal log, * which keeps everything). Matched by substring so the daemon's * [harness]-prefixed wordings and the web's own phrasing all stay * covered; substantive system rows (aborts, requeues, retries) * never match. */ public static bool comment_is_mechanical(string body) { return body.contains("session context restored") || body.contains("parked until"); } /** * The stable verdict wordings (plan 11): they are stored as the * block answers AND ride the resuming allocation's * `resume_notice` verbatim, so agent and human always read the * same text. */ public const string NOTICE_PLAN_APPROVED = "Plan approved. Proceed with implementation."; public const string NOTICE_PLAN_FEEDBACK = "Plan feedback received:"; public const string NOTICE_REVIEW_APPROVED = "Review approved. Continue with the task."; public const string NOTICE_REVIEW_FEEDBACK = "Review feedback received:"; /** The display label of a goal-run parked on a review-family block (U3). */ public const string DISPLAY_AWAITING_REVIEW = "awaiting your review"; /** * A task's computed presentation state (stored status + unanswered * blocks): `review requested` while a review block is unanswered * (even when the stored status is terminal — the post-mortem's * `done`-with-open-review case), `parked` for the suspended state * (a goal agent between events), else the stored status. Plan 16 * S5 (U3): a GOAL-RUN task parked on a review-family block (plan * or completion review) reads "awaiting your review" — agreeing * with its goal's label, never a bare "blocked". Pure * presentation — the stored statuses and the dispatcher are * untouched. */ public class TaskDisplay : Object { public string status { get; set; default = ""; } public bool terminal { get; set; } public bool failed { get; set; } public static TaskDisplay compute(TaskEntity task, bool review_pending, bool goal_review_pending = false) { var view = new TaskDisplay() { terminal = task.status == STATUS_DONE || task.status == STATUS_FAILED, failed = task.status == STATUS_FAILED }; if (goal_review_pending && task.kind == TASK_KIND_GOAL_RUN) { view.status = DISPLAY_AWAITING_REVIEW; } else if (review_pending) { view.status = "review requested"; } else if (task.status == STATUS_SUSPENDED) { view.status = "parked"; } else { view.status = task.status; } return view; } } /** * One LLM-reported progress line on a task — the ledger's progress * events projected back into the feed shape the pages read (id is * the event id, summary the payload's text). */ public class TaskProgress : Object { public int64 id { get; set; } public int64 task_id { get; set; } public string summary { get; set; default = ""; } public DateTime? created_at { get; set; default = null; } } /** * The full task aggregate loaded by {@link TaskService.get_task}: the * task itself plus its comments, blocks, progress history and * session history. */ public class TaskDetail : Object { public TaskEntity task { get; set; } public ImmutableLot comments { get; set; } public ImmutableLot blocks { get; set; } public ImmutableLot progress { get; set; } public ImmutableLot sessions { get; set; } } /** * Task storage and lifecycle: creation, enqueue-after-aware * readiness, agent interaction primitives (comments, blocks, * progress, reviews) and completion. Saddle bookkeeping rows are * kept consistent on every terminal transition. The plan-12 task * graph is goal-shaped: no parents, no dependencies, no plans — * goals own their tasks. */ public class TaskService : Object { private OrmSession db = inject(); private Connection connection = inject(); private StatePulse pulse = inject(); private PushService push = inject(); private EventService events = inject(); private SaddlePollState poll_state = inject(); // Goal 28 task F: answers whose owner is BUSY ride the // singleton delivery hub (this service is request-scoped and // deliberately holds no SaddleService dependency — the // dispatcher's tick drains the queue through one). private AnswerDeliveryHub answer_hub = inject(); /** * Inserts a task (status 'draft' until its creator enqueues * it). `description`, `task_branch` and `branch_from` may be * null; `branch_from` prefills from the project's default * branch; priority clamps to 1–9 (5 is the default, 1 the * highest). The plan-12 shape: `kind` ("task"/"goal_run"), * `mode` (implement/investigate/review/test, orchestrate for * goal runs), `creator` ("human"/"goal" — enqueueing is * creator-only), `goal_id` (the owning goal) and * `enqueue_after_task_id` (human-created tasks only: the * auto-enqueue predecessor, cycle-checked by the caller). */ public async TaskEntity create_task(int64 project_id, string title, string? description, int64 priority, string? task_branch = null, string? branch_from = null, string status = STATUS_DRAFT, string kind = TASK_KIND_TASK, string mode = "implement", string creator = CREATOR_HUMAN, int64 goal_id = 0, int64 enqueue_after_task_id = 0) throws Error { // Goal 8 task 4: project_id 0 is a STANDALONE task — its // own ephemeral workspace, no project row anywhere. if (project_id < 0) { throw new Error(Quark.from_string("clanker-saddle"), 76, @"Invalid project id $(project_id)"); } string resolved_branch_from = branch_from ?? ""; if (resolved_branch_from.length == 0 && project_id > 0) { var project = yield db.query() .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"); yield append_state_change(task.id, "", status, "Task created", creator == CREATOR_HUMAN ? EVENT_ACTOR_USER : EVENT_ACTOR_MODEL); return task; } /** * Appends a task-scoped state_change event carrying `from`/`to` * plus a human-readable line the timeline and narrative read * back verbatim. */ private async void append_state_change(int64 task_id, string from, string to, string text, string actor) throws Error { var payload = new Json.Object(); payload.set_string_member("from", from); payload.set_string_member("to", to); payload.set_string_member("text", text); yield events.append(EVENT_STATE_CHANGE, actor, EVENT_SCOPE_TASK, task_id, payload); } /** * Loads the full aggregate for `id`: the task, its comments, * blocks, progress and sessions. Returns null when the task * does not exist. */ public async TaskDetail? get_task(int64 id) throws Error { var task = yield db.query() .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_desc(expr("id")) .materialise_async() }; } /** * Computed display states for a batch of tasks (all lists share * this), keyed by task id string. Unanswered review-family * blocks load in ONE batched query over the wanted tasks, not * per task. Plan 16 S5 (U3): a goal-run task with a pending * review-family block (plan/completion/review) reads "awaiting * your review" — the run-task card agrees with its goal. */ public async Dictionary display_map_for( Lot tasks) throws Error { var wanted = new Invercargill.DataStructures.HashSet(); var goal_runs = new Invercargill.DataStructures.HashSet(); foreach (var task in tasks) { wanted.add(task.id.to_string()); if (task.kind == TASK_KIND_GOAL_RUN) { goal_runs.add(task.id.to_string()); } } var review_pending = new Invercargill.DataStructures.HashSet(); var goal_review_pending = new Invercargill.DataStructures.HashSet(); if (wanted.length > 0) { foreach (var block in yield db.query() .where(id_in("task_id", wanted)) .materialise_async()) { if (block.answered_at != null || !wanted.has(block.task_id.to_string())) { continue; } if (block.block_type == BLOCK_REVIEW) { review_pending.add(block.task_id.to_string()); } if (block.block_type == BLOCK_REVIEW || block.block_type == BLOCK_COMPLETION || block.block_type == BLOCK_PLAN) { if (goal_runs.has(block.task_id.to_string())) { goal_review_pending.add(block.task_id.to_string()); } } } } var map = new Dictionary(); foreach (var task in tasks) { map.set(task.id.to_string(), TaskDisplay.compute(task, review_pending.has(task.id.to_string()), goal_review_pending.has(task.id.to_string()))); } return map; } /** * The display state of a single task — the detail page's path, * computed from its already-loaded aggregate (no extra queries). */ public static TaskDisplay display_of_detail(TaskDetail detail) { bool review_pending = false; bool goal_review_pending = false; foreach (var block in detail.blocks) { if (block.answered_at != null) { continue; } if (block.block_type == BLOCK_REVIEW) { review_pending = true; } if (detail.task.kind == TASK_KIND_GOAL_RUN && (block.block_type == BLOCK_REVIEW || block.block_type == BLOCK_COMPLETION || block.block_type == BLOCK_PLAN)) { goal_review_pending = true; } } return TaskDisplay.compute(detail.task, review_pending, goal_review_pending); } /** * Lists tasks, optionally filtered by status and/or project, * newest first (creation descending, id as the same-second * tiebreak) — the list-page consistency sweep: every table * leads with the most recent row. Priority stays a visible * column, no longer a sort key; the dispatcher's candidate * order ({@link get_ready_tasks}) keeps priority-first. */ public async ImmutableLot list_tasks(string? status, int64? project_id) throws Error { var query = db.query(); if (status != null && status.length > 0) { query.where(expr("status == $0", elem(status))); } if (project_id != null) { query.where(expr("project_id == $0", elem(project_id))); } return yield query .order_by_desc(expr("created_at")) .order_by_desc(expr("id")) .materialise_async(); } /** * Every standalone human-workable task row (kind task, no * goal), any status, ordered like {@link list_tasks} * (newest first) — the auto-enqueue-after dropdowns' option * list, queried for exactly the rows they show instead of * every task in the table. */ public async ImmutableLot list_standalone_tasks() throws Error { return yield db.query() .where(expr("kind == $0 && goal_id == $1", elem(TASK_KIND_TASK), elem(0))) .order_by_desc(expr("created_at")) .order_by_desc(expr("id")) .materialise_async(); } /** * Saves the human-editable fields of the task identified by `id` * under the draft-lifecycle edit matrix: in DRAFT all fields * except status are editable (title, description/mandate, * priority, branch fields, the plan-12 work mode and — * human-created tasks only — the auto-enqueue predecessor, * cycle-validated); in non-draft non-terminal * states only priority applies; in terminal states nothing * does. Fields not editable in the current state are silently * left unchanged (null arguments always mean "leave * unchanged"; an empty branch field clears it; an * `enqueue_after_task_id` of -1 CLEARS the predecessor). * Returns null when the task does not exist. The pulse only * fires when a value actually moved, so periodic status * refreshes do not spam live pages. */ public async TaskEntity? edit_task(int64 id, string? title, string? description, int64? priority, string? task_branch, string? branch_from, string? mode = null, int64? enqueue_after_task_id = null) throws Error { var task = yield db.query() .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) { if (!is_valid_mode(mode)) { throw new Error(Quark.from_string("clanker-saddle"), 79, @"Unknown mode '$(mode)' — valid modes: " + string.joinv(", ", TASK_MODES)); } if (((!)task).mode != mode) { ((!)task).mode = mode; changed = true; } } // Human-created tasks only: the auto-enqueue // predecessor is settable/clearable while draft, and // setting it validates the chain (no cycles). if (enqueue_after_task_id != null && ((!)task).creator == CREATOR_HUMAN && ((!)task).kind == TASK_KIND_TASK) { int64 next = (!)enqueue_after_task_id; if (next < 0) { // The -1 sentinel clears the predecessor. next = 0; } if (next > 0) { yield validate_enqueue_after_cycle(id, next); } if (((!)task).enqueue_after_task_id != next) { ((!)task).enqueue_after_task_id = next; changed = true; } } } // Priority stays editable in every non-terminal state. if ((((!)task).status != STATUS_DONE && ((!)task).status != STATUS_FAILED) && priority != null) { int64 clamped = ((!)priority).clamp(1, 9); if (((!)task).priority != clamped) { ((!)task).priority = clamped; changed = true; } } yield db.update_async(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) { string from = ((!)task).status; if (status != STATUS_DONE && status != STATUS_FAILED) { // Terminal states are sticky (plan 13 B2's // invariant): a NON-terminal transition must never // downgrade a done/failed task. The entity writes // below interleave with completions on the main // loop (the terminal checks in the callers have // already yielded by then — a late 'working' push // racing a 'done' push used to resurrect the task // as working AFTER its session closed done), so // the guard lives in the UPDATE's WHERE: atomic. // The row is re-read afterwards so the caller sees // the truth (terminal included), never a stale // entity it might write back. var guard = connection.create_command( "UPDATE tasks SET status = :status WHERE id = :id" + " AND status NOT IN ('done', 'failed')"); guard.with_parameter("status", elem(status)); guard.with_parameter("id", elem(id)); yield guard.execute_non_query_async(); task = yield db.query() .where(expr("id == $0", elem(id))) .first_async(); if (task == null) { return null; } } else { ((!)task).status = status; if (status == STATUS_DONE && ((!)task).completed_at == null) { ((!)task).completed_at = new DateTime.now_local(); } yield db.update_async((!)task); } yield pulse.bump(id, @"task $(status)"); var moved = new Json.Object(); moved.set_string_member("from", from); moved.set_string_member("to", status); moved.set_string_member("text", @"Status → $(status)"); yield events.append(EVENT_STATE_CHANGE, EVENT_ACTOR_SYSTEM, EVENT_SCOPE_TASK, id, moved); } return task; } /** Every task under `goal_id` (both kinds), newest first. */ public async ImmutableLot list_tasks_for_goal(int64 goal_id) throws Error { return yield db.query() .where(expr("goal_id == $0", elem(goal_id))) .order_by_desc(expr("id")) .materialise_async(); } /** * The Enqueue action (creator-only, plan 12): moves a draft * task to the queue. Humans enqueue theirs, goal agents theirs * — an actor that is not the task's creator is refused (the * goal agent sequences its own tasks). A pending * auto-enqueue predecessor does not refuse the enqueue — the * dispatcher's readiness check holds dispatch back until it * finishes. The internal auto-enqueue-after pass calls this as * the dependent's creator (the creator's pre-authorisation). */ public async void enqueue_task(int64 id, string actor = CREATOR_HUMAN) throws Error { var task = yield db.query() .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"); yield append_state_change(id, STATUS_DRAFT, STATUS_QUEUED, "Task enqueued", actor == CREATOR_HUMAN ? EVENT_ACTOR_USER : EVENT_ACTOR_MODEL); } /** * The cycle check for `enqueue_after_task_id` (human-created * tasks only, settable/clearable while the dependent is * draft): following the predecessor chain from * `predecessor_id` must never reach `task_id` — a cycle would * strand every task on it in draft forever. */ public async void validate_enqueue_after_cycle(int64 task_id, int64 predecessor_id) throws Error { if (predecessor_id <= 0) { return; } if (predecessor_id == task_id) { throw new Error(Quark.from_string("clanker-saddle"), 77, "A task cannot enqueue after itself"); } var seen = new Invercargill.DataStructures.HashSet(); 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 row teardown every task deletion runs (the shared core * of the human Delete action and the goal cascade's quiet * delete): attached rules, capabilities, progress, comments * and blocks die with the task, predecessor references clear, * the row goes and the pulse fires. */ private async void delete_task_rows(TaskEntity task) throws Error { foreach (var link in yield db.query() .where(expr("task_id == $0", elem(task.id))) .materialise_async()) { yield db.delete_async(link); } foreach (var link in yield db.query() .where(expr("task_id == $0", elem(task.id))) .materialise_async()) { yield db.delete_async(link); } foreach (var row in yield db.query() .where(expr("task_id == $0", elem(task.id))) .materialise_async()) { yield db.delete_async(row); } foreach (var row in yield db.query() .where(expr("task_id == $0", elem(task.id))) .materialise_async()) { yield db.delete_async(row); } yield events.delete_for_task(task.id); yield clear_enqueue_after_references(task.id); yield db.delete_async(task); yield pulse.bump(task.id, "task deleted"); } /** * The delete the goal cascade uses on its own draft tasks: * every attached row dies with the task and no dependents * guard applies (the whole goal is going away together). */ public async void delete_task_quietly(int64 id) throws Error { var task = yield db.query() .where(expr("id == $0", elem(id))) .first_async(); if (task == null) { return; } yield delete_task_rows((!)task); } /** * The human-only Delete action: removes a DRAFT task. Any * auto-enqueue-after predecessor references pointing at it are * cleared first (a dangling predecessor would strand its * dependents in not-ready forever). Attached rows (comments, * blocks, progress, rules, capabilities) die with the task. */ public async void delete_task(int64 id) throws Error { var task = yield db.query() .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)"); } yield delete_task_rows((!)task); } /** * Clears `enqueue_after_task_id` references pointing at * `id`: a deleted predecessor must not strand its dependents * (their readiness check would wait forever on a task that no * longer exists). */ private async void clear_enqueue_after_references(int64 id) throws Error { foreach (var dependent in yield db.query() .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 to the task's ledger (the daemon's * progress events land in the same ledger through the chunks * push; the goal-era results-entry path that wrote * `task_progress` rows is gone). */ public async TaskProgress add_progress(int64 task_id, string summary) throws Error { var payload = new Json.Object(); payload.set_string_member("text", summary); var record = yield events.append(EVENT_PROGRESS, EVENT_ACTOR_MODEL, EVENT_SCOPE_TASK, task_id, payload); yield pulse.bump(task_id, "progress"); return new TaskProgress() { id = record.id, task_id = task_id, summary = summary, created_at = record.created_at }; } /** The task's progress history, newest first. */ public async ImmutableLot list_progress(int64 task_id) throws Error { var rows = new Vector(); foreach (var record in yield events.list_for_task(task_id, { EVENT_PROGRESS })) { rows.add(new TaskProgress() { id = record.id, task_id = task_id, summary = EventService.payload_string(record, "text"), created_at = record.created_at }); } return rows.to_immutable_buffer(); } /** * Appends a comment; `author` is 'agent', 'user:' or * 'system' (the harness-authored bookkeeping rows — plan 15 * C2's distinct system-event class, never agent content). The * comment event references the row (the task_comments table * stays; the ledger carries the reference plus the body). */ public async TaskComment add_comment(int64 task_id, string author, string body, string? quote = null) throws Error { var comment = new TaskComment() { task_id = task_id, author = author, body = body, quote = quote != null && quote.strip().length > 0 ? quote.strip() : null, created_at = new DateTime.now_local() }; yield db.insert_async(comment); var payload = new Json.Object(); payload.set_int_member("comment_id", comment.id); payload.set_string_member("body", body); payload.set_string_member("author", author); if (comment.quote != null) { payload.set_string_member("quote", (!)comment.quote); } yield events.append(EVENT_COMMENT, author == "agent" ? EVENT_ACTOR_MODEL : author.has_prefix("user:") ? EVENT_ACTOR_USER : EVENT_ACTOR_SYSTEM, EVENT_SCOPE_TASK, task_id, payload); yield pulse.bump(task_id, "comment"); return comment; } /** * Records a harness/system event on the task's thread (plan * 15 C2): park/resume bookkeeping (aborts, requeues, * retries), the session-context restoration, unreachable * notices — authored 'system', rendered with the distinct * muted system styling, never an agent comment row. */ public async TaskComment add_system_comment(int64 task_id, string body) throws Error { return yield add_comment(task_id, COMMENT_AUTHOR_SYSTEM, body); } /** * Materialises an AGENT comment row from its already-stored * ledger event: the plan-17 chunks push carries the remark as * a comment event, and this creates the task_comments row the * page's comment section renders. No second event is appended * (the ledger already carries the remark); only fresh * (session, seq) pairs reach here, so a redelivered push can * never double the row. */ public async TaskComment adopt_agent_comment(int64 task_id, string body) throws Error { var comment = new TaskComment() { task_id = task_id, author = COMMENT_AUTHOR_AGENT, body = body, created_at = new DateTime.now_local() }; yield db.insert_async(comment); yield pulse.bump(task_id, "agent comment"); return comment; } /** * Raises a blocking prompt on a task — `question` (optionally with * multiple-choice `options` and a markdown `body`), `rfc` or * `review` — and flips it to 'blocked' (or 'review_requested') * while it waits for a human answer. Returns the new block. */ public async TaskBlock add_block(int64 task_id, string question, string block_type = BLOCK_QUESTION, string[]? options = null, string? body = null, bool quiet = false) throws Error { string options_json = null; if (options != null && options.length > 0) { var array = new Json.Array(); foreach (var option in options) { array.add_string_element(option); } options_json = Json.to_string(new Json.Node(Json.NodeType.ARRAY).init_array(array), false); } var block = new TaskBlock() { task_id = task_id, question = question, block_type = block_type.length > 0 ? block_type : BLOCK_QUESTION, options = options_json, body = body != null && body.length > 0 ? body : null, created_at = new DateTime.now_local() }; yield db.insert_async(block); var task = yield db.query() .where(expr("id == $0", elem(task_id))) .first_async(); string previous_status = task != null ? ((!)task).status : ""; if (task != null && (((!)task).status == STATUS_ASSIGNED || ((!)task).status == STATUS_WORKING || ((!)task).status == STATUS_BLOCKED || ((!)task).status == STATUS_REVIEW_REQUESTED)) { // The park flip is a guarded update: a completion // interleaving the raise keeps its terminal state (the // read above already yielded — the entity is stale by // write time under load; plan 13 B2's sticky rule). var guard = connection.create_command( "UPDATE tasks SET status = :status WHERE id = :id" + " AND status IN ('assigned', 'working', 'blocked', 'review_requested')"); guard.with_parameter("status", elem(block.block_type == BLOCK_REVIEW ? STATUS_REVIEW_REQUESTED : STATUS_BLOCKED)); guard.with_parameter("id", elem(task_id)); yield guard.execute_non_query_async(); // Plan 16 D4: a raised block parks the round — the // session row CLOSES 'parked' right here (every raise // path funnels through this flip, entry-driven raises // included, so no parked task ever keeps a dangling // open session). The workspace awaits its resuming // allocation on a fresh row. yield park_open_sessions(task_id); } yield pulse.bump(task_id, "block raised"); var raised_payload = new Json.Object(); raised_payload.set_string_member("from", previous_status); raised_payload.set_string_member("to", block.block_type == BLOCK_REVIEW ? STATUS_REVIEW_REQUESTED : STATUS_BLOCKED); raised_payload.set_string_member("block_type", block.block_type); raised_payload.set_string_member("question", block.question); raised_payload.set_string_member("text", @"$(block.block_type) raised: $(block.question)"); yield events.append(EVENT_STATE_CHANGE, EVENT_ACTOR_SYSTEM, EVENT_SCOPE_TASK, task_id, raised_payload); // Web Push (plan 11 task E): every raised block notifies all // push-subscribed users globally — the goal agent's mirrored // blocks stay quiet (the goal-level block owns the // notification with the goal's title and URL). if (!quiet && task != null) { push.notify_global(PushService.block_notification( ((!)task).title, task_id, block.block_type, block.question)); } return block; } /** * Records a review request (`saddle review`): a review-type block * carrying the summary markdown, with the task flipped to * 'review_requested' while it waits for the human verdict. */ public async TaskBlock request_review(int64 task_id, string summary) throws Error { return yield add_block(task_id, "Review requested", BLOCK_REVIEW, null, summary.length > 0 ? summary : null); } /** * Records a human's answer to a block. A parked round (its * session closed 'parked' at the raise — plan 16 D4) stays * 'blocked'/'review_requested': the dispatcher resumes the * paused workspace and hands the answer over. Only a block on * a task that NEVER RAN (no session rows at all) flips * straight back to 'working'. */ public async void answer_block(int64 block_id, string answer, string? answered_by = null) throws Error { var block = yield db.query() .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); bool review_family = ((!)block).block_type == BLOCK_REVIEW || ((!)block).block_type == BLOCK_PLAN || ((!)block).block_type == BLOCK_COMPLETION; var verdict_payload = new Json.Object(); verdict_payload.set_int_member("block_id", block_id); verdict_payload.set_string_member("block_type", ((!)block).block_type); if (((!)block).answered_by != null) { verdict_payload.set_string_member("by", ((!)block).answered_by); } if (review_family) { bool approved = answer == NOTICE_PLAN_APPROVED || answer == NOTICE_REVIEW_APPROVED; verdict_payload.set_string_member("verdict", approved ? "approved" : "feedback"); verdict_payload.set_string_member("text", @"$(approved ? "approved" : "feedback") — $(answer)"); yield events.append(EVENT_VERDICT, ((!)block).answered_by != null ? EVENT_ACTOR_USER : EVENT_ACTOR_SYSTEM, EVENT_SCOPE_TASK, ((!)block).task_id, verdict_payload); } else { verdict_payload.set_string_member("answer", answer); verdict_payload.set_string_member("text", @"$(((!)block).block_type) answered: $(answer)"); yield events.append(EVENT_ANSWER, ((!)block).answered_by != null ? EVENT_ACTOR_USER : EVENT_ACTOR_SYSTEM, EVENT_SCOPE_TASK, ((!)block).task_id, verdict_payload); } var task = yield db.query() .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 && (yield newest_session_for_task(((!)block).task_id)) == null) { ((!)task).status = STATUS_WORKING; yield db.update_async((!)task); } } // Goal 28 task F (busy-answer delivery): when the task's // affine saddle holds a RUNNING session right now (the // race-restarted round, or a crash-restore that adopted // the parked workspace), NO affinity resume can ever // deliver this verdict — the busy saddle cannot ask. Park // it on the singleton hub instead: the dispatcher's tick // injects it into the running session through the // daemon's task-update channel ({@link // SaddleService.deliver_pending_answers}). A task with no // open session keeps the resume path (the verdict rides // the next resuming assignment's resume_notice, as // before). var running = yield open_session_for_task(((!)block).task_id); if (running != null) { answer_hub.offer(new PendingAnswerDelivery() { block_id = block_id, task_id = ((!)block).task_id, notice = verdict_notice_for((!)block) }); } yield pulse.bump(((!)block).task_id, "block answered"); } /** * The git-host compare URL for the task's branches — * `{git_url}/compare/{branch_from}...{task_branch}` — or "" when * the task has no working branch. */ public async string diff_url_for_task(int64 task_id) throws Error { var task = yield db.query() .where(expr("id == $0", elem(task_id))) .first_async(); if (task == null || ((!)task).task_branch == null || ((!)task).task_branch.length == 0) { return ""; } if (((!)task).project_id <= 0) { // Standalone tasks own no repository — no compare URL. return ""; } var project = yield db.query() .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. * Batched: the candidate tasks load once, their sessions in * one query, the saddles in another. */ public async Invercargill.DataStructures.HashSet unreachable_task_ids() throws Error { var ids = new Invercargill.DataStructures.HashSet(); var candidate_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()) { candidate_ids.add(task.id.to_string()); } } if (candidate_ids.length == 0) { return ids; } // The newest open session per task: sessions ordered by id // ascending, the LAST open row per task wins (the // pre-batch semantics) — walked backwards, first hit per // task keeps the slot. var row_tasks = new Vector(); var row_saddles = new Vector(); foreach (var session in yield db.query() .where(id_in("task_id", candidate_ids)) .order_by(expr("id")) .materialise_async()) { if (session.ended_at == null) { row_tasks.add(session.task_id.to_string()); row_saddles.add(session.saddle_id.to_string()); } } var open_task_ids = new Vector(); var open_saddle_ids = new Vector(); var taken_tasks = new Invercargill.DataStructures.HashSet(); for (int index = (int) row_tasks.length - 1; index >= 0; index--) { if (taken_tasks.add(row_tasks.get(index))) { open_task_ids.add(row_tasks.get(index)); open_saddle_ids.add(row_saddles.get(index)); } } if (open_task_ids.length == 0) { return ids; } var saddle_ids = new Invercargill.DataStructures.HashSet(); foreach (var saddle_id in open_saddle_ids) { saddle_ids.add(saddle_id); } var offline = new Invercargill.DataStructures.HashSet(); foreach (var saddle in yield db.query() .where(id_in("id", saddle_ids)) .materialise_async()) { if (saddle.status == SADDLE_OFFLINE) { offline.add(saddle.id.to_string()); } } for (int index = 0; index < open_task_ids.length; index++) { if (offline.has(open_saddle_ids.get(index))) { ids.add(open_task_ids.get(index)); } } return ids; } /** * An OR-chain of `column == id` comparisons for a batch of ids * (the grammar has no IN operator), or null for an empty * batch — the WHERE-in the batched lookups build. */ /** * The OR-chain batch convention (plan 16 item 23 note): id * batches compare through decimal STRINGS — Invercargill * collections corrupt boxed int64 vectors, and the expression * layer's string compare against integer columns reads fine * (SQLite's dynamic typing). Shared with SaddleService's * derived fleet lists. */ internal static Expression? id_in(string column, Invercargill.DataStructures.HashSet ids) throws Error { Expression? chain = null; foreach (var id in ids) { var comparison = expr(column + " == $0", elem(id)); chain = chain == null ? comparison : new BinaryExpression((!)chain, comparison, BinaryOperator.OR); } return chain; } /** * The resume_notice wording for an answered block (the plan-11 * stable verdicts): review and plan answers are already stored * in the stable wording and pass through verbatim; question * replies are formatted the way the daemon used to format * delivered answers ("Human selected: …" for option matches, * "Human answered (other): …" otherwise); RFC/plain answers * carry the "Human answer: …" prefix. */ public static string verdict_notice_for(TaskBlock block) { string answer = block.answer ?? ""; if (block.block_type == BLOCK_REVIEW || block.block_type == BLOCK_PLAN) { return answer; } if (block.block_type == BLOCK_QUESTION) { foreach (var option in EntryJson.parse_string_array(block.options)) { if (option == answer) { return "Human selected: " + answer; } } return "Human answered (other): " + answer; } return "Human answer: " + answer; } /** * The ONE terminal stamping path (complete/fail/cancel): owns * the status flip, the `completed_at` stamp and the summary * precedence — a non-empty `summary` replaces whatever the * row holds, else `empty_summary` applies only when the row * holds nothing. Callers keep their own idempotence guard * (terminal states are sticky, plan 13 B2). */ private async void mark_terminal(TaskEntity task, string status, string? summary, string? empty_summary = null) throws Error { string from = task.status; task.status = status; task.completed_at = new DateTime.now_local(); if (summary != null && summary.length > 0) { task.summary = summary; } else if ((task.summary ?? "").length == 0 && empty_summary != null) { task.summary = empty_summary; } yield db.update_async(task); var payload = new Json.Object(); payload.set_string_member("from", from); payload.set_string_member("to", status); payload.set_string_member("text", status == STATUS_DONE ? "Task completed" : "Task failed"); if ((task.summary ?? "").length > 0) { payload.set_string_member("summary", (!)task.summary); } yield events.append(EVENT_STATE_CHANGE, EVENT_ACTOR_SYSTEM, EVENT_SCOPE_TASK, task.id, payload); } /** * The ONE park stamping path: `suspended` plus the * `suspended_at` stamp, stamped once per parking window so the * resume trigger only sees goal events that landed after the * suspension. No-op when already parked (the stamp survives). * Plan 16 D4: parking closes the round's open sessions * ('parked') — the pause record the resume machinery keys on * is the newest session row, open or closed. */ public async void set_parked(int64 task_id) throws Error { var task = yield db.query() .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 park_open_sessions(task_id); yield pulse.bump(task_id, "task suspended"); yield append_state_change(task_id, STATUS_WORKING, STATUS_SUSPENDED, "Parked — awaiting the resume", EVENT_ACTOR_SYSTEM); // Plan 16 S5 (U2): the mechanical park notice persists in // the audit ledger (a system comment event) and never // renders on a user-facing timeline. yield add_system_comment(task_id, "goal agent parked until an owned task completes or fails"); } /** * Marks a task done: records `summary`, stamps `completed_at`, * closes any open session (outcome `outcome`) and frees the * saddle that was running it. Terminal transitions project ONCE * (plan 13 B2): a task already done or failed keeps its first * outcome — a redelivered push or a second terminal entry with * a fresh id re-runs no bookkeeping, no comment, no event * (retry_task is the one path that un-terminals). */ public async void complete_task(int64 task_id, string? summary, string outcome) throws Error { var task = yield db.query() .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; } yield mark_terminal((!)task, STATUS_DONE, summary); yield release_task(task_id, outcome, false); yield pulse.bump(task_id, "task completed"); // Web Push (plan 11 task E): completion notifies the users // holding a "Notify me" pref for this task. push.notify_opted_in(task_id, PushService.lifecycle_notification( ((!)task).title, task_id, "completed", summary)); // Plan 12: the goal's narrative hears the terminal event // (a parked goal agent resumes on it), and a done // predecessor auto-enqueues its dependents. yield enqueue_dependents_of(((!)task)); } /** * Cancels a task: status 'failed' with a cancellation summary. Any * open session is closed with outcome 'stopped' and its saddle * freed. Draft tasks are refused — they are deleted, not * cancelled (nothing ever ran). Idempotent on terminal tasks * (plan 13 B2): the first outcome stands. */ public async void cancel_task(int64 task_id) throws Error { var task = yield db.query() .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"); } yield mark_terminal((!)task, STATUS_FAILED, null, "Cancelled by user"); { var payload = new Json.Object(); payload.set_string_member("text", "Task cancelled by user"); yield events.append(EVENT_CANCEL, EVENT_ACTOR_USER, EVENT_SCOPE_TASK, task_id, payload); } yield release_task(task_id, OUTCOME_STOPPED, false); yield pulse.bump(task_id, "task cancelled"); } /** * Records a saddle-reported failure: status 'failed' with the * daemon's message as summary, the open session closed with * outcome 'failed' and its saddle freed. Plan 16 S4 * (terminal-text single representation): `tasks.summary` IS * the terminal cache — no duplicate agent comment is written * (the narrative renders `summary_lead`; the task page renders * the summary card). Terminal transitions project ONCE (plan * 13 B2): the live failure's honest burst — a dozen-plus * distinct `saddle failure` entry ids in one session — still * fails the task exactly once, with one summary write and one * goal event. */ public async void fail_task(int64 task_id, string? summary) throws Error { var task = yield db.query() .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; } yield mark_terminal((!)task, STATUS_FAILED, summary, "Saddle reported a failure"); yield release_task(task_id, OUTCOME_FAILED, false); yield pulse.bump(task_id, "task failed"); // Web Push (plan 11 task E): an agent-reported failure // notifies the opted-in users (a human cancellation does // not — the cancellator already knows). push.notify_opted_in(task_id, PushService.lifecycle_notification( ((!)task).title, task_id, "failed", summary)); } /** Withdraws the suspension stamp after a successful resume. */ public async void clear_suspension(int64 task_id) throws Error { var task = yield db.query() .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 record in yield events.list_for_task(task_id, new string[0])) { if (record.created_at != null) { park = (!)record.created_at; break; } } } if (park == null) { // Never parked (or nothing recorded): nothing could have // advanced past a marker that does not exist. return false; } foreach (var sibling in yield sibling_tasks((!)me)) { if (sibling.completed_at != null && ((!)sibling.completed_at).compare((!)park) > 0) { return true; } foreach (var session in yield db.query() .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 record in yield events.list_for_task(sibling.id, { EVENT_PROGRESS })) { if (record.created_at != null && ((!)record.created_at).compare((!)park) > 0) { return true; } } } return false; } /** * The tasks sharing `task`'s working context (excluding it): * same-project tasks on the same non-empty working branch, and * — for a goal agent — every other task of its goal (their * completions merge into the goal branch, so the goal agent's * post-park pull must hear them). */ private async ImmutableLot 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 — {@link release_task} with the requeue, the * single owner of the bookkeeping. Plan 14 B1: a parked * (suspended) task requeues too — the goal page's * requeue-to-resume reaches a goal agent parked on a dead * saddle; the suspension stamp clears so the next dispatch * starts a fresh parking window. */ public async void requeue_task(int64 task_id) throws Error { yield release_task(task_id, OUTCOME_STOPPED, true); yield pulse.bump(task_id, "task requeued"); } /** * Crash recovery at the ask: a daemon that asks for work is * provably idle — its own state machine parks the ask loop * while any session runs — so an open session on an * assigned/working task can only be a crash orphan (both * sides restart, nobody closes the row, and the busy record * then wedges the fleet: the saddle asks, the claim rolls * back, forever, while the UI shows it busy). Every orphan * older than the grace window requeues — the very ask that * found it can win the task straight back. The grace window * protects a manual /start whose session row lands while an * ask is in flight. */ public const int64 ASK_RECOVERY_GRACE_US = 120 * 1000000L; /** * Goal 28 task F: how long a round's last push keeps its * session un-requeueable after the daemon asks — the window * in which a park's transition may still be in flight * (matches the daemon-side park settle). */ public const int64 PARK_SETTLE_GRACE_S = 5; public async int requeue_interrupted_sessions(Saddle saddle) throws Error { var requeued = 0; foreach (var session in yield db.query() .where(expr("saddle_id == $0", elem(saddle.id))) .order_by(expr("id")) .materialise_async()) { if (session.ended_at != null) { continue; } var started = session.started_at ?? new DateTime.now_local(); if (new DateTime.now_local().difference(started) < ASK_RECOVERY_GRACE_US) { continue; } // Goal 28 task F (the allocation guard's requeue leg): // a round that pushed chunks/results moments ago was // LIVE — its closing transition (a park carrying the // block payload) may still be queued in the daemon's // outbox. Requeueing here handed the parked task back // as a crash restore with its question gone (the // task-187 race); hold instead — the ask 204s, the // transition lands, the next ask sees the parked row. if (poll_state.seconds_since_push(session.id) < PARK_SETTLE_GRACE_S) { continue; } var task = yield db.query() .where(expr("id == $0", elem(session.task_id))) .first_async(); if (task == null || (((!)task).status != STATUS_ASSIGNED && ((!)task).status != STATUS_WORKING)) { continue; } yield requeue_task(((!)task).id); yield add_system_comment(((!)task).id, @"Session interrupted (saddle '$(saddle.name)' restarted mid-run); requeued"); requeued++; } return requeued; } /** * Returns a failed task to the queue for another attempt. * Plan 16 S4: the failure summary is the terminal cache of the * attempt that earned it — retry clears it and copies it * NOWHERE (the narrative rendered its lead, the page rendered * its card; the next attempt writes a fresh one), so no * summary→comment duplicate joins the thread. `completed_at` * is withdrawn, any open session closes and its saddle frees — * the same bookkeeping as {@link requeue_task} (retry's flip * is unconditional: the attempt resets whatever it held), * plus the audit comment "Task retried by user". */ public async void retry_task(int64 task_id) throws Error { var task = yield db.query() .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_QUEUED; ((!)task).summary = null; ((!)task).completed_at = null; yield db.update_async((!)task); yield release_task(task_id, OUTCOME_STOPPED, false); yield add_system_comment(task_id, "Task retried by user"); yield append_state_change(task_id, STATUS_FAILED, STATUS_QUEUED, "Task retried by user", EVENT_ACTOR_USER); yield pulse.bump(task_id, "task retried"); } /** * Queued tasks whose auto-enqueue predecessor (if any) is * done, ordered by priority (1 first) then creation time — * the dispatcher's candidate list. Only the CANDIDATES' * predecessors are queried (not every task row). */ public async ImmutableLot get_ready_tasks() throws Error { var queued = yield db.query() .where(expr("status == $0", elem(STATUS_QUEUED))) .order_by(expr("priority")) .order_by(expr("created_at")) .materialise_async(); var predecessor_ids = new Invercargill.DataStructures.HashSet(); foreach (var task in queued) { if (task.enqueue_after_task_id > 0) { predecessor_ids.add(task.enqueue_after_task_id.to_string()); } } var predecessor_status = new Dictionary(); if (predecessor_ids.length > 0) { foreach (var predecessor in yield db.query() .where(id_in("id", predecessor_ids)) .materialise_async()) { predecessor_status.set(predecessor.id.to_string(), predecessor.status); } } 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. A predecessor row that // does not resolve keeps the dependent waiting. string? status = predecessor_status.get_or_default( task.enqueue_after_task_id.to_string()); ready = status != null && status == STATUS_DONE; } if (ready) { rows.add(task); } } return rows.to_immutable_buffer(); } /** Comments on a task, newest first — the thread leads with * the latest remark (the list-page consistency sweep). */ public async ImmutableLot list_comments(int64 task_id) throws Error { return yield db.query() .where(expr("task_id == $0", elem(task_id))) .order_by_desc(expr("id")) .materialise_async(); } /** Blocks on a task, newest first — the blocking sections * lead with the most recent question, RFC or review (the * list-page consistency sweep). */ public async ImmutableLot list_blocks(int64 task_id) throws Error { return yield db.query() .where(expr("task_id == $0", elem(task_id))) .order_by_desc(expr("id")) .materialise_async(); } /** The task's newest block, or null when it has none — the * FIRST row (list_blocks is newest-first since the ordering * sweep; the walker keeps the semantics explicit). */ public async TaskBlock? newest_block(int64 task_id) throws Error { foreach (var block in yield list_blocks(task_id)) { return block; } return null; } /** * True when the task's newest block exists and carries an * answer — a verdict is waiting to reach the parked workspace * (the answer-delivery pass, the affinity resume, or a later * assignment payload). Such a task must never be requeued: the * requeue closes the session the resume machinery needs. * Plan 15 C1: goal blocks live on the run task as single rows * — no mirror pair exists to fall out of sync, so the newest * row's own answer is the whole truth. */ public async bool newest_block_answered(int64 task_id) throws Error { var newest = yield newest_block(task_id); return newest != null && ((!)newest).answer != null; } /** * Goal 28 task F (batched verdict composition): every block * ANSWERED since `since` (the resumed session's start, or the * suspension stamp), oldest first — the resume notice joins * their verdict texts so a newer block can never swallow an * older answer. Undelivered-only filtering is the caller's * (the poll-state ledger decides what the agent has already * seen mid-session). */ public async ImmutableLot blocks_answered_since(int64 task_id, DateTime? since) throws Error { var rows = new Vector(); // list_blocks is newest-first; walk it backwards so the // batch reads oldest-first. var newest_first = yield list_blocks(task_id); var reversed = new Vector(); foreach (var block in newest_first) { reversed.add(block); } for (int i = (int) reversed.length - 1; i >= 0; i--) { var block = reversed[i]; if (block.answer == null || block.answered_at == null) { continue; } if (since != null && block.answered_at.compare(((!)since)) < 0) { continue; } rows.add(block); } return rows.to_immutable_buffer(); } /** * The joined verdict text for a batch of answered blocks * (oldest first — {@link blocks_answered_since}): each * block's {@link verdict_notice_for} wording on its own * paragraph, in answer order. An empty batch yields "". */ public static string batched_verdict_notice(ImmutableLot blocks) { var parts = new Vector(); foreach (var block in blocks) { parts.add(verdict_notice_for(block)); } return string.joinv("\n\n", parts.to_array()); } /** * The session still open for `task_id` (no `ended_at`), or null. * The null check runs in Vala: the expression layer cannot compare * columns against null. */ public async Session? open_session_for_task(int64 task_id) throws Error { var sessions = yield db.query() .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; } /** * The task's newest session row, open or closed (plan 16 D4: * a parked round's row is closed 'parked' — it is still the * pause record the affinity resume, the session-scoped task * CLI and the goal page's run surfaces key on), or null when * the task never ran. */ public async Session? newest_session_for_task(int64 task_id) throws Error { Session? newest = null; foreach (var session in yield db.query() .where(expr("task_id == $0", elem(task_id))) .order_by(expr("id")) .materialise_async()) { newest = session; } return newest; } /** * Closes the task's open sessions with outcome 'parked' * (plan 16 D4 — the ONE park close every raise and suspend * path runs; idempotent, nothing to do for rows already * closed). The workspace stays on its saddle awaiting the * resuming allocation that opens the successor row. */ public async void park_open_sessions(int64 task_id) throws Error { foreach (var session in yield db.query() .where(expr("task_id == $0", elem(task_id))) .materialise_async()) { if (session.ended_at != null) { continue; } session.ended_at = new DateTime.now_local(); session.outcome = OUTCOME_PARKED; yield db.update_async(session); } } /** * The ONE release path (plan 16 R2, arch item 13): every path * that takes a task off its saddle runs through here. Closes * the task's open sessions with `outcome` — the close IS the * free (plan 16 item 23: busy/idle derive from the open * sessions; no saddle-row write remains to race an ask * allocation). `requeue` first returns a non-terminal task to * the queue (assigned/working/blocked/review_requested/ * suspended — the suspension stamp withdraws so the next * dispatch opens a fresh parking window). * * The terminal callers (complete/cancel/fail) pass their own * outcome with the status already stamped; requeue_task and * the unregister path ride the requeue; retry_task does its * own unconditional flip (summary and completed_at die with * the attempt) and releases without one. The offline variant * ({@link SaddleService.mark_saddle_unreachable}) calls this * for its derived held task before its own guarded OFFLINE * stamp; its keep-parked gate (a task that parked mid-flight) * skips the release entirely. */ /** * Goal 8 task 4 (comment-resurrect): re-queues a TERMINAL * non-goal task (goal_id == 0, status done/failed) for another * round after a human comment. The completion summary and * `completed_at` STAY (the report of the earlier round remains * readable — the daemon's next finish overwrites the summary), * and a fresh parked session row is seeded on the task's last * saddle so the plain-task requeue path (which keys on the * newest session's affine saddle) has a workspace to resume; * no session row at all (a fixture-shaped task) requeues onto * the open queue instead. The resurrection is idempotent from * the caller's side: only a terminal task flips. */ public async void requeue_terminal_task(int64 task_id) throws Error { var task = yield db.query() .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; } string from = ((!)task).status; ((!)task).status = STATUS_QUEUED; ((!)task).completed_at = null; yield db.update_async((!)task); yield append_state_change(task_id, from, STATUS_QUEUED, @"Task re-opened by comment ($(from))", EVENT_ACTOR_USER); yield pulse.bump(task_id, "task re-opened by comment"); } /** * Seeds a CLOSED 'parked' session row for `task_id` on the * saddle of its newest prior session (the resurrect paths' * resume anchor — the allocation's parked-resume arm needs a * session row to key the affine saddle and the successor row * on). Returns true when a row was seeded; false when the task * has no session history at all. */ public async bool seed_parked_session(int64 task_id) throws Error { var newest = yield newest_session_for_task(task_id); if (newest == null) { return false; } var seeded = new Session() { task_id = task_id, saddle_id = ((!)newest).saddle_id, started_at = ((!)newest).started_at ?? new DateTime.now_local(), ended_at = new DateTime.now_local(), outcome = OUTCOME_PARKED }; yield db.insert_async(seeded); return true; } public async void release_task(int64 task_id, string outcome, bool requeue) throws Error { if (requeue) { 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)) { string from = ((!)task).status; ((!)task).status = STATUS_QUEUED; if (((!)task).suspended_at != null) { ((!)task).suspended_at = null; } yield db.update_async((!)task); yield append_state_change(task_id, from, STATUS_QUEUED, "Task requeued", EVENT_ACTOR_SYSTEM); } } 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); } } } }