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 like a phase-1 daemon),
  POST /start (captures the payload — the rollback path), GET /status
  (scripted state machine), GET /output (empty), GET /log?since= (raw
  PTY chunks, one new chunk per request so since-paging is exercised)

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: session_raw_log polling + 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

Prints PASS/FAIL lines Statum-style; exits non-zero on any failure.
"""
import json
import os
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

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")
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 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 = json.loads(directive["snapshot_frame"]["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.5):
    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))
    return None


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


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

    Phase-1 ask behaviour: the daemon asks POST /api/saddle/next-task
    whenever idle-awake — at registration (its token arrives via
    /register), on POST /wake, after its task reaches a terminal state,
    and whenever a /status poll observes it parked idle (throttled).
    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.chunks = []               # [{seq, text}] — the raw log ring
        self.next_seq = 1
        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.pending_output = []        # entries drained by the next /output poll
        self.output_seq = 100
        self.last_activity_epoch = int(time.time())
        self.subtask_status_events = []  # recorded POST /subtask-status payloads
        self.fail_subtask_status = False  # exercise the web's retry path
        self.hang_next = {}             # path suffix -> hang count (never responds)
        self.fail_log_count = 0          # /log requests failing (500) while > 0
        self.scripted_log = None         # fixed chunk set served by /log
        self.scripted_next = 0           # the daemon-style next for it
        self.scripted_corpus = None      # chunks "produced" when a task starts
        self.race_texts = None           # chunks appended DURING /log response
                                        # construction (the daemon-side bug)
        self.answer_bodies = []          # recorded POST /answer payloads
        self.abort_failures = 0          # /abort requests failing (500) while > 0
        self.abort_log = []              # task ids aborted via POST /abort
        self.running_task = 0            # task id the daemon believes it runs
        self.running_title = ""          # its title (suspend targeting)
        self.zombie_409 = 0              # /start 409s once per unit (then decrements)
        self.conflict_running = 0        # while > 0, /start persistently 409s with this id
        self.hold_state = False          # freeze /status lifecycle transitions
        self.ssh_public_key = ""         # reported on /status + /register
        self.started_mono = time.monotonic()
        self.freeze_activity = False     # hold last_activity_epoch still
        # --- 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.serve_register = True         # False: 404 /register (pre-flip daemon)
        self.last_ask_trigger = 0.0
        self.port = None

    def set_state(self, state, summary=None):
        """Advances the scripted lifecycle.

        Time-based on purpose: the web app may probe /status more than
        once per tick (status polls, stale-busy healing, manual
        refreshes), and a poll-counted machine would conflate them and
        race dependent sections.
        """
        self.state = state
        if summary is not None:
            self.summary = summary
        self.state_entered = time.monotonic()

    def queue_output(self, entry_type, command, output):
        """Queues a typed entry for the dispatcher's next /output poll."""
        with self.lock:
            self.pending_output.append({
                "id": self.output_seq,
                "command": command,
                "output": output,
                "exit_code": 1,
                "type": entry_type,
                "timestamp": "2026-09-06T00:00:00Z",
            })
            self.output_seq += 1

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

    def append_chunk(self, text):
        with self.lock:
            self.chunks.append({"seq": self.next_seq, "text": text})
            self.next_seq += 1

    def log_since(self, since):
        with self.lock:
            if self.scripted_log is not None:
                chunks = [c for c in self.scripted_log if c["seq"] >= since]
                return chunks, self.scripted_next
            # `since` is the inclusive resume point (wire contract: a
            # response ending at seq 12 carries next=13, and polling
            # with since=13 must still deliver the chunk numbered 13).
            chunks = [c for c in self.chunks if c["seq"] >= since]
            return chunks, self.next_seq

    # --- 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)."""
        def run():
            try:
                self.ask_now()
            except OSError:
                pass  # app restarting / shutting down mid-test
        threading.Thread(target=lambda: (time.sleep(0.2), run()), daemon=True).start()

    def maybe_trigger_ask(self, reason):
        now = time.monotonic()
        if 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 ""
        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.append_chunk(self.PROMPT)
        self.append_chunk("echo hello\r\n")
        self.append_chunk("hello\r\n")
        self.set_state("failing" if self.fail_next else "working")
        self.fail_next = False
        if self.scripted_corpus is not None:
            with self.lock:
                for chunk in self.scripted_corpus:
                    self.scripted_log.append(dict(chunk))
                self.scripted_next = self.scripted_corpus[-1]["seq"] + 1

    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):
            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.
                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 == "/start":
                if self.daemon.conflict_running > 0 or self.daemon.zombie_409 > 0:
                    # Wire contract: a machine-readable conflict.
                    running = self.daemon.conflict_running
                    if running == 0:
                        running = self.daemon.running_task
                        self.daemon.zombie_409 -= 1
                    body = json.dumps({"ok": False, "error": "busy",
                                       "state": "working",
                                       "running_task_id": running}).encode()
                    self.send_response(409)
                    self.send_header("Content-Type", "application/json")
                    self.send_header("Content-Length", str(len(body)))
                    self.end_headers()
                    self.wfile.write(body)
                    return
                try:
                    payload = json.loads(raw)
                except ValueError:
                    payload = {"raw": raw}
                self.daemon.accept_assignment(payload)
                self._send(200, {})
            elif self.path == "/answer":
                try:
                    self.daemon.answer_bodies.append(json.loads(raw))
                except ValueError:
                    pass
                self._send(200, {})
            elif self.path == "/abort":
                if self.daemon.abort_failures > 0:
                    self.daemon.abort_failures -= 1
                    self._send(500, {"error": "abort failed (simulated)"})
                    return
                aborted = self.daemon.running_task
                self.daemon.set_state("idle")
                self.daemon.summary = ""
                self.daemon.running_task = 0
                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 self.path == "/subtask-status":
                try:
                    payload = json.loads(raw)
                except ValueError:
                    self._send(400, {"error": "bad json"})
                    return
                if self.daemon.fail_subtask_status:
                    self._send(500, {"error": "simulated failure"})
                    return
                self.daemon.subtask_status_events.append(payload)
                self._send(200, {"ok": True})
            else:
                self._send(404, {"error": "no such route"})

        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 wedge signal
                # must reflect loop rounds / command completions.
                self.daemon.touch_activity()
            if self.path == "/status":
                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
                if self.daemon.hold_state:
                    self._send(200, response)
                    return
                if self.daemon.state == "failing":
                    # Fails fast (a rejected assignment script).
                    if time.monotonic() - self.daemon.state_entered >= 1:
                        self.daemon.set_state("failed",
                                              "bash -n rejected the script (simulated)")
                elif self.daemon.state == "working":
                    # Time-based, not poll-based: whoever probes /status
                    # (dispatcher polls, stale-busy healing, refreshes)
                    # must not advance the lifecycle.
                    if time.monotonic() - self.daemon.state_entered >= 8:
                        if self.daemon.suspend_on_title \
                                and self.daemon.suspend_on_title == self.daemon.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.daemon.set_state("suspended")
                            self.daemon.suspend_on_title = None
                        else:
                            self.daemon.set_state("done", "fake daemon finished")
                # 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")
                self._send(200, response)
            elif self.path.startswith("/output"):
                with self.daemon.lock:
                    entries = self.daemon.pending_output
                    self.daemon.pending_output = []
                self._send(200, entries)
            elif self.path.startswith("/log"):
                if self.daemon.fail_log_count > 0:
                    self.daemon.fail_log_count -= 1
                    self._send(500, {"error": "log unavailable (simulated)"})
                    return
                query = urllib.parse.urlparse(self.path).query
                since = int(urllib.parse.parse_qs(query).get("since", ["0"])[0])
                # The daemon-side snapshot race: the served array is
                # snapshotted, chunks are appended, and `next` is computed
                # AFTER the appends — the response's next skips chunks the
                # array never carried (the live run's 33 lost chunks).
                raced = False
                if since > 0 and self.daemon.race_texts:
                    raced = True
                    with self.daemon.lock:
                        next_seq = self.daemon.scripted_next
                        for text in self.daemon.race_texts:
                            self.daemon.scripted_log.append(
                                {"seq": next_seq, "text": text})
                            next_seq += 1
                        self.daemon.scripted_next = next_seq
                        self.daemon.race_texts = None
                # Simulate the live PTY: each poll produces one more chunk
                # (before slicing, so the poller actually sees it).
                self.daemon.append_chunk("stream-line-%d\r\n" % self.daemon.next_seq)
                chunks, nxt = self.daemon.log_since(since)
                if raced:
                    nxt = self.daemon.scripted_next
                self._send(200, {"chunks": chunks, "next": nxt})
            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, not when the web happens to poll):
        working ages into done/suspended, failing into failed — and a
        task END makes the idle-awake daemon ask for work again."""
        while True:
            time.sleep(0.5)
            if self.hold_state:
                continue
            if self.state == "failing":
                if time.monotonic() - self.state_entered >= 1:
                    self.set_state("failed", "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 >= 8:
                    if self.suspend_on_title and self.suspend_on_title == self.running_title:
                        self.set_state("suspended")
                        self.suspend_on_title = None
                        # 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")
                        if self.ask_on_terminal:
                            self.maybe_trigger_ask("terminal")

    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):
        """Stops listening: subsequent connections are refused — a daemon
        that died mid-task."""
        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": "5",
        "SADDLE_RAW_PROBE_INTERVAL_S": "5",
        "SADDLE_STRAND_GRACE_S": "10",
        "SADDLE_WEDGE_THRESHOLD_S": "8",
        "SADDLE_LOST_WAKE_S": "2",
        # 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


def main():
    if not os.path.exists(APP):
        raise SystemExit("app binary not found at %s (build first)" % APP)

    fake = FakeSaddle()
    fake.serve()

    # 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
    # sidesteps it. Only tolerated while nothing has been checked yet.
    auth = None
    app = None
    for attempt in range(3):
        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)

    try:
        check(True, "login yields an auth slot")
        print("== feature 3: plan review hides duplicate body ==")
        state = slot_public(auth, "/projects", "projects")
        act(auth, state["create"], {
            "name": "demo",
            "git_url": "https://example.com/demo.git",
            "default_branch": "main",
        })
        conn = db()
        conn.execute("insert into tasks (project_id, title, status, priority,"
                     " is_design_task, plan, created_at)"
                     " values (1, 'Plan review task', 'working', 5, 1, '# The plan\n\n1. Do it',"
                     " strftime('%s','now'))")
        conn.commit()
        plan_task = conn.execute("select max(id) from tasks").fetchone()[0]

        saddles_state = slot_public(auth, "/saddles", "saddles")
        act(auth, saddles_state["register"], {"name": "fake-saddle",
                                              "host": "127.0.0.1:%d" % DAEMON_PORT})
        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")
        fake_id = int(registered[0]["id"])
        fake_token = registered[0]["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)")

        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": "rfc",
                                      "body": "# The plan\n\n1. Do it"}})
        check(status == 200 and body.get("block_id"), "plan-review 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["rfc"].get("is_plan_review") is True,
              "rfc block flagged as the plan review")
        check(blocks["rfc"].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")

        act(auth, blocks["rfc"]["answer_action"], {"answer": "Plan approved. Proceed."})
        detail = task_state(auth, plan_task)
        rfc = [b for b in detail["blocks"] if b["block_type"] == "rfc"][0]
        check(rfc.get("show_body") is True,
              "answered plan review shows the body again (review section gone)")

        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": "Human approved the review"})
        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("== 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")
        act(auth, tasks_state["create"], {
            "project_id": "1",
            "title": "Rules dispatch",
            "description": "Effective rule set rides the /start payload.",
            "priority": "2",
        })
        rule_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                         if t["title"] == "Rules dispatch"][0])
        detail = task_state(auth, rule_task)
        act(auth, detail["attach_rule"], {"rule_id": str(rule_id("Write Vala docs"))})
        act(auth, detail["attach_rule"], {"rule_id": str(rule_id("Use Invercargill"))})
        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()

        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "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()
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "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()
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "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("== raw log polling + terminal raw view ==")
        # Wait for the last dispatch to finish so its final polls ran.
        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 daemon's chunks (%d)" % len(rows))
        seqs = [r[0] for r in rows]
        check(seqs == sorted(seqs) and len(set(seqs)) == len(seqs),
              "raw 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,
              "raw log keeps the prompt and echoed output byte-exact")
        check("stream-line-" in text, "incremental /log polls append later chunks")

        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.
        conn = db()
        conn.execute("insert into session_commands (session_id, command_text, output_text,"
                     " type, started_at) values (?, 'ls', 'old-output', 'command',"
                     " strftime('%s','now'))", (session_id,))
        conn.commit()
        conn.close()
        legacy = slot_public(auth, "/tasks/%d/terminal" % plan_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 raw chunks in seq order (prompt first)")

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

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

        fake_card = saddle_card("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")
        act(auth, tasks_state["create"], {
            "project_id": "1",
            "title": "GPU task",
            "priority": "2",
        })
        gpu_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                        if t["title"] == "GPU task"][0])
        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")

        act(auth, tasks_state["create"], {
            "project_id": "1",
            "title": "Any-saddle task",
            "priority": "3",
        })
        any_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                        if t["title"] == "Any-saddle task"][0])

        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")
        with open(os.path.join(WORK, "app.log")) as handle:
            app_log = handle.read()
        check("waits for a capable saddle" in app_log and "requires: gpu" in app_log,
              "dispatcher logs the capability wait")

        # Attaching the tag to the saddle releases the task.
        fake_card = saddle_card("fake-saddle")
        act(auth, fake_card["attach_capability"], {"capability_id": str(int(caps["gpu"]["id"]))})
        fake_card = saddle_card("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.
        act(auth, tasks_state["create"], {
            "project_id": "1",
            "title": "Big task",
            "priority": "2",
        })
        big_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                        if t["title"] == "Big task"][0])
        detail = task_state(auth, big_task)
        act(auth, detail["attach_capability"], {"capability_id": str(int(caps["gpu"]["id"]))})
        act(auth, detail["attach_capability"], {"capability_id": str(int(caps["bigmem"]["id"]))})
        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("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; delete a capability everywhere.
        detail = task_state(auth, big_task)
        act(auth, [c for c in detail["capabilities"] if c["name"] == "bigmem"][0]["detach"], {})
        check([c["name"] for c in task_state(auth, big_task)["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("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")

        act(auth, tasks_state["create"], {
            "project_id": "1",
            "title": "Cap-on-create task",
            "capability_id": [gpu_option["id"]],
        })
        cap_create = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                          if t["title"] == "Cap-on-create task"][0])
        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("== B6: computed display state ==")
        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("== retry + resume ==")
        # A daemon-side failure fails the task with its explanation.
        fake.fail_next = True
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Flaky task",
            "priority": "2",
        })
        flaky = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                     if t["title"] == "Flaky task"][0])
        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")
        check(wait_task_done(auth, flaky, timeout=60) is not None,
              "retried task dispatches again and completes")
        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.
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Stuck task",
            "priority": "2",
        })
        stuck = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                     if t["title"] == "Stuck task"][0])
        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")
        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")
        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("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")
        check(wait_task_done(auth, stuck, timeout=60) is not None,
              "requeued task resumes on the refreshed saddle and completes")

        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" % stuck, 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" % stuck, 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 from /output flows into the audit
        # trail without failing (or finishing) the task.
        fake.queue_output("rejected", "echo 'too long' | saddle",
                          "max_tokens exceeded (recoverable)")
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Rejected ride",
            "priority": "2",
        })
        rejected_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                             if t["title"] == "Rejected ride"][0])
        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("== B7: suspended state + per-subtask resume ==")
        fake.suspend_on_title = "Suspend parent"
        pushes_before = len(fake.start_payloads)
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Suspend parent",
            "priority": "2",
        })
        parent7 = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                       if t["title"] == "Suspend parent"][0])
        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")
        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"
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Push parent",
            "priority": "2",
        })
        push_parent = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                           if t["title"] == "Push parent"][0])
        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")

        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Push child",
            "parent_task_id": str(push_parent),
        })
        push_child = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                          if t["title"] == "Push child"][0])

        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" and "finished" in (e.get("summary") or "")
                                   for e in events_for(push_child)) or False,
                       timeout=90, what="done subtask event") is not None,
              "subtask completion pushes done with the summary")
        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"
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Push child two",
            "parent_task_id": str(push_parent),
        })
        push_child_two = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                              if t["title"] == "Push child two"][0])
        time.sleep(7)
        check(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")


        print("== R3 B2: stale-busy saddle healing ==")
        # The observed incident: a saddle left busy with current_task_id 0.
        conn = db()
        conn.execute("update saddles set status = 'busy', current_task_id = 0,"
                     " allocated_at = strftime('%s','now') where name = 'fake-saddle'")
        conn.commit()
        conn.close()
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Heal dispatch",
            "priority": "2",
        })
        heal_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                         if t["title"] == "Heal dispatch"][0])
        check(wait_task_done(auth, heal_task, timeout=60) is not None,
              "busy-without-task heals and the queued task dispatches")
        with open(os.path.join(WORK, "app.log")) as handle:
            app_log = handle.read()
        check("Healing stale-busy saddle" in app_log
              and "busy with no current task" in app_log,
              "the busy@0 heal is logged")

        # Contradiction: the daemon reports idle while the web believes the
        # saddle is running the orphaned task. Drain everything in flight
        # first — forcing the daemon's state mid-run requeues whatever it
        # is really executing.
        def in_flight_count():
            conn = db()
            try:
                # The plan-review fixture (task 1) is inserted 'working'
                # on purpose and never dispatches; it is not in flight.
                return conn.execute(
                    "select count(*) from tasks where status in"
                    " ('queued', 'assigned', 'working')"
                    " and title != 'Plan review task'").fetchone()[0]
            finally:
                conn.close()

        def in_flight_titles():
            conn = db()
            try:
                return [r[0] for r in conn.execute(
                    "select title from tasks where status in"
                    " ('queued', 'assigned', 'working')"
                    " and title != 'Plan review task'")]
            finally:
                conn.close()
        check(wait_for(lambda: in_flight_count() == 0 or False,
                       timeout=90, what="in-flight work settles (%s)" % in_flight_titles()) is not None,
              "in-flight work settles before the contradiction run")
        conn = db()
        conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
                     " values (1, 'Orphan task', 'working', 2, strftime('%s','now'))")
        orphan = conn.execute("select max(id) from tasks").fetchone()[0]
        conn.execute("update saddles set status = 'busy', current_task_id = ?,"
                     " allocated_at = strftime('%s','now') where name = 'fake-saddle'",
                     (orphan,))
        conn.commit()
        conn.close()
        fake.state = "idle"
        check(wait_task_done(auth, orphan, timeout=60) is not None,
              "contradicted saddle heals, orphan requeues and completes")
        with open(os.path.join(WORK, "app.log")) as handle:
            app_log = handle.read()
        check("daemon idle while the web believes" in app_log,
              "the contradiction heal is logged")


        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"
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Callback parent",
            "priority": "2",
        })
        cb_parent = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                         if t["title"] == "Callback parent"][0])
        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")

        anchor = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                      if t["title"] == "Heal dispatch"][0])
        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("== R4 B2: subtask auto-approve ==")
        # 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")

        # 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"
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Auto parent",
            "priority": "2",
        })
        auto_parent = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                           if t["title"] == "Auto parent"][0])
        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"])
        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)})

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

        # 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 (lift the capability requirement first).
        act(auth, task_state(auth, auto_child)["capabilities"][0]["detach"], {})
        act(auth, task_state(auth, auto_grand)["capabilities"][0]["detach"], {})

        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("== 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 (the auto-approve
        # stragglers must finish — their parked resumes stay offerable
        # otherwise and would steal the race's asks), freeze every
        # daemon's self-ask triggers, put all three to sleep (204),
        # then fire one task worth of concurrent asks.
        check(wait_task_done(auth, auto_child, timeout=90) is not None
              and wait_task_done(auth, auto_grand, timeout=90) is not None
              and wait_task_done(auth, auto_parent, timeout=90) is not None,
              "the auto-approve stragglers settle before the race")
        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
        time.sleep(1.5)  # let any in-flight triggered asks land
        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")

        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Race task",
            "priority": "2",
        })
        race_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                         if t["title"] == "Race task"][0])

        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"
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Affinity parent",
            "priority": "2",
            "capability_id": [str(affinity_cap)],
        })
        affinity_parent = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                               if t["title"] == "Affinity parent"][0])
        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 child's completion is stamped
        # slightly in the future: the poller may not have observed (and
        # stamped) the parent's suspension yet, and the resume trigger
        # needs completed_at > suspended_at. Freeze racer-two's ask
        # triggers so the manual ask below is the only contender (its
        # park-time idle-status ask would otherwise race this claim).
        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','+15 seconds'),"
                     " strftime('%s','now'))", (affinity_parent,))
        conn.commit()
        conn.close()
        fake2.ask_on_wake = False
        fake2.ask_on_idle_status = False
        time.sleep(1.2)  # any in-flight park-time trigger ask lands (204)
        act(auth, racer3_card["attach_capability"], {"capability_id": str(affinity_cap)})
        status, _ = fake3.ask_now()
        check(status == 204,
              "a tag-capable non-affine saddle is refused the parked task (204)")

        wakes_before_2 = len(fake2.wake_log)
        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
        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")

        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Lost wake task",
            "priority": "2",
            "capability_id": [str(lostwake_cap)],
        })
        lost_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                         if t["title"] == "Lost wake task"][0])
        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("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

        # The racers leave the dispatch pool: the remaining sections
        # (rollback resume, R5-R11) predate multi-daemon routing and
        # assert against the primary fake's payloads. racer-three is
        # already offline (the lost-wake guard); racer-two joins it.
        conn = db()
        conn.execute("update saddles set status = 'offline' where name = 'racer-two'")
        conn.commit()
        conn.close()

        # Tasks caught working on a saddle that went offline requeue with a
        # bot comment each tick.
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Offline saddle task",
            "priority": "2",
        })
        offline_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                            if t["title"] == "Offline saddle task"][0])
        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")
        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")
        check(wait_task_done(auth, offline_task, timeout=60) is not None,
              "requeued task completes again on the healthy daemon")


        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()
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Lineage subtask",
            "priority": "3",
        })
        lineage_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                            if t["title"] == "Lineage subtask"][0])
        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) == offline_task]
        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")

        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)
        other_project = int([p["id"] for p in slot_public(auth, "/projects", "projects")["projects"]
                             if p["name"] == "diff-fabrica"][0])
        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")


        print("== R6 F1: a hanging request cannot freeze the tick ==")
        # One wedged /output (never responds) blocks a poll; the Soup
        # timeout must fail it and later ticks keep dispatching.
        task_state(auth, lineage_task)
        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()
        fake.hang_next["/output"] = 1
        # Dispatch a task first so a poll is in flight against the hang,
        # then verify a later task still dispatches (tick not frozen).
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Hang runner",
            "priority": "3",
        })
        hang_runner = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                           if t["title"] == "Hang runner"][0])
        check(wait_task_done(auth, hang_runner, timeout=90) is not None,
              "dispatching task survives a hung daemon request")
        # The hung poll itself fails after the timeout; its task requeues
        # or completes on later ticks — 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"
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Parked refresh parent",
            "priority": "2",
        })
        parked = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                      if t["title"] == "Parked refresh parent"][0])
        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("fake-saddle")
        act(auth, fake_card["refresh"], {})
        fake_card = saddle_card("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("== R6 F3: raw-log probe retries instead of deciding once ==")
        fake.fail_log_count = 1  # the seed-time probe fails exactly once
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Raw retry task",
            "priority": "2",
        })
        raw_retry = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                         if t["title"] == "Raw retry task"][0])

        def raw_rows_for(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_rows_for(raw_retry) > 0 or False,
                       timeout=60, what="raw log rows appear after retry") is not None,
              "a failed seed probe retries and the raw log fills")
        check(wait_task_done(auth, raw_retry, timeout=60) is not None,
              "raw retry task completes")


        print("== R6 F4: stranded assigned recovery ==")
        # The live incident's shape: task assigned, session open, zero
        # output ever, daemon idle. Grace period 10s in the harness env.
        fake.state = "idle"
        conn = db()
        conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
                     " values (1, 'Stranded assigned', 'assigned', 2, strftime('%s','now'))")
        stranded = conn.execute("select max(id) from tasks").fetchone()[0]
        conn.execute("insert into sessions (task_id, saddle_id, started_at)"
                     " values (?, ?, strftime('%s','now','-180 seconds'))", (stranded, fake_id))
        # Saddle row idle (the busy@task shape is the stale-busy heal's
        # case); this exercises the strand pass specifically.
        conn.execute("update saddles set status = 'idle', current_task_id = 0,"
                     " allocated_at = null where name = 'fake-saddle'")
        conn.commit()
        conn.close()
        check(wait_for(lambda: any(c["body"] == "task stranded with no daemon output; task requeued"
                                   for c in task_state(auth, stranded).get("comments", []))
                       or False, timeout=40, what="stranded requeue comment") is not None,
              "stranded assignment requeues with a bot comment")
        check(wait_task_done(auth, stranded, timeout=90) is not None,
              "recovered task dispatches and completes")


        print("== R7 B1: cancel reaches the daemon ==")
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Cancel live",
            "priority": "2",
        })
        cancel_live = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                           if t["title"] == "Cancel live"][0])
        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 /abort 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")

        # Transport/500 failure on abort fences the saddle; recovery probe
        # clears the fence; nothing dispatches while fenced.
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Fence runner",
            "priority": "2",
        })
        fence_runner = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                            if t["title"] == "Fence runner"][0])
        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 fenced"
                  for c in task_state(auth, fence_runner).get("comments", [])),
              "failed abort narrates the fence")
        check(saddle_card("fake-saddle")["status"] == "recovering",
              "saddle fenced as recovering")
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "During fence",
            "priority": "3",
        })
        during_fence = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                            if t["title"] == "During fence"][0])
        time.sleep(7)
        check(task_state(auth, during_fence)["status"] == "queued",
              "fenced saddle is not dispatchable")
        # A real daemon lands idle once its (unaborted) round ends; the
        # fake's scripted lifecycle parks on 'done', so model that.
        fake.state = "idle"
        check(wait_for(lambda: saddle_card("fake-saddle")["status"] == "idle" or False,
                       timeout=30, what="fence clears") is not None,
              "an idle probe clears the fence")
        check(wait_task_done(auth, during_fence, timeout=60) is not None,
              "queued work dispatches once the fence clears")


        print("== phase 1: rollback resume path (pre-flip daemon) ==")
        # Phase 1 keeps the /answer + /start resume machinery for daemons
        # that never ask (the rollback route; the 409-aware /start push
        # machinery itself is rollback-only now that dispatch flows
        # through asks). A pre-flip daemon's parked suspended parent is
        # resumed by a /start push carrying the subtask report.
        fake4.ask_on_register = False
        fake4.ask_on_wake = False
        fake4.ask_on_terminal = False
        fake4.ask_on_idle_status = False
        fake4.serve_register = False  # a pre-flip daemon has no /register
        registers_before = len(fake4.register_log)
        saddles_state = slot_public(auth, "/saddles", "saddles")
        act(auth, saddles_state["register"], {"name": "preflip-daemon",
                                              "host": "127.0.0.1:%d" % fake4.port})
        preflip_card = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
                        if s["name"] == "preflip-daemon"][0]
        preflip_id = int(preflip_card["id"])
        check(len(fake4.register_log) == registers_before,
              "the pre-flip daemon 404s /register (registration tolerated)")

        conn = db()
        conn.execute("insert into tasks (project_id, title, status, priority, suspended_at,"
                     " created_at) values (1, 'Preflip parked parent', 'suspended', 2,"
                     " strftime('%s','now','-60 seconds'), strftime('%s','now','-120 seconds'))")
        preflip_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','-90 seconds'))",
                     (preflip_parent, preflip_id))
        conn.execute("insert into tasks (project_id, title, status, priority, parent_task_id,"
                     " summary, completed_at, created_at) values (1, 'Preflip child', 'done',"
                     " 5, ?, 'child of the pre-flip era',"
                     " strftime('%s','now','-30 seconds'),"
                     " strftime('%s','now','-100 seconds'))", (preflip_parent,))
        conn.commit()
        conn.close()

        pushes_before = len(fake4.start_payloads)
        check(wait_for(lambda: len(fake4.start_payloads) > pushes_before or False,
                       timeout=45, what="pre-flip /start resume push") is not None,
              "the pre-flip daemon's parked parent resumes via POST /start")
        preflip_push = [p for p in fake4.start_payloads
                        if int(p.get("task", {}).get("id") or 0) == preflip_parent][-1]
        check("Preflip child" in (preflip_push.get("resume_notice") or "")
              and "child of the pre-flip era" in (preflip_push.get("resume_notice") or ""),
              "the rollback /start resume carries the subtask report notice")
        check(wait_task_done(auth, preflip_parent, timeout=60) is not None,
              "the pre-flip parent completes through the rollback path")


        print("== R7 B3: wedged daemon, health panel, raw log, vanishing daemon ==")
        # Raw-log polling starts from the FIRST push (no probe failure):
        # a normally dispatched task accumulates raw rows immediately.
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Raw from first push",
            "priority": "2",
        })
        raw_first = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                         if t["title"] == "Raw from first push"][0])

        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,
              "raw-log polling runs from the first push (session-4 gap)")
        check(wait_task_done(auth, raw_first, timeout=60) is not None,
              "raw-first task completes")

        # Health panel: after the polls above the snapshot knows the fake.
        card = saddle_card("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")

        # Alive-but-wedged: a working daemon whose last activity is stale
        # (threshold 3s in the harness) fences the saddle and requeues the
        # task with a bot comment.
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Wedge victim",
            "priority": "2",
        })
        wedge = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                     if t["title"] == "Wedge victim"][0])
        check(wait_for(lambda: task_state(auth, wedge).get("status") == "working"
                       or False, timeout=60, what="wedge victim working") is not None,
              "wedge victim reaches working")
        fake.hold_state = True
        fake.freeze_activity = True
        check(wait_for(lambda: any("saddle wedged" in c["body"]
                                   for c in task_state(auth, wedge).get("comments", []))
                       or False, timeout=40, what="wedge comment") is not None,
              "stale activity wedges the saddle with a bot comment")
        check(saddle_card("fake-saddle")["status"] == "recovering",
              "wedged saddle is fenced")
        card = saddle_card("fake-saddle")
        check(card.get("health") == "fenced",
              "health badge reports fenced")
        fake.hold_state = False
        fake.freeze_activity = False
        # The fake's scripted lifecycle parks on 'done' when released
        # mid-working; a real daemon lands idle after its round.
        fake.state = "idle"
        check(wait_for(lambda: saddle_card("fake-saddle")["health"] in ("healthy", "stale")
                       or False, timeout=40, what="health recovers") is not None,
              "health badge recovers with the fence")
        check(wait_task_done(auth, wedge, timeout=90) is not None,
              "wedged task requeues and completes after recovery")

        print("== R8: raw-log ingestion is lossless ==")
        # Dynamic scripted ring: empty at seed (the pre-push probe
        # baselines at next=1), then the daemon "produces" the corpus on
        # /start — seq 1 coloured prompt run, 2 plain, 3 a bare CRLF
        # between prompts, 4..6 following. Every non-empty served chunk
        # must land, exactly once.
        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.scripted_log = []
        fake.scripted_next = 1
        fake.state = "working"
        fake.hold_state = True
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Lossless log task",
            "priority": "2",
        })
        lossless = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                        if t["title"] == "Lossless log task"][0])

        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 with a pre-push probe on an empty ring")
        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 served chunk lands in session_raw_log, CRLF-only included (%s)"
              % stored_seqs(lossless))
        time.sleep(7)
        check(stored_seqs(lossless) == [1, 2, 3, 4, 5, 6],
              "no duplicates or losses on later polls")
        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.state = "idle"
        fake.scripted_log = None
        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.
        fake.ssh_public_key = "ssh-ed25519 AAAAC3NzaC1lZDI1teStKey saddle-1@fedora"
        fake.state = "idle"
        fake_card = saddle_card("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("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 /start payload.
        fake_card = saddle_card("fake-saddle")
        act(auth, fake_card["update_identity"],
            {"git_name": "clanker-bot", "git_email": "bot@unitatem.net"})
        fake_card = saddle_card("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")

        pushes_before = len(fake.start_payloads)
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Identity task",
            "priority": "2",
        })
        identity_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                             if t["title"] == "Identity task"][0])
        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 pushed") 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",
              "/start carries the saddle's git_identity")

        # Earlier pushes (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 pushes.
        fake_card = saddle_card("fake-saddle")
        act(auth, fake_card["update_identity"], {"git_name": "", "git_email": ""})
        check(saddle_card("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("== R10: branch_advanced + gap forensics ==")
        # --- branch_advanced on the /poll verdict channel ---
        fake.state = "working"
        fake.hold_state = True
        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))
        poll_session = conn.execute("select max(id) from sessions").fetchone()[0]
        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 task: cancel it so the held-working poll loop
        # does not carry it into the race subsection.
        act(auth, task_state(auth, poll_parent)["cancel"], {})
        fake.hold_state = False
        fake.state = "idle"

        # --- the /log snapshot race loses nothing ---
        fake.state = "working"
        fake.hold_state = True
        fake.scripted_corpus = [
            {"seq": 1, "text": "prompt-one\r\n"},
            {"seq": 2, "text": "echo a\r\n"},
            {"seq": 3, "text": "a\r\n"},
            {"seq": 4, "text": "prompt-two\r\n"},
        ]
        fake.scripted_log = []
        fake.scripted_next = 1
        fake.race_texts = ["raced-five\r\n", "raced-six\r\n"]
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Race forensics task",
            "priority": "2",
        })
        race_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                         if t["title"] == "Race forensics task"][0])

        def race_seqs():
            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", (race_task,))]
            finally:
                conn.close()

        check(wait_for(lambda: race_seqs() == [1, 2, 3, 4, 5, 6] or False,
                       timeout=40, what="race chunks all stored") is not None,
              "the snapshot race loses nothing: complete 1..6, no gaps (%s)"
              % race_seqs())
        time.sleep(7)
        check(race_seqs() == [1, 2, 3, 4, 5, 6],
              "race chunks stored exactly once")
        fake.hold_state = False
        fake.state = "idle"
        fake.scripted_log = None
        fake.scripted_corpus = None
        check(wait_task_done(auth, race_task, timeout=60) is not None,
              "race forensics task completes")


        print("== R11: live blocks arrive without a reload ==")
        # The browser flow on each pulse: statum.act(task.refresh) — the
        # sealed GET action must rehydrate the task slot with the new
        # block. (The channel chain itself — entrypoint subscribe on
        # load, SSE set events on bump — was verified end-to-end against
        # a real channel client during the investigation.)
        fake.state = "working"
        fake.hold_state = True
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Live block task",
            "priority": "2",
        })
        live_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                         if t["title"] == "Live block task"][0])
        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")
        conn = db()
        live_session = conn.execute(
            "select id from sessions where task_id = ?", (live_task,)).fetchone()[0]
        conn.close()
        status, _ = saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
                               {"type": "block", "session_id": live_session, "data": {
                                   "question": "Pick one", "type": "question",
                                   "options": ["A", "B"]}})
        check(status == 200, "block raised via the daemon")
        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 raised block reaches the task state")
        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, detail["answer_action"] if "answer_action" in detail
            else detail["blocks"][0]["answer_action"], {"answer": "A"})
        fake.hold_state = False
        fake.state = "idle"
        check(wait_task_done(auth, live_task, timeout=60) is not None,
              "live-block task completes after answering")

        # The daemon vanishing mid-task: three refused polls → offline,
        # task requeued with a bot comment, UI reflects.
        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
            "project_id": "1",
            "title": "Vanish victim",
            "priority": "2",
        })
        vanish = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                      if t["title"] == "Vanish victim"][0])
        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("fake-saddle")["status"] == "offline" or False,
                       timeout=60, what="daemon death → offline") is not None,
              "vanishing daemon drives the saddle offline")
        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("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.
        conn = db()
        conn.execute("insert into api_tokens (saddle_id, token, created_at)"
                     " values (?, 'saddle-legacy-copied-token', strftime('%s','now'))",
                     (int(racer_cards[fake2]["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")

        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, fake3):
            daemon.ask_now()
        check(True, "fleet re-asks after the restart (sleeping re-established)")


    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)" % app.poll())
        if app.poll() is None and sys.exc_info()[0] is not None:
            dump_app_stacks(app.pid)
        app.terminate()
        try:
            app.wait(timeout=5)
        except subprocess.TimeoutExpired:
            app.kill()

    print("%d passed, %d failed" % (passes, failures))
    return 1 if failures else 0


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