clanker-saddle/test/web_stack_test.py

#!/usr/bin/env python3
"""Self-contained web-stack feature tests: app + fake saddle daemon.

Unlike integration_test.py (which needs a real saddle daemon and a mock
LLM), this suite spawns everything it needs:

- the clanker-saddle app binary (path from argv[1], or builddir default)
  in a scratch directory with its own web-config.json (statum keys copied
  from the repo, plus `saddle_callback_base_url`) and SQLite database
- in-process fake saddle daemons speaking the wire contract: POST
  /register (plan 10: receives {saddle_token, web_base_url}, answers
  {ok, ssh_public_key, version}), POST /wake (records the wake; the
  daemon then asks POST /api/saddle/next-task), POST /task-update
  (cancel/comment/subtask-status), GET /status (scripted reflection —
  exceptional use only), and the phase-2 push channels home:
  POST /api/saddle/chunks (raw PTY batches) and POST /api/saddle/results
  (round entries + state transitions). The retired /start, /output and
  /log routes 404 like a post-phase-2 daemon.

It then drives the app the way statum.js does (entrypoint loads + action
POSTs) and asserts the Workstream B features against the wire contract:

3. plan-review blocks hide their duplicate body preview
5. rules CRUD/attach + the effective `rules` array in assignments
6. project notes, POST /api/saddle/note, `project_notes` + `callback`
   in assignments
   plus the raw-log side: pushed chunks -> session_raw_log + terminal
   raw view

Plan 10 phase 1 (work acquisition flip):

- per-saddle tokens: registration mints + /register-delivers them,
  legacy tokens migrate onto the row at startup, removal revokes
- token auth on every /api/saddle/* endpoint
- next-task allocation: concurrent asks race exactly one winner,
  affinity pinning for parked tasks, resume_notice on suspended resumes
- dispatcher: wake -> ask -> allocate roundtrip, the lost-wake guard
  (no ask after a wake -> offline, re-wake the next capable), and the
  pre-flip /answer + /start resume paths kept as the rollback route

Plan 10 phase 2 (telemetry flip):

- pushed chunks: idempotent on (session_id, seq), ordered, landing in
  session_raw_log so the SSE terminal keeps working
- pushed results: entries store like the /output path did (redelivery
  idempotent on the daemon entry ids), blocked raises the block row +
  parks + bumps the pulse (sealed refresh shows it), done/failed
  complete/fail the task, suspended parks for the affinity resume
- answer resumes, both daemon shapes: a SLEEPING affine saddle (its
  post-park ask returned 204) is woken, its ask allocating the resume
  with the verdict riding the assignment; a daemon parked MID-ROUND
  (its round holds the block's answer wait open, so it can never ask)
  gets the verdict delivered into that wait via POST /answer — same
  session, no wake. A daemon dying with a parked-with-answered task
  never requeues it: the verdict and session survive for the resume
- staleness: a silent runner gets ONE exceptional GET /status probe —
  alive keeps it running, dead requeues the task with a bot comment and
  marks the saddle offline; a recovered daemon's ask flips offline rows
  back to idle
- the deleted layers are gone: /output, /log and /start 404 on the
  daemon side, and nothing on the web side calls them

Plan 10 phase 3 (capability awareness + UI):

- assignment capability context: {saddle_tags, fleet_tags} on every
  payload, GET /api/saddle/capabilities (per-saddle token) feeding
  the daemon's `saddle capabilities` CLI
- subtask callback capabilities: explicit-only attachment validated
  against the operator tag set, unknown tags rejected with the valid
  list in the error
- task page: status block pinned to the top, blocking sections
  newest-first (sealed GET state)
- home dashboard: pinned OPEN blocks as newest-first link-through
  cards (no inline answer forms) above the cross-task activity
  timeline (last ~50 events newest-first)
- lost-wake-guard regression: a saddle that asks SLOWLY after its
  wake (busy winding down) still claims the task — the wake is never
  silently dropped, no re-wake storm fires, and the never-asks guard
  path (offline + re-wake of the next capable) stays green

Prints PASS/FAIL lines Statum-style; exits non-zero on any failure.

Plan 11 task D+E (UI + notifications), the `push` shard:

- task page: progress feed above the status card, status + description
  in a 50/50 grid, edit expander under the history expander, plan
  behind an expander, answered blocks behind an expander; queue screen:
  New Task expander above the filters; dashboard waiting-on-drafts
  block with links to where Enqueue lives (pages shard)
- real Web Push over VAPID: keypair generated at startup into
  web-config.json, GET /sw.js service worker with click-through,
  subscribe/unsubscribe + per-task Notify me actions, block events and
  waiting-on-drafts pushed globally, first start/failure/completion to
  opted-in users, best-effort queue with dispatcher-tick retry, 410/404
  endpoint pruning — payloads verified protocol-level (RFC 8291
  decryption + ES256 JWT signature) because the harness owns the
  browser-side keys

Heredoc completion reports (pages shard):

- terminal tasks render tasks.summary (set by done/fail) as a
  markdown completion-report card below the status/description grid —
  "Completion report" when done, "Failure summary" when failed, no
  card when empty; the summary renders through the shipped
  stm-markdown renderer (headings/code, no syntax leaks, HTML
  escaped) and the dashboard timeline snippets its first line

Speed pass (plan 11, task 0):

- SHARDING: the scenario groups are partitioned into seven independent
  bundles (pages, dispatch, subtasks, lineage, allocation, lifecycle,
  answers), each running in its own process against its own app
  instance, fake daemons and SQLite database. The default invocation
  runs all bundles in parallel and aggregates the verdicts and check
  counts into one exit code — meson sees a single green/red test.
  `--bundle NAME` runs one shard for debugging; WEBSTACK_SEQUENTIAL=1
  runs them in-process, one after another.
- Within a bundle, scenario groups SHARE the app+daemon instance and
  carry per-group state as database rows (task/session/block
  fixtures), never process restarts; the one deliberate restart (the
  startup token migration) lives in the answers bundle.
- EVENT-DRIVEN WAITS: wait_for polls at 50ms; the fixed sleeps were
  converted either to observable-condition waits (daemon-side
  counters, e.g. failed-delivery hits) or to quiet_window() — a
  bounded negative-assertion poll that FAILS LOUDLY the moment the
  event it guards against happens (a sleep could silently pass).
- ENV KNOBS: SADDLE_DISPATCH_TICK_S=1 (dispatcher cadence; default 5
  in production), SADDLE_HTTP_TIMEOUT_S=2, SADDLE_LOST_WAKE_S=2,
  SADDLE_STALE_S=3, plus the fake daemon's scripted timings — the
  wire-traffic shapes stay realistic, just faster.
"""
import json
import os
import re
import shutil
import socket
import sqlite3
import subprocess
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

# Web Push protocol verification (plan 11 task E): the test harness
# owns the browser-side keys, so it can do everything the push service
# does — decrypt the RFC 8291 body and verify the ES256 VAPID JWT.
import base64
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.asymmetric.utils import encode_dss_signature
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat

REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
APP = os.path.abspath(sys.argv[1]) if len(sys.argv) > 1 else os.path.join(REPO, "builddir", "clanker-saddle")

# --- speed-pass tuning -----------------------------------------------------
# The fake daemon's scripted timings and the dispatch-tick cadence the
# quiet windows are sized against. All overridable for debugging.
TICK_S = float(os.environ.get("SADDLE_DISPATCH_TICK_S", "1") or 1)
FAKE_WORK_S = float(os.environ.get("WEBSTACK_FAKE_WORK_S", "1.2") or 1.2)
FAKE_FAIL_S = float(os.environ.get("WEBSTACK_FAKE_FAIL_S", "0.5") or 0.5)
FAKE_HEARTBEAT_S = float(os.environ.get("WEBSTACK_FAKE_HEARTBEAT_S", "0.4") or 0.4)
FAKE_TRIGGER_S = float(os.environ.get("WEBSTACK_FAKE_TRIGGER_S", "0.05") or 0.05)
SHARD_TIMEOUT_S = float(os.environ.get("WEBSTACK_SHARD_TIMEOUT_S", "600") or 600)


def quiet_seconds():
    """A negative-assertion window spanning several dispatch ticks (the
    converted fixed sleeps were 3-7s at the old 5s tick; the bound never
    exceeds the old timeout)."""
    return min(2 * TICK_S + 0.8, 7.0)


WORK = os.environ.get("WEBSTACK_WORK", "/tmp/kilo/webstack-e2e")
DB_PATH = os.path.join(WORK, "web.sqlite")

passes = 0
failures = 0


def check(condition, label):
    global passes, failures
    if condition:
        passes += 1
        print("PASS %s" % label)
    else:
        failures += 1
        print("FAIL %s" % label)


def free_port():
    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        return sock.getsockname()[1]


APP_PORT = free_port()
DAEMON_PORT = free_port()
WEB = "http://127.0.0.1:%d" % APP_PORT


def bundle_env(name):
    """Point the module's runtime globals at one shard's scratch dir,
    database and ports (each bundle is a process with its own app,
    daemons and SQLite file)."""
    global WORK, DB_PATH, APP_PORT, DAEMON_PORT, WEB
    base = os.environ.get("WEBSTACK_WORK", "/tmp/kilo/webstack-e2e")
    WORK = "%s-%s" % (base, name)
    DB_PATH = os.path.join(WORK, "web.sqlite")
    APP_PORT = free_port()
    DAEMON_PORT = free_port()
    WEB = "http://127.0.0.1:%d" % APP_PORT


def http(method, url, data=None, headers=None, timeout=30):
    body = None
    if data is not None:
        pairs = []
        for k, v in data.items():
            for item in (v if isinstance(v, (list, tuple)) else [v]):
                pairs.append((k, item))
        body = "&".join("%s=%s" % (k, urllib.parse.quote(str(item))) for k, item in pairs).encode()
    request = urllib.request.Request(WEB + url if url.startswith("/") else url,
                                     data=body, method=method)
    if data is not None:
        request.add_header("Content-Type", "application/x-www-form-urlencoded")
    for key, value in (headers or {}).items():
        request.add_header(key, value)
    # Retry dropped/refused connections: the freshly started app can
    # close a very early keep-alive connection once while warming up,
    # and its first seconds can wobble under load.
    for attempt in range(3):
        try:
            with urllib.request.urlopen(request, timeout=timeout) as response:
                return response.status, response.read().decode()
        except urllib.error.HTTPError as error:
            return error.code, error.read().decode()
        except (urllib.error.URLError, OSError):
            if attempt >= 2:
                raise
            time.sleep(0.75)


def directives(uri, auth=None):
    headers = {}
    if auth:
        headers["X-Statum-Slot"] = "key=%s; as_at=%s" % (auth["key"], auth["as_at"])
    status, body = http("GET", "/_statum/entrypoint?uri=" + urllib.parse.quote(uri.split("?")[0]),
                        headers=headers)
    if status != 200:
        raise SystemExit("entrypoint %s -> HTTP %d: %s" % (uri, status, body[:300]))
    return json.loads(body)


def slot_public(auth, uri, type_name):
    for directive in directives(uri, auth):
        if directive["type"] == "set":
            content = directive["snapshot_frame"].get("content")
            if not content:
                continue
            content = json.loads(content)
            if content["slot"]["type"] == type_name:
                return content["public"]
    raise SystemExit("no %s slot in entrypoint state for %s" % (type_name, uri))


def act(auth, action, fields):
    headers = {"X-Statum-Slot": "key=%s; as_at=%s" % (auth["key"], auth["as_at"])}
    if action.get("private"):
        headers["X-Statum-Private"] = action["private"]
    status, body = http(action.get("method", "POST"), action["uri"], fields, headers)
    if status != 200:
        raise SystemExit("action %s -> HTTP %d: %s" % (action["uri"], status, body[:300]))
    return json.loads(body)


def login(username, password):
    frames = directives("/login")
    content = json.loads(frames[0]["snapshot_frame"]["content"])
    action = content["public"]["login"]
    status, body = http(action.get("method", "POST"), action["uri"],
                        {"username": username, "password": password})
    if status != 200:
        raise SystemExit("login -> HTTP %d: %s" % (status, body[:300]))
    for directive in json.loads(body):
        if directive["type"] == "set":
            frame = json.loads(directive["snapshot_frame"]["content"])
            if frame["slot"]["type"] == "auth":
                return {"key": frame["slot"]["key"], "as_at": frame["as_at"]}
    raise SystemExit("login response carried no auth slot: " + body[:300])


def wait_for(predicate, timeout=60, what="condition", interval=0.05):
    deadline = time.monotonic() + timeout
    last = None
    while time.monotonic() < deadline:
        last = predicate()
        if last is not False and last is not None:
            return last
        time.sleep(interval)
    print("  (timeout waiting for %s; last: %s)" % (what, last))
    dump_threads()
    return None


def quiet_window(seconds, predicate, what):
    """A converted fixed sleep: asserts a NEGATIVE condition (nothing may
    happen) across a window spanning several dispatch ticks, polling at
    50ms. The moment the condition breaks the window fails loudly
    (diagnostics + thread dump) instead of silently passing a sleep."""
    deadline = time.monotonic() + seconds
    while time.monotonic() < deadline:
        value = predicate()
        if value is False or value is None:
            remaining = deadline - time.monotonic()
            print("  (quiet window '%s' broken after %.2fs of %.2fs — the"
                  " guarded event DID happen)" % (what, seconds - remaining, seconds))
            dump_threads()
            return False
        time.sleep(0.05)
    return True


def dump_threads():
    """Stacks of suite threads not parked in sleep — pinpoints a wedged
    fake lifecycle when a wait times out."""
    import traceback
    shown = 0
    for thread_id, frame in sys._current_frames().items():
        stack = "".join(traceback.format_stack(frame)[-4:])
        if ("web_stack_test.py" in stack) and ("sleep" not in stack) and shown < 6:
            shown += 1
            print("  --- thread %s ---\n%s" % (thread_id, stack))


def task_state(auth, task_id):
    return slot_public(auth, "/tasks/%d" % task_id, "task")


def wait_task_done(auth, task_id, timeout=90):
    return wait_for(lambda: task_state(auth, task_id).get("status") == "done" or False,
                    timeout=timeout, what="task %d -> done" % task_id)


def task_id_of(title):
    conn = db()
    try:
        return conn.execute("select id from tasks where title = ?", (title,)).fetchone()[0]
    finally:
        conn.close()


def enqueue_task(auth, task_id):
    """The human-only Enqueue action (plan 11): draft -> queued."""
    detail = task_state(auth, task_id)
    assert detail["status"] == "draft", "task %d not in draft (%s)" % (task_id, detail["status"])
    act(auth, detail["enqueue"], {})
    return task_state(auth, task_id)["status"]


def create_queued(auth, fields):
    """Create via the queue form (lands in draft) and enqueue it — the
    standard 'this task should dispatch' fixture of the plan-11 world."""
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], fields)
    task_id = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                   if t["title"] == fields["title"]][0])
    enqueue_task(auth, task_id)
    return task_id


def saddle_api(method, path, token, payload=None):
    """Raw call against the saddle ingest API; returns (status, json)."""
    request = urllib.request.Request(WEB + path,
                                     data=json.dumps(payload or {}).encode(), method=method)
    request.add_header("Content-Type", "application/json")
    if token is not None:
        request.add_header("Authorization", "Bearer " + token)
    try:
        with urllib.request.urlopen(request, timeout=15) as response:
            return response.status, json.loads(response.read().decode() or "{}")
    except urllib.error.HTTPError as error:
        body = error.read().decode()
        try:
            return error.code, json.loads(body)
        except ValueError:
            return error.code, {"raw": body}


def db():
    return sqlite3.connect(DB_PATH)


def saddle_card(auth, name):
    rows = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
            if s["name"] == name]
    return rows[0]


def queued_count():
    conn = db()
    try:
        return conn.execute(
            "select count(*) from tasks where status = 'queued'").fetchone()[0]
    finally:
        conn.close()


def command_rows(session):
    conn = db()
    try:
        return conn.execute(
            "select command_text, type from session_commands"
            " where session_id = ? order by id", (session,)).fetchall()
    finally:
        conn.close()


