TaskService.vala 73 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588
  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 blocks load
  243. * once, not per task. Plan 16 S5 (U3): a goal-run task with a
  244. * pending review-family block (plan/completion/review) reads
  245. * "awaiting your review" — the run-task card agrees with its
  246. * 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. foreach (var family in new string[] { BLOCK_REVIEW, BLOCK_COMPLETION, BLOCK_PLAN }) {
  261. foreach (var block in yield db.query<TaskBlock>()
  262. .where(expr("type == $0", elem(family)))
  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 (goal_runs.has(block.task_id.to_string())) {
  272. goal_review_pending.add(block.task_id.to_string());
  273. }
  274. }
  275. }
  276. var map = new Dictionary<string, TaskDisplay>();
  277. foreach (var task in tasks) {
  278. map.set(task.id.to_string(),
  279. TaskDisplay.compute(task, review_pending.has(task.id.to_string()),
  280. goal_review_pending.has(task.id.to_string())));
  281. }
  282. return map;
  283. }
  284. /**
  285. * The display state of a single task — the detail page's path,
  286. * computed from its already-loaded aggregate (no extra queries).
  287. */
  288. public static TaskDisplay display_of_detail(TaskDetail detail) {
  289. bool review_pending = false;
  290. bool goal_review_pending = false;
  291. foreach (var block in detail.blocks) {
  292. if (block.answered_at != null) {
  293. continue;
  294. }
  295. if (block.block_type == BLOCK_REVIEW) {
  296. review_pending = true;
  297. }
  298. if (detail.task.kind == TASK_KIND_GOAL_RUN
  299. && (block.block_type == BLOCK_REVIEW
  300. || block.block_type == BLOCK_COMPLETION
  301. || block.block_type == BLOCK_PLAN)) {
  302. goal_review_pending = true;
  303. }
  304. }
  305. return TaskDisplay.compute(detail.task, review_pending, goal_review_pending);
  306. }
  307. /**
  308. * Lists tasks, optionally filtered by status and/or project, ordered
  309. * by priority (1 first) then creation time.
  310. */
  311. public async ImmutableLot<TaskEntity> list_tasks(string? status, int64? project_id) throws Error {
  312. var tasks = yield db.query<TaskEntity>().materialise_async();
  313. var rows = new Vector<TaskEntity>();
  314. foreach (var task in tasks) {
  315. if (status != null && status.length > 0 && task.status != status) {
  316. continue;
  317. }
  318. if (project_id != null && task.project_id != project_id) {
  319. continue;
  320. }
  321. rows.add(task);
  322. }
  323. var sorted = rows.sort((a, b) => {
  324. if (a.priority != b.priority) {
  325. return (int)(a.priority - b.priority);
  326. }
  327. int64 a_created = a.created_at != null ? ((!)a.created_at).to_unix() : 0;
  328. int64 b_created = b.created_at != null ? ((!)b.created_at).to_unix() : 0;
  329. return (int)(a_created - b_created);
  330. });
  331. return sorted.to_immutable_buffer();
  332. }
  333. /**
  334. * Saves the human-editable fields of the task identified by `id`
  335. * under the draft-lifecycle edit matrix: in DRAFT all fields
  336. * except status are editable (title, description/mandate,
  337. * priority, branch fields, the plan-12 work mode and —
  338. * human-created tasks only — the auto-enqueue predecessor,
  339. * cycle-validated); in non-draft non-terminal
  340. * states only priority applies; in terminal states nothing
  341. * does. Fields not editable in the current state are silently
  342. * left unchanged (null arguments always mean "leave
  343. * unchanged"; an empty branch field clears it; an
  344. * `enqueue_after_task_id` of -1 CLEARS the predecessor).
  345. * Returns null when the task does not exist. The pulse only
  346. * fires when a value actually moved, so periodic status
  347. * refreshes do not spam live pages.
  348. */
  349. public async TaskEntity? edit_task(int64 id, string? title, string? description,
  350. int64? priority, string? task_branch, string? branch_from,
  351. string? mode = null, int64? enqueue_after_task_id = null) throws Error {
  352. var task = yield db.query<TaskEntity>()
  353. .where(expr("id == $0", elem<int64?>(id)))
  354. .first_async();
  355. if (task == null) {
  356. return null;
  357. }
  358. bool changed = false;
  359. bool draft = ((!)task).status == STATUS_DRAFT;
  360. if (draft) {
  361. if (title != null && title.length > 0 && ((!)task).title != title) {
  362. ((!)task).title = title;
  363. changed = true;
  364. }
  365. if (description != null) {
  366. string? next_description = description.length > 0 ? description : null;
  367. if (((!)task).description != next_description) {
  368. ((!)task).description = next_description;
  369. changed = true;
  370. }
  371. }
  372. if (task_branch != null) {
  373. string? next_branch = task_branch.length > 0 ? task_branch : null;
  374. if (((!)task).task_branch != next_branch) {
  375. ((!)task).task_branch = next_branch;
  376. changed = true;
  377. }
  378. }
  379. if (branch_from != null) {
  380. string? next_from = branch_from.length > 0 ? branch_from : null;
  381. if (((!)task).branch_from != next_from) {
  382. ((!)task).branch_from = next_from;
  383. changed = true;
  384. }
  385. }
  386. if (mode != null && mode.length > 0) {
  387. if (!is_valid_mode(mode)) {
  388. throw new Error(Quark.from_string("clanker-saddle"), 79,
  389. @"Unknown mode '$(mode)' — valid modes: " + string.joinv(", ", TASK_MODES));
  390. }
  391. if (((!)task).mode != mode) {
  392. ((!)task).mode = mode;
  393. changed = true;
  394. }
  395. }
  396. // Human-created tasks only: the auto-enqueue
  397. // predecessor is settable/clearable while draft, and
  398. // setting it validates the chain (no cycles).
  399. if (enqueue_after_task_id != null
  400. && ((!)task).creator == CREATOR_HUMAN
  401. && ((!)task).kind == TASK_KIND_TASK) {
  402. int64 next = (!)enqueue_after_task_id;
  403. if (next < 0) {
  404. // The -1 sentinel clears the predecessor.
  405. next = 0;
  406. }
  407. if (next > 0) {
  408. yield validate_enqueue_after_cycle(id, next);
  409. }
  410. if (((!)task).enqueue_after_task_id != next) {
  411. ((!)task).enqueue_after_task_id = next;
  412. changed = true;
  413. }
  414. }
  415. }
  416. // Priority stays editable in every non-terminal state.
  417. if ((((!)task).status != STATUS_DONE && ((!)task).status != STATUS_FAILED)
  418. && priority != null) {
  419. int64 clamped = ((!)priority).clamp(1, 9);
  420. if (((!)task).priority != clamped) {
  421. ((!)task).priority = clamped;
  422. changed = true;
  423. }
  424. }
  425. yield db.update_async<TaskEntity>(task);
  426. if (changed) {
  427. yield pulse.bump(id, "task saved");
  428. }
  429. return task;
  430. }
  431. /**
  432. * The internal status-transition path (daemon pushes, park and
  433. * result machinery) — never reachable from a human form.
  434. * Stamps `completed_at` on done. Returns null when the task
  435. * does not exist.
  436. */
  437. public async TaskEntity? set_status(int64 id, string status) throws Error {
  438. var task = yield db.query<TaskEntity>()
  439. .where(expr("id == $0", elem<int64?>(id)))
  440. .first_async();
  441. if (task == null) {
  442. return null;
  443. }
  444. if (((!)task).status != status) {
  445. if (status != STATUS_DONE && status != STATUS_FAILED) {
  446. // Terminal states are sticky (plan 13 B2's
  447. // invariant): a NON-terminal transition must never
  448. // downgrade a done/failed task. The entity writes
  449. // below interleave with completions on the main
  450. // loop (the terminal checks in the callers have
  451. // already yielded by then — a late 'working' push
  452. // racing a 'done' push used to resurrect the task
  453. // as working AFTER its session closed done), so
  454. // the guard lives in the UPDATE's WHERE: atomic.
  455. // The row is re-read afterwards so the caller sees
  456. // the truth (terminal included), never a stale
  457. // entity it might write back.
  458. var guard = connection.create_command(
  459. "UPDATE tasks SET status = :status WHERE id = :id"
  460. + " AND status NOT IN ('done', 'failed')");
  461. guard.with_parameter<Invercargill.Element>("status",
  462. elem(status));
  463. guard.with_parameter<Invercargill.Element>("id",
  464. elem<int64?>(id));
  465. yield guard.execute_non_query_async();
  466. task = yield db.query<TaskEntity>()
  467. .where(expr("id == $0", elem<int64?>(id)))
  468. .first_async();
  469. if (task == null) {
  470. return null;
  471. }
  472. } else {
  473. ((!)task).status = status;
  474. if (status == STATUS_DONE && ((!)task).completed_at == null) {
  475. ((!)task).completed_at = new DateTime.now_local();
  476. }
  477. yield db.update_async<TaskEntity>((!)task);
  478. }
  479. yield pulse.bump(id, @"task $(status)");
  480. }
  481. return task;
  482. }
  483. /** Every task under `goal_id` (both kinds), oldest first. */
  484. public async ImmutableLot<TaskEntity> list_tasks_for_goal(int64 goal_id) throws Error {
  485. return yield db.query<TaskEntity>()
  486. .where(expr("goal_id == $0", elem<int64?>(goal_id)))
  487. .order_by(expr("id"))
  488. .materialise_async();
  489. }
  490. /**
  491. * The Enqueue action (creator-only, plan 12): moves a draft
  492. * task to the queue. Humans enqueue theirs, goal agents theirs
  493. * — an actor that is not the task's creator is refused (the
  494. * goal agent sequences its own tasks). A pending
  495. * auto-enqueue predecessor does not refuse the enqueue — the
  496. * dispatcher's readiness check holds dispatch back until it
  497. * finishes. The internal auto-enqueue-after pass calls this as
  498. * the dependent's creator (the creator's pre-authorisation).
  499. */
  500. public async void enqueue_task(int64 id, string actor = CREATOR_HUMAN) throws Error {
  501. var task = yield db.query<TaskEntity>()
  502. .where(expr("id == $0", elem<int64?>(id)))
  503. .first_async();
  504. if (task == null) {
  505. throw new Error(Quark.from_string("clanker-saddle"), 50,
  506. @"No task exists with id $(id)");
  507. }
  508. if (((!)task).status != STATUS_DRAFT) {
  509. throw new Error(Quark.from_string("clanker-saddle"), 51,
  510. "Only draft tasks can be enqueued");
  511. }
  512. if (((!)task).creator != actor) {
  513. throw new Error(Quark.from_string("clanker-saddle"), 76,
  514. @"Only the task's creator may enqueue it (task created by '$(((!)task).creator)', actor '$(actor)')");
  515. }
  516. ((!)task).status = STATUS_QUEUED;
  517. yield db.update_async<TaskEntity>((!)task);
  518. yield pulse.bump(id, "task enqueued");
  519. }
  520. /**
  521. * The cycle check for `enqueue_after_task_id` (human-created
  522. * tasks only, settable/clearable while the dependent is
  523. * draft): following the predecessor chain from
  524. * `predecessor_id` must never reach `task_id` — a cycle would
  525. * strand every task on it in draft forever.
  526. */
  527. public async void validate_enqueue_after_cycle(int64 task_id,
  528. int64 predecessor_id) throws Error {
  529. if (predecessor_id <= 0) {
  530. return;
  531. }
  532. if (predecessor_id == task_id) {
  533. throw new Error(Quark.from_string("clanker-saddle"), 77,
  534. "A task cannot enqueue after itself");
  535. }
  536. var seen = new Invercargill.DataStructures.HashSet<string>();
  537. int64 current = predecessor_id;
  538. while (current > 0 && seen.add(current.to_string())) {
  539. if (current == task_id) {
  540. throw new Error(Quark.from_string("clanker-saddle"), 78,
  541. @"enqueue-after would form a cycle through task $(predecessor_id)");
  542. }
  543. var row = yield db.query<TaskEntity>()
  544. .where(expr("id == $0", elem<int64?>(current)))
  545. .first_async();
  546. if (row == null) {
  547. return;
  548. }
  549. current = ((!)row).enqueue_after_task_id;
  550. }
  551. if (!seen.has(task_id.to_string())) {
  552. return;
  553. }
  554. throw new Error(Quark.from_string("clanker-saddle"), 78,
  555. @"enqueue-after would form a cycle through task $(predecessor_id)");
  556. }
  557. /**
  558. * The auto-enqueue-after pass (plan 12): when a predecessor
  559. * reaches `done`, its draft dependents enqueue as their
  560. * creator (the pre-authorised enqueue). A failed or cancelled
  561. * predecessor leaves them in draft — the creator decides.
  562. */
  563. private async void enqueue_dependents_of(TaskEntity predecessor) throws Error {
  564. if (predecessor.status != STATUS_DONE) {
  565. return;
  566. }
  567. foreach (var dependent in yield db.query<TaskEntity>()
  568. .where(expr("enqueue_after_task_id == $0 && status == $1",
  569. elem<int64?>(predecessor.id), elem(STATUS_DRAFT)))
  570. .materialise_async()) {
  571. try {
  572. yield enqueue_task(dependent.id, dependent.creator);
  573. } catch (Error e) {
  574. warning("[ClankerSaddle] Auto-enqueue of task %lld after %lld failed: %s",
  575. dependent.id, predecessor.id, e.message);
  576. }
  577. }
  578. }
  579. /**
  580. * The row teardown every task deletion runs (the shared core
  581. * of the human Delete action and the goal cascade's quiet
  582. * delete): attached rules, capabilities, progress, comments
  583. * and blocks die with the task, predecessor references clear,
  584. * the row goes and the pulse fires.
  585. */
  586. private async void delete_task_rows(TaskEntity task) throws Error {
  587. foreach (var link in yield db.query<TaskRule>()
  588. .where(expr("task_id == $0", elem<int64?>(task.id)))
  589. .materialise_async()) {
  590. yield db.delete_async<TaskRule>(link);
  591. }
  592. foreach (var link in yield db.query<TaskCapability>()
  593. .where(expr("task_id == $0", elem<int64?>(task.id)))
  594. .materialise_async()) {
  595. yield db.delete_async<TaskCapability>(link);
  596. }
  597. foreach (var row in yield db.query<TaskProgress>()
  598. .where(expr("task_id == $0", elem<int64?>(task.id)))
  599. .materialise_async()) {
  600. yield db.delete_async<TaskProgress>(row);
  601. }
  602. foreach (var row in yield db.query<TaskComment>()
  603. .where(expr("task_id == $0", elem<int64?>(task.id)))
  604. .materialise_async()) {
  605. yield db.delete_async<TaskComment>(row);
  606. }
  607. foreach (var row in yield db.query<TaskBlock>()
  608. .where(expr("task_id == $0", elem<int64?>(task.id)))
  609. .materialise_async()) {
  610. yield db.delete_async<TaskBlock>(row);
  611. }
  612. yield clear_enqueue_after_references(task.id);
  613. yield db.delete_async<TaskEntity>(task);
  614. yield pulse.bump(task.id, "task deleted");
  615. }
  616. /**
  617. * The delete the goal cascade uses on its own draft tasks:
  618. * every attached row dies with the task and no dependents
  619. * guard applies (the whole goal is going away together).
  620. */
  621. public async void delete_task_quietly(int64 id) throws Error {
  622. var task = yield db.query<TaskEntity>()
  623. .where(expr("id == $0", elem<int64?>(id)))
  624. .first_async();
  625. if (task == null) {
  626. return;
  627. }
  628. yield delete_task_rows((!)task);
  629. }
  630. /**
  631. * The human-only Delete action: removes a DRAFT task. Any
  632. * auto-enqueue-after predecessor references pointing at it are
  633. * cleared first (a dangling predecessor would strand its
  634. * dependents in not-ready forever). Attached rows (comments,
  635. * blocks, progress, rules, capabilities) die with the task.
  636. */
  637. public async void delete_task(int64 id) throws Error {
  638. var task = yield db.query<TaskEntity>()
  639. .where(expr("id == $0", elem<int64?>(id)))
  640. .first_async();
  641. if (task == null) {
  642. return;
  643. }
  644. if (((!)task).status != STATUS_DRAFT) {
  645. throw new Error(Quark.from_string("clanker-saddle"), 53,
  646. "Only draft tasks can be deleted (cancel or retry handles the rest)");
  647. }
  648. yield delete_task_rows((!)task);
  649. }
  650. /**
  651. * Clears `enqueue_after_task_id` references pointing at
  652. * `id`: a deleted predecessor must not strand its dependents
  653. * (their readiness check would wait forever on a task that no
  654. * longer exists).
  655. */
  656. private async void clear_enqueue_after_references(int64 id) throws Error {
  657. foreach (var dependent in yield db.query<TaskEntity>()
  658. .where(expr("enqueue_after_task_id == $0", elem<int64?>(id)))
  659. .materialise_async()) {
  660. dependent.enqueue_after_task_id = 0;
  661. yield db.update_async<TaskEntity>(dependent);
  662. }
  663. }
  664. /**
  665. * Appends a progress line (`saddle progress-update`) to the task's
  666. * history and returns the new row.
  667. */
  668. public async TaskProgress add_progress(int64 task_id, string summary) throws Error {
  669. var row = new TaskProgress() {
  670. task_id = task_id,
  671. summary = summary,
  672. created_at = new DateTime.now_local()
  673. };
  674. yield db.insert_async<TaskProgress>(row);
  675. yield pulse.bump(task_id, "progress");
  676. return row;
  677. }
  678. /** The task's progress history, newest first. */
  679. public async ImmutableLot<TaskProgress> list_progress(int64 task_id) throws Error {
  680. return yield db.query<TaskProgress>()
  681. .where(expr("task_id == $0", elem<int64?>(task_id)))
  682. .order_by_desc(expr("id"))
  683. .materialise_async();
  684. }
  685. /**
  686. * Every task id whose sessions carry a `yeehaw` entry (stringly
  687. * keyed) — the batch presence check the task list and the
  688. * dashboard's task references badge from, one DISTINCT query
  689. * instead of one EXISTS per row.
  690. */
  691. public async Invercargill.DataStructures.HashSet<string> yeehaw_task_ids() throws Error {
  692. var ids = new Invercargill.DataStructures.HashSet<string>();
  693. var command = connection.create_command(
  694. "SELECT DISTINCT s.task_id AS task_id FROM sessions s"
  695. + " JOIN session_commands sc ON sc.session_id = s.id"
  696. + " WHERE sc.type = :kind");
  697. command.with_parameter<Invercargill.Element>("kind", elem(ENTRY_YEEHAW));
  698. foreach (var row in yield command.execute_query_async()) {
  699. var element = row.get("task_id");
  700. int64? value = 0;
  701. if (element != null && element.try_get_as<int64?>(out value) && value != null) {
  702. ids.add(((!)value).to_string());
  703. }
  704. }
  705. return ids;
  706. }
  707. /**
  708. * The task's yeehaw card data: the newest
  709. * {@link YEEHAW_CARD_MAX} celebrations (message + when, newest
  710. * first) plus the uncapped total. Reads the existing
  711. * `session_commands` rows through the sessions join — the card
  712. * stores nothing of its own.
  713. */
  714. public async YeehawFeed yeehaw_feed(int64 task_id) throws Error {
  715. var total_command = connection.create_command(
  716. "SELECT COUNT(*) AS total FROM session_commands sc"
  717. + " JOIN sessions s ON s.id = sc.session_id"
  718. + " WHERE s.task_id = :task AND sc.type = :kind");
  719. total_command.with_parameter<Invercargill.Element>("task", elem<int64?>(task_id));
  720. total_command.with_parameter<Invercargill.Element>("kind", elem(ENTRY_YEEHAW));
  721. var total = yield total_command.execute_scalar_async();
  722. int64? total_value = 0;
  723. if (total != null) {
  724. total.try_get_as<int64?>(out total_value);
  725. }
  726. var feed = new YeehawFeed() { total = total_value };
  727. var rows_command = connection.create_command(
  728. "SELECT sc.output_text AS message, sc.command_text AS command,"
  729. + " sc.started_at AS started, sc.completed_at AS completed"
  730. + " FROM session_commands sc JOIN sessions s ON s.id = sc.session_id"
  731. + " WHERE s.task_id = :task AND sc.type = :kind"
  732. + " ORDER BY sc.id DESC LIMIT :cap");
  733. rows_command.with_parameter<Invercargill.Element>("task", elem<int64?>(task_id));
  734. rows_command.with_parameter<Invercargill.Element>("kind", elem(ENTRY_YEEHAW));
  735. rows_command.with_parameter<Invercargill.Element>("cap", elem<int64?>(YEEHAW_CARD_MAX));
  736. foreach (var row in yield rows_command.execute_query_async()) {
  737. string message = string_of(row.get("message"));
  738. if (message.length == 0) {
  739. // The message rides the output field; the command
  740. // line is the fallback for empty-output pushes.
  741. message = string_of(row.get("command"));
  742. }
  743. DateTime? at = date_time_of(row.get("started"));
  744. if (at == null) {
  745. at = date_time_of(row.get("completed"));
  746. }
  747. feed.latest.add(new YeehawEntry() { message = message, at = at });
  748. }
  749. return feed;
  750. }
  751. /** A raw query column back as a string ("" when null/unreadable). */
  752. private static string string_of(Invercargill.Element? element) {
  753. if (element == null) {
  754. return "";
  755. }
  756. string value = "";
  757. return ((!)element).try_get_as<string>(out value) ? value : "";
  758. }
  759. /** A raw query column back as a DateTime (null when null/unreadable). */
  760. private static DateTime? date_time_of(Invercargill.Element? element) {
  761. if (element == null) {
  762. return null;
  763. }
  764. DateTime value;
  765. return ((!)element).try_get_as<DateTime>(out value) ? value : null;
  766. }
  767. /**
  768. * Appends a comment; `author` is 'agent', 'user:<name>' or
  769. * 'system' (the harness-authored bookkeeping rows — plan 15
  770. * C2's distinct system-event class, never agent content).
  771. */
  772. public async TaskComment add_comment(int64 task_id, string author, string body,
  773. string? quote = null) throws Error {
  774. var comment = new TaskComment() {
  775. task_id = task_id,
  776. author = author,
  777. body = body,
  778. quote = quote != null && quote.strip().length > 0 ? quote.strip() : null,
  779. created_at = new DateTime.now_local()
  780. };
  781. yield db.insert_async<TaskComment>(comment);
  782. yield pulse.bump(task_id, "comment");
  783. return comment;
  784. }
  785. /**
  786. * Records a harness/system event on the task's thread (plan
  787. * 15 C2): park/resume bookkeeping (aborts, requeues,
  788. * retries), the session-context restoration, unreachable
  789. * notices — authored 'system', rendered with the distinct
  790. * muted system styling, never an agent comment row.
  791. */
  792. public async TaskComment add_system_comment(int64 task_id, string body) throws Error {
  793. return yield add_comment(task_id, COMMENT_AUTHOR_SYSTEM, body);
  794. }
  795. /**
  796. * Appends a terminal entry to the task's open session so
  797. * conversation-level events (blocks, plans, progress, reviews,
  798. * human comments) appear in the streamed log. A no-op when the
  799. * task has no open session.
  800. */
  801. public async void log_session_entry(int64 task_id, string entry_type,
  802. string command, string? output) throws Error {
  803. var session = yield open_session_for_task(task_id);
  804. if (session == null) {
  805. return;
  806. }
  807. var entry = new SessionCommand() {
  808. session_id = ((!)session).id,
  809. command_text = command,
  810. output_text = output,
  811. entry_type = entry_type,
  812. started_at = new DateTime.now_local()
  813. };
  814. yield db.insert_async<SessionCommand>(entry);
  815. }
  816. /**
  817. * Raises a blocking prompt on a task — `question` (optionally with
  818. * multiple-choice `options` and a markdown `body`), `rfc` or
  819. * `review` — and flips it to 'blocked' (or 'review_requested')
  820. * while it waits for a human answer. Returns the new block.
  821. */
  822. public async TaskBlock add_block(int64 task_id, string question, string block_type = BLOCK_QUESTION,
  823. string[]? options = null, string? body = null,
  824. bool quiet = false) throws Error {
  825. string options_json = null;
  826. if (options != null && options.length > 0) {
  827. var array = new Json.Array();
  828. foreach (var option in options) {
  829. array.add_string_element(option);
  830. }
  831. options_json = Json.to_string(new Json.Node(Json.NodeType.ARRAY).init_array(array), false);
  832. }
  833. var block = new TaskBlock() {
  834. task_id = task_id,
  835. question = question,
  836. block_type = block_type.length > 0 ? block_type : BLOCK_QUESTION,
  837. options = options_json,
  838. body = body != null && body.length > 0 ? body : null,
  839. created_at = new DateTime.now_local()
  840. };
  841. yield db.insert_async<TaskBlock>(block);
  842. var task = yield db.query<TaskEntity>()
  843. .where(expr("id == $0", elem<int64?>(task_id)))
  844. .first_async();
  845. if (task != null && (((!)task).status == STATUS_ASSIGNED || ((!)task).status == STATUS_WORKING
  846. || ((!)task).status == STATUS_BLOCKED || ((!)task).status == STATUS_REVIEW_REQUESTED)) {
  847. // The park flip is a guarded update: a completion
  848. // interleaving the raise keeps its terminal state (the
  849. // read above already yielded — the entity is stale by
  850. // write time under load; plan 13 B2's sticky rule).
  851. var guard = connection.create_command(
  852. "UPDATE tasks SET status = :status WHERE id = :id"
  853. + " AND status IN ('assigned', 'working', 'blocked', 'review_requested')");
  854. guard.with_parameter<Invercargill.Element>("status",
  855. elem(block.block_type == BLOCK_REVIEW
  856. ? STATUS_REVIEW_REQUESTED : STATUS_BLOCKED));
  857. guard.with_parameter<Invercargill.Element>("id",
  858. elem<int64?>(task_id));
  859. yield guard.execute_non_query_async();
  860. }
  861. yield pulse.bump(task_id, "block raised");
  862. // Web Push (plan 11 task E): every raised block notifies all
  863. // push-subscribed users globally — the goal agent's mirrored
  864. // blocks stay quiet (the goal-level block owns the
  865. // notification with the goal's title and URL).
  866. if (!quiet && task != null) {
  867. push.notify_global(PushService.block_notification(
  868. ((!)task).title, task_id, block.block_type, block.question));
  869. }
  870. return block;
  871. }
  872. /**
  873. * Records a review request (`saddle review`): a review-type block
  874. * carrying the summary markdown, with the task flipped to
  875. * 'review_requested' while it waits for the human verdict.
  876. */
  877. public async TaskBlock request_review(int64 task_id, string summary) throws Error {
  878. return yield add_block(task_id, "Review requested", BLOCK_REVIEW, null,
  879. summary.length > 0 ? summary : null);
  880. }
  881. /**
  882. * Records a human's answer to a block. With an open session the
  883. * task stays 'blocked'/'review_requested' — the dispatcher resumes
  884. * the paused workspace and hands the answer over; without one the
  885. * task flips straight back to 'working'.
  886. */
  887. public async void answer_block(int64 block_id, string answer,
  888. string? answered_by = null) throws Error {
  889. var block = yield db.query<TaskBlock>()
  890. .where(expr("id == $0", elem<int64?>(block_id)))
  891. .first_async();
  892. if (block == null) {
  893. throw new Error(Quark.from_string("clanker-saddle"), 5,
  894. @"No block exists with id $(block_id)");
  895. }
  896. ((!)block).answer = answer;
  897. ((!)block).answered_at = new DateTime.now_local();
  898. ((!)block).answered_by = answered_by;
  899. yield db.update_async<TaskBlock>((!)block);
  900. var task = yield db.query<TaskEntity>()
  901. .where(expr("id == $0", elem<int64?>(((!)block).task_id)))
  902. .first_async();
  903. if (task != null && (((!)task).status == STATUS_BLOCKED
  904. || ((!)task).status == STATUS_REVIEW_REQUESTED)) {
  905. var session = yield open_session_for_task(((!)block).task_id);
  906. if (session == null) {
  907. ((!)task).status = STATUS_WORKING;
  908. yield db.update_async<TaskEntity>((!)task);
  909. }
  910. }
  911. yield pulse.bump(((!)block).task_id, "block answered");
  912. }
  913. /**
  914. * The git-host compare URL for the task's branches —
  915. * `{git_url}/compare/{branch_from}...{task_branch}` — or "" when
  916. * the task has no working branch.
  917. */
  918. public async string diff_url_for_task(int64 task_id) throws Error {
  919. var task = yield db.query<TaskEntity>()
  920. .where(expr("id == $0", elem<int64?>(task_id)))
  921. .first_async();
  922. if (task == null || ((!)task).task_branch == null
  923. || ((!)task).task_branch.length == 0) {
  924. return "";
  925. }
  926. var project = yield db.query<Project>()
  927. .where(expr("id == $0", elem<int64?>(((!)task).project_id)))
  928. .first_async();
  929. if (project == null) {
  930. return "";
  931. }
  932. string base_branch = ((!)task).branch_from ?? ((!)project).default_branch;
  933. return @"$(https_git_url(((!)project).git_url))/compare/$(base_branch)...$(((!)task).task_branch)";
  934. }
  935. /**
  936. * Translates a project's git remote into its browsable HTTPS form
  937. * so compare links open in the web UI instead of an SSH checkout:
  938. * `ssh://git@host:port/owner/repo.git` and `git@host:owner/repo(.git)`
  939. * both become `https://host/owner/repo` (the port is dropped —
  940. * SSH-only ports like fabrica.unitatem.net:222 have no HTTPS
  941. * meaning). Anything already http(s) (or unrecognised) is kept
  942. * as-is, minus a trailing `.git`.
  943. */
  944. public static string https_git_url(string git_url) {
  945. string url = git_url.strip();
  946. if (url.has_prefix("http://") || url.has_prefix("https://")) {
  947. return url.has_suffix(".git") ? url.substring(0, url.length - 4) : url;
  948. }
  949. // ssh://git@host[:port]/owner/repo.git
  950. if (url.has_prefix("ssh://")) {
  951. string rest = url.substring("ssh://".length);
  952. if (rest.has_prefix("git@")) {
  953. rest = rest.substring("git@".length);
  954. }
  955. int slash = rest.index_of("/");
  956. if (slash < 0) {
  957. return url;
  958. }
  959. string host_port = rest.substring(0, slash);
  960. string path = rest.substring(slash);
  961. if (host_port.contains(":")) {
  962. // SSH-only ports (fabrica.unitatem.net:222) carry no
  963. // HTTPS meaning — the browse host is plain.
  964. host_port = host_port.split(":")[0];
  965. }
  966. if (path.has_suffix(".git")) {
  967. path = path.substring(0, path.length - 4);
  968. }
  969. return @"https://$(host_port)$(path)";
  970. }
  971. // git@host:owner/repo(.git) — scp-style (the colon sits before
  972. // any path slash)
  973. int at = url.index_of("@");
  974. int colon = url.index_of(":");
  975. int slash = url.index_of("/");
  976. if (at >= 0 && colon > at && (slash < 0 || colon < slash)) {
  977. string host = url.substring(at + 1, colon - at - 1);
  978. string path = url.substring(colon + 1);
  979. if (path.has_suffix(".git")) {
  980. path = path.substring(0, path.length - 4);
  981. }
  982. return @"https://$(host)/$(path)";
  983. }
  984. return url;
  985. }
  986. /**
  987. * Task ids currently `assigned`/`working` whose open session sits
  988. * on an offline or fenced saddle — the queue marks them
  989. * "saddle unreachable — requeueing" while recovery runs.
  990. */
  991. public async Invercargill.DataStructures.HashSet<string> unreachable_task_ids() throws Error {
  992. var ids = new Invercargill.DataStructures.HashSet<string>();
  993. foreach (var status in new string[] { STATUS_ASSIGNED, STATUS_WORKING }) {
  994. foreach (var task in yield db.query<TaskEntity>()
  995. .where(expr("status == $0", elem(status)))
  996. .materialise_async()) {
  997. var session = yield open_session_for_task(task.id);
  998. if (session == null) {
  999. continue;
  1000. }
  1001. var saddle = yield db.query<Saddle>()
  1002. .where(expr("id == $0", elem<int64?>(((!)session).saddle_id)))
  1003. .first_async();
  1004. if (saddle != null && ((!)saddle).status == SADDLE_OFFLINE) {
  1005. ids.add(task.id.to_string());
  1006. }
  1007. }
  1008. }
  1009. return ids;
  1010. }
  1011. /**
  1012. * The resume_notice wording for an answered block (the plan-11
  1013. * stable verdicts): review and plan answers are already stored
  1014. * in the stable wording and pass through verbatim; question
  1015. * replies are formatted the way the daemon used to format
  1016. * delivered answers ("Human selected: …" for option matches,
  1017. * "Human answered (other): …" otherwise); RFC/plain answers
  1018. * carry the "Human answer: …" prefix.
  1019. */
  1020. public static string verdict_notice_for(TaskBlock block) {
  1021. string answer = block.answer ?? "";
  1022. if (block.block_type == BLOCK_REVIEW || block.block_type == BLOCK_PLAN) {
  1023. return answer;
  1024. }
  1025. if (block.block_type == BLOCK_QUESTION) {
  1026. foreach (var option in EntryJson.parse_string_array(block.options)) {
  1027. if (option == answer) {
  1028. return "Human selected: " + answer;
  1029. }
  1030. }
  1031. return "Human answered (other): " + answer;
  1032. }
  1033. return "Human answer: " + answer;
  1034. }
  1035. /**
  1036. * The ONE terminal stamping path (complete/fail/cancel): owns
  1037. * the status flip, the `completed_at` stamp and the summary
  1038. * precedence — a non-empty `summary` replaces whatever the
  1039. * row holds, else `empty_summary` applies only when the row
  1040. * holds nothing. `working_branch` records when non-empty (the
  1041. * done push's branch report). Callers keep their own
  1042. * idempotence guard (terminal states are sticky, plan 13 B2).
  1043. */
  1044. private async void mark_terminal(TaskEntity task, string status, string? summary,
  1045. string? working_branch = null, string? empty_summary = null) throws Error {
  1046. task.status = status;
  1047. task.completed_at = new DateTime.now_local();
  1048. if (summary != null && summary.length > 0) {
  1049. task.summary = summary;
  1050. } else if ((task.summary ?? "").length == 0 && empty_summary != null) {
  1051. task.summary = empty_summary;
  1052. }
  1053. if (working_branch != null && working_branch.length > 0) {
  1054. task.working_branch = working_branch;
  1055. }
  1056. yield db.update_async<TaskEntity>(task);
  1057. }
  1058. /**
  1059. * The ONE park stamping path: `suspended` plus the
  1060. * `suspended_at` stamp, stamped once per parking window so the
  1061. * resume trigger only sees goal events that landed after the
  1062. * suspension. No-op when already parked (the stamp survives).
  1063. */
  1064. public async void set_parked(int64 task_id) throws Error {
  1065. var task = yield db.query<TaskEntity>()
  1066. .where(expr("id == $0", elem<int64?>(task_id)))
  1067. .first_async();
  1068. if (task == null || ((!)task).status == STATUS_SUSPENDED) {
  1069. return;
  1070. }
  1071. ((!)task).status = STATUS_SUSPENDED;
  1072. ((!)task).suspended_at = new DateTime.now_local();
  1073. yield db.update_async<TaskEntity>((!)task);
  1074. yield pulse.bump(task_id, "task suspended");
  1075. }
  1076. /**
  1077. * Marks a task done: records `summary` and `working_branch`, stamps
  1078. * `completed_at`, closes any open session (outcome `outcome`) and
  1079. * frees the saddle that was running it. Terminal transitions
  1080. * project ONCE (plan 13 B2): a task already done or failed keeps
  1081. * its first outcome — a redelivered push or a second terminal
  1082. * entry with a fresh id re-runs no bookkeeping, no comment, no
  1083. * event (retry_task is the one path that un-terminals).
  1084. */
  1085. public async void complete_task(int64 task_id, string? summary,
  1086. string? working_branch, string outcome) throws Error {
  1087. var task = yield db.query<TaskEntity>()
  1088. .where(expr("id == $0", elem<int64?>(task_id)))
  1089. .first_async();
  1090. if (task == null) {
  1091. throw new Error(Quark.from_string("clanker-saddle"), 8,
  1092. @"No task exists with id $(task_id)");
  1093. }
  1094. if (((!)task).status == STATUS_DONE || ((!)task).status == STATUS_FAILED) {
  1095. return;
  1096. }
  1097. yield mark_terminal((!)task, STATUS_DONE, summary, working_branch);
  1098. yield release_task(task_id, outcome, false);
  1099. yield pulse.bump(task_id, "task completed");
  1100. // Web Push (plan 11 task E): completion notifies the users
  1101. // holding a "Notify me" pref for this task.
  1102. push.notify_opted_in(task_id, PushService.lifecycle_notification(
  1103. ((!)task).title, task_id, "completed", summary));
  1104. // Plan 12: the goal's narrative hears the terminal event
  1105. // (a parked goal agent resumes on it), and a done
  1106. // predecessor auto-enqueues its dependents.
  1107. yield enqueue_dependents_of(((!)task));
  1108. }
  1109. /**
  1110. * Cancels a task: status 'failed' with a cancellation summary. Any
  1111. * open session is closed with outcome 'stopped' and its saddle
  1112. * freed. Draft tasks are refused — they are deleted, not
  1113. * cancelled (nothing ever ran). Idempotent on terminal tasks
  1114. * (plan 13 B2): the first outcome stands.
  1115. */
  1116. public async void cancel_task(int64 task_id) throws Error {
  1117. var task = yield db.query<TaskEntity>()
  1118. .where(expr("id == $0", elem<int64?>(task_id)))
  1119. .first_async();
  1120. if (task == null) {
  1121. throw new Error(Quark.from_string("clanker-saddle"), 9,
  1122. @"No task exists with id $(task_id)");
  1123. }
  1124. if (((!)task).status == STATUS_DONE || ((!)task).status == STATUS_FAILED) {
  1125. return;
  1126. }
  1127. if (((!)task).status == STATUS_DRAFT) {
  1128. throw new Error(Quark.from_string("clanker-saddle"), 55,
  1129. "Draft tasks are deleted, not cancelled");
  1130. }
  1131. yield mark_terminal((!)task, STATUS_FAILED, null, null, "Cancelled by user");
  1132. yield release_task(task_id, OUTCOME_STOPPED, false);
  1133. yield pulse.bump(task_id, "task cancelled");
  1134. }
  1135. /**
  1136. * Records a saddle-reported failure: status 'failed' with the
  1137. * daemon's message as summary, the open session closed with
  1138. * outcome 'failed' and its saddle freed. Plan 16 S4
  1139. * (terminal-text single representation): `tasks.summary` IS
  1140. * the terminal cache — no duplicate agent comment is written
  1141. * (the narrative renders `summary_lead`; the task page renders
  1142. * the summary card). Terminal transitions project ONCE (plan
  1143. * 13 B2): the live failure's honest burst — a dozen-plus
  1144. * distinct `saddle failure` entry ids in one session — still
  1145. * fails the task exactly once, with one summary write and one
  1146. * goal event.
  1147. */
  1148. public async void fail_task(int64 task_id, string? summary) throws Error {
  1149. var task = yield db.query<TaskEntity>()
  1150. .where(expr("id == $0", elem<int64?>(task_id)))
  1151. .first_async();
  1152. if (task == null) {
  1153. throw new Error(Quark.from_string("clanker-saddle"), 25,
  1154. @"No task exists with id $(task_id)");
  1155. }
  1156. if (((!)task).status == STATUS_DONE || ((!)task).status == STATUS_FAILED) {
  1157. return;
  1158. }
  1159. yield mark_terminal((!)task, STATUS_FAILED, summary, null,
  1160. "Saddle reported a failure");
  1161. yield release_task(task_id, OUTCOME_FAILED, false);
  1162. yield pulse.bump(task_id, "task failed");
  1163. // Web Push (plan 11 task E): an agent-reported failure
  1164. // notifies the opted-in users (a human cancellation does
  1165. // not — the cancellator already knows).
  1166. push.notify_opted_in(task_id, PushService.lifecycle_notification(
  1167. ((!)task).title, task_id, "failed", summary));
  1168. }
  1169. /** Withdraws the suspension stamp after a successful resume. */
  1170. public async void clear_suspension(int64 task_id) throws Error {
  1171. var task = yield db.query<TaskEntity>()
  1172. .where(expr("id == $0", elem<int64?>(task_id)))
  1173. .first_async();
  1174. if (task == null || ((!)task).suspended_at == null) {
  1175. return;
  1176. }
  1177. ((!)task).suspended_at = null;
  1178. yield db.update_async<TaskEntity>((!)task);
  1179. }
  1180. /**
  1181. * Whether any OTHER task sharing this task's working context
  1182. * moved while it was parked — a sibling under the same parent,
  1183. * or any task of the same project on the same shared working
  1184. * branch. "Moved" means any recorded transition after the park
  1185. * timestamp: a completion/failure (`completed_at`), a dispatch
  1186. * (session start), or a progress entry. The park marker is the
  1187. * suspension stamp, else the newest block's creation (the
  1188. * blocked-at marker), else the session's last recorded
  1189. * activity. False with no siblings or nothing moved — the wire's
  1190. * `branch_advanced`, so the daemon can skip the pointless
  1191. * `git pull --rebase` when the shared branch did not advance.
  1192. */
  1193. public async bool branch_advanced_since(int64 task_id) throws Error {
  1194. var me = yield db.query<TaskEntity>()
  1195. .where(expr("id == $0", elem<int64?>(task_id)))
  1196. .first_async();
  1197. if (me == null) {
  1198. return false;
  1199. }
  1200. DateTime? park = ((!)me).suspended_at;
  1201. if (park == null) {
  1202. foreach (var block in yield db.query<TaskBlock>()
  1203. .where(expr("task_id == $0", elem<int64?>(task_id)))
  1204. .order_by(expr("id"))
  1205. .materialise_async()) {
  1206. if (block.created_at != null) {
  1207. park = ((!)block).created_at;
  1208. }
  1209. }
  1210. }
  1211. if (park == null) {
  1212. foreach (var session in yield db.query<Session>()
  1213. .where(expr("task_id == $0", elem<int64?>(task_id)))
  1214. .materialise_async()) {
  1215. foreach (var entry in yield db.query<SessionCommand>()
  1216. .where(expr("session_id == $0", elem<int64?>(session.id)))
  1217. .materialise_async()) {
  1218. if (entry.started_at != null
  1219. && (park == null
  1220. || ((!)entry.started_at).compare((!)park) > 0)) {
  1221. park = (!)entry.started_at;
  1222. }
  1223. }
  1224. }
  1225. }
  1226. if (park == null) {
  1227. // Never parked (or nothing recorded): nothing could have
  1228. // advanced past a marker that does not exist.
  1229. return false;
  1230. }
  1231. foreach (var sibling in yield sibling_tasks((!)me)) {
  1232. if (sibling.completed_at != null
  1233. && ((!)sibling.completed_at).compare((!)park) > 0) {
  1234. return true;
  1235. }
  1236. foreach (var session in yield db.query<Session>()
  1237. .where(expr("task_id == $0", elem<int64?>(sibling.id)))
  1238. .materialise_async()) {
  1239. if (session.started_at != null
  1240. && session.started_at.compare((!)park) > 0) {
  1241. return true;
  1242. }
  1243. }
  1244. foreach (var progress in yield db.query<TaskProgress>()
  1245. .where(expr("task_id == $0", elem<int64?>(sibling.id)))
  1246. .materialise_async()) {
  1247. if (progress.created_at != null
  1248. && progress.created_at.compare((!)park) > 0) {
  1249. return true;
  1250. }
  1251. }
  1252. }
  1253. return false;
  1254. }
  1255. /**
  1256. * The tasks sharing `task`'s working context (excluding it):
  1257. * same-project tasks on the same non-empty working branch, and
  1258. * — for a goal agent — every other task of its goal (their
  1259. * completions merge into the goal branch, so the goal agent's
  1260. * post-park pull must hear them).
  1261. */
  1262. private async ImmutableLot<TaskEntity> sibling_tasks(TaskEntity task) throws Error {
  1263. var rows = new Vector<TaskEntity>();
  1264. var seen = new Invercargill.DataStructures.HashSet<string>();
  1265. if (task.goal_id > 0) {
  1266. foreach (var sibling in yield db.query<TaskEntity>()
  1267. .where(expr("goal_id == $0", elem<int64?>(task.goal_id)))
  1268. .materialise_async()) {
  1269. if (sibling.id != task.id && seen.add(sibling.id.to_string())) {
  1270. rows.add(sibling);
  1271. }
  1272. }
  1273. }
  1274. if (task.task_branch != null && ((!)task.task_branch).length > 0) {
  1275. foreach (var sibling in yield db.query<TaskEntity>()
  1276. .where(expr("project_id == $0 && task_branch == $1",
  1277. elem<int64?>(task.project_id), elem(((!)task).task_branch)))
  1278. .materialise_async()) {
  1279. if (sibling.id != task.id && seen.add(sibling.id.to_string())) {
  1280. rows.add(sibling);
  1281. }
  1282. }
  1283. }
  1284. return rows.to_immutable_buffer();
  1285. }
  1286. /**
  1287. * Returns a non-terminal task to the queue: closes its open
  1288. * sessions (outcome 'stopped') and frees the saddle that was
  1289. * running it — {@link release_task} with the requeue, the
  1290. * single owner of the bookkeeping. Plan 14 B1: a parked
  1291. * (suspended) task requeues too — the goal page's
  1292. * requeue-to-resume reaches a goal agent parked on a dead
  1293. * saddle; the suspension stamp clears so the next dispatch
  1294. * starts a fresh parking window.
  1295. */
  1296. public async void requeue_task(int64 task_id) throws Error {
  1297. yield release_task(task_id, OUTCOME_STOPPED, true);
  1298. yield pulse.bump(task_id, "task requeued");
  1299. }
  1300. /**
  1301. * Returns a failed task to the queue for another attempt.
  1302. * Plan 16 S4: the failure summary is the terminal cache of the
  1303. * attempt that earned it — retry clears it and copies it
  1304. * NOWHERE (the narrative rendered its lead, the page rendered
  1305. * its card; the next attempt writes a fresh one), so no
  1306. * summary→comment duplicate joins the thread. `completed_at`
  1307. * is withdrawn, any open session closes and its saddle frees —
  1308. * the same bookkeeping as {@link requeue_task} (retry's flip
  1309. * is unconditional: the attempt resets whatever it held),
  1310. * plus the audit comment "Task retried by user".
  1311. */
  1312. public async void retry_task(int64 task_id) throws Error {
  1313. var task = yield db.query<TaskEntity>()
  1314. .where(expr("id == $0", elem<int64?>(task_id)))
  1315. .first_async();
  1316. if (task == null) {
  1317. throw new Error(Quark.from_string("clanker-saddle"), 8,
  1318. @"No task exists with id $(task_id)");
  1319. }
  1320. ((!)task).status = STATUS_QUEUED;
  1321. ((!)task).summary = null;
  1322. ((!)task).completed_at = null;
  1323. yield db.update_async<TaskEntity>((!)task);
  1324. yield release_task(task_id, OUTCOME_STOPPED, false);
  1325. yield add_system_comment(task_id, "Task retried by user");
  1326. yield pulse.bump(task_id, "task retried");
  1327. }
  1328. /**
  1329. * Queued tasks whose auto-enqueue predecessor (if any) is
  1330. * done, ordered by priority (1 first) then creation time —
  1331. * the dispatcher's candidate list.
  1332. */
  1333. public async ImmutableLot<TaskEntity> get_ready_tasks() throws Error {
  1334. var queued = yield db.query<TaskEntity>()
  1335. .where(expr("status == $0", elem(STATUS_QUEUED)))
  1336. .materialise_async();
  1337. var all_tasks = yield db.query<TaskEntity>().materialise_async();
  1338. var rows = new Vector<TaskEntity>();
  1339. foreach (var task in queued) {
  1340. bool ready = true;
  1341. if (task.enqueue_after_task_id > 0) {
  1342. // The auto-enqueue predecessor also gates dispatch:
  1343. // a dependent that reached the queue some other way
  1344. // (a predecessor set after a manual enqueue) still
  1345. // waits for it to finish.
  1346. ready = false;
  1347. foreach (var other in all_tasks) {
  1348. if (other.id == task.enqueue_after_task_id) {
  1349. ready = other.status == STATUS_DONE;
  1350. break;
  1351. }
  1352. }
  1353. }
  1354. if (ready) {
  1355. rows.add(task);
  1356. }
  1357. }
  1358. var sorted = rows.sort((a, b) => {
  1359. if (a.priority != b.priority) {
  1360. return (int)(a.priority - b.priority);
  1361. }
  1362. int64 a_created = a.created_at != null ? ((!)a.created_at).to_unix() : 0;
  1363. int64 b_created = b.created_at != null ? ((!)b.created_at).to_unix() : 0;
  1364. return (int)(a_created - b_created);
  1365. });
  1366. return sorted.to_immutable_buffer();
  1367. }
  1368. /** Comments on a task, oldest first. */
  1369. public async ImmutableLot<TaskComment> list_comments(int64 task_id) throws Error {
  1370. return yield db.query<TaskComment>()
  1371. .where(expr("task_id == $0", elem<int64?>(task_id)))
  1372. .order_by(expr("id"))
  1373. .materialise_async();
  1374. }
  1375. /** Blocks on a task, oldest first. */
  1376. public async ImmutableLot<TaskBlock> list_blocks(int64 task_id) throws Error {
  1377. return yield db.query<TaskBlock>()
  1378. .where(expr("task_id == $0", elem<int64?>(task_id)))
  1379. .order_by(expr("id"))
  1380. .materialise_async();
  1381. }
  1382. /** The task's newest block, or null when it has none. */
  1383. public async TaskBlock? newest_block(int64 task_id) throws Error {
  1384. TaskBlock? newest = null;
  1385. foreach (var block in yield list_blocks(task_id)) {
  1386. newest = block;
  1387. }
  1388. return newest;
  1389. }
  1390. /**
  1391. * True when the task's newest block exists and carries an
  1392. * answer — a verdict is waiting to reach the parked workspace
  1393. * (the answer-delivery pass, the affinity resume, or a later
  1394. * assignment payload). Such a task must never be requeued: the
  1395. * requeue closes the session the resume machinery needs.
  1396. * Plan 15 C1: goal blocks live on the run task as single rows
  1397. * — no mirror pair exists to fall out of sync, so the newest
  1398. * row's own answer is the whole truth.
  1399. */
  1400. public async bool newest_block_answered(int64 task_id) throws Error {
  1401. var newest = yield newest_block(task_id);
  1402. return newest != null && ((!)newest).answer != null;
  1403. }
  1404. /** All commands recorded against a session, oldest first. */
  1405. public async ImmutableLot<SessionCommand> list_commands(int64 session_id) throws Error {
  1406. return yield db.query<SessionCommand>()
  1407. .where(expr("session_id == $0", elem<int64?>(session_id)))
  1408. .order_by(expr("id"))
  1409. .materialise_async();
  1410. }
  1411. /**
  1412. * True when the session carries raw PTY log rows — the terminal
  1413. * view then renders the concatenated raw log instead of the
  1414. * typed entry stream.
  1415. */
  1416. public async bool has_raw_log(int64 session_id) throws Error {
  1417. var probe = yield db.query<SessionRawLog>()
  1418. .where(expr("session_id == $0", elem<int64?>(session_id)))
  1419. .first_async();
  1420. return probe != null;
  1421. }
  1422. /**
  1423. * The session still open for `task_id` (no `ended_at`), or null.
  1424. * The null check runs in Vala: the expression layer cannot compare
  1425. * columns against null.
  1426. */
  1427. public async Session? open_session_for_task(int64 task_id) throws Error {
  1428. var sessions = yield db.query<Session>()
  1429. .where(expr("task_id == $0", elem<int64?>(task_id)))
  1430. .order_by(expr("id"))
  1431. .materialise_async();
  1432. foreach (var session in sessions) {
  1433. if (session.ended_at == null) {
  1434. return session;
  1435. }
  1436. }
  1437. return null;
  1438. }
  1439. /**
  1440. * The ONE release path (plan 16 R2, arch item 13): every path
  1441. * that takes a task off its saddle runs through here. Closes
  1442. * the task's open sessions with `outcome` and frees whatever
  1443. * saddle row still runs it — one conditional UPDATE, atomic
  1444. * against a concurrent ask allocation claiming a new task for
  1445. * the saddle between a read and a full-row write. `requeue`
  1446. * first returns a non-terminal task to the queue
  1447. * (assigned/working/blocked/review_requested/suspended — the
  1448. * suspension stamp withdraws so the next dispatch opens a
  1449. * fresh parking window).
  1450. *
  1451. * The terminal callers (complete/cancel/fail) pass their own
  1452. * outcome with the status already stamped; requeue_task and
  1453. * the unregister path ride the requeue; retry_task does its
  1454. * own unconditional flip (summary and completed_at die with
  1455. * the attempt) and releases without one. The offline variant
  1456. * ({@link SaddleService.mark_saddle_unreachable}) keeps its
  1457. * own guarded OFFLINE write for the saddle row and calls this
  1458. * after it — the idle-free below no-ops against that row (its
  1459. * current_task_id already cleared under the entry-seen guard)
  1460. * while the session close applies; its keep-parked gate (an
  1461. * answered block awaiting delivery) skips the release
  1462. * entirely.
  1463. */
  1464. public async void release_task(int64 task_id, string outcome, bool requeue) throws Error {
  1465. if (requeue) {
  1466. var task = yield db.query<TaskEntity>()
  1467. .where(expr("id == $0", elem<int64?>(task_id)))
  1468. .first_async();
  1469. if (task != null && (((!)task).status == STATUS_ASSIGNED
  1470. || ((!)task).status == STATUS_WORKING || ((!)task).status == STATUS_BLOCKED
  1471. || ((!)task).status == STATUS_REVIEW_REQUESTED
  1472. || ((!)task).status == STATUS_SUSPENDED)) {
  1473. ((!)task).status = STATUS_QUEUED;
  1474. if (((!)task).suspended_at != null) {
  1475. ((!)task).suspended_at = null;
  1476. }
  1477. yield db.update_async<TaskEntity>((!)task);
  1478. }
  1479. }
  1480. var sessions = yield db.query<Session>()
  1481. .where(expr("task_id == $0", elem<int64?>(task_id)))
  1482. .materialise_async();
  1483. foreach (var session in sessions) {
  1484. if (session.ended_at != null) {
  1485. continue;
  1486. }
  1487. session.ended_at = new DateTime.now_local();
  1488. session.outcome = outcome;
  1489. yield db.update_async<Session>(session);
  1490. }
  1491. var command = connection.create_command(
  1492. "UPDATE saddles SET status = '" + SADDLE_IDLE
  1493. + "', current_task_id = 0, allocated_at = NULL"
  1494. + " WHERE current_task_id = :task");
  1495. command.with_parameter<Invercargill.Element>("task", elem<int64?>(task_id));
  1496. yield command.execute_non_query_async();
  1497. }
  1498. }
  1499. }