| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588 |
- using Invercargill;
- using Invercargill.DataStructures;
- using Invercargill.Expressions;
- using InvercargillSql;
- using InvercargillSql.Orm;
- using Inversion;
- using ClankerSaddle.Data;
- namespace ClankerSaddle.Services {
- /** Valid task statuses (the `tasks.status` wire values). */
- public const string STATUS_DRAFT = "draft";
- public const string STATUS_QUEUED = "queued";
- public const string STATUS_ASSIGNED = "assigned";
- public const string STATUS_WORKING = "working";
- public const string STATUS_BLOCKED = "blocked";
- public const string STATUS_REVIEW_REQUESTED = "review_requested";
- public const string STATUS_SUSPENDED = "suspended";
- public const string STATUS_DONE = "done";
- public const string STATUS_FAILED = "failed";
- /** Valid {@link TaskBlock.block_type} values. */
- public const string BLOCK_QUESTION = "question";
- public const string BLOCK_RFC = "rfc";
- public const string BLOCK_REVIEW = "review";
- public const string BLOCK_PLAN = "plan";
- /**
- * The goal agent's completion report (plan 15 C1): raised on the
- * goal-run task by {@link GoalService.submit_completion} — the
- * goal's review kinds are plan and completion.
- */
- public const string BLOCK_COMPLETION = "completion";
- /**
- * The system-event comment author (plan 15 C2): harness-authored
- * bookkeeping rows (aborts, requeues, retries, session
- * restorations) — never agent content.
- */
- public const string COMMENT_AUTHOR_SYSTEM = "system";
- /**
- * Plan 16 S5 (U2): MECHANICAL harness bookkeeping that every
- * user-facing timeline filters out — the session-restore footnote
- * and the park-until notice (both remain in the raw terminal log,
- * which keeps everything). Matched by substring so the daemon's
- * [harness]-prefixed wordings and the web's own phrasing all stay
- * covered; substantive system rows (aborts, requeues, retries)
- * never match.
- */
- public static bool comment_is_mechanical(string body) {
- return body.contains("session context restored")
- || body.contains("parked until");
- }
- /**
- * The agent-celebration entry type (`saddle yeehaw <message>`): the
- * daemon pushes it as a round-results entry carrying the message in
- * the output field, so it stores like any other `session_commands`
- * row — nothing new is persisted for the badge or the card.
- */
- public const string ENTRY_YEEHAW = "yeehaw";
- /** The task page's yeehaw card cap (newest entries kept). */
- public const int YEEHAW_CARD_MAX = 3;
- /**
- * The stable verdict wordings (plan 11): they are stored as the
- * block answers AND ride the resuming allocation's
- * `resume_notice` verbatim, so agent and human always read the
- * same text.
- */
- public const string NOTICE_PLAN_APPROVED = "Plan approved. Proceed with implementation.";
- public const string NOTICE_PLAN_FEEDBACK = "Plan feedback received:";
- public const string NOTICE_REVIEW_APPROVED = "Review approved. Continue with the task.";
- public const string NOTICE_REVIEW_FEEDBACK = "Review feedback received:";
- /** The display label of a goal-run parked on a review-family block (U3). */
- public const string DISPLAY_AWAITING_REVIEW = "awaiting your review";
- /**
- * A task's computed presentation state (stored status + unanswered
- * blocks): `review requested` while a review block is unanswered
- * (even when the stored status is terminal — the post-mortem's
- * `done`-with-open-review case), `parked` for the suspended state
- * (a goal agent between events), else the stored status. Plan 16
- * S5 (U3): a GOAL-RUN task parked on a review-family block (plan
- * or completion review) reads "awaiting your review" — agreeing
- * with its goal's label, never a bare "blocked". Pure
- * presentation — the stored statuses and the dispatcher are
- * untouched.
- */
- public class TaskDisplay : Object {
- public string status { get; set; default = ""; }
- public bool terminal { get; set; }
- public bool failed { get; set; }
- public static TaskDisplay compute(TaskEntity task, bool review_pending,
- bool goal_review_pending = false) {
- var view = new TaskDisplay() {
- terminal = task.status == STATUS_DONE || task.status == STATUS_FAILED,
- failed = task.status == STATUS_FAILED
- };
- if (goal_review_pending && task.kind == TASK_KIND_GOAL_RUN) {
- view.status = DISPLAY_AWAITING_REVIEW;
- } else if (review_pending) {
- view.status = "review requested";
- } else if (task.status == STATUS_SUSPENDED) {
- view.status = "parked";
- } else {
- view.status = task.status;
- }
- return view;
- }
- }
- /**
- * The full task aggregate loaded by {@link TaskService.get_task}: the
- * task itself plus its comments, blocks, progress history and
- * session history.
- */
- public class TaskDetail : Object {
- public TaskEntity task { get; set; }
- public ImmutableLot<TaskComment> comments { get; set; }
- public ImmutableLot<TaskBlock> blocks { get; set; }
- public ImmutableLot<TaskProgress> progress { get; set; }
- public ImmutableLot<Session> sessions { get; set; }
- }
- /**
- * One agent celebration (`saddle yeehaw`) read back for display: the
- * message from the entry's output field and when it landed.
- */
- public class YeehawEntry : Object {
- public string message { get; set; default = ""; }
- public DateTime? at { get; set; }
- }
- /**
- * The task's celebration feed for the yeehaw card: the newest
- * {@link YEEHAW_CARD_MAX} entries plus the uncapped total (the
- * "+N more" hint derives from the two).
- */
- public class YeehawFeed : Object {
- public Vector<YeehawEntry> latest { get; set; default = new Vector<YeehawEntry>(); }
- public int64 total { get; set; }
- }
- /**
- * Task storage and lifecycle: creation, enqueue-after-aware
- * readiness, agent interaction primitives (comments, blocks,
- * progress, reviews) and completion. Saddle bookkeeping rows are
- * kept consistent on every terminal transition. The plan-12 task
- * graph is goal-shaped: no parents, no dependencies, no plans —
- * goals own their tasks.
- */
- public class TaskService : Object {
- private OrmSession db = inject<OrmSession>();
- private Connection connection = inject<Connection>();
- private StatePulse pulse = inject<StatePulse>();
- private PushService push = inject<PushService>();
- /**
- * Inserts a task (status 'draft' until its creator enqueues
- * it). `description`, `task_branch` and `branch_from` may be
- * null; `branch_from` prefills from the project's default
- * branch; priority clamps to 1–9 (5 is the default, 1 the
- * highest). The plan-12 shape: `kind` ("task"/"goal_run"),
- * `mode` (implement/investigate/review/test, orchestrate for
- * goal runs), `creator` ("human"/"goal" — enqueueing is
- * creator-only), `goal_id` (the owning goal) and
- * `enqueue_after_task_id` (human-created tasks only: the
- * auto-enqueue predecessor, cycle-checked by the caller).
- */
- public async TaskEntity create_task(int64 project_id, string title, string? description,
- int64 priority,
- string? task_branch = null, string? branch_from = null,
- string status = STATUS_DRAFT,
- string kind = TASK_KIND_TASK, string mode = "implement",
- string creator = CREATOR_HUMAN, int64 goal_id = 0,
- int64 enqueue_after_task_id = 0) throws Error {
- string resolved_branch_from = branch_from ?? "";
- if (resolved_branch_from.length == 0) {
- var project = yield db.query<Project>()
- .where(expr("id == $0", elem<int64?>(project_id)))
- .first_async();
- if (project != null) {
- resolved_branch_from = ((!)project).default_branch;
- }
- }
- string? resolved_task_branch = task_branch != null && task_branch.length > 0 ? task_branch : null;
- if (goal_id > 0 && kind == TASK_KIND_TASK) {
- // Goal worker tasks exist only under a live goal.
- var goal = yield db.query<Goal>()
- .where(expr("id == $0", elem<int64?>(goal_id)))
- .first_async();
- if (goal == null) {
- throw new Error(Quark.from_string("clanker-saddle"), 75,
- @"No goal exists with id $(goal_id)");
- }
- }
- var task = new TaskEntity() {
- project_id = project_id,
- title = title,
- description = description != null && description.length > 0 ? description : null,
- priority = priority.clamp(1, 9),
- status = status,
- task_branch = resolved_task_branch,
- branch_from = resolved_branch_from.length > 0 ? resolved_branch_from : null,
- kind = kind,
- mode = mode,
- creator = creator,
- goal_id = goal_id,
- enqueue_after_task_id = enqueue_after_task_id,
- created_at = new DateTime.now_local()
- };
- yield db.insert_async<TaskEntity>(task);
- if (goal_id > 0 && kind == TASK_KIND_TASK) {
- // Goal worker tasks run on ephemeral auto-named
- // branches cut from the goal branch (the daemon builds
- // the worktrees; the stored name feeds compare links).
- var goal = yield db.query<Goal>()
- .where(expr("id == $0", elem<int64?>(goal_id)))
- .first_async();
- if (goal != null) {
- task.task_branch = @"task/$(goal_id)-$(task.id)";
- task.branch_from = ((!)goal).branch_name;
- yield db.update_async<TaskEntity>(task);
- }
- }
- yield pulse.bump(task.id, "task created");
- return task;
- }
- /**
- * Loads the full aggregate for `id`: the task, its comments,
- * blocks, progress and sessions. Returns null when the task
- * does not exist.
- */
- public async TaskDetail? get_task(int64 id) throws Error {
- var task = yield db.query<TaskEntity>()
- .where(expr("id == $0", elem<int64?>(id)))
- .first_async();
- if (task == null) {
- return null;
- }
- return new TaskDetail() {
- task = (!)task,
- comments = yield list_comments(id),
- blocks = yield list_blocks(id),
- progress = yield list_progress(id),
- sessions = yield db.query<Session>()
- .where(expr("task_id == $0", elem<int64?>(id)))
- .order_by(expr("id"))
- .materialise_async()
- };
- }
- /**
- * Computed display states for a batch of tasks (all lists share
- * this), keyed by task id string. Unanswered review blocks load
- * once, not per task. Plan 16 S5 (U3): a goal-run task with a
- * pending review-family block (plan/completion/review) reads
- * "awaiting your review" — the run-task card agrees with its
- * goal.
- */
- public async Dictionary<string, TaskDisplay> display_map_for(
- Lot<TaskEntity> tasks) throws Error {
- var wanted = new Invercargill.DataStructures.HashSet<string>();
- var goal_runs = new Invercargill.DataStructures.HashSet<string>();
- foreach (var task in tasks) {
- wanted.add(task.id.to_string());
- if (task.kind == TASK_KIND_GOAL_RUN) {
- goal_runs.add(task.id.to_string());
- }
- }
- var review_pending = new Invercargill.DataStructures.HashSet<string>();
- var goal_review_pending = new Invercargill.DataStructures.HashSet<string>();
- foreach (var family in new string[] { BLOCK_REVIEW, BLOCK_COMPLETION, BLOCK_PLAN }) {
- foreach (var block in yield db.query<TaskBlock>()
- .where(expr("type == $0", elem(family)))
- .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 (goal_runs.has(block.task_id.to_string())) {
- goal_review_pending.add(block.task_id.to_string());
- }
- }
- }
- var map = new Dictionary<string, TaskDisplay>();
- foreach (var task in tasks) {
- map.set(task.id.to_string(),
- TaskDisplay.compute(task, review_pending.has(task.id.to_string()),
- goal_review_pending.has(task.id.to_string())));
- }
- return map;
- }
- /**
- * The display state of a single task — the detail page's path,
- * computed from its already-loaded aggregate (no extra queries).
- */
- public static TaskDisplay display_of_detail(TaskDetail detail) {
- bool review_pending = false;
- bool goal_review_pending = false;
- foreach (var block in detail.blocks) {
- if (block.answered_at != null) {
- continue;
- }
- if (block.block_type == BLOCK_REVIEW) {
- review_pending = true;
- }
- if (detail.task.kind == TASK_KIND_GOAL_RUN
- && (block.block_type == BLOCK_REVIEW
- || block.block_type == BLOCK_COMPLETION
- || block.block_type == BLOCK_PLAN)) {
- goal_review_pending = true;
- }
- }
- return TaskDisplay.compute(detail.task, review_pending, goal_review_pending);
- }
- /**
- * Lists tasks, optionally filtered by status and/or project, ordered
- * by priority (1 first) then creation time.
- */
- public async ImmutableLot<TaskEntity> list_tasks(string? status, int64? project_id) throws Error {
- var tasks = yield db.query<TaskEntity>().materialise_async();
- var rows = new Vector<TaskEntity>();
- foreach (var task in tasks) {
- if (status != null && status.length > 0 && task.status != status) {
- continue;
- }
- if (project_id != null && task.project_id != project_id) {
- continue;
- }
- rows.add(task);
- }
- var sorted = rows.sort((a, b) => {
- if (a.priority != b.priority) {
- return (int)(a.priority - b.priority);
- }
- int64 a_created = a.created_at != null ? ((!)a.created_at).to_unix() : 0;
- int64 b_created = b.created_at != null ? ((!)b.created_at).to_unix() : 0;
- return (int)(a_created - b_created);
- });
- return sorted.to_immutable_buffer();
- }
- /**
- * Saves the human-editable fields of the task identified by `id`
- * under the draft-lifecycle edit matrix: in DRAFT all fields
- * except status are editable (title, description/mandate,
- * priority, branch fields, the plan-12 work mode and —
- * human-created tasks only — the auto-enqueue predecessor,
- * cycle-validated); in non-draft non-terminal
- * states only priority applies; in terminal states nothing
- * does. Fields not editable in the current state are silently
- * left unchanged (null arguments always mean "leave
- * unchanged"; an empty branch field clears it; an
- * `enqueue_after_task_id` of -1 CLEARS the predecessor).
- * Returns null when the task does not exist. The pulse only
- * fires when a value actually moved, so periodic status
- * refreshes do not spam live pages.
- */
- public async TaskEntity? edit_task(int64 id, string? title, string? description,
- int64? priority, string? task_branch, string? branch_from,
- string? mode = null, int64? enqueue_after_task_id = null) throws Error {
- var task = yield db.query<TaskEntity>()
- .where(expr("id == $0", elem<int64?>(id)))
- .first_async();
- if (task == null) {
- return null;
- }
- bool changed = false;
- bool draft = ((!)task).status == STATUS_DRAFT;
- if (draft) {
- if (title != null && title.length > 0 && ((!)task).title != title) {
- ((!)task).title = title;
- changed = true;
- }
- if (description != null) {
- string? next_description = description.length > 0 ? description : null;
- if (((!)task).description != next_description) {
- ((!)task).description = next_description;
- changed = true;
- }
- }
- if (task_branch != null) {
- string? next_branch = task_branch.length > 0 ? task_branch : null;
- if (((!)task).task_branch != next_branch) {
- ((!)task).task_branch = next_branch;
- changed = true;
- }
- }
- if (branch_from != null) {
- string? next_from = branch_from.length > 0 ? branch_from : null;
- if (((!)task).branch_from != next_from) {
- ((!)task).branch_from = next_from;
- changed = true;
- }
- }
- if (mode != null && mode.length > 0) {
- if (!is_valid_mode(mode)) {
- throw new Error(Quark.from_string("clanker-saddle"), 79,
- @"Unknown mode '$(mode)' — valid modes: " + string.joinv(", ", TASK_MODES));
- }
- if (((!)task).mode != mode) {
- ((!)task).mode = mode;
- changed = true;
- }
- }
- // Human-created tasks only: the auto-enqueue
- // predecessor is settable/clearable while draft, and
- // setting it validates the chain (no cycles).
- if (enqueue_after_task_id != null
- && ((!)task).creator == CREATOR_HUMAN
- && ((!)task).kind == TASK_KIND_TASK) {
- int64 next = (!)enqueue_after_task_id;
- if (next < 0) {
- // The -1 sentinel clears the predecessor.
- next = 0;
- }
- if (next > 0) {
- yield validate_enqueue_after_cycle(id, next);
- }
- if (((!)task).enqueue_after_task_id != next) {
- ((!)task).enqueue_after_task_id = next;
- changed = true;
- }
- }
- }
- // Priority stays editable in every non-terminal state.
- if ((((!)task).status != STATUS_DONE && ((!)task).status != STATUS_FAILED)
- && priority != null) {
- int64 clamped = ((!)priority).clamp(1, 9);
- if (((!)task).priority != clamped) {
- ((!)task).priority = clamped;
- changed = true;
- }
- }
- yield db.update_async<TaskEntity>(task);
- if (changed) {
- yield pulse.bump(id, "task saved");
- }
- return task;
- }
- /**
- * The internal status-transition path (daemon pushes, park and
- * result machinery) — never reachable from a human form.
- * Stamps `completed_at` on done. Returns null when the task
- * does not exist.
- */
- public async TaskEntity? set_status(int64 id, string status) throws Error {
- var task = yield db.query<TaskEntity>()
- .where(expr("id == $0", elem<int64?>(id)))
- .first_async();
- if (task == null) {
- return null;
- }
- if (((!)task).status != status) {
- if (status != STATUS_DONE && status != STATUS_FAILED) {
- // Terminal states are sticky (plan 13 B2's
- // invariant): a NON-terminal transition must never
- // downgrade a done/failed task. The entity writes
- // below interleave with completions on the main
- // loop (the terminal checks in the callers have
- // already yielded by then — a late 'working' push
- // racing a 'done' push used to resurrect the task
- // as working AFTER its session closed done), so
- // the guard lives in the UPDATE's WHERE: atomic.
- // The row is re-read afterwards so the caller sees
- // the truth (terminal included), never a stale
- // entity it might write back.
- var guard = connection.create_command(
- "UPDATE tasks SET status = :status WHERE id = :id"
- + " AND status NOT IN ('done', 'failed')");
- guard.with_parameter<Invercargill.Element>("status",
- elem(status));
- guard.with_parameter<Invercargill.Element>("id",
- elem<int64?>(id));
- yield guard.execute_non_query_async();
- task = yield db.query<TaskEntity>()
- .where(expr("id == $0", elem<int64?>(id)))
- .first_async();
- if (task == null) {
- return null;
- }
- } else {
- ((!)task).status = status;
- if (status == STATUS_DONE && ((!)task).completed_at == null) {
- ((!)task).completed_at = new DateTime.now_local();
- }
- yield db.update_async<TaskEntity>((!)task);
- }
- yield pulse.bump(id, @"task $(status)");
- }
- return task;
- }
- /** Every task under `goal_id` (both kinds), oldest first. */
- public async ImmutableLot<TaskEntity> list_tasks_for_goal(int64 goal_id) throws Error {
- return yield db.query<TaskEntity>()
- .where(expr("goal_id == $0", elem<int64?>(goal_id)))
- .order_by(expr("id"))
- .materialise_async();
- }
- /**
- * The Enqueue action (creator-only, plan 12): moves a draft
- * task to the queue. Humans enqueue theirs, goal agents theirs
- * — an actor that is not the task's creator is refused (the
- * goal agent sequences its own tasks). A pending
- * auto-enqueue predecessor does not refuse the enqueue — the
- * dispatcher's readiness check holds dispatch back until it
- * finishes. The internal auto-enqueue-after pass calls this as
- * the dependent's creator (the creator's pre-authorisation).
- */
- public async void enqueue_task(int64 id, string actor = CREATOR_HUMAN) throws Error {
- var task = yield db.query<TaskEntity>()
- .where(expr("id == $0", elem<int64?>(id)))
- .first_async();
- if (task == null) {
- throw new Error(Quark.from_string("clanker-saddle"), 50,
- @"No task exists with id $(id)");
- }
- if (((!)task).status != STATUS_DRAFT) {
- throw new Error(Quark.from_string("clanker-saddle"), 51,
- "Only draft tasks can be enqueued");
- }
- if (((!)task).creator != actor) {
- throw new Error(Quark.from_string("clanker-saddle"), 76,
- @"Only the task's creator may enqueue it (task created by '$(((!)task).creator)', actor '$(actor)')");
- }
- ((!)task).status = STATUS_QUEUED;
- yield db.update_async<TaskEntity>((!)task);
- yield pulse.bump(id, "task enqueued");
- }
- /**
- * The cycle check for `enqueue_after_task_id` (human-created
- * tasks only, settable/clearable while the dependent is
- * draft): following the predecessor chain from
- * `predecessor_id` must never reach `task_id` — a cycle would
- * strand every task on it in draft forever.
- */
- public async void validate_enqueue_after_cycle(int64 task_id,
- int64 predecessor_id) throws Error {
- if (predecessor_id <= 0) {
- return;
- }
- if (predecessor_id == task_id) {
- throw new Error(Quark.from_string("clanker-saddle"), 77,
- "A task cannot enqueue after itself");
- }
- var seen = new Invercargill.DataStructures.HashSet<string>();
- int64 current = predecessor_id;
- while (current > 0 && seen.add(current.to_string())) {
- if (current == task_id) {
- throw new Error(Quark.from_string("clanker-saddle"), 78,
- @"enqueue-after would form a cycle through task $(predecessor_id)");
- }
- var row = yield db.query<TaskEntity>()
- .where(expr("id == $0", elem<int64?>(current)))
- .first_async();
- if (row == null) {
- return;
- }
- current = ((!)row).enqueue_after_task_id;
- }
- if (!seen.has(task_id.to_string())) {
- return;
- }
- throw new Error(Quark.from_string("clanker-saddle"), 78,
- @"enqueue-after would form a cycle through task $(predecessor_id)");
- }
- /**
- * The auto-enqueue-after pass (plan 12): when a predecessor
- * reaches `done`, its draft dependents enqueue as their
- * creator (the pre-authorised enqueue). A failed or cancelled
- * predecessor leaves them in draft — the creator decides.
- */
- private async void enqueue_dependents_of(TaskEntity predecessor) throws Error {
- if (predecessor.status != STATUS_DONE) {
- return;
- }
- foreach (var dependent in yield db.query<TaskEntity>()
- .where(expr("enqueue_after_task_id == $0 && status == $1",
- elem<int64?>(predecessor.id), elem(STATUS_DRAFT)))
- .materialise_async()) {
- try {
- yield enqueue_task(dependent.id, dependent.creator);
- } catch (Error e) {
- warning("[ClankerSaddle] Auto-enqueue of task %lld after %lld failed: %s",
- dependent.id, predecessor.id, e.message);
- }
- }
- }
- /**
- * The row teardown every task deletion runs (the shared core
- * of the human Delete action and the goal cascade's quiet
- * delete): attached rules, capabilities, progress, comments
- * and blocks die with the task, predecessor references clear,
- * the row goes and the pulse fires.
- */
- private async void delete_task_rows(TaskEntity task) throws Error {
- foreach (var link in yield db.query<TaskRule>()
- .where(expr("task_id == $0", elem<int64?>(task.id)))
- .materialise_async()) {
- yield db.delete_async<TaskRule>(link);
- }
- foreach (var link in yield db.query<TaskCapability>()
- .where(expr("task_id == $0", elem<int64?>(task.id)))
- .materialise_async()) {
- yield db.delete_async<TaskCapability>(link);
- }
- foreach (var row in yield db.query<TaskProgress>()
- .where(expr("task_id == $0", elem<int64?>(task.id)))
- .materialise_async()) {
- yield db.delete_async<TaskProgress>(row);
- }
- foreach (var row in yield db.query<TaskComment>()
- .where(expr("task_id == $0", elem<int64?>(task.id)))
- .materialise_async()) {
- yield db.delete_async<TaskComment>(row);
- }
- foreach (var row in yield db.query<TaskBlock>()
- .where(expr("task_id == $0", elem<int64?>(task.id)))
- .materialise_async()) {
- yield db.delete_async<TaskBlock>(row);
- }
- yield clear_enqueue_after_references(task.id);
- yield db.delete_async<TaskEntity>(task);
- yield pulse.bump(task.id, "task deleted");
- }
- /**
- * The delete the goal cascade uses on its own draft tasks:
- * every attached row dies with the task and no dependents
- * guard applies (the whole goal is going away together).
- */
- public async void delete_task_quietly(int64 id) throws Error {
- var task = yield db.query<TaskEntity>()
- .where(expr("id == $0", elem<int64?>(id)))
- .first_async();
- if (task == null) {
- return;
- }
- yield delete_task_rows((!)task);
- }
- /**
- * The human-only Delete action: removes a DRAFT task. Any
- * auto-enqueue-after predecessor references pointing at it are
- * cleared first (a dangling predecessor would strand its
- * dependents in not-ready forever). Attached rows (comments,
- * blocks, progress, rules, capabilities) die with the task.
- */
- public async void delete_task(int64 id) throws Error {
- var task = yield db.query<TaskEntity>()
- .where(expr("id == $0", elem<int64?>(id)))
- .first_async();
- if (task == null) {
- return;
- }
- if (((!)task).status != STATUS_DRAFT) {
- throw new Error(Quark.from_string("clanker-saddle"), 53,
- "Only draft tasks can be deleted (cancel or retry handles the rest)");
- }
- yield delete_task_rows((!)task);
- }
- /**
- * Clears `enqueue_after_task_id` references pointing at
- * `id`: a deleted predecessor must not strand its dependents
- * (their readiness check would wait forever on a task that no
- * longer exists).
- */
- private async void clear_enqueue_after_references(int64 id) throws Error {
- foreach (var dependent in yield db.query<TaskEntity>()
- .where(expr("enqueue_after_task_id == $0", elem<int64?>(id)))
- .materialise_async()) {
- dependent.enqueue_after_task_id = 0;
- yield db.update_async<TaskEntity>(dependent);
- }
- }
- /**
- * Appends a progress line (`saddle progress-update`) to the task's
- * history and returns the new row.
- */
- public async TaskProgress add_progress(int64 task_id, string summary) throws Error {
- var row = new TaskProgress() {
- task_id = task_id,
- summary = summary,
- created_at = new DateTime.now_local()
- };
- yield db.insert_async<TaskProgress>(row);
- yield pulse.bump(task_id, "progress");
- return row;
- }
- /** The task's progress history, newest first. */
- public async ImmutableLot<TaskProgress> list_progress(int64 task_id) throws Error {
- return yield db.query<TaskProgress>()
- .where(expr("task_id == $0", elem<int64?>(task_id)))
- .order_by_desc(expr("id"))
- .materialise_async();
- }
- /**
- * Every task id whose sessions carry a `yeehaw` entry (stringly
- * keyed) — the batch presence check the task list and the
- * dashboard's task references badge from, one DISTINCT query
- * instead of one EXISTS per row.
- */
- public async Invercargill.DataStructures.HashSet<string> yeehaw_task_ids() throws Error {
- var ids = new Invercargill.DataStructures.HashSet<string>();
- var command = connection.create_command(
- "SELECT DISTINCT s.task_id AS task_id FROM sessions s"
- + " JOIN session_commands sc ON sc.session_id = s.id"
- + " WHERE sc.type = :kind");
- command.with_parameter<Invercargill.Element>("kind", elem(ENTRY_YEEHAW));
- foreach (var row in yield command.execute_query_async()) {
- var element = row.get("task_id");
- int64? value = 0;
- if (element != null && element.try_get_as<int64?>(out value) && value != null) {
- ids.add(((!)value).to_string());
- }
- }
- return ids;
- }
- /**
- * The task's yeehaw card data: the newest
- * {@link YEEHAW_CARD_MAX} celebrations (message + when, newest
- * first) plus the uncapped total. Reads the existing
- * `session_commands` rows through the sessions join — the card
- * stores nothing of its own.
- */
- public async YeehawFeed yeehaw_feed(int64 task_id) throws Error {
- var total_command = connection.create_command(
- "SELECT COUNT(*) AS total FROM session_commands sc"
- + " JOIN sessions s ON s.id = sc.session_id"
- + " WHERE s.task_id = :task AND sc.type = :kind");
- total_command.with_parameter<Invercargill.Element>("task", elem<int64?>(task_id));
- total_command.with_parameter<Invercargill.Element>("kind", elem(ENTRY_YEEHAW));
- var total = yield total_command.execute_scalar_async();
- int64? total_value = 0;
- if (total != null) {
- total.try_get_as<int64?>(out total_value);
- }
- var feed = new YeehawFeed() { total = total_value };
- var rows_command = connection.create_command(
- "SELECT sc.output_text AS message, sc.command_text AS command,"
- + " sc.started_at AS started, sc.completed_at AS completed"
- + " FROM session_commands sc JOIN sessions s ON s.id = sc.session_id"
- + " WHERE s.task_id = :task AND sc.type = :kind"
- + " ORDER BY sc.id DESC LIMIT :cap");
- rows_command.with_parameter<Invercargill.Element>("task", elem<int64?>(task_id));
- rows_command.with_parameter<Invercargill.Element>("kind", elem(ENTRY_YEEHAW));
- rows_command.with_parameter<Invercargill.Element>("cap", elem<int64?>(YEEHAW_CARD_MAX));
- foreach (var row in yield rows_command.execute_query_async()) {
- string message = string_of(row.get("message"));
- if (message.length == 0) {
- // The message rides the output field; the command
- // line is the fallback for empty-output pushes.
- message = string_of(row.get("command"));
- }
- DateTime? at = date_time_of(row.get("started"));
- if (at == null) {
- at = date_time_of(row.get("completed"));
- }
- feed.latest.add(new YeehawEntry() { message = message, at = at });
- }
- return feed;
- }
- /** A raw query column back as a string ("" when null/unreadable). */
- private static string string_of(Invercargill.Element? element) {
- if (element == null) {
- return "";
- }
- string value = "";
- return ((!)element).try_get_as<string>(out value) ? value : "";
- }
- /** A raw query column back as a DateTime (null when null/unreadable). */
- private static DateTime? date_time_of(Invercargill.Element? element) {
- if (element == null) {
- return null;
- }
- DateTime value;
- return ((!)element).try_get_as<DateTime>(out value) ? value : null;
- }
- /**
- * Appends a comment; `author` is 'agent', 'user:<name>' or
- * 'system' (the harness-authored bookkeeping rows — plan 15
- * C2's distinct system-event class, never agent content).
- */
- public async TaskComment add_comment(int64 task_id, string author, string body,
- string? quote = null) throws Error {
- var comment = new TaskComment() {
- task_id = task_id,
- author = author,
- body = body,
- quote = quote != null && quote.strip().length > 0 ? quote.strip() : null,
- created_at = new DateTime.now_local()
- };
- yield db.insert_async<TaskComment>(comment);
- yield pulse.bump(task_id, "comment");
- return comment;
- }
- /**
- * Records a harness/system event on the task's thread (plan
- * 15 C2): park/resume bookkeeping (aborts, requeues,
- * retries), the session-context restoration, unreachable
- * notices — authored 'system', rendered with the distinct
- * muted system styling, never an agent comment row.
- */
- public async TaskComment add_system_comment(int64 task_id, string body) throws Error {
- return yield add_comment(task_id, COMMENT_AUTHOR_SYSTEM, body);
- }
- /**
- * Appends a terminal entry to the task's open session so
- * conversation-level events (blocks, plans, progress, reviews,
- * human comments) appear in the streamed log. A no-op when the
- * task has no open session.
- */
- public async void log_session_entry(int64 task_id, string entry_type,
- string command, string? output) throws Error {
- var session = yield open_session_for_task(task_id);
- if (session == null) {
- return;
- }
- var entry = new SessionCommand() {
- session_id = ((!)session).id,
- command_text = command,
- output_text = output,
- entry_type = entry_type,
- started_at = new DateTime.now_local()
- };
- yield db.insert_async<SessionCommand>(entry);
- }
- /**
- * Raises a blocking prompt on a task — `question` (optionally with
- * multiple-choice `options` and a markdown `body`), `rfc` or
- * `review` — and flips it to 'blocked' (or 'review_requested')
- * while it waits for a human answer. Returns the new block.
- */
- public async TaskBlock add_block(int64 task_id, string question, string block_type = BLOCK_QUESTION,
- string[]? options = null, string? body = null,
- bool quiet = false) throws Error {
- string options_json = null;
- if (options != null && options.length > 0) {
- var array = new Json.Array();
- foreach (var option in options) {
- array.add_string_element(option);
- }
- options_json = Json.to_string(new Json.Node(Json.NodeType.ARRAY).init_array(array), false);
- }
- var block = new TaskBlock() {
- task_id = task_id,
- question = question,
- block_type = block_type.length > 0 ? block_type : BLOCK_QUESTION,
- options = options_json,
- body = body != null && body.length > 0 ? body : null,
- created_at = new DateTime.now_local()
- };
- yield db.insert_async<TaskBlock>(block);
- var task = yield db.query<TaskEntity>()
- .where(expr("id == $0", elem<int64?>(task_id)))
- .first_async();
- if (task != null && (((!)task).status == STATUS_ASSIGNED || ((!)task).status == STATUS_WORKING
- || ((!)task).status == STATUS_BLOCKED || ((!)task).status == STATUS_REVIEW_REQUESTED)) {
- // The park flip is a guarded update: a completion
- // interleaving the raise keeps its terminal state (the
- // read above already yielded — the entity is stale by
- // write time under load; plan 13 B2's sticky rule).
- var guard = connection.create_command(
- "UPDATE tasks SET status = :status WHERE id = :id"
- + " AND status IN ('assigned', 'working', 'blocked', 'review_requested')");
- guard.with_parameter<Invercargill.Element>("status",
- elem(block.block_type == BLOCK_REVIEW
- ? STATUS_REVIEW_REQUESTED : STATUS_BLOCKED));
- guard.with_parameter<Invercargill.Element>("id",
- elem<int64?>(task_id));
- yield guard.execute_non_query_async();
- }
- 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<TaskBlock>()
- .where(expr("id == $0", elem<int64?>(block_id)))
- .first_async();
- if (block == null) {
- throw new Error(Quark.from_string("clanker-saddle"), 5,
- @"No block exists with id $(block_id)");
- }
- ((!)block).answer = answer;
- ((!)block).answered_at = new DateTime.now_local();
- ((!)block).answered_by = answered_by;
- yield db.update_async<TaskBlock>((!)block);
- var task = yield db.query<TaskEntity>()
- .where(expr("id == $0", elem<int64?>(((!)block).task_id)))
- .first_async();
- if (task != null && (((!)task).status == STATUS_BLOCKED
- || ((!)task).status == STATUS_REVIEW_REQUESTED)) {
- var session = yield open_session_for_task(((!)block).task_id);
- if (session == null) {
- ((!)task).status = STATUS_WORKING;
- yield db.update_async<TaskEntity>((!)task);
- }
- }
- yield pulse.bump(((!)block).task_id, "block answered");
- }
- /**
- * The git-host compare URL for the task's branches —
- * `{git_url}/compare/{branch_from}...{task_branch}` — or "" when
- * the task has no working branch.
- */
- public async string diff_url_for_task(int64 task_id) throws Error {
- var task = yield db.query<TaskEntity>()
- .where(expr("id == $0", elem<int64?>(task_id)))
- .first_async();
- if (task == null || ((!)task).task_branch == null
- || ((!)task).task_branch.length == 0) {
- return "";
- }
- var project = yield db.query<Project>()
- .where(expr("id == $0", elem<int64?>(((!)task).project_id)))
- .first_async();
- if (project == null) {
- return "";
- }
- string base_branch = ((!)task).branch_from ?? ((!)project).default_branch;
- return @"$(https_git_url(((!)project).git_url))/compare/$(base_branch)...$(((!)task).task_branch)";
- }
- /**
- * Translates a project's git remote into its browsable HTTPS form
- * so compare links open in the web UI instead of an SSH checkout:
- * `ssh://git@host:port/owner/repo.git` and `git@host:owner/repo(.git)`
- * both become `https://host/owner/repo` (the port is dropped —
- * SSH-only ports like fabrica.unitatem.net:222 have no HTTPS
- * meaning). Anything already http(s) (or unrecognised) is kept
- * as-is, minus a trailing `.git`.
- */
- public static string https_git_url(string git_url) {
- string url = git_url.strip();
- if (url.has_prefix("http://") || url.has_prefix("https://")) {
- return url.has_suffix(".git") ? url.substring(0, url.length - 4) : url;
- }
- // ssh://git@host[:port]/owner/repo.git
- if (url.has_prefix("ssh://")) {
- string rest = url.substring("ssh://".length);
- if (rest.has_prefix("git@")) {
- rest = rest.substring("git@".length);
- }
- int slash = rest.index_of("/");
- if (slash < 0) {
- return url;
- }
- string host_port = rest.substring(0, slash);
- string path = rest.substring(slash);
- if (host_port.contains(":")) {
- // SSH-only ports (fabrica.unitatem.net:222) carry no
- // HTTPS meaning — the browse host is plain.
- host_port = host_port.split(":")[0];
- }
- if (path.has_suffix(".git")) {
- path = path.substring(0, path.length - 4);
- }
- return @"https://$(host_port)$(path)";
- }
- // git@host:owner/repo(.git) — scp-style (the colon sits before
- // any path slash)
- int at = url.index_of("@");
- int colon = url.index_of(":");
- int slash = url.index_of("/");
- if (at >= 0 && colon > at && (slash < 0 || colon < slash)) {
- string host = url.substring(at + 1, colon - at - 1);
- string path = url.substring(colon + 1);
- if (path.has_suffix(".git")) {
- path = path.substring(0, path.length - 4);
- }
- return @"https://$(host)/$(path)";
- }
- return url;
- }
- /**
- * Task ids currently `assigned`/`working` whose open session sits
- * on an offline or fenced saddle — the queue marks them
- * "saddle unreachable — requeueing" while recovery runs.
- */
- public async Invercargill.DataStructures.HashSet<string> unreachable_task_ids() throws Error {
- var ids = new Invercargill.DataStructures.HashSet<string>();
- foreach (var status in new string[] { STATUS_ASSIGNED, STATUS_WORKING }) {
- foreach (var task in yield db.query<TaskEntity>()
- .where(expr("status == $0", elem(status)))
- .materialise_async()) {
- var session = yield open_session_for_task(task.id);
- if (session == null) {
- continue;
- }
- var saddle = yield db.query<Saddle>()
- .where(expr("id == $0", elem<int64?>(((!)session).saddle_id)))
- .first_async();
- if (saddle != null && ((!)saddle).status == SADDLE_OFFLINE) {
- 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 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. `working_branch` records when non-empty (the
- * done push's branch report). 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? working_branch = null, string? empty_summary = null) throws Error {
- task.status = status;
- task.completed_at = new DateTime.now_local();
- if (summary != null && summary.length > 0) {
- task.summary = summary;
- } else if ((task.summary ?? "").length == 0 && empty_summary != null) {
- task.summary = empty_summary;
- }
- if (working_branch != null && working_branch.length > 0) {
- task.working_branch = working_branch;
- }
- yield db.update_async<TaskEntity>(task);
- }
- /**
- * The ONE park stamping path: `suspended` plus the
- * `suspended_at` stamp, stamped once per parking window so the
- * resume trigger only sees goal events that landed after the
- * suspension. No-op when already parked (the stamp survives).
- */
- public async void set_parked(int64 task_id) throws Error {
- var task = yield db.query<TaskEntity>()
- .where(expr("id == $0", elem<int64?>(task_id)))
- .first_async();
- if (task == null || ((!)task).status == STATUS_SUSPENDED) {
- return;
- }
- ((!)task).status = STATUS_SUSPENDED;
- ((!)task).suspended_at = new DateTime.now_local();
- yield db.update_async<TaskEntity>((!)task);
- yield pulse.bump(task_id, "task suspended");
- }
- /**
- * 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<TaskEntity>()
- .where(expr("id == $0", elem<int64?>(task_id)))
- .first_async();
- if (task == null) {
- throw new Error(Quark.from_string("clanker-saddle"), 8,
- @"No task exists with id $(task_id)");
- }
- if (((!)task).status == STATUS_DONE || ((!)task).status == STATUS_FAILED) {
- return;
- }
- yield mark_terminal((!)task, STATUS_DONE, summary, working_branch);
- yield release_task(task_id, outcome, false);
- yield pulse.bump(task_id, "task completed");
- // Web Push (plan 11 task E): completion notifies the users
- // holding a "Notify me" pref for this task.
- push.notify_opted_in(task_id, PushService.lifecycle_notification(
- ((!)task).title, task_id, "completed", summary));
- // Plan 12: the goal's narrative hears the terminal event
- // (a parked goal agent resumes on it), and a done
- // predecessor auto-enqueues its dependents.
- yield enqueue_dependents_of(((!)task));
- }
- /**
- * Cancels a task: status 'failed' with a cancellation summary. Any
- * open session is closed with outcome 'stopped' and its saddle
- * freed. Draft tasks are refused — they are deleted, not
- * cancelled (nothing ever ran). Idempotent on terminal tasks
- * (plan 13 B2): the first outcome stands.
- */
- public async void cancel_task(int64 task_id) throws Error {
- var task = yield db.query<TaskEntity>()
- .where(expr("id == $0", elem<int64?>(task_id)))
- .first_async();
- if (task == null) {
- throw new Error(Quark.from_string("clanker-saddle"), 9,
- @"No task exists with id $(task_id)");
- }
- if (((!)task).status == STATUS_DONE || ((!)task).status == STATUS_FAILED) {
- return;
- }
- if (((!)task).status == STATUS_DRAFT) {
- throw new Error(Quark.from_string("clanker-saddle"), 55,
- "Draft tasks are deleted, not cancelled");
- }
- yield mark_terminal((!)task, STATUS_FAILED, null, null, "Cancelled by user");
- yield release_task(task_id, OUTCOME_STOPPED, false);
- yield pulse.bump(task_id, "task cancelled");
- }
- /**
- * Records a saddle-reported failure: status 'failed' with the
- * daemon's message as summary, the open session closed with
- * outcome 'failed' and its saddle freed. Plan 16 S4
- * (terminal-text single representation): `tasks.summary` IS
- * the terminal cache — no duplicate agent comment is written
- * (the narrative renders `summary_lead`; the task page renders
- * the summary card). Terminal transitions project ONCE (plan
- * 13 B2): the live failure's honest burst — a dozen-plus
- * distinct `saddle failure` entry ids in one session — still
- * fails the task exactly once, with one summary write and one
- * goal event.
- */
- public async void fail_task(int64 task_id, string? summary) throws Error {
- var task = yield db.query<TaskEntity>()
- .where(expr("id == $0", elem<int64?>(task_id)))
- .first_async();
- if (task == null) {
- throw new Error(Quark.from_string("clanker-saddle"), 25,
- @"No task exists with id $(task_id)");
- }
- if (((!)task).status == STATUS_DONE || ((!)task).status == STATUS_FAILED) {
- return;
- }
- yield mark_terminal((!)task, STATUS_FAILED, summary, null,
- "Saddle reported a failure");
- yield release_task(task_id, OUTCOME_FAILED, false);
- yield pulse.bump(task_id, "task failed");
- // Web Push (plan 11 task E): an agent-reported failure
- // notifies the opted-in users (a human cancellation does
- // not — the cancellator already knows).
- push.notify_opted_in(task_id, PushService.lifecycle_notification(
- ((!)task).title, task_id, "failed", summary));
- }
- /** Withdraws the suspension stamp after a successful resume. */
- public async void clear_suspension(int64 task_id) throws Error {
- var task = yield db.query<TaskEntity>()
- .where(expr("id == $0", elem<int64?>(task_id)))
- .first_async();
- if (task == null || ((!)task).suspended_at == null) {
- return;
- }
- ((!)task).suspended_at = null;
- yield db.update_async<TaskEntity>((!)task);
- }
- /**
- * Whether any OTHER task sharing this task's working context
- * moved while it was parked — a sibling under the same parent,
- * or any task of the same project on the same shared working
- * branch. "Moved" means any recorded transition after the park
- * timestamp: a completion/failure (`completed_at`), a dispatch
- * (session start), or a progress entry. The park marker is the
- * suspension stamp, else the newest block's creation (the
- * blocked-at marker), else the session's last recorded
- * activity. False with no siblings or nothing moved — the wire's
- * `branch_advanced`, so the daemon can skip the pointless
- * `git pull --rebase` when the shared branch did not advance.
- */
- public async bool branch_advanced_since(int64 task_id) throws Error {
- var me = yield db.query<TaskEntity>()
- .where(expr("id == $0", elem<int64?>(task_id)))
- .first_async();
- if (me == null) {
- return false;
- }
- DateTime? park = ((!)me).suspended_at;
- if (park == null) {
- foreach (var block in yield db.query<TaskBlock>()
- .where(expr("task_id == $0", elem<int64?>(task_id)))
- .order_by(expr("id"))
- .materialise_async()) {
- if (block.created_at != null) {
- park = ((!)block).created_at;
- }
- }
- }
- if (park == null) {
- foreach (var session in yield db.query<Session>()
- .where(expr("task_id == $0", elem<int64?>(task_id)))
- .materialise_async()) {
- foreach (var entry in yield db.query<SessionCommand>()
- .where(expr("session_id == $0", elem<int64?>(session.id)))
- .materialise_async()) {
- if (entry.started_at != null
- && (park == null
- || ((!)entry.started_at).compare((!)park) > 0)) {
- park = (!)entry.started_at;
- }
- }
- }
- }
- if (park == null) {
- // Never parked (or nothing recorded): nothing could have
- // advanced past a marker that does not exist.
- return false;
- }
- foreach (var sibling in yield sibling_tasks((!)me)) {
- if (sibling.completed_at != null
- && ((!)sibling.completed_at).compare((!)park) > 0) {
- return true;
- }
- foreach (var session in yield db.query<Session>()
- .where(expr("task_id == $0", elem<int64?>(sibling.id)))
- .materialise_async()) {
- if (session.started_at != null
- && session.started_at.compare((!)park) > 0) {
- return true;
- }
- }
- foreach (var progress in yield db.query<TaskProgress>()
- .where(expr("task_id == $0", elem<int64?>(sibling.id)))
- .materialise_async()) {
- if (progress.created_at != null
- && progress.created_at.compare((!)park) > 0) {
- return true;
- }
- }
- }
- return false;
- }
- /**
- * The tasks sharing `task`'s working context (excluding it):
- * same-project tasks on the same non-empty working branch, and
- * — for a goal agent — every other task of its goal (their
- * completions merge into the goal branch, so the goal agent's
- * post-park pull must hear them).
- */
- private async ImmutableLot<TaskEntity> sibling_tasks(TaskEntity task) throws Error {
- var rows = new Vector<TaskEntity>();
- var seen = new Invercargill.DataStructures.HashSet<string>();
- if (task.goal_id > 0) {
- foreach (var sibling in yield db.query<TaskEntity>()
- .where(expr("goal_id == $0", elem<int64?>(task.goal_id)))
- .materialise_async()) {
- if (sibling.id != task.id && seen.add(sibling.id.to_string())) {
- rows.add(sibling);
- }
- }
- }
- if (task.task_branch != null && ((!)task.task_branch).length > 0) {
- foreach (var sibling in yield db.query<TaskEntity>()
- .where(expr("project_id == $0 && task_branch == $1",
- elem<int64?>(task.project_id), elem(((!)task).task_branch)))
- .materialise_async()) {
- if (sibling.id != task.id && seen.add(sibling.id.to_string())) {
- rows.add(sibling);
- }
- }
- }
- return rows.to_immutable_buffer();
- }
- /**
- * Returns a non-terminal task to the queue: closes its open
- * sessions (outcome 'stopped') and frees the saddle that was
- * running it — {@link release_task} with the requeue, the
- * single owner of the bookkeeping. Plan 14 B1: a parked
- * (suspended) task requeues too — the goal page's
- * requeue-to-resume reaches a goal agent parked on a dead
- * saddle; the suspension stamp clears so the next dispatch
- * starts a fresh parking window.
- */
- public async void requeue_task(int64 task_id) throws Error {
- yield release_task(task_id, OUTCOME_STOPPED, true);
- yield pulse.bump(task_id, "task requeued");
- }
- /**
- * Returns a failed task to the queue for another attempt.
- * Plan 16 S4: the failure summary is the terminal cache of the
- * attempt that earned it — retry clears it and copies it
- * NOWHERE (the narrative rendered its lead, the page rendered
- * its card; the next attempt writes a fresh one), so no
- * summary→comment duplicate joins the thread. `completed_at`
- * is withdrawn, any open session closes and its saddle frees —
- * the same bookkeeping as {@link requeue_task} (retry's flip
- * is unconditional: the attempt resets whatever it held),
- * plus the audit comment "Task retried by user".
- */
- public async void retry_task(int64 task_id) throws Error {
- var task = yield db.query<TaskEntity>()
- .where(expr("id == $0", elem<int64?>(task_id)))
- .first_async();
- if (task == null) {
- throw new Error(Quark.from_string("clanker-saddle"), 8,
- @"No task exists with id $(task_id)");
- }
- ((!)task).status = STATUS_QUEUED;
- ((!)task).summary = null;
- ((!)task).completed_at = null;
- yield db.update_async<TaskEntity>((!)task);
- yield release_task(task_id, OUTCOME_STOPPED, false);
- yield add_system_comment(task_id, "Task retried by user");
- yield pulse.bump(task_id, "task retried");
- }
- /**
- * Queued tasks whose auto-enqueue predecessor (if any) is
- * done, ordered by priority (1 first) then creation time —
- * the dispatcher's candidate list.
- */
- public async ImmutableLot<TaskEntity> get_ready_tasks() throws Error {
- var queued = yield db.query<TaskEntity>()
- .where(expr("status == $0", elem(STATUS_QUEUED)))
- .materialise_async();
- var all_tasks = yield db.query<TaskEntity>().materialise_async();
- var rows = new Vector<TaskEntity>();
- foreach (var task in queued) {
- bool ready = true;
- if (task.enqueue_after_task_id > 0) {
- // The auto-enqueue predecessor also gates dispatch:
- // a dependent that reached the queue some other way
- // (a predecessor set after a manual enqueue) still
- // waits for it to finish.
- 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<TaskComment> list_comments(int64 task_id) throws Error {
- return yield db.query<TaskComment>()
- .where(expr("task_id == $0", elem<int64?>(task_id)))
- .order_by(expr("id"))
- .materialise_async();
- }
- /** Blocks on a task, oldest first. */
- public async ImmutableLot<TaskBlock> list_blocks(int64 task_id) throws Error {
- return yield db.query<TaskBlock>()
- .where(expr("task_id == $0", elem<int64?>(task_id)))
- .order_by(expr("id"))
- .materialise_async();
- }
- /** The task's newest block, or null when it has none. */
- public async TaskBlock? newest_block(int64 task_id) throws Error {
- TaskBlock? newest = null;
- foreach (var block in yield list_blocks(task_id)) {
- newest = block;
- }
- return newest;
- }
- /**
- * True when the task's newest block exists and carries an
- * answer — a verdict is waiting to reach the parked workspace
- * (the answer-delivery pass, the affinity resume, or a later
- * assignment payload). Such a task must never be requeued: the
- * requeue closes the session the resume machinery needs.
- * Plan 15 C1: goal blocks live on the run task as single rows
- * — no mirror pair exists to fall out of sync, so the newest
- * row's own answer is the whole truth.
- */
- public async bool newest_block_answered(int64 task_id) throws Error {
- var newest = yield newest_block(task_id);
- return newest != null && ((!)newest).answer != null;
- }
- /** All commands recorded against a session, oldest first. */
- public async ImmutableLot<SessionCommand> list_commands(int64 session_id) throws Error {
- return yield db.query<SessionCommand>()
- .where(expr("session_id == $0", elem<int64?>(session_id)))
- .order_by(expr("id"))
- .materialise_async();
- }
- /**
- * True when the session carries raw PTY log rows — the terminal
- * view then renders the concatenated raw log instead of the
- * typed entry stream.
- */
- public async bool has_raw_log(int64 session_id) throws Error {
- var probe = yield db.query<SessionRawLog>()
- .where(expr("session_id == $0", elem<int64?>(session_id)))
- .first_async();
- return probe != null;
- }
- /**
- * The session still open for `task_id` (no `ended_at`), or null.
- * The null check runs in Vala: the expression layer cannot compare
- * columns against null.
- */
- public async Session? open_session_for_task(int64 task_id) throws Error {
- var sessions = yield db.query<Session>()
- .where(expr("task_id == $0", elem<int64?>(task_id)))
- .order_by(expr("id"))
- .materialise_async();
- foreach (var session in sessions) {
- if (session.ended_at == null) {
- return session;
- }
- }
- return null;
- }
- /**
- * The ONE release path (plan 16 R2, arch item 13): every path
- * that takes a task off its saddle runs through here. Closes
- * the task's open sessions with `outcome` and frees whatever
- * saddle row still runs it — one conditional UPDATE, atomic
- * against a concurrent ask allocation claiming a new task for
- * the saddle between a read and a full-row write. `requeue`
- * first returns a non-terminal task to the queue
- * (assigned/working/blocked/review_requested/suspended — the
- * suspension stamp withdraws so the next dispatch opens a
- * fresh parking window).
- *
- * The terminal callers (complete/cancel/fail) pass their own
- * outcome with the status already stamped; requeue_task and
- * the unregister path ride the requeue; retry_task does its
- * own unconditional flip (summary and completed_at die with
- * the attempt) and releases without one. The offline variant
- * ({@link SaddleService.mark_saddle_unreachable}) keeps its
- * own guarded OFFLINE write for the saddle row and calls this
- * after it — the idle-free below no-ops against that row (its
- * current_task_id already cleared under the entry-seen guard)
- * while the session close applies; its keep-parked gate (an
- * answered block awaiting delivery) skips the release
- * entirely.
- */
- public async void release_task(int64 task_id, string outcome, bool requeue) throws Error {
- if (requeue) {
- var task = yield db.query<TaskEntity>()
- .where(expr("id == $0", elem<int64?>(task_id)))
- .first_async();
- if (task != null && (((!)task).status == STATUS_ASSIGNED
- || ((!)task).status == STATUS_WORKING || ((!)task).status == STATUS_BLOCKED
- || ((!)task).status == STATUS_REVIEW_REQUESTED
- || ((!)task).status == STATUS_SUSPENDED)) {
- ((!)task).status = STATUS_QUEUED;
- if (((!)task).suspended_at != null) {
- ((!)task).suspended_at = null;
- }
- yield db.update_async<TaskEntity>((!)task);
- }
- }
- var sessions = yield db.query<Session>()
- .where(expr("task_id == $0", elem<int64?>(task_id)))
- .materialise_async();
- foreach (var session in sessions) {
- if (session.ended_at != null) {
- continue;
- }
- session.ended_at = new DateTime.now_local();
- session.outcome = outcome;
- yield db.update_async<Session>(session);
- }
- var command = connection.create_command(
- "UPDATE saddles SET status = '" + SADDLE_IDLE
- + "', current_task_id = 0, allocated_at = NULL"
- + " WHERE current_task_id = :task");
- command.with_parameter<Invercargill.Element>("task", elem<int64?>(task_id));
- yield command.execute_non_query_async();
- }
- }
- }
|