class FakeSaddle:
    """In-process saddle daemon implementing the wire contract.

    Phase-2 push behaviour: the daemon asks POST /api/saddle/next-task
    whenever idle-awake (at registration, on POST /wake, after its task
    reaches a terminal state, and when a /status probe observes it
    parked idle — throttled), accepts assignments like POST /start used
    to, then PUSHES its telemetry home: raw PTY chunks via POST
    /api/saddle/chunks and round results with state transitions via
    POST /api/saddle/results (working at round start, terminal states
    when the scripted lifecycle ends them). /output, /log and /start
    are gone (404) like a post-phase-2 daemon; /task-update serves the
    cancel/comment/subtask-status kinds; POST /answer receives a
    delivered verdict when the round is parked in a block's answer
    wait (409 when nothing waits — the real daemon's contract; the
    resumed round continues after the response). Every trigger can be
    disabled per test for determinism; ask_now() performs a
    synchronous manual ask.
    """

    PROMPT = "bbarrow@fedora:/workspace/task-1$ "

    def __init__(self):
        self.lock = threading.Lock()
        self.start_payloads = []       # captured assignment payloads (/start + next-task 200s)
        self.state = "idle"
        self.summary = ""
        self.state_entered = time.monotonic()
        self.fail_next = False          # next task drives the task to failed
        self.start_counts = {}          # task id -> assignment count
        self.resumed_pending = False    # crash-resume flag on the next /status
        self.suspend_on_title = None    # this task (by title) parks as suspended
        self.output_seq = 100           # daemon-global entry ids for pushed results
        self.last_activity_epoch = int(time.time())
        self.subtask_status_events = []  # recorded /task-update subtask-status payloads
        self.subtask_fail_hits = 0        # delivery attempts that 500ed (fail_subtask_status)
        self.fail_subtask_status = False  # exercise the web's retry path
        self.hang_next = {}             # path suffix -> hang count (never responds)
        self.abort_failures = 0          # cancel task-updates failing (500) while > 0
        self.abort_log = []              # task ids aborted via task-update cancel
        self.running_task = 0            # task id the daemon believes it runs
        self.running_title = ""          # its title (suspend targeting)
        self.session_id = None           # the web-side run row of the running task
        self.hold_state = False          # freeze lifecycle + heartbeat pushes
        self.ssh_public_key = ""         # reported on /status + /register
        self.started_mono = time.monotonic()
        self.freeze_activity = False     # hold last_activity_epoch still
        self.silent = False              # suppress pushes (staleness tests)
        self.dead = False                # kill(): no serving AND no calls home
        self.status_hits = 0             # GET /status arrivals (probe accounting)
        self.comment_log = []            # recorded /task-update comment payloads
        self.scripted_corpus = None      # chunk batch pushed when a task starts
        # --- plan 10 phase 1: ask/wake/register ---
        self.token = None                # per-saddle token from POST /register
        self.register_log = []           # recorded /register bodies
        self.wake_log = []               # recorded POST /wake arrivals
        self.asks = []                   # [{"status": int, "payload": dict|None}]
        self.ask_on_register = True
        self.ask_on_wake = True
        self.ask_on_terminal = True
        self.ask_on_idle_status = True
        self.ask_delay = 0.0             # seconds; a slow daemon delays its wake-triggered ask
        self.serve_register = True         # False: 404 /register (pre-flip daemon)
        self.last_ask_trigger = 0.0
        self.port = None
        self.ask_threads = []            # scheduled ask-trigger threads (settle_asks joins them)
        # --- answer delivery (the parked-session resume) ---
        self.pending_block = None        # the block payload this daemon waits on (state blocked)
        self.answer_log = []             # recorded POST /answer bodies
        self.answer_push_delay = 0.2     # the resumed round's first push, after the response

    def set_state(self, state, summary=None):
        """Advances the scripted lifecycle."""
        self.state = state
        if summary is not None:
            self.summary = summary
        self.state_entered = time.monotonic()

    def touch_activity(self):
        if not self.freeze_activity:
            self.last_activity_epoch = int(time.time())

    def queue_output(self, entry_type, command, output):
        """Pushes a typed entry as a round-results push (the daemon's
        entry log riding the results channel)."""
        with self.lock:
            entry = {
                "id": self.output_seq,
                "command": command,
                "output": output,
                "exit_code": 1,
                "type": entry_type,
                "timestamp": "2026-09-06T00:00:00Z",
            }
            self.output_seq += 1
        self.push_results(self.state if self.state in ("working",) else "working",
                          entries=[entry])

    # --- plan 10 phase 2: the push channels ---

    def push_chunks(self, chunks):
        """Pushes raw PTY chunks home: elements are texts (assigned
        fresh seqs) or {seq, text} dicts (a redelivery or explicit
        corpus). Pushes are suppressed while silent/hold_state (the
        staleness tests' silent runner)."""
        if self.dead or self.silent or self.token is None or self.session_id is None:
            return
        batch = []
        with self.lock:
            for chunk in chunks:
                if isinstance(chunk, dict):
                    batch.append({"seq": int(chunk["seq"]), "text": chunk["text"]})
                else:
                    batch.append({"seq": self.next_push_seq(), "text": chunk})
        try:
            saddle_api("POST", "/api/saddle/chunks", self.token,
                       {"session_id": self.session_id, "chunks": batch})
        except (OSError, ValueError):
            pass  # app restarting / shutting down mid-test

    def next_push_seq(self):
        # The daemon-side chunk counter for ad-hoc pushes.
        self._push_seq = getattr(self, "_push_seq", 0) + 1
        return self._push_seq

    def push_results(self, state, entries=None, block=None, summary=None, session_id=None):
        """Pushes one round-results payload home (state transition plus
        its entries). Returns the HTTP status when it went out, None
        when suppressed."""
        if self.dead or self.silent:
            return None
        if self.token is None:
            return None
        target = session_id if session_id is not None else self.session_id
        if target is None:
            return None
        payload = {"session_id": target, "state": state}
        if entries is not None:
            payload["entries"] = entries
        if block is not None:
            payload["block"] = block
            if state == "blocked":
                # The daemon parks its round on this block's answer wait
                # (the real daemon's wait_for_answer).
                self.pending_block = dict(block)
        if summary is not None:
            payload["summary"] = summary
        try:
            status, _ = saddle_api("POST", "/api/saddle/results", self.token, payload)
            if status != 200:
                print("  [fake %s] push %s -> HTTP %s (task=%s)"
                      % (self.port, state, status, self.running_task))
            return status
        except (OSError, ValueError) as error:
            print("  [fake %s] push %s failed: %r (task=%s)"
                  % (self.port, state, error, self.running_task))
            return None

    # --- plan 10 phase 1: the ask path ---

    def ask_now(self):
        """Synchronously asks the web for work (a phase-1 daemon's idle
        ask). A 200 assignment is accepted exactly like POST /start;
        returns (status, payload|None)."""
        token = self.token
        if token is None:
            raise SystemExit("fake daemon asked before receiving its token")
        status, payload = saddle_api("POST", "/api/saddle/next-task", token)
        with self.lock:
            self.asks.append({"status": status, "payload": payload})
        if status == 200:
            self.accept_assignment(payload)
        return status, payload

    def _trigger_ask(self, reason):
        """Schedules a delayed ask (the daemon woke / went idle-awake).
        ask_delay models a SLOW asker — a daemon still winding down its
        round when the wake lands: it honours the wake, but its ask
        arrives late (still inside the web's lost-wake guard window)."""
        def run():
            if self.state in ("working", "failing"):
                return  # a running daemon does not ask
            if self.ask_delay > 0:
                time.sleep(self.ask_delay)
            for attempt in range(3):
                try:
                    status, _ = self.ask_now()
                    if status in (200, 204):
                        return
                    print("  (ask %s -> HTTP %s, attempt %d)" % (reason, status, attempt + 1))
                except Exception as error:  # app restarting / blips: retry
                    print("  (ask %s failed: %r, attempt %d)" % (reason, error, attempt + 1))
                time.sleep(1.0)

        def starter():
            time.sleep(FAKE_TRIGGER_S)
            run()

        thread = threading.Thread(target=starter, daemon=True)
        with self.lock:
            self.ask_threads.append(thread)
        thread.start()

    def settle_asks(self, timeout=15):
        """Joins every ask-trigger thread scheduled so far — the
        event-driven replacement for the fixed sleeps that waited for
        in-flight asks to land after triggers were frozen."""
        with self.lock:
            threads = [t for t in self.ask_threads if t.is_alive()]
        deadline = time.monotonic() + timeout
        for thread in threads:
            thread.join(timeout=max(0.1, deadline - time.monotonic()))
        with self.lock:
            self.ask_threads = [t for t in self.ask_threads if t.is_alive()]
        return all(not thread.is_alive() for thread in threads)

    def maybe_trigger_ask(self, reason, throttle=False):
        """Schedules an idle-awake ask. Protocol events (wake, abort,
        round end, park) ask singularly and immediately; only the
        /status-observation path throttles (a burst of probes must not
        hammer the ask endpoint)."""
        now = time.monotonic()
        if throttle and now - self.last_ask_trigger < 1.0:
            return
        self.last_ask_trigger = now
        self._trigger_ask(reason)

    def accept_assignment(self, payload):
        """A 200 next-task assignment: adopted exactly like POST /start."""
        self.start_payloads.append(payload)
        task = payload.get("task") or {}
        task_id = task.get("id")
        self.running_task = int(task_id or 0)
        self.running_title = task.get("title") or ""
        self.session_id = payload.get("session_id")
        # The assignment's answers resolve any block this daemon was
        # parked on (resume-carries-answer).
        self.pending_block = None
        count = self.start_counts.get(task_id, 0) + 1
        self.start_counts[task_id] = count
        if count > 1:
            # Crash-resume contract: re-pushing a task whose
            # workspace still exists restores it; the first
            # /status after the restore reports "resumed": true.
            self.resumed_pending = True
        self.set_state("failing" if self.fail_next else "working")
        self.fail_next = False
        # The round's opening telemetry: the flushed-so-far raw PTY
        # batch plus the working state push (>= 2 flushes per round
        # and the 10s timer are the real daemon's contract; the fake
        # pushes its batch up front and heartbeats while working).
        if self.scripted_corpus is not None:
            self.push_chunks([dict(chunk) for chunk in self.scripted_corpus])
        else:
            self.push_chunks([self.PROMPT, "echo hello\r\n", "hello\r\n"])
        self.push_results("working")

    class Handler(BaseHTTPRequestHandler):
        daemon = None  # set on the class before serving
        protocol_version = "HTTP/1.1"

        def log_message(self, fmt, *args):
            pass

        def maybe_hang(self, path):
            """True when this request should hang forever (never respond),
            consuming one unit of the path's hang budget."""
            for suffix, count in list(self.daemon.hang_next.items()):
                if count > 0 and path.startswith(suffix):
                    self.daemon.hang_next[suffix] = count - 1
                    return True
            return False

        def _send(self, status, obj):
            body = json.dumps(obj).encode()
            self.send_response(status)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)

        def do_POST(self):
            if self.maybe_hang(self.path):
                time.sleep(600)
                return
            self.daemon.touch_activity()
            length = int(self.headers.get("Content-Length") or 0)
            raw = self.rfile.read(length).decode() if length else "{}"
            if self.path == "/register":
                # Plan 10: the web delivers the per-saddle token + its
                # own base URL; the daemon persists the token for its
                # calls home and reports its ssh key + version.
                if not self.daemon.serve_register:
                    self._send(404, {"error": "no such route"})
                    return
                try:
                    payload = json.loads(raw)
                except ValueError:
                    payload = {"raw": raw}
                self.daemon.register_log.append(payload)
                if payload.get("saddle_token"):
                    self.daemon.token = payload["saddle_token"]
                self._send(200, {"ok": True,
                                 "ssh_public_key": self.daemon.ssh_public_key,
                                 "version": "0.1-fake"})
                if self.daemon.ask_on_register:
                    self.daemon.maybe_trigger_ask("register")
                return
            if self.path == "/wake":
                # Plan 10: a capable task is ready; a sleeping daemon
                # asks for work (immediately — a wake is a direct
                # command, never throttled).
                self.daemon.wake_log.append(time.monotonic())
                self._send(200, {"ok": True})
                if self.daemon.ask_on_wake:
                    self.daemon.maybe_trigger_ask("wake")
                return
            if self.path == "/answer":
                # Plan 11 (daemon Task A): parks FREE the daemon, so no
                # block ever waits in-process — the verdict travels in
                # the resuming allocation's resume_notice. Always 409,
                # exactly like the real daemon's contract.
                try:
                    payload = json.loads(raw)
                except ValueError:
                    self._send(400, {"error": "bad json"})
                    return
                self.daemon.answer_log.append({
                    "refused": True,
                    "answer": payload.get("answer", "")})
                self._send(409, {"error": "no block is waiting for an answer — parks resume via allocation"})
                return
            if self.path == "/task-update":
                # Plan 10: cancel (abort machinery), comment (injected
                # turn) and subtask-status (live subtask set) — delivered
                # directly to a RUNNING daemon, no auth.
                try:
                    payload = json.loads(raw)
                except ValueError:
                    self._send(400, {"error": "bad json"})
                    return
                kind = payload.get("kind")
                if kind == "cancel":
                    if self.daemon.abort_failures > 0:
                        self.daemon.abort_failures -= 1
                        self._send(500, {"error": "cancel failed (simulated)"})
                        return
                    aborted = self.daemon.running_task
                    self.daemon.set_state("idle")
                    self.daemon.summary = ""
                    self.daemon.running_task = 0
                    self.daemon.session_id = None
                    self.daemon.abort_log.append(aborted)
                    self._send(200, {"ok": True, "aborted_task_id": aborted})
                    # An aborted daemon returns to its idle loop: it asks.
                    self.daemon.maybe_trigger_ask("aborted")
                elif kind == "comment":
                    self.daemon.comment_log.append(payload)
                    self._send(200, {"ok": True})
                elif kind == "subtask-status":
                    if self.daemon.fail_subtask_status:
                        self.daemon.subtask_fail_hits += 1
                        self._send(500, {"error": "simulated failure"})
                        return
                    self.daemon.subtask_status_events.append(payload)
                    self._send(200, {"ok": True})
                else:
                    self._send(400, {"error": "unknown task-update kind"})
                return
            self._send(404, {"error": "no such route"})
            self.close_connection = True

        def do_GET(self):
            if self.maybe_hang(self.path):
                time.sleep(600)
                return
            if self.path != "/status":
                # /status itself is not daemon activity: the liveness
                # signal must reflect loop rounds / command completions.
                self.daemon.touch_activity()
            if self.path == "/status":
                self.daemon.status_hits += 1
                response = {"state": self.daemon.state, "summary": self.daemon.summary,
                            "uptime_seconds": int(time.monotonic() - self.daemon.started_mono),
                            "last_activity_epoch": self.daemon.last_activity_epoch,
                            "ssh_public_key": self.daemon.ssh_public_key}
                if self.daemon.resumed_pending:
                    response["resumed"] = True
                    self.daemon.resumed_pending = False
                # A parked/terminal daemon is idle-awake: a phase-1
                # daemon asks (and sleeps on 204). Throttled so status
                # poll bursts do not hammer the ask endpoint.
                if self.daemon.ask_on_idle_status and self.daemon.state in (
                        "idle", "done", "failed", "suspended"):
                    self.daemon.maybe_trigger_ask("idle-status", throttle=True)
                self._send(200, response)
            else:
                self._send(404, {"error": "no such route"})
                self.close_connection = True

    def run_lifecycle(self):
        """Applies the scripted lifecycle on a timer (like a real daemon
        whose task ends on its own): working heartbeats chunks (the
        flush timer's stand-in, keeping the staleness clock fed), works
        for FAKE_WORK_S, then pushes done/suspended; failing pushes
        failed after FAKE_FAIL_S — and a task END makes the idle-awake
        daemon ask again."""
        while True:
            time.sleep(0.1)
            if self.hold_state or self.silent:
                # hold_state freezes the script; silent models a daemon
                # mid-round with nothing to flush — its round has not
                # ended, so no terminal push fires while quiet.
                continue
            if self.state == "failing":
                if time.monotonic() - self.state_entered >= FAKE_FAIL_S:
                    self.set_state("failed", "bash -n rejected the script (simulated)")
                    self.push_results("failed",
                                      summary="bash -n rejected the script (simulated)")
                    if self.ask_on_terminal:
                        self.maybe_trigger_ask("terminal")
            elif self.state == "working":
                if time.monotonic() - self.state_entered >= FAKE_WORK_S:
                    if self.suspend_on_title and self.suspend_on_title == self.running_title:
                        # `saddle suspend`: park until a fresh assignment
                        # resumes the workspace (title-targeted one-shot,
                        # so an interleaved dispatch cannot eat the flag).
                        self.set_state("suspended")
                        self.suspend_on_title = None
                        self.push_results("suspended")
                        # A parked daemon is idle-awake: it asks (and
                        # sleeps on 204) so a later wake can reach it.
                        if self.ask_on_idle_status:
                            self.maybe_trigger_ask("parked")
                    else:
                        self.set_state("done", "fake daemon finished")
                        self.push_results("done", summary="fake daemon finished")
                        if self.ask_on_terminal:
                            self.maybe_trigger_ask("terminal")
                elif time.monotonic() - getattr(self, "last_heartbeat", 0) >= FAKE_HEARTBEAT_S:
                    # The flush timer: output keeps flowing, so chunks
                    # keep arriving (the staleness window never opens on
                    # a healthy runner).
                    self.last_heartbeat = time.monotonic()
                    self.push_chunks(["working-heartbeat\r\n"])

    def serve(self, port=None):
        """Binds the daemon's HTTP server. A requested port that is
        already taken (free_port() hands out recyclable ports — two
        calls can collide) falls back to an OS-assigned one instead of
        silently dying and impersonating whatever owns that port."""
        if port is None:
            port = DAEMON_PORT
        handler = type("BoundHandler", (FakeSaddle.Handler,), {"daemon": self})
        self.server = None
        try:
            self.server = ThreadingHTTPServer(("127.0.0.1", port), handler)
        except OSError:
            print("(fake daemon: port %d taken, falling back)" % port)
            self.server = ThreadingHTTPServer(("127.0.0.1", 0), handler)
        self.port = self.server.server_address[1]
        threading.Thread(target=self.server.serve_forever, daemon=True).start()
        threading.Thread(target=self.run_lifecycle, daemon=True).start()
        return self.server

    def kill(self):
        """The daemon dies: it stops listening (connections refused) AND
        stops calling home — no pushes, no asks (a dead process has no
        outbound anything; in-flight handler threads may linger but the
        wire contract's traffic is one-directional now)."""
        self.dead = True
        if getattr(self, "server", None) is not None:
            self.server.shutdown()
            self.server.server_close()
            self.server = None


def start_app():
    config = json.load(open(os.path.join(REPO, "web-config.json")))
    config["saddle_callback_base_url"] = WEB
    with open(os.path.join(WORK, "web-config.json"), "w") as handle:
        json.dump(config, handle)

    env = dict(os.environ)
    env.update({
        "SPRY_DB_PATH": DB_PATH,
        "SPRY_SEED_ADMIN_USERNAME": "admin",
        "SPRY_SEED_ADMIN_PASSWORD": "adminpass123",
        "SADDLE_LLM_BASE_URL": "http://127.0.0.1:%d" % DAEMON_PORT,
        "SADDLE_PROJECTS_ROOT": os.path.join(WORK, "workspace"),
        "SADDLE_HTTP_TIMEOUT_S": "2",
        "SADDLE_LOST_WAKE_S": "2",
        "SADDLE_STALE_S": "3",
        "SADDLE_DISPATCH_TICK_S": "1",
        # Never inherit a caller-poisoned LD_LIBRARY_PATH (e.g. one that
        # resolves the ancient /lib64 copies): empty makes the loader use
        # the binary's RUNPATH, which is the dev stack it was built
        # against.
        "LD_LIBRARY_PATH": "",
    })
    log_previous = os.path.join(WORK, "app.log.1")
    if os.path.exists(os.path.join(WORK, "app.log")):
        shutil.move(os.path.join(WORK, "app.log"), log_previous)
    log = open(os.path.join(WORK, "app.log"), "w")
    command = [APP, str(APP_PORT)]
    if os.environ.get("WEBSTACK_GDB"):
        # Debugging aid: run the app under gdb; a crash backtrace lands in
        # app.log. WEBSTACK_GDB=1 meson test ... to enable.
        command = ["gdb", "-batch", "-quiet",
                   "-ex", "handle SIGPIPE nostop noprint pass",
                   "-ex", "run", "-ex", "bt", "--args", APP, str(APP_PORT)]
    process = subprocess.Popen(command, cwd=WORK, env=env,
                               stdout=log, stderr=subprocess.STDOUT)
    outcome = wait_for(lambda: (app_ready() and "ready")
                       or (process.poll() is not None and "app-exited-%d" % process.poll())
                       or False, timeout=30, what="app ready")
    if outcome is None:
        raise SystemExit("app did not come up; see %s/app.log" % WORK)
    if str(outcome).startswith("app-exited-"):
        raise SystemExit("app exited during startup (%s); see %s/app.log"
                         % outcome, WORK)
    return process


def app_ready():
    try:
        status, _ = http("GET", "/login", timeout=3)
        return status == 200
    except (urllib.error.URLError, OSError):
        return False


def dump_app_stacks(pid):
    """Best-effort thread backtrace of the (possibly wedged) app."""
    try:
        fds = os.listdir("/proc/%d/fd" % pid)
        print("(app open fds at dump: %d)" % len(fds))
        kinds = {}
        for fd in fds[:1200]:
            try:
                target = os.readlink("/proc/%d/fd/%s" % (pid, fd))
            except OSError:
                continue
            key = target.split(":", 1)[0] if ":" in target else target
            kinds[key] = kinds.get(key, 0) + 1
        print("(fd kinds: %s)" % sorted(kinds.items(), key=lambda kv: -kv[1])[:6])
    except OSError:
        pass
    for command in (["eu-stack", "-p", str(pid)],
                    ["gdb", "-batch", "-quiet", "-p", str(pid),
                     "-ex", "thread apply all bt"]):
        try:
            out = subprocess.run(command, capture_output=True, timeout=60)
            text = out.stdout.decode(errors="replace") + out.stderr.decode(errors="replace")
            if text.strip():
                print("(app stacks via %s)\n%s" % (command[0], text[:12000]))
                return
        except (OSError, subprocess.SubprocessError):
            continue


# =============================================================================
# Fake push service + Web Push protocol helpers (plan 11 task E)
# =============================================================================

def b64url(raw):
    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode()


def b64url_dec(text):
    return base64.urlsafe_b64decode(text + "=" * ((4 - len(text) % 4) % 4))


class FakePushService:
    """In-process Web Push endpoint server: every subscription the tests
    register points at a path here (a distinct "browser endpoint").
    Responses are scriptable per path — a fixed status (the 410-prune
    fixture) or N transient 500s (the retry-next-tick fixture)."""

    def __init__(self):
        self.requests = []            # [{"path", "headers", "body", "at"}]
        self.fail_next = {}           # path -> remaining 500s
        self.status_for = {}          # path -> fixed status
        self.lock = threading.Lock()
        self.port = None
        self.server = None

    def serve(self):
        service = self

        class Handler(BaseHTTPRequestHandler):
            protocol_version = "HTTP/1.1"

            def log_message(self, fmt, *args):
                pass

            def do_POST(self):
                length = int(self.headers.get("Content-Length") or 0)
                body = self.rfile.read(length)
                status = 201
                if self.path in service.status_for:
                    status = service.status_for[self.path]
                elif service.fail_next.get(self.path, 0) > 0:
                    service.fail_next[self.path] -= 1
                    status = 500
                with service.lock:
                    service.requests.append({
                        "path": self.path,
                        "headers": {key.lower(): value
                                    for key, value in self.headers.items()},
                        "body": body,
                        "at": time.monotonic(),
                    })
                payload = b"{}"
                self.send_response(status)
                self.send_header("Content-Type", "application/json")
                self.send_header("Content-Length", str(len(payload)))
                self.end_headers()
                self.wfile.write(payload)

        self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
        self.port = self.server.server_address[1]
        threading.Thread(target=self.server.serve_forever, daemon=True).start()
        return self.server

    def url(self, path):
        return "http://127.0.0.1:%d%s" % (self.port, path)

    def hits(self, prefix):
        with self.lock:
            return [request for request in self.requests
                    if request["path"].startswith(prefix)]


def decrypt_webpush(request, client_private_key, client_public, auth_secret):
    """RFC 8291 decryption of one captured push: returns
    (record_size, payload dict) — the payload builder's exact output."""
    body = request["body"]
    salt = body[:16]
    record_size = int.from_bytes(body[16:20], "big")
    key_id_len = body[20]
    eph_public = body[21:21 + key_id_len]
    ciphertext = body[21 + key_id_len:]
    eph_point = ec.EllipticCurvePublicKey.from_encoded_point(ec.SECP256R1(),
                                                             eph_public)
    shared = client_private_key.exchange(ec.ECDH(), eph_point)
    info = b"WebPush: info\x00" + eph_public + client_public
    cek = HKDF(algorithm=hashes.SHA256(), length=16, salt=salt,
               info=info).derive(shared + auth_secret)
    nonce = HKDF(algorithm=hashes.SHA256(), length=12, salt=salt,
                 info=info).derive(shared + auth_secret)
    plaintext = AESGCM(cek).decrypt(nonce, ciphertext, None)
    return record_size, json.loads(plaintext[:-1])  # strip the 0x02 delimiter


