clanker-saddle/test/bundles/answerwake.py

"""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()