TaskService.vala 78 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713
  1. using Invercargill;
  2. using Invercargill.DataStructures;
  3. using Invercargill.Expressions;
  4. using InvercargillSql;
  5. using InvercargillSql.Orm;
  6. using Inversion;
  7. using ClankerSaddle.Data;
  8. namespace ClankerSaddle.Services {
  9. /** Valid task statuses (the `tasks.status` wire values). */
  10. public const string STATUS_DRAFT = "draft";
  11. public const string STATUS_QUEUED = "queued";
  12. public const string STATUS_ASSIGNED = "assigned";
  13. public const string STATUS_WORKING = "working";
  14. public const string STATUS_BLOCKED = "blocked";
  15. public const string STATUS_REVIEW_REQUESTED = "review_requested";
  16. public const string STATUS_SUSPENDED = "suspended";
  17. public const string STATUS_DONE = "done";
  18. public const string STATUS_FAILED = "failed";
  19. /** Valid {@link TaskBlock.block_type} values. */
  20. public const string BLOCK_QUESTION = "question";
  21. public const string BLOCK_RFC = "rfc";
  22. public const string BLOCK_REVIEW = "review";
  23. public const string BLOCK_PLAN = "plan";
  24. /**
  25. * The goal agent's completion report (plan 15 C1): raised on the
  26. * goal-run task by {@link GoalService.submit_completion} — the
  27. * goal's review kinds are plan and completion.
  28. */
  29. public const string BLOCK_COMPLETION = "completion";
  30. /**
  31. * The system-event comment author (plan 15 C2): harness-authored
  32. * bookkeeping rows (aborts, requeues, retries, session
  33. * restorations) — never agent content.
  34. */
  35. public const string COMMENT_AUTHOR_SYSTEM = "system";
  36. /**
  37. * Plan 16 S5 (U2): MECHANICAL harness bookkeeping that every
  38. * user-facing timeline filters out — the session-restore footnote
  39. * and the park-until notice (both remain in the raw terminal log,
  40. * which keeps everything). Matched by substring so the daemon's
  41. * [harness]-prefixed wordings and the web's own phrasing all stay
  42. * covered; substantive system rows (aborts, requeues, retries)
  43. * never match.
  44. */
  45. public static bool comment_is_mechanical(string body) {
  46. return body.contains("session context restored")
  47. || body.contains("parked until");
  48. }
  49. /**
  50. * The agent-celebration entry type (`saddle yeehaw <message>`): the
  51. * daemon pushes it as a round-results entry carrying the message in
  52. * the output field, so it stores like any other `session_commands`
  53. * row — nothing new is persisted for the badge or the card.
  54. */
  55. public const string ENTRY_YEEHAW = "yeehaw";
  56. /** The task page's yeehaw card cap (newest entries kept). */
  57. public const int YEEHAW_CARD_MAX = 3;
  58. /**
  59. * The stable verdict wordings (plan 11): they are stored as the
  60. * block answers AND ride the resuming allocation's
  61. * `resume_notice` verbatim, so agent and human always read the
  62. * same text.
  63. */
  64. public const string NOTICE_PLAN_APPROVED = "Plan approved. Proceed with implementation.";
  65. public const string NOTICE_PLAN_FEEDBACK = "Plan feedback received:";
  66. public const string NOTICE_REVIEW_APPROVED = "Review approved. Continue with the task.";
  67. public const string NOTICE_REVIEW_FEEDBACK = "Review feedback received:";
  68. /** The display label of a goal-run parked on a review-family block (U3). */
  69. public const string DISPLAY_AWAITING_REVIEW = "awaiting your review";
  70. /**
  71. * A task's computed presentation state (stored status + unanswered
  72. * blocks): `review requested` while a review block is unanswered
  73. * (even when the stored status is terminal — the post-mortem's
  74. * `done`-with-open-review case), `parked` for the suspended state
  75. * (a goal agent between events), else the stored status. Plan 16
  76. * S5 (U3): a GOAL-RUN task parked on a review-family block (plan
  77. * or completion review) reads "awaiting your review" — agreeing
  78. * with its goal's label, never a bare "blocked". Pure
  79. * presentation — the stored statuses and the dispatcher are
  80. * untouched.
  81. */
  82. public class TaskDisplay : Object {
  83. public string status { get; set; default = ""; }
  84. public bool terminal { get; set; }
  85. public bool failed { get; set; }
  86. public static TaskDisplay compute(TaskEntity task, bool review_pending,
  87. bool goal_review_pending = false) {
  88. var view = new TaskDisplay() {
  89. terminal = task.status == STATUS_DONE || task.status == STATUS_FAILED,
  90. failed = task.status == STATUS_FAILED
  91. };
  92. if (goal_review_pending && task.kind == TASK_KIND_GOAL_RUN) {
  93. view.status = DISPLAY_AWAITING_REVIEW;
  94. } else if (review_pending) {
  95. view.status = "review requested";
  96. } else if (task.status == STATUS_SUSPENDED) {
  97. view.status = "parked";
  98. } else {
  99. view.status = task.status;
  100. }
  101. return view;
  102. }
  103. }
  104. /**
  105. * The full task aggregate loaded by {@link TaskService.get_task}: the
  106. * task itself plus its comments, blocks, progress history and
  107. * session history.
  108. */
  109. public class TaskDetail : Object {
  110. public TaskEntity task { get; set; }
  111. public ImmutableLot<TaskComment> comments { get; set; }
  112. public ImmutableLot<TaskBlock> blocks { get; set; }
  113. public ImmutableLot<TaskProgress> progress { get; set; }
  114. public ImmutableLot<Session> sessions { get; set; }
  115. }
  116. /**
  117. * One agent celebration (`saddle yeehaw`) read back for display: the
  118. * message from the entry's output field and when it landed.
  119. */
  120. public class YeehawEntry : Object {
  121. public string message { get; set; default = ""; }
  122. public DateTime? at { get; set; }
  123. }
  124. /**
  125. * The task's celebration feed for the yeehaw card: the newest
  126. * {@link YEEHAW_CARD_MAX} entries plus the uncapped total (the
  127. * "+N more" hint derives from the two).
  128. */
  129. public class YeehawFeed : Object {
  130. public Vector<YeehawEntry> latest { get; set; default = new Vector<YeehawEntry>(); }
  131. public int64 total { get; set; }
  132. }
  133. /**
  134. * Task storage and lifecycle: creation, enqueue-after-aware
  135. * readiness, agent interaction primitives (comments, blocks,
  136. * progress, reviews) and completion. Saddle bookkeeping rows are
  137. * kept consistent on every terminal transition. The plan-12 task
  138. * graph is goal-shaped: no parents, no dependencies, no plans —
  139. * goals own their tasks.
  140. */
  141. public class TaskService : Object {
  142. private OrmSession db = inject<OrmSession>();
  143. private Connection connection = inject<Connection>();
  144. private StatePulse pulse = inject<StatePulse>();
  145. private PushService push = inject<PushService>();
  146. /**
  147. * Inserts a task (status 'draft' until its creator enqueues
  148. * it). `description`, `task_branch` and `branch_from` may be
  149. * null; `branch_from` prefills from the project's default
  150. * branch; priority clamps to 1–9 (5 is the default, 1 the
  151. * highest). The plan-12 shape: `kind` ("task"/"goal_run"),
  152. * `mode` (implement/investigate/review/test, orchestrate for
  153. * goal runs), `creator` ("human"/"goal" — enqueueing is
  154. * creator-only), `goal_id` (the owning goal) and
  155. * `enqueue_after_task_id` (human-created tasks only: the
  156. * auto-enqueue predecessor, cycle-checked by the caller).
  157. */
  158. public async TaskEntity create_task(int64 project_id, string title, string? description,
  159. int64 priority,
  160. string? task_branch = null, string? branch_from = null,
  161. string status = STATUS_DRAFT,
  162. string kind = TASK_KIND_TASK, string mode = "implement",
  163. string creator = CREATOR_HUMAN, int64 goal_id = 0,
  164. int64 enqueue_after_task_id = 0) throws Error {
  165. string resolved_branch_from = branch_from ?? "";
  166. if (resolved_branch_from.length == 0) {
  167. var project = yield db.query<Project>()
  168. .where(expr("id == $0", elem<int64?>(project_id)))
  169. .first_async();
  170. if (project != null) {
  171. resolved_branch_from = ((!)project).default_branch;
  172. }
  173. }
  174. string? resolved_task_branch = task_branch != null && task_branch.length > 0 ? task_branch : null;
  175. if (goal_id > 0 && kind == TASK_KIND_TASK) {
  176. // Goal worker tasks exist only under a live goal.
  177. var goal = yield db.query<Goal>()
  178. .where(expr("id == $0", elem<int64?>(goal_id)))
  179. .first_async();
  180. if (goal == null) {
  181. throw new Error(Quark.from_string("clanker-saddle"), 75,
  182. @"No goal exists with id $(goal_id)");
  183. }
  184. }
  185. var task = new TaskEntity() {
  186. project_id = project_id,
  187. title = title,
  188. description = description != null && description.length > 0 ? description : null,
  189. priority = priority.clamp(1, 9),
  190. status = status,
  191. task_branch = resolved_task_branch,
  192. branch_from = resolved_branch_from.length > 0 ? resolved_branch_from : null,
  193. kind = kind,
  194. mode = mode,
  195. creator = creator,
  196. goal_id = goal_id,
  197. enqueue_after_task_id = enqueue_after_task_id,
  198. created_at = new DateTime.now_local()
  199. };
  200. yield db.insert_async<TaskEntity>(task);
  201. if (goal_id > 0 && kind == TASK_KIND_TASK) {
  202. // Goal worker tasks run on ephemeral auto-named
  203. // branches cut from the goal branch (the daemon builds
  204. // the worktrees; the stored name feeds compare links).
  205. var goal = yield db.query<Goal>()
  206. .where(expr("id == $0", elem<int64?>(goal_id)))
  207. .first_async();
  208. if (goal != null) {
  209. task.task_branch = @"task/$(goal_id)-$(task.id)";
  210. task.branch_from = ((!)goal).branch_name;
  211. yield db.update_async<TaskEntity>(task);
  212. }
  213. }
  214. yield pulse.bump(task.id, "task created");
  215. return task;
  216. }
  217. /**
  218. * Loads the full aggregate for `id`: the task, its comments,
  219. * blocks, progress and sessions. Returns null when the task
  220. * does not exist.
  221. */
  222. public async TaskDetail? get_task(int64 id) throws Error {
  223. var task = yield db.query<TaskEntity>()
  224. .where(expr("id == $0", elem<int64?>(id)))
  225. .first_async();
  226. if (task == null) {
  227. return null;
  228. }
  229. return new TaskDetail() {
  230. task = (!)task,
  231. comments = yield list_comments(id),
  232. blocks = yield list_blocks(id),
  233. progress = yield list_progress(id),
  234. sessions = yield db.query<Session>()
  235. .where(expr("task_id == $0", elem<int64?>(id)))
  236. .order_by(expr("id"))
  237. .materialise_async()
  238. };
  239. }
  240. /**
  241. * Computed display states for a batch of tasks (all lists share
  242. * this), keyed by task id string. Unanswered review-family
  243. * blocks load in ONE batched query over the wanted tasks, not
  244. * per task. Plan 16 S5 (U3): a goal-run task with a pending
  245. * review-family block (plan/completion/review) reads "awaiting
  246. * your review" — the run-task card agrees with its goal.
  247. */
  248. public async Dictionary<string, TaskDisplay> display_map_for(
  249. Lot<TaskEntity> tasks) throws Error {
  250. var wanted = new Invercargill.DataStructures.HashSet<string>();
  251. var goal_runs = new Invercargill.DataStructures.HashSet<string>();
  252. foreach (var task in tasks) {
  253. wanted.add(task.id.to_string());
  254. if (task.kind == TASK_KIND_GOAL_RUN) {
  255. goal_runs.add(task.id.to_string());
  256. }
  257. }
  258. var review_pending = new Invercargill.DataStructures.HashSet<string>();
  259. var goal_review_pending = new Invercargill.DataStructures.HashSet<string>();
  260. if (wanted.length > 0) {
  261. foreach (var block in yield db.query<TaskBlock>()
  262. .where(id_in("task_id", wanted))
  263. .materialise_async()) {
  264. if (block.answered_at != null
  265. || !wanted.has(block.task_id.to_string())) {
  266. continue;
  267. }
  268. if (block.block_type == BLOCK_REVIEW) {
  269. review_pending.add(block.task_id.to_string());
  270. }
  271. if (block.block_type == BLOCK_REVIEW
  272. || block.block_type == BLOCK_COMPLETION
  273. || block.block_type == BLOCK_PLAN) {
  274. if (goal_runs.has(block.task_id.to_string())) {
  275. goal_review_pending.add(block.task_id.to_string());
  276. }
  277. }
  278. }
  279. }
  280. var map = new Dictionary<string, TaskDisplay>();
  281. foreach (var task in tasks) {
  282. map.set(task.id.to_string(),
  283. TaskDisplay.compute(task, review_pending.has(task.id.to_string()),
  284. goal_review_pending.has(task.id.to_string())));
  285. }
  286. return map;
  287. }
  288. /**
  289. * The display state of a single task — the detail page's path,
  290. * computed from its already-loaded aggregate (no extra queries).
  291. */
  292. public static TaskDisplay display_of_detail(TaskDetail detail) {
  293. bool review_pending = false;
  294. bool goal_review_pending = false;
  295. foreach (var block in detail.blocks) {
  296. if (block.answered_at != null) {
  297. continue;
  298. }
  299. if (block.block_type == BLOCK_REVIEW) {
  300. review_pending = true;
  301. }
  302. if (detail.task.kind == TASK_KIND_GOAL_RUN
  303. && (block.block_type == BLOCK_REVIEW
  304. || block.block_type == BLOCK_COMPLETION
  305. || block.block_type == BLOCK_PLAN)) {
  306. goal_review_pending = true;
  307. }
  308. }
  309. return TaskDisplay.compute(detail.task, review_pending, goal_review_pending);
  310. }
  311. /**
  312. * Lists tasks, optionally filtered by status and/or project, ordered
  313. * by priority (1 first) then creation time — filtered and sorted
  314. * in the query, one materialisation.
  315. */
  316. public async ImmutableLot<TaskEntity> list_tasks(string? status, int64? project_id) throws Error {
  317. var query = db.query<TaskEntity>();
  318. if (status != null && status.length > 0) {
  319. query.where(expr("status == $0", elem(status)));
  320. }
  321. if (project_id != null) {
  322. query.where(expr("project_id == $0", elem<int64?>(project_id)));
  323. }
  324. return yield query
  325. .order_by(expr("priority"))
  326. .order_by(expr("created_at"))
  327. .materialise_async();
  328. }
  329. /**
  330. * Every standalone human-workable task row (kind task, no
  331. * goal), any status, ordered like {@link list_tasks} — the
  332. * auto-enqueue-after dropdowns' option list, queried for
  333. * exactly the rows they show instead of every task in the
  334. * table.
  335. */
  336. public async ImmutableLot<TaskEntity> list_standalone_tasks() throws Error {
  337. return yield db.query<TaskEntity>()
  338. .where(expr("kind == $0 && goal_id == $1",
  339. elem(TASK_KIND_TASK), elem<int64?>(0)))
  340. .order_by(expr("priority"))
  341. .order_by(expr("created_at"))
  342. .materialise_async();
  343. }
  344. /**
  345. * Saves the human-editable fields of the task identified by `id`
  346. * under the draft-lifecycle edit matrix: in DRAFT all fields
  347. * except status are editable (title, description/mandate,
  348. * priority, branch fields, the plan-12 work mode and —
  349. * human-created tasks only — the auto-enqueue predecessor,
  350. * cycle-validated); in non-draft non-terminal
  351. * states only priority applies; in terminal states nothing
  352. * does. Fields not editable in the current state are silently
  353. * left unchanged (null arguments always mean "leave
  354. * unchanged"; an empty branch field clears it; an
  355. * `enqueue_after_task_id` of -1 CLEARS the predecessor).
  356. * Returns null when the task does not exist. The pulse only
  357. * fires when a value actually moved, so periodic status
  358. * refreshes do not spam live pages.
  359. */
  360. public async TaskEntity? edit_task(int64 id, string? title, string? description,
  361. int64? priority, string? task_branch, string? branch_from,
  362. string? mode = null, int64? enqueue_after_task_id = null) throws Error {
  363. var task = yield db.query<TaskEntity>()
  364. .where(expr("id == $0", elem<int64?>(id)))
  365. .first_async();
  366. if (task == null) {
  367. return null;
  368. }
  369. bool changed = false;
  370. bool draft = ((!)task).status == STATUS_DRAFT;
  371. if (draft) {
  372. if (title != null && title.length > 0 && ((!)task).title != title) {
  373. ((!)task).title = title;
  374. changed = true;
  375. }
  376. if (description != null) {
  377. string? next_description = description.length > 0 ? description : null;
  378. if (((!)task).description != next_description) {
  379. ((!)task).description = next_description;
  380. changed = true;
  381. }
  382. }
  383. if (task_branch != null) {
  384. string? next_branch = task_branch.length > 0 ? task_branch : null;
  385. if (((!)task).task_branch != next_branch) {
  386. ((!)task).task_branch = next_branch;
  387. changed = true;
  388. }
  389. }
  390. if (branch_from != null) {
  391. string? next_from = branch_from.length > 0 ? branch_from : null;
  392. if (((!)task).branch_from != next_from) {
  393. ((!)task).branch_from = next_from;
  394. changed = true;
  395. }
  396. }
  397. if (mode != null && mode.length > 0) {
  398. if (!is_valid_mode(mode)) {
  399. throw new Error(Quark.from_string("clanker-saddle"), 79,
  400. @"Unknown mode '$(mode)' — valid modes: " + string.joinv(", ", TASK_MODES));
  401. }
  402. if (((!)task).mode != mode) {
  403. ((!)task).mode = mode;
  404. changed = true;
  405. }
  406. }
  407. // Human-created tasks only: the auto-enqueue
  408. // predecessor is settable/clearable while draft, and
  409. // setting it validates the chain (no cycles).
  410. if (enqueue_after_task_id != null
  411. && ((!)task).creator == CREATOR_HUMAN
  412. && ((!)task).kind == TASK_KIND_TASK) {
  413. int64 next = (!)enqueue_after_task_id;
  414. if (next < 0) {
  415. // The -1 sentinel clears the predecessor.
  416. next = 0;
  417. }
  418. if (next > 0) {
  419. yield validate_enqueue_after_cycle(id, next);
  420. }
  421. if (((!)task).enqueue_after_task_id != next) {
  422. ((!)task).enqueue_after_task_id = next;
  423. changed = true;
  424. }
  425. }
  426. }
  427. // Priority stays editable in every non-terminal state.
  428. if ((((!)task).status != STATUS_DONE && ((!)task).status != STATUS_FAILED)
  429. && priority != null) {
  430. int64 clamped = ((!)priority).clamp(1, 9);
  431. if (((!)task).priority != clamped) {
  432. ((!)task).priority = clamped;
  433. changed = true;
  434. }
  435. }
  436. yield db.update_async<TaskEntity>(task);
  437. if (changed) {
  438. yield pulse.bump(id, "task saved");
  439. }
  440. return task;
  441. }
  442. /**
  443. * The internal status-transition path (daemon pushes, park and
  444. * result machinery) — never reachable from a human form.
  445. * Stamps `completed_at` on done. Returns null when the task
  446. * does not exist.
  447. */
  448. public async TaskEntity? set_status(int64 id, string status) throws Error {
  449. var task = yield db.query<TaskEntity>()
  450. .where(expr("id == $0", elem<int64?>(id)))
  451. .first_async();
  452. if (task == null) {
  453. return null;
  454. }
  455. if (((!)task).status != status) {
  456. if (status != STATUS_DONE && status != STATUS_FAILED) {
  457. // Terminal states are sticky (plan 13 B2's
  458. // invariant): a NON-terminal transition must never
  459. // downgrade a done/failed task. The entity writes
  460. // below interleave with completions on the main
  461. // loop (the terminal checks in the callers have
  462. // already yielded by then — a late 'working' push
  463. // racing a 'done' push used to resurrect the task
  464. // as working AFTER its session closed done), so
  465. // the guard lives in the UPDATE's WHERE: atomic.
  466. // The row is re-read afterwards so the caller sees
  467. // the truth (terminal included), never a stale
  468. // entity it might write back.
  469. var guard = connection.create_command(
  470. "UPDATE tasks SET status = :status WHERE id = :id"
  471. + " AND status NOT IN ('done', 'failed')");
  472. guard.with_parameter<Invercargill.Element>("status",
  473. elem(status));
  474. guard.with_parameter<Invercargill.Element>("id",
  475. elem<int64?>(id));
  476. yield guard.execute_non_query_async();
  477. task = yield db.query<TaskEntity>()
  478. .where(expr("id == $0", elem<int64?>(id)))
  479. .first_async();
  480. if (task == null) {
  481. return null;
  482. }
  483. } else {
  484. ((!)task).status = status;
  485. if (status == STATUS_DONE && ((!)task).completed_at == null) {
  486. ((!)task).completed_at = new DateTime.now_local();
  487. }
  488. yield db.update_async<TaskEntity>((!)task);
  489. }
  490. yield pulse.bump(id, @"task $(status)");
  491. }
  492. return task;
  493. }
  494. /** Every task under `goal_id` (both kinds), oldest first. */
  495. public async ImmutableLot<TaskEntity> list_tasks_for_goal(int64 goal_id) throws Error {
  496. return yield db.query<TaskEntity>()
  497. .where(expr("goal_id == $0", elem<int64?>(goal_id)))
  498. .order_by(expr("id"))
  499. .materialise_async();
  500. }
  501. /**
  502. * The Enqueue action (creator-only, plan 12): moves a draft
  503. * task to the queue. Humans enqueue theirs, goal agents theirs
  504. * — an actor that is not the task's creator is refused (the
  505. * goal agent sequences its own tasks). A pending
  506. * auto-enqueue predecessor does not refuse the enqueue — the
  507. * dispatcher's readiness check holds dispatch back until it
  508. * finishes. The internal auto-enqueue-after pass calls this as
  509. * the dependent's creator (the creator's pre-authorisation).
  510. */
  511. public async void enqueue_task(int64 id, string actor = CREATOR_HUMAN) throws Error {
  512. var task = yield db.query<TaskEntity>()
  513. .where(expr("id == $0", elem<int64?>(id)))
  514. .first_async();
  515. if (task == null) {
  516. throw new Error(Quark.from_string("clanker-saddle"), 50,
  517. @"No task exists with id $(id)");
  518. }
  519. if (((!)task).status != STATUS_DRAFT) {
  520. throw new Error(Quark.from_string("clanker-saddle"), 51,
  521. "Only draft tasks can be enqueued");
  522. }
  523. if (((!)task).creator != actor) {
  524. throw new Error(Quark.from_string("clanker-saddle"), 76,
  525. @"Only the task's creator may enqueue it (task created by '$(((!)task).creator)', actor '$(actor)')");
  526. }
  527. ((!)task).status = STATUS_QUEUED;
  528. yield db.update_async<TaskEntity>((!)task);
  529. yield pulse.bump(id, "task enqueued");
  530. }
  531. /**
  532. * The cycle check for `enqueue_after_task_id` (human-created
  533. * tasks only, settable/clearable while the dependent is
  534. * draft): following the predecessor chain from
  535. * `predecessor_id` must never reach `task_id` — a cycle would
  536. * strand every task on it in draft forever.
  537. */
  538. public async void validate_enqueue_after_cycle(int64 task_id,
  539. int64 predecessor_id) throws Error {
  540. if (predecessor_id <= 0) {
  541. return;
  542. }
  543. if (predecessor_id == task_id) {
  544. throw new Error(Quark.from_string("clanker-saddle"), 77,
  545. "A task cannot enqueue after itself");
  546. }
  547. var seen = new Invercargill.DataStructures.HashSet<string>();
  548. int64 current = predecessor_id;
  549. while (current > 0 && seen.add(current.to_string())) {
  550. if (current == task_id) {
  551. throw new Error(Quark.from_string("clanker-saddle"), 78,
  552. @"enqueue-after would form a cycle through task $(predecessor_id)");
  553. }
  554. var row = yield db.query<TaskEntity>()
  555. .where(expr("id == $0", elem<int64?>(current)))
  556. .first_async();
  557. if (row == null) {
  558. return;
  559. }
  560. current = ((!)row).enqueue_after_task_id;
  561. }
  562. if (!seen.has(task_id.to_string())) {
  563. return;
  564. }
  565. throw new Error(Quark.from_string("clanker-saddle"), 78,
  566. @"enqueue-after would form a cycle through task $(predecessor_id)");
  567. }
  568. /**
  569. * The auto-enqueue-after pass (plan 12): when a predecessor
  570. * reaches `done`, its draft dependents enqueue as their
  571. * creator (the pre-authorised enqueue). A failed or cancelled
  572. * predecessor leaves them in draft — the creator decides.
  573. */
  574. private async void enqueue_dependents_of(TaskEntity predecessor) throws Error {
  575. if (predecessor.status != STATUS_DONE) {
  576. return;
  577. }
  578. foreach (var dependent in yield db.query<TaskEntity>()
  579. .where(expr("enqueue_after_task_id == $0 && status == $1",
  580. elem<int64?>(predecessor.id), elem(STATUS_DRAFT)))
  581. .materialise_async()) {
  582. try {
  583. yield enqueue_task(dependent.id, dependent.creator);
  584. } catch (Error e) {
  585. warning("[ClankerSaddle] Auto-enqueue of task %lld after %lld failed: %s",
  586. dependent.id, predecessor.id, e.message);
  587. }
  588. }
  589. }
  590. /**
  591. * The row teardown every task deletion runs (the shared core
  592. * of the human Delete action and the goal cascade's quiet
  593. * delete): attached rules, capabilities, progress, comments
  594. * and blocks die with the task, predecessor references clear,
  595. * the row goes and the pulse fires.
  596. */
  597. private async void delete_task_rows(TaskEntity task) throws Error {
  598. foreach (var link in yield db.query<TaskRule>()
  599. .where(expr("task_id == $0", elem<int64?>(task.id)))
  600. .materialise_async()) {
  601. yield db.delete_async<TaskRule>(link);
  602. }
  603. foreach (var link in yield db.query<TaskCapability>()
  604. .where(expr("task_id == $0", elem<int64?>(task.id)))
  605. .materialise_async()) {
  606. yield db.delete_async<TaskCapability>(link);
  607. }
  608. foreach (var row in yield db.query<TaskProgress>()
  609. .where(expr("task_id == $0", elem<int64?>(task.id)))
  610. .materialise_async()) {
  611. yield db.delete_async<TaskProgress>(row);
  612. }
  613. foreach (var row in yield db.query<TaskComment>()
  614. .where(expr("task_id == $0", elem<int64?>(task.id)))
  615. .materialise_async()) {
  616. yield db.delete_async<TaskComment>(row);
  617. }
  618. foreach (var row in yield db.query<TaskBlock>()
  619. .where(expr("task_id == $0", elem<int64?>(task.id)))
  620. .materialise_async()) {
  621. yield db.delete_async<TaskBlock>(row);
  622. }
  623. yield clear_enqueue_after_references(task.id);
  624. yield db.delete_async<TaskEntity>(task);
  625. yield pulse.bump(task.id, "task deleted");
  626. }
  627. /**
  628. * The delete the goal cascade uses on its own draft tasks:
  629. * every attached row dies with the task and no dependents
  630. * guard applies (the whole goal is going away together).
  631. */
  632. public async void delete_task_quietly(int64 id) throws Error {
  633. var task = yield db.query<TaskEntity>()
  634. .where(expr("id == $0", elem<int64?>(id)))
  635. .first_async();
  636. if (task == null) {
  637. return;
  638. }
  639. yield delete_task_rows((!)task);
  640. }
  641. /**
  642. * The human-only Delete action: removes a DRAFT task. Any
  643. * auto-enqueue-after predecessor references pointing at it are
  644. * cleared first (a dangling predecessor would strand its
  645. * dependents in not-ready forever). Attached rows (comments,
  646. * blocks, progress, rules, capabilities) die with the task.
  647. */
  648. public async void delete_task(int64 id) throws Error {
  649. var task = yield db.query<TaskEntity>()
  650. .where(expr("id == $0", elem<int64?>(id)))
  651. .first_async();
  652. if (task == null) {
  653. return;
  654. }
  655. if (((!)task).status != STATUS_DRAFT) {
  656. throw new Error(Quark.from_string("clanker-saddle"), 53,
  657. "Only draft tasks can be deleted (cancel or retry handles the rest)");
  658. }
  659. yield delete_task_rows((!)task);
  660. }
  661. /**
  662. * Clears `enqueue_after_task_id` references pointing at
  663. * `id`: a deleted predecessor must not strand its dependents
  664. * (their readiness check would wait forever on a task that no
  665. * longer exists).
  666. */
  667. private async void clear_enqueue_after_references(int64 id) throws Error {
  668. foreach (var dependent in yield db.query<TaskEntity>()
  669. .where(expr("enqueue_after_task_id == $0", elem<int64?>(id)))
  670. .materialise_async()) {
  671. dependent.enqueue_after_task_id = 0;
  672. yield db.update_async<TaskEntity>(dependent);
  673. }
  674. }
  675. /**
  676. * Appends a progress line (`saddle progress-update`) to the task's
  677. * history and returns the new row.
  678. */
  679. public async TaskProgress add_progress(int64 task_id, string summary) throws Error {
  680. var row = new TaskProgress() {
  681. task_id = task_id,
  682. summary = summary,
  683. created_at = new DateTime.now_local()
  684. };
  685. yield db.insert_async<TaskProgress>(row);
  686. yield pulse.bump(task_id, "progress");
  687. return row;
  688. }
  689. /** The task's progress history, newest first. */
  690. public async ImmutableLot<TaskProgress> list_progress(int64 task_id) throws Error {
  691. return yield db.query<TaskProgress>()
  692. .where(expr("task_id == $0", elem<int64?>(task_id)))
  693. .order_by_desc(expr("id"))
  694. .materialise_async();
  695. }
  696. /**
  697. * Every task id whose sessions carry a `yeehaw` entry (stringly
  698. * keyed) — the batch presence check the task list and the
  699. * dashboard's task references badge from, one DISTINCT query
  700. * instead of one EXISTS per row.
  701. */
  702. public async Invercargill.DataStructures.HashSet<string> yeehaw_task_ids() throws Error {
  703. var ids = new Invercargill.DataStructures.HashSet<string>();
  704. var command = connection.create_command(
  705. "SELECT DISTINCT s.task_id AS task_id FROM sessions s"
  706. + " JOIN session_commands sc ON sc.session_id = s.id"
  707. + " WHERE sc.type = :kind");
  708. command.with_parameter<Invercargill.Element>("kind", elem(ENTRY_YEEHAW));
  709. foreach (var row in yield command.execute_query_async()) {
  710. var element = row.get("task_id");
  711. int64? value = 0;
  712. if (element != null && element.try_get_as<int64?>(out value) && value != null) {
  713. ids.add(((!)value).to_string());
  714. }
  715. }
  716. return ids;
  717. }
  718. /**
  719. * The task's yeehaw card data: the newest
  720. * {@link YEEHAW_CARD_MAX} celebrations (message + when, newest
  721. * first) plus the uncapped total. Reads the existing
  722. * `session_commands` rows through the sessions join — the card
  723. * stores nothing of its own.
  724. */
  725. public async YeehawFeed yeehaw_feed(int64 task_id) throws Error {
  726. var total_command = connection.create_command(
  727. "SELECT COUNT(*) AS total FROM session_commands sc"
  728. + " JOIN sessions s ON s.id = sc.session_id"
  729. + " WHERE s.task_id = :task AND sc.type = :kind");
  730. total_command.with_parameter<Invercargill.Element>("task", elem<int64?>(task_id));
  731. total_command.with_parameter<Invercargill.Element>("kind", elem(ENTRY_YEEHAW));
  732. var total = yield total_command.execute_scalar_async();
  733. int64? total_value = 0;
  734. if (total != null) {
  735. total.try_get_as<int64?>(out total_value);
  736. }
  737. var feed = new YeehawFeed() { total = total_value };
  738. var rows_command = connection.create_command(
  739. "SELECT sc.output_text AS message, sc.command_text AS command,"
  740. + " sc.started_at AS started, sc.completed_at AS completed"
  741. + " FROM session_commands sc JOIN sessions s ON s.id = sc.session_id"
  742. + " WHERE s.task_id = :task AND sc.type = :kind"
  743. + " ORDER BY sc.id DESC LIMIT :cap");
  744. rows_command.with_parameter<Invercargill.Element>("task", elem<int64?>(task_id));
  745. rows_command.with_parameter<Invercargill.Element>("kind", elem(ENTRY_YEEHAW));
  746. rows_command.with_parameter<Invercargill.Element>("cap", elem<int64?>(YEEHAW_CARD_MAX));
  747. foreach (var row in yield rows_command.execute_query_async()) {
  748. string message = string_of(row.get("message"));
  749. if (message.length == 0) {
  750. // The message rides the output field; the command
  751. // line is the fallback for empty-output pushes.
  752. message = string_of(row.get("command"));
  753. }
  754. DateTime? at = date_time_of(row.get("started"));
  755. if (at == null) {
  756. at = date_time_of(row.get("completed"));
  757. }
  758. feed.latest.add(new YeehawEntry() { message = message, at = at });
  759. }
  760. return feed;
  761. }
  762. /** A raw query column back as a string ("" when null/unreadable). */
  763. private static string string_of(Invercargill.Element? element) {
  764. if (element == null) {
  765. return "";
  766. }
  767. string value = "";
  768. return ((!)element).try_get_as<string>(out value) ? value : "";
  769. }
  770. /** A raw query column back as a DateTime (null when null/unreadable). */
  771. private static DateTime? date_time_of(Invercargill.Element? element) {
  772. if (element == null) {
  773. return null;
  774. }
  775. DateTime value;
  776. return ((!)element).try_get_as<DateTime>(out value) ? value : null;
  777. }
  778. /**
  779. * Appends a comment; `author` is 'agent', 'user:<name>' or
  780. * 'system' (the harness-authored bookkeeping rows — plan 15
  781. * C2's distinct system-event class, never agent content).
  782. */
  783. public async TaskComment add_comment(int64 task_id, string author, string body,
  784. string? quote = null) throws Error {
  785. var comment = new TaskComment() {
  786. task_id = task_id,
  787. author = author,
  788. body = body,
  789. quote = quote != null && quote.strip().length > 0 ? quote.strip() : null,
  790. created_at = new DateTime.now_local()
  791. };
  792. yield db.insert_async<TaskComment>(comment);
  793. yield pulse.bump(task_id, "comment");
  794. return comment;
  795. }
  796. /**
  797. * Records a harness/system event on the task's thread (plan
  798. * 15 C2): park/resume bookkeeping (aborts, requeues,
  799. * retries), the session-context restoration, unreachable
  800. * notices — authored 'system', rendered with the distinct
  801. * muted system styling, never an agent comment row.
  802. */
  803. public async TaskComment add_system_comment(int64 task_id, string body) throws Error {
  804. return yield add_comment(task_id, COMMENT_AUTHOR_SYSTEM, body);
  805. }
  806. /**
  807. * Appends a terminal entry to the task's open session so
  808. * conversation-level events (blocks, plans, progress, reviews,
  809. * human comments) appear in the streamed log. A no-op when the
  810. * task has no open session.
  811. */
  812. public async void log_session_entry(int64 task_id, string entry_type,
  813. string command, string? output) throws Error {
  814. var session = yield open_session_for_task(task_id);
  815. if (session == null) {
  816. return;
  817. }
  818. var entry = new SessionCommand() {
  819. session_id = ((!)session).id,
  820. command_text = command,
  821. output_text = output,
  822. entry_type = entry_type,
  823. started_at = new DateTime.now_local()
  824. };
  825. yield db.insert_async<SessionCommand>(entry);
  826. }
  827. /**
  828. * Raises a blocking prompt on a task — `question` (optionally with
  829. * multiple-choice `options` and a markdown `body`), `rfc` or
  830. * `review` — and flips it to 'blocked' (or 'review_requested')
  831. * while it waits for a human answer. Returns the new block.
  832. */
  833. public async TaskBlock add_block(int64 task_id, string question, string block_type = BLOCK_QUESTION,
  834. string[]? options = null, string? body = null,
  835. bool quiet = false) throws Error {
  836. string options_json = null;
  837. if (options != null && options.length > 0) {
  838. var array = new Json.Array();
  839. foreach (var option in options) {
  840. array.add_string_element(option);
  841. }
  842. options_json = Json.to_string(new Json.Node(Json.NodeType.ARRAY).init_array(array), false);
  843. }
  844. var block = new TaskBlock() {
  845. task_id = task_id,
  846. question = question,
  847. block_type = block_type.length > 0 ? block_type : BLOCK_QUESTION,
  848. options = options_json,
  849. body = body != null && body.length > 0 ? body : null,
  850. created_at = new DateTime.now_local()
  851. };
  852. yield db.insert_async<TaskBlock>(block);
  853. var task = yield db.query<TaskEntity>()
  854. .where(expr("id == $0", elem<int64?>(task_id)))
  855. .first_async();
  856. if (task != null && (((!)task).status == STATUS_ASSIGNED || ((!)task).status == STATUS_WORKING
  857. || ((!)task).status == STATUS_BLOCKED || ((!)task).status == STATUS_REVIEW_REQUESTED)) {
  858. // The park flip is a guarded update: a completion
  859. // interleaving the raise keeps its terminal state (the
  860. // read above already yielded — the entity is stale by
  861. // write time under load; plan 13 B2's sticky rule).
  862. var guard = connection.create_command(
  863. "UPDATE tasks SET status = :status WHERE id = :id"
  864. + " AND status IN ('assigned', 'working', 'blocked', 'review_requested')");
  865. guard.with_parameter<Invercargill.Element>("status",
  866. elem(block.block_type == BLOCK_REVIEW
  867. ? STATUS_REVIEW_REQUESTED : STATUS_BLOCKED));
  868. guard.with_parameter<Invercargill.Element>("id",
  869. elem<int64?>(task_id));
  870. yield guard.execute_non_query_async();
  871. // Plan 16 D4: a raised block parks the round — the
  872. // session row CLOSES 'parked' right here (every raise
  873. // path funnels through this flip, entry-driven raises
  874. // included, so no parked task ever keeps a dangling
  875. // open session). The workspace awaits its resuming
  876. // allocation on a fresh row.
  877. yield park_open_sessions(task_id);
  878. }
  879. yield pulse.bump(task_id, "block raised");
  880. // Web Push (plan 11 task E): every raised block notifies all
  881. // push-subscribed users globally — the goal agent's mirrored
  882. // blocks stay quiet (the goal-level block owns the
  883. // notification with the goal's title and URL).
  884. if (!quiet && task != null) {
  885. push.notify_global(PushService.block_notification(
  886. ((!)task).title, task_id, block.block_type, block.question));
  887. }
  888. return block;
  889. }
  890. /**
  891. * Records a review request (`saddle review`): a review-type block
  892. * carrying the summary markdown, with the task flipped to
  893. * 'review_requested' while it waits for the human verdict.
  894. */
  895. public async TaskBlock request_review(int64 task_id, string summary) throws Error {
  896. return yield add_block(task_id, "Review requested", BLOCK_REVIEW, null,
  897. summary.length > 0 ? summary : null);
  898. }
  899. /**
  900. * Records a human's answer to a block. A parked round (its
  901. * session closed 'parked' at the raise — plan 16 D4) stays
  902. * 'blocked'/'review_requested': the dispatcher resumes the
  903. * paused workspace and hands the answer over. Only a block on
  904. * a task that NEVER RAN (no session rows at all) flips
  905. * straight back to 'working'.
  906. */
  907. public async void answer_block(int64 block_id, string answer,
  908. string? answered_by = null) throws Error {
  909. var block = yield db.query<TaskBlock>()
  910. .where(expr("id == $0", elem<int64?>(block_id)))
  911. .first_async();
  912. if (block == null) {
  913. throw new Error(Quark.from_string("clanker-saddle"), 5,
  914. @"No block exists with id $(block_id)");
  915. }
  916. ((!)block).answer = answer;
  917. ((!)block).answered_at = new DateTime.now_local();
  918. ((!)block).answered_by = answered_by;
  919. yield db.update_async<TaskBlock>((!)block);
  920. var task = yield db.query<TaskEntity>()
  921. .where(expr("id == $0", elem<int64?>(((!)block).task_id)))
  922. .first_async();
  923. if (task != null && (((!)task).status == STATUS_BLOCKED
  924. || ((!)task).status == STATUS_REVIEW_REQUESTED)) {
  925. var session = yield open_session_for_task(((!)block).task_id);
  926. if (session == null
  927. && (yield newest_session_for_task(((!)block).task_id)) == null) {
  928. ((!)task).status = STATUS_WORKING;
  929. yield db.update_async<TaskEntity>((!)task);
  930. }
  931. }
  932. yield pulse.bump(((!)block).task_id, "block answered");
  933. }
  934. /**
  935. * The git-host compare URL for the task's branches —
  936. * `{git_url}/compare/{branch_from}...{task_branch}` — or "" when
  937. * the task has no working branch.
  938. */
  939. public async string diff_url_for_task(int64 task_id) throws Error {
  940. var task = yield db.query<TaskEntity>()
  941. .where(expr("id == $0", elem<int64?>(task_id)))
  942. .first_async();
  943. if (task == null || ((!)task).task_branch == null
  944. || ((!)task).task_branch.length == 0) {
  945. return "";
  946. }
  947. var project = yield db.query<Project>()
  948. .where(expr("id == $0", elem<int64?>(((!)task).project_id)))
  949. .first_async();
  950. if (project == null) {
  951. return "";
  952. }
  953. string base_branch = ((!)task).branch_from ?? ((!)project).default_branch;
  954. return @"$(https_git_url(((!)project).git_url))/compare/$(base_branch)...$(((!)task).task_branch)";
  955. }
  956. /**
  957. * Translates a project's git remote into its browsable HTTPS form
  958. * so compare links open in the web UI instead of an SSH checkout:
  959. * `ssh://git@host:port/owner/repo.git` and `git@host:owner/repo(.git)`
  960. * both become `https://host/owner/repo` (the port is dropped —
  961. * SSH-only ports like fabrica.unitatem.net:222 have no HTTPS
  962. * meaning). Anything already http(s) (or unrecognised) is kept
  963. * as-is, minus a trailing `.git`.
  964. */
  965. public static string https_git_url(string git_url) {
  966. string url = git_url.strip();
  967. if (url.has_prefix("http://") || url.has_prefix("https://")) {
  968. return url.has_suffix(".git") ? url.substring(0, url.length - 4) : url;
  969. }
  970. // ssh://git@host[:port]/owner/repo.git
  971. if (url.has_prefix("ssh://")) {
  972. string rest = url.substring("ssh://".length);
  973. if (rest.has_prefix("git@")) {
  974. rest = rest.substring("git@".length);
  975. }
  976. int slash = rest.index_of("/");
  977. if (slash < 0) {
  978. return url;
  979. }
  980. string host_port = rest.substring(0, slash);
  981. string path = rest.substring(slash);
  982. if (host_port.contains(":")) {
  983. // SSH-only ports (fabrica.unitatem.net:222) carry no
  984. // HTTPS meaning — the browse host is plain.
  985. host_port = host_port.split(":")[0];
  986. }
  987. if (path.has_suffix(".git")) {
  988. path = path.substring(0, path.length - 4);
  989. }
  990. return @"https://$(host_port)$(path)";
  991. }
  992. // git@host:owner/repo(.git) — scp-style (the colon sits before
  993. // any path slash)
  994. int at = url.index_of("@");
  995. int colon = url.index_of(":");
  996. int slash = url.index_of("/");
  997. if (at >= 0 && colon > at && (slash < 0 || colon < slash)) {
  998. string host = url.substring(at + 1, colon - at - 1);
  999. string path = url.substring(colon + 1);
  1000. if (path.has_suffix(".git")) {
  1001. path = path.substring(0, path.length - 4);
  1002. }
  1003. return @"https://$(host)/$(path)";
  1004. }
  1005. return url;
  1006. }
  1007. /**
  1008. * Task ids currently `assigned`/`working` whose open session sits
  1009. * on an offline or fenced saddle — the queue marks them
  1010. * "saddle unreachable — requeueing" while recovery runs.
  1011. * Batched: the candidate tasks load once, their sessions in
  1012. * one query, the saddles in another.
  1013. */
  1014. public async Invercargill.DataStructures.HashSet<string> unreachable_task_ids() throws Error {
  1015. var ids = new Invercargill.DataStructures.HashSet<string>();
  1016. var candidate_ids = new Invercargill.DataStructures.HashSet<string>();
  1017. foreach (var status in new string[] { STATUS_ASSIGNED, STATUS_WORKING }) {
  1018. foreach (var task in yield db.query<TaskEntity>()
  1019. .where(expr("status == $0", elem(status)))
  1020. .materialise_async()) {
  1021. candidate_ids.add(task.id.to_string());
  1022. }
  1023. }
  1024. if (candidate_ids.length == 0) {
  1025. return ids;
  1026. }
  1027. // The newest open session per task: sessions ordered by id
  1028. // ascending, the LAST open row per task wins (the
  1029. // pre-batch semantics) — walked backwards, first hit per
  1030. // task keeps the slot.
  1031. var row_tasks = new Vector<string>();
  1032. var row_saddles = new Vector<string>();
  1033. foreach (var session in yield db.query<Session>()
  1034. .where(id_in("task_id", candidate_ids))
  1035. .order_by(expr("id"))
  1036. .materialise_async()) {
  1037. if (session.ended_at == null) {
  1038. row_tasks.add(session.task_id.to_string());
  1039. row_saddles.add(session.saddle_id.to_string());
  1040. }
  1041. }
  1042. var open_task_ids = new Vector<string>();
  1043. var open_saddle_ids = new Vector<string>();
  1044. var taken_tasks = new Invercargill.DataStructures.HashSet<string>();
  1045. for (int index = (int) row_tasks.length - 1; index >= 0; index--) {
  1046. if (taken_tasks.add(row_tasks.get(index))) {
  1047. open_task_ids.add(row_tasks.get(index));
  1048. open_saddle_ids.add(row_saddles.get(index));
  1049. }
  1050. }
  1051. if (open_task_ids.length == 0) {
  1052. return ids;
  1053. }
  1054. var saddle_ids = new Invercargill.DataStructures.HashSet<string>();
  1055. foreach (var saddle_id in open_saddle_ids) {
  1056. saddle_ids.add(saddle_id);
  1057. }
  1058. var offline = new Invercargill.DataStructures.HashSet<string>();
  1059. foreach (var saddle in yield db.query<Saddle>()
  1060. .where(id_in("id", saddle_ids))
  1061. .materialise_async()) {
  1062. if (saddle.status == SADDLE_OFFLINE) {
  1063. offline.add(saddle.id.to_string());
  1064. }
  1065. }
  1066. for (int index = 0; index < open_task_ids.length; index++) {
  1067. if (offline.has(open_saddle_ids.get(index))) {
  1068. ids.add(open_task_ids.get(index));
  1069. }
  1070. }
  1071. return ids;
  1072. }
  1073. /**
  1074. * An OR-chain of `column == id` comparisons for a batch of ids
  1075. * (the grammar has no IN operator), or null for an empty
  1076. * batch — the WHERE-in the batched lookups build.
  1077. */
  1078. private static Expression? id_in(string column,
  1079. Invercargill.DataStructures.HashSet<string> ids) throws Error {
  1080. Expression? chain = null;
  1081. foreach (var id in ids) {
  1082. var comparison = expr(column + " == $0", elem(id));
  1083. chain = chain == null
  1084. ? comparison : new BinaryExpression((!)chain, comparison,
  1085. BinaryOperator.OR);
  1086. }
  1087. return chain;
  1088. }
  1089. /**
  1090. * The resume_notice wording for an answered block (the plan-11
  1091. * stable verdicts): review and plan answers are already stored
  1092. * in the stable wording and pass through verbatim; question
  1093. * replies are formatted the way the daemon used to format
  1094. * delivered answers ("Human selected: …" for option matches,
  1095. * "Human answered (other): …" otherwise); RFC/plain answers
  1096. * carry the "Human answer: …" prefix.
  1097. */
  1098. public static string verdict_notice_for(TaskBlock block) {
  1099. string answer = block.answer ?? "";
  1100. if (block.block_type == BLOCK_REVIEW || block.block_type == BLOCK_PLAN) {
  1101. return answer;
  1102. }
  1103. if (block.block_type == BLOCK_QUESTION) {
  1104. foreach (var option in EntryJson.parse_string_array(block.options)) {
  1105. if (option == answer) {
  1106. return "Human selected: " + answer;
  1107. }
  1108. }
  1109. return "Human answered (other): " + answer;
  1110. }
  1111. return "Human answer: " + answer;
  1112. }
  1113. /**
  1114. * The ONE terminal stamping path (complete/fail/cancel): owns
  1115. * the status flip, the `completed_at` stamp and the summary
  1116. * precedence — a non-empty `summary` replaces whatever the
  1117. * row holds, else `empty_summary` applies only when the row
  1118. * holds nothing. Callers keep their own idempotence guard
  1119. * (terminal states are sticky, plan 13 B2).
  1120. */
  1121. private async void mark_terminal(TaskEntity task, string status, string? summary,
  1122. string? empty_summary = null) throws Error {
  1123. task.status = status;
  1124. task.completed_at = new DateTime.now_local();
  1125. if (summary != null && summary.length > 0) {
  1126. task.summary = summary;
  1127. } else if ((task.summary ?? "").length == 0 && empty_summary != null) {
  1128. task.summary = empty_summary;
  1129. }
  1130. yield db.update_async<TaskEntity>(task);
  1131. }
  1132. /**
  1133. * The ONE park stamping path: `suspended` plus the
  1134. * `suspended_at` stamp, stamped once per parking window so the
  1135. * resume trigger only sees goal events that landed after the
  1136. * suspension. No-op when already parked (the stamp survives).
  1137. * Plan 16 D4: parking closes the round's open sessions
  1138. * ('parked') — the pause record the resume machinery keys on
  1139. * is the newest session row, open or closed.
  1140. */
  1141. public async void set_parked(int64 task_id) throws Error {
  1142. var task = yield db.query<TaskEntity>()
  1143. .where(expr("id == $0", elem<int64?>(task_id)))
  1144. .first_async();
  1145. if (task == null || ((!)task).status == STATUS_SUSPENDED) {
  1146. return;
  1147. }
  1148. ((!)task).status = STATUS_SUSPENDED;
  1149. ((!)task).suspended_at = new DateTime.now_local();
  1150. yield db.update_async<TaskEntity>((!)task);
  1151. yield park_open_sessions(task_id);
  1152. yield pulse.bump(task_id, "task suspended");
  1153. }
  1154. /**
  1155. * Marks a task done: records `summary`, stamps `completed_at`,
  1156. * closes any open session (outcome `outcome`) and frees the
  1157. * saddle that was running it. Terminal transitions project ONCE
  1158. * (plan 13 B2): a task already done or failed keeps its first
  1159. * outcome — a redelivered push or a second terminal entry with
  1160. * a fresh id re-runs no bookkeeping, no comment, no event
  1161. * (retry_task is the one path that un-terminals).
  1162. */
  1163. public async void complete_task(int64 task_id, string? summary,
  1164. string outcome) throws Error {
  1165. var task = yield db.query<TaskEntity>()
  1166. .where(expr("id == $0", elem<int64?>(task_id)))
  1167. .first_async();
  1168. if (task == null) {
  1169. throw new Error(Quark.from_string("clanker-saddle"), 8,
  1170. @"No task exists with id $(task_id)");
  1171. }
  1172. if (((!)task).status == STATUS_DONE || ((!)task).status == STATUS_FAILED) {
  1173. return;
  1174. }
  1175. yield mark_terminal((!)task, STATUS_DONE, summary);
  1176. yield release_task(task_id, outcome, false);
  1177. yield pulse.bump(task_id, "task completed");
  1178. // Web Push (plan 11 task E): completion notifies the users
  1179. // holding a "Notify me" pref for this task.
  1180. push.notify_opted_in(task_id, PushService.lifecycle_notification(
  1181. ((!)task).title, task_id, "completed", summary));
  1182. // Plan 12: the goal's narrative hears the terminal event
  1183. // (a parked goal agent resumes on it), and a done
  1184. // predecessor auto-enqueues its dependents.
  1185. yield enqueue_dependents_of(((!)task));
  1186. }
  1187. /**
  1188. * Cancels a task: status 'failed' with a cancellation summary. Any
  1189. * open session is closed with outcome 'stopped' and its saddle
  1190. * freed. Draft tasks are refused — they are deleted, not
  1191. * cancelled (nothing ever ran). Idempotent on terminal tasks
  1192. * (plan 13 B2): the first outcome stands.
  1193. */
  1194. public async void cancel_task(int64 task_id) throws Error {
  1195. var task = yield db.query<TaskEntity>()
  1196. .where(expr("id == $0", elem<int64?>(task_id)))
  1197. .first_async();
  1198. if (task == null) {
  1199. throw new Error(Quark.from_string("clanker-saddle"), 9,
  1200. @"No task exists with id $(task_id)");
  1201. }
  1202. if (((!)task).status == STATUS_DONE || ((!)task).status == STATUS_FAILED) {
  1203. return;
  1204. }
  1205. if (((!)task).status == STATUS_DRAFT) {
  1206. throw new Error(Quark.from_string("clanker-saddle"), 55,
  1207. "Draft tasks are deleted, not cancelled");
  1208. }
  1209. yield mark_terminal((!)task, STATUS_FAILED, null, "Cancelled by user");
  1210. yield release_task(task_id, OUTCOME_STOPPED, false);
  1211. yield pulse.bump(task_id, "task cancelled");
  1212. }
  1213. /**
  1214. * Records a saddle-reported failure: status 'failed' with the
  1215. * daemon's message as summary, the open session closed with
  1216. * outcome 'failed' and its saddle freed. Plan 16 S4
  1217. * (terminal-text single representation): `tasks.summary` IS
  1218. * the terminal cache — no duplicate agent comment is written
  1219. * (the narrative renders `summary_lead`; the task page renders
  1220. * the summary card). Terminal transitions project ONCE (plan
  1221. * 13 B2): the live failure's honest burst — a dozen-plus
  1222. * distinct `saddle failure` entry ids in one session — still
  1223. * fails the task exactly once, with one summary write and one
  1224. * goal event.
  1225. */
  1226. public async void fail_task(int64 task_id, string? summary) throws Error {
  1227. var task = yield db.query<TaskEntity>()
  1228. .where(expr("id == $0", elem<int64?>(task_id)))
  1229. .first_async();
  1230. if (task == null) {
  1231. throw new Error(Quark.from_string("clanker-saddle"), 25,
  1232. @"No task exists with id $(task_id)");
  1233. }
  1234. if (((!)task).status == STATUS_DONE || ((!)task).status == STATUS_FAILED) {
  1235. return;
  1236. }
  1237. yield mark_terminal((!)task, STATUS_FAILED, summary,
  1238. "Saddle reported a failure");
  1239. yield release_task(task_id, OUTCOME_FAILED, false);
  1240. yield pulse.bump(task_id, "task failed");
  1241. // Web Push (plan 11 task E): an agent-reported failure
  1242. // notifies the opted-in users (a human cancellation does
  1243. // not — the cancellator already knows).
  1244. push.notify_opted_in(task_id, PushService.lifecycle_notification(
  1245. ((!)task).title, task_id, "failed", summary));
  1246. }
  1247. /** Withdraws the suspension stamp after a successful resume. */
  1248. public async void clear_suspension(int64 task_id) throws Error {
  1249. var task = yield db.query<TaskEntity>()
  1250. .where(expr("id == $0", elem<int64?>(task_id)))
  1251. .first_async();
  1252. if (task == null || ((!)task).suspended_at == null) {
  1253. return;
  1254. }
  1255. ((!)task).suspended_at = null;
  1256. yield db.update_async<TaskEntity>((!)task);
  1257. }
  1258. /**
  1259. * Whether any OTHER task sharing this task's working context
  1260. * moved while it was parked — a sibling under the same parent,
  1261. * or any task of the same project on the same shared working
  1262. * branch. "Moved" means any recorded transition after the park
  1263. * timestamp: a completion/failure (`completed_at`), a dispatch
  1264. * (session start), or a progress entry. The park marker is the
  1265. * suspension stamp, else the newest block's creation (the
  1266. * blocked-at marker), else the session's last recorded
  1267. * activity. False with no siblings or nothing moved — the wire's
  1268. * `branch_advanced`, so the daemon can skip the pointless
  1269. * `git pull --rebase` when the shared branch did not advance.
  1270. */
  1271. public async bool branch_advanced_since(int64 task_id) throws Error {
  1272. var me = yield db.query<TaskEntity>()
  1273. .where(expr("id == $0", elem<int64?>(task_id)))
  1274. .first_async();
  1275. if (me == null) {
  1276. return false;
  1277. }
  1278. DateTime? park = ((!)me).suspended_at;
  1279. if (park == null) {
  1280. foreach (var block in yield db.query<TaskBlock>()
  1281. .where(expr("task_id == $0", elem<int64?>(task_id)))
  1282. .order_by(expr("id"))
  1283. .materialise_async()) {
  1284. if (block.created_at != null) {
  1285. park = ((!)block).created_at;
  1286. }
  1287. }
  1288. }
  1289. if (park == null) {
  1290. foreach (var session in yield db.query<Session>()
  1291. .where(expr("task_id == $0", elem<int64?>(task_id)))
  1292. .materialise_async()) {
  1293. foreach (var entry in yield db.query<SessionCommand>()
  1294. .where(expr("session_id == $0", elem<int64?>(session.id)))
  1295. .materialise_async()) {
  1296. if (entry.started_at != null
  1297. && (park == null
  1298. || ((!)entry.started_at).compare((!)park) > 0)) {
  1299. park = (!)entry.started_at;
  1300. }
  1301. }
  1302. }
  1303. }
  1304. if (park == null) {
  1305. // Never parked (or nothing recorded): nothing could have
  1306. // advanced past a marker that does not exist.
  1307. return false;
  1308. }
  1309. foreach (var sibling in yield sibling_tasks((!)me)) {
  1310. if (sibling.completed_at != null
  1311. && ((!)sibling.completed_at).compare((!)park) > 0) {
  1312. return true;
  1313. }
  1314. foreach (var session in yield db.query<Session>()
  1315. .where(expr("task_id == $0", elem<int64?>(sibling.id)))
  1316. .materialise_async()) {
  1317. if (session.started_at != null
  1318. && session.started_at.compare((!)park) > 0) {
  1319. return true;
  1320. }
  1321. }
  1322. foreach (var progress in yield db.query<TaskProgress>()
  1323. .where(expr("task_id == $0", elem<int64?>(sibling.id)))
  1324. .materialise_async()) {
  1325. if (progress.created_at != null
  1326. && progress.created_at.compare((!)park) > 0) {
  1327. return true;
  1328. }
  1329. }
  1330. }
  1331. return false;
  1332. }
  1333. /**
  1334. * The tasks sharing `task`'s working context (excluding it):
  1335. * same-project tasks on the same non-empty working branch, and
  1336. * — for a goal agent — every other task of its goal (their
  1337. * completions merge into the goal branch, so the goal agent's
  1338. * post-park pull must hear them).
  1339. */
  1340. private async ImmutableLot<TaskEntity> sibling_tasks(TaskEntity task) throws Error {
  1341. var rows = new Vector<TaskEntity>();
  1342. var seen = new Invercargill.DataStructures.HashSet<string>();
  1343. if (task.goal_id > 0) {
  1344. foreach (var sibling in yield db.query<TaskEntity>()
  1345. .where(expr("goal_id == $0", elem<int64?>(task.goal_id)))
  1346. .materialise_async()) {
  1347. if (sibling.id != task.id && seen.add(sibling.id.to_string())) {
  1348. rows.add(sibling);
  1349. }
  1350. }
  1351. }
  1352. if (task.task_branch != null && ((!)task.task_branch).length > 0) {
  1353. foreach (var sibling in yield db.query<TaskEntity>()
  1354. .where(expr("project_id == $0 && task_branch == $1",
  1355. elem<int64?>(task.project_id), elem(((!)task).task_branch)))
  1356. .materialise_async()) {
  1357. if (sibling.id != task.id && seen.add(sibling.id.to_string())) {
  1358. rows.add(sibling);
  1359. }
  1360. }
  1361. }
  1362. return rows.to_immutable_buffer();
  1363. }
  1364. /**
  1365. * Returns a non-terminal task to the queue: closes its open
  1366. * sessions (outcome 'stopped') and frees the saddle that was
  1367. * running it — {@link release_task} with the requeue, the
  1368. * single owner of the bookkeeping. Plan 14 B1: a parked
  1369. * (suspended) task requeues too — the goal page's
  1370. * requeue-to-resume reaches a goal agent parked on a dead
  1371. * saddle; the suspension stamp clears so the next dispatch
  1372. * starts a fresh parking window.
  1373. */
  1374. public async void requeue_task(int64 task_id) throws Error {
  1375. yield release_task(task_id, OUTCOME_STOPPED, true);
  1376. yield pulse.bump(task_id, "task requeued");
  1377. }
  1378. /**
  1379. * Returns a failed task to the queue for another attempt.
  1380. * Plan 16 S4: the failure summary is the terminal cache of the
  1381. * attempt that earned it — retry clears it and copies it
  1382. * NOWHERE (the narrative rendered its lead, the page rendered
  1383. * its card; the next attempt writes a fresh one), so no
  1384. * summary→comment duplicate joins the thread. `completed_at`
  1385. * is withdrawn, any open session closes and its saddle frees —
  1386. * the same bookkeeping as {@link requeue_task} (retry's flip
  1387. * is unconditional: the attempt resets whatever it held),
  1388. * plus the audit comment "Task retried by user".
  1389. */
  1390. public async void retry_task(int64 task_id) throws Error {
  1391. var task = yield db.query<TaskEntity>()
  1392. .where(expr("id == $0", elem<int64?>(task_id)))
  1393. .first_async();
  1394. if (task == null) {
  1395. throw new Error(Quark.from_string("clanker-saddle"), 8,
  1396. @"No task exists with id $(task_id)");
  1397. }
  1398. ((!)task).status = STATUS_QUEUED;
  1399. ((!)task).summary = null;
  1400. ((!)task).completed_at = null;
  1401. yield db.update_async<TaskEntity>((!)task);
  1402. yield release_task(task_id, OUTCOME_STOPPED, false);
  1403. yield add_system_comment(task_id, "Task retried by user");
  1404. yield pulse.bump(task_id, "task retried");
  1405. }
  1406. /**
  1407. * Queued tasks whose auto-enqueue predecessor (if any) is
  1408. * done, ordered by priority (1 first) then creation time —
  1409. * the dispatcher's candidate list. Only the CANDIDATES'
  1410. * predecessors are queried (not every task row).
  1411. */
  1412. public async ImmutableLot<TaskEntity> get_ready_tasks() throws Error {
  1413. var queued = yield db.query<TaskEntity>()
  1414. .where(expr("status == $0", elem(STATUS_QUEUED)))
  1415. .order_by(expr("priority"))
  1416. .order_by(expr("created_at"))
  1417. .materialise_async();
  1418. var predecessor_ids = new Invercargill.DataStructures.HashSet<string>();
  1419. foreach (var task in queued) {
  1420. if (task.enqueue_after_task_id > 0) {
  1421. predecessor_ids.add(task.enqueue_after_task_id.to_string());
  1422. }
  1423. }
  1424. var predecessor_status = new Dictionary<string, string>();
  1425. if (predecessor_ids.length > 0) {
  1426. foreach (var predecessor in yield db.query<TaskEntity>()
  1427. .where(id_in("id", predecessor_ids))
  1428. .materialise_async()) {
  1429. predecessor_status.set(predecessor.id.to_string(), predecessor.status);
  1430. }
  1431. }
  1432. var rows = new Vector<TaskEntity>();
  1433. foreach (var task in queued) {
  1434. bool ready = true;
  1435. if (task.enqueue_after_task_id > 0) {
  1436. // The auto-enqueue predecessor also gates dispatch:
  1437. // a dependent that reached the queue some other way
  1438. // (a predecessor set after a manual enqueue) still
  1439. // waits for it to finish. A predecessor row that
  1440. // does not resolve keeps the dependent waiting.
  1441. string? status = predecessor_status.get_or_default(
  1442. task.enqueue_after_task_id.to_string());
  1443. ready = status != null && status == STATUS_DONE;
  1444. }
  1445. if (ready) {
  1446. rows.add(task);
  1447. }
  1448. }
  1449. return rows.to_immutable_buffer();
  1450. }
  1451. /** Comments on a task, oldest first. */
  1452. public async ImmutableLot<TaskComment> list_comments(int64 task_id) throws Error {
  1453. return yield db.query<TaskComment>()
  1454. .where(expr("task_id == $0", elem<int64?>(task_id)))
  1455. .order_by(expr("id"))
  1456. .materialise_async();
  1457. }
  1458. /** Blocks on a task, oldest first. */
  1459. public async ImmutableLot<TaskBlock> list_blocks(int64 task_id) throws Error {
  1460. return yield db.query<TaskBlock>()
  1461. .where(expr("task_id == $0", elem<int64?>(task_id)))
  1462. .order_by(expr("id"))
  1463. .materialise_async();
  1464. }
  1465. /** The task's newest block, or null when it has none. */
  1466. public async TaskBlock? newest_block(int64 task_id) throws Error {
  1467. TaskBlock? newest = null;
  1468. foreach (var block in yield list_blocks(task_id)) {
  1469. newest = block;
  1470. }
  1471. return newest;
  1472. }
  1473. /**
  1474. * True when the task's newest block exists and carries an
  1475. * answer — a verdict is waiting to reach the parked workspace
  1476. * (the answer-delivery pass, the affinity resume, or a later
  1477. * assignment payload). Such a task must never be requeued: the
  1478. * requeue closes the session the resume machinery needs.
  1479. * Plan 15 C1: goal blocks live on the run task as single rows
  1480. * — no mirror pair exists to fall out of sync, so the newest
  1481. * row's own answer is the whole truth.
  1482. */
  1483. public async bool newest_block_answered(int64 task_id) throws Error {
  1484. var newest = yield newest_block(task_id);
  1485. return newest != null && ((!)newest).answer != null;
  1486. }
  1487. /** All commands recorded against a session, oldest first. */
  1488. public async ImmutableLot<SessionCommand> list_commands(int64 session_id) throws Error {
  1489. return yield db.query<SessionCommand>()
  1490. .where(expr("session_id == $0", elem<int64?>(session_id)))
  1491. .order_by(expr("id"))
  1492. .materialise_async();
  1493. }
  1494. /**
  1495. * True when the session carries raw PTY log rows — the terminal
  1496. * view then renders the concatenated raw log instead of the
  1497. * typed entry stream.
  1498. */
  1499. public async bool has_raw_log(int64 session_id) throws Error {
  1500. var probe = yield db.query<SessionRawLog>()
  1501. .where(expr("session_id == $0", elem<int64?>(session_id)))
  1502. .first_async();
  1503. return probe != null;
  1504. }
  1505. /**
  1506. * The session still open for `task_id` (no `ended_at`), or null.
  1507. * The null check runs in Vala: the expression layer cannot compare
  1508. * columns against null.
  1509. */
  1510. public async Session? open_session_for_task(int64 task_id) throws Error {
  1511. var sessions = yield db.query<Session>()
  1512. .where(expr("task_id == $0", elem<int64?>(task_id)))
  1513. .order_by(expr("id"))
  1514. .materialise_async();
  1515. foreach (var session in sessions) {
  1516. if (session.ended_at == null) {
  1517. return session;
  1518. }
  1519. }
  1520. return null;
  1521. }
  1522. /**
  1523. * The task's newest session row, open or closed (plan 16 D4:
  1524. * a parked round's row is closed 'parked' — it is still the
  1525. * pause record the affinity resume, the session-scoped task
  1526. * CLI and the goal page's run surfaces key on), or null when
  1527. * the task never ran.
  1528. */
  1529. public async Session? newest_session_for_task(int64 task_id) throws Error {
  1530. Session? newest = null;
  1531. foreach (var session in yield db.query<Session>()
  1532. .where(expr("task_id == $0", elem<int64?>(task_id)))
  1533. .order_by(expr("id"))
  1534. .materialise_async()) {
  1535. newest = session;
  1536. }
  1537. return newest;
  1538. }
  1539. /**
  1540. * Closes the task's open sessions with outcome 'parked'
  1541. * (plan 16 D4 — the ONE park close every raise and suspend
  1542. * path runs; idempotent, nothing to do for rows already
  1543. * closed). The workspace stays on its saddle awaiting the
  1544. * resuming allocation that opens the successor row.
  1545. */
  1546. public async void park_open_sessions(int64 task_id) throws Error {
  1547. foreach (var session in yield db.query<Session>()
  1548. .where(expr("task_id == $0", elem<int64?>(task_id)))
  1549. .materialise_async()) {
  1550. if (session.ended_at != null) {
  1551. continue;
  1552. }
  1553. session.ended_at = new DateTime.now_local();
  1554. session.outcome = OUTCOME_PARKED;
  1555. yield db.update_async<Session>(session);
  1556. }
  1557. }
  1558. /**
  1559. * The ONE release path (plan 16 R2, arch item 13): every path
  1560. * that takes a task off its saddle runs through here. Closes
  1561. * the task's open sessions with `outcome` and frees whatever
  1562. * saddle row still runs it — one conditional UPDATE, atomic
  1563. * against a concurrent ask allocation claiming a new task for
  1564. * the saddle between a read and a full-row write. `requeue`
  1565. * first returns a non-terminal task to the queue
  1566. * (assigned/working/blocked/review_requested/suspended — the
  1567. * suspension stamp withdraws so the next dispatch opens a
  1568. * fresh parking window).
  1569. *
  1570. * The terminal callers (complete/cancel/fail) pass their own
  1571. * outcome with the status already stamped; requeue_task and
  1572. * the unregister path ride the requeue; retry_task does its
  1573. * own unconditional flip (summary and completed_at die with
  1574. * the attempt) and releases without one. The offline variant
  1575. * ({@link SaddleService.mark_saddle_unreachable}) keeps its
  1576. * own guarded OFFLINE write for the saddle row and calls this
  1577. * after it — the idle-free below no-ops against that row (its
  1578. * current_task_id already cleared under the entry-seen guard)
  1579. * while the session close applies; its keep-parked gate (an
  1580. * answered block awaiting delivery) skips the release
  1581. * entirely.
  1582. */
  1583. public async void release_task(int64 task_id, string outcome, bool requeue) throws Error {
  1584. if (requeue) {
  1585. var task = yield db.query<TaskEntity>()
  1586. .where(expr("id == $0", elem<int64?>(task_id)))
  1587. .first_async();
  1588. if (task != null && (((!)task).status == STATUS_ASSIGNED
  1589. || ((!)task).status == STATUS_WORKING || ((!)task).status == STATUS_BLOCKED
  1590. || ((!)task).status == STATUS_REVIEW_REQUESTED
  1591. || ((!)task).status == STATUS_SUSPENDED)) {
  1592. ((!)task).status = STATUS_QUEUED;
  1593. if (((!)task).suspended_at != null) {
  1594. ((!)task).suspended_at = null;
  1595. }
  1596. yield db.update_async<TaskEntity>((!)task);
  1597. }
  1598. }
  1599. var sessions = yield db.query<Session>()
  1600. .where(expr("task_id == $0", elem<int64?>(task_id)))
  1601. .materialise_async();
  1602. foreach (var session in sessions) {
  1603. if (session.ended_at != null) {
  1604. continue;
  1605. }
  1606. session.ended_at = new DateTime.now_local();
  1607. session.outcome = outcome;
  1608. yield db.update_async<Session>(session);
  1609. }
  1610. var command = connection.create_command(
  1611. "UPDATE saddles SET status = '" + SADDLE_IDLE
  1612. + "', current_task_id = 0, allocated_at = NULL"
  1613. + " WHERE current_task_id = :task");
  1614. command.with_parameter<Invercargill.Element>("task", elem<int64?>(task_id));
  1615. yield command.execute_non_query_async();
  1616. }
  1617. }
  1618. }