def verify_vapid(request, vapid_public, vapid_private):
    """Verifies the ES256 VAPID JWT on one captured push; returns its
    claims dict (aud/exp/sub) — raises on a bad signature."""
    header = request["headers"].get("authorization", "")
    assert header.startswith("vapid t="), "no vapid authorization header"
    token_part = header[len("vapid t="):].split(",")[0]
    sent_key = header.split("k=")[1]
    head_b64, claims_b64, sig_b64 = token_part.split(".")
    signature = b64url_dec(sig_b64)
    key = ec.derive_private_key(
        int.from_bytes(b64url_dec(vapid_private), "big"), ec.SECP256R1())
    key.public_key().verify(
        encode_dss_signature(int.from_bytes(signature[:32], "big"),
                             int.from_bytes(signature[32:], "big")),
        (head_b64 + "." + claims_b64).encode(),
        ec.ECDSA(hashes.SHA256()))
    claims = json.loads(b64url_dec(claims_b64))
    assert sent_key == vapid_public, "the k= key is not the app's VAPID key"
    return claims


def action_notify(result):
    """The message of the first notify directive an action returned."""
    for directive in result:
        if directive.get("type") == "notify":
            return directive.get("message", "")
    return None


# =============================================================================
# Shard orchestration
# =============================================================================

class Ctx:
    """Per-bundle shared fixtures: the app process, the admin auth slot
    and the primary fake saddle (registered + token delivered)."""


def start_bundle(name):
    """Boots one shard's environment: scratch dir, database, ports, the
    primary fake daemon and the app, then the shared fixtures every
    bundle builds on (the demo project + the registered fake saddle).
    Scenario groups inside the bundle share this app+daemon instance
    and sequence their state as DB rows, never process restarts."""
    global APP_PORT, WEB, DAEMON_PORT
    bundle_env(name)
    fake = FakeSaddle()
    fake.serve()
    DAEMON_PORT = fake.port  # the actually-bound port (fallback-safe)

    # A rare framework-level crash (stack corruption in Statum slot
    # signing during the FIRST login — shared-library codegen bug, moved
    # twice already) kills the app before any test runs; a clean restart
    # (with fresh ports) sidesteps it. Only tolerated while nothing has
    # been checked yet.
    auth = None
    app = None
    for attempt in range(3):
        APP_PORT = free_port()
        WEB = "http://127.0.0.1:%d" % APP_PORT
        if os.path.exists(WORK):
            shutil.rmtree(WORK)
        os.makedirs(os.path.join(WORK, "workspace"))
        try:
            app = start_app()
            auth = login("admin", "adminpass123")
            break
        except (SystemExit, OSError, urllib.error.URLError):
            if passes > 0:
                raise
            if app is not None and app.poll() is None:
                app.terminate()
                try:
                    app.wait(timeout=5)
                except subprocess.TimeoutExpired:
                    app.kill()
            app = None
            print("(app died before login; restarting cleanly, attempt %d)" % (attempt + 1))
    if app is None or auth is None:
        raise SystemExit("app never came up; see %s/app.log" % WORK)

    state = slot_public(auth, "/projects", "projects")
    act(auth, state["create"], {
        "name": "demo",
        "git_url": "https://example.com/demo.git",
        "default_branch": "main",
    })
    saddles_state = slot_public(auth, "/saddles", "saddles")
    act(auth, saddles_state["register"], {"name": "fake-saddle",
                                          "host": "127.0.0.1:%d" % fake.port})
    registered = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
                  if s["name"] == "fake-saddle"]
    if wait_for(lambda: len(fake.register_log) > 0 or False,
                timeout=15, what="/register delivery") is None:
        raise SystemExit("fake saddle never received /register")

    ctx = Ctx()
    ctx.name = name
    ctx.app = app
    ctx.auth = auth
    ctx.fake = fake
    ctx.fake_id = int(registered[0]["id"])
    ctx.fake_token = registered[0]["token"]
    return ctx


