clanker-saddle/test/bundles/allocation.py

"""The allocation 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: allocation — the ask/wake/allocate machinery: concurrent ask
# race, token revocation, workspace affinity, the lost-wake guard, the
# slow-ask regression, assignment capability context + the tag
# catalogue, and the pinned-blocks/activity dashboard.
# =============================================================================

def run(ctx):
    auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token
    app = ctx.app
    r4_token = fake_token
    # gpu rides the primary fake from the start (the routing bundle's
    # early attach, replayed): its attach row precedes lostwake's, so
    # the assignment's saddle_tags order matches the assertion in the
    # capability-context section below.
    saddles_state = slot_public(auth, "/saddles", "saddles")
    act(auth, saddles_state["create_capability"], {"name": "gpu"})
    act(auth, saddle_card(auth, "fake-saddle")["attach_capability"],
        {"capability_id": str(int([c["id"] for c in
                                   slot_public(auth, "/saddles", "saddles")["capabilities"]
                                   if c["name"] == "gpu"][0]))})

    print("== phase 1: ask/allocation — token auth + concurrent race ==")
    # Plan 10 phase 1. Registration mints per-saddle tokens; the
    # ask endpoint authenticates by them; concurrent asks race to
    # exactly one winner through the allocation transaction.
    status, _ = saddle_api("POST", "/api/saddle/next-task", None)
    check(status == 401, "next-task rejects a missing token")
    status, _ = saddle_api("POST", "/api/saddle/next-task", "saddle-bogus-token")
    check(status == 401, "next-task rejects a wrong token")
    status, body = saddle_api("POST", "/api/saddle/next-task", fake_token)
    check(status == 204, "next-task answers 204 (nothing offerable — daemon sleeps)")

    # Two more racers: real daemons on their own ports; fake2 reports
    # its ssh key at /register so the key lands on the row directly.
    fake2 = FakeSaddle()
    fake2.ssh_public_key = "ssh-ed25519 AAAAFake2RacerKey saddle-2@fedora"
    fake2.serve(port=free_port())
    fake3 = FakeSaddle()
    fake3.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})
    act(auth, saddles_state["register"], {"name": "racer-three",
                                          "host": "127.0.0.1:%d" % fake3.port})
    check(wait_for(lambda: len(fake2.register_log) > 0 and len(fake3.register_log) > 0
                   or False, timeout=15, what="racers registered") is not None,
          "both racers received /register")
    check(wait_for(lambda: "AAAAFake2RacerKey" in
                   [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
                    if s["name"] == "racer-two"][0].get("ssh_public_key", "") or False,
                   timeout=15, what="register-reported key") is not None,
          "the ssh key reported at /register lands on the saddle row")

    racer_cards = {}
    for name, daemon in (("fake-saddle", fake), ("racer-two", fake2), ("racer-three", fake3)):
        card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
                if s["name"] == name][0]
        racer_cards[daemon] = {"id": int(card["id"]), "token": card["token"]}
    # Revocation: a registered saddle's token works until the saddle
    # is removed, then nothing authorises its calls.
    fake4 = FakeSaddle()
    fake4.ask_on_register = False  # a pre-flip daemon: never asks
    fake4.serve(port=free_port())
    saddles_state = slot_public(auth, "/saddles", "saddles")
    act(auth, saddles_state["register"], {"name": "old-daemon",
                                          "host": "127.0.0.1:%d" % fake4.port})
    old_card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
                if s["name"] == "old-daemon"][0]
    status, _ = saddle_api("POST", "/api/saddle/next-task", old_card["token"])
    check(status in (200, 204),
          "the old daemon's fresh token authenticates next-task")
    status, _ = saddle_api("POST", "/api/saddle/note", old_card["token"],
                           {"project": "demo", "body": "from the old daemon"})
    check(status == 200, "note accepts any registered saddle's token")
    act(auth, old_card["unregister"], {})
    status, _ = saddle_api("POST", "/api/saddle/next-task", old_card["token"])
    check(status == 401, "removing the saddle revokes its token")
    status, _ = saddle_api("POST", "/api/saddle/note", old_card["token"],
                           {"project": "demo", "body": "should fail"})
    check(status == 401, "revocation covers the note endpoint too")

    # Deterministic race: settle the queue first (this shard runs the
    # race before any parked work exists, so only in-flight asks need
    # draining), freeze every daemon's self-ask triggers, put all three
    # to sleep (204), then fire one task worth of concurrent asks.
    for daemon in (fake, fake2, fake3):
        daemon.ask_on_register = False
        daemon.ask_on_wake = False
        daemon.ask_on_terminal = False
        daemon.ask_on_idle_status = False
        daemon.settle_asks()  # converted sleep: in-flight asks land (204)
    def queued_count():
        conn = db()
        try:
            return conn.execute(
                "select count(*) from tasks where status = 'queued'").fetchone()[0]
        finally:
            conn.close()

    check(wait_for(lambda: queued_count() == 0 or False,
                   timeout=60, what="queue settles before the race") is not None,
          "the queue settles before the race")

    sleep_statuses = [daemon.ask_now()[0] for daemon in (fake, fake2, fake3)]
    check(sleep_statuses == [204, 204, 204],
          "all three racers sleep (204) before the task exists")

    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Race task",
        "priority": "2",
    })
    race_task = task_id_tmp

    race_results = []
    def race_ask(daemon):
        # A real racing daemon accepts the assignment it wins — the
        # manual ask path does exactly that on a 200.
        race_results.append(daemon.ask_now())
    threads = [threading.Thread(target=race_ask, args=(daemon,))
               for daemon in (fake, fake2, fake3)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()
    race_statuses = sorted(r[0] for r in race_results)
    check(race_statuses == [200, 204, 204],
          "concurrent asks race to exactly one winner (%s)" % race_statuses)
    winner = [r for r in race_results if r[0] == 200][0][1]
    check(int(winner["task"]["id"]) == race_task, "the winner's payload carries the task")
    check(int(winner["session_id"]) > 0, "the assignment carries the session id")
    conn = db()
    race_sessions = conn.execute(
        "select count(*) from sessions where task_id = ?", (race_task,)).fetchone()[0]
    conn.close()
    check(race_sessions == 1,
          "exactly one session row — the transaction serialised the claims (%d)"
          % race_sessions)
    check(wait_task_done(auth, race_task, timeout=60) is not None,
          "the raced task completes on the winning daemon")
    for daemon in (fake, fake2, fake3):
        daemon.ask_on_register = True
        daemon.ask_on_wake = True
        daemon.ask_on_terminal = True
        daemon.ask_on_idle_status = True
    # Re-establish sleeping explicitly: the winner's post-race
    # re-ask timing is not deterministic, and the affinity section
    # below needs every racer provably asleep. Any stray offerable
    # drains into these asks first.
    def all_asleep():
        return all(daemon.ask_now()[0] == 204 for daemon in (fake, fake2, fake3))
    check(wait_for(lambda: all_asleep() or False, timeout=30,
                   what="racers re-sleep after the race") is not None,
          "all racers re-sleep after the race")

    print("== phase 1: workspace affinity pins parked tasks ==")
    # A suspended parent parked on racer-two may be resumed ONLY by
    # racer-two: a tag-capable non-affine saddle asking gets 204, and
    # the wake targets the affine saddle. Only racer-two carries the
    # tag at dispatch time (the race's winner is nondeterministic, so
    # the park must not depend on which racers are free); racer-three
    # gains it afterwards to prove the pinning is by workspace, not
    # tags.
    saddles_state = slot_public(auth, "/saddles", "saddles")
    act(auth, saddles_state["create_capability"], {"name": "affinity-tag"})
    affinity_cap = int([c["id"] for c in slot_public(auth, "/saddles", "saddles")["capabilities"]
                        if c["name"] == "affinity-tag"][0])
    racer2_card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
                   if s["name"] == "racer-two"][0]
    racer3_card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
                   if s["name"] == "racer-three"][0]
    act(auth, racer2_card["attach_capability"], {"capability_id": str(affinity_cap)})

    # Plan 12: the parked workspace is the goal agent's run task. The
    # other racers sleep (frozen ask triggers), so racer-two claims
    # the goal run and parks it; the goal-task event then resumes the
    # agent AFFINE to racer-two — racer-three (tag-capable or not)
    # gets 204, the dispatcher's wake targets racer-two.
    fake.ask_on_wake = False
    fake.ask_on_idle_status = False
    fake.settle_asks()
    fake3.ask_on_wake = False
    fake3.ask_on_idle_status = False
    fake3.settle_asks()

    act(auth, slot_public(auth, "/goals", "goals")["create"], {
        "project_id": "1", "title": "Affinity goal",
        "brief": "Parks on racer-two.", "branch_name": "goal/affinity"})
    conn = db()
    affinity_goal = conn.execute(
        "select id from goals where title = 'Affinity goal'").fetchone()[0]
    conn.close()
    fake2.goal_script = [("plan", "# Affinity plan\n\n1. Park")]
    act(auth, slot_public(auth, "/goals/%d" % affinity_goal, "goal")["start"], {})
    check(wait_for(lambda: slot_public(auth, "/goals/%d" % affinity_goal,
                                       "goal")["status"] == "plan_review" or False,
                   timeout=60, what="affinity goal plans") is not None,
          "the affinity goal reaches plan review (racer-two hosting)")
    fake2.goal_script = [("suspend",)]
    # Freeze racer-two's ask triggers across the approval too (the
    # later wake measurement already does): the mirror-sync fix makes
    # the run resume-offerable the moment its goal row carries the
    # verdict, so a wake-triggered background ask would race this
    # manual ask_now for the claim.
    # (Parking guard: the scripted park needs a live wake source.)
    conn = db()
    conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
                 " priority, mode, creator, created_at) values (1, ?, 'task',"
                 " 'Affinity live worker', 'blocked', 5, 'implement', 'goal',"
                 " strftime('%s','now'))", (affinity_goal,))
    conn.commit()
    conn.close()
    fake2.ask_on_wake = False
    fake2.ask_on_idle_status = False
    fake2.settle_asks()
    act(auth, slot_public(auth, "/goals/%d" % affinity_goal, "goal")
        ["plan_pending_block"]["answer_action"],
        {"answer": "Plan approved. Proceed with implementation."})
    conn = db()
    affinity_parent = conn.execute(
        "select id from tasks where goal_id = ? and kind = 'goal_run'",
        (affinity_goal,)).fetchone()[0]
    agent_session = conn.execute(
        "select max(id) from sessions where task_id = ?",
        (affinity_parent,)).fetchone()[0]
    conn.close()
    status, resume_payload = fake2.ask_now()
    check(status == 200 and int(resume_payload["task"]["id"]) == affinity_parent,
          "the approved plan resumes the agent on racer-two (its workspace)")
    check(wait_for(lambda: task_state(auth, affinity_parent).get("status") == "suspended"
                   or False, timeout=60, what="affinity agent parks") is not None,
          "the affinity goal agent parks on racer-two")
    conn = db()
    affinity_session = conn.execute(
        "select saddle_id from sessions where task_id = ?"
        " order by id desc limit 1",
        (affinity_parent,)).fetchone()[0]
    conn.close()
    check(affinity_session == int(racer2_card["id"]),
          "the parked workspace sits on racer-two")

    # Now racer-three carries the tag too — the pinning is by
    # workspace, not tags. With nothing due, a non-affine ask gets 204.
    act(auth, racer3_card["attach_capability"], {"capability_id": str(affinity_cap)})
    status, _ = fake3.ask_now()
    check(status == 204,
          "a tag-capable non-affine saddle is refused the parked agent (204)")

    # A goal task completing makes the parked agent resume-offerable —
    # affine to racer-two only. The other racers wake back up (they
    # host the worker; their claims of the parked agent get refused as
    # non-affine), while racer-two's ask triggers stay frozen so the
    # wake measurement stays clean. A wake may have hit the frozen
    # racers while the goal run queued (the lost-wake guard marks them
    # offline): the refresh probe brings them back before the worker.
    fake.ask_on_wake = True
    fake.ask_on_idle_status = True
    fake3.ask_on_wake = True
    fake3.ask_on_idle_status = True
    for racer_name in ("fake-saddle", "racer-three"):
        act(auth, saddle_card(auth, racer_name)["refresh"], {})
        check(wait_for(lambda name=racer_name: [s for s in
                                                slot_public(auth, "/saddles", "saddles")["saddles"]
                                                if s["name"] == name][0]["status"] == "idle"
                       or False, timeout=15, what="racer re-onlined") is not None,
              "%s re-onlined after its frozen-window wake" % racer_name)
    fake2.ask_on_wake = False
    fake2.ask_on_idle_status = False
    wakes_before_2 = len(fake2.wake_log)
    fake2.settle_asks()  # converted sleep: park-time ask lands (204)
    status, body = saddle_api("POST", "/api/saddle/task/create", fake2.token, {
        "session_id": agent_session, "title": "Affinity child",
        "mandate": "finish the affinity fixture"})
    check(status == 200, "the parked agent fans its worker out via the CLI")
    affinity_child = int(body["task_id"])
    saddle_api("POST", "/api/saddle/task/enqueue", fake2.token,
               {"session_id": agent_session, "task_id": affinity_child})
    check(wait_task_done(auth, affinity_child, timeout=90) is not None,
          "the affinity worker completes (on a non-affine racer)")

    check(wait_for(lambda: len(fake2.wake_log) > wakes_before_2 or False,
                   timeout=90, what="affine wake") is not None,
          "the dispatcher's wake targets the affine saddle")
    status, resume_payload = fake2.ask_now()
    fake2.ask_on_wake = True       # restored whatever the claim returned
    fake2.ask_on_idle_status = True
    check(status == 200 and int(resume_payload["task"]["id"]) == affinity_parent,
          "the affine saddle's ask claims the parked agent")
    check("Affinity child" in (resume_payload.get("resume_notice") or "")
          and "fake daemon finished" in (resume_payload.get("resume_notice") or ""),
          "the resume assignment carries the goal's batched report notice")
    check(wait_for(lambda: task_state(auth, affinity_parent).get("status")
                   == "suspended" or False,
                   timeout=60, what="affinity agent re-parks") is not None,
          "the resumed agent re-parks (script)")
    act(auth, slot_public(auth, "/goals/%d" % affinity_goal, "goal")["cancel"], {})
    check(wait_for(lambda: slot_public(auth, "/goals/%d" % affinity_goal,
                                       "goal")["status"] == "cancelled" or False,
                   timeout=30, what="affinity goal cancelled") is not None,
          "the affinity goal cancels (drain)")

    print("== phase 1: lost wake -> offline + re-wake of the next capable ==")
    # racer-three sleeps through wakes (a daemon that died after its
    # last 204): the guard window (SADDLE_LOST_WAKE_S=2) passes with
    # no ask, the saddle goes offline, and the work moves to the
    # next capable sleeping saddle.
    fake3.ask_on_wake = False
    fake3.ask_on_idle_status = False  # sleeps through everything
    saddles_state = slot_public(auth, "/saddles", "saddles")
    act(auth, saddles_state["create_capability"], {"name": "lostwake"})
    lostwake_cap = int([c["id"] for c in slot_public(auth, "/saddles", "saddles")["capabilities"]
                        if c["name"] == "lostwake"][0])
    racer3_card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
                   if s["name"] == "racer-three"][0]
    act(auth, racer3_card["attach_capability"], {"capability_id": str(lostwake_cap)})
    # Racer-three must be asleep to be woken: park it with a manual
    # 204 ask (nothing offerable for it — the task below is not
    # created yet).
    status, _ = fake3.ask_now()
    check(status == 204, "racer-three sleeps before the lost-wake run")

    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Lost wake task",
        "priority": "2",
        "capability_id": [str(lostwake_cap)],
    })
    lost_task = task_id_tmp
    check(wait_for(lambda: len(fake3.wake_log) > 0 or False,
                   timeout=30, what="wake sent to racer-three") is not None,
          "the only capable sleeping saddle is woken")
    check(wait_for(lambda: [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
                            if s["name"] == "racer-three"][0]["status"] == "offline"
                   or False, timeout=30, what="lost-wake -> offline") is not None,
          "no ask after the wake: the saddle goes offline (lost-wake guard)")
    check(task_state(auth, lost_task)["status"] == "queued",
          "the task is untouched while its wake was pending")

    # Re-wake the next capable: the healthy saddle gains the tag and
    # picks the work up via its own wake -> ask -> allocate roundtrip.
    fake_card = saddle_card(auth, "fake-saddle")
    act(auth, fake_card["attach_capability"], {"capability_id": str(lostwake_cap)})
    wakes_before = len(fake.wake_log)
    check(wait_for(lambda: len(fake.wake_log) > wakes_before or False,
                   timeout=30, what="re-wake of the next capable") is not None,
          "the next capable sleeping saddle is woken for the task")
    check(wait_task_done(auth, lost_task, timeout=60) is not None,
          "wake -> ask -> allocate: the task completes on the healthy saddle")
    check(wait_for(lambda: any(int(p.get("task", {}).get("id") or 0) == lost_task
                               for p in fake.start_payloads) or False,
                   timeout=10, what="lost task assignment captured") is not None,
          "the re-woken saddle's ask produced the assignment")
    fake3.ask_on_wake = True
    fake3.ask_on_idle_status = True

    print("== phase 3: capability context, catalogue, subtask tags ==")
    # Assignment capability context (plan 10 phase 3): every
    # assignment carries {saddle_tags: the allocated saddle's tags,
    # fleet_tags: every tag name known across the fleet} — the
    # daemon renders it into its system prompt. fake-saddle carries
    # gpu + lostwake; the catalogue is affinity-tag, gpu, lostwake.
    fake2.ask_on_register = False
    fake2.ask_on_wake = False
    fake2.ask_on_terminal = False
    fake2.ask_on_idle_status = False
    fake2.settle_asks()  # converted sleep: in-flight trigger asks land
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Capabilities context",
        "priority": "2",
    })
    cap_task = task_id_tmp
    check(wait_for(lambda: any(int(p.get("task", {}).get("id") or 0) == cap_task
                               for p in fake.start_payloads) or False,
                   timeout=60, what="capabilities-context task dispatched") is not None,
          "the capabilities-context task dispatches")
    cap_payload = [p for p in fake.start_payloads
                   if int(p.get("task", {}).get("id") or 0) == cap_task][-1]
    context = cap_payload.get("capabilities") or {}
    check(context.get("saddle_tags") == ["gpu", "lostwake"],
          "assignment saddle_tags = the allocated saddle's tags (%s)"
          % context.get("saddle_tags"))
    check(context.get("fleet_tags") == ["affinity-tag", "gpu", "lostwake"],
          "assignment fleet_tags = every tag name known across the fleet (%s)"
          % context.get("fleet_tags"))
    check(wait_task_done(auth, cap_task, timeout=60) is not None,
          "capabilities-context task completes")

    # GET /api/saddle/capabilities feeds the daemon's `saddle
    # capabilities` CLI: per-saddle-token auth, tag order by name,
    # each tag with the saddles carrying it (name order).
    status, _ = saddle_api("GET", "/api/saddle/capabilities", None)
    check(status == 401, "capabilities endpoint rejects a missing token")
    status, body = saddle_api("GET", "/api/saddle/capabilities", "saddle-bogus")
    check(status == 401, "capabilities endpoint rejects a wrong token")
    status, body = saddle_api("GET", "/api/saddle/capabilities", fake_token)
    catalogue = {t["name"]: t["saddles"] for t in body.get("tags", [])}
    check(status == 200 and body.get("ok") is True
          and [t["name"] for t in body.get("tags", [])]
          == ["affinity-tag", "gpu", "lostwake"],
          "catalogue lists every tag by name")
    check(catalogue.get("gpu") == ["fake-saddle"]
          and catalogue.get("affinity-tag") == ["racer-three", "racer-two"]
          and catalogue.get("lostwake") == ["fake-saddle", "racer-three"],
          "each tag names its carrying saddles in name order (%s)" % catalogue)

    # Task CLI capabilities (`task create --capability`): validated
    # against the operator tag set; unknown tags reject with the valid
    # list in the error (the agent reads it verbatim in the CLI
    # output). A goal agent's parked session hosts the calls.
    fake.suspend_on_title = "Tag host"
    tag_host = create_queued(auth, {
        "project_id": "1", "title": "Tag host", "priority": "3"})
    check(wait_for(lambda: task_state(auth, tag_host)["status"] == "suspended"
                   or False, timeout=60, what="tag host parks") is not None,
          "a parked plain task hosts nothing (no goal scope)")
    # The goal callbacks need a real goal agent: reuse the sequencing
    # goal pattern — a fresh parked agent under this shard.
    act(auth, slot_public(auth, "/goals", "goals")["create"], {
        "project_id": "1", "title": "Tag goal", "brief": "b",
        "branch_name": "goal/tags"})
    conn = db()
    tag_goal = conn.execute("select id from goals where title = 'Tag goal'").fetchone()[0]
    # (Parking guard: the parked agent hosts the CLI calls below — the
    # park needs a live wake source under the goal.)
    conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
                 " priority, mode, creator, created_at) values (1, ?, 'task',"
                 " 'Tag live worker', 'blocked', 5, 'implement', 'goal',"
                 " strftime('%s','now'))", (tag_goal,))
    conn.commit()
    conn.close()
    fake.goal_script = [("suspend",)]
    act(auth, slot_public(auth, "/goals/%d" % tag_goal, "goal")["start"], {})
    conn = db()
    tag_run = conn.execute(
        "select id from tasks where goal_id = ? and kind = 'goal_run'",
        (tag_goal,)).fetchone()[0]
    conn.close()
    check(wait_for(lambda: task_state(auth, tag_run)["status"] == "suspended"
                   or False, timeout=60, what="tag goal agent parks") is not None,
          "the tag goal agent parks")
    conn = db()
    tag_session = conn.execute(
        "select max(id) from sessions where task_id = ?",
        (tag_run,)).fetchone()[0]
    conn.close()
    conn = db()
    tag_host_saddle = conn.execute(
        "select saddle_id from sessions where id = ?", (tag_session,)).fetchone()[0]
    conn.close()
    racer2_card_id = int([c for c in slot_public(auth, "/saddles", "saddles")["saddles"]
                          if c["name"] == "racer-two"][0]["id"])
    racer3_card_id = int([c for c in slot_public(auth, "/saddles", "saddles")["saddles"]
                          if c["name"] == "racer-three"][0]["id"])
    host_daemon = {int(fake_id): fake, racer2_card_id: fake2,
                   racer3_card_id: fake3}[int(tag_host_saddle)]
    status, body = saddle_api("POST", "/api/saddle/task/create", host_daemon.token, {
        "session_id": tag_session, "title": "Bad tag worker",
        "mandate": "m", "capabilities": ["gpu", "warp-drive"]})
    check(status == 400, "an unknown capability tag rejects the task call")
    error_text = body.get("error", "")
    check("warp-drive" in error_text and "valid tags" in error_text
          and "gpu" in error_text and "lostwake" in error_text,
          "the rejection names the unknown tag and the valid list (%s)" % error_text)
    status, _ = saddle_api("POST", "/api/saddle/task/create", host_daemon.token, {
        "session_id": tag_session, "title": "Bad shape worker",
        "mandate": "m", "capabilities": "gpu"})
    check(status == 400, "a non-array capabilities member rejects")

    # A valid tag creates the goal worker in draft with exactly the
    # explicit tag attached; the goal agent enqueues it.
    status, body = saddle_api("POST", "/api/saddle/task/create", host_daemon.token, {
        "session_id": tag_session, "title": "Explicit tag worker",
        "mandate": "do the tagged work", "capabilities": ["gpu"]})
    check(status == 200 and body.get("ok") is True,
          "a valid tag creates the task synchronously")
    tagged_worker = int(body["task_id"])
    detail = task_state(auth, tagged_worker)
    check([c["name"] for c in detail.get("capabilities", [])] == ["gpu"],
          "the worker carries exactly the explicit tag")
    conn = db()
    tag_rows = conn.execute(
        "select c.name from task_capabilities tc"
        " join capabilities c on c.id = tc.capability_id"
        " where tc.task_id = ?", (tagged_worker,)).fetchall()
    conn.close()
    check([r[0] for r in tag_rows] == ["gpu"],
          "the task_capabilities rows record the explicit tag")
    check(detail["status"] == "draft",
          "the callback's task lands in draft until its agent enqueues")
    saddle_api("POST", "/api/saddle/task/enqueue", host_daemon.token,
               {"session_id": tag_session, "task_id": tagged_worker})
    check(wait_task_done(auth, tagged_worker, timeout=90) is not None,
          "the tagged worker dispatches and completes")
    act(auth, slot_public(auth, "/goals/%d" % tag_goal, "goal")["cancel"], {})
    act(auth, task_state(auth, tag_host)["cancel"], {})

    print("== phase 3: dashboard — pinned blocks + activity timeline ==")
    # Pinned OPEN blocking items as link-through cards, newest
    # first; no inline answer forms (the card links to the task
    # page). Fixture: one open question + one open review raised on a
    # local task (stand-ins for the auto-approve grandchild's
    # leftovers) join the probe RFC raised below.
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
                 " values (1, 'Dash fixture task', 'working', 5, strftime('%s','now'))")
    dash_task = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.execute("insert into sessions (task_id, saddle_id, started_at)"
                 " values (?, ?, strftime('%s','now'))", (dash_task, fake_id))
    dash_session = conn.execute("select max(id) from sessions").fetchone()[0]
    conn.commit()
    conn.close()
    def raise_on_dash(question, kind="question", body=None, summary=None):
        """Push-era raises on the dashboard fixture task (the poll-era
        /update channel is gone): a blocked results push carries the
        block payload; a review verb rides a review entry."""
        if kind == "review":
            block = {"question": question, "type": "review"}
            if summary is not None:
                block["body"] = summary
            return saddle_api("POST", "/api/saddle/results", fake_token, {
                "session_id": dash_session, "state": "blocked",
                "block": block})[0]
        block = {"question": question, "type": kind}
        if body is not None:
            block["body"] = body
        return saddle_api("POST", "/api/saddle/results", fake_token, {
            "session_id": dash_session, "state": "blocked",
            "block": block})[0]

    raise_on_dash("Dash fixture question?", kind="question")
    raise_on_dash("dash fixture review", kind="review", summary="dash fixture review")

    # The activity timeline: last <= 50 events, newest first, fed
    # from the existing tables. Fixture rows a few seconds old
    # guarantee the progress kind and a human-authored comment ride
    # along; the probe block below proves same-source ordering.
    conn = db()
    conn.execute("insert into task_comments (task_id, author, body, created_at)"
                 " values (?, 'user:admin', 'timeline fixture comment by a human',"
                 " strftime('%s','now','-5 seconds'))", (cap_task,))
    conn.execute(
        "insert into events (kind, actor, scope_type, scope_id, payload,"
        " created_at) values ('comment', 'user', 'task', ?,"
        " ?, strftime('%s','now','-5 seconds'))",
        (cap_task, json.dumps({"comment_id": 0, "author": "user:admin",
                               "body": "timeline fixture comment by a human"})))
    conn.execute(
        "insert into events (kind, actor, scope_type, scope_id, payload,"
        " created_at) values ('progress', 'model', 'task', ?,"
        " ?, strftime('%s','now','-5 seconds'))",
        (cap_task, json.dumps({"text": "timeline fixture progress"})))
    conn.commit()
    conn.close()
    # The probe RFC raised last: the newest pinned card and the newest
    # timeline event (its raise+answer below prove same-source ordering
    # both ways). The timeline sorts same-second events by source
    # materialisation order, so the probe must land in a LATER clock
    # second than the fixtures — wait for the boundary (bounded).
    fixtures_epoch = int(time.time())
    wait_for(lambda: int(time.time()) > fixtures_epoch or False,
             timeout=3, what="next clock second before the probe raise")
    raise_on_dash("Timeline probe: approve the dashboard flow",
                  kind="rfc", body="# Probe")
    dash = slot_public(auth, "/", "dashboard")
    pinned = dash.get("pinned", [])
    pinned_ids = [int(b["id"]) for b in pinned]
    check(len(pinned_ids) >= 3 and pinned_ids == sorted(pinned_ids, reverse=True),
          "pinned open blocks order newest-first (%s)" % pinned_ids)
    check({b["block_type"] for b in pinned} >= {"question", "rfc", "review"},
          "pinned cards cover questions, RFCs and reviews")
    check(all(b["question"] and b["task_title"] and int(b["task_id"]) > 0
              for b in pinned)
          and all("answer_action" not in b for b in pinned),
          "pinned cards are link-through only (task title, kind badge, question)")
    activity = dash.get("activity", [])
    epochs = [int(e["at_epoch"]) for e in activity]
    check(0 < len(activity) <= 50 and epochs == sorted(epochs, reverse=True),
          "the timeline shows the last <=50 events newest-first (%d events)"
          % len(activity))
    kinds = {e["kind"] for e in activity}
    check({"created", "started", "dispatch", "comment", "progress", "completed"} <= kinds,
          "the timeline covers status changes, dispatches, comments, progress and"
          " completions (%s)" % sorted(kinds))
    check(any(e["kind"] == "comment" and e["text"].startswith("user:admin:")
              and "fixture comment" in e["text"] for e in activity),
          "comment events carry author + snippet")

    # The freshly raised probe pinned as the newest card and landed
    # atop the timeline as block-raised; answering it records
    # block-answered.
    check(dash["pinned"][0]["question"] == "Timeline probe: approve the dashboard flow"
          and int(dash["pinned"][0]["task_id"]) == dash_task,
          "the freshest open block pins as the first card")
    check(dash["activity"][0]["kind"] == "block-raised"
          and "Timeline probe" in dash["activity"][0]["text"],
          "the raised block lands atop the timeline")
    probe_block = [b for b in task_state(auth, dash_task)["blocks"]
                   if b["question"] == "Timeline probe: approve the dashboard flow"][0]
    act(auth, probe_block["answer_action"], {"answer": "Ship the dashboard"})
    dash = slot_public(auth, "/", "dashboard")
    activity = dash.get("activity", [])
    raised_at = next((i for i, e in enumerate(activity)
                      if e["kind"] == "block-raised" and "Timeline probe" in e["text"]), -1)
    answered_at = next((i for i, e in enumerate(activity)
                        if e["kind"] == "block-answered"
                        and "Ship the dashboard" in e["text"]), -1)
    check(answered_at >= 0 and raised_at >= 0 and answered_at < raised_at,
          "the answered event sorts above its raised event (newest-first, same source)")
    check(all(b["question"] != "Timeline probe: approve the dashboard flow"
              for b in dash["pinned"]),
          "the answered block leaves the pinned cards")

    print("== phase 3 regression: a slow ask after a wake is never dropped ==")
    # The daemon worker's lost-wake-guard regression: a saddle whose
    # ask arrives SLOWLY after the wake (busy winding down when the
    # wake lands) must not lose the wake — the delayed ask (inside
    # the guard window) claims the task, no re-wake storm fires
    # while it is pending, and the saddle stays healthy. The
    # never-asks path (guard -> offline + re-wake of the next
    # capable) is proven by the racer-three section above.
    check(wait_for(lambda: queued_count() == 0 or False,
                   timeout=60, what="queue settles before the slow-wake run") is not None,
          "the queue settles before the slow-wake run")
    status, _ = fake2.ask_now()
    check(status == 204, "the slow-ask racer sleeps before its wake")
    saddles_state = slot_public(auth, "/saddles", "saddles")
    act(auth, saddles_state["create_capability"], {"name": "slowwake"})
    slowwake_cap = int([c["id"] for c in slot_public(auth, "/saddles", "saddles")
                        ["capabilities"] if c["name"] == "slowwake"][0])
    racer2_card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
                   if s["name"] == "racer-two"][0]
    act(auth, racer2_card["attach_capability"], {"capability_id": str(slowwake_cap)})
    fake2.ask_delay = 1.2   # slow: asks well after the wake, inside the 2s guard
    fake2.ask_on_wake = True
    wakes_before_slow = len(fake2.wake_log)
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Slow wake task",
        "priority": "2",
        "capability_id": [str(slowwake_cap)],
    })
    slow_task = task_id_tmp
    check(wait_for(lambda: len(fake2.wake_log) > wakes_before_slow or False,
                   timeout=30, what="the slow-ask racer woken") is not None,
          "the only capable sleeping saddle is woken")
    check(wait_task_done(auth, slow_task, timeout=120) is not None,
          "the delayed ask claims the task — the wake was not dropped")
    check(any(a.get("status") == 200
              and int((a.get("payload") or {}).get("task", {}).get("id") or 0)
              == slow_task for a in fake2.asks),
          "the slow ask's 200 carries the task")
    check(len(fake2.wake_log) == wakes_before_slow + 1,
          "the pending-wake skip re-wakes nobody while the slow ask is in flight")
    racer2_card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
                   if s["name"] == "racer-two"][0]
    check(racer2_card["status"] in ("idle", "busy"),
          "the slow asker never went offline (guard did not fire)")
    fake2.ask_delay = 0.0
    fake2.ask_on_wake = False  # stays out of the pool below

    print("== goal 4 t1: plain-task force wake — offerable on its affine saddle ==")
    # A suspended PLAIN task stays parked until the human force-wakes
    # it; the resuming claim carries the explicit wake notice and
    # consumes the flag (one wake per press).
    fake2.ask_on_wake = False
    fake2.ask_on_idle_status = False
    fake2.settle_asks()
    conn = db()
    conn.execute("insert into tasks (project_id, kind, title, status, priority, mode,"
                 " creator, created_at, suspended_at) values (1, 'task',"
                 " 'Park me', 'suspended', 5, 'implement', 'human',"
                 " strftime('%s','now'), strftime('%s','now'))")
    plain_parked = conn.execute("select max(id) from tasks").fetchone()[0]
    # parked_offerable keys the affine saddle off the newest SESSION
    # row (plan 16 D4): a parked round's row is closed 'parked' — give
    # the fixture one so the wake offers on the fake saddle.
    conn.execute("insert into sessions (task_id, saddle_id, started_at, ended_at,"
                 " outcome) values (?, ?, strftime('%s','now'),"
                 " strftime('%s','now'), 'parked')",
                 (plain_parked, racer2_card_id))
    conn.commit()
    conn.close()
    task_page = http_get_page(auth, "/tasks/%d" % plain_parked)
    check('stm-action="task.force_wake"' in task_page
          and 'stm-if="task.force_wake != null"' in task_page,
          "the task page authors the Force wake button for a suspended plain task")
    # No press, no wake: frozen-ask racer-two gets 204.
    status, _ = fake2.ask_now()
    check(status == 204,
          "without a press the parked plain task stays parked (204)")
    detail = task_state(auth, plain_parked)
    check(detail.get("force_wake") is not None,
          "the task state exposes the offered force-wake action")
    # Determinism: the wake's trigger-ask (and any idle-probe ask) must
    # not race the MANUAL claim the section asserts on.
    fake2.ask_on_wake = False
    fake2.ask_on_idle_status = False
    act(auth, detail["force_wake"], {})
    status, resume_payload = fake2.ask_now()
    check(status == 200 and int(resume_payload["task"]["id"]) == plain_parked,
          "the force-woke plain task is claimable again")
    woke_notice = resume_payload.get("resume_notice") or ""
    check("force-woke this parked task" in woke_notice
          and "resume where you left off" in woke_notice,
          "the plain-task resume carries the explicit wake notice")
    conn = db()
    flag_row = conn.execute("select force_wake from tasks where id = ?",
                            (plain_parked,)).fetchone()
    conn.close()
    check(flag_row is not None and flag_row[0] in (0, False),
          "the claim consumed the task force-wake flag")
    fake2.ask_on_wake = True
    fake2.ask_on_idle_status = True
    # The woken run re-parks (fake daemon finishes then suspends? no —
    # fake.start_payloads script: the plain task runs to done via the
    # daemon's own finish path in later suites; here just let it be
    # requeued by completing it directly to keep the fixture short).
    check(wait_task_done(auth, plain_parked, timeout=60) is not None,
          "the woken plain task completes (daemon finish path)")
    # Guard: a non-suspended task offers no wake action (the template
    # ships the gated markup statically; absence is a STATE property).
    check(task_state(auth, plain_parked).get("force_wake") is None,
          "a done task offers no force-wake action")