clanker-saddle/test/harness.py

#!/usr/bin/env python3
"""The web-stack test harness (plan 16 R5, arch 18): everything the
scenario bundles share, plus the runner (sequential by default,
`--parallel [N]` for capped parallel shards).

Layout: test/harness.py (this file — the FakeSaddle daemon, the
http/act/login/slot_public wire helpers, wait_for/quiet_window/db/check,
the per-shard fixtures via start_bundle/Ctx, and run_bundle +
run_all_parallel); test/bundles/<name>.py (one module per bundle, its
run(ctx) driving that shard's scenarios); test/web_stack_test.py (the
thin entry shim meson invokes; it also stays directly runnable for the
per-bundle shard processes the parallel runner spawns).

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

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

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

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

Plan 10 phase 1 (work acquisition flip):

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

Plan 10 phase 2 (telemetry flip):

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

Plan 10 phase 3 (capability awareness + UI):

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

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

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

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

Heredoc completion reports (pages shard):

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

Speed pass (plan 11, task 0):

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

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

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

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


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


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

passes = 0
failures = 0

# The current shard's fixture context (goal_state reads its auth — the
# restart scenarios re-login and re-point ctx.auth mid-run)
CTX = None


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


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


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


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


def http(method, url, data=None, headers=None, timeout=30):
    body = None
    if data is not None:
        pairs = []
        for k, v in data.items():
            for item in (v if isinstance(v, (list, tuple)) else [v]):
                pairs.append((k, item))
        body = "&".join("%s=%s" % (k, urllib.parse.quote(str(item))) for k, item in pairs).encode()
    request = urllib.request.Request(WEB + url if url.startswith("/") else url,
                                     data=body, method=method)
    if data is not None:
        request.add_header("Content-Type", "application/x-www-form-urlencoded")
    for key, value in (headers or {}).items():
        request.add_header(key, value)
    # Retry dropped/refused connections: the freshly started app can
    # close a very early keep-alive connection once while warming up,
    # and its first seconds can wobble under load.
    for attempt in range(6):
        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 >= 5:
                raise
            time.sleep(0.75)


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


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


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


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


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


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


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


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


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


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


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


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


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


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


def http_get_page(auth, uri):
    """The compiled page HTML for a statum route (auth via the slot
    header), for asserting page-level embeds the slot state cannot
    see."""
    status, body = http("GET", uri, headers={
        "X-Statum-Slot": "key=%s; as_at=%s" % (auth["key"], auth["as_at"])})
    if status != 200:
        raise SystemExit("GET %s -> HTTP %d" % (uri, status))
    return body


def goal_state(goal_id):
    """The goal slot for the CURRENT shard's admin login (the x7
    per-bundle duplicates collapsed here, plan 16 R5). Reads the
    runner's fixture context — the restart scenarios re-login and
    re-point ctx.auth mid-run, and late binding follows."""
    return slot_public(CTX.auth, "/goals/%d" % goal_id, "goal")


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


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


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


def event_rows(task_id, kinds=None):
    """The task's ledger rows (kind, payload JSON), oldest first."""
    conn = db()
    try:
        if kinds:
            marks = ",".join("?" for _ in kinds)
            return conn.execute(
                "select kind, payload from events"
                " where scope_type = 'task' and scope_id = ? and kind in (%s)"
                " order by id" % marks, (task_id,) + tuple(kinds)).fetchall()
        return conn.execute(
            "select kind, payload from events"
            " where scope_type = 'task' and scope_id = ? order by id",
            (task_id,)).fetchall()
    finally:
        conn.close()


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

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

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

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

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

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

    def queue_output(self, entry_type, command, output):
        """Pushes one command_output event home (the daemon's entry log
        now rides the events channel; the old results-entries path is
        gone)."""
        self.push_events([{"kind": "command_output",
                           "payload": {"command": command, "text": output}}])

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

    def push_events(self, events):
        """Pushes a batch of ledger events home (plan 17 §7.2): each
        element a {seq?, kind, actor?, ts?, payload?} dict; a missing
        seq is assigned from the per-session monotonic counter.
        Pushes are suppressed while silent/hold_state (the staleness
        tests' silent runner)."""
        if self.dead or self.silent or self.token is None or self.session_id is None:
            return
        batch = []
        with self.lock:
            for event in events:
                item = dict(event)
                item["seq"] = int(item.pop("seq", 0) or self.next_push_seq())
                item.setdefault("kind", "progress")
                item.setdefault("actor", "model")
                item.setdefault("payload", {})
                batch.append(item)
        try:
            saddle_api("POST", "/api/saddle/chunks", self.token,
                       {"task_id": self.running_task,
                        "session_id": self.session_id, "events": batch})
        except (OSError, ValueError):
            pass  # app restarting / shutting down mid-test

    def push_chunks(self, chunks):
        """Legacy-shaped helper: raw texts (or {seq, text} dicts, a
        redelivery) push as progress events carrying the text."""
        batch = []
        for chunk in chunks:
            if isinstance(chunk, dict):
                batch.append({"seq": chunk.get("seq"),
                              "kind": chunk.get("kind", "progress"),
                              "payload": chunk.get("payload",
                                                   {"text": chunk.get("text", "")})})
            else:
                batch.append({"kind": "progress", "payload": {"text": chunk}})
        self.push_events(batch)

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

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

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

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

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

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

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

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

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

    def accept_assignment(self, payload):
        """A 200 next-task assignment: adopted exactly like POST /start."""
        self.start_payloads.append(payload)
        task = payload.get("task") or {}
        task_id = task.get("id")
        self.running_task = int(task_id or 0)
        self.running_title = task.get("title") or ""
        self.session_id = payload.get("session_id")
        # The assignment's answers resolve any block this daemon was
        # parked on (resume-carries-answer).
        self.pending_block = None
        count = self.start_counts.get(task_id, 0) + 1
        self.start_counts[task_id] = count
        if count > 1:
            # Crash-resume contract: re-pushing a task whose
            # workspace still exists restores it; the first
            # /status after the restore reports "resumed": true.
            self.resumed_pending = True
        # Plan 12: a goal-run assignment plays its scripted behaviour
        # instead of the plain done push (plan submission, parking,
        # question, completion, failure — consumed one per round).
        self.goal_behaviour = None
        if payload.get("kind") == "goal_run":
            self.goal_assignments.append(payload)
            self.goal_behaviour = self.goal_script.pop(0) if self.goal_script else ("suspend",)
        self.set_state("failing" if self.fail_next else "working")
        self.fail_next = False
        # The round's opening telemetry: the §7.2 event batch (turn 1:
        # lifecycle, turn text, tool call + result, usage) plus the
        # working state push. A new session restarts the per-session
        # seq counter; a scripted corpus supplements the batch.
        self._push_seq = 0
        opening = [
            {"kind": "lifecycle",
             "actor": "harness",
             "payload": {"action": "start", "text": "round started"}},
            {"kind": "turn_text",
             "payload": {"turn": 1, "text": "Starting the assigned task."}},
            {"kind": "tool_call",
             "payload": {"turn": 1, "tool": "run_command",
                         "args": {"command": "echo hello"}}},
            {"kind": "tool_result",
             "actor": "harness",
             "payload": {"turn": 1, "stdout": "hello\n", "exit_code": 0}},
            {"kind": "usage",
             "actor": "harness",
             "payload": {"turn": 1, "input_tokens": 120, "output_tokens": 30}},
        ]
        if self.scripted_corpus is not None:
            opening = [dict(chunk) for chunk in self.scripted_corpus]
        self.push_events(opening)
        self.push_results("working")

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

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

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

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

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

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

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

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

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


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

    env = dict(os.environ)
    env.update({
        "SPRY_DB_PATH": DB_PATH,
        "SPRY_SEED_ADMIN_USERNAME": "admin",
        "SPRY_SEED_ADMIN_PASSWORD": "adminpass123",
        "SADDLE_LLM_BASE_URL": "http://127.0.0.1:%d" % DAEMON_PORT,
        "SADDLE_PROJECTS_ROOT": os.path.join(WORK, "workspace"),
        "SADDLE_HTTP_TIMEOUT_S": "2",
        "SADDLE_LOST_WAKE_S": "2",
        "SADDLE_STALE_S": "3",
        "SADDLE_DISPATCH_TICK_S": "1",
        # Plan 12: the goal-agent resume coalescing window shrinks to
        # one second so the event-driven resume tests settle quickly.
        "SADDLE_GOAL_RESUME_S": os.environ.get("SADDLE_GOAL_RESUME_S", "1"),
        # Never inherit a caller-poisoned LD_LIBRARY_PATH (e.g. one that
        # resolves the ancient /lib64 copies): empty makes the loader use
        # the binary's RUNPATH, which is the dev stack it was built
        # against.
        "LD_LIBRARY_PATH": "",
    })
    log_previous = os.path.join(WORK, "app.log.1")
    if os.path.exists(os.path.join(WORK, "app.log")):
        shutil.move(os.path.join(WORK, "app.log"), log_previous)
    log = open(os.path.join(WORK, "app.log"), "w")
    command = [APP, str(APP_PORT)]
    if os.environ.get("WEBSTACK_GDB"):
        # Debugging aid: run the app under gdb; a crash backtrace lands in
        # app.log. WEBSTACK_GDB=1 meson test ... to enable.
        command = ["gdb", "-batch", "-quiet",
                   "-ex", "handle SIGPIPE nostop noprint pass",
                   "-ex", "run", "-ex", "bt", "--args", APP, str(APP_PORT)]
    process = subprocess.Popen(command, cwd=WORK, env=env,
                               stdout=log, stderr=subprocess.STDOUT)
    # The ready probe must require OUR process alive: a free_port()
    # TOCTOU can hand this port to another shard's app — ours then dies
    # at bind (MHD error()) while the probe happily answers from the
    # intruder (same admin seed), and every check would silently drive
    # the WRONG app. poll() != None with a serving port is exactly that
    # shape: take the exit branch, the start_bundle retry re-ports.
    outcome = wait_for(lambda: (app_ready() and process.poll() is None and "ready")
                       or (process.poll() is not None and "app-exited-%d" % process.poll())
                       or False, timeout=30, what="app ready")
    if outcome is None:
        raise SystemExit("app did not come up; see %s/app.log" % WORK)
    if str(outcome).startswith("app-exited-"):
        raise SystemExit("app exited during startup (%s); see %s/app.log"
                         % outcome, WORK)
    return process


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


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


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

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


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


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

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

    def serve(self):
        service = self

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

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

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

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

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

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


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


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


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


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

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


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

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

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

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