# =============================================================================
# Bundle: pages — block presentation, diff URLs, computed display state,
# terminal rendering (rejected entries, ANSI SGR), task-page layout and
# the /poll verdict channel. Fixture-driven: tasks and sessions are
# inserted as rows; only the rejected-ride scenario dispatches.
# =============================================================================

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

    check(True, "login yields an auth slot")
    print("== plan blocks: every task's saddle plan parks for acceptance ==")
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority,"
                 " created_at)"
                 " values (1, 'Plan review task', 'working', 5,"
                 " strftime('%s','now'))")
    plan_task = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.execute("insert into sessions (task_id, saddle_id, started_at, ended_at, outcome)"
                 " values (?, ?, strftime('%s','now'), strftime('%s','now'), 'done')",
                 (plan_task, fake_id))
    conn.commit()
    session_id = conn.execute("select max(id) from sessions").fetchone()[0]
    conn.close()

    status, body = saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
                              {"type": "block", "session_id": session_id, "data": {
                                  "question": "Plan submitted for review",
                                  "type": "plan",
                                  "body": "# The plan\n\n1. Do it"}})
    check(status == 200 and body.get("block_id"), "plan block raised")
    status, _ = saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
                           {"type": "block", "session_id": session_id, "data": {
                               "question": "Which database?",
                               "type": "question",
                               "options": ["SQLite", "Postgres"],
                               "body": "# Storage choice"}})
    check(status == 200, "question block raised")

    detail = task_state(auth, plan_task)
    blocks = {b["block_type"]: b for b in detail["blocks"]}
    check(blocks["plan"].get("is_plan_review") is True,
          "plan block flagged as the plan review")
    check(blocks["plan"].get("show_body") is False,
          "pending plan review hides the duplicate body preview")
    check(blocks["question"].get("show_body") is True,
          "question block keeps its body preview")

    # Accept applies the submitted plan markdown to task.plan and stores
    # the stable approval wording as the answer (it becomes the resume
    # notice the daemon reads).
    act(auth, blocks["plan"]["answer_action"],
        {"answer": "Plan approved. Proceed with implementation."})
    detail = task_state(auth, plan_task)
    plan_block = [b for b in detail["blocks"] if b["block_type"] == "plan"][0]
    check(plan_block.get("show_body") is True,
          "answered plan review shows the body again (review section gone)")
    check(plan_block.get("answered") is True
          and plan_block.get("answer") == "Plan approved. Proceed with implementation.",
          "accept stores the stable approval wording")
    check(detail.get("plan", "").startswith("# The plan"),
          "accepting a plan applies its markdown to task.plan")

    print("== B1: highlight-to-comment quote dialog ==")
    status, page = http("GET", "/tasks/%d" % plan_task, timeout=15)
    check(status == 200 and "quote-comment-dialog" in page and "quote-text" in page,
          "task page carries the quote-comment dialog the popover opens")
    check('name="quote"' in page and 'name="return_to"' in page,
          "quote dialog posts the quote and return-to fields")
    status, page = http("GET", "/tasks/%d/terminal" % plan_task, timeout=15)
    check(status == 200 and "quote-comment-dialog" in page,
          "terminal page keeps its quote dialog")

    print("== B3: review section hides the duplicate body ==")
    status, body = saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
                              {"type": "block", "session_id": session_id, "data": {
                                  "question": "Review the diff",
                                  "type": "review",
                                  "body": "# The change\n\nReworks the flange module."}})
    check(status == 200 and body.get("block_id"), "review block raised")
    detail = task_state(auth, plan_task)
    review_block = [b for b in detail["blocks"] if b["block_type"] == "review"][0]
    check(detail.get("review_pending") is True,
          "review section shows while the review is unanswered")
    check(review_block.get("show_body") is False,
          "pending review block hides its duplicate body preview")
    check(review_block.get("body", "").startswith("# The change"),
          "review body itself still arrives in state (B2 counterpart)")
    act(auth, review_block["answer_action"],
        {"answer": "Review approved. Continue with the task."})
    detail = task_state(auth, plan_task)
    review_block = [b for b in detail["blocks"] if b["block_type"] == "review"][0]
    check(review_block.get("show_body") is True,
          "answered review shows the body again (banner gone)")

    print("== B2: question block body above options ==")
    # Contract side (A2 will restore bodies end-to-end): a question
    # block raised with a heredoc body carries it; the body preview
    # rides above the options (template order) and survives answering
    # alongside the selected answer.
    detail = task_state(auth, plan_task)
    question = [b for b in detail["blocks"] if b["block_type"] == "question"][0]
    check(question.get("body") == "# Storage choice"
          and question.get("show_body") is True,
          "question block keeps its body markdown with the preview on")
    act(auth, question["answer_action"], {"option": "SQLite"})
    detail = task_state(auth, plan_task)
    question = [b for b in detail["blocks"] if b["block_type"] == "question"][0]
    check(question.get("answered") is True and question.get("answer") == "SQLite"
          and question.get("body") == "# Storage choice"
          and question.get("show_body") is True,
          "answered question shows body + selected answer")

    print("== B5: compare-changes SSH -> HTTPS ==")
    def diff_url_for_project(git_url, suffix):
        state = slot_public(auth, "/projects", "projects")
        act(auth, state["create"], {"name": "diff-%s" % suffix, "git_url": git_url})
        project_id = int([p["id"] for p in slot_public(auth, "/projects", "projects")["projects"]
                          if p["name"] == "diff-%s" % suffix][0])
        conn = db()
        conn.execute("insert into tasks (project_id, title, status, task_branch,"
                     " branch_from, created_at) values (?, ?, 'working', 'feature/x',"
                     " 'main', strftime('%s','now'))", (project_id, "Diff %s" % suffix))
        conn.commit()
        task_id = conn.execute("select max(id) from tasks").fetchone()[0]
        conn.execute("insert into sessions (task_id, saddle_id, started_at, ended_at,"
                     " outcome) values (?, ?, strftime('%s','now'), strftime('%s','now'),"
                     " 'done')", (task_id, fake_id))
        conn.commit()
        session = conn.execute("select max(id) from sessions").fetchone()[0]
        conn.close()
        saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
                   {"type": "block", "session_id": session, "data": {
                       "question": "Review the diff", "type": "review"}})
        return task_state(auth, task_id).get("review_diff_url") or ""

    check(diff_url_for_project("ssh://fabrica.unitatem.net:222/clanker/clanker-saddle.git",
                               "fabrica")
          == "https://fabrica.unitatem.net/clanker/clanker-saddle/compare/main...feature/x",
          "ssh://git@host:port/owner/repo.git translates (port dropped)")
    check(diff_url_for_project("git@github.com:acme/widgets.git", "scp")
          == "https://github.com/acme/widgets/compare/main...feature/x",
          "git@host:owner/repo.git translates to the browse host")
    check(diff_url_for_project("https://example.com/plain/repo", "plain")
          == "https://example.com/plain/repo/compare/main...feature/x",
          "HTTPS URLs pass through unchanged")

    print("== B6: computed display state ==")
    # Draft tasks display as draft (plan 11) and never dispatch.
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1",
        "title": "Draft display task",
        "priority": "5",
    })
    draft_display = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                         if t["title"] == "Draft display task"][0])
    detail = task_state(auth, draft_display)
    check(detail["status"] == "draft" and detail.get("display") == "draft",
          "new tasks default to draft and display it")
    check(detail.get("draft") is True and detail.get("enqueue") is not None
          and detail.get("delete") is not None,
          "draft tasks expose the enqueue + delete actions")
    check(quiet_window(quiet_seconds(),
                       lambda: task_state(auth, draft_display)["status"] == "draft",
                       "draft task never dispatches"),
          "a draft task is never dispatched or woken for")

    # A suspended parent with subtasks still in draft shows the clear
    # enqueue-pending wording (plan 11).
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
                 " values (1, 'Draft-waiting parent', 'suspended', 5, strftime('%s','now'))")
    waiting_parent = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.execute("insert into tasks (project_id, title, status, priority, parent_task_id,"
                 " created_at) values (1, 'Draft child', 'draft', 5, ?,"
                 " strftime('%s','now'))", (waiting_parent,))
    conn.commit()
    conn.close()
    detail = task_state(auth, waiting_parent)
    check(detail.get("display") == "waiting on subtasks to be enqueued"
          and detail.get("subtask_draft") == 1,
          "suspended parent with draft children shows the enqueue-pending wording (%s)"
          % detail.get("display"))
    queue = [t for t in slot_public(auth, "/tasks", "tasks")["tasks"]
             if int(t["id"]) == waiting_parent][0]
    check(queue["display"] == "waiting on subtasks to be enqueued",
          "queue row carries the enqueue-pending display too")
    dash = slot_public(auth, "/", "dashboard")
    check(any(int(w["task_id"]) == waiting_parent and int(w["draft_subtasks"]) == 1
              for w in dash.get("waiting_drafts", [])),
          "dashboard data carries the waiting-on-drafts block")
    waiting_view = [w for w in dash.get("waiting_drafts", [])
                    if int(w["task_id"]) == waiting_parent][0]
    check(len(waiting_view.get("drafts", [])) == 1,
          "the waiting block links the draft subtask where Enqueue lives")
    status, home_page = http("GET", "/", timeout=15)
    check(status == 200 and "Waiting on subtasks to be enqueued" in home_page
          and "stm-attribute.href=\"'/tasks/' + waiting.task_id\"" in home_page
          and "stm-attribute.href=\"'/tasks/' + draft.id\"" in home_page,
          "the dashboard renders the waiting-on-drafts block with task-page links")

    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
                 " values (1, 'Parent', 'done', 5, strftime('%s','now'))")
    parent6 = conn.execute("select max(id) from tasks").fetchone()[0]
    # The post-mortem case: stored done WITH an unanswered review block.
    conn.execute("insert into sessions (task_id, saddle_id, started_at, ended_at, outcome)"
                 " values (?, ?, strftime('%s','now'), strftime('%s','now'), 'done')",
                 (parent6, fake_id))
    session6 = conn.execute("select max(id) from sessions").fetchone()[0]
    conn.commit()
    conn.close()
    saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
               {"type": "block", "session_id": session6, "data": {
                   "question": "Review the final diff", "type": "review"}})

    detail = task_state(auth, parent6)
    check(detail.get("display") == "review requested" and detail.get("terminal") is True,
          "unanswered review overrides the stored done status (terminal flag keeps stored truth)")
    queue = [t for t in slot_public(auth, "/tasks", "tasks")["tasks"]
             if int(t["id"]) == parent6][0]
    check(queue["display"] == "review requested",
          "queue row shows the computed review-requested display")

    # Delegation: with the review answered, open subtasks delegate.
    conn = db()
    conn.execute("update task_blocks set answered_at = strftime('%s','now'),"
                 " answer = 'Human approved the review' where task_id = ?", (parent6,))
    conn.execute("insert into tasks (project_id, title, status, priority, parent_task_id,"
                 " created_at) values (1, 'Sub A', 'working', 5, ?, strftime('%s','now'))",
                 (parent6,))
    conn.execute("insert into tasks (project_id, title, status, priority, parent_task_id,"
                 " created_at) values (1, 'Sub B', 'done', 5, ?, strftime('%s','now'))",
                 (parent6,))
    conn.commit()
    conn.close()
    detail = task_state(auth, parent6)
    check(detail.get("display") == "delegating 1/2" and detail.get("subtask_open") == 1,
          "open subtasks show the delegating breakdown (%s)" % detail.get("display"))

    # Flags track the stored status once the display falls through
    # (subtasks terminal first — delegating leads while any are open).
    conn = db()
    conn.execute("update tasks set status = 'failed', completed_at = strftime('%s','now')"
                 " where id = ?", (parent6,))
    conn.execute("update tasks set status = 'failed', completed_at = strftime('%s','now')"
                 " where title = 'Sub A'")
    conn.commit()
    conn.close()
    detail = task_state(auth, parent6)
    check(detail.get("display") == "failed" and detail.get("failed") is True
          and detail.get("terminal") is True,
          "failed task with answered review and terminal subtasks shows failed + terminal flags")

    print("== rejected entries (non-fatal) ==")
    # Ingest API: a rejected entry stores on the session and streams
    # through SSE; the task is untouched.
    status, _ = saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
                           {"type": "output", "session_id": session_id, "data": {
                               "command": "bash -n script.sh",
                               "output": "syntax error near line 3 (recoverable)",
                               "exit_code": 1,
                               "type": "rejected"}})
    check(status == 200, "rejected entry accepted by the ingest API")
    request = urllib.request.Request(WEB + "/terminal-stream/%d" % session_id)
    with urllib.request.urlopen(request, timeout=15) as stream:
        rejected_event = None
        deadline = time.monotonic() + 8
        while time.monotonic() < deadline and rejected_event is None:
            line = stream.readline().decode().strip()
            if not line.startswith("data: ") or line == "data: ping":
                continue
            event = json.loads(line[6:])
            if event.get("type") == "rejected":
                rejected_event = event
        check(rejected_event is not None
              and "syntax error near line 3" in (rejected_event.get("output") or ""),
              "terminal stream carries rejected entries with their reason")

    # The page template renders them with the distinct non-fatal style.
    status, page = http("GET", "/tasks/%d/terminal" % plan_task, timeout=15)
    check(status == 200 and "term-rejected" in page,
          "terminal page renders rejected entries in their own style")

    print("== B8: ANSI SGR in the terminal view ==")
    # Byte-exact SGR delivery: the daemon's coloured chunks flow through
    # the store + SSE untouched (the client renderer styles them).
    coloured = "\x1b[1;32mbold green\x1b[0m then \x1b[38;5;196m256-red\x1b[0m\n"
    conn = db()
    conn.execute("insert into session_raw_log (session_id, seq, chunk)"
                 " values (?, 900, ?)", (session_id, coloured))
    conn.commit()
    conn.close()
    request = urllib.request.Request(WEB + "/terminal-stream/%d" % session_id)
    with urllib.request.urlopen(request, timeout=15) as stream:
        coloured_event = None
        deadline = time.monotonic() + 8
        while time.monotonic() < deadline and coloured_event is None:
            line = stream.readline().decode().strip()
            if not line.startswith("data: ") or line == "data: ping":
                continue
            event = json.loads(line[6:])
            if event.get("type") == "raw" and "\x1b[1;32m" in (event.get("text") or ""):
                coloured_event = event
        check(coloured_event is not None
              and "\x1b[38;5;196m" in coloured_event["text"],
              "SGR sequences stream byte-exact to the client renderer")
    status, page = http("GET", "/tasks/%d/terminal" % plan_task, timeout=15)
    check(status == 200 and "makeAnsiRenderer" in page and "colour256" in page,
          "terminal page embeds the ANSI renderer (SGR + 256-colour palette)")

    # Dispatch path: a rejected entry in a round's results push flows
    # into the audit trail without failing (or finishing) the task.
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Rejected ride",
        "priority": "2",
    })
    rejected_task = task_id_tmp
    check(wait_for(lambda: task_state(auth, rejected_task).get("status") == "working"
                   or False, timeout=60, what="rejected-ride working") is not None,
          "rejected-ride task reaches working")
    fake.queue_output("rejected", "echo 'too long' | saddle",
                      "max_tokens exceeded (recoverable)")
    check(wait_task_done(auth, rejected_task, timeout=60) is not None,
          "task completes normally past rejected entries")
    conn = db()
    stored = conn.execute(
        "select count(*) from session_commands sc join sessions s on s.id = sc.session_id"
        " where s.task_id = ? and sc.type = 'rejected'", (rejected_task,)).fetchone()[0]
    conn.close()
    check(stored >= 1,
          "rejected entries stored in the audit trail (%d rows)" % stored)
    check(task_state(auth, rejected_task)["status"] == "done"
          and "fake daemon finished" in (task_state(auth, rejected_task).get("summary") or ""),
          "rejected entries never fail the task")

    print("== plan 11 task D: task page — progress first, 50/50 grid, expanders ==")
    detail = task_state(auth, plan_task)
    block_ids = [int(b["id"]) for b in detail.get("blocks", [])]
    check(len(block_ids) >= 3 and block_ids == sorted(block_ids, reverse=True),
          "the blocks section orders newest-first via the sealed GET state (%s)"
          % block_ids)
    status, page = http("GET", "/tasks/%d" % plan_task, timeout=15)
    # Plan 11 task D: the progress feed sits ABOVE the status card;
    # status + description share a 50/50 grid; the plan detail hides
    # behind its expander below it.
    check(status == 200 and "grid-2" in page
          and page.index("<h3>Progress</h3>") < page.index("status-head"),
          "the progress feed renders above the status card")
    check(page.index("grid-2") < page.index("status-head")
          and page.index("status-head") < page.index("<h3>Description</h3>"),
          "status and description sit side by side inside the grid (status first)")
    check("<h3>Description</h3>" in page
          and page.index("<h3>Description</h3>") < page.index("plan-expander"),
          "the plan detail hides behind its expander below the grid")
    # Answered blocks fold behind the history expander (this fixture's
    # three blocks are all answered by now).
    check("blocks-history" in page and "Answered blocks" in page,
          "answered blocks hide behind the blocks history expander")
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority,"
                 " created_at) values (1, 'Progress layout task', 'working', 5,"
                 " strftime('%s','now','-120 seconds'))")
    layout_task = conn.execute("select max(id) from tasks").fetchone()[0]
    # backdated so the dashboard-timeline fixtures below stay the
    # newest events
    conn.execute("insert into task_progress (task_id, summary, created_at)"
                 " values (?, 'first update', strftime('%s','now','-90 seconds'))",
                 (layout_task,))
    conn.execute("insert into task_progress (task_id, summary, created_at)"
                 " values (?, 'second update', strftime('%s','now','-89 seconds'))",
                 (layout_task,))
    conn.commit()
    conn.close()
    status, history_page = http("GET", "/tasks/%d" % layout_task, timeout=15)
    check(status == 200
          and history_page.index("progress-latest") < history_page.index("progress-history"),
          "the progress timeline renders newest-first (latest above the history)")
    # The edit expander sits UNDER the "Show history" expander.
    check(history_page.index("Show history") < history_page.index("edit-expander"),
          "the edit expander sits under the Show history expander")

    print("== heredoc completion reports: markdown card + timeline snippet ==")
    # The daemon's `cat << 'EOF' | saddle done` stores a markdown
    # summary (the failure path stores the same field); the task page
    # renders it as a full-width completion-report card directly below
    # the status/description grid — markdown via stm-markdown
    # (XSS-safe), "Completion report" when done, "Failure summary"
    # when failed, no card when the summary is empty. The dashboard
    # timeline snippets the summary's FIRST line.
    report_md = ("## Flange wired and tested\n\n"
                 "Ran `meson test` over the full matrix.\n\n"
                 "```sh\nmeson test -C builddir\n```\n\n"
                 "<b>not html</b> & escaped\n")
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority,"
                 " summary, started_at, completed_at, created_at)"
                 " values (1, 'Report done task', 'done', 5, ?,"
                 " strftime('%s','now'), strftime('%s','now'),"
                 " strftime('%s','now'))", (report_md,))
    report_done = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.execute("insert into tasks (project_id, title, status, priority,"
                 " summary, started_at, completed_at, created_at)"
                 " values (1, 'Report failed task', 'failed', 5,"
                 " 'Boom: tests failed\n\nsecond line carries the detail',"
                 " strftime('%s','now'), strftime('%s','now'),"
                 " strftime('%s','now'))")
    report_failed = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.execute("insert into tasks (project_id, title, status, priority,"
                 " started_at, completed_at, created_at)"
                 " values (1, 'Report empty task', 'done', 5,"
                 " strftime('%s','now'), strftime('%s','now'),"
                 " strftime('%s','now'))")
    report_empty = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.commit()
    conn.close()

    done_state = task_state(auth, report_done)
    check(done_state["report_title"] == "Completion report"
          and "## Flange wired" in (done_state.get("summary") or ""),
          "a done task's summary rides the report card state")
    check(task_state(auth, report_failed)["report_title"] == "Failure summary",
          "a failed task's card heading reads Failure summary")
    empty_state = task_state(auth, report_empty)
    check(empty_state["report_title"] == "" and not (empty_state.get("summary") or ""),
          "an empty summary renders no report card (no heading)")
    check(task_state(auth, layout_task)["report_title"] == "",
          "non-terminal tasks never show the report card")

    status, page = http("GET", "/tasks/%d" % report_done, timeout=15)
    check(status == 200 and "completion-report" in page
          and page.index("grid-2") < page.index("completion-report")
          < page.index("edit-expander"),
          "the completion-report card sits below the status/description grid, full width")
    card = page[page.index("completion-report"):page.index("edit-expander")]
    check('stm-markdown="task.summary"' in card
          and 'stm-text="task.report_title"' in card,
          "the card renders the summary as markdown under its state-driven heading")
    check("<strong>Summary:</strong>" not in page,
          "the old plain-text summary line is gone (the card owns the summary)")

    # The markdown itself renders client-side: drive the SHIPPED
    # stm-markdown renderer (statum.js) on the fixture summary via
    # node — headings and code come out as HTML, no markdown syntax
    # leaks, raw HTML stays escaped.
    status, statum_js = http("GET", "/_statum/resource/statum.js", timeout=15)
    skip = None if status == 200 else "statum.js not served"
    parts = []
    if skip is None:
        def extract_md(name, source):
            start = source.index("function %s(" % name)
            depth = 0
            index = source.index("{", start)
            end = index
            while True:
                if source[end] == "{":
                    depth += 1
                elif source[end] == "}":
                    depth -= 1
                    if depth == 0:
                        break
                end += 1
            return source[start:end + 1]
        try:
            parts = [extract_md(name, statum_js)
                     for name in ("safeMarkdownUrl", "mdLinks", "mdInline", "renderMarkdown")]
        except ValueError:
            skip = "renderer functions missing"
    rendered = ""
    if skip is None:
        try:
            result = subprocess.run(
                ["node", "-e", "%s\n%s\n%s\n%s\nconsole.log(renderMarkdown(process.argv[1]));"
                 % tuple(parts), report_md],
                capture_output=True, text=True, timeout=20)
            rendered = result.stdout
        except (OSError, subprocess.SubprocessError):
            skip = "node unavailable"
    if skip is not None:
        check(True, "stm-markdown renderer present (%s, deep check skipped)" % skip)
    else:
        check("<h2>Flange wired and tested</h2>" in rendered
              and "<pre><code" in rendered and "<code>meson test</code>" in rendered,
              "the report summary renders headings and code through stm-markdown")
        check("## Flange" not in rendered and "```sh" not in rendered
              and "`meson test`" not in rendered,
              "no raw markdown syntax leaks into the rendered report")
        check("<b>not html</b>" not in rendered and "&lt;b&gt;not html&lt;/b&gt;" in rendered
              and "&amp; escaped" in rendered,
              "raw HTML in the summary stays escaped (XSS-safe)")

    # The dashboard timeline: completion/failure events snippet the
    # summary's first line, falling back to the bare verdict text.
    dash = slot_public(auth, "/", "dashboard")
    done_event = next((e for e in dash.get("activity", [])
                       if e["kind"] == "completed" and int(e["task_id"]) == report_done), None)
    check(done_event is not None
          and done_event["text"] == "Task completed: ## Flange wired and tested",
          "the timeline completion event snippets the summary's first line")
    failed_event = next((e for e in dash.get("activity", [])
                         if e["kind"] == "failed" and int(e["task_id"]) == report_failed), None)
    check(failed_event is not None
          and failed_event["text"] == "Task failed: Boom: tests failed",
          "the timeline failure event snippets the summary's first line")
    empty_event = next((e for e in dash.get("activity", [])
                        if e["kind"] == "completed" and int(e["task_id"]) == report_empty), None)
    check(empty_event is not None and empty_event["text"] == "Task completed",
          "a missing summary falls back to the bare timeline text")

    print("== plan 11 task D: block split + queue-screen layout ==")
    # One open and one answered question: the open block renders in
    # full, the answered one folds behind the history expander.
    conn = db()
    conn.execute("insert into sessions (task_id, saddle_id, started_at)"
                 " values (?, ?, strftime('%s','now'))", (layout_task, fake_id))
    conn.commit()
    split_session = conn.execute("select max(id) from sessions").fetchone()[0]
    conn.close()
    saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
               {"type": "block", "session_id": split_session, "data": {
                   "question": "Open question for layout?", "type": "question"}})
    saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
               {"type": "block", "session_id": split_session, "data": {
                   "question": "Answered question for layout?", "type": "question"}})
    answered_block = [b for b in task_state(auth, layout_task)["blocks"]
                      if b["question"] == "Answered question for layout?"][0]
    act(auth, answered_block["answer_action"], {"answer": "Yes"})
    detail = task_state(auth, layout_task)
    check(len(detail.get("open_blocks", [])) == 1
          and len(detail.get("answered_blocks", [])) == 1
          and detail["open_blocks"][0]["question"] == "Open question for layout?"
          and detail["answered_blocks"][0]["question"] == "Answered question for layout?",
          "the task slot splits blocks into open (current) and answered lists")
    status, page = http("GET", "/tasks/%d" % layout_task, timeout=15)
    check(status == 200
          and page.index('stm-for-block-in="task.open_blocks"') < page.index("blocks-history")
          < page.index('stm-for-block-in="task.answered_blocks"'),
          "current blocks render in full above the answered-behind-expander history")
    check(page.index("status-head") < page.index("Cancel task"),
          "the lifecycle buttons ride the status card inside the grid")
    status, queue_page = http("GET", "/tasks", timeout=15)
    check(status == 200
          and queue_page.index("new-task-expander") < queue_page.index("filter-status")
          < queue_page.index("<table"),
          "the New Task expander sits above the filter controls and the table")

    print("== yeehaw: ingestion, festive card, badge, timeline ==")
    # `saddle yeehaw <message>` arrives as a round-results entry with the
    # message in the output field: it stores like any entry type, earns
    # the task its 🤠 badge (derived, nothing new stored), renders as the
    # task page's festive card (last 3, newest first, "+N more") and
    # surfaces on the dashboard timeline as its own event kind.
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
                 " values (1, 'Yeehaw task', 'working', 5, strftime('%s','now'))")
    yeehaw_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'))", (yeehaw_task, fake_id))
    yeehaw_session = conn.execute("select max(id) from sessions").fetchone()[0]
    conn.commit()
    conn.close()
    now_epoch = int(time.time())

    def push_yeehaw(entry_id, message, offset=0):
        """One celebration riding a round-results push, timestamped
        `offset` seconds from now (distinct epochs keep the timeline's
        newest-first ordering observable)."""
        return saddle_api("POST", "/api/saddle/results", fake_token, {
            "session_id": yeehaw_session, "state": "working",
            "entries": [{
                "id": entry_id, "command": "saddle yeehaw", "output": message,
                "exit_code": 0, "type": "yeehaw",
                "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ",
                                           time.gmtime(now_epoch + offset)),
            }]})[0]

    def yeehaw_rows():
        conn = db()
        try:
            return conn.execute(
                "select count(*) from session_commands where session_id = ?"
                " and type = 'yeehaw'", (yeehaw_session,)).fetchone()[0]
        finally:
            conn.close()

    check(push_yeehaw(9101, "Shipped the flange") == 200,
          "yeehaw entries ingest through the results push like any entry type")
    check(yeehaw_rows() == 1,
          "the yeehaw entry stores as a session_commands row (message in output)")
    check(push_yeehaw(9101, "Shipped the flange") == 200 and yeehaw_rows() == 1,
          "a redelivered yeehaw push is idempotent on the daemon entry id")
    check(task_state(auth, yeehaw_task)["status"] == "working",
          "a yeehaw entry never derails the task state")

    push_yeehaw(9102, "Second hoedown", 1)
    push_yeehaw(9103, "Third hoedown\nsecond line carries the detail", 2)
    push_yeehaw(9104, "Fourth hoedown", 3)
    push_yeehaw(9105, "Fifth hoedown", 4)
    detail = task_state(auth, yeehaw_task)
    check(detail.get("yeehaw") is True and detail.get("yeehaw_total") == 5,
          "the detail state flags the badge presence and the uncapped total")
    check([y["message"] for y in detail.get("yeehaws", [])]
          == ["Fifth hoedown", "Fourth hoedown",
              "Third hoedown\nsecond line carries the detail"],
          "the festive card keeps the last 3 yeehaws, newest first (full messages)")
    check(detail.get("yeehaw_more") == 2,
          "the '+N more' hint counts the capped tail")
    status, page = http("GET", "/tasks/%d" % yeehaw_task, timeout=15)
    # The template compiler HTML-escapes the emoji (&#129312; = 🤠).
    check(status == 200 and "yeehaw-card" in page and "&#129312; Yeehaw!" in page,
          "the task page renders the festive yeehaw card")
    check(page.index("<h3>Progress</h3>") < page.index("yeehaw-card")
          < page.index("status-head"),
          "the card sits between the progress feed and the status grid")
    check('stm-text="task.yeehaw_more"' in page and "more yeehaw" in page,
          "the card carries the +N more hint")
    check("badge-yeehaw" in page and "status-corner" in page,
          "the status card corner carries the earned 🤠 badge")
    plain_detail = task_state(auth, layout_task)
    check(plain_detail.get("yeehaw") is False and plain_detail.get("yeehaw_total") == 0
          and plain_detail.get("yeehaws", []) == [],
          "tasks without yeehaw entries carry no card state")
    status, plain_page = http("GET", "/tasks/%d" % layout_task, timeout=15)
    check(status == 200 and 'section class="card yeehaw-card" stm-if="task.yeehaw"' in plain_page,
          "the card template only renders under state gating (stm-if task.yeehaw)")

    queue_rows = slot_public(auth, "/tasks", "tasks")["tasks"]
    check([t for t in queue_rows if int(t["id"]) == yeehaw_task][0].get("has_yeehaw") is True,
          "queue rows badge yeehaw tasks")
    check([t for t in queue_rows if int(t["id"]) == layout_task][0].get("has_yeehaw") is False,
          "queue rows leave yeehaw-less tasks bare")
    status, queue_page = http("GET", "/tasks", timeout=15)
    check(status == 200 and 'stm-if="task.has_yeehaw"' in queue_page,
          "the queue template renders the row badge")

    dash = slot_public(auth, "/", "dashboard")
    yeehaw_events = [e for e in dash.get("activity", []) if e["kind"] == "yeehaw"]
    check(len(yeehaw_events) == 5
          and all(int(e["task_id"]) == yeehaw_task for e in yeehaw_events),
          "every yeehaw entry surfaces as its own timeline event kind")
    check(yeehaw_events[0]["text"] == "🤖 yeehaw'd: Fifth hoedown",
          "the newest yeehaw event leads the timeline rows")
    check(all("second line" not in e["text"] for e in yeehaw_events),
          "multiline yeehaw messages snippet to their first line")
    referenced = [e for e in dash.get("activity", []) if int(e["task_id"]) == yeehaw_task]
    check(len(referenced) > 0 and all(e.get("task_yeehaw") is True for e in referenced),
          "dashboard timeline task references carry the badge flag")
    plain_refs = [e for e in dash.get("activity", []) if int(e["task_id"]) == layout_task]
    check(len(plain_refs) > 0 and all(e.get("task_yeehaw") is False for e in plain_refs),
          "references to yeehaw-less tasks stay bare")
    status, home_page = http("GET", "/", timeout=15)
    check(status == 200 and 'stm-if="item.task_yeehaw"' in home_page
          and "badge-yeehaw" in home_page,
          "the dashboard renders the badge on timeline task references")

    print("== R10: branch_advanced on the /poll verdict channel ==")
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority,"
                 " created_at) values (1, 'Poll grandparent', 'working', 5,"
                 " strftime('%s','now','-300 seconds'))")
    poll_grand = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.execute("insert into tasks (project_id, title, status, priority,"
                 " parent_task_id, created_at) values (1, 'Poll answer parent',"
                 " 'blocked', 5, ?, strftime('%s','now','-120 seconds'))",
                 (poll_grand,))
    poll_parent = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.execute("insert into sessions (task_id, saddle_id, started_at)"
                 " values (?, ?, strftime('%s','now','-120 seconds'))",
                 (poll_parent, fake_id))
    conn.execute("insert into task_blocks (task_id, question, type, answer,"
                 " answered_at, created_at) values (?, 'Which DB?', 'question',"
                 " 'SQLite', strftime('%s','now','-60 seconds'),"
                 " strftime('%s','now','-100 seconds'))", (poll_parent,))
    conn.execute("update saddles set status = 'busy', current_task_id = ?,"
                 " allocated_at = strftime('%s','now') where name = 'fake-saddle'",
                 (poll_parent,))
    conn.commit()
    conn.close()
    status, body = saddle_api("GET", "/api/saddle/%d/poll" % fake_id, fake_token)
    check(status == 200 and len(body.get("answers", [])) == 1
          and body["answers"][0].get("branch_advanced") is False,
          "/poll verdicts carry branch_advanced=false with no siblings")

    # A sibling under the same parent completed after the park marker.
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority,"
                 " parent_task_id, summary, completed_at, created_at)"
                 " values (1, 'Poll sibling', 'done', 5, ?, 'sibling finished',"
                 " strftime('%s','now'), strftime('%s','now','-200 seconds'))",
                 (poll_grand,))
    conn.execute("insert into task_blocks (task_id, question, type, answer,"
                 " answered_at, created_at) values (?, 'Second question?',"
                 " 'question', 'Yes', strftime('%s','now'),"
                 " strftime('%s','now','-90 seconds'))", (poll_parent,))
    conn.commit()
    conn.close()
    status, body = saddle_api("GET", "/api/saddle/%d/poll" % fake_id, fake_token)
    check(status == 200 and len(body.get("answers", [])) == 1
          and body["answers"][0].get("branch_advanced") is True,
          "sibling completion after park flips branch_advanced to true")

    # Release the staged fixture: cancel it so its busy@task row does
    # not linger into the sections below.
    act(auth, task_state(auth, poll_parent)["cancel"], {})


# =============================================================================
# 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 bundle_dispatch(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")
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Rules dispatch",
        "description": "Effective rule set rides the /start payload.",
        "priority": "2",
    })
    rule_task = task_id_tmp
    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"))})
    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 chunks -> raw log + terminal view ==")
    # 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, chunk from session_raw_log where session_id = ?"
                        " order by seq", (raw_session,)).fetchall()
    conn.close()
    check(len(rows) >= 3, "session_raw_log captured the pushed chunks (%d)" % len(rows))
    seqs = [r[0] for r in rows]
    check(seqs == sorted(seqs) and len(set(seqs)) == len(seqs),
          "pushed chunks stored once, seq strictly increasing")
    text = "".join(r[1] for r in rows)
    check(fake.PROMPT in text and "echo hello" in text and "hello" in text,
          "pushed raw log keeps the prompt and echoed output byte-exact")
    check("working-heartbeat" in text,
          "the flush-timer heartbeat chunks keep arriving mid-round")

    terminal = slot_public(auth, "/tasks/%d/terminal" % task_id_of("Char cap dispatch"),
                           "terminal")
    check(terminal.get("has_raw_log") is True and int(terminal["session_id"]) == raw_session,
          "terminal view selects raw rendering for the raw session")

    # Old sessions (typed entries, no raw rows) keep the fallback view
    # (a dedicated fixture task: its only session is the legacy one,
    # so the terminal view cannot prefer a dispatched raw session).
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
                 " values (1, 'Legacy terminal task', 'working', 5, strftime('%s','now'))")
    legacy_task = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.execute("insert into sessions (task_id, saddle_id, started_at, ended_at, outcome)"
                 " values (?, ?, strftime('%s','now'), strftime('%s','now'), 'done')",
                 (legacy_task, fake_id))
    conn.commit()
    legacy_session = conn.execute("select max(id) from sessions").fetchone()[0]
    conn.execute("insert into session_commands (session_id, command_text, output_text,"
                 " type, started_at) values (?, 'ls', 'old-output', 'command',"
                 " strftime('%s','now'))", (legacy_session,))
    conn.commit()
    conn.close()
    legacy = slot_public(auth, "/tasks/%d/terminal" % legacy_task, "terminal")
    check(legacy.get("has_raw_log") is False,
          "legacy session falls back to the entry-type terminal view")

    request = urllib.request.Request(WEB + "/terminal-stream/%d" % raw_session)
    with urllib.request.urlopen(request, timeout=15) as stream:
        raw_events = []
        deadline = time.monotonic() + 8
        while time.monotonic() < deadline and len(raw_events) < 3:
            line = stream.readline().decode().strip()
            if not line.startswith("data: ") or line == "data: ping":
                continue
            event = json.loads(line[6:])
            if event.get("type") == "raw":
                raw_events.append(event)
        check(len(raw_events) >= 3 and raw_events[0]["text"] == fake.PROMPT
              and raw_events[0]["seq"] < raw_events[1]["seq"],
              "terminal stream serves pushed chunks in seq order (prompt first)")

    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, "chunks": []})
    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, "chunks": []})
    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 chunk idempotency: re-pushing a batch (the outbox's
    # at-least-once redelivery after a lost response) consumes the
    # duplicates and stores only the tail.
    conn = db()
    before = conn.execute("select count(*) from session_raw_log where session_id = ?",
                          (raw_session,)).fetchone()[0]
    conn.close()
    redelivery = [{"seq": r[0], "text": r[1]} for r in rows[:3]]
    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
                              {"session_id": raw_session, "chunks": 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, "text": "late-flush\r\n"},
             {"seq": 9001, "text": ""}]  # truly empty text: no row
    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
                              {"session_id": raw_session, "chunks": fresh})
    check(status == 200 and int(body.get("stored") or 0) == 1,
          "a partial-redelivery batch stores only the new chunks")
    conn = db()
    after = conn.execute("select count(*) from session_raw_log where session_id = ?",
                         (raw_session,)).fetchone()[0]
    empty_row = conn.execute(
        "select count(*) from session_raw_log where session_id = ? and seq = 9001",
        (raw_session,)).fetchone()[0]
    conn.close()
    check(after == before + 1 and empty_row == 0,
          "redelivered chunks never duplicate rows; empty text stores nothing")

    # Pushed entries: today's OutputEntry shape with the daemon's
    # global entry ids — a redelivered results push stores its
    # entries exactly once.
    def command_rows(session):
        conn = db()
        try:
            return conn.execute(
                "select command_text, type from session_commands"
                " where session_id = ? order by id", (session,)).fetchall()
        finally:
            conn.close()

    entries = [{"id": 8101, "command": "sed -n 1p README.md", "output": "# demo",
                "exit_code": 0, "type": "command", "timestamp": "2026-09-06T00:00:00Z"}]
    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                           {"session_id": raw_session, "state": "working",
                            "entries": entries})
    check(status == 200, "entries push accepted")
    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                           {"session_id": raw_session, "state": "working",
                            "entries": entries})
    check(status == 200, "redelivered entries push accepted (at-least-once)")
    stored = command_rows(raw_session)
    check(len([r for r in stored if r[0] == "sed -n 1p README.md"]) == 1,
          "redelivered entries store exactly once (entry-id cursor)")

    # 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"]))})
    detail = task_state(auth, gpu_task)
    check([c["name"] for c in detail["capabilities"]] == ["gpu"],
          "task page lists the required capability")
    # The only idle saddle lacks the tag, so the queue flags the wait
    # (read before any other task claims the idle saddle).
    queue = [t for t in slot_public(auth, "/tasks", "tasks")["tasks"]
             if int(t["id"]) == gpu_task][0]
    check(queue["status"] == "queued" and queue["capabilities"] == ["gpu"]
          and queue.get("waiting_for_saddle") is True,
          "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=20,
                   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")

    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=60, 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: summary moves into a comment and clears, 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(any("bash -n rejected" in c["body"] for c in detail["comments"]),
          "failure explanation kept as a bot comment")
    check(any(c["body"] == "Task retried by user" for c in detail["comments"]),
          "retry lands its audit comment")
    check(len(detail["comments"]) >= comments_before + 1,
          "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=60,
                   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(any("restored the session context" in p["summary"]
              for p in detail.get("progress", [])),
          "daemon resume reported on the first status after the restore")
    check(len(detail["progress"]) > progress_before,
          "resume note appended to the progress timeline")

    # 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"
              for c in detail["comments"]),
          "manual resume lands the interruption comment")
    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")


