"""The answerwake bundle (goal 28 task F): the park/answer-wake race. The task-187 arc, replayed end-to-end against the real allocation: - an agent's ask_question parks the session (the blocked push); - the FREED daemon asks again IMMEDIATELY — the very race that used to hand the same task back as a bare crash-restore — and the ask must 204 while the block is unanswered (the allocation guard); - a CRASHED parked task resumes with its question still open: the requeued task re-allocates and the assignment flags `pending_question` so the restored conversation's opening turn says the question is pending (F6) instead of barreling on; - a human answers while that session RUNS: the verdict is injected into the running round through the daemon's task-update channel (kind "answer"), delivery-once (F2) — no affinity resume can deliver it, the busy saddle cannot ask; - the running round parks on a further question; answering it resumes through the allocation and the resume notice carries the FRESH verdict while the already-injected one is never repeated (F3's batched composition with the delivery ledger). """ from harness import * # noqa: F401,F403 — the shared wire helpers # The web-side park-in-flight hold (TaskService.PARK_SETTLE_GRACE_S is # 5s) plus a slack for the dispatcher tick — the crash-recovery sweep # must not fire until the crashed round's last push has aged past it. PARK_SETTLE_WAIT = 7.0 def run(ctx): auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token print("== goal 28 F1: the park race — an immediate re-ask is held ==") # A private capability tag keeps the shard's other sleeping # saddles out of these tasks. saddles_state = slot_public(auth, "/saddles", "saddles") act(auth, saddles_state["create_capability"], {"name": "answerwake"}) cap_id = int([c["id"] for c in slot_public(auth, "/saddles", "saddles")["capabilities"] if c["name"] == "answerwake"][0]) def race_daemon(name, **kwargs): daemon = FakeSaddle() daemon.ask_on_register = False # manual, deterministic asks daemon.hold_state = True # the test drives the lifecycle for key, value in kwargs.items(): setattr(daemon, key, value) daemon.serve() 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(cap_id)}) return daemon def race_task(title): return create_queued(auth, { "project_id": "1", "title": title, "priority": "2", "capability_id": [str(cap_id)]}) runner = race_daemon("race-runner") task_id = race_task("Race: parked then re-asked") ask_status, assignment = runner.ask_now() check(ask_status == 200 and int(assignment["task"]["id"]) == task_id, "the ask claims the queued task") first_session = assignment.get("session_id") # The park: the round blocks on a question. The web raises the # block row and closes the session 'parked' synchronously inside # the /results push. runner.state = "blocked" check(runner.push_results("blocked", block={ "question": "First: which side?", "type": "question", "options": ["Port", "Starboard"]}) == 200, "the blocked push parks the session") check(wait_for(lambda: task_state(auth, task_id).get("status") == "blocked" or False, timeout=30, what="raced task parked") is not None, "the task parks as blocked") conn = db() row = conn.execute( "select outcome from sessions where id = ?", (first_session,)).fetchone() conn.close() check(row is not None and row[0] == "parked", "the park closed the round's session row 'parked'") # THE RACE: the freed daemon asks again IMMEDIATELY — the block is # unanswered, so the allocation guard must hold the task (204), # never hand it back. The pre-fix web re-allocated it here and the # resumed round woke as a crash restore with its question gone. ask_status, payload = runner.ask_now() check(ask_status == 204, "F1: the immediate re-ask returns 204 while the block is unanswered") check(len([p for p in runner.start_payloads if int(p.get("task", {}).get("id") or 0) == task_id]) == 1, "F1: the parked task was never re-allocated") # The guard holds for the pre-block race too: a task whose row # still reads working/assigned while its round's session just # closed 'parked' (the transition in flight) is not claimable. conn = db() conn.execute("update tasks set status = 'working' where id = ?", (task_id,)) conn.commit() conn.close() ask_status, _ = runner.ask_now() check(ask_status == 204, "F1: a working row with a fresh parked session is held (transition in flight)") conn = db() conn.execute("update tasks set status = 'blocked' where id = ?", (task_id,)) conn.commit() conn.close() print("== goal 28 F6: a crashed parked task resumes with its question open ==") # The LEGIT unanswered-block resume (mandate F6): the parked task's # daemon crashed and the crash-recovery sweep requeued it — the # block row stays unanswered while the task re-enters the queue. # The sweep's post-state is applied directly (it needs a stale # OPEN session; this round's row is closed 'parked'). second_task = race_task("Race: crashed parked task") ask_status, assignment = runner.ask_now() check(ask_status == 200 and int(assignment["task"]["id"]) == second_task, "the ask claims the second task") second_first_session = assignment.get("session_id") runner.state = "blocked" check(runner.push_results("blocked", block={ "question": "Which side first?", "type": "question", "options": ["Port", "Starboard"]}) == 200, "the second task parks on a question") check(wait_for(lambda: task_state(auth, second_task).get("status") == "blocked" or False, timeout=30, what="second task parked") is not None, "the second task parks as blocked") # Answering is the resume: the wake -> ask -> allocation carries # the verdict on a fresh session row. detail = task_state(auth, second_task) side_block = [b for b in detail["blocks"] if b["question"] == "Which side first?"][0] act(auth, side_block["answer_action"], {"option": "Port"}) check(wait_for(lambda: any( int(p.get("task", {}).get("id") or 0) == second_task and p.get("session_id") != second_first_session and p.get("resume_notice") == "Human selected: Port" for p in runner.start_payloads[1:]) or False, timeout=30, what="second task resume allocation") is not None, "the answer resumes the parked task through the allocation") resumed = [p for p in runner.start_payloads if int(p.get("task", {}).get("id") or 0) == second_task][-1] running_session = resumed.get("session_id") check(wait_for(lambda: task_state(auth, second_task).get("status") == "working" or False, timeout=15, what="second round working") is not None, "the resumed round is running") print("== goal 28 F2: an answer while a session runs is injected ==") # The running round parks on a SECOND question, then the round is # lost to a crash — the task-187 shape: the newest block is # UNANSWERED while a fresh round is re-allocated (the crash # sweep's post-state, applied directly for determinism). runner.state = "blocked" check(runner.push_results("blocked", block={ "question": "Which database?", "type": "question", "options": ["SQLite", "Postgres"]}) == 200, "the running round parks on a second question") check(wait_for(lambda: task_state(auth, second_task).get("status") == "blocked" or False, timeout=30, what="second question raised") is not None, "the second question parks the round") # The crash: the round's session row is left OPEN and the task # row reads working (the daemon died between its park push and # its unwind). The next ask's crash-recovery sweep requeues it — # but NOT while the round's last push is fresh (the park-in-flight # hold); after the settle the sweep fires and the requeued task # re-allocates through the QUEUE with its question still open. conn = db() # Aged past the sweep's 120s grace (the round ran for a while # before it crashed) yet its LAST PUSH is the park transition, # seconds old — the exact task-187 shape. conn.execute( "update sessions set ended_at = null, outcome = null," " started_at = strftime('%s','now','-10 minutes') where id = ?", (running_session,)) conn.execute("update tasks set status = 'working' where id = ?", (second_task,)) conn.commit() conn.close() # Park-in-flight hold: an immediate ask requeues nothing (the # round pushed moments ago — its transition may be queued). ask_status, _ = runner.ask_now() check(ask_status == 204, "F1: the crash-recovery sweep holds a round whose last push is" " fresh (park in flight)") conn = db() still_open = conn.execute( "select count(*) from sessions where id = ? and ended_at is null", (running_session,)).fetchone()[0] task_row = conn.execute( "select status from tasks where id = ?", (second_task,)).fetchone()[0] conn.close() check(still_open == 1 and task_row == "working", "F1: nothing was requeued by the early ask") # After the settle window the sweep fires on the very next ask: # requeue (the crashed row closes 'stopped') and the freed task is # claimed straight back by the same ask — one round-trip. time.sleep(PARK_SETTLE_WAIT) ask_status, assignment = runner.ask_now() check(ask_status == 200 and assignment.get("session_id") != running_session, "the settled sweep requeues the crashed round and the ask claims it" " (fresh session row)") # The answer below must find the round genuinely RUNNING: wait for # the daemon to key on the new session and push its opening batch # (the busy-owner state the delivery path keys on). live_session = assignment.get("session_id") check(wait_for(lambda: runner.session_id == live_session or False, timeout=15, what="daemon keyed on the live session") is not None, "the daemon keyed its pushes on the live session") runner.push_results("working") check(wait_for(lambda: task_state(auth, second_task).get("status") == "working" or False, timeout=15, what="live round working") is not None, "the re-allocated round is running") check(assignment.get("pending_question") is True, "F6: the assignment flags its unanswered newest block (pending_question)") # The human answers the OPEN question while this round runs: no # affinity resume can deliver it (the saddle is busy), so the # verdict must ride the task-update injection channel. The # dispatcher drains the hub every tick (TICK_S=1 in the suite). # (The barrier above guarantees the open session exists BEFORE the # answer fires — an answer against a closed task takes the resume # path instead, which is F3's business.) detail = task_state(auth, second_task) db_block = [b for b in detail["blocks"] if b["question"] == "Which database?"][0] act(auth, db_block["answer_action"], {"option": "SQLite"}) check(wait_for(lambda: any(e.get("injected") and e.get("text") == "Human selected: SQLite" for e in runner.answer_log) or False, timeout=30, what="answer injected into the running session") is not None, "F2: the answer reaches the RUNNING session via task-update kind answer") conn = db() check(wait_for(lambda: conn.execute( "select count(*) from task_comments where task_id = ? and body = ?", (second_task, "verdict delivered into the running session (block #%s)" % db_block["id"]) ).fetchone()[0] > 0 or False, timeout=15, what="delivery system comment") is not None, "F2: the delivery is recorded on the task timeline") conn.close() # Delivery is exactly-once: no second injection may land. check(quiet_window(quiet_seconds(), lambda: len([e for e in runner.answer_log if e.get("injected")]) == 1, "no duplicate answer injection"), "F2: the answer is injected exactly once") conn = db() injected = conn.execute( "select answers_injected from sessions where id = ?", (live_session,)).fetchone() conn.close() check(injected is not None and injected[0] >= 1, "F2: the injection is counted on the session row (answers_injected)") print("== goal 28 F3: the resume notice composes the batch ==") # The live round parks on a THIRD question; answering it resumes # through the allocation. The notice carries the FRESH verdict and # never repeats the already-injected one (the delivery ledger is # the filter) — a newer block can no longer swallow an older # answer silently. runner.state = "blocked" check(runner.push_results("blocked", block={ "question": "Final check?", "type": "question", "options": ["Ship", "Hold"]}) == 200, "the live round parks on a third question") check(wait_for(lambda: task_state(auth, second_task).get("status") == "blocked" or False, timeout=30, what="third park") is not None, "the task parks on its third question") ask_status, _ = runner.ask_now() check(ask_status == 204, "the third park also holds the immediate re-ask (unanswered block)") detail = task_state(auth, second_task) final_block = [b for b in detail["blocks"] if b["question"] == "Final check?"][0] act(auth, final_block["answer_action"], {"option": "Ship"}) check(wait_for(lambda: any( int(p.get("task", {}).get("id") or 0) == second_task and p.get("session_id") != live_session and "Human selected: Ship" in (p.get("resume_notice") or "") for p in runner.start_payloads[1:]) or False, timeout=30, what="third resume allocation") is not None, "F3: the answer resumes the parked task with its verdict") final = [p for p in runner.start_payloads if int(p.get("task", {}).get("id") or 0) == second_task][-1] notice = final.get("resume_notice") or "" check("Human selected: Ship" in notice, "F3: the fresh verdict rides the resume notice") check("Human selected: SQLite" not in notice, "F3: the already-injected verdict is NOT repeated as a resume verdict") check(final.get("pending_question") is not True, "the resumed assignment has no open question left") # The resume_notice persists on the session row (the terminal # page's resume-info panel reads it) and the watermark was stored. conn = db() stored = conn.execute( "select resume_notice from sessions where id = ?", (final.get("session_id"),)).fetchone() watermark = conn.execute( "select resume_since is not null from sessions where id = ?", (final.get("session_id"),)).fetchone() conn.close() check(stored is not None and "Human selected: Ship" in (stored[0] or ""), "the resume notice persists on the resumed session row") check(watermark is not None and watermark[0] == 1, "the batched-verdict watermark persists (resume_since)") # The final round completes cleanly. runner.push_results("done", summary="race rounds complete") check(wait_task_done(auth, second_task, timeout=30) is not None, "the resumed round continues to completion") runner.kill()