"""The answers bundle (moved verbatim from the pre-split web_stack_test.py; see test/harness.py for the shared machinery and the runner).""" from harness import * # noqa: F401,F403 — the shared wire helpers # ============================================================================= # Bundle: answers — the answer-resume machinery in both daemon shapes # (sleeping affine wake, mid-round verdict delivery), offline requeue # keeping parked-with-answered tasks, the staleness probe, and the # startup token migration (this shard's one deliberate app restart). # ============================================================================= def run(ctx): auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token app = ctx.app print("== phase 2: answer resume — wake -> ask -> affinity claim ==") # The full live sequence against a daemon that asks after its # park: ask -> assignment -> blocked push (park) -> ask -> 204 # -> sleeps (row idle) -> human answers -> the dispatcher WAKES # the sleeping affine saddle -> its ask allocates the resume, # the verdict riding the assignment -> the session continues. # A private capability tag keeps the fleet's other sleeping # saddles from claiming these tasks. saddles_state = slot_public(auth, "/saddles", "saddles") act(auth, saddles_state["create_capability"], {"name": "answerflow"}) answerflow_cap = int([c["id"] for c in slot_public(auth, "/saddles", "saddles")["capabilities"] if c["name"] == "answerflow"][0]) def answerflow_daemon(**kwargs): daemon = FakeSaddle() daemon.ask_on_register = False # manual, deterministic asks daemon.hold_state = True # the tests drive the lifecycle for key, value in kwargs.items(): setattr(daemon, key, value) daemon.serve() return daemon def register_answerflow(daemon, name): act(auth, slot_public(auth, "/saddles", "saddles")["register"], { "name": name, "host": "127.0.0.1:%d" % daemon.port}) act(auth, saddle_card(auth, name)["attach_capability"], {"capability_id": str(answerflow_cap)}) def create_answerflow_task(title): return create_queued(auth, { "project_id": "1", "title": title, "priority": "2", "capability_id": [str(answerflow_cap)]}) def retire_daemon(daemon, name): # deterministic teardown: a dead daemon and an offline row # never interfere with the sections that follow daemon.kill() act(auth, saddle_card(auth, name)["refresh"], {}) sleeper = answerflow_daemon(ask_on_wake=True) register_answerflow(sleeper, "resume-sleeper") sleeping_task = create_answerflow_task("Sleeping resume task") ask_status, assignment = sleeper.ask_now() check(ask_status == 200 and int(assignment["task"]["id"]) == sleeping_task, "the daemon's ask claims the queued task") sleeping_session = assignment.get("session_id") sleeper.state = "blocked" check(sleeper.push_results("blocked", block={ "question": "Pick a side", "type": "question", "options": ["Port", "Starboard"]}) == 200, "the blocked push parks the session") check(wait_for(lambda: task_state(auth, sleeping_task).get("status") == "blocked" or False, timeout=30, what="sleeping task parked") is not None, "the task parks as blocked") check(saddle_card(auth, "resume-sleeper")["status"] == "idle", "the parked session's saddle row reads idle") conn = db() parked_row = conn.execute( "select outcome, ended_at is not null from sessions where id = ?", (sleeping_session,)).fetchone() conn.close() check(tuple(parked_row) == ("parked", 1), "the park closes its session row 'parked' at park time (D4: the row" " aligns with the daemon's file-log park bracket)") ask_status, _ = sleeper.ask_now() check(ask_status == 204, "the parked daemon's ask returns 204 — it sleeps until woken") # Converted sleep(3): ticks pass with the daemon asleep and no # wake may arrive — a wake during the window fails loudly. check(quiet_window(quiet_seconds(), lambda: len(sleeper.wake_log) == 0, "sleeping daemon never woken"), "the sleeping daemon is never woken while the block waits") detail = task_state(auth, sleeping_task) side_block = [b for b in detail["blocks"] if b["question"] == "Pick a side"][0] act(auth, side_block["answer_action"], {"option": "Starboard"}) check(wait_for(lambda: len(sleeper.wake_log) > 0 or False, timeout=30, what="wake after the answer") is not None, "answering the block wakes the sleeping affine saddle") # Answering = allocating (plan 11): the resume assignment carries # the verdict as resume_notice (stable wording) AND a FRESH # session id — the parked row closed 'parked' at its park (D4), # the daemon keys its resumed pushes on the new one. check(wait_for(lambda: any( int(p.get("task", {}).get("id") or 0) == sleeping_task and p.get("session_id") != sleeping_session and p.get("resume_notice") == "Human selected: Starboard" and any(a.get("answer") == "Starboard" for a in p.get("task", {}).get("answers", [])) for p in sleeper.start_payloads[1:]) or False, timeout=30, what="resume allocation carries the verdict") is not None, "the woken daemon's ask allocates the resume — fresh session, notice carried") # Barrier: the resume's own working push must commit before the # manual done push (a late working write would otherwise race the # completion's full-row write). check(wait_for(lambda: task_state(auth, sleeping_task).get("status") == "working" or False, timeout=15, what="resumed round working") is not None, "the resumed round reports working before its completion push") # The resumed round runs to completion keyed on the fresh session. sleeper.push_results("done", summary="sleep-resumed round complete") check(wait_task_done(auth, sleeping_task, timeout=30) is not None, "the resumed session continues to completion") conn = db() rows = conn.execute( "select id, outcome, ended_at is not null from sessions where task_id = ?" " order by id", (sleeping_task,)).fetchall() conn.close() check(len(rows) == 2 and rows[0][1] == "parked" and rows[0][2] and rows[1][1] == "done" and rows[1][2], "parked session closed 'parked'; the resumed round ran on a fresh row" " to done (%s)" % (rows,)) retire_daemon(sleeper, "resume-sleeper") print("== phase 2: answer resume — the parked daemon is woken (POST /answer retired) ==") # Post Task A every park FREES the daemon: after the blocked push # it lands idle and asks (204). The retired POST /answer channel # always refuses; the verdict reaches the parked workspace purely # through the wake -> ask -> affinity allocation carrying the # resume_notice + a fresh session id. parked = answerflow_daemon(ask_on_wake=True, ask_on_terminal=False) register_answerflow(parked, "resume-parked") parked_task = create_answerflow_task("Parked resume task") ask_status, assignment = parked.ask_now() check(ask_status == 200 and int(assignment["task"]["id"]) == parked_task, "the daemon's ask claims the task") parked_session = assignment.get("session_id") parked.state = "blocked" check(parked.push_results("blocked", block={ "question": "Which database?", "type": "question", "options": ["SQLite", "Postgres"]}) == 200, "the blocked push parks the session mid-round") check(wait_for(lambda: task_state(auth, parked_task).get("status") == "blocked" or False, timeout=30, what="parked task blocked") is not None, "the task parks as blocked") check(saddle_card(auth, "resume-parked")["status"] == "idle", "the parked session's saddle row reads idle") # The freed daemon re-asks (its idle loop) and sleeps on 204. ask_status, _ = parked.ask_now() check(ask_status == 204, "the freed parked daemon asks again and sleeps (204)") # The retired channel always refuses — the web never depends on it. request = urllib.request.Request( "http://127.0.0.1:%d/answer" % parked.port, data=json.dumps({"answer": "SQLite", "branch_advanced": False}).encode(), method="POST") request.add_header("Content-Type", "application/json") try: urllib.request.urlopen(request, timeout=5) refused = False except urllib.error.HTTPError as error: refused = error.code == 409 check(refused, "POST /answer always refuses (409) — parks resume via allocation") detail = task_state(auth, parked_task) db_block = [b for b in detail["blocks"] if b["question"] == "Which database?"][0] # branch_advanced on the resume assignment (the /poll verdict # channel's hint lives here now): a sibling on the same shared # working branch completed strictly after the block's park marker # — the resumed allocation must report branch_advanced=true so the # daemon knows to pull before continuing. conn = db() parked_block_created = conn.execute( "select created_at from task_blocks where id = ?", (db_block["id"],) ).fetchone()[0] conn.execute("update tasks set task_branch = 'shared/resume-branch'" " where id = ?", (parked_task,)) conn.commit() conn.close() wait_for(lambda: int(time.time()) > parked_block_created or False, timeout=3, what="next clock second past the park marker") conn = db() conn.execute("insert into tasks (project_id, title, status, priority," " task_branch, summary, completed_at, created_at)" " values (1, 'Resume sibling', 'done', 5, 'shared/resume-branch'," " 'sibling finished', strftime('%s','now')," " strftime('%s','now','-300 seconds'))") conn.commit() conn.close() act(auth, db_block["answer_action"], {"option": "SQLite"}) check(wait_for(lambda: len(parked.wake_log) > 0 or False, timeout=30, what="wake after the answer") is not None, "answering the block wakes the parked (idle) affine saddle") check(wait_for(lambda: any( int(p.get("task", {}).get("id") or 0) == parked_task and p.get("session_id") != parked_session and p.get("resume_notice") == "Human selected: SQLite" and p.get("branch_advanced") is True for p in parked.start_payloads[1:]) or False, timeout=30, what="parked resume allocation") is not None, "the parked daemon's ask allocates the resume — fresh session, formatted notice," " branch_advanced=true after the sibling's completion") # The hold_state daemon drives its round manually: the resumed # round completes on the fresh session row. parked.push_results("done", summary="parked round resumed and finished") check(wait_task_done(auth, parked_task, timeout=90) is not None, "the parked session continues to completion") conn = db() row = conn.execute("select outcome from sessions where id = ?", (parked_session,)).fetchone() session_rows = conn.execute( "select id, outcome from sessions where task_id = ? order by id", (parked_task,)).fetchall() conn.close() check(row is not None and row[0] == "parked" and len(session_rows) == 2 and session_rows[0] == (parked_session, "parked"), "the parked session closed 'parked' (at its park, D4) and the resumed round" " kept its own row (%s)" % (session_rows,)) retire_daemon(parked, "resume-parked") print("== plan 11: plan acceptance — accept / reject / auto-approve ==") # A type 'plan' block parks like any other; Accept and # Reject-with-feedback drive the same wake -> ask -> allocation # resume as any verdict, the stable wording riding the # resume_notice. Plan 12: the daemon only SENDS plan blocks on # goal runs now (the goal page owns the review); this covers the # generic verdict path a goal-run's mirrored block also uses. The # auto-approve option is gone. planner = answerflow_daemon(ask_on_wake=True, ask_on_terminal=False) register_answerflow(planner, "plan-planner") def park_on_plan(task_id, markdown): planner.state = "blocked" check(planner.push_results("blocked", block={ "question": "Plan submitted for review", "type": "plan", "body": markdown}) == 200, "the plan push parks the session") # Wait for the block row itself (the auto-approve variant is # answered — and may resume — before any status poll lands). check(wait_for(lambda: any(b["block_type"] == "plan" for b in task_state(auth, task_id)["blocks"]) or False, timeout=30, what="plan block raised") is not None, "the plan block reaches the task") return [b for b in task_state(auth, task_id)["blocks"] if b["block_type"] == "plan"][0] # (a) Accept resumes with the approval notice. accept_task = create_answerflow_task("Plan accept task") ask_status, accept_assignment = planner.ask_now() check(ask_status == 200 and int(accept_assignment["task"]["id"]) == accept_task, "the planner claims the accept task") accept_parked_session = accept_assignment.get("session_id") plan_block = park_on_plan(accept_task, "# Steps\n\n1. Wire\n2. Test") check(plan_block.get("answered") is not True, "the plan block waits for a human verdict") # Plan 16 S2: the accept form posts the verdict marker; the server # builds the stable constant (byte-identical wire). act(auth, plan_block["answer_action"], {"verdict": "accept"}) check(wait_for(lambda: any( int(p.get("task", {}).get("id") or 0) == accept_task and p.get("session_id") != accept_parked_session and p.get("resume_notice") == "Plan approved. Proceed with implementation." for p in planner.start_payloads[1:]) or False, timeout=30, what="accept resume allocation") is not None, "accept resumes the parked daemon with the approval notice") planner.push_results("done", summary="plan accepted and implemented") check(wait_task_done(auth, accept_task, timeout=60) is not None, "the accepted plan's task completes") planner.set_state("idle") # (b) Reject-with-feedback loops the feedback notice back. reject_task = create_answerflow_task("Plan reject task") ask_status, reject_assignment = planner.ask_now() check(ask_status == 200 and int(reject_assignment["task"]["id"]) == reject_task, "the planner claims the reject task") reject_parked_session = reject_assignment.get("session_id") plan_block = park_on_plan(reject_task, "# Draft\n\nJust wing it") act(auth, plan_block["answer_action"], {"feedback": "Add a test section"}) reject_block = [b for b in task_state(auth, reject_task)["blocks"] if b["block_type"] == "plan"][0] check(reject_block.get("answer") == "Plan feedback received:\nAdd a test section", "reject stores the feedback in the stable wording (%s)" % reject_block.get("answer")) check(wait_for(lambda: any( int(p.get("task", {}).get("id") or 0) == reject_task and p.get("session_id") != reject_parked_session and p.get("resume_notice") == "Plan feedback received:\nAdd a test section" for p in planner.start_payloads[1:]) or False, timeout=30, what="reject resume allocation") is not None, "reject resumes the parked daemon with the feedback notice") planner.push_results("done", summary="plan revised and delivered") check(wait_task_done(auth, reject_task, timeout=60) is not None, "the revised plan's task completes") planner.set_state("idle") retire_daemon(planner, "plan-planner") print("== phase 2: offline requeue keeps parked-with-answered tasks ==") # A daemon dying while its parked task carries an answered # verdict: going offline must not requeue the task — the resume # machinery (delivery once the daemon returns, or the affinity # allocation on its next ask) needs the parked record (the # session row closed 'parked' at its park, D4) and the task left # parked. Both park kinds (question and review) exercise the # unreachable path's requeue decision. stranded = answerflow_daemon() register_answerflow(stranded, "resume-stranded") stranded_task = create_answerflow_task("Stranded answered task") ask_status, _ = stranded.ask_now() check(ask_status == 200, "the stranded daemon claims its task") stranded.state = "blocked" stranded.push_results("blocked", block={ "question": "Hold or fold?", "type": "question", "options": ["Hold", "Fold"]}) check(wait_for(lambda: task_state(auth, stranded_task).get("status") == "blocked" or False, timeout=30, what="stranded task parked") is not None, "the stranded task parks as blocked") detail = task_state(auth, stranded_task) hold_block = [b for b in detail["blocks"] if b["question"] == "Hold or fold?"][0] act(auth, hold_block["answer_action"], {"option": "Hold"}) stranded.kill() # the daemon dies before the verdict is delivered act(auth, saddle_card(auth, "resume-stranded")["refresh"], {}) check(wait_for(lambda: saddle_card(auth, "resume-stranded")["status"] == "offline" or False, timeout=30, what="stranded saddle offline") is not None, "the dead daemon's saddle row goes offline") # Converted sleep(7): offline ticks run while the task MUST stay # parked — a requeue during the window fails loudly. check(quiet_window(quiet_seconds(), lambda: task_state(auth, stranded_task)["status"] == "blocked", "stranded task not requeued"), "the parked-with-answered task is not requeued by the offline pass") conn = db() stranded_rows = conn.execute( "select outcome, ended_at is not null from sessions where task_id = ?" " order by id desc limit 1", (stranded_task,)).fetchone() conn.close() check(tuple(stranded_rows) == ("parked", 1), "the parked-with-answered task keeps its parked session record ('parked')" " for the resume") reviewer = answerflow_daemon(ask_on_terminal=False) register_answerflow(reviewer, "resume-reviewer") review_task = create_answerflow_task("Review park task") ask_status, _ = reviewer.ask_now() check(ask_status == 200, "the reviewer claims its task") # Post Task A the review park frees the daemon: the blocked push # carries the review block and parks the row idle. reviewer.state = "blocked" check(reviewer.push_results("blocked", block={ "question": "Review requested", "type": "review", "body": "Ready for review: wired and tested."}) == 200, "the review park push carries the review block") check(wait_for(lambda: task_state(auth, review_task).get("status") == "review_requested" or False, timeout=30, what="review park raised") is not None, "the review block parks the task as review_requested (row freed)") # The verdict lands (nothing delivers it anymore); THEN the daemon # dies and the manual refresh offlines it — the parked-with-answered # task must keep its session for the eventual resume. detail = task_state(auth, review_task) verdict_block = [b for b in detail["blocks"] if b["block_type"] == "review"][0] act(auth, verdict_block["answer_action"], {"verdict": "accept"}) reviewer.kill() act(auth, saddle_card(auth, "resume-reviewer")["refresh"], {}) check(wait_for(lambda: saddle_card(auth, "resume-reviewer")["status"] == "offline" or False, timeout=30, what="reviewer offline") is not None, "the dead reviewer goes offline on the failed probe") # Converted sleep(7): offline ticks run while the task MUST stay # review-parked — a requeue during the window fails loudly. check(quiet_window(quiet_seconds(), lambda: task_state(auth, review_task)["status"] == "review_requested", "review-parked task not requeued"), "the review-parked task with an answered verdict is never requeued") conn = db() review_rows = conn.execute( "select outcome, ended_at is not null from sessions where task_id = ?" " order by id desc limit 1", (review_task,)).fetchone() conn.close() check(tuple(review_rows) == ("parked", 1), "the review-parked task keeps its parked session record ('parked')" " for the resume") print("== phase 2: staleness — one probe decides ==") # A silent-but-alive runner (long think, nothing to flush): after # SADDLE_STALE_S (3s here) of silence it owes ONE exceptional # GET /status; the answer keeps it running and the silence # episode gets no second probe. task_id_tmp = create_queued(auth, { "project_id": "1", "title": "Silent thinker", "priority": "2", }) thinker = task_id_tmp check(wait_for(lambda: task_state(auth, thinker).get("status") == "working" or False, timeout=60, what="thinker working") is not None, "silent thinker reaches working") fake.silent = True # the daemon goes quiet mid-round (alive) probes_at_start = fake.status_hits check(wait_for(lambda: fake.status_hits > probes_at_start or False, timeout=30, what="exceptional probe arrives") is not None, "a silent runner gets its exceptional GET /status probe") # Converted sleep(8)+sleep(6): a quiet window across several # ticks in which the probe count MUST NOT move — a second probe # (the poll drumbeat this guards against) fails loudly. second_window = fake.status_hits check(quiet_window(quiet_seconds(), lambda: fake.status_hits == second_window, "no second staleness probe"), "the silence episode is probed exactly once (no poll drumbeat)") check(task_state(auth, thinker).get("status") == "working" and saddle_card(auth, "fake-saddle")["status"] == "busy", "the answered probe keeps the task running (no requeue)") # Output flows again: the episode ends, the clock resets. fake.silent = False fake.push_chunks(["resumed output\r\n"]) check(wait_task_done(auth, thinker, timeout=60) is not None, "the silent thinker completes once output flows again") # The daemon vanishing mid-task: silence for the window, the # probe fails, the task requeues with an agent comment and the # saddle goes offline. task_id_tmp = create_queued(auth, { "project_id": "1", "title": "Vanish victim", "priority": "2", }) vanish = task_id_tmp check(wait_for(lambda: task_state(auth, vanish).get("status") == "working" or False, timeout=60, what="vanish victim working") is not None, "vanish victim reaches working") fake.kill() check(wait_for(lambda: saddle_card(auth, "fake-saddle")["status"] == "offline" or False, timeout=60, what="daemon death -> offline") is not None, "vanishing daemon drives the saddle offline (failed staleness probe)") check(wait_for(lambda: any("unreachable" in c["body"] and c.get("is_system") for c in task_state(auth, vanish).get("comments", [])) or False, timeout=30, what="unreachable comment") is not None, "the affected task gets an unreachable system row (never an agent comment)") check(task_state(auth, vanish)["status"] == "queued", "the task requeues after the daemon death") card = saddle_card(auth, "fake-saddle") check(card.get("health") == "offline", "health badge reports offline") # The unreachable row flag marks the detection window: a task the # web still sees working while its saddle is offline/fenced, before # the requeue pass moves it. Controlled window (the real # transitions above are atomic with their requeues). conn = db() conn.execute("insert into tasks (project_id, title, status, priority," " created_at) values (1, 'Unreachable row', 'working', 5," " strftime('%s','now'))") unreachable_row = conn.execute("select max(id) from tasks").fetchone()[0] conn.execute("insert into sessions (task_id, saddle_id, started_at)" " values (?, ?, strftime('%s','now'))", (unreachable_row, fake_id)) conn.commit() conn.close() check(wait_for(lambda: (lambda row: row.get("unreachable") is True)( [t for t in slot_public(auth, "/tasks", "tasks")["tasks"] if int(t["id"]) == unreachable_row][0]) or False, timeout=20, what="unreachable flag") is not None, "queue row shows the unreachable state in the window") print("== phase 1: saddles.token is the single credential store (arch 24) ==") # The legacy api_tokens table and every fallback read died with the # poll era: a restart proves the migrated schema carries no # api_tokens table, the row token survives and still authenticates, # and a legacy-shaped token string authorises nothing (no fallback # read exists to honour it). Fixture: this shard's own racer. fake2 = FakeSaddle() fake2.ask_on_register = False fake2.serve(port=free_port()) saddles_state = slot_public(auth, "/saddles", "saddles") act(auth, saddles_state["register"], {"name": "racer-two", "host": "127.0.0.1:%d" % fake2.port}) racer_two_id = int([s for s in slot_public(auth, "/saddles", "saddles")["saddles"] if s["name"] == "racer-two"][0]["id"]) racer_two_token = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"] if s["name"] == "racer-two"][0]["token"] app.terminate() app.wait(timeout=10) app = start_app() auth = login("admin", "adminpass123") ctx.app = app # teardown tracks the restarted process ctx.auth = auth def schema_tables(): conn = db() try: return {row[0] for row in conn.execute( "select name from sqlite_master where type = 'table'")} finally: conn.close() check("api_tokens" not in schema_tables(), "the fresh schema carries no api_tokens table (arch 24)") conn = db() row_token = conn.execute("select token from saddles where id = ?", (racer_two_id,)).fetchone()[0] conn.close() check(row_token == racer_two_token and (row_token or "").startswith("saddle-"), "the per-saddle row token survives the restart unchanged") status, _ = saddle_api("POST", "/api/saddle/next-task", racer_two_token) check(status in (200, 204), "the row token still authenticates after the restart") status, _ = saddle_api("POST", "/api/saddle/next-task", "saddle-legacy-copied-token") check(status == 401, "a legacy api_tokens-shaped credential authorises nothing (fallback gone)") fake2.token = racer_two_token fake2.ask_now() check(True, "the fleet re-asks after the restart (sleeping re-established)")