# =============================================================================
# Bundle: subtasks — suspension parking, the subtask status push (with
# its best-effort retry), the synchronous subtask callback, and the
# settling waits for the parked parents this shard created.
# =============================================================================

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

    # A hold tag nothing carries: R3's second child stays queued (a
    # stable parked session for the failed-delivery leg) until released.
    saddles_state = slot_public(auth, "/saddles", "saddles")
    act(auth, saddles_state["create_capability"], {"name": "gpu"})

    print("== B7: suspended state + per-subtask resume ==")
    fake.suspend_on_title = "Suspend parent"
    pushes_before = len(fake.start_payloads)
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Suspend parent",
        "priority": "2",
    })
    parent7 = task_id_tmp
    check(wait_for(lambda: task_state(auth, parent7).get("status") == "suspended"
                   or False, timeout=60, what="parent parks as suspended") is not None,
          "daemon 'suspended' state parks the task")
    detail = task_state(auth, parent7)
    check(detail.get("display") == "waiting on subtasks",
          "suspended task displays as waiting on subtasks")
    check(any(s["status"] == "idle"
              for s in slot_public(auth, "/saddles", "saddles")["saddles"]),
          "suspension frees the saddle")
    first_push = fake.start_payloads[pushes_before]
    check(first_push.get("subtasks") == [],
          "the /start payload carries the subtasks array (empty here)")

    # Two subtask events land while parked (staged, not dispatched):
    # both must coalesce into one resume with a full report.
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority, parent_task_id,"
                 " summary, completed_at, created_at) values (1, 'Child one', 'done', 5, ?,"
                 " 'built and verified the flange', strftime('%s','now','+1 second'),"
                 " strftime('%s','now'))", (parent7,))
    conn.execute("insert into tasks (project_id, title, status, priority, parent_task_id,"
                 " summary, completed_at, created_at) values (1, 'Child two', 'failed', 5, ?,"
                 " 'the harness refused to start', strftime('%s','now','+1 second'),"
                 " strftime('%s','now'))", (parent7,))
    conn.commit()
    conn.close()

    def parent_push_count():
        return len([p for p in fake.start_payloads
                    if int(p.get("task", {}).get("id") or 0) == parent7])

    check(wait_for(lambda: parent_push_count() >= 2 or False,
                   timeout=45, what="suspended parent resumes") is not None,
          "subtask events resume the suspended parent")
    parent_pushes = [p for p in fake.start_payloads
                     if int(p.get("task", {}).get("id") or 0) == parent7]
    check(len(parent_pushes) >= 2,
          "the parent's resume is a fresh /start for the same task")
    resume_push = parent_pushes[-1]
    notice = resume_push.get("resume_notice") or ""
    check("Child one" in notice and "completed" in notice
          and "built and verified the flange" in notice,
          "resume notice reports the finished subtask with its summary")
    check("Child two" in notice and "failed" in notice
          and "the harness refused to start" in notice,
          "coalesced: the second event rides the same resume notice")
    subtask_entries = {(s["title"], s["status"]) for s in resume_push.get("subtasks", [])}
    check(("Child one", "done") in subtask_entries
          and ("Child two", "failed") in subtask_entries,
          "subtasks array carries the children's states at dispatch time")
    # The resumed arc surfaces on a /status read while the resumed
    # round still runs — the operator's manual refresh in the push
    # world.
    act(auth, saddle_card(auth, "fake-saddle")["refresh"], {})
    check(wait_task_done(auth, parent7, timeout=60) is not None,
          "resumed parent completes")
    detail = task_state(auth, parent7)
    check(any("restored the session context" in p["summary"]
              for p in detail.get("progress", [])),
          "resumed flag reported on the first status after the restore")

    print("== R3 B1: subtask status push ==")
    # A parked (suspended) parent gives delivery a stable open session.
    fake.suspend_on_title = "Push parent"
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Push parent",
        "priority": "2",
    })
    push_parent = task_id_tmp
    check(wait_for(lambda: task_state(auth, push_parent).get("status") == "suspended"
                   or False, timeout=60, what="push parent parks") is not None,
          "push parent parks with an open session")

    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Push child",
        "parent_task_id": str(push_parent),
    })
    push_child = task_id_tmp

    def events_for(subtask_id):
        return [e for e in fake.subtask_status_events
                if int(e.get("subtask_id") or 0) == subtask_id]

    check(wait_for(lambda: len(events_for(push_child)) > 0 or False,
                   timeout=30, what="queued subtask event") is not None,
          "subtask creation pushes a queued status event")
    first_event = events_for(push_child)[0]
    check(int(first_event.get("parent_task_id") or 0) == push_parent
          and first_event.get("title") == "Push child"
          and first_event.get("status") in ("queued", "working"),
          "status event carries parent id, title and a wire status (%s)"
          % first_event.get("status"))

    # Real lifecycle: the child dispatches (working) and completes (done
    # with summary) — the parent's daemon hears every transition.
    check(wait_for(lambda: any(e.get("status") == "done"
                               for e in events_for(push_child)) or False,
                   timeout=90, what="done subtask event") is not None,
          "subtask completion pushes the done transition")
    check(any(e.get("status") == "working" for e in events_for(push_child)),
          "dispatch pushes a working status event")

    # Best-effort retry: while the daemon 500s, events stay queued and
    # land once it recovers. The parent's resumed run re-parks (the
    # ask-flow resume races this section otherwise: its completion
    # would close the parked session the events are delivered to).
    fake.fail_subtask_status = True
    fake.suspend_on_title = "Push parent"
    # The child carries the hold tag (nothing can satisfy it) so it
    # stays queued: its status events keep targeting the parent's
    # stable parked session until the hold is released below.
    gpu_cap_id = [c["id"] for c in slot_public(auth, "/saddles", "saddles")["capabilities"]
                  if c["name"] == "gpu"][0]
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Push child two",
        "parent_task_id": str(push_parent),
        "capability_id": [str(gpu_cap_id)],
    })
    push_child_two = task_id_tmp
    # Converted sleep(7): wait until the dispatcher has actually
    # attempted (and failed) the delivery twice — two ticks' worth of
    # evidence the events stay queued — instead of sleeping blindly.
    check(wait_for(lambda: fake.subtask_fail_hits >= 2 or False,
                   timeout=15, what="failed subtask-event deliveries") is not None
          and len(events_for(push_child_two)) == 0,
          "failed pushes stay queued (no event delivered)")
    fake.fail_subtask_status = False
    check(wait_for(lambda: len(events_for(push_child_two)) > 0 or False,
                   timeout=30, what="retried subtask event") is not None,
          "recovered pushes deliver on a later tick")
    # Release the hold: the child dispatches and completes; the settle
    # waits at the end of this bundle cover the parent's final resume.
    act(auth, task_state(auth, push_child_two)["capabilities"][0]["detach"], {})

    print("== R4 B1: synchronous subtask callback ==")
    # Plan 10: the subtask callback is per-saddle-token authenticated.
    r4_token = fake_token

    # Parent with a live parked session (suspended) hosts the callback.
    fake.suspend_on_title = "Callback parent"
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Callback parent",
        "priority": "2",
    })
    cb_parent = task_id_tmp
    check(wait_for(lambda: task_state(auth, cb_parent).get("status") == "suspended"
                   or False, timeout=60, what="callback parent parks") is not None,
          "callback parent parks")

    status, _ = saddle_api("POST", "/api/saddle/subtask", None,
                           {"parent_task_id": cb_parent, "title": "no token"})
    check(status == 401, "subtask callback rejects a missing token")

    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
                 " values (1, 'Anchor task', 'done', 5, strftime('%s','now'))")
    anchor = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.commit()
    conn.close()
    status, body = saddle_api("POST", "/api/saddle/subtask", r4_token, {
        "parent_task_id": cb_parent,
        "title": "Wire the flange",
        "description": "torque to spec",
        "depends_on": [anchor],
        "session_id": 1,
    })
    check(status == 200 and body.get("ok") is True and int(body.get("task_id") or 0) > 0,
          "callback creates the subtask synchronously")
    subtask_id = int(body.get("task_id"))
    detail = task_state(auth, subtask_id)
    check(detail["title"] == "Wire the flange" and int(detail["project_id"]) == 1
          and int(detail["parent_task_id"]) == cb_parent,
          "subtask row inherits the parent's project and linkage")
    check(detail.get("task_branch") == task_state(auth, cb_parent).get("task_branch"),
          "subtask shares the parent's task branch")
    conn = db()
    dep_rows = conn.execute(
        "select depends_on_task_id from task_dependencies where task_id = ?",
        (subtask_id,)).fetchall()
    conn.close()
    check([r[0] for r in dep_rows] == [anchor],
          "depends_on wired as task_dependencies")

    # Idempotency + entry-poll fallback dedup.
    status, body2 = saddle_api("POST", "/api/saddle/subtask", r4_token, {
        "parent_task_id": cb_parent, "title": "Wire the flange"})
    check(status == 200 and int(body2.get("task_id") or 0) == subtask_id,
          "retried callback returns the same task id")
    conn = db()
    before = conn.execute("select count(*) from tasks where parent_task_id = ?",
                          (cb_parent,)).fetchone()[0]
    conn.close()
    conn = db()
    cb_session = conn.execute(
        "select id from sessions where task_id = ? and ended_at is null",
        (cb_parent,)).fetchone()[0]
    conn.close()
    saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
               {"type": "subtask", "session_id": cb_session, "data": {
                   "title": "Wire the flange", "description": "again"}})
    conn = db()
    after = conn.execute("select count(*) from tasks where parent_task_id = ?",
                         (cb_parent,)).fetchone()[0]
    timeline = conn.execute(
        "select count(*) from session_commands where session_id = ?"
        " and type = 'subtask'", (cb_session,)).fetchone()[0]
    conn.close()
    check(after == before,
          "entry-poll fallback dedupes against the callback's row")
    check(timeline >= 1, "timeline entry recorded on the parent's session")

    # Validation: unknown parent, unknown depends_on, ancestor cycle.
    status, body = saddle_api("POST", "/api/saddle/subtask", r4_token,
                              {"parent_task_id": 99999, "title": "ghost"})
    check(status == 400, "unknown parent rejected")
    status, body = saddle_api("POST", "/api/saddle/subtask", r4_token,
                              {"parent_task_id": cb_parent, "title": "bad dep",
                               "depends_on": [99999]})
    check(status == 400 and "unknown task" in body.get("error", ""),
          "unknown depends_on id rejected with a clear error")
    status, body = saddle_api("POST", "/api/saddle/subtask", r4_token,
                              {"parent_task_id": cb_parent, "title": "cycle dep",
                               "depends_on": [cb_parent]})
    check(status == 400 and "cycle" in body.get("error", ""),
          "depending on an ancestor rejected as a cycle")

    print("== plan 11: auto-enqueue agent-created subtasks ==")
    # The per-task option: a subtask created under a parent carrying it
    # enqueues immediately (via the entry-log fallback AND the
    # synchronous callback); without it the subtask lands in draft.
    # Either way the child inherits the parent's value as its own
    # setting.
    conn = db()
    conn.execute("update tasks set auto_enqueue_subtasks = 1 where id = ?", (cb_parent,))
    conn.commit()
    cb_session = conn.execute(
        "select id from sessions where task_id = ? and ended_at is null",
        (cb_parent,)).fetchone()
    conn.close()
    check(cb_session is not None, "callback parent still parked for the fallback leg")

    # The entry-log fallback applies the rule (run first: the parked
    # session is guaranteed open before any enqueued child completes
    # and resumes the parent).
    saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
               {"type": "subtask", "session_id": cb_session[0], "data": {
                   "title": "Fallback enqueued child"}})
    fallback_child = task_id_of("Fallback enqueued child")
    check(wait_for(lambda: task_state(auth, fallback_child)["status"] == "queued"
                   or False, timeout=15, what="fallback child queued") is not None,
          "the entry-log fallback enqueues under the parent's option too")

    status, body = saddle_api("POST", "/api/saddle/subtask", r4_token, {
        "parent_task_id": cb_parent, "title": "Auto-enqueued child"})
    check(status == 200, "callback under an auto-enqueue parent accepted")
    auto_enqueued = int(body["task_id"])
    check(wait_for(lambda: task_state(auth, auto_enqueued)["status"] == "queued"
                   or False, timeout=15, what="auto-enqueued child queued") is not None,
          "the parent's option enqueues the callback subtask immediately")
    check(task_state(auth, auto_enqueued).get("auto_enqueue_subtasks") is True,
          "the child inherits the parent's auto-enqueue setting for its own children")

    status, body = saddle_api("POST", "/api/saddle/subtask", r4_token, {
        "parent_task_id": auto_enqueued, "title": "Grandchild under inheritance"})
    grandchild_ae = int(body["task_id"])
    check(task_state(auth, grandchild_ae)["status"] == "queued",
          "inheritance applies: a subtask under the child enqueues too")

    # Without the option the callback drafts (reset first — the child
    # inherits the parent's value at creation time).
    conn = db()
    conn.execute("update tasks set auto_enqueue_subtasks = 0 where id = ?", (cb_parent,))
    conn.commit()
    conn.close()
    status, body = saddle_api("POST", "/api/saddle/subtask", r4_token, {
        "parent_task_id": cb_parent, "title": "Draft callback child"})
    draft_cb_child = int(body["task_id"])
    detail = task_state(auth, draft_cb_child)
    check(detail["status"] == "draft"
          and detail.get("auto_enqueue_subtasks") is False,
          "without the option the callback subtask lands in draft (setting off)")

    print("== settle: this shard's parked parents resume and finish ==")
    # Settle earlier stragglers first: the round-3/B1 suspended parents
    # resume (a fresh /start) once their last subtask completes — those
    # pushes would consume the one-shot suspend flag below.
    check(wait_task_done(auth, push_parent, timeout=90) is not None
          and wait_task_done(auth, push_child_two, timeout=90) is not None
          and wait_task_done(auth, cb_parent, timeout=90) is not None,
          "earlier suspended parents settle before the auto-approve run")


# =============================================================================
# Bundle: lineage — subtask auto-approve (transitive coverage, the
# toggle's descendant-only scope, the final resume), /start parent
# context and the daemon task-reading API.
# =============================================================================

def bundle_lineage(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 exists only as a hold (no saddle carries it): the auto-approve
    # children stay queued until the final-resume leg lifts it.
    saddles_state = slot_public(auth, "/saddles", "saddles")
    act(auth, saddles_state["create_capability"], {"name": "gpu"})

    # Parent parks; its toggle then covers all descendants. Children are
    # created with an unmet 'gpu' requirement so they stay queued (and
    # untouched) until the requirement is lifted for the final-resume leg.
    fake.suspend_on_title = "Auto parent"
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Auto parent",
        "priority": "2",
    })
    auto_parent = task_id_tmp
    check(wait_for(lambda: task_state(auth, auto_parent).get("status") == "suspended"
                   or False, timeout=60, what="auto parent parks") is not None,
          "auto parent parks")

    detail = task_state(auth, auto_parent)
    check(detail.get("auto_approve") is False,
          "auto-approve defaults to off")
    act(auth, detail["toggle_auto_approve"], {})
    check(task_state(auth, auto_parent).get("auto_approve") is True,
          "toggle turns auto-approve on")

    gpu_option = int([o["id"] for o in slot_public(auth, "/tasks", "tasks")["capability_options"]
                      if o["name"] == "gpu"][0])
    status, body = saddle_api("POST", "/api/saddle/subtask", r4_token,
                              {"parent_task_id": auto_parent,
                               "title": "Auto child", "depends_on": []})
    auto_child = int(body["task_id"])
    status, body = saddle_api("POST", "/api/saddle/subtask", r4_token,
                              {"parent_task_id": auto_child, "title": "Auto grandchild"})
    auto_grand = int(body["task_id"])
    # Fixture: one more held subtask whose title the task-reading API's
    # q= search matches below (this bundle's "Wire the flange").
    status, body = saddle_api("POST", "/api/saddle/subtask", r4_token, {
        "parent_task_id": auto_parent, "title": "Wire the flange",
        "description": "torque to spec"})
    flange_task = int(body["task_id"])
    act(auth, task_state(auth, auto_child)["attach_capability"],
        {"capability_id": str(gpu_option)})
    act(auth, task_state(auth, auto_grand)["attach_capability"],
        {"capability_id": str(gpu_option)})
    act(auth, task_state(auth, flange_task)["attach_capability"],
        {"capability_id": str(gpu_option)})
    # The callback children land in draft (auto_parent carries no
    # auto-enqueue option); the held gpu requirement attaches while
    # editable, then the human enqueues each — they sit queued (and
    # un-dispatchable) until the release loop lifts the hold.
    for held in (auto_child, auto_grand, flange_task):
        enqueue_task(auth, held)

    # Transitive coverage: the GRANDCHILD's review is auto-answered
    # (any ancestor set), answered-by 'auto', timeline entry recorded.
    conn = db()
    conn.execute("insert into sessions (task_id, saddle_id, started_at)"
                 " values (?, ?, strftime('%s','now'))", (auto_grand, fake_id))
    conn.commit()
    grand_session = conn.execute("select max(id) from sessions").fetchone()[0]
    conn.close()
    saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
               {"type": "review", "session_id": grand_session, "data": {
                   "summary": "the flange is wired"}})
    grand_detail = task_state(auth, auto_grand)
    grand_review = [b for b in grand_detail["blocks"] if b["block_type"] == "review"][0]
    check(grand_review.get("answered") is True
          and grand_review.get("answer") == "Auto-approved (subtask auto-approve is on)"
          and grand_review.get("answered_by") == "auto",
          "covered grandchild review auto-answered as 'auto'")
    conn = db()
    timeline_auto = conn.execute(
        "select count(*) from session_commands where session_id = ?"
        " and type = 'auto-approve'", (grand_session,)).fetchone()[0]
    conn.close()
    check(timeline_auto >= 1, "auto action visible in the timeline")

    # Questions and RFCs still block for a human.
    saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
               {"type": "block", "session_id": grand_session, "data": {
                   "question": "Which torque spec?", "type": "question"}})
    saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
               {"type": "block", "session_id": grand_session, "data": {
                   "question": "RFC: approve the design", "type": "rfc",
                   "body": "# Design"}})
    grand_detail = task_state(auth, auto_grand)
    check(all(b.get("answered") is not True for b in grand_detail["blocks"]
              if b["block_type"] in ("question", "rfc")),
          "questions and RFCs are never auto-answered")

    # Plan 11: the transitive toggle also covers descendant PLAN blocks
    # (instant approve, plan markdown applied).
    saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
               {"type": "block", "session_id": grand_session, "data": {
                   "question": "Plan submitted for review", "type": "plan",
                   "body": "# Grandchild plan\n\n1. Follow the parent plan"}})
    grand_detail = task_state(auth, auto_grand)
    grand_plan = [b for b in grand_detail["blocks"] if b["block_type"] == "plan"][0]
    check(grand_plan.get("answered") is True
          and grand_plan.get("answer") == "Plan approved. Proceed with implementation."
          and grand_plan.get("answered_by") == "auto",
          "the ancestor toggle instantly approves descendant plan blocks")
    check(grand_detail.get("plan", "").startswith("# Grandchild plan"),
          "the auto-approved plan markdown is applied to the task")

    # The toggle covers DESCENDANTS only: the parent's own review waits.
    conn = db()
    conn.execute("insert into sessions (task_id, saddle_id, started_at, ended_at,"
                 " outcome) values (?, ?, strftime('%s','now'), strftime('%s','now'),"
                 " 'done')", (auto_parent, fake_id))
    conn.commit()
    parent_session = conn.execute("select max(id) from sessions").fetchone()[0]
    conn.close()
    saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
               {"type": "review", "session_id": parent_session, "data": {
                   "summary": "the whole change"}})
    parent_detail = task_state(auth, auto_parent)
    parent_review = [b for b in parent_detail["blocks"] if b["block_type"] == "review"][0]
    check(parent_review.get("answered") is not True,
          "the top-level parent's own review is never auto-approved")

    # All descendants terminal → the suspended parent resumes with the
    # final-review instruction. The resume fires per completion (ANY
    # terminal child makes the parent offerable again), so release the
    # held descendants one at a time and re-park the parent between
    # releases: only the last resume sees every descendant terminal,
    # and its notice then carries the final-review instruction.
    for step, held_task in enumerate((auto_child, auto_grand, flange_task)):
        if step < 2:
            fake.suspend_on_title = "Auto parent"  # interim resumes re-park
        act(auth, task_state(auth, held_task)["capabilities"][0]["detach"], {})
        check(wait_task_done(auth, held_task, timeout=90) is not None,
              "released descendant settles before the next release (step %d)" % step)

    def pushes_for(task_id):
        return len([p for p in fake.start_payloads
                    if int(p.get("task", {}).get("id") or 0) == task_id])

    before_pushes = pushes_for(auto_parent)
    check(wait_for(lambda: pushes_for(auto_parent) > before_pushes or False,
                   timeout=90, what="auto parent resumes") is not None,
          "terminal subtasks resume the suspended parent")
    notice = ""
    for p in fake.start_payloads:
        if int(p.get("task", {}).get("id") or 0) == auto_parent:
            notice = p.get("resume_notice") or ""
    check("All subtasks are complete" in notice
          and "ONE final review" in notice and "saddle review" in notice,
          "resume notice ends with the final-review instruction")

    print("== R5 B1: /start parent context ==")
    # A fresh subtask dispatch carries the parent object (plan clamped).
    conn = db()
    conn.execute("update tasks set plan = ? where id = ?",
                 ("# Parent plan\n\n" + "step. " * 1200, auto_parent))
    conn.commit()
    conn.close()
    # Barrier: the dispatch reads the parent row through the app's own
    # connection — wait until the app serves the externally written
    # plan before creating the dispatchable task (a stale snapshot
    # would otherwise ship an empty parent plan).
    check(wait_for(lambda: "Parent plan" in
                   (((saddle_api("GET", "/api/saddle/task/%d" % auto_parent,
                                 r4_token)[1].get("task")) or {}).get("plan") or "")
                   or False, timeout=15, what="app serves the parent plan") is not None,
          "the task API serves the parent plan before the lineage dispatch")
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Lineage subtask",
        "priority": "3",
    })
    lineage_task = task_id_tmp
    conn = db()
    conn.execute("update tasks set parent_task_id = ?, description = 'torque to spec'"
                 " where id = ?", (auto_parent, lineage_task))
    conn.commit()
    conn.close()
    check(wait_for(lambda: any(int(p.get("task", {}).get("id") or 0) == lineage_task
                               for p in fake.start_payloads) or False,
                   timeout=60, what="lineage subtask dispatches") is not None,
          "lineage subtask dispatches")
    lineage_push = [p for p in fake.start_payloads
                    if int(p.get("task", {}).get("id") or 0) == lineage_task][-1]
    parent_obj = lineage_push.get("parent")
    check(parent_obj is not None
          and int(parent_obj.get("id") or 0) == auto_parent
          and parent_obj.get("title") == "Auto parent"
          and parent_obj.get("status") in ("suspended", "working", "done"),
          "subtask /start carries the parent object")
    check(parent_obj.get("description") == "torque to spec"
          or "Parent plan" in (parent_obj.get("plan") or ""),
          "parent description/plan content flows through")
    check(len(parent_obj.get("plan") or "") <= 4000,
          "parent plan clamped to ~4000 chars (%d)"
          % len(parent_obj.get("plan") or ""))
    root_pushes = [p for p in fake.start_payloads
                   if int(p.get("task", {}).get("id") or 0) == auto_parent]
    check(all("parent" not in p for p in root_pushes),
          "root task /start carries no parent object")

    print("== R5 B2: task-reading API ==")
    status, _ = saddle_api("GET", "/api/saddle/tasks", None)
    check(status == 401, "tasks endpoint rejects a missing token")
    status, _ = saddle_api("GET", "/api/saddle/task/%d" % auto_child, None)
    check(status == 401, "task detail endpoint rejects a missing token")

    # Fixture: a second project whose tasks all carry the "Diff " title
    # prefix (this bundle's own copy of the pages bundle's diff
    # projects) for the project_id filter checks — created first so the
    # unfiltered list has the population the count check expects.
    state = slot_public(auth, "/projects", "projects")
    act(auth, state["create"], {"name": "diff-fabrica",
                                "git_url": "ssh://fabrica.unitatem.net:222/clanker/clanker-saddle.git"})
    conn = db()
    other_project = int([p["id"] for p in slot_public(auth, "/projects", "projects")["projects"]
                         if p["name"] == "diff-fabrica"][0])
    for suffix in ("fabrica", "scp"):
        conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
                     " values (?, ?, 'done', 5, strftime('%s','now'))",
                     (other_project, "Diff %s" % suffix))
    conn.commit()
    conn.close()
    status, body = saddle_api("GET", "/api/saddle/tasks", r4_token)
    check(status == 200 and len(body.get("tasks", [])) > 5,
          "tasks endpoint lists tasks without filters")
    row = [t for t in body["tasks"] if int(t["id"]) == auto_child][0]
    check(row["title"] == "Auto child" and int(row["parent_task_id"]) == auto_parent
          and row["status"] in ("done", "queued", "working"),
          "compact rows carry id/title/status/parent_task_id")

    status, body = saddle_api("GET", "/api/saddle/tasks?project_id=1", r4_token)
    check(all(True for _ in [1])  # project 1 dominates the data set
          and not any(int(t["id"]) == 0 for t in body["tasks"]),
          "project_id filter accepted")
    status, body = saddle_api("GET",
                              "/api/saddle/tasks?project_id=%d" % other_project, r4_token)
    check(all(t["title"].startswith("Diff") for t in body["tasks"]),
          "other project lists only its own tasks")

    status, body = saddle_api("GET", "/api/saddle/tasks?q=FLANGE", r4_token)
    titles = [t["title"] for t in body["tasks"]]
    check("Wire the flange" in titles,
          "q matches titles case-insensitively")
    status, body = saddle_api("GET", "/api/saddle/tasks?q=torque%20to%20spec", r4_token)
    check(any(t["title"] == "Lineage subtask" for t in body["tasks"]),
          "q matches description text")
    status, body = saddle_api("GET", "/api/saddle/tasks?q=parent%20plan", r4_token)
    check(any(t["title"] == "Auto parent" for t in body["tasks"]),
          "q matches plan text")

    status, body = saddle_api("GET", "/api/saddle/task/%d" % auto_child, r4_token)
    task_view = body.get("task", {})
    check(status == 200 and int(task_view.get("id") or 0) == auto_child
          and task_view.get("parent_title") == "Auto parent"
          and int(task_view.get("parent_task_id") or 0) == auto_parent,
          "single-task view joins the parent title")
    check("status" in task_view and "description" in task_view
          and "plan" in task_view and "summary" in task_view,
          "single-task view carries the full detail fields")
    status, body = saddle_api("GET", "/api/saddle/task/99999", r4_token)
    check(status == 404, "unknown task id returns 404")