BUNDLES_ORDER = ("pages", "dispatch", "subtasks", "lineage",
                 "allocation", "lifecycle", "answers", "push", "goals",
                 "replay", "goalui", "forcewake", "users", "stacks",
                 "resurrect")


def run_bundle(name):
    """Runs one shard to completion; returns a failure count."""
    global CTX
    started = time.monotonic()
    ctx = start_bundle(name)
    CTX = ctx
    import importlib
    bundle = importlib.import_module("bundles." + name)
    # Runtime globals the star-import captured at module load can be
    # stale (the sequential debug path reuses modules across shards) —
    # re-point them at this shard's environment. The shared helpers
    # read them from harness globals already.
    for runtime_name in ("WORK", "DB_PATH", "APP_PORT", "DAEMON_PORT", "WEB"):
        setattr(bundle, runtime_name, globals()[runtime_name])
    runner = bundle.run
    try:
        runner(ctx)
    finally:
        # Exit-code diagnostic: distinguishes a crashed app (negative
        # returncode, e.g. -11 SIGSEGV) from a clean shutdown. When
        # tearing down mid-exception, also capture the still-alive
        # app's stacks before killing it (a wedged-but-alive server
        # otherwise leaves no evidence).
        print("(app returncode at teardown: %s)" % ctx.app.poll())
        if ctx.app.poll() is None and sys.exc_info()[0] is not None:
            dump_app_stacks(ctx.app.pid)
        ctx.fake.kill()
        try:
            ctx.app.terminate()
            try:
                ctx.app.wait(timeout=5)
            except subprocess.TimeoutExpired:
                ctx.app.kill()
            check(ctx.app.poll() in (0, -15),
                  "bundle %s app shut down cleanly (exit %s)"
                  % (name, ctx.app.poll()))
        except OSError:
            pass
    print("%d passed, %d failed" % (passes, failures))
    print("[bundle %s] %.1fs" % (name, time.monotonic() - started))
    return failures


