lifecycle.py 40 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788
  1. """The lifecycle bundle (moved verbatim from the pre-split web_stack_test.py;
  2. see test/harness.py for the shared machinery and the runner)."""
  3. from harness import * # noqa: F401,F403 — the shared wire helpers
  4. # =============================================================================
  5. # Bundle: lifecycle — offline requeues, the hung-request tick survival,
  6. # parked-refresh bookkeeping, cancellation, the health panel, raw-log
  7. # losslessness, git identity and the live blocked-push roundtrip.
  8. # =============================================================================
  9. def run(ctx):
  10. auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token
  11. app = ctx.app
  12. print("== plan 11: draft state machine, edit matrix, enqueue/delete guards ==")
  13. # New tasks default to draft; Enqueue is the only path to the
  14. # queue (there is no agent path — the ingest API refuses to queue).
  15. act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
  16. "project_id": "1",
  17. "title": "Draft lifecycle task",
  18. "priority": "4",
  19. })
  20. draft_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
  21. if t["title"] == "Draft lifecycle task"][0])
  22. detail = task_state(auth, draft_task)
  23. check(detail["status"] == "draft", "a created task lands in draft")
  24. check(detail.get("enqueue") is not None and detail.get("delete") is not None,
  25. "draft exposes the enqueue + delete actions")
  26. check(detail.get("cancel") is not None, "the cancel action stays authored")
  27. status, page = http("GET", "/tasks/%d" % draft_task, timeout=15)
  28. check(status == 200 and "Enqueue task" in page and ">Delete<" in page,
  29. "the task page shows Enqueue + Delete in draft")
  30. check("Mark done" not in page,
  31. "the user Mark-complete action is gone (agent-only completion)")
  32. # Plan 12: the auto-enqueue-after predecessor REPLACES draft
  33. # dependencies — setting it is allowed while the predecessor is
  34. # still pending (the enqueue itself succeeds; the DISPATCHER holds
  35. # dispatch back until the predecessor is done).
  36. act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
  37. "project_id": "1",
  38. "title": "Draft dependency",
  39. "priority": "4",
  40. })
  41. draft_dep = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
  42. if t["title"] == "Draft dependency"][0])
  43. act(auth, detail["update"], {"enqueue_after_task_id": str(draft_dep)})
  44. check(int(task_state(auth, draft_task).get("enqueue_after_task_id") or 0)
  45. == draft_dep,
  46. "the draft edit sets the auto-enqueue predecessor")
  47. act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
  48. "project_id": "1", "title": "Eager dependent", "priority": "4",
  49. "enqueue_after_task_id": str(draft_dep)})
  50. eager = task_id_of("Eager dependent")
  51. enqueue_task(auth, eager)
  52. check(task_state(auth, eager)["status"] == "queued",
  53. "enqueue succeeds while the predecessor is pending (no draft guard)")
  54. check(quiet_window(quiet_seconds(),
  55. lambda: task_state(auth, eager)["status"] == "queued",
  56. "pending predecessor holds dispatch"),
  57. "the pending predecessor holds dispatch back (never assigned)")
  58. # The edit matrix starts in DRAFT: every field except status.
  59. act(auth, detail["update"], {
  60. "title": "Draft lifecycle task (edited)",
  61. "description": "Now with a description",
  62. "priority": "2",
  63. "task_branch": "feature/draft",
  64. "branch_from": "main",
  65. "mode": "investigate",
  66. })
  67. detail = task_state(auth, draft_task)
  68. check(detail["title"] == "Draft lifecycle task (edited)"
  69. and detail["description"] == "Now with a description"
  70. and detail["priority"] == 2
  71. and detail["task_branch"] == "feature/draft"
  72. and detail["branch_from"] == "main"
  73. and detail["mode"] == "investigate",
  74. "draft edits apply to every field (incl. the mode)")
  75. # Clear the predecessor (the empty sentinel): dispatch gates on
  76. # the hold tag alone now.
  77. act(auth, detail["update"], {"enqueue_after_task_id": "0"})
  78. check(not int(task_state(auth, draft_task).get("enqueue_after_task_id") or 0),
  79. "the draft edit clears the auto-enqueue predecessor")
  80. # A hold tag nothing carries keeps the dependent QUEUED (never
  81. # dispatched) through the remaining edit-matrix checks.
  82. saddles_state = slot_public(auth, "/saddles", "saddles")
  83. act(auth, saddles_state["create_capability"], {"name": "holdlock"})
  84. holdlock_cap = int([c["id"] for c in slot_public(auth, "/saddles", "saddles")["capabilities"]
  85. if c["name"] == "holdlock"][0])
  86. act(auth, detail["attach_capability"], {"capability_id": str(holdlock_cap)})
  87. # Enqueue the now-cleared dependent: the hold tag keeps it queued
  88. # (never dispatched), so the non-draft edit matrix applies.
  89. enqueue_task(auth, draft_task)
  90. check(quiet_window(3, lambda: task_state(auth, draft_task)["status"] == "queued",
  91. "held dependent stays queued"),
  92. "the held dependent stays queued (never dispatched)")
  93. # The predecessor settles for the shard drain.
  94. enqueue_task(auth, draft_dep)
  95. check(wait_task_done(auth, draft_dep, timeout=90) is not None,
  96. "the predecessor settles")
  97. # Drafts are never dispatched even with an idle saddle available.
  98. act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
  99. "project_id": "1",
  100. "title": "Never-dispatched draft",
  101. "priority": "1",
  102. })
  103. silent_draft = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
  104. if t["title"] == "Never-dispatched draft"][0])
  105. check(quiet_window(quiet_seconds(),
  106. lambda: task_state(auth, silent_draft)["status"] == "draft",
  107. "draft never dispatched"),
  108. "an idle saddle never picks a draft up (no agent enqueue path)")
  109. # Once work can begin (queued here), only priority is editable.
  110. act(auth, detail["update"], {
  111. "title": "Should not apply",
  112. "priority": "3",
  113. })
  114. detail = task_state(auth, draft_task)
  115. check(detail["title"] == "Draft lifecycle task (edited)" and detail["priority"] == 3,
  116. "non-draft edits apply priority only (title locked)")
  117. # A terminal task edits nothing.
  118. conn = db()
  119. conn.execute("update tasks set status = 'done', completed_at = strftime('%s','now')"
  120. " where id = ?", (draft_task,))
  121. conn.commit()
  122. conn.close()
  123. act(auth, task_state(auth, draft_task)["update"], {"priority": "8"})
  124. check(task_state(auth, draft_task)["priority"] == 3,
  125. "terminal tasks edit nothing (priority locked)")
  126. # Capabilities lock once a worker has the job: editable while
  127. # queued, locked once working.
  128. act(auth, saddles_state["create_capability"], {"name": "editlock"})
  129. editlock_cap = int([c["id"] for c in slot_public(auth, "/saddles", "saddles")
  130. ["capabilities"] if c["name"] == "editlock"][0])
  131. act(auth, saddle_card(auth, "fake-saddle")["attach_capability"],
  132. {"capability_id": str(editlock_cap)})
  133. locked_task = create_queued(auth, {"project_id": "1", "title": "Caplock task",
  134. "priority": "2"})
  135. detail = task_state(auth, locked_task)
  136. act(auth, detail["attach_capability"], {"capability_id": str(editlock_cap)})
  137. check(wait_for(lambda: any(c["name"] == "editlock"
  138. for c in task_state(auth,
  139. locked_task)["capabilities"])
  140. or False, timeout=60, what="editlock attached") is not None,
  141. "capabilities attach while queued (no worker yet)")
  142. check(wait_for(lambda: task_state(auth, locked_task).get("status") == "working"
  143. or False, timeout=60, what="caplock working") is not None,
  144. "the caplock task reaches working")
  145. detach_directives = act(auth, task_state(auth, locked_task)["capabilities"][0]["detach"], {})
  146. check(any(d.get("type") == "notify" and d.get("kind") == "error"
  147. for d in detach_directives),
  148. "capabilities are locked once a worker has the job")
  149. # The ingest API has no queue transition either (only working and
  150. # blocked are accepted states).
  151. conn = db()
  152. caplock_session = conn.execute(
  153. "select id from sessions where task_id = ? and ended_at is null",
  154. (locked_task,)).fetchone()
  155. conn.close()
  156. if caplock_session is not None:
  157. status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
  158. {"session_id": caplock_session[0],
  159. "state": "queued"})
  160. check(status == 200
  161. and task_state(auth, locked_task)["status"] in ("working", "done"),
  162. "a queued state push is a no-op (only the wire's states drive bookkeeping)")
  163. # Delete guards: a clean draft deletes; non-draft refuses; a draft
  164. # with dependents is refused naming them.
  165. act(auth, task_state(auth, silent_draft)["delete"], {})
  166. check(not any(t["id"] == silent_draft
  167. for t in slot_public(auth, "/tasks", "tasks")["tasks"]),
  168. "a clean draft task deletes")
  169. delete_directives = act(auth, task_state(auth, draft_task)["delete"], {})
  170. check(any(d.get("type") == "notify" and d.get("kind") == "error"
  171. for d in delete_directives),
  172. "a non-draft task refuses the delete")
  173. act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
  174. "project_id": "1", "title": "Guarded draft", "priority": "4"})
  175. guarded = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
  176. if t["title"] == "Guarded draft"][0])
  177. act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
  178. "project_id": "1", "title": "Dependent namer", "priority": "4"})
  179. namer = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
  180. if t["title"] == "Dependent namer"][0])
  181. act(auth, task_state(auth, namer)["update"],
  182. {"enqueue_after_task_id": str(guarded)})
  183. check(int(task_state(auth, namer).get("enqueue_after_task_id") or 0) == guarded,
  184. "the namer points at the guarded predecessor")
  185. act(auth, task_state(auth, guarded)["delete"], {})
  186. check(not any(t["id"] == guarded
  187. for t in slot_public(auth, "/tasks", "tasks")["tasks"]),
  188. "deleting a predecessor succeeds")
  189. check(not int(task_state(auth, namer).get("enqueue_after_task_id") or 0),
  190. "the delete clears the dependent's dangling predecessor reference")
  191. print("== offline requeues ==")
  192. # Tasks caught working on a saddle that went offline requeue with a
  193. # agent comment each tick.
  194. task_id_tmp = create_queued(auth, {
  195. "project_id": "1",
  196. "title": "Offline saddle task",
  197. "priority": "2",
  198. })
  199. offline_task = task_id_tmp
  200. check(wait_for(lambda: task_state(auth, offline_task).get("status") == "working"
  201. or False, timeout=60, what="offline task starts working") is not None,
  202. "task reaches working before its saddle disappears")
  203. # Hold the daemon mid-round across the requeue window (the
  204. # original relied on its 8s work period outlasting the 5s tick):
  205. # the task completing on its own would race the offline pass and
  206. # skip the requeue the check exists to observe.
  207. fake.hold_state = True
  208. conn = db()
  209. conn.execute("update saddles set status = 'offline' where name = 'fake-saddle'")
  210. conn.commit()
  211. conn.close()
  212. # The requeue frees the saddle in the same tick, so 'queued' is
  213. # transient; the durable evidence is the agent comment (and the task
  214. # completing again on the healthy daemon).
  215. check(wait_for(lambda: any(c["body"] == "saddle went offline; task requeued"
  216. for c in task_state(auth, offline_task).get("comments", []))
  217. or False, timeout=30, what="offline requeue comment") is not None,
  218. "offline saddle requeues the task")
  219. fake.hold_state = False # release: the round ends, the daemon re-asks
  220. check(wait_task_done(auth, offline_task, timeout=60) is not None,
  221. "requeued task completes again on the healthy daemon")
  222. # R2 (arch 13): the release owner's outcomes — the requeue closed
  223. # the held attempt's session 'stopped'; the held daemon's terminal
  224. # push then completes the task against that closed session (the
  225. # outcome stands — a closed row never reopens). A task that
  226. # completes a LIVE round closes its session 'done' (draft_dep
  227. # above ran one open session to its terminal push).
  228. conn = db()
  229. outcome_rows = conn.execute(
  230. "select outcome from sessions where task_id = ? order by id",
  231. (offline_task,)).fetchall()
  232. dep_outcome = conn.execute(
  233. "select outcome from sessions where task_id = ? order by id desc",
  234. (draft_dep,)).fetchone()
  235. conn.close()
  236. check(outcome_rows and outcome_rows[0][0] == "stopped",
  237. "the one release path closes the requeued attempt 'stopped'")
  238. check(dep_outcome is not None and dep_outcome[0] == "done",
  239. "the terminal push closes a live round's session 'done'")
  240. print("== R6 F1: a hanging request cannot freeze the tick ==")
  241. # One wedged /task-update (never responds) blocks the comment/
  242. # subtask-event delivery pass; the Soup timeout must fail it and
  243. # later ticks keep dispatching.
  244. conn = db()
  245. conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
  246. " values (1, 'Hang survivor', 'queued', 2, strftime('%s','now'))")
  247. hang_task = conn.execute("select max(id) from tasks").fetchone()[0]
  248. conn.close()
  249. task_id_tmp = create_queued(auth, {
  250. "project_id": "1",
  251. "title": "Hang runner",
  252. "priority": "3",
  253. })
  254. hang_runner = task_id_tmp
  255. # A human comment on the running task gives the tick a
  256. # /task-update delivery to make; the next one hangs.
  257. check(wait_for(lambda: task_state(auth, hang_runner).get("status") == "working"
  258. or False, timeout=60, what="hang runner working") is not None,
  259. "hang runner reaches working")
  260. act(auth, task_state(auth, hang_runner)["comment"],
  261. {"body": "operator ping while wedged"})
  262. fake.hang_next["/task-update"] = 1
  263. check(wait_task_done(auth, hang_runner, timeout=90) is not None,
  264. "dispatching task survives a hung daemon request")
  265. check(wait_for(lambda: any(c.get("text") == "operator ping while wedged"
  266. or c.get("body") == "operator ping while wedged"
  267. for c in fake.comment_log) or False,
  268. timeout=30, what="wedged comment eventually delivered") is not None,
  269. "the comment delivery retries after the hang")
  270. # The hung delivery itself fails after the timeout; the system
  271. # kept moving either way.
  272. with open(os.path.join(WORK, "app.log")) as handle:
  273. app_log = handle.read()
  274. check("tick exceeded" not in app_log or "Dispatch tick failed" in app_log,
  275. "a hung request never freezes the tick latch")
  276. print("== R6 F2: refresh must not corrupt parked bookkeeping ==")
  277. fake.suspend_on_title = "Parked refresh parent"
  278. task_id_tmp = create_queued(auth, {
  279. "project_id": "1",
  280. "title": "Parked refresh parent",
  281. "priority": "2",
  282. })
  283. parked = task_id_tmp
  284. check(wait_for(lambda: task_state(auth, parked).get("status") == "suspended"
  285. or False, timeout=60, what="parked parent suspends") is not None,
  286. "parent parks before the refresh")
  287. fake_card = saddle_card(auth, "fake-saddle")
  288. act(auth, fake_card["refresh"], {})
  289. fake_card = saddle_card(auth, "fake-saddle")
  290. check(fake_card["status"] == "idle" and int(fake_card["current_task_id"] or 0) == 0,
  291. "refresh on a parked (suspended) daemon leaves the saddle idle")
  292. # The parked task keeps its open session and stays put: plain
  293. # tasks have no subtask events to resume them (the goal machinery
  294. # owns resumes); the human resumes it explicitly.
  295. check(quiet_window(quiet_seconds(),
  296. lambda: task_state(auth, parked)["status"] == "suspended",
  297. "parked task stays parked"),
  298. "a parked plain task stays parked after the refresh (no resume channel)")
  299. act(auth, task_state(auth, parked)["cancel"], {})
  300. check(wait_for(lambda: task_state(auth, parked)["status"] == "failed" or False,
  301. timeout=30, what="parked task cancelled") is not None,
  302. "the parked task cancels cleanly after the refresh")
  303. print("== R7 B1: cancel reaches the daemon ==")
  304. # Plan 10 phase 2: cancellation is POST /task-update {kind: cancel}
  305. # to the RUNNING daemon (the abort machinery, workspace kept).
  306. task_id_tmp = create_queued(auth, {
  307. "project_id": "1",
  308. "title": "Cancel live",
  309. "priority": "2",
  310. })
  311. cancel_live = task_id_tmp
  312. check(wait_for(lambda: task_state(auth, cancel_live).get("status") == "working"
  313. or False, timeout=60, what="cancel-live working") is not None,
  314. "task reaches working before the cancel")
  315. act(auth, task_state(auth, cancel_live)["cancel"], {})
  316. check(wait_for(lambda: cancel_live in fake.abort_log or False,
  317. timeout=30, what="daemon abort recorded") is not None,
  318. "cancelling POSTs task-update cancel to the daemon")
  319. check(any(c["body"] == "cancelled; daemon aborted"
  320. for c in task_state(auth, cancel_live).get("comments", [])),
  321. "the abort outcome is narrated")
  322. check(task_state(auth, cancel_live)["status"] == "failed",
  323. "cancelled task lands failed as before")
  324. # A 500 on the cancel marks the saddle offline (no fence layer
  325. # anymore); nothing dispatches to it while offline, and the
  326. # daemon's own ask flips the row back to idle (proof of life).
  327. task_id_tmp = create_queued(auth, {
  328. "project_id": "1",
  329. "title": "Fence runner",
  330. "priority": "2",
  331. })
  332. fence_runner = task_id_tmp
  333. check(wait_for(lambda: task_state(auth, fence_runner).get("status") == "working"
  334. or False, timeout=60, what="fence runner working") is not None,
  335. "second task reaches working")
  336. fake.abort_failures = 1
  337. act(auth, task_state(auth, fence_runner)["cancel"], {})
  338. check(any(c["body"] == "cancelled; daemon unreachable — saddle marked offline"
  339. for c in task_state(auth, fence_runner).get("comments", [])),
  340. "failed cancel narrates the offline marking")
  341. check(saddle_card(auth, "fake-saddle")["status"] == "offline",
  342. "saddle marked offline after the failed cancel")
  343. task_id_tmp = create_queued(auth, {
  344. "project_id": "1",
  345. "title": "During fence",
  346. "priority": "3",
  347. })
  348. during_fence = task_id_tmp
  349. # Converted sleep(7): a quiet window spanning several dispatch
  350. # ticks in which the task MUST stay queued — if the offline
  351. # saddle ever dispatches it, this fails loudly. The daemon is held
  352. # mid-round for the window (the original relied on its 8s work
  353. # period outlasting the sleep): a finishing round would ask, prove
  354. # liveness and flip the offline row — only after the window may it.
  355. fake.hold_state = True
  356. check(quiet_window(quiet_seconds(),
  357. lambda: task_state(auth, during_fence)["status"] == "queued",
  358. "offline saddle does not dispatch during_fence"),
  359. "offline saddle is not dispatchable")
  360. fake.hold_state = False
  361. # The (unaborted) daemon finishes its round, goes idle-awake and
  362. # asks — the ask is the proof of life that flips the row back.
  363. check(wait_for(lambda: saddle_card(auth, "fake-saddle")["status"] == "idle" or False,
  364. timeout=30, what="ask flips offline to idle") is not None,
  365. "the daemon's ask flips the offline row back to idle")
  366. check(wait_task_done(auth, during_fence, timeout=60) is not None,
  367. "queued work dispatches once the saddle proves it is alive")
  368. print("== R7 B3: health panel + pushed events from the first round ==")
  369. # A normally dispatched task pushes its event batch immediately:
  370. # ledger rows exist from the first round (the session-4 gap stays
  371. # fixed without any probe machinery).
  372. task_id_tmp = create_queued(auth, {
  373. "project_id": "1",
  374. "title": "Raw from first push",
  375. "priority": "2",
  376. })
  377. raw_first = task_id_tmp
  378. def event_count(task_id):
  379. conn = db()
  380. try:
  381. return conn.execute(
  382. "select count(*) from events where session_id in"
  383. " (select id from sessions where task_id = ?)", (task_id,)).fetchone()[0]
  384. finally:
  385. conn.close()
  386. check(wait_for(lambda: event_count(raw_first) > 0 or False,
  387. timeout=60, what="event rows from first push") is not None,
  388. "pushed events land from the first round (session-4 gap)")
  389. check(wait_task_done(auth, raw_first, timeout=60) is not None,
  390. "raw-first task completes")
  391. # Health panel: a manual refresh populates the snapshot.
  392. fake_card = saddle_card(auth, "fake-saddle")
  393. act(auth, fake_card["refresh"], {})
  394. card = saddle_card(auth, "fake-saddle")
  395. check(card.get("health") in ("healthy", "stale")
  396. and card.get("daemon_state") in ("working", "idle", "done")
  397. and card.get("uptime_display", "") != ""
  398. and card.get("last_contact_display", "") not in ("", "never"),
  399. "saddles page shows daemon state, uptime and last contact")
  400. print("== R8: pushed chunk ingestion is lossless ==")
  401. # The daemon's opening flush: a coloured prompt run, plain text,
  402. # a bare CRLF between prompts and the tail — every non-empty
  403. # chunk must land, exactly once, in seq order. A mid-flight
  404. # redelivery of the whole batch (the outbox's at-least-once
  405. # retry) consumes the duplicates.
  406. corpus = [
  407. {"seq": 1, "kind": "command_output",
  408. "payload": {"text": "\x1b[01;32muser@host\x1b[00m:\x1b[01;34m~/task-1\x1b[00m$ "}},
  409. {"seq": 2, "kind": "command_output", "payload": {"text": "echo hello\r\n"}},
  410. {"seq": 3, "kind": "command_output", "payload": {"text": "\r\n"}},
  411. {"seq": 4, "kind": "command_output", "payload": {"text": "hello\r\n"}},
  412. {"seq": 5, "kind": "command_output",
  413. "payload": {"text": "\x1b[01;32muser@host\x1b[00m:\x1b[01;34m~/task-1\x1b[00m$ "}},
  414. {"seq": 6, "kind": "command_output", "payload": {"text": "done\r\n"}},
  415. ]
  416. fake.scripted_corpus = corpus
  417. fake.hold_state = True
  418. task_id_tmp = create_queued(auth, {
  419. "project_id": "1",
  420. "title": "Lossless log task",
  421. "priority": "2",
  422. })
  423. lossless = task_id_tmp
  424. def stored_seqs(task_id):
  425. conn = db()
  426. try:
  427. return [r[0] for r in conn.execute(
  428. "select seq from events where session_id in"
  429. " (select id from sessions where task_id = ?)"
  430. " and seq is not null order by seq", (task_id,))]
  431. finally:
  432. conn.close()
  433. check(wait_for(lambda: task_state(auth, lossless).get("status") == "working"
  434. or False, timeout=60, what="lossless task working") is not None,
  435. "lossless task dispatches and pushes its opening flush")
  436. check(wait_for(lambda: stored_seqs(lossless) == [1, 2, 3, 4, 5, 6] or False,
  437. timeout=30, what="all six chunks stored") is not None,
  438. "every pushed event lands in the ledger, CRLF-only included (%s)"
  439. % stored_seqs(lossless))
  440. status, _ = saddle_api("POST", "/api/saddle/chunks", fake.token,
  441. {"session_id": fake.session_id,
  442. "events": [dict(chunk) for chunk in corpus]})
  443. check(status == 200 and stored_seqs(lossless) == [1, 2, 3, 4, 5, 6],
  444. "a full-batch redelivery stores nothing new (exactly-once)")
  445. request = urllib.request.Request(
  446. WEB + "/tasks/%d/events/stream" % lossless)
  447. with urllib.request.urlopen(request, timeout=15) as stream:
  448. texts = []
  449. deadline = time.monotonic() + 8
  450. while time.monotonic() < deadline and len(texts) < 6:
  451. line = stream.readline().decode().strip()
  452. if not line.startswith("data: ") or line == "data: ping":
  453. continue
  454. event = json.loads(line[6:])
  455. if event.get("kind") == "command_output":
  456. texts.append(event.get("payload", {}).get("text") or "")
  457. check("\r\n" in texts and len(texts) == 6,
  458. "SSE serves the whitespace payload unfiltered")
  459. fake.hold_state = False
  460. fake.scripted_corpus = None
  461. check(wait_task_done(auth, lossless, timeout=60) is not None,
  462. "lossless task completes")
  463. print("== R9 B1/B2: git identity + ssh key ==")
  464. # The key travels on /status into the saddle row and the page view
  465. # (the manual refresh's exceptional probe).
  466. fake.ssh_public_key = "ssh-ed25519 AAAAC3NzaC1lZDI1teStKey saddle-1@fedora"
  467. fake_card = saddle_card(auth, "fake-saddle")
  468. check(fake_card.get("ssh_public_key", "") == "",
  469. "no key shown before the daemon reports one")
  470. act(auth, fake_card["refresh"], {})
  471. check(wait_for(lambda: "AAAAC3NzaC1lZDI1teStKey" in
  472. saddle_card(auth, "fake-saddle").get("ssh_public_key", "") or False,
  473. timeout=30, what="key lands on the row") is not None,
  474. "reported ssh_public_key persists onto the saddle row")
  475. conn = db()
  476. row_key = conn.execute("select ssh_public_key from saddles where name = 'fake-saddle'"
  477. ).fetchone()[0]
  478. conn.close()
  479. check(row_key == fake.ssh_public_key,
  480. "the key is stored verbatim in the database")
  481. # Identity edit flows through the UI into the assignment payload.
  482. fake_card = saddle_card(auth, "fake-saddle")
  483. act(auth, fake_card["update_identity"],
  484. {"git_name": "clanker-bot", "git_email": "bot@unitatem.net"})
  485. fake_card = saddle_card(auth, "fake-saddle")
  486. check(fake_card.get("git_name") == "clanker-bot"
  487. and fake_card.get("git_email") == "bot@unitatem.net",
  488. "inline edit saves the git identity")
  489. task_id_tmp = create_queued(auth, {
  490. "project_id": "1",
  491. "title": "Identity task",
  492. "priority": "2",
  493. })
  494. identity_task = task_id_tmp
  495. check(wait_for(lambda: any(int(p.get("task", {}).get("id") or 0) == identity_task
  496. for p in fake.start_payloads) or False,
  497. timeout=60, what="identity task dispatched") is not None,
  498. "identity task dispatches")
  499. payload = [p for p in fake.start_payloads
  500. if int(p.get("task", {}).get("id") or 0) == identity_task][-1]
  501. identity = payload.get("git_identity") or {}
  502. check(identity.get("name") == "clanker-bot"
  503. and identity.get("email") == "bot@unitatem.net",
  504. "the assignment carries the saddle's git_identity")
  505. # Earlier assignments (identity unset) carry no git_identity member.
  506. check(all("git_identity" not in p for p in fake.start_payloads[:-1]),
  507. "unset identity omits the member (daemon fallback applies)")
  508. # Clearing the identity omits it from later assignments.
  509. fake_card = saddle_card(auth, "fake-saddle")
  510. act(auth, fake_card["update_identity"], {"git_name": "", "git_email": ""})
  511. check(saddle_card(auth, "fake-saddle").get("git_name", "") == "",
  512. "clearing the identity empties the row")
  513. check(wait_task_done(auth, identity_task, timeout=60) is not None,
  514. "identity task completes")
  515. print("== goal 8 task 5: per-saddle prompt note ==")
  516. # The admin edits the note on the /saddles card; it persists on the
  517. # row and rides EVERY later assignment as `prompt_note`.
  518. fake_card = saddle_card(auth, "fake-saddle")
  519. check(fake_card.get("prompt_note", "") == "",
  520. "no prompt note before the admin sets one")
  521. check(all(p.get("prompt_note") == "" for p in fake.start_payloads),
  522. "unset note carries an empty string (never a null)")
  523. act(auth, fake_card["update_prompt_note"],
  524. {"prompt_note": "You have root and may install development tooling via DNF."})
  525. fake_card = saddle_card(auth, "fake-saddle")
  526. check(fake_card.get("prompt_note") == "You have root and may install development tooling via DNF.",
  527. "inline edit saves the prompt note")
  528. note_task = create_queued(auth, {
  529. "project_id": "1",
  530. "title": "Prompt note task",
  531. "priority": "2",
  532. })
  533. check(wait_for(lambda: any(int(p.get("task", {}).get("id") or 0) == note_task
  534. for p in fake.start_payloads) or False,
  535. timeout=60, what="prompt-note task dispatched") is not None,
  536. "prompt-note task dispatches")
  537. note_payload = [p for p in fake.start_payloads
  538. if int(p.get("task", {}).get("id") or 0) == note_task][-1]
  539. check(note_payload.get("prompt_note") == "You have root and may install development tooling via DNF.",
  540. "the assignment carries the saddle's prompt_note verbatim")
  541. check(wait_task_done(auth, note_task, timeout=60) is not None,
  542. "prompt-note task completes")
  543. # Clearing the note empties the member on later assignments.
  544. fake_card = saddle_card(auth, "fake-saddle")
  545. act(auth, fake_card["update_prompt_note"], {"prompt_note": ""})
  546. check(saddle_card(auth, "fake-saddle").get("prompt_note", "") == "",
  547. "clearing the prompt note empties the row")
  548. # A fat note is clamped: the row keeps the full text but the wire
  549. # serves at most PROMPT_NOTE_MAX_CHARS.
  550. fat = "N" * 5000
  551. fake_card = saddle_card(auth, "fake-saddle")
  552. act(auth, fake_card["update_prompt_note"], {"prompt_note": fat})
  553. clamp_task = create_queued(auth, {
  554. "project_id": "1",
  555. "title": "Clamp note task",
  556. "priority": "2",
  557. })
  558. check(wait_for(lambda: any(int(p.get("task", {}).get("id") or 0) == clamp_task
  559. for p in fake.start_payloads) or False,
  560. timeout=60, what="clamp-note task dispatched") is not None,
  561. "clamp-note task dispatches")
  562. clamp_payload = [p for p in fake.start_payloads
  563. if int(p.get("task", {}).get("id") or 0) == clamp_task][-1]
  564. check(len(clamp_payload.get("prompt_note") or "") == 2000,
  565. "a fat prompt_note is clamped to 2000 chars on the wire")
  566. check(wait_task_done(auth, clamp_task, timeout=60) is not None,
  567. "clamp-note task completes")
  568. fake_card = saddle_card(auth, "fake-saddle")
  569. act(auth, fake_card["update_prompt_note"], {"prompt_note": ""})
  570. print("== phase 2: blocked push -> block row + pulse ==")
  571. # The browser flow on each pulse: statum.act(task.refresh) — the
  572. # sealed GET action must rehydrate the task slot with the new
  573. # block the daemon just PUSHED (state blocked + block payload).
  574. task_id_tmp = create_queued(auth, {
  575. "project_id": "1",
  576. "title": "Live block task",
  577. "priority": "2",
  578. })
  579. live_task = task_id_tmp
  580. check(wait_for(lambda: task_state(auth, live_task).get("status") == "working"
  581. or False, timeout=60, what="live task working") is not None,
  582. "live-block task reaches working")
  583. fake.hold_state = True
  584. fake.state = "blocked" # the daemon parks on its question
  585. status = fake.push_results("blocked", block={
  586. "question": "Pick one", "type": "question", "options": ["A", "B"]})
  587. check(status == 200, "blocked state push accepted")
  588. # The parked daemon is idle-awake: it asks (and sleeps on 204).
  589. fake.maybe_trigger_ask("parked")
  590. check(wait_for(lambda: any(b.get("question") == "Pick one"
  591. for b in task_state(auth, live_task).get("blocks", []))
  592. or False, timeout=30, what="block lands in state") is not None,
  593. "the pushed block reaches the sealed refresh without a reload")
  594. detail = task_state(auth, live_task)
  595. check(detail.get("status") == "blocked",
  596. "the blocked push flips the task to blocked")
  597. pushed_block = [b for b in detail["blocks"] if b.get("question") == "Pick one"][0]
  598. conn = db()
  599. options_json = conn.execute(
  600. "select options from task_blocks where id = ?", (int(pushed_block["id"]),)
  601. ).fetchone()[0]
  602. conn.close()
  603. check(json.loads(options_json) == ["A", "B"],
  604. "the block row carries its options payload")
  605. check(any(s["status"] == "idle"
  606. for s in slot_public(auth, "/saddles", "saddles")["saddles"]),
  607. "the blocked push parks the saddle")
  608. detail = task_state(auth, live_task)
  609. check(detail.get("refresh") is not None,
  610. "the sealed GET refresh action is still authored")
  611. status, _ = http("GET", detail["refresh"]["uri"], timeout=15)
  612. check(status in (200, 401),
  613. "the refresh URI resolves as a GET (client roundtrip shape)")
  614. act(auth, pushed_block["answer_action"], {"answer": "A"})
  615. # The answer makes the parked task resume-offerable: the affine
  616. # saddle is woken, asks, and its assignment carries the answer.
  617. pushes_before = len([p for p in fake.start_payloads
  618. if int(p.get("task", {}).get("id") or 0) == live_task])
  619. check(wait_for(lambda: len([p for p in fake.start_payloads
  620. if int(p.get("task", {}).get("id") or 0) == live_task])
  621. > pushes_before or False, timeout=60,
  622. what="blocked task resumes") is not None,
  623. "answering the pushed block resumes the parked workspace")
  624. fake.hold_state = False # the resumed round runs to completion
  625. resume_payload = [p for p in fake.start_payloads
  626. if int(p.get("task", {}).get("id") or 0) == live_task][-1]
  627. answers = resume_payload.get("task", {}).get("answers", [])
  628. check(any(a.get("answer") == "A" for a in answers),
  629. "the resume assignment carries the human's answer")
  630. check(wait_task_done(auth, live_task, timeout=60) is not None,
  631. "live-block task completes after answering")
  632. print("== R2: the one release path behind unregister ==")
  633. # Arch 13 + plan 16 item 23: removing a saddle releases the task it
  634. # is DERIVED to run (its open session) through the same owner as
  635. # every requeue — never an orphan open session on a deleted row, and
  636. # no saddle-row cache to key the release on. Route 1: a DEAD host —
  637. # the abort delivery fails fast, the offline bookkeeping releases
  638. # (queued + comment + session 'stopped'), the unregister then
  639. # deletes the row.
  640. conn = db()
  641. no_cache_columns = conn.execute(
  642. "select count(*) from pragma_table_info('saddles')"
  643. " where name in ('current_task_id', 'allocated_at')").fetchone()[0]
  644. conn.close()
  645. check(no_cache_columns == 0,
  646. "the saddle row carries no work cache (item 23: sessions are the truth)")
  647. saddles_state = slot_public(auth, "/saddles", "saddles")
  648. act(auth, saddles_state["register"], {"name": "dead-host",
  649. "host": "127.0.0.1:1"})
  650. dead_card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
  651. if s["name"] == "dead-host"][0]
  652. dead_id = int(dead_card["id"])
  653. task_id_tmp = create_queued(auth, {
  654. "project_id": "1",
  655. "title": "Dead-host unregister task",
  656. "priority": "2",
  657. })
  658. dead_task = task_id_tmp
  659. conn = db()
  660. conn.execute("update tasks set status = 'working' where id = ?",
  661. (dead_task,))
  662. conn.execute("insert into sessions (task_id, saddle_id, started_at)"
  663. " values (?, ?, strftime('%s','now'))", (dead_task, dead_id))
  664. conn.commit()
  665. dead_session = conn.execute(
  666. "select id from sessions where task_id = ? and ended_at is null",
  667. (dead_task,)).fetchone()[0]
  668. conn.close()
  669. act(auth, dead_card["unregister"], {})
  670. conn = db()
  671. dead_row = conn.execute("select status from tasks where id = ?",
  672. (dead_task,)).fetchone()
  673. dead_session_row = conn.execute(
  674. "select outcome, ended_at is not null from sessions where id = ?",
  675. (dead_session,)).fetchone()
  676. dead_saddle_count = conn.execute(
  677. "select count(*) from saddles where id = ?", (dead_id,)).fetchone()[0]
  678. conn.close()
  679. check(dead_row is not None and dead_row[0] == "queued",
  680. "unregister of a dead-host saddle releases the held task to the queue")
  681. check(dead_session_row is not None and dead_session_row[0] == "stopped"
  682. and dead_session_row[1],
  683. "the offline release closes the held task's session 'stopped'")
  684. check(dead_saddle_count == 0, "the dead-host saddle row is gone")
  685. # Route 2: a REACHABLE host holding a running session — the abort
  686. # delivers (the daemon answers 200; its idle loop aborts nothing),
  687. # and the unregister itself carries the derived release through the
  688. # one owner (requeue flip + close + row deleted).
  689. saddles_state = slot_public(auth, "/saddles", "saddles")
  690. act(auth, saddles_state["register"], {"name": "quiet-host",
  691. "host": "127.0.0.1:%d" % fake.port})
  692. quiet_card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
  693. if s["name"] == "quiet-host"][0]
  694. quiet_id = int(quiet_card["id"])
  695. task_id_tmp = create_queued(auth, {
  696. "project_id": "1",
  697. "title": "Quiet-host unregister task",
  698. "priority": "2",
  699. })
  700. quiet_task = task_id_tmp
  701. conn = db()
  702. conn.execute("update tasks set status = 'working' where id = ?",
  703. (quiet_task,))
  704. conn.execute("insert into sessions (task_id, saddle_id, started_at)"
  705. " values (?, ?, strftime('%s','now'))", (quiet_task, quiet_id))
  706. conn.commit()
  707. quiet_session = conn.execute(
  708. "select id from sessions where task_id = ? and ended_at is null",
  709. (quiet_task,)).fetchone()[0]
  710. conn.close()
  711. act(auth, quiet_card["unregister"], {})
  712. conn = db()
  713. quiet_row = conn.execute("select status from tasks where id = ?",
  714. (quiet_task,)).fetchone()
  715. quiet_session_row = conn.execute(
  716. "select outcome, ended_at is not null from sessions where id = ?",
  717. (quiet_session,)).fetchone()
  718. quiet_saddle_count = conn.execute(
  719. "select count(*) from saddles where id = ?", (quiet_id,)).fetchone()[0]
  720. conn.close()
  721. check(quiet_row is not None and quiet_row[0] == "queued",
  722. "unregister's own release returns the derived held task to the queue")
  723. check(quiet_session_row is not None and quiet_session_row[0] == "stopped"
  724. and quiet_session_row[1],
  725. "the derived release closes the held task's session 'stopped'")
  726. check(quiet_saddle_count == 0, "the quiet-host saddle row is gone")
  727. # Route 3 (the derived-view pin): a 'working'-status task with NO
  728. # session derives NO hold — with the cache gone there is nothing to
  729. # key a release on, and the unregister must leave the drift row
  730. # untouched (a shape only fixtures can fabricate now: every real
  731. # path that flips a task running opens its session in the same
  732. # breath).
  733. saddles_state = slot_public(auth, "/saddles", "saddles")
  734. act(auth, saddles_state["register"], {"name": "empty-host",
  735. "host": "127.0.0.1:1"})
  736. empty_card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
  737. if s["name"] == "empty-host"][0]
  738. empty_id = int(empty_card["id"])
  739. task_id_tmp = create_queued(auth, {
  740. "project_id": "1",
  741. "title": "Empty-host drift task",
  742. "priority": "2",
  743. })
  744. empty_task = task_id_tmp
  745. conn = db()
  746. conn.execute("update tasks set status = 'working' where id = ?",
  747. (empty_task,))
  748. conn.commit()
  749. conn.close()
  750. act(auth, empty_card["unregister"], {})
  751. conn = db()
  752. empty_row = conn.execute("select status from tasks where id = ?",
  753. (empty_task,)).fetchone()
  754. empty_saddle_count = conn.execute(
  755. "select count(*) from saddles where id = ?", (empty_id,)).fetchone()[0]
  756. conn.close()
  757. check(empty_row is not None and empty_row[0] == "working",
  758. "no session means no derived hold — unregister releases nothing")
  759. check(empty_saddle_count == 0, "the empty-host saddle row is gone")