answers.py 27 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519
  1. """The answers 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: answers — the answer-resume machinery in both daemon shapes
  6. # (sleeping affine wake, mid-round verdict delivery), offline requeue
  7. # keeping parked-with-answered tasks, the staleness probe, and the
  8. # startup token migration (this shard's one deliberate app restart).
  9. # =============================================================================
  10. def run(ctx):
  11. auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token
  12. app = ctx.app
  13. print("== phase 2: answer resume — wake -> ask -> affinity claim ==")
  14. # The full live sequence against a daemon that asks after its
  15. # park: ask -> assignment -> blocked push (park) -> ask -> 204
  16. # -> sleeps (row idle) -> human answers -> the dispatcher WAKES
  17. # the sleeping affine saddle -> its ask allocates the resume,
  18. # the verdict riding the assignment -> the session continues.
  19. # A private capability tag keeps the fleet's other sleeping
  20. # saddles from claiming these tasks.
  21. saddles_state = slot_public(auth, "/saddles", "saddles")
  22. act(auth, saddles_state["create_capability"], {"name": "answerflow"})
  23. answerflow_cap = int([c["id"] for c in slot_public(auth, "/saddles", "saddles")["capabilities"]
  24. if c["name"] == "answerflow"][0])
  25. def answerflow_daemon(**kwargs):
  26. daemon = FakeSaddle()
  27. daemon.ask_on_register = False # manual, deterministic asks
  28. daemon.hold_state = True # the tests drive the lifecycle
  29. for key, value in kwargs.items():
  30. setattr(daemon, key, value)
  31. daemon.serve()
  32. return daemon
  33. def register_answerflow(daemon, name):
  34. act(auth, slot_public(auth, "/saddles", "saddles")["register"], {
  35. "name": name, "host": "127.0.0.1:%d" % daemon.port})
  36. act(auth, saddle_card(auth, name)["attach_capability"],
  37. {"capability_id": str(answerflow_cap)})
  38. def create_answerflow_task(title):
  39. return create_queued(auth, {
  40. "project_id": "1", "title": title, "priority": "2",
  41. "capability_id": [str(answerflow_cap)]})
  42. def retire_daemon(daemon, name):
  43. # deterministic teardown: a dead daemon and an offline row
  44. # never interfere with the sections that follow
  45. daemon.kill()
  46. act(auth, saddle_card(auth, name)["refresh"], {})
  47. sleeper = answerflow_daemon(ask_on_wake=True)
  48. register_answerflow(sleeper, "resume-sleeper")
  49. sleeping_task = create_answerflow_task("Sleeping resume task")
  50. ask_status, assignment = sleeper.ask_now()
  51. check(ask_status == 200 and int(assignment["task"]["id"]) == sleeping_task,
  52. "the daemon's ask claims the queued task")
  53. sleeping_session = assignment.get("session_id")
  54. sleeper.state = "blocked"
  55. check(sleeper.push_results("blocked", block={
  56. "question": "Pick a side", "type": "question",
  57. "options": ["Port", "Starboard"]}) == 200,
  58. "the blocked push parks the session")
  59. check(wait_for(lambda: task_state(auth, sleeping_task).get("status") == "blocked"
  60. or False, timeout=30, what="sleeping task parked") is not None,
  61. "the task parks as blocked")
  62. check(saddle_card(auth, "resume-sleeper")["status"] == "idle",
  63. "the parked session's saddle row reads idle")
  64. conn = db()
  65. parked_row = conn.execute(
  66. "select outcome, ended_at is not null from sessions where id = ?",
  67. (sleeping_session,)).fetchone()
  68. conn.close()
  69. check(tuple(parked_row) == ("parked", 1),
  70. "the park closes its session row 'parked' at park time (D4: the row"
  71. " aligns with the daemon's file-log park bracket)")
  72. ask_status, _ = sleeper.ask_now()
  73. check(ask_status == 204,
  74. "the parked daemon's ask returns 204 — it sleeps until woken")
  75. # Converted sleep(3): ticks pass with the daemon asleep and no
  76. # wake may arrive — a wake during the window fails loudly.
  77. check(quiet_window(quiet_seconds(),
  78. lambda: len(sleeper.wake_log) == 0,
  79. "sleeping daemon never woken"),
  80. "the sleeping daemon is never woken while the block waits")
  81. detail = task_state(auth, sleeping_task)
  82. side_block = [b for b in detail["blocks"] if b["question"] == "Pick a side"][0]
  83. act(auth, side_block["answer_action"], {"option": "Starboard"})
  84. check(wait_for(lambda: len(sleeper.wake_log) > 0 or False,
  85. timeout=30, what="wake after the answer") is not None,
  86. "answering the block wakes the sleeping affine saddle")
  87. # Answering = allocating (plan 11): the resume assignment carries
  88. # the verdict as resume_notice (stable wording) AND a FRESH
  89. # session id — the parked row closed 'parked' at its park (D4),
  90. # the daemon keys its resumed pushes on the new one.
  91. check(wait_for(lambda: any(
  92. int(p.get("task", {}).get("id") or 0) == sleeping_task
  93. and p.get("session_id") != sleeping_session
  94. and p.get("resume_notice") == "Human selected: Starboard"
  95. and any(a.get("answer") == "Starboard"
  96. for a in p.get("task", {}).get("answers", []))
  97. for p in sleeper.start_payloads[1:]) or False,
  98. timeout=30, what="resume allocation carries the verdict") is not None,
  99. "the woken daemon's ask allocates the resume — fresh session, notice carried")
  100. # Barrier: the resume's own working push must commit before the
  101. # manual done push (a late working write would otherwise race the
  102. # completion's full-row write).
  103. check(wait_for(lambda: task_state(auth, sleeping_task).get("status") == "working"
  104. or False, timeout=15, what="resumed round working") is not None,
  105. "the resumed round reports working before its completion push")
  106. # The resumed round runs to completion keyed on the fresh session.
  107. sleeper.push_results("done", summary="sleep-resumed round complete")
  108. check(wait_task_done(auth, sleeping_task, timeout=30) is not None,
  109. "the resumed session continues to completion")
  110. conn = db()
  111. rows = conn.execute(
  112. "select id, outcome, ended_at is not null from sessions where task_id = ?"
  113. " order by id", (sleeping_task,)).fetchall()
  114. conn.close()
  115. check(len(rows) == 2 and rows[0][1] == "parked" and rows[0][2]
  116. and rows[1][1] == "done" and rows[1][2],
  117. "parked session closed 'parked'; the resumed round ran on a fresh row"
  118. " to done (%s)" % (rows,))
  119. retire_daemon(sleeper, "resume-sleeper")
  120. print("== phase 2: answer resume — the parked daemon is woken (POST /answer retired) ==")
  121. # Post Task A every park FREES the daemon: after the blocked push
  122. # it lands idle and asks (204). The retired POST /answer channel
  123. # always refuses; the verdict reaches the parked workspace purely
  124. # through the wake -> ask -> affinity allocation carrying the
  125. # resume_notice + a fresh session id.
  126. parked = answerflow_daemon(ask_on_wake=True,
  127. ask_on_terminal=False)
  128. register_answerflow(parked, "resume-parked")
  129. parked_task = create_answerflow_task("Parked resume task")
  130. ask_status, assignment = parked.ask_now()
  131. check(ask_status == 200 and int(assignment["task"]["id"]) == parked_task,
  132. "the daemon's ask claims the task")
  133. parked_session = assignment.get("session_id")
  134. parked.state = "blocked"
  135. check(parked.push_results("blocked", block={
  136. "question": "Which database?", "type": "question",
  137. "options": ["SQLite", "Postgres"]}) == 200,
  138. "the blocked push parks the session mid-round")
  139. check(wait_for(lambda: task_state(auth, parked_task).get("status") == "blocked"
  140. or False, timeout=30, what="parked task blocked") is not None,
  141. "the task parks as blocked")
  142. check(saddle_card(auth, "resume-parked")["status"] == "idle",
  143. "the parked session's saddle row reads idle")
  144. # The freed daemon re-asks (its idle loop) and sleeps on 204.
  145. ask_status, _ = parked.ask_now()
  146. check(ask_status == 204,
  147. "the freed parked daemon asks again and sleeps (204)")
  148. # The retired channel always refuses — the web never depends on it.
  149. request = urllib.request.Request(
  150. "http://127.0.0.1:%d/answer" % parked.port,
  151. data=json.dumps({"answer": "SQLite", "branch_advanced": False}).encode(),
  152. method="POST")
  153. request.add_header("Content-Type", "application/json")
  154. try:
  155. urllib.request.urlopen(request, timeout=5)
  156. refused = False
  157. except urllib.error.HTTPError as error:
  158. refused = error.code == 409
  159. check(refused, "POST /answer always refuses (409) — parks resume via allocation")
  160. detail = task_state(auth, parked_task)
  161. db_block = [b for b in detail["blocks"] if b["question"] == "Which database?"][0]
  162. # branch_advanced on the resume assignment (the /poll verdict
  163. # channel's hint lives here now): a sibling on the same shared
  164. # working branch completed strictly after the block's park marker
  165. # — the resumed allocation must report branch_advanced=true so the
  166. # daemon knows to pull before continuing.
  167. conn = db()
  168. parked_block_created = conn.execute(
  169. "select created_at from task_blocks where id = ?", (db_block["id"],)
  170. ).fetchone()[0]
  171. conn.execute("update tasks set task_branch = 'shared/resume-branch'"
  172. " where id = ?", (parked_task,))
  173. conn.commit()
  174. conn.close()
  175. wait_for(lambda: int(time.time()) > parked_block_created or False,
  176. timeout=3, what="next clock second past the park marker")
  177. conn = db()
  178. conn.execute("insert into tasks (project_id, title, status, priority,"
  179. " task_branch, summary, completed_at, created_at)"
  180. " values (1, 'Resume sibling', 'done', 5, 'shared/resume-branch',"
  181. " 'sibling finished', strftime('%s','now'),"
  182. " strftime('%s','now','-300 seconds'))")
  183. conn.commit()
  184. conn.close()
  185. act(auth, db_block["answer_action"], {"option": "SQLite"})
  186. check(wait_for(lambda: len(parked.wake_log) > 0 or False,
  187. timeout=30, what="wake after the answer") is not None,
  188. "answering the block wakes the parked (idle) affine saddle")
  189. check(wait_for(lambda: any(
  190. int(p.get("task", {}).get("id") or 0) == parked_task
  191. and p.get("session_id") != parked_session
  192. and p.get("resume_notice") == "Human selected: SQLite"
  193. and p.get("branch_advanced") is True
  194. for p in parked.start_payloads[1:]) or False,
  195. timeout=30, what="parked resume allocation") is not None,
  196. "the parked daemon's ask allocates the resume — fresh session, formatted notice,"
  197. " branch_advanced=true after the sibling's completion")
  198. # The hold_state daemon drives its round manually: the resumed
  199. # round completes on the fresh session row.
  200. parked.push_results("done", summary="parked round resumed and finished")
  201. check(wait_task_done(auth, parked_task, timeout=90) is not None,
  202. "the parked session continues to completion")
  203. conn = db()
  204. row = conn.execute("select outcome from sessions where id = ?",
  205. (parked_session,)).fetchone()
  206. session_rows = conn.execute(
  207. "select id, outcome from sessions where task_id = ? order by id",
  208. (parked_task,)).fetchall()
  209. conn.close()
  210. check(row is not None and row[0] == "parked"
  211. and len(session_rows) == 2 and session_rows[0] == (parked_session, "parked"),
  212. "the parked session closed 'parked' (at its park, D4) and the resumed round"
  213. " kept its own row (%s)" % (session_rows,))
  214. retire_daemon(parked, "resume-parked")
  215. print("== plan 11: plan acceptance — accept / reject / auto-approve ==")
  216. # A type 'plan' block parks like any other; Accept and
  217. # Reject-with-feedback drive the same wake -> ask -> allocation
  218. # resume as any verdict, the stable wording riding the
  219. # resume_notice. Plan 12: the daemon only SENDS plan blocks on
  220. # goal runs now (the goal page owns the review); this covers the
  221. # generic verdict path a goal-run's mirrored block also uses. The
  222. # auto-approve option is gone.
  223. planner = answerflow_daemon(ask_on_wake=True, ask_on_terminal=False)
  224. register_answerflow(planner, "plan-planner")
  225. def park_on_plan(task_id, markdown):
  226. planner.state = "blocked"
  227. check(planner.push_results("blocked", block={
  228. "question": "Plan submitted for review", "type": "plan",
  229. "body": markdown}) == 200,
  230. "the plan push parks the session")
  231. # Wait for the block row itself (the auto-approve variant is
  232. # answered — and may resume — before any status poll lands).
  233. check(wait_for(lambda: any(b["block_type"] == "plan"
  234. for b in task_state(auth, task_id)["blocks"])
  235. or False, timeout=30, what="plan block raised") is not None,
  236. "the plan block reaches the task")
  237. return [b for b in task_state(auth, task_id)["blocks"]
  238. if b["block_type"] == "plan"][0]
  239. # (a) Accept resumes with the approval notice.
  240. accept_task = create_answerflow_task("Plan accept task")
  241. ask_status, accept_assignment = planner.ask_now()
  242. check(ask_status == 200 and int(accept_assignment["task"]["id"]) == accept_task,
  243. "the planner claims the accept task")
  244. accept_parked_session = accept_assignment.get("session_id")
  245. plan_block = park_on_plan(accept_task, "# Steps\n\n1. Wire\n2. Test")
  246. check(plan_block.get("answered") is not True,
  247. "the plan block waits for a human verdict")
  248. # Plan 16 S2: the accept form posts the verdict marker; the server
  249. # builds the stable constant (byte-identical wire).
  250. act(auth, plan_block["answer_action"], {"verdict": "accept"})
  251. check(wait_for(lambda: any(
  252. int(p.get("task", {}).get("id") or 0) == accept_task
  253. and p.get("session_id") != accept_parked_session
  254. and p.get("resume_notice") == "Plan approved. Proceed with implementation."
  255. for p in planner.start_payloads[1:]) or False,
  256. timeout=30, what="accept resume allocation") is not None,
  257. "accept resumes the parked daemon with the approval notice")
  258. planner.push_results("done", summary="plan accepted and implemented")
  259. check(wait_task_done(auth, accept_task, timeout=60) is not None,
  260. "the accepted plan's task completes")
  261. planner.set_state("idle")
  262. # (b) Reject-with-feedback loops the feedback notice back.
  263. reject_task = create_answerflow_task("Plan reject task")
  264. ask_status, reject_assignment = planner.ask_now()
  265. check(ask_status == 200 and int(reject_assignment["task"]["id"]) == reject_task,
  266. "the planner claims the reject task")
  267. reject_parked_session = reject_assignment.get("session_id")
  268. plan_block = park_on_plan(reject_task, "# Draft\n\nJust wing it")
  269. act(auth, plan_block["answer_action"], {"feedback": "Add a test section"})
  270. reject_block = [b for b in task_state(auth, reject_task)["blocks"]
  271. if b["block_type"] == "plan"][0]
  272. check(reject_block.get("answer")
  273. == "Plan feedback received:\nAdd a test section",
  274. "reject stores the feedback in the stable wording (%s)"
  275. % reject_block.get("answer"))
  276. check(wait_for(lambda: any(
  277. int(p.get("task", {}).get("id") or 0) == reject_task
  278. and p.get("session_id") != reject_parked_session
  279. and p.get("resume_notice") == "Plan feedback received:\nAdd a test section"
  280. for p in planner.start_payloads[1:]) or False,
  281. timeout=30, what="reject resume allocation") is not None,
  282. "reject resumes the parked daemon with the feedback notice")
  283. planner.push_results("done", summary="plan revised and delivered")
  284. check(wait_task_done(auth, reject_task, timeout=60) is not None,
  285. "the revised plan's task completes")
  286. planner.set_state("idle")
  287. retire_daemon(planner, "plan-planner")
  288. print("== phase 2: offline requeue keeps parked-with-answered tasks ==")
  289. # A daemon dying while its parked task carries an answered
  290. # verdict: going offline must not requeue the task — the resume
  291. # machinery (delivery once the daemon returns, or the affinity
  292. # allocation on its next ask) needs the parked record (the
  293. # session row closed 'parked' at its park, D4) and the task left
  294. # parked. Both park kinds (question and review) exercise the
  295. # unreachable path's requeue decision.
  296. stranded = answerflow_daemon()
  297. register_answerflow(stranded, "resume-stranded")
  298. stranded_task = create_answerflow_task("Stranded answered task")
  299. ask_status, _ = stranded.ask_now()
  300. check(ask_status == 200, "the stranded daemon claims its task")
  301. stranded.state = "blocked"
  302. stranded.push_results("blocked", block={
  303. "question": "Hold or fold?", "type": "question", "options": ["Hold", "Fold"]})
  304. check(wait_for(lambda: task_state(auth, stranded_task).get("status") == "blocked"
  305. or False, timeout=30, what="stranded task parked") is not None,
  306. "the stranded task parks as blocked")
  307. detail = task_state(auth, stranded_task)
  308. hold_block = [b for b in detail["blocks"] if b["question"] == "Hold or fold?"][0]
  309. act(auth, hold_block["answer_action"], {"option": "Hold"})
  310. stranded.kill() # the daemon dies before the verdict is delivered
  311. act(auth, saddle_card(auth, "resume-stranded")["refresh"], {})
  312. check(wait_for(lambda: saddle_card(auth, "resume-stranded")["status"] == "offline"
  313. or False, timeout=30, what="stranded saddle offline") is not None,
  314. "the dead daemon's saddle row goes offline")
  315. # Converted sleep(7): offline ticks run while the task MUST stay
  316. # parked — a requeue during the window fails loudly.
  317. check(quiet_window(quiet_seconds(),
  318. lambda: task_state(auth, stranded_task)["status"] == "blocked",
  319. "stranded task not requeued"),
  320. "the parked-with-answered task is not requeued by the offline pass")
  321. conn = db()
  322. stranded_rows = conn.execute(
  323. "select outcome, ended_at is not null from sessions where task_id = ?"
  324. " order by id desc limit 1", (stranded_task,)).fetchone()
  325. conn.close()
  326. check(tuple(stranded_rows) == ("parked", 1),
  327. "the parked-with-answered task keeps its parked session record ('parked')"
  328. " for the resume")
  329. reviewer = answerflow_daemon(ask_on_terminal=False)
  330. register_answerflow(reviewer, "resume-reviewer")
  331. review_task = create_answerflow_task("Review park task")
  332. ask_status, _ = reviewer.ask_now()
  333. check(ask_status == 200, "the reviewer claims its task")
  334. # Post Task A the review park frees the daemon: the blocked push
  335. # carries the review block and parks the row idle.
  336. reviewer.state = "blocked"
  337. check(reviewer.push_results("blocked", block={
  338. "question": "Review requested", "type": "review",
  339. "body": "Ready for review: wired and tested."}) == 200,
  340. "the review park push carries the review block")
  341. check(wait_for(lambda: task_state(auth, review_task).get("status") == "review_requested"
  342. or False, timeout=30, what="review park raised") is not None,
  343. "the review block parks the task as review_requested (row freed)")
  344. # The verdict lands (nothing delivers it anymore); THEN the daemon
  345. # dies and the manual refresh offlines it — the parked-with-answered
  346. # task must keep its session for the eventual resume.
  347. detail = task_state(auth, review_task)
  348. verdict_block = [b for b in detail["blocks"]
  349. if b["block_type"] == "review"][0]
  350. act(auth, verdict_block["answer_action"], {"verdict": "accept"})
  351. reviewer.kill()
  352. act(auth, saddle_card(auth, "resume-reviewer")["refresh"], {})
  353. check(wait_for(lambda: saddle_card(auth, "resume-reviewer")["status"] == "offline"
  354. or False, timeout=30, what="reviewer offline") is not None,
  355. "the dead reviewer goes offline on the failed probe")
  356. # Converted sleep(7): offline ticks run while the task MUST stay
  357. # review-parked — a requeue during the window fails loudly.
  358. check(quiet_window(quiet_seconds(),
  359. lambda: task_state(auth, review_task)["status"] == "review_requested",
  360. "review-parked task not requeued"),
  361. "the review-parked task with an answered verdict is never requeued")
  362. conn = db()
  363. review_rows = conn.execute(
  364. "select outcome, ended_at is not null from sessions where task_id = ?"
  365. " order by id desc limit 1", (review_task,)).fetchone()
  366. conn.close()
  367. check(tuple(review_rows) == ("parked", 1),
  368. "the review-parked task keeps its parked session record ('parked')"
  369. " for the resume")
  370. print("== phase 2: staleness — one probe decides ==")
  371. # A silent-but-alive runner (long think, nothing to flush): after
  372. # SADDLE_STALE_S (3s here) of silence it owes ONE exceptional
  373. # GET /status; the answer keeps it running and the silence
  374. # episode gets no second probe.
  375. task_id_tmp = create_queued(auth, {
  376. "project_id": "1",
  377. "title": "Silent thinker",
  378. "priority": "2",
  379. })
  380. thinker = task_id_tmp
  381. check(wait_for(lambda: task_state(auth, thinker).get("status") == "working"
  382. or False, timeout=60, what="thinker working") is not None,
  383. "silent thinker reaches working")
  384. fake.silent = True # the daemon goes quiet mid-round (alive)
  385. probes_at_start = fake.status_hits
  386. check(wait_for(lambda: fake.status_hits > probes_at_start or False,
  387. timeout=30, what="exceptional probe arrives") is not None,
  388. "a silent runner gets its exceptional GET /status probe")
  389. # Converted sleep(8)+sleep(6): a quiet window across several
  390. # ticks in which the probe count MUST NOT move — a second probe
  391. # (the poll drumbeat this guards against) fails loudly.
  392. second_window = fake.status_hits
  393. check(quiet_window(quiet_seconds(),
  394. lambda: fake.status_hits == second_window,
  395. "no second staleness probe"),
  396. "the silence episode is probed exactly once (no poll drumbeat)")
  397. check(task_state(auth, thinker).get("status") == "working"
  398. and saddle_card(auth, "fake-saddle")["status"] == "busy",
  399. "the answered probe keeps the task running (no requeue)")
  400. # Output flows again: the episode ends, the clock resets.
  401. fake.silent = False
  402. fake.push_chunks(["resumed output\r\n"])
  403. check(wait_task_done(auth, thinker, timeout=60) is not None,
  404. "the silent thinker completes once output flows again")
  405. # The daemon vanishing mid-task: silence for the window, the
  406. # probe fails, the task requeues with an agent comment and the
  407. # saddle goes offline.
  408. task_id_tmp = create_queued(auth, {
  409. "project_id": "1",
  410. "title": "Vanish victim",
  411. "priority": "2",
  412. })
  413. vanish = task_id_tmp
  414. check(wait_for(lambda: task_state(auth, vanish).get("status") == "working"
  415. or False, timeout=60, what="vanish victim working") is not None,
  416. "vanish victim reaches working")
  417. fake.kill()
  418. check(wait_for(lambda: saddle_card(auth, "fake-saddle")["status"] == "offline" or False,
  419. timeout=60, what="daemon death -> offline") is not None,
  420. "vanishing daemon drives the saddle offline (failed staleness probe)")
  421. check(wait_for(lambda: any("unreachable" in c["body"]
  422. and c.get("is_system")
  423. for c in task_state(auth, vanish).get("comments", []))
  424. or False, timeout=30, what="unreachable comment") is not None,
  425. "the affected task gets an unreachable system row (never an agent comment)")
  426. check(task_state(auth, vanish)["status"] == "queued",
  427. "the task requeues after the daemon death")
  428. card = saddle_card(auth, "fake-saddle")
  429. check(card.get("health") == "offline",
  430. "health badge reports offline")
  431. # The unreachable row flag marks the detection window: a task the
  432. # web still sees working while its saddle is offline/fenced, before
  433. # the requeue pass moves it. Controlled window (the real
  434. # transitions above are atomic with their requeues).
  435. conn = db()
  436. conn.execute("insert into tasks (project_id, title, status, priority,"
  437. " created_at) values (1, 'Unreachable row', 'working', 5,"
  438. " strftime('%s','now'))")
  439. unreachable_row = conn.execute("select max(id) from tasks").fetchone()[0]
  440. conn.execute("insert into sessions (task_id, saddle_id, started_at)"
  441. " values (?, ?, strftime('%s','now'))", (unreachable_row, fake_id))
  442. conn.commit()
  443. conn.close()
  444. check(wait_for(lambda: (lambda row: row.get("unreachable") is True)(
  445. [t for t in slot_public(auth, "/tasks", "tasks")["tasks"]
  446. if int(t["id"]) == unreachable_row][0]) or False,
  447. timeout=20, what="unreachable flag") is not None,
  448. "queue row shows the unreachable state in the window")
  449. print("== phase 1: saddles.token is the single credential store (arch 24) ==")
  450. # The legacy api_tokens table and every fallback read died with the
  451. # poll era: a restart proves the migrated schema carries no
  452. # api_tokens table, the row token survives and still authenticates,
  453. # and a legacy-shaped token string authorises nothing (no fallback
  454. # read exists to honour it). Fixture: this shard's own racer.
  455. fake2 = FakeSaddle()
  456. fake2.ask_on_register = False
  457. fake2.serve(port=free_port())
  458. saddles_state = slot_public(auth, "/saddles", "saddles")
  459. act(auth, saddles_state["register"], {"name": "racer-two",
  460. "host": "127.0.0.1:%d" % fake2.port})
  461. racer_two_id = int([s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
  462. if s["name"] == "racer-two"][0]["id"])
  463. racer_two_token = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
  464. if s["name"] == "racer-two"][0]["token"]
  465. app.terminate()
  466. app.wait(timeout=10)
  467. app = start_app()
  468. auth = login("admin", "adminpass123")
  469. ctx.app = app # teardown tracks the restarted process
  470. ctx.auth = auth
  471. def schema_tables():
  472. conn = db()
  473. try:
  474. return {row[0] for row in conn.execute(
  475. "select name from sqlite_master where type = 'table'")}
  476. finally:
  477. conn.close()
  478. check("api_tokens" not in schema_tables(),
  479. "the fresh schema carries no api_tokens table (arch 24)")
  480. conn = db()
  481. row_token = conn.execute("select token from saddles where id = ?",
  482. (racer_two_id,)).fetchone()[0]
  483. conn.close()
  484. check(row_token == racer_two_token and (row_token or "").startswith("saddle-"),
  485. "the per-saddle row token survives the restart unchanged")
  486. status, _ = saddle_api("POST", "/api/saddle/next-task", racer_two_token)
  487. check(status in (200, 204),
  488. "the row token still authenticates after the restart")
  489. status, _ = saddle_api("POST", "/api/saddle/next-task", "saddle-legacy-copied-token")
  490. check(status == 401,
  491. "a legacy api_tokens-shaped credential authorises nothing (fallback gone)")
  492. fake2.token = racer_two_token
  493. fake2.ask_now()
  494. check(True, "the fleet re-asks after the restart (sleeping re-established)")