# =============================================================================
# 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 bundle_allocation(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)})

    fake2.suspend_on_title = "Affinity parent"
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Affinity parent",
        "priority": "2",
        "capability_id": [str(affinity_cap)],
    })
    affinity_parent = task_id_tmp
    check(wait_for(lambda: task_state(auth, affinity_parent).get("status") == "suspended"
                   or False, timeout=60, what="affinity parent parks") is not None,
          "affinity parent parks (suspended) on the tag-carrying racer")
    conn = db()
    affinity_session = conn.execute(
        "select saddle_id from sessions where task_id = ? and ended_at is null",
        (affinity_parent,)).fetchone()[0]
    conn.close()
    check(affinity_session == int(racer2_card["id"]),
          "the parked workspace sits on racer-two (the only tag-carrier)")

    # Now racer-three carries the tag too — the pinning is by
    # workspace, not tags.

    # A terminal subtask makes the parked task resume-offerable —
    # affine to racer-two only. The park is observed the moment the
    # suspended push lands (no poller lag to compensate), so the
    # child is inserted only after racer-two's ask triggers are
    # frozen and the wake measurement armed — its completion stamped
    # one second ahead so completed_at > suspended_at holds.
    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)
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority, parent_task_id,"
                 " summary, completed_at, created_at) values (1, 'Affinity child', 'done',"
                 " 5, ?, 'affinity child finished', strftime('%s','now','+1 second'),"
                 " strftime('%s','now'))", (affinity_parent,))
    conn.commit()
    conn.close()
    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 task (204)")

    check(wait_for(lambda: len(fake2.wake_log) > wakes_before_2 or False,
                   timeout=30, 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 task")
    check("Affinity child" in (resume_payload.get("resume_notice") or "")
          and "affinity child finished" in (resume_payload.get("resume_notice") or ""),
          "the resume assignment carries the subtask report notice")
    check(wait_task_done(auth, affinity_parent, timeout=60) is not None,
          "the resumed parent completes")

    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)

    # Subtask capabilities (`saddle subtask --capability`): explicit
    # only — never inherited — and 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).
    conn = db()
    tasks_before = conn.execute("select count(*) from tasks").fetchone()[0]
    conn.close()
    status, body = saddle_api("POST", "/api/saddle/subtask", r4_token, {
        "parent_task_id": cap_task, "title": "Bad tag subtask",
        "capabilities": ["gpu", "warp-drive"]})
    conn = db()
    tasks_after = conn.execute("select count(*) from tasks").fetchone()[0]
    conn.close()
    check(status == 400 and tasks_after == tasks_before,
          "an unknown capability tag rejects the subtask 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/subtask", r4_token, {
        "parent_task_id": cap_task, "title": "Bad shape subtask",
        "capabilities": "gpu"})
    check(status == 400, "a non-array capabilities member rejects")

    # Explicit-only: a parent requiring 'lostwake' (the lost-wake
    # task) gets a subtask tagged just 'gpu' — nothing inherited.
    status, body = saddle_api("POST", "/api/saddle/subtask", r4_token, {
        "parent_task_id": lost_task, "title": "Explicit tag subtask",
        "capabilities": ["gpu"]})
    check(status == 200 and body.get("ok") is True,
          "a valid tag creates the subtask synchronously")
    tagged_subtask = int(body["task_id"])
    detail = task_state(auth, tagged_subtask)
    check([c["name"] for c in detail.get("capabilities", [])] == ["gpu"],
          "the subtask carries exactly the explicit tag (no inheritance)")
    check([c["name"] for c in task_state(auth, lost_task).get("capabilities", [])]
          == ["lostwake"],
          "the parent keeps its own requirement")
    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_subtask,)).fetchall()
    conn.close()
    check([r[0] for r in tag_rows] == ["gpu"],
          "the task_capabilities rows record the explicit tag")
    # Without the parent's auto-enqueue option the callback's subtask
    # lands in draft — the human Enqueue action queues it (plan 11).
    check(detail["status"] == "draft", "callback subtask without auto-enqueue lands in draft")
    enqueue_task(auth, tagged_subtask)
    # The queued subtask dispatches (gpu-carrier) and settles before
    # the slow-wake run needs an empty queue.
    check(wait_task_done(auth, tagged_subtask, timeout=90) is not None,
          "the tagged subtask dispatches and completes")

    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()
    saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
               {"type": "block", "session_id": dash_session, "data": {
                   "question": "Dash fixture question?", "type": "question"}})
    saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
               {"type": "review", "session_id": dash_session, "data": {
                   "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 task_progress (task_id, summary, created_at)"
                 " values (?, 'timeline fixture progress',"
                 " strftime('%s','now','-5 seconds'))",
                 (cap_task,))
    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")
    saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
               {"type": "block", "session_id": dash_session, "data": {
                   "question": "Timeline probe: approve the dashboard flow",
                   "type": "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


# =============================================================================
# Bundle: lifecycle — offline requeues, the hung-request tick survival,
# parked-refresh bookkeeping, cancellation, the health panel, raw-log
# losslessness, git identity and the live blocked-push roundtrip.
# =============================================================================

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

    print("== plan 11: draft state machine, edit matrix, enqueue/delete guards ==")
    # New tasks default to draft; Enqueue is the only path to the
    # queue (there is no agent path — the ingest API refuses to queue).
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1",
        "title": "Draft lifecycle task",
        "priority": "4",
    })
    draft_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                      if t["title"] == "Draft lifecycle task"][0])
    detail = task_state(auth, draft_task)
    check(detail["status"] == "draft", "a created task lands in draft")
    check(detail.get("enqueue") is not None and detail.get("delete") is not None,
          "draft exposes the enqueue + delete actions")
    check(detail.get("cancel") is not None, "the cancel action stays authored")
    status, page = http("GET", "/tasks/%d" % draft_task, timeout=15)
    check(status == 200 and "Enqueue task" in page and ">Delete<" in page,
          "the task page shows Enqueue + Delete in draft")
    check("Mark done" not in page,
          "the user Mark-complete action is gone (bot-only completion)")
    # A draft dependency blocks enqueue with a shown error naming it.
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1",
        "title": "Draft dependency",
        "priority": "4",
    })
    draft_dep = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                     if t["title"] == "Draft dependency"][0])
    act(auth, detail["dependency"], {"depends_on_task_id": str(draft_dep)})
    directives_result = act(auth, task_state(auth, draft_task)["enqueue"], {})
    check(any(d.get("type") == "notify" and d.get("kind") == "error"
              and "Draft dependency" in (d.get("message") or "")
              for d in directives_result),
          "enqueue is refused with an error naming the draft dependency (%s)"
          % [d.get("message") for d in directives_result if d.get("type") == "notify"])
    check(task_state(auth, draft_task)["status"] == "draft",
          "the blocked enqueue leaves the task in draft")
    # The edit matrix starts in DRAFT: every field except status.
    act(auth, detail["update"], {
        "title": "Draft lifecycle task (edited)",
        "description": "Now with a description",
        "priority": "2",
        "task_branch": "feature/draft",
        "branch_from": "main",
        "plan_auto_approve": "on",
        "auto_enqueue_subtasks": "on",
        "auto_approve": "on",
    })
    detail = task_state(auth, draft_task)
    check(detail["title"] == "Draft lifecycle task (edited)"
          and detail["description"] == "Now with a description"
          and detail["priority"] == 2
          and detail["task_branch"] == "feature/draft"
          and detail["branch_from"] == "main",
          "draft edits apply to every field")
    check(detail.get("plan_auto_approve") is True
          and detail.get("auto_enqueue_subtasks") is True
          and detail.get("auto_approve") is True,
          "draft edits apply the task options")
    # A hold tag nothing carries keeps the dependent QUEUED (never
    # dispatched) through the remaining edit-matrix checks.
    saddles_state = slot_public(auth, "/saddles", "saddles")
    act(auth, saddles_state["create_capability"], {"name": "holdlock"})
    holdlock_cap = int([c["id"] for c in slot_public(auth, "/saddles", "saddles")
                        ["capabilities"] if c["name"] == "holdlock"][0])
    act(auth, detail["attach_capability"], {"capability_id": str(holdlock_cap)})
    # Enqueue the dependency, then the dependent goes through.
    enqueue_task(auth, draft_dep)
    enqueue_task(auth, draft_task)
    check(task_state(auth, draft_task)["status"] == "queued",
          "enqueue succeeds once no dependency is in draft")

    # Drafts are never dispatched even with an idle saddle available.
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1",
        "title": "Never-dispatched draft",
        "priority": "1",
    })
    silent_draft = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                        if t["title"] == "Never-dispatched draft"][0])
    check(quiet_window(quiet_seconds(),
                       lambda: task_state(auth, silent_draft)["status"] == "draft",
                       "draft never dispatched"),
          "an idle saddle never picks a draft up (no agent enqueue path)")

    # Once work can begin (queued here), only priority is editable.
    act(auth, detail["update"], {
        "title": "Should not apply",
        "priority": "3",
    })
    detail = task_state(auth, draft_task)
    check(detail["title"] == "Draft lifecycle task (edited)" and detail["priority"] == 3,
          "non-draft edits apply priority only (title locked)")
    # A terminal task edits nothing.
    conn = db()
    conn.execute("update tasks set status = 'done', completed_at = strftime('%s','now')"
                 " where id = ?", (draft_task,))
    conn.commit()
    conn.close()
    act(auth, task_state(auth, draft_task)["update"], {"priority": "8"})
    check(task_state(auth, draft_task)["priority"] == 3,
          "terminal tasks edit nothing (priority locked)")
    # Capabilities lock once a worker has the job: editable while
    # queued, locked once working.
    act(auth, saddles_state["create_capability"], {"name": "editlock"})
    editlock_cap = int([c["id"] for c in slot_public(auth, "/saddles", "saddles")
                        ["capabilities"] if c["name"] == "editlock"][0])
    act(auth, saddle_card(auth, "fake-saddle")["attach_capability"],
        {"capability_id": str(editlock_cap)})
    locked_task = create_queued(auth, {"project_id": "1", "title": "Caplock task",
                                       "priority": "2"})
    detail = task_state(auth, locked_task)
    act(auth, detail["attach_capability"], {"capability_id": str(editlock_cap)})
    check(any(c["name"] == "editlock"
              for c in task_state(auth, locked_task)["capabilities"]),
          "capabilities attach while queued (no worker yet)")
    check(wait_for(lambda: task_state(auth, locked_task).get("status") == "working"
                   or False, timeout=60, what="caplock working") is not None,
          "the caplock task reaches working")
    detach_directives = act(auth, task_state(auth, locked_task)["capabilities"][0]["detach"], {})
    check(any(d.get("type") == "notify" and d.get("kind") == "error"
              for d in detach_directives),
          "capabilities are locked once a worker has the job")
    # The ingest API has no queue transition either (only working and
    # blocked are accepted states).
    conn = db()
    caplock_session = conn.execute(
        "select id from sessions where task_id = ? and ended_at is null",
        (locked_task,)).fetchone()
    conn.close()
    if caplock_session is not None:
        status, _ = saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
                               {"type": "status", "session_id": caplock_session[0],
                                "data": {"state": "queued"}})
        check(status == 200
              and task_state(auth, locked_task)["status"] in ("working", "done"),
              "the ingest API's status push never queues a task")

    # Delete guards: a clean draft deletes; non-draft refuses; a draft
    # with dependents is refused naming them.
    act(auth, task_state(auth, silent_draft)["delete"], {})
    check(not any(t["id"] == silent_draft
                  for t in slot_public(auth, "/tasks", "tasks")["tasks"]),
          "a clean draft task deletes")
    delete_directives = act(auth, task_state(auth, draft_task)["delete"], {})
    check(any(d.get("type") == "notify" and d.get("kind") == "error"
              for d in delete_directives),
          "a non-draft task refuses the delete")
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1", "title": "Guarded draft", "priority": "4"})
    guarded = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                   if t["title"] == "Guarded draft"][0])
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1", "title": "Dependent namer", "priority": "4"})
    namer = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                 if t["title"] == "Dependent namer"][0])
    act(auth, task_state(auth, namer)["dependency"], {"depends_on_task_id": str(guarded)})
    delete_directives = act(auth, task_state(auth, guarded)["delete"], {})
    check(any(d.get("type") == "notify" and d.get("kind") == "error"
              and "Dependent namer" in (d.get("message") or "")
              for d in delete_directives),
          "deleting a task with dependents is refused naming them")

    print("== offline requeues ==")
    # Tasks caught working on a saddle that went offline requeue with a
    # bot comment each tick.
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Offline saddle task",
        "priority": "2",
    })
    offline_task = task_id_tmp
    check(wait_for(lambda: task_state(auth, offline_task).get("status") == "working"
                   or False, timeout=60, what="offline task starts working") is not None,
          "task reaches working before its saddle disappears")
    # Hold the daemon mid-round across the requeue window (the
    # original relied on its 8s work period outlasting the 5s tick):
    # the task completing on its own would race the offline pass and
    # skip the requeue the check exists to observe.
    fake.hold_state = True
    conn = db()
    conn.execute("update saddles set status = 'offline' where name = 'fake-saddle'")
    conn.commit()
    conn.close()
    # The requeue frees the saddle in the same tick, so 'queued' is
    # transient; the durable evidence is the bot comment (and the task
    # completing again on the healthy daemon).
    check(wait_for(lambda: any(c["body"] == "saddle went offline; task requeued"
                               for c in task_state(auth, offline_task).get("comments", []))
                   or False, timeout=30, what="offline requeue comment") is not None,
          "offline saddle requeues the task")
    fake.hold_state = False  # release: the round ends, the daemon re-asks
    check(wait_task_done(auth, offline_task, timeout=60) is not None,
          "requeued task completes again on the healthy daemon")

    print("== R6 F1: a hanging request cannot freeze the tick ==")
    # One wedged /task-update (never responds) blocks the comment/
    # subtask-event delivery pass; the Soup timeout must fail it and
    # later ticks keep dispatching.
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
                 " values (1, 'Hang survivor', 'queued', 2, strftime('%s','now'))")
    hang_task = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.close()
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Hang runner",
        "priority": "3",
    })
    hang_runner = task_id_tmp
    # A human comment on the running task gives the tick a
    # /task-update delivery to make; the next one hangs.
    check(wait_for(lambda: task_state(auth, hang_runner).get("status") == "working"
                   or False, timeout=60, what="hang runner working") is not None,
          "hang runner reaches working")
    act(auth, task_state(auth, hang_runner)["comment"],
        {"body": "operator ping while wedged"})
    fake.hang_next["/task-update"] = 1
    check(wait_task_done(auth, hang_runner, timeout=90) is not None,
          "dispatching task survives a hung daemon request")
    check(wait_for(lambda: any(c.get("text") == "operator ping while wedged"
                               or c.get("body") == "operator ping while wedged"
                               for c in fake.comment_log) or False,
                   timeout=30, what="wedged comment eventually delivered") is not None,
          "the comment delivery retries after the hang")
    # The hung delivery itself fails after the timeout; the system
    # kept moving either way.
    with open(os.path.join(WORK, "app.log")) as handle:
        app_log = handle.read()
    check("tick exceeded" not in app_log or "Dispatch tick failed" in app_log,
          "a hung request never freezes the tick latch")

    print("== R6 F2: refresh must not corrupt parked bookkeeping ==")
    fake.suspend_on_title = "Parked refresh parent"
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Parked refresh parent",
        "priority": "2",
    })
    parked = task_id_tmp
    check(wait_for(lambda: task_state(auth, parked).get("status") == "suspended"
                   or False, timeout=60, what="parked parent suspends") is not None,
          "parent parks before the refresh")
    fake_card = saddle_card(auth, "fake-saddle")
    act(auth, fake_card["refresh"], {})
    fake_card = saddle_card(auth, "fake-saddle")
    check(fake_card["status"] == "idle" and int(fake_card["current_task_id"] or 0) == 0,
          "refresh on a parked (suspended) daemon leaves the saddle idle")
    # The parked task keeps its open session and resumes normally.
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority, parent_task_id,"
                 " summary, completed_at, created_at) values (1, 'Parked child', 'done', 5,"
                 " ?, 'child finished', strftime('%s','now','+1 second'),"
                 " strftime('%s','now'))", (parked,))
    conn.commit()
    conn.close()
    check(wait_task_done(auth, parked, timeout=90) is not None,
          "parked parent still resumes after the refresh")

    print("== R7 B1: cancel reaches the daemon ==")
    # Plan 10 phase 2: cancellation is POST /task-update {kind: cancel}
    # to the RUNNING daemon (the abort machinery, workspace kept).
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Cancel live",
        "priority": "2",
    })
    cancel_live = task_id_tmp
    check(wait_for(lambda: task_state(auth, cancel_live).get("status") == "working"
                   or False, timeout=60, what="cancel-live working") is not None,
          "task reaches working before the cancel")
    act(auth, task_state(auth, cancel_live)["cancel"], {})
    check(wait_for(lambda: cancel_live in fake.abort_log or False,
                   timeout=30, what="daemon abort recorded") is not None,
          "cancelling POSTs task-update cancel to the daemon")
    check(any(c["body"] == "cancelled; daemon aborted"
              for c in task_state(auth, cancel_live).get("comments", [])),
          "the abort outcome is narrated")
    check(task_state(auth, cancel_live)["status"] == "failed",
          "cancelled task lands failed as before")

    # A 500 on the cancel marks the saddle offline (no fence layer
    # anymore); nothing dispatches to it while offline, and the
    # daemon's own ask flips the row back to idle (proof of life).
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Fence runner",
        "priority": "2",
    })
    fence_runner = task_id_tmp
    check(wait_for(lambda: task_state(auth, fence_runner).get("status") == "working"
                   or False, timeout=60, what="fence runner working") is not None,
          "second task reaches working")
    fake.abort_failures = 1
    act(auth, task_state(auth, fence_runner)["cancel"], {})
    check(any(c["body"] == "cancelled; daemon unreachable — saddle marked offline"
              for c in task_state(auth, fence_runner).get("comments", [])),
          "failed cancel narrates the offline marking")
    check(saddle_card(auth, "fake-saddle")["status"] == "offline",
          "saddle marked offline after the failed cancel")
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "During fence",
        "priority": "3",
    })
    during_fence = task_id_tmp
    # Converted sleep(7): a quiet window spanning several dispatch
    # ticks in which the task MUST stay queued — if the offline
    # saddle ever dispatches it, this fails loudly. The daemon is held
    # mid-round for the window (the original relied on its 8s work
    # period outlasting the sleep): a finishing round would ask, prove
    # liveness and flip the offline row — only after the window may it.
    fake.hold_state = True
    check(quiet_window(quiet_seconds(),
                       lambda: task_state(auth, during_fence)["status"] == "queued",
                       "offline saddle does not dispatch during_fence"),
          "offline saddle is not dispatchable")
    fake.hold_state = False
    # The (unaborted) daemon finishes its round, goes idle-awake and
    # asks — the ask is the proof of life that flips the row back.
    check(wait_for(lambda: saddle_card(auth, "fake-saddle")["status"] == "idle" or False,
                   timeout=30, what="ask flips offline to idle") is not None,
          "the daemon's ask flips the offline row back to idle")
    check(wait_task_done(auth, during_fence, timeout=60) is not None,
          "queued work dispatches once the saddle proves it is alive")

    print("== R7 B3: health panel + pushed raw log from the first round ==")
    # A normally dispatched task pushes its chunks immediately: raw
    # rows exist from the first round (the session-4 gap stays
    # fixed without any probe machinery).
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Raw from first push",
        "priority": "2",
    })
    raw_first = task_id_tmp

    def raw_count(task_id):
        conn = db()
        try:
            return conn.execute(
                "select count(*) from session_raw_log where session_id in"
                " (select id from sessions where task_id = ?)", (task_id,)).fetchone()[0]
        finally:
            conn.close()

    check(wait_for(lambda: raw_count(raw_first) > 0 or False,
                   timeout=60, what="raw rows from first push") is not None,
          "pushed chunks land from the first round (session-4 gap)")
    check(wait_task_done(auth, raw_first, timeout=60) is not None,
          "raw-first task completes")

    # Health panel: a manual refresh populates the snapshot.
    fake_card = saddle_card(auth, "fake-saddle")
    act(auth, fake_card["refresh"], {})
    card = saddle_card(auth, "fake-saddle")
    check(card.get("health") in ("healthy", "stale")
          and card.get("daemon_state") in ("working", "idle", "done")
          and card.get("uptime_display", "") != ""
          and card.get("last_contact_display", "") not in ("", "never"),
          "saddles page shows daemon state, uptime and last contact")

    print("== R8: pushed chunk ingestion is lossless ==")
    # The daemon's opening flush: a coloured prompt run, plain text,
    # a bare CRLF between prompts and the tail — every non-empty
    # chunk must land, exactly once, in seq order. A mid-flight
    # redelivery of the whole batch (the outbox's at-least-once
    # retry) consumes the duplicates.
    corpus = [
        {"seq": 1, "text": "\x1b[01;32muser@host\x1b[00m:\x1b[01;34m~/task-1\x1b[00m$ "},
        {"seq": 2, "text": "echo hello\r\n"},
        {"seq": 3, "text": "\r\n"},
        {"seq": 4, "text": "hello\r\n"},
        {"seq": 5, "text": "\x1b[01;32muser@host\x1b[00m:\x1b[01;34m~/task-1\x1b[00m$ "},
        {"seq": 6, "text": "done\r\n"},
    ]
    fake.scripted_corpus = corpus
    fake.hold_state = True
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Lossless log task",
        "priority": "2",
    })
    lossless = task_id_tmp

    def stored_seqs(task_id):
        conn = db()
        try:
            return [r[0] for r in conn.execute(
                "select seq from session_raw_log where session_id in"
                " (select id from sessions where task_id = ?)"
                " order by seq", (task_id,))]
        finally:
            conn.close()

    check(wait_for(lambda: task_state(auth, lossless).get("status") == "working"
                   or False, timeout=60, what="lossless task working") is not None,
          "lossless task dispatches and pushes its opening flush")
    check(wait_for(lambda: stored_seqs(lossless) == [1, 2, 3, 4, 5, 6] or False,
                   timeout=30, what="all six chunks stored") is not None,
          "every pushed chunk lands in session_raw_log, CRLF-only included (%s)"
          % stored_seqs(lossless))
    status, _ = saddle_api("POST", "/api/saddle/chunks", fake.token,
                           {"session_id": fake.session_id,
                            "chunks": [dict(chunk) for chunk in corpus]})
    check(status == 200 and stored_seqs(lossless) == [1, 2, 3, 4, 5, 6],
          "a full-batch redelivery stores nothing new (exactly-once)")
    request = urllib.request.Request(
        WEB + "/terminal-stream/%d" % [r[0] for r in db().execute(
            "select id from sessions where task_id = ?", (lossless,))][0])
    with urllib.request.urlopen(request, timeout=15) as stream:
        texts = []
        deadline = time.monotonic() + 8
        while time.monotonic() < deadline and len(texts) < 6:
            line = stream.readline().decode().strip()
            if not line.startswith("data: ") or line == "data: ping":
                continue
            event = json.loads(line[6:])
            if event.get("type") == "raw":
                texts.append(event.get("text") or "")
        check("\r\n" in texts and len(texts) == 6,
              "SSE serves the whitespace chunk unfiltered")
    fake.hold_state = False
    fake.scripted_corpus = None
    check(wait_task_done(auth, lossless, timeout=60) is not None,
          "lossless task completes")

    print("== R9 B1/B2: git identity + ssh key ==")
    # The key travels on /status into the saddle row and the page view
    # (the manual refresh's exceptional probe).
    fake.ssh_public_key = "ssh-ed25519 AAAAC3NzaC1lZDI1teStKey saddle-1@fedora"
    fake_card = saddle_card(auth, "fake-saddle")
    check(fake_card.get("ssh_public_key", "") == "",
          "no key shown before the daemon reports one")
    act(auth, fake_card["refresh"], {})
    check(wait_for(lambda: "AAAAC3NzaC1lZDI1teStKey" in
                   saddle_card(auth, "fake-saddle").get("ssh_public_key", "") or False,
                   timeout=30, what="key lands on the row") is not None,
          "reported ssh_public_key persists onto the saddle row")
    conn = db()
    row_key = conn.execute("select ssh_public_key from saddles where name = 'fake-saddle'"
                           ).fetchone()[0]
    conn.close()
    check(row_key == fake.ssh_public_key,
          "the key is stored verbatim in the database")

    # Identity edit flows through the UI into the assignment payload.
    fake_card = saddle_card(auth, "fake-saddle")
    act(auth, fake_card["update_identity"],
        {"git_name": "clanker-bot", "git_email": "bot@unitatem.net"})
    fake_card = saddle_card(auth, "fake-saddle")
    check(fake_card.get("git_name") == "clanker-bot"
          and fake_card.get("git_email") == "bot@unitatem.net",
          "inline edit saves the git identity")

    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Identity task",
        "priority": "2",
    })
    identity_task = task_id_tmp
    check(wait_for(lambda: any(int(p.get("task", {}).get("id") or 0) == identity_task
                               for p in fake.start_payloads) or False,
                   timeout=60, what="identity task dispatched") is not None,
          "identity task dispatches")
    payload = [p for p in fake.start_payloads
               if int(p.get("task", {}).get("id") or 0) == identity_task][-1]
    identity = payload.get("git_identity") or {}
    check(identity.get("name") == "clanker-bot"
          and identity.get("email") == "bot@unitatem.net",
          "the assignment carries the saddle's git_identity")

    # Earlier assignments (identity unset) carry no git_identity member.
    check(all("git_identity" not in p for p in fake.start_payloads[:-1]),
          "unset identity omits the member (daemon fallback applies)")

    # Clearing the identity omits it from later assignments.
    fake_card = saddle_card(auth, "fake-saddle")
    act(auth, fake_card["update_identity"], {"git_name": "", "git_email": ""})
    check(saddle_card(auth, "fake-saddle").get("git_name", "") == "",
          "clearing the identity empties the row")
    check(wait_task_done(auth, identity_task, timeout=60) is not None,
          "identity task completes")

    print("== phase 2: blocked push -> block row + pulse ==")
    # The browser flow on each pulse: statum.act(task.refresh) — the
    # sealed GET action must rehydrate the task slot with the new
    # block the daemon just PUSHED (state blocked + block payload).
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Live block task",
        "priority": "2",
    })
    live_task = task_id_tmp
    check(wait_for(lambda: task_state(auth, live_task).get("status") == "working"
                   or False, timeout=60, what="live task working") is not None,
          "live-block task reaches working")
    fake.hold_state = True
    fake.state = "blocked"  # the daemon parks on its question
    status = fake.push_results("blocked", block={
        "question": "Pick one", "type": "question", "options": ["A", "B"]})
    check(status == 200, "blocked state push accepted")
    # The parked daemon is idle-awake: it asks (and sleeps on 204).
    fake.maybe_trigger_ask("parked")
    check(wait_for(lambda: any(b.get("question") == "Pick one"
                               for b in task_state(auth, live_task).get("blocks", []))
                   or False, timeout=30, what="block lands in state") is not None,
          "the pushed block reaches the sealed refresh without a reload")
    detail = task_state(auth, live_task)
    check(detail.get("status") == "blocked",
          "the blocked push flips the task to blocked")
    pushed_block = [b for b in detail["blocks"] if b.get("question") == "Pick one"][0]
    conn = db()
    options_json = conn.execute(
        "select options from task_blocks where id = ?", (int(pushed_block["id"]),)
    ).fetchone()[0]
    conn.close()
    check(json.loads(options_json) == ["A", "B"],
          "the block row carries its options payload")
    check(any(s["status"] == "idle"
              for s in slot_public(auth, "/saddles", "saddles")["saddles"]),
          "the blocked push parks the saddle")
    detail = task_state(auth, live_task)
    check(detail.get("refresh") is not None,
          "the sealed GET refresh action is still authored")
    status, _ = http("GET", detail["refresh"]["uri"], timeout=15)
    check(status in (200, 401),
          "the refresh URI resolves as a GET (client roundtrip shape)")
    act(auth, pushed_block["answer_action"], {"answer": "A"})
    # The answer makes the parked task resume-offerable: the affine
    # saddle is woken, asks, and its assignment carries the answer.
    pushes_before = len([p for p in fake.start_payloads
                         if int(p.get("task", {}).get("id") or 0) == live_task])
    check(wait_for(lambda: len([p for p in fake.start_payloads
                                if int(p.get("task", {}).get("id") or 0) == live_task])
                   > pushes_before or False, timeout=60,
                   what="blocked task resumes") is not None,
          "answering the pushed block resumes the parked workspace")
    fake.hold_state = False  # the resumed round runs to completion
    resume_payload = [p for p in fake.start_payloads
                      if int(p.get("task", {}).get("id") or 0) == live_task][-1]
    answers = resume_payload.get("task", {}).get("answers", [])
    check(any(a.get("answer") == "A" for a in answers),
          "the resume assignment carries the human's answer")
    check(wait_task_done(auth, live_task, timeout=60) is not None,
          "live-block task completes after answering")


# =============================================================================
# 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 bundle_answers(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]
    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"
        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")
    # 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=30) 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 ==")
    # Every task's `saddle plan` parks with block.type 'plan'; Accept
    # and Reject-with-feedback drive the same wake -> ask ->
    # allocation resume as any verdict, the stable wording riding the
    # resume_notice; the per-task auto-approve option answers the plan
    # block instantly web-side.
    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 applies the plan markdown and resumes with approval.
    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")
    act(auth, plan_block["answer_action"],
        {"answer": "Plan approved. Proceed with implementation."})
    detail = task_state(auth, accept_task)
    check(detail.get("plan", "").startswith("# Steps"),
          "accept applies the submitted markdown to task.plan")
    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=30) 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"})
    detail = task_state(auth, reject_task)
    reject_block = [b for b in detail["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=30) is not None,
          "the revised plan's task completes")
    planner.set_state("idle")

    # (c) The per-task option answers the plan block instantly
    # web-side — the agent never sees a human gate, and the resume
    # still rides the allocation with the approval notice.
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1", "title": "Auto-approved plan task", "priority": "2",
        "plan_auto_approve": "on",
        "capability_id": [str(answerflow_cap)],
    })
    auto_plan = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                     if t["title"] == "Auto-approved plan task"][0])
    check(task_state(auth, auto_plan).get("plan_auto_approve") is True,
          "the create form stores the plan-auto-approve option")
    enqueue_task(auth, auto_plan)
    ask_status, _ = planner.ask_now()
    check(ask_status == 200, "the planner claims the auto-approve task")
    auto_parked_session = planner.session_id
    plan_block = park_on_plan(auto_plan, "# Auto plan\n\nStraightforward.")
    detail = task_state(auth, auto_plan)
    auto_block = [b for b in detail["blocks"] if b["block_type"] == "plan"][0]
    check(auto_block.get("answered") is True
          and auto_block.get("answer") == "Plan approved. Proceed with implementation."
          and auto_block.get("answered_by") == "auto",
          "the plan block is answered instantly web-side (answered-by auto)")
    check(detail.get("plan", "").startswith("# Auto plan"),
          "the auto-approval applies the plan markdown")
    check(wait_for(lambda: any(
        int(p.get("task", {}).get("id") or 0) == auto_plan
        and p.get("session_id") != auto_parked_session
        and p.get("resume_notice") == "Plan approved. Proceed with implementation."
        for p in planner.start_payloads[1:]) or False,
        timeout=30, what="auto-approve resume allocation") is not None,
          "the auto-approved plan resumes through the allocation with the notice")
    planner.push_results("done", summary="auto-approved plan implemented")
    check(wait_task_done(auth, auto_plan, timeout=30) is not None,
          "the auto-approved plan's task completes")
    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"],
        {"answer": "Review approved. Continue with the task."})
    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 a bot 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"]
                               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 bot 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()
    row = [t for t in slot_public(auth, "/tasks", "tasks")["tasks"]
           if int(t["id"]) == unreachable_row][0]
    check(row.get("unreachable") is True,
          "queue row shows the unreachable state in the window")

    print("== phase 1: startup migrates the per-saddle token ==")
    # Saddles registered before the flip hold legacy api_tokens rows.
    # On startup every saddle row gets its canonical token: the
    # legacy value it already holds (daemons keep their credential)
    # or a freshly minted one. 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"])
    conn = db()
    conn.execute("insert into api_tokens (saddle_id, token, created_at)"
                 " values (?, 'saddle-legacy-copied-token', strftime('%s','now'))",
                 (racer_two_id,))
    conn.execute("update saddles set token = null where name = 'racer-two'")
    conn.execute("insert into saddles (name, host, status, current_task_id)"
                 " values ('tokenless-orphan', '127.0.0.1:1', 'offline', 0)")
    conn.commit()
    conn.close()

    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 saddle_row(name):
        conn = db()
        try:
            return conn.execute("select id, token from saddles where name = ?",
                                (name,)).fetchone()
        finally:
            conn.close()

    check(wait_for(lambda: saddle_row("tokenless-orphan") is not None
                   and (saddle_row("tokenless-orphan")[1] or "").startswith("saddle-")
                   or False, timeout=20, what="tokenless row minted") is not None,
          "startup mints a token for a saddle row without one")
    check(wait_for(lambda: saddle_row("racer-two") is not None
                   and saddle_row("racer-two")[1] == "saddle-legacy-copied-token"
                   or False, timeout=20, what="legacy token copied") is not None,
          "startup copies the legacy api_tokens value onto the row")
    fake2.token = "saddle-legacy-copied-token"  # the daemon asks with it
    status, _ = saddle_api("POST", "/api/saddle/next-task", "saddle-legacy-copied-token")
    check(status in (200, 204),
          "the legacy token still authenticates (copied, not replaced)")
    conn = db()
    conn.execute("delete from saddles where name = 'tokenless-orphan'")
    conn.commit()
    conn.close()

    # The restart wiped the in-memory ask bookkeeping; the fleet's
    # daemons re-establish sleeping with a fresh ask each (the real
    # daemons' idle-awake ask / 15-minute safety self-ask).
    for daemon in (fake2,):
        daemon.ask_now()
    check(True, "fleet re-asks after the restart (sleeping re-established)")