def run_all_parallel(limit):
    """The explicit opt-in: one subprocess per bundle (each with its own
    app, daemons and database), at most `limit` in flight at a time
    (every shard is an app plus fake daemons — an unbounded fan-out
    OOMs the machine), verdicts aggregated into one exit code and one
    combined check-count summary."""
    base = os.environ.get("WEBSTACK_WORK", "/tmp/kilo/webstack-e2e")
    log_dir = os.path.join(os.path.dirname(base) or "/tmp/kilo",
                           "webstack-shards")
    os.makedirs(log_dir, exist_ok=True)
    started = time.monotonic()
    limit = max(1, int(limit))
    pending = list(BUNDLES_ORDER)
    running = []
    finished = []
    while pending or running:
        while pending and len(running) < limit:
            name = pending.pop(0)
            out_path = os.path.join(log_dir, "bundle-%s.log" % name)
            handle = open(out_path, "w")
            proc = subprocess.Popen(
                [sys.executable, "-u",
                 os.path.join(os.path.dirname(os.path.abspath(__file__)),
                              "web_stack_test.py"), APP, "--bundle", name],
                stdout=handle, stderr=subprocess.STDOUT)
            running.append((name, proc, handle, out_path, time.monotonic()))
        time.sleep(0.2)
        still = []
        for name, proc, handle, out_path, spawned in running:
            if proc.poll() is None:
                if time.monotonic() - spawned > SHARD_TIMEOUT_S:
                    proc.kill()
                    proc.wait()
                else:
                    still.append((name, proc, handle, out_path, spawned))
                continue
            handle.close()
            finished.append((name, proc, out_path))
        running = still
    total_pass = total_fail = 0
    broken = []
    by_name = {name: (proc, out_path) for name, proc, out_path in finished}
    for name in BUNDLES_ORDER:
        proc, out_path = by_name[name]
        text = open(out_path).read()
        match = re.search(r"(\d+) passed, (\d+) failed", text)
        took = re.search(r"\[bundle %s\] ([0-9.]+)s" % name, text)
        bundle_pass = int(match.group(1)) if match else 0
        bundle_fail = int(match.group(2)) if match else -1
        total_pass += max(bundle_pass, 0)
        total_fail += max(bundle_fail, 0)
        status = "ok"
        if proc.returncode != 0 or bundle_fail != 0 or match is None:
            status = "BROKEN (exit %s)" % proc.returncode
            broken.append(name)
        print("[shard %-10s] %6ss  %3d passed %3d failed  %s"
              % (name, took.group(1) if took else "?",
                 bundle_pass, bundle_fail, status))
        if status != "ok":
            tail = "\n".join(text.rstrip().split("\n")[-60:])
            print("--- web-stack/%s tail (full log: %s) ---" % (name, out_path))
            print(tail)
            print("--- end web-stack/%s tail ---" % name)
    print("shards wall: %.1fs  (logs under %s)" % (time.monotonic() - started, log_dir))
    print("ALL SHARDS: %d passed, %d failed%s"
          % (total_pass, total_fail,
             "" if not broken else "  BROKEN: " + ", ".join(broken)))
    return 1 if broken or total_fail else 0


def main():
    if not os.path.exists(APP):
        raise SystemExit("app binary not found at %s (build first)" % APP)
    argv = sys.argv[2:]
    if "--bundle" in argv:
        sys.exit(1 if run_bundle(argv[argv.index("--bundle") + 1]) else 0)
    if "--parallel" in argv:
        at = argv.index("--parallel")
        limit = 4
        if len(argv) > at + 1 and argv[at + 1].isdigit():
            limit = int(argv[at + 1])
        return run_all_parallel(limit)
    for name in BUNDLES_ORDER:
        if run_bundle(name):
            return 1
    return 0


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