clanker-saddle/test/bundles/answers.py

"""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")
    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, 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] == "resumed" and rows[0][2]
          and rows[1][1] == "done" and rows[1][2],
          "parked session closed 'resumed'; 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] == "resumed"
          and len(session_rows) == 2 and session_rows[0] == (parked_session, "resumed"),
          "the parked session closed 'resumed' 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 or close its
    # session — the resume machinery (delivery once the daemon
    # returns, or the affinity allocation on its next ask) needs
    # both. The question park leaves the row idle; the review park
    # keeps it busy@task (`saddle review` blocks the round without
    # a parking state push), exercising the unreachable path's
    # requeue decision both ways.
    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()
    open_session = conn.execute(
        "select count(*) from sessions where task_id = ? and ended_at is null",
        (stranded_task,)).fetchone()[0]
    conn.close()
    check(open_session == 1,
          "the parked-with-answered task keeps its open session 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()
    open_session = conn.execute(
        "select count(*) from sessions where task_id = ? and ended_at is null",
        (review_task,)).fetchone()[0]
    conn.close()
    check(open_session == 1,
          "the review-parked task keeps its open session 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)")