# =============================================================================
# Bundle: push — plan 11 task E: real Web Push over VAPID. VAPID
# persistence, the service worker, subscription CRUD + validation,
# notify-me prefs, the block/lifecycle dispatch matrix (global vs
# opted-in), 410 pruning, retry-next-tick delivery and protocol-level
# payload verification (RFC 8291 decryption + ES256 JWT) against the
# fake push service.
# =============================================================================

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

    def subscription_count(endpoint):
        conn = db()
        try:
            return conn.execute(
                "select count(*) from push_subscriptions where endpoint = ?",
                (endpoint,)).fetchone()[0]
        finally:
            conn.close()

    print("== task E: service worker asset ==")
    status, sw = http("GET", "/sw.js", timeout=15)
    check(status == 200 and "addEventListener('push'" in sw
          and "notificationclick" in sw and "clients.openWindow" in sw,
          "the service worker serves at the origin root with push + click routing")

    print("== task E: VAPID keypair generation + persistence ==")
    config = json.load(open(os.path.join(WORK, "web-config.json")))
    vapid = config.get("vapid") or {}
    check(bool(vapid.get("public_key")) and bool(vapid.get("private_key")),
          "the VAPID keypair persists into web-config.json's vapid section")
    vapid_private_key = ec.derive_private_key(
        int.from_bytes(b64url_dec(vapid["private_key"]), "big"), ec.SECP256R1())
    derived_public = vapid_private_key.public_key().public_bytes(
        Encoding.X962, PublicFormat.UncompressedPoint)
    check(b64url_dec(vapid["public_key"]) == derived_public,
          "the persisted public key matches the private scalar (a valid P-256 pair)")
    dash = slot_public(auth, "/", "dashboard")
    check(dash.get("push_vapid_key") == vapid["public_key"],
          "the dashboard exposes the same VAPID key to subscribers")
    check(dash.get("push_subscribed") is False and bool(dash.get("push_subscribe")),
          "the bell starts unsubscribed and carries the subscribe action")

    print("== task E: subscription CRUD + auth validation ==")
    browser = ec.generate_private_key(ec.SECP256R1())
    browser_public = browser.public_key().public_bytes(
        Encoding.X962, PublicFormat.UncompressedPoint)
    browser_auth = os.urandom(16)
    subscribe_fields = {"p256dh": b64url(browser_public), "auth": b64url(browser_auth)}
    valid_endpoint = "http://127.0.0.1:1/x"
    check("absolute http(s)" in (action_notify(act(auth, dash["push_subscribe"],
        {"endpoint": "ftp://example.com/push", **subscribe_fields})) or ""),
        "a non-http endpoint is rejected")
    check("p256dh" in (action_notify(act(auth, dash["push_subscribe"],
        {"endpoint": valid_endpoint, "p256dh": b64url(os.urandom(64)),
         "auth": subscribe_fields["auth"]})) or ""),
        "a malformed p256dh point is rejected")
    check("auth" in (action_notify(act(auth, dash["push_subscribe"],
        {"endpoint": valid_endpoint, "p256dh": subscribe_fields["p256dh"],
         "auth": b64url(os.urandom(15))})) or ""),
        "a malformed auth secret is rejected")

    push = FakePushService()
    push.serve()
    endpoint_main = push.url("/main")
    act(auth, dash["push_subscribe"],
        {"endpoint": endpoint_main, **subscribe_fields})
    conn = db()
    rows = conn.execute("select user_id, p256dh, auth from push_subscriptions"
                        " where endpoint = ?", (endpoint_main,)).fetchall()
    admin_id = conn.execute("select id from spry_users where username = 'admin'"
                            ).fetchone()[0]
    conn.close()
    check(len(rows) == 1 and rows[0][0] == admin_id,
          "a valid subscription stores against the signed-in user")
    # Re-subscribing the same endpoint replaces (never duplicates).
    rotated_auth = os.urandom(16)
    act(auth, dash["push_subscribe"],
        {"endpoint": endpoint_main, "p256dh": subscribe_fields["p256dh"],
         "auth": b64url(rotated_auth)})
    conn = db()
    rows = conn.execute("select auth from push_subscriptions where endpoint = ?",
                        (endpoint_main,)).fetchall()
    conn.close()
    check(len(rows) == 1 and rows[0][0] == b64url(rotated_auth),
          "re-subscribing an endpoint replaces its keys in place")

    print("== task E: notify-me pref persistence ==")
    act(auth, slot_public(auth, "/tasks", "tasks")["create"],
        {"project_id": "1", "title": "Push pref task", "priority": "5"})
    pref_task = task_id_of("Push pref task")
    act(auth, task_state(auth, pref_task)["toggle_notify"], {})
    conn = db()
    check(conn.execute("select count(*) from task_notify_prefs where task_id = ?"
                       " and user_id = ?", (pref_task, admin_id)).fetchone()[0] == 1,
          "Notify me persists a task_notify_prefs row for the user")
    conn.close()
    check(task_state(auth, pref_task).get("notify_me") is True,
          "the task slot reflects notify_me after the toggle")
    act(auth, task_state(auth, pref_task)["toggle_notify"], {})
    conn = db()
    check(conn.execute("select count(*) from task_notify_prefs where task_id = ?",
                       (pref_task,)).fetchone()[0] == 0,
          "Don't notify me removes the pref row")
    conn.close()
    check(task_state(auth, pref_task).get("notify_me") is False,
          "the task slot reflects the opt-out")

    print("== task E: dispatch matrix — global blocks vs opted-in lifecycle ==")
    # Two "browsers" for admin (/main, /gone) and one for another user
    # (id 999, rows seeded directly — push targeting is per user).
    endpoint_gone = push.url("/gone")
    push.status_for["/gone"] = 410
    act(auth, dash["push_subscribe"],
        {"endpoint": endpoint_gone, **subscribe_fields})
    other_browser = ec.generate_private_key(ec.SECP256R1())
    other_public = other_browser.public_key().public_bytes(
        Encoding.X962, PublicFormat.UncompressedPoint)
    other_auth = os.urandom(16)
    endpoint_other = push.url("/other")
    conn = db()
    conn.execute("insert into push_subscriptions (user_id, endpoint, p256dh, auth,"
                 " created_at) values (999, ?, ?, ?, strftime('%s','now'))",
                 (endpoint_other, b64url(other_public), b64url(other_auth)))
    conn.commit()
    conn.close()

    # Task 1: admin opted in via the UI toggle; the other user is not.
    # Baselines are captured BEFORE the enqueue: the first-start push
    # fires at the allocation, the instant the dispatch lands.
    act(auth, slot_public(auth, "/tasks", "tasks")["create"],
        {"project_id": "1", "title": "Push lifecycle task", "priority": "5"})
    task1 = task_id_of("Push lifecycle task")
    act(auth, task_state(auth, task1)["toggle_notify"], {})
    main_before = len(push.hits("/main"))
    other_before = len(push.hits("/other"))
    gone_before = len(push.hits("/gone"))
    enqueue_task(auth, task1)
    check(wait_for(lambda: fake.running_task == task1 or False, timeout=30,
                   what="task 1 assignment") is not None,
          "task 1 dispatches to the fake daemon")
    task1_session = fake.session_id
    check(wait_for(lambda: len(push.hits("/main")) > main_before, timeout=15,
                   what="first-start push to the opted-in admin") is not None,
          "first start notifies the opted-in user's subscription")
    check(wait_for(lambda: len(push.hits("/gone")) > gone_before, timeout=15,
                   what="first-start push to the gone endpoint") is not None,
          "first start reaches every subscription of the opted-in user")
    check(quiet_window(quiet_seconds(),
                       lambda: len(push.hits("/other")) == other_before,
                       "no first-start push to the not-opted-in user"),
          "first start never reaches users without the notify pref")
    check(wait_task_done(auth, task1, timeout=60) is not None,
          "task 1 completes")
    _, done_payload = decrypt_webpush(push.hits("/main")[-1], browser,
                                      browser_public, rotated_auth)
    check(done_payload["title"] == "Push lifecycle task — completed"
          and done_payload["url"] == "/tasks/%d" % task1,
          "completion notifies the opted-in user (decrypted payload)")
    check(len(push.hits("/other")) == other_before,
          "completion stays with the opted-in user only")
    check(wait_for(lambda: subscription_count(endpoint_gone) == 0, timeout=15,
                   what="410 endpoint pruned") is not None,
          "a 410 Gone endpoint is pruned from push_subscriptions")

    # Block events go to EVERY push-subscribed user globally.
    main_before = len(push.hits("/main"))
    other_before = len(push.hits("/other"))
    saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
               {"type": "block", "session_id": task1_session, "data": {
                   "question": "Global question for everyone?", "type": "question"}})
    check(wait_for(lambda: len(push.hits("/main")) > main_before
                   and len(push.hits("/other")) > other_before, timeout=15,
                   what="global block push") is not None,
          "a raised block notifies every push-subscribed user globally")
    block_hit = push.hits("/other")[-1]
    _, payload = decrypt_webpush(block_hit, other_browser, other_public, other_auth)
    check(payload["title"] == "Push lifecycle task — question raised"
          and payload["body"] == "Global question for everyone?"
          and payload["url"] == "/tasks/%d" % task1,
          "the block payload carries task title + kind, question and click URL (%s)"
          % payload)
    ciphertext_len = len(block_hit["body"]) - (21 + 65)
    envelope = block_hit["body"]
    record_size = int.from_bytes(envelope[16:20], "big")
    # rs = plaintext + delimiter (1) + tag (16) == ciphertext-with-tag len.
    check(record_size == ciphertext_len,
          "the RFC 8291 envelope record size covers payload + delimiter + tag")
    claims = verify_vapid(block_hit, vapid["public_key"], vapid["private_key"])
    check(claims.get("aud") == "http://127.0.0.1:%d" % push.port
          and claims.get("sub", "").startswith("mailto:")
          and claims.get("exp", 0) > time.time(),
          "the VAPID JWT verifies: aud = the endpoint origin, mailto sub, future exp")
    check(block_hit["headers"].get("ttl") == "60"
          and block_hit["headers"].get("content-type") == "application/octet-stream",
          "the push POST carries TTL and the encrypted content type")

    # Task 2: only the OTHER user opted in — start + failure reach them
    # alone. Baseline again before the enqueue (first start fires at
    # the allocation).
    fake.fail_next = True
    act(auth, slot_public(auth, "/tasks", "tasks")["create"],
        {"project_id": "1", "title": "Push failure task", "priority": "5"})
    task2 = task_id_of("Push failure task")
    conn = db()
    conn.execute("insert into task_notify_prefs (task_id, user_id) values (?, 999)",
                 (task2,))
    conn.commit()
    conn.close()
    main_before2 = len(push.hits("/main"))
    other_before2 = len(push.hits("/other"))
    enqueue_task(auth, task2)
    check(wait_for(lambda: fake.running_task == task2 or False, timeout=30,
                   what="task 2 assignment") is not None,
          "task 2 dispatches (scripted to fail)")
    check(wait_for(lambda: task_state(auth, task2)["status"] == "failed"
                   or False, timeout=30, what="task 2 failed") is not None,
          "task 2 fails")
    check(wait_for(lambda: len(push.hits("/other")) >= other_before2 + 2,
                   timeout=15, what="opted-in user's started+failed pushes"),
          "the opted-in user gets first start AND failure for the task")
    started_hit = [hit for hit in push.hits("/other")[other_before2:]][0]
    _, started_payload = decrypt_webpush(started_hit, other_browser,
                                         other_public, other_auth)
    check(started_payload["title"] == "Push failure task — started",
          "the first-start payload names task + kind (%s)" % started_payload)
    _, failure_payload = decrypt_webpush(push.hits("/other")[-1], other_browser,
                                         other_public, other_auth)
    check(failure_payload["title"] == "Push failure task — failed"
          and failure_payload["url"] == "/tasks/%d" % task2,
          "the failure payload names the task and links its page")
    check(len(push.hits("/main")) == main_before2,
          "an agent failure never reaches users without the pref")

    print("== task E: retry next tick (best-effort delivery) ==")
    endpoint_flaky = push.url("/flaky")
    act(auth, dash["push_subscribe"],
        {"endpoint": endpoint_flaky, **subscribe_fields})
    push.fail_next["/flaky"] = 1
    flaky_before = len(push.hits("/flaky"))
    saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
               {"type": "block", "session_id": task1_session, "data": {
                   "question": "Flaky delivery question?", "type": "question"}})
    check(wait_for(lambda: len(push.hits("/flaky")) >= flaky_before + 2,
                   timeout=20, what="flaky endpoint retried") is not None,
          "a failed send retries on the next dispatcher tick and lands")

    print("== task E: waiting-on-drafts notifies like a block event ==")
    fake.suspend_on_title = "Push parked task"
    act(auth, slot_public(auth, "/tasks", "tasks")["create"],
        {"project_id": "1", "title": "Push parked task", "priority": "5"})
    task3 = task_id_of("Push parked task")
    enqueue_task(auth, task3)
    check(wait_for(lambda: fake.running_task == task3 or False, timeout=30,
                   what="task 3 assignment") is not None,
          "task 3 dispatches (scripted to suspend)")
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority,"
                 " parent_task_id, created_at)"
                 " values (1, 'Push draft subtask', 'draft', 5, ?,"
                 " strftime('%s','now'))", (task3,))
    conn.commit()
    conn.close()
    main_before = len(push.hits("/main"))
    other_before = len(push.hits("/other"))
    check(wait_for(lambda: task_state(auth, task3)["status"] == "suspended"
                   or False, timeout=30, what="task 3 suspended") is not None,
          "task 3 parks (suspended)")
    check(wait_for(lambda: len(push.hits("/main")) > main_before
                   and len(push.hits("/other")) > other_before, timeout=15,
                   what="waiting-on-drafts global push") is not None,
          "the waiting-on-drafts condition notifies globally like a block event")
    _, waiting_payload = decrypt_webpush(push.hits("/main")[-1], browser,
                                         browser_public, rotated_auth)
    check(waiting_payload["title"]
          == "Push parked task — waiting on subtasks to be enqueued"
          and "still in draft" in waiting_payload["body"]
          and waiting_payload["url"] == "/tasks/%d" % task3,
          "the waiting-on-drafts payload carries the clear wording + link (%s)"
          % waiting_payload)
    # A subtask created under the ALREADY-parked parent re-arms the push.
    main_before = len(push.hits("/main"))
    saddle_api("POST", "/api/saddle/subtask", fake_token,
               {"parent_task_id": task3, "title": "Parked-late child"})
    check(wait_for(lambda: len(push.hits("/main")) > main_before, timeout=15,
                   what="waiting-on-drafts re-armed push") is not None,
          "a draft subtask created under a parked parent re-arms the push")
    check(task_state(auth, task_id_of("Parked-late child"))["status"] == "draft",
          "the parked parent's child lands in draft (no auto-enqueue)")

    print("== task E: unsubscribe removes the subscription ==")
    dash = slot_public(auth, "/", "dashboard")
    check(dash.get("push_subscribed") is True,
          "the bell shows subscribed while endpoints exist")
    act(auth, dash["push_unsubscribe"], {"endpoint": endpoint_main})
    check(subscription_count(endpoint_main) == 0,
          "unsubscribing removes the endpoint's row")


BUNDLES_ORDER = ("pages", "dispatch", "subtasks", "lineage",
                 "allocation", "lifecycle", "answers", "push")


def run_bundle(name):
    """Runs one shard to completion; returns a failure count."""
    started = time.monotonic()
    ctx = start_bundle(name)
    runner = globals()["bundle_" + name]
    try:
        runner(ctx)
    finally:
        # Exit-code diagnostic: distinguishes a crashed app (negative
        # returncode, e.g. -11 SIGSEGV) from a clean shutdown. When
        # tearing down mid-exception, also capture the still-alive
        # app's stacks before killing it (a wedged-but-alive server
        # otherwise leaves no evidence).
        print("(app returncode at teardown: %s)" % ctx.app.poll())
        if ctx.app.poll() is None and sys.exc_info()[0] is not None:
            dump_app_stacks(ctx.app.pid)
        ctx.fake.kill()
        try:
            ctx.app.terminate()
            try:
                ctx.app.wait(timeout=5)
            except subprocess.TimeoutExpired:
                ctx.app.kill()
            check(ctx.app.poll() in (0, -15),
                  "bundle %s app shut down cleanly (exit %s)"
                  % (name, ctx.app.poll()))
        except OSError:
            pass
    print("%d passed, %d failed" % (passes, failures))
    print("[bundle %s] %.1fs" % (name, time.monotonic() - started))
    return failures


def run_all_parallel():
    """The default meson entry: one subprocess per bundle (each with its
    own app, daemons and database), verdicts aggregated into one exit
    code and one combined check-count summary."""
    base = os.environ.get("WEBSTACK_WORK", "/tmp/kilo/webstack-e2e")
    log_dir = os.path.join(os.path.dirname(base) or "/tmp/kilo",
                           "webstack-shards")
    os.makedirs(log_dir, exist_ok=True)
    procs = {}
    started = time.monotonic()
    for name in BUNDLES_ORDER:
        out_path = os.path.join(log_dir, "bundle-%s.log" % name)
        handle = open(out_path, "w")
        procs[name] = (subprocess.Popen(
            [sys.executable, "-u", os.path.abspath(__file__), APP,
             "--bundle", name],
            stdout=handle, stderr=subprocess.STDOUT), handle, out_path)
    total_pass = total_fail = 0
    broken = []
    for name, (proc, handle, out_path) in procs.items():
        try:
            proc.wait(timeout=SHARD_TIMEOUT_S)
        except subprocess.TimeoutExpired:
            proc.kill()
            proc.wait()
        handle.close()
        text = open(out_path).read()
        match = re.search(r"(\d+) passed, (\d+) failed", text)
        took = re.search(r"\[bundle %s\] ([0-9.]+)s" % name, text)
        bundle_pass = int(match.group(1)) if match else 0
        bundle_fail = int(match.group(2)) if match else -1
        total_pass += max(bundle_pass, 0)
        total_fail += max(bundle_fail, 0)
        status = "ok"
        if proc.returncode != 0 or bundle_fail != 0 or match is None:
            status = "BROKEN (exit %s)" % proc.returncode
            broken.append(name)
        print("[shard %-10s] %6ss  %3d passed %3d failed  %s"
              % (name, took.group(1) if took else "?",
                 bundle_pass, bundle_fail, status))
        if status != "ok":
            tail = "\n".join(text.rstrip().split("\n")[-60:])
            print("--- web-stack/%s tail (full log: %s) ---" % (name, out_path))
            print(tail)
            print("--- end web-stack/%s tail ---" % name)
    print("shards wall: %.1fs  (logs under %s)" % (time.monotonic() - started, log_dir))
    print("ALL SHARDS: %d passed, %d failed%s"
          % (total_pass, total_fail,
             "" if not broken else "  BROKEN: " + ", ".join(broken)))
    return 1 if broken or total_fail else 0


def main():
    if not os.path.exists(APP):
        raise SystemExit("app binary not found at %s (build first)" % APP)
    argv = sys.argv[2:]
    if "--bundle" in argv:
        sys.exit(1 if run_bundle(argv[argv.index("--bundle") + 1]) else 0)
    if os.environ.get("WEBSTACK_SEQUENTIAL"):
        # Debug path: every bundle in-process, one after another.
        for name in BUNDLES_ORDER:
            if run_bundle(name):
                return 1
        return 0
    return run_all_parallel()


if __name__ == "__main__":
    sys.exit(main())