"""The dispatch 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: dispatch — the task pipeline end to end: registration tokens,
# rules, project notes, pushed chunks/results ingestion, capability
# routing, the create form, and retry/manual resume.
# =============================================================================

def run(ctx):
    auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token
    app = ctx.app

    print("== registration: token minting and delivery ==")
    registered = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
                  if s["name"] == "fake-saddle"]
    check(len(registered) == 1 and registered[0]["token"].startswith("saddle-"),
          "fake saddle registered with an API token")

    # Plan 10 phase 1: registration delivers the per-saddle token to
    # the daemon via POST /register {saddle_token, web_base_url}.
    check(wait_for(lambda: len(fake.register_log) > 0 or False,
                   timeout=15, what="/register delivery") is not None,
          "registration POSTs /register to the daemon")
    register_body = fake.register_log[0]
    check(register_body.get("saddle_token") == fake_token,
          "/register carries the minted per-saddle token")
    check(register_body.get("web_base_url") == WEB,
          "/register carries the web base URL")
    check(fake.token == fake_token,
          "daemon persisted the token (its asks use it)")
    token_conn = db()
    row_token = token_conn.execute("select token from saddles where id = ?",
                                   (fake_id,)).fetchone()[0]
    token_conn.close()
    check(row_token == fake_token,
          "per-saddle token lives on the saddle row")
    # The register-time ask (idle-awake daemon): 204, nothing
    # offerable yet — the saddle is asleep and wakeable.
    check(wait_for(lambda: any(a["status"] == 204 for a in fake.asks) or False,
                   timeout=15, what="register-time ask") is not None,
          "the daemon's idle ask returns 204 (nothing offerable)")

    print("== feature 5: rules ==")
    rules_state = slot_public(auth, "/rules", "rules")
    for name, body in [
        ("Use Invercargill", "Never use libgee; use Invercargill enumerables."),
        ("Write Vala docs", "Every public class carries a doc comment."),
        ("Prefer meson", "Builds go through meson; no makefiles."),
        ("Co-dev rule", "Keep the wire contract exactly."),
        ("Reference rule", "Must never ride along."),
    ]:
        act(auth, rules_state["create"], {"name": name, "body": body})
    listed = {r["name"]: r for r in slot_public(auth, "/rules", "rules")["rules"]}
    check(len(listed) == 5, "five rules created and listed on /rules")

    def rule_id(name):
        rows = [r for r in slot_public(auth, "/rules", "rules")["rules"] if r["name"] == name]
        return int(rows[0]["id"])

    state = slot_public(auth, "/projects", "projects")
    act(auth, state["create"], {"name": "co", "git_url": "https://example.com/co.git"})
    act(auth, state["create"], {"name": "ref", "git_url": "https://example.com/ref.git"})
    co_id = int([p["id"] for p in slot_public(auth, "/projects", "projects")["projects"]
                 if p["name"] == "co"][0])
    ref_id = int([p["id"] for p in slot_public(auth, "/projects", "projects")["projects"]
                  if p["name"] == "ref"][0])

    project = slot_public(auth, "/projects/1", "project")
    act(auth, project["add_link"], {"linked_project_id": str(co_id),
                                    "relationship": "co-developed"})
    act(auth, project["add_link"], {"linked_project_id": str(ref_id),
                                    "relationship": "reference"})
    act(auth, project["attach_rule"], {"rule_id": str(rule_id("Use Invercargill"))})
    act(auth, project["attach_rule"], {"rule_id": str(rule_id("Prefer meson"))})
    project = slot_public(auth, "/projects/1", "project")
    check(sorted(r["name"] for r in project["rules"]) == ["Prefer meson", "Use Invercargill"],
          "project page lists its attached rules")

    co_project = slot_public(auth, "/projects/%d" % co_id, "project")
    act(auth, co_project["attach_rule"], {"rule_id": str(rule_id("Co-dev rule"))})
    ref_project = slot_public(auth, "/projects/%d" % ref_id, "project")
    act(auth, ref_project["attach_rule"], {"rule_id": str(rule_id("Reference rule"))})

    tasks_state = slot_public(auth, "/tasks", "tasks")
    # Draft-first (deterministic rule attachment): the dispatcher claims
    # a queued task on its 1s tick, which under load could beat the
    # attach_rule calls below and ship an assignment without the task's
    # own rules — attach while draft, enqueue last.
    act(auth, tasks_state["create"], {
        "project_id": "1",
        "title": "Rules dispatch",
        "description": "Effective rule set rides the /start payload.",
        "priority": "2",
    })
    rule_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                     if t["title"] == "Rules dispatch"][0])
    detail = task_state(auth, rule_task)
    act(auth, detail["attach_rule"], {"rule_id": str(rule_id("Write Vala docs"))})
    act(auth, detail["attach_rule"], {"rule_id": str(rule_id("Use Invercargill"))})
    enqueue_task(auth, rule_task)
    detail = task_state(auth, rule_task)
    check(sorted(r["name"] for r in detail["rules"]) == ["Use Invercargill", "Write Vala docs"],
          "task page lists its attached rules")

    check(wait_for(lambda: len(fake.start_payloads) > 0 or False,
                   timeout=30, what="/start push") is not None,
          "dispatcher pushes the task to the fake saddle")
    payload = fake.start_payloads[-1]
    rule_names = [r["name"] for r in payload.get("rules", [])]
    check(rule_names == ["Write Vala docs", "Use Invercargill", "Prefer meson", "Co-dev rule"],
          "effective rules in /start: task, primary, co-developed order (%s)" % rule_names)
    check(all(r.get("body") for r in payload.get("rules", [])),
          "rules carry their bodies")
    check("Reference rule" not in rule_names,
          "reference project rules excluded")

    detail = task_state(auth, rule_task)
    act(auth, detail["rules"][0]["detach"], {})
    check([r["name"] for r in task_state(auth, rule_task)["rules"]] == ["Use Invercargill"],
          "detached task rule disappears")

    rules_rows = slot_public(auth, "/rules", "rules")["rules"]
    docs_row = [r for r in rules_rows if r["name"] == "Write Vala docs"][0]
    act(auth, docs_row["update"], {"name": "Write Vala docs", "body": "Doc comments or bust."})
    check(any(r["body"] == "Doc comments or bust."
              for r in slot_public(auth, "/rules", "rules")["rules"]),
          "rule edit saves the body")

    check(wait_task_done(auth, rule_task, timeout=60) is not None,
          "dispatched task completes against the fake daemon")

    print("== feature 6: project notes + callback ==")
    # Plan 10 phase 1: the shared saddle-callback token is gone —
    # the per-saddle token authorises the callback endpoints.
    callback_token = fake_token
    check(callback_token.startswith("saddle-") and "saddle-callback-" not in callback_token,
          "note callback authorised by the per-saddle token")

    status, _ = saddle_api("POST", "/api/saddle/note", None,
                           {"project": "demo", "body": "x"})
    check(status == 401, "note endpoint rejects a missing token")
    status, _ = saddle_api("POST", "/api/saddle/note", "saddle-callback-not-real",
                           {"project": "demo", "body": "x"})
    check(status == 401, "note endpoint rejects a wrong (legacy-style) token")

    status, body = saddle_api("POST", "/api/saddle/note", callback_token,
                              {"task_id": rule_task, "project": "demo",
                               "body": "style.css tokens live in :root"})
    check(status == 200 and body.get("ok") is True, "note accepted with the callback token")
    status, _ = saddle_api("POST", "/api/saddle/note", callback_token,
                           {"project": "no-such-project", "body": "x"})
    check(status == 400, "unknown project rejected")
    status, _ = saddle_api("POST", "/api/saddle/note", callback_token,
                           {"project": "demo", "body": "   "})
    check(status == 400, "empty note body rejected")

    project = slot_public(auth, "/projects/1", "project")
    check(any("style.css tokens" in n["body"] and int(n["source_task_id"]) == rule_task
              for n in project["notes"]),
          "project page lists the note with its source task")
    act(auth, [n for n in project["notes"] if "style.css tokens" in n["body"]][0]["delete"], {})
    check(all("style.css tokens" not in n["body"]
              for n in slot_public(auth, "/projects/1", "project")["notes"]),
          "note deleted from the project page")

    # /start payload: notes of primary + co-developed (not reference),
    # oldest first; callback carries the configured base URL + token.
    saddle_api("POST", "/api/saddle/note", callback_token,
               {"project": "demo", "body": "note-one"})
    saddle_api("POST", "/api/saddle/note", callback_token,
               {"project": "demo", "body": "note-two"})
    saddle_api("POST", "/api/saddle/note", callback_token,
               {"project": "co", "body": "co-note"})
    conn = db()
    conn.execute("insert into project_notes (project_id, body, created_at)"
                 " values (?, 'ref-note-must-not-ship', strftime('%s','now'))", (ref_id,))
    conn.commit()
    conn.close()

    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Notes dispatch",
        "description": "Project notes ride the /start payload.",
        "priority": "2",
    })
    check(wait_for(lambda: len(fake.start_payloads) >= 2 or False,
                   timeout=30, what="second /start push") is not None,
          "dispatcher pushes the notes task")
    payload = fake.start_payloads[-1]
    notes = payload.get("project_notes", [])
    check([n["body"] for n in notes] == ["note-one", "note-two", "co-note"],
          "project_notes: primary + co-developed, oldest first (%s)" % [n["body"] for n in notes])
    check(all(n.get("project") for n in notes)
          and notes[2]["project"] == "co",
          "project_notes carry the project name")
    check("ref-note-must-not-ship" not in [n["body"] for n in notes],
          "reference project notes excluded")
    callback = payload.get("callback", {})
    check(callback.get("base_url") == WEB and callback.get("token") == callback_token,
          "callback carries the configured base URL and the minted token")

    # entry cap: 30 of 33 candidates
    conn = db()
    for i in range(33):
        conn.execute("insert into project_notes (project_id, body, created_at)"
                     " values (?, ?, strftime('%s','now'))", (1, "bulk-%02d" % i))
    conn.commit()
    conn.close()
    create_queued(auth, {
        "project_id": "1",
        "title": "Entry cap dispatch",
        "priority": "2",
    })
    check(wait_for(lambda: len(fake.start_payloads) >= 3 or False,
                   timeout=30, what="third /start push") is not None,
          "dispatcher pushes the entry-cap task")
    notes = fake.start_payloads[-1].get("project_notes", [])
    check(len(notes) == 30 and notes[0]["body"] == "note-one" and notes[1]["body"] == "note-two",
          "project_notes capped at 30 entries, oldest kept (%d)" % len(notes))

    # char cap: the giant note is truncated so the total stays <= 20000
    conn = db()
    conn.execute("delete from project_notes where body like 'bulk-%' or body = 'co-note'")
    conn.execute("insert into project_notes (project_id, body, created_at)"
                 " values (?, ?, strftime('%s','now'))", (1, "giant!" + "x" * 25000,))
    conn.commit()
    conn.close()
    create_queued(auth, {
        "project_id": "1",
        "title": "Char cap dispatch",
        "priority": "2",
    })
    check(wait_for(lambda: len(fake.start_payloads) >= 4 or False,
                   timeout=30, what="fourth /start push") is not None,
          "dispatcher pushes the char-cap task")
    notes = fake.start_payloads[-1].get("project_notes", [])
    total = sum(len(n["body"]) for n in notes)
    check(len(notes) == 3 and total == 20000 and notes[2]["body"].startswith("giant!"),
          "project_notes char cap: head kept, giant truncated (entries=%d, chars=%d)"
          % (len(notes), total))

    print("== phase 2: pushed events -> the unified ledger ==")
    # Wait for the last dispatch to finish so its pushes landed.
    check(wait_task_done(auth, task_id_of("Char cap dispatch"), timeout=60) is not None,
          "char-cap task completes")
    conn = db()
    raw_session = conn.execute(
        "select s.id from sessions s join tasks t on t.id = s.task_id"
        " where t.title = 'Char cap dispatch' order by s.id desc limit 1").fetchone()[0]
    rows = conn.execute(
        "select seq, kind from events where session_id = ? order by seq",
        (raw_session,)).fetchall()
    conn.close()
    check(len(rows) >= 5,
          "the ledger captured the round's event batch (%d)" % len(rows))
    kinds = [r[1] for r in rows]
    check(all(k in kinds for k in
              ("lifecycle", "turn_text", "tool_call", "tool_result", "usage")),
          "the opening batch carries the §7.2 kinds (%s)" % sorted(set(kinds)))
    seqs = [r[0] for r in rows]
    check(seqs == sorted(seqs) and len(set(seqs)) == len(seqs),
          "stored events keep one strictly increasing seq per session")

    # The plan-17 event log wire: one newest-first JSON page per task.
    cap_task = task_id_of("Char cap dispatch")
    status, page = saddle_api(
        "GET", "/tasks/%d/events?limit=30" % cap_task, None)
    check(status == 200 and page.get("ok") is True
          and len(page.get("events", [])) >= 5,
          "GET /tasks/<id>/events serves the newest-first page")
    page_events = page["events"]
    ids = [e["id"] for e in page_events]
    check(ids == sorted(ids, reverse=True),
          "the JSON page lists events newest first")
    check(page_events[0]["kind"] == "state_change"
          and page_events[0]["payload"].get("to") == "done",
          "the newest event is the terminal state change")

    # The SSE stream seeds the most-recent window then stays open.
    request = urllib.request.Request(WEB + "/tasks/%d/events/stream" % cap_task)
    stream = urllib.request.urlopen(request, timeout=15)
    stream_events = []
    deadline = time.monotonic() + 6
    while time.monotonic() < deadline and len(stream_events) < 30:
        line = stream.readline().decode().strip()
        if not line.startswith("data: ") or line == "data: ping":
            continue
        event = json.loads(line[6:])
        stream_events.append(event)
    stream.close()
    check(len(stream_events) >= 5,
          "the event stream seeds the most-recent window (%d)" % len(stream_events))
    stream_ids = [e["id"] for e in stream_events]
    check(stream_ids == sorted(stream_ids) and len(set(stream_ids)) == len(stream_ids),
          "the seed batch is oldest first inside the window")

    print("== phase 2: chunks/results ingestion — auth, idempotency ==")
    # Token auth on both push channels (plan 10: the per-saddle
    # bearer authorises every daemon->web call).
    status, _ = saddle_api("POST", "/api/saddle/chunks", None,
                           {"session_id": raw_session, "events": []})
    check(status == 401, "chunks endpoint rejects a missing token")
    status, _ = saddle_api("POST", "/api/saddle/results", "saddle-bogus-token",
                           {"session_id": raw_session, "state": "working"})
    check(status == 401, "results endpoint rejects a wrong token")
    status, _ = saddle_api("POST", "/api/saddle/chunks", fake_token,
                           {"session_id": 99999, "events": []})
    check(status == 400, "chunks for an unknown session rejected")
    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                           {"session_id": raw_session, "state": "done"})
    check(status == 200, "the owning saddle's session accepts its results")

    # Pushed event idempotency: re-pushing a batch (the outbox's
    # at-least-once redelivery after a lost response) consumes the
    # duplicates and stores only the tail.
    def session_event_count(session):
        conn = db()
        try:
            return conn.execute(
                "select count(*) from events where session_id = ?",
                (session,)).fetchone()[0]
        finally:
            conn.close()

    before = session_event_count(raw_session)
    redelivery = [{"seq": r[0], "kind": r[1]} for r in rows[:3]]
    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
                              {"session_id": raw_session, "events": redelivery})
    check(status == 200 and body.get("ok") is True and int(body.get("stored") or 0) == 0,
          "a fully-duplicate batch stores nothing (duplicates consumed)")
    fresh = [{"seq": 9000, "kind": "progress",
              "payload": {"text": "late-flush\r\n"}},
             {"seq": 9001, "kind": "progress", "payload": {}}]
    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
                              {"session_id": raw_session, "events": fresh})
    check(status == 200 and int(body.get("stored") or 0) == 2,
          "a partial-redelivery batch stores only the new events")
    check(session_event_count(raw_session) == before + 2,
          "redelivered events never duplicate rows")
    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
                              {"session_id": raw_session,
                               "events": [{"kind": "progress", "payload": {}}]})
    check(status == 400, "an event without a positive seq is refused")

    # A seq-less batch shape is the daemon contract violation the web
    # rejects wholesale (nothing of the batch stores).
    count_before = session_event_count(raw_session)
    status, _ = saddle_api("POST", "/api/saddle/chunks", fake_token,
                           {"session_id": raw_session,
                            "events": [{"seq": 9100, "kind": "progress"},
                                       {"kind": "usage", "payload": {}}]})
    check(status == 400 and session_event_count(raw_session) == count_before,
          "a batch with a seq-less event is refused atomically")

    # The retired daemon routes 404 (phase 2 deleted /output, /log
    # and /start serving) — and the web app never calls them.
    for method, path in (("GET", "/output?since=0"), ("GET", "/log?since=0"),
                         ("POST", "/start")):
        request = urllib.request.Request(
            "http://127.0.0.1:%d%s" % (fake.port, path),
            data=b"{}" if method == "POST" else None, method=method)
        try:
            urllib.request.urlopen(request, timeout=5)
            gone = False
        except urllib.error.HTTPError as error:
            gone = error.code == 404
        check(gone, "the fake daemon's %s %s is deleted (404)" % (method, path))
    with open(os.path.join(WORK, "app.log")) as handle:
        app_log = handle.read()
    check("Saddle output request" not in app_log and "Saddle log request" not in app_log,
          "the web never polls /output or /log anymore")

    print("== capabilities: tag-based routing ==")
    saddles_state = slot_public(auth, "/saddles", "saddles")
    act(auth, saddles_state["create_capability"], {"name": "gpu"})
    act(auth, saddles_state["create_capability"], {"name": "bigmem"})
    saddles_state = slot_public(auth, "/saddles", "saddles")
    caps = {c["name"]: c for c in saddles_state["capabilities"]}
    check(len(caps) == 2 and all(int(c["saddle_count"]) == 0 and int(c["task_count"]) == 0
                                 for c in caps.values()),
          "capabilities created and listed with zero usage")


    fake_card = saddle_card(auth, "fake-saddle")
    check(fake_card["capabilities"] == [],
          "saddle card starts with no capability chips")

    # A task requiring 'gpu' waits while the only idle saddle lacks it;
    # a later task without requirements still dispatches (skip-and-continue).
    tasks_state = slot_public(auth, "/tasks", "tasks")
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "GPU task",
        "priority": "2",
    })
    gpu_task = task_id_tmp
    detail = task_state(auth, gpu_task)
    act(auth, detail["attach_capability"], {"capability_id": str(int(caps["gpu"]["id"]))})
    check(wait_for(lambda: [c["name"] for c in
                            task_state(auth, gpu_task)["capabilities"]] == ["gpu"]
                   or False, timeout=90, what="capability attached") is not None,
          "task page lists the required capability")
    detail = task_state(auth, gpu_task)
    # The only idle saddle lacks the tag, so the queue flags the wait
    # (read before any other task claims the idle saddle). The state is
    # stable until the any-saddle task below dispatches — poll for it
    # rather than sampling once (a single read under load once sampled
    # the row before the attach's pulse settled).
    check(wait_for(lambda: (lambda row: row["status"] == "queued"
                            and row["capabilities"] == ["gpu"]
                            and row.get("waiting_for_saddle") is True)(
                        [t for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                         if int(t["id"]) == gpu_task][0]) or False,
                   timeout=20, what="waiting flag") is not None,
          "queue row shows the waiting task with its tag and wait flag")

    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Any-saddle task",
        "priority": "3",
    })
    any_task = task_id_tmp

    check(wait_for(lambda: task_state(auth, any_task).get("status") != "queued" or False,
                   timeout=30, what="uncapable task dispatches") is not None,
          "task without requirements dispatches past the waiting task")
    def capability_wait_logged():
        with open(os.path.join(WORK, "app.log")) as handle:
            app_log = handle.read()
        return "waits for a capable saddle" in app_log and "requires: gpu" in app_log
    check(wait_for(capability_wait_logged, timeout=90,
                   what="capability wait log") is not None,
          "dispatcher logs the capability wait")

    # Attaching the tag to the saddle releases the task.
    fake_card = saddle_card(auth, "fake-saddle")
    act(auth, fake_card["attach_capability"], {"capability_id": str(int(caps["gpu"]["id"]))})
    fake_card = saddle_card(auth, "fake-saddle")
    check([c["name"] for c in fake_card["capabilities"]] == ["gpu"],
          "saddle card shows the attached chip")
    check(wait_task_done(auth, gpu_task, timeout=60) is not None,
          "task dispatches once the saddle carries the required tag")
    queue = [t for t in slot_public(auth, "/tasks", "tasks")["tasks"]
             if int(t["id"]) == gpu_task][0]
    check(queue.get("waiting_for_saddle") is not True,
          "wait flag clears once dispatched")

    # Multi-tag: gpu + bigmem required, only gpu carried → waits; adding
    # bigmem completes the superset. Both requirements ride the create
    # form (atomic — attaching them one at a time would race a tick
    # into dispatching on the partial set).
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Big task",
        "priority": "2",
        "capability_id": [str(int(caps["gpu"]["id"])), str(int(caps["bigmem"]["id"]))],
    })
    big_task = task_id_tmp
    check(wait_for(lambda: ([t for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                             if int(t["id"]) == big_task][0].get("waiting_for_saddle")
                            is True) or False, timeout=30, what="big task waits") is not None,
          "task waits when the saddle's tags are not a superset")
    fake_card = saddle_card(auth, "fake-saddle")
    act(auth, fake_card["attach_capability"], {"capability_id": str(int(caps["bigmem"]["id"]))})
    check(wait_task_done(auth, big_task, timeout=60) is not None,
          "task dispatches once the superset is complete")

    # Detach a requirement (on a draft — capabilities lock once a
    # worker has the job); delete a capability everywhere.
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1",
        "title": "Detach probe task",
        "priority": "4",
        "capability_id": [str(int(caps["gpu"]["id"])), str(int(caps["bigmem"]["id"]))],
    })
    detach_probe = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                        if t["title"] == "Detach probe task"][0])
    act(auth, [c for c in task_state(auth, detach_probe)["capabilities"]
               if c["name"] == "bigmem"][0]["detach"], {})
    check([c["name"] for c in task_state(auth, detach_probe)["capabilities"]] == ["gpu"],
          "detached requirement disappears from the task")
    bigmem_row = [c for c in slot_public(auth, "/saddles", "saddles")["capabilities"]
                  if c["name"] == "bigmem"][0]
    act(auth, bigmem_row["delete"], {})
    saddles_state = slot_public(auth, "/saddles", "saddles")
    check(all(c["name"] != "bigmem" for c in saddles_state["capabilities"]),
          "deleted capability gone from the catalogue")
    check([c["name"] for c in task_state(auth, big_task)["capabilities"]] == ["gpu"],
          "deleting a capability drops its task requirements")
    check(all(c["name"] != "bigmem"
              for c in saddle_card(auth, "fake-saddle")["capabilities"]),
          "deleting a capability drops it from saddles")

    print("== B4: required capabilities on the create form ==")
    tasks_state = slot_public(auth, "/tasks", "tasks")
    gpu_option = [o for o in tasks_state.get("capability_options", [])
                  if o["name"] == "gpu"][0]
    check(len(tasks_state.get("capability_options", [])) >= 1,
          "queue page offers capability options in the create form")

    conn = db()
    before = conn.execute("select count(*) from tasks").fetchone()[0]
    conn.close()
    act(auth, tasks_state["create"], {
        "project_id": "1",
        "title": "Bad cap task",
        "capability_id": [gpu_option["id"], "999"],
    })
    conn = db()
    after = conn.execute("select count(*) from tasks").fetchone()[0]
    conn.close()
    check(after == before, "unknown capability id rejects the create")

    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Cap-on-create task",
        "capability_id": [gpu_option["id"]],
    })
    cap_create = task_id_tmp
    detail = task_state(auth, cap_create)
    check([c["name"] for c in detail["capabilities"]] == ["gpu"],
          "capabilities from the create form land on the task")
    queue = [t for t in slot_public(auth, "/tasks", "tasks")["tasks"]
             if int(t["id"]) == cap_create][0]
    check(queue["capabilities"] == ["gpu"],
          "queue row shows the created task's requirements")
    # The waiting flag must hold it back until the capable saddle frees
    # up (the fake saddle still carries gpu from the routing checks).
    check(queue.get("waiting_for_saddle") is not True,
          "capable saddle available: no wait flag on the new task")
    # Drain barrier: the cap task must be CLAIMED before the flaky
    # arc arms fail_next (under parallel-shard load the dispatch tick
    # can lag task creation; an unclaimed predecessor would consume
    # the one-shot failure flag instead of the flaky task).
    check(wait_for(lambda: task_state(auth, cap_create).get("status")
                   in ("working", "done", "failed") or False,
                   timeout=60, what="cap task claimed") is not None,
          "the capable task is claimed before the flaky arc starts")

    print("== retry + resume ==")
    # A daemon-side failure fails the task with its explanation.
    fake.fail_next = True
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Flaky task",
        "priority": "2",
    })
    flaky = task_id_tmp
    check(wait_for(lambda: task_state(auth, flaky).get("status") == "failed" or False,
                   timeout=90, what="flaky task fails") is not None,
          "daemon failure fails the task")
    detail = task_state(auth, flaky)
    check("bash -n rejected" in (detail.get("summary") or ""),
          "failure summary shows the daemon's explanation")

    # Retry: the terminal summary dies with its attempt (plan 16 S4 —
    # it is NOT copied to the thread), the task requeues, history
    # stays, and the re-dispatch restores the workspace.
    comments_before = len(detail.get("comments", []))
    progress_before = len(detail.get("progress", []))
    act(auth, detail["retry"], {})
    detail = task_state(auth, flaky)
    check(detail["status"] == "queued" and (detail.get("summary") or "") == "",
          "retry requeues the task and clears the stale summary")
    check(not any("bash -n rejected" in c["body"] for c in detail["comments"]),
          "no summary→comment copy joins the thread (the terminal cache dies with its attempt, S4)")
    check(any(c["body"] == "Task retried by user" and c["author"] == "system"
              and c.get("is_system") for c in detail["comments"]),
          "retry lands its audit comment as a system row (never an agent row)")
    check(len(detail["comments"]) >= comments_before,
          "retry keeps the comment history")
    pushes_before = len([p for p in fake.start_payloads
                         if int(p.get("task", {}).get("id") or 0) == flaky])
    check(wait_for(lambda: len([p for p in fake.start_payloads
                                if int(p.get("task", {}).get("id") or 0) == flaky])
                   > pushes_before or False, timeout=90,
                   what="flaky task re-dispatched") is not None,
          "retried task dispatches again")
    # The resumed arc surfaces on the next /status read — in the
    # push world that is the operator's manual refresh.
    act(auth, saddle_card(auth, "fake-saddle")["refresh"], {})
    check(wait_task_done(auth, flaky, timeout=60) is not None,
          "retried task completes on the restored workspace")
    detail = task_state(auth, flaky)
    check(all("restored the session context" not in c["body"]
              for c in detail.get("comments", [])),
          "the session-restore stays OUT of the user-facing thread (U2: raw terminal only)")
    check(all("restored the session context" not in p["summary"]
              for p in detail.get("progress", [])),
          "the resume note stays OUT of the progress feed (F8: status is not progress)")
    # Plan 16 S4: terminal text has ONE representation — the failed
    # attempt's summary lived in tasks.summary alone (asserted above:
    # no agent comment), and a HUMAN comment lands as exactly one
    # user-authored row (the web is the single writer; the daemon's
    # terminal echo is raw-log only).
    act(auth, task_state(auth, flaky)["comment"],
        {"body": "The retry landed — nice recovery"})
    flaky_comments = task_state(auth, flaky)["comments"]
    check(len([c for c in flaky_comments
               if c["body"] == "The retry landed — nice recovery"]) == 1
          and all(c["author"] != "agent"
                  for c in flaky_comments
                  if c["body"] == "The retry landed — nice recovery"),
          "a human comment lands as exactly ONE user row (single writer, S4)")

    # Manual resume: a task stuck working on an offline saddle.
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Stuck task",
        "priority": "2",
    })
    stuck = task_id_tmp
    check(wait_for(lambda: task_state(auth, stuck).get("status") == "working" or False,
                   timeout=60, what="stuck task starts working") is not None,
          "stuck task reaches working")
    # Deterministic claim for the resume leg below: the round's end,
    # the refresh probe AND the dispatcher's wake for the requeued task
    # all trigger self-asks that could steal it — freeze every trigger
    # until the manual ask below has made the claim.
    fake.ask_on_terminal = False
    fake.ask_on_idle_status = False
    fake.ask_on_wake = False
    fake.settle_asks()
    conn = db()
    conn.execute("update saddles set status = 'offline' where name = 'fake-saddle'")
    conn.commit()
    conn.close()
    detail = task_state(auth, stuck)
    check(detail.get("resume_available") is True,
          "task page offers resume while its saddle is offline")
    # The resume POSTs task-update cancel to the daemon; the abort
    # handler fires an ungated re-ask that would race the manual claim
    # below — slow it past the manual ask (the knob the slow-ask
    # regression uses).
    fake.ask_delay = 5.0
    act(auth, detail["resume"], {})
    detail = task_state(auth, stuck)
    check(detail["status"] == "queued",
          "manual resume requeues the stuck task")
    check(any(c["body"] == "Saddle became unreachable; task requeued"
              and c["author"] == "system" and c.get("is_system")
              for c in detail["comments"]),
          "manual resume lands the interruption comment as a system row")
    check(all(s["status"] == "offline"
              for s in slot_public(auth, "/saddles", "saddles")["saddles"]),
          "saddle stays offline after the manual requeue")

    # Bringing the saddle back lets the requeued task resume and finish.
    fake.state = "idle"
    fake_card = saddle_card(auth, "fake-saddle")
    act(auth, fake_card["refresh"], {})
    check(any(s["status"] == "idle"
              for s in slot_public(auth, "/saddles", "saddles")["saddles"]
              if s["name"] == "fake-saddle"),
          "refreshed saddle returns idle")
    # The refreshed daemon is idle-awake: it asks (the real loop's
    # own ask; driven directly for determinism — every self-ask trigger
    # is frozen and the abort re-ask is slowed past it; joining that
    # delayed thread here would fire its ask first, so it is left to
    # sleep out the rest of the bundle).
    status, payload = fake.ask_now()
    fake.ask_on_idle_status = True
    fake.ask_on_terminal = True
    fake.ask_on_wake = True
    fake.ask_delay = 0.0
    check(status == 200, "the refreshed daemon's ask claims the requeued task")
    check(wait_task_done(auth, stuck, timeout=60) is not None,
          "requeued task resumes on the refreshed saddle and completes")