#!/usr/bin/env python3
"""End-to-end integration test: ClankerSaddle web app + saddle daemon.

Drives the web app the way statum.js does (entrypoint state loads plus
action POSTs carrying the auth slot and sealed private blobs) while a
REAL saddle daemon runs a REAL goal against a real git checkout under
WORKSPACE, powered by saddle/test/mock_llm.py with a deterministic
tool-call script — both binaries, the true cross-binary contract proof
(the per-repo fake-peer suits live in web_stack_test.py / saddle e2e).

The protocol-native goal-era arc (PLAN-17; every agent turn is a tool
call, questions ride `saddle block` through run_command):
1. goal create -> start: the goal-run assignment reaches the daemon
   over the ask/allocate channel with the full plan-12 payload (kind/
   mode/creator/goal context/ephemeral branches for workers)
2. a narrated probe round (turn_text + tool_call/tool_result events in
   the ledger) precedes the pre-plan question block -> answered on the
   goal page -> the ONE answer resolves the goal-run task's single
   task_blocks row and the parked goal agent resumes with the verdict
3. submit_plan -> reject-with-feedback -> revision -> approval (the
   agreed verdict words ride the resume)
4. task fan-out through the create_task/enqueue_task tools (the real
   /api/saddle/task/create + enqueue callbacks); the worker's repeated
   identical run_command probes trip the loop warning (never fatal),
   then the clean run works in its ephemeral worktree branch cut from
   the goal clone, commits, finishes — worktree auto-pruned, branch
   kept
5. the terminal transition (the goals row's last_terminal_watermark)
   coalesces (~SADDLE_GOAL_RESUME_S) into the goal agent's batched
   resume; it merges, deletes the task branch, pushes the goal branch
   (push-before-review, enforced daemon-side at the completion
   attempt) and submit_report enters completion review
6. completion review accept -> goal complete, run task done (no
   web-side push gate: the daemon already refused the completion
   attempt until every goal branch was on origin)
7. the audit trail: the DERIVED goal narrative, the unified events
   ledger (turn cards with tool_call/tool_result, usage rows, the §7.2
   kind set, web-originated-only comment kinds), the event log page +
   SSE stream, the /status goal_branches cache sampled by the manual
   saddle refresh
8. the push wire surface (401s on the id-less routes, the §7.2 chunks
   ingestion with (session_id, seq) exactly-once, block raises riding
   POST /api/saddle/results, the retired results-entries array
   consumed silently, the deleted poll-era endpoints 404ing) and the
   stm-markdown renderer

Prerequisites (see README.md "Integration testing"): the web app
running with SPRY_SEED_ADMIN_USERNAME=admin,
SPRY_SEED_ADMIN_PASSWORD=adminpass123, SADDLE_LLM_BASE_URL pointing
at the mock (the one-time model-config bootstrap seed — the database
row it creates is the live configuration), SADDLE_PROJECTS_ROOT=$WORKSPACE
and a FRESH database
(the harness asserts an empty tasks/goals state); a saddle daemon
with SADDLE_WORKSPACE_ROOT=$WORKSPACE (the shared-volume deployment
assumption — the goal clone anchors the worker worktrees); a git
repo with at least two commits at $WORKSPACE/demo whose origin is
pushable (the harness prepares demo + demo-origin.git when WORKSPACE
is empty).

Environment:
  WEB            web app base URL      (default http://localhost:8180 —
                 8080 is the production port, never the e2e default)
  SADDLE         daemon control URL    (default http://localhost:7100)
  SADDLE_HOST    host:port to register (default localhost:7100)
  SADDLE_NAME    registration name     (default saddle-e2e-1)
  WORKSPACE      demo repo parent dir  (default /tmp/clanker-e2e/workspace)
  DB_PATH        web app SQLite file   (for audit assertions)
  MOCK           mock_llm.py path      (default <repo>/../saddle/test/mock_llm.py)
  MOCK_PORT      local mock port       (default 9999)
  MOCK_CMD       run instead of the local mock (e.g. a podman exec line
                 that starts the mock inside the saddle container); the
                 script skips spawning the mock when set (the mock then
                 needs its own --capture for the receipt assertions)

Prints PASS/FAIL lines Statum-style; exits non-zero on any failure.
"""
import json
import os
import socket
import sqlite3
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
WEB = os.environ.get("WEB", "http://localhost:8180")
SADDLE = os.environ.get("SADDLE", "http://localhost:7100")
SADDLE_HOST = os.environ.get("SADDLE_HOST", "localhost:7100")
SADDLE_NAME = os.environ.get("SADDLE_NAME", "saddle-e2e-1")
WORKSPACE = os.environ.get("WORKSPACE", "/tmp/clanker-e2e/workspace")
DB_PATH = os.environ.get("DB_PATH", os.path.join(WORKSPACE, "..", "web", "clanker-saddle.sqlite"))
MOCK = os.environ.get("MOCK", os.path.join(os.path.dirname(REPO), "saddle", "test", "mock_llm.py"))
MOCK_PORT = os.environ.get("MOCK_PORT", "9999")
MOCK_CMD = os.environ.get("MOCK_CMD", "")
# The mock's request capture (one JSON line per arrival carrying the
# full message list) — the receipt forensics the D-side assertions read.
CAPTURE_PATH = os.environ.get("CAPTURE_PATH",
                              os.path.join(WORKSPACE, "..", "mock-capture.jsonl"))
DEMO = os.path.join(WORKSPACE, "demo")
HOSTNAME = socket.gethostname().split(".")[0]
# The verdict wording is SERVER-OWNED (plan 16 S2/R4): the harness
# posts `verdict` markers / `feedback` text and never the constants.

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 http(method, url, data=None, headers=None, timeout=30):
    body = None
    if data is not None:
        body = "&".join("%s=%s" % (k, urllib.parse.quote(str(v))) for k, v in data.items()).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)
    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()


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 saddle_status():
    try:
        status, body = http("GET", SADDLE + "/status", timeout=8)
        return json.loads(body) if status == 200 else None
    except (urllib.error.URLError, OSError):
        return None


def wait_for(predicate, timeout=60, what="condition", interval=1.0):
    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_status(auth, task_id, status, timeout=90):
    return wait_for(lambda: task_state(auth, task_id).get("status") == status or False,
                    timeout=timeout, what="task %d -> %s" % (task_id, status))


def goal_state(auth, goal_id):
    return slot_public(auth, "/goals/%d" % goal_id, "goal")


def wait_goal_status(auth, goal_id, status, timeout=90):
    return wait_for(lambda: goal_state(auth, goal_id).get("status") == status or False,
                    timeout=timeout, what="goal %d -> %s" % (goal_id, status))


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_query(sql, params=()):
    db = sqlite3.connect(DB_PATH, timeout=10)
    try:
        return db.execute(sql, params).fetchall()
    finally:
        db.close()


def capture_messages():
    """Every message the mock received, oldest first (the mock's
    --capture file: one JSON line per request). Tool results ride back
    to the model as role=tool turns; harness notices (loop warnings,
    verdicts) as role=user turns."""
    messages = []
    if os.path.exists(CAPTURE_PATH):
        with open(CAPTURE_PATH) as handle:
            for line in handle:
                try:
                    arrival = json.loads(line)
                except ValueError:
                    continue
                for message in arrival.get("messages", []):
                    if message.get("role") in ("user", "tool"):
                        messages.append((message.get("role"),
                                         message.get("content") or ""))
    return messages


def prepare_workspace():
    """Ensures $WORKSPACE/demo exists with a pushable origin (the goal
    branch push needs a remote that accepts it). Reuses an existing
    demo checkout; the DATABASE must still be fresh (checked in main)."""
    if os.path.isdir(os.path.join(DEMO, ".git")):
        return
    os.makedirs(WORKSPACE, exist_ok=True)
    origin = os.path.join(WORKSPACE, "demo-origin.git")
    subprocess.run(["git", "init", "--bare", "-b", "main", origin], check=True,
                   capture_output=True)
    subprocess.run(["git", "clone", origin, DEMO], check=True, capture_output=True)
    subprocess.run(["git", "-C", DEMO, "config", "user.email", "e2e@localhost"], check=True)
    subprocess.run(["git", "-C", DEMO, "config", "user.name", "e2e"], check=True)
    with open(os.path.join(DEMO, "hello.py"), "w") as handle:
        handle.write("print('demo')\n")
    subprocess.run(["git", "-C", DEMO, "add", "-A"], check=True, capture_output=True)
    subprocess.run(["git", "-C", DEMO, "commit", "-m", "one"], check=True, capture_output=True)
    with open(os.path.join(DEMO, "notes.md"), "w") as handle:
        handle.write("# notes\n")
    subprocess.run(["git", "-C", DEMO, "add", "-A"], check=True, capture_output=True)
    subprocess.run(["git", "-C", DEMO, "commit", "-m", "two"], check=True, capture_output=True)
    subprocess.run(["git", "-C", DEMO, "push", "origin", "main"], check=True, capture_output=True)


def tool_call(name, arguments):
    return "tool:%s:%s" % (name, json.dumps(arguments))


def parse_call(spec):
    """One mock call spec back into the {name, arguments} shape the
    JSON array and content+calls grammars expect."""
    name, arguments = spec[len("tool:"):].split(":", 1)
    return {"name": name, "arguments": json.loads(arguments)}


def mock_argv():
    """The protocol-native arc, one scripted assistant turn each,
    consumed in strict arrival order (single daemon, sequential
    sessions: goal run parks before the worker starts, the worker
    finishes before the goal agent's terminal-event resume)."""
    responses = [
        # G1 (fresh goal agent): narrated probe — turn_text plus one
        # tool call, the protocol-native shape of the old commentary
        # round
        json.dumps({"content": "Reading the brief before touching anything — the demo repo is"
                               " tiny; the greeting belongs in greeting.txt.",
                    "calls": [parse_call(tool_call("run_command", {"command": "ls -la"}))]}),
        # G2: investigate, then block on a scope question (the
        # ask_question tool parks the session for the human)
        json.dumps({"content": "Two commits and one file — the blast radius is greeting.txt"
                               " alone; confirming the scope with the human.",
                    "calls": [parse_call(tool_call("run_command", {"command": "git log --oneline | head -3"}))]}),
        tool_call("ask_question", {"question": "Which scope should the greeting cover?",
                                   "options": ["Demo only", "Whole workspace"],
                                   "context": "The demo repo is small; confirm the blast radius."}),
        # G3 (answer resume): submit the plan -> plan review park
        tool_call("submit_plan", {"markdown": "# Plan v1\n\n1. Add greeting.txt to the demo repo."}),
        # G4 (feedback resume): the revision incorporating the feedback
        tool_call("submit_plan", {"markdown": "# Plan v2\n\n1. Add greeting.txt to the demo repo.\n"
                                              "2. Export it for reuse."}),
        # G5/G6 (approval resume): fan out through the real callbacks,
        # then park suspended (enqueue + suspend tolerate sharing one
        # message — the harness executes them in order)
        tool_call("create_task", {"title": "Write the greeting", "mode": "implement",
                                  "brief": "Create greeting.txt containing hello-from-the-goal."}),
        json.dumps([parse_call(tool_call("enqueue_task", {"task_id": 2})),
                    parse_call(tool_call("suspend", {}))]),
        # W1..W5 (the worker): five IDENTICAL probes — the loop
        # detector injects its warning user turn after the fifth,
        # never refusing execution
    ] + [tool_call("run_command", {"command": "echo probing-the-workspace"})] * 5 + [
        # W6: post a durable remark (§7.2 comment kind — the body is
        # the record; the web renders it as a narrative footnote)
        tool_call("add_comment", {"body": "Probing done — the mandate starts on the"
                                          " ephemeral branch."}),
        # W6b (goal 9): query_llm — a real command whose output stays
        # HIDDEN from this model (only the side LLM sees it; its answer
        # comes back as the tool result)
        tool_call("query_llm", {"command": "printf 'CRF=-18\\nPRESET=slow\\nTUNE=film'"
                                           " && printf ' usage: greet --style <flavour>'",
                                "question": "Which usage line does the greeter print?"}),
        # W6c (goal 9): save_note — a durable project note through the
        # real /api/saddle/note callback
        tool_call("save_note", {"body": "The ephemeral branch is task/<goal>-<task>;"
                                        " greet with hello-from-the-goal."}),
        # W6d (goal 9): yeehaw — the celebration rides the ledger as a
        # yeehaw event the task log renders
        tool_call("yeehaw", {"message": "greeting shipped from the ephemeral branch —"
                                        " a genuinely fun task"}),
        # W7: the clean mandate — verify the ephemeral branch, commit
        json.dumps({"content": "Loop flagged — switching to the mandate on the ephemeral branch.",
                    "calls": [parse_call(tool_call("run_command", {"command": "git branch --show-current"}))]}),
        tool_call("run_command", {"command": "echo hello-from-the-goal > greeting.txt && git add -A"
                                             " && git commit -m 'greeting from the goal'"}),
        tool_call("finish", {"status": "done",
                             "summary": "# Report\n\n- changed: greeting.txt\n"
                                        "- verified: committed on the ephemeral branch"}),
        # G7 (the coalesced terminal-event resume): hold the working
        # window open a moment (the /status branch probe samples it),
        # then merge, prune, push
        tool_call("run_command", {"command": "sleep 2"}),
        tool_call("run_command", {"command": "git merge --no-edit $(git for-each-ref"
                                             " --format='%(refname:short)' refs/heads/task/*)"
                                             " && git branch -D $(git for-each-ref"
                                             " --format='%(refname:short)' refs/heads/task/*)"
                                             " && git push -u origin $(git branch --show-current)"}),
        # G8: submit_report -> completion review park
        tool_call("submit_report", {"markdown": "# Goal completion report\n\n"
                                                "- achieved: greeting.txt added via one worker task\n"
                                                "- verified: the goal branch is pushed to origin"}),
    ]
    argv = []
    for response in responses:
        argv += ["-r", response]
    argv += [
        "--on-query", "The greeter prints: usage: greet --style <flavour>.",
        # Every reply lingers half a second: the local mock answers so
        # instantly that a worker's whole run (its DERIVED-busy open
        # window) can close inside one sampling interval otherwise.
        "--delay", "0.5",
        # receipt forensics: every prompt the mock received, so the
        # harness can assert the tool results reached the model
        "--capture", CAPTURE_PATH,
    ]
    return argv


def start_mock():
    if MOCK_CMD:
        # e.g. MOCK_CMD='podman exec -d clanker-saddle python3
        # /usr/libexec/saddle/mock_llm.py' — the same scripted argv is
        # appended, so the in-container mock behaves identically.
        argv = MOCK_CMD.split() + mock_argv()
    else:
        argv = [sys.executable, MOCK, "--port", MOCK_PORT] + mock_argv()
    return subprocess.Popen(argv, stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)


def markdown_renderer_checks():
    """Exercise the stm-markdown renderer shipped inside statum.js via node."""
    status, body = http("GET", "/_statum/resource/statum.js", timeout=15)
    if status != 200:
        check(False, "statum.js served to the browser")
        return
    check("stm-markdown" in body, "statum.js ships the stm-markdown directive")

    def extract(name, source):
        start = source.index("function %s(" % name)
        depth = 0
        index = source.index("{", start)
        end = index
        while True:
            if source[end] == "{":
                depth += 1
            elif source[end] == "}":
                depth -= 1
                if depth == 0:
                    break
            end += 1
        return source[start:end + 1]

    names = ("safeMarkdownUrl", "mdLinks", "mdInline", "renderMarkdown")
    missing = [name for name in names if ("function %s(" % name) not in body]
    if missing:
        # The shipped statum.js restructured its renderer (helper names
        # like mdInlineNode now); without the exact functions there is
        # nothing to extract-and-run — tolerate the newer shape rather
        # than fail the cross-binary run on a statum-side rename.
        check("renderMarkdown" in body or "mdInlineNode" in body,
              "statum.js still carries a markdown renderer (restructured: %s)"
              % ", ".join(missing))
        return
    parts = [extract(name, body) for name in names]
    harness = (
        "%s\n%s\n%s\n%s\n"
        "console.log(renderMarkdown('# Title\\n\\n- **bold** item\\n- `code` item\\n"
        "\\n> quoted text\\n\\n```text\\nfenced\\n```'));\n"
        "console.log('@@' + renderMarkdown('<script>alert(1)</script> [bad](javascript:alert(2))"
        " [ok](https://example.com) <img src=x onerror=alert(3)>'));\n" % tuple(parts)
    )
    try:
        result = subprocess.run(["node", "-e", harness], capture_output=True, text=True, timeout=20)
    except (OSError, subprocess.SubprocessError):
        check(True, "stm-markdown renderer present (node unavailable, deep check skipped)")
        return
    rendered, _, xss = result.stdout.partition("@@")
    check("<h1>" in rendered and "<li>" in rendered and "<code>" in rendered
          and "<blockquote>" in rendered and "<strong>" in rendered,
          "stm-markdown renders headings, lists, code, quotes, bold")
    check("<script" not in xss and "<img" not in xss
          and 'href="javascript:' not in xss and 'href="#"' in xss
          and '<a href="https://example.com">ok</a>' in xss,
          "stm-markdown escapes raw tags, defangs javascript: links, keeps safe ones")


def main():
    prepare_workspace()
    if os.path.exists(CAPTURE_PATH):
        os.remove(CAPTURE_PATH)

    if db_query("select count(*) from tasks")[0][0] or db_query("select count(*) from goals")[0][0]:
        raise SystemExit("the web database is not fresh (tasks/goals exist) — "
                         "wipe the DB and restart the app+daemon for a clean goal-era run")

    mock = start_mock()
    time.sleep(2.0)

    try:
        # A previous suite run (or a crashed orchestrator) can leave the
        # daemon parked on a block; /stop releases it so dispatch works.
        try:
            http("POST", SADDLE + "/stop", timeout=5)
        except (urllib.error.URLError, OSError):
            pass

        auth = login("admin", "adminpass123")
        check(True, "login yields an auth slot")

        state = slot_public(auth, "/projects", "projects")
        act(auth, state["create"], {
            "name": "demo",
            "git_url": DEMO,
            "default_branch": "main",
            "summary": "Integration test repository",
        })
        check(any(p["name"] == "demo" for p in slot_public(auth, "/projects", "projects")["projects"]),
              "project demo created (local git_url, pushable origin)")

        state = slot_public(auth, "/saddles", "saddles")
        act(auth, state["register"], {"name": SADDLE_NAME, "host": SADDLE_HOST})
        registered = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
                      if s["name"] == SADDLE_NAME]
        check(len(registered) == 1 and registered[0]["token"].startswith("saddle-"),
              "saddle registered at %s with an API token" % SADDLE_HOST)

        print("== goal create -> start ==")
        goals_state = slot_public(auth, "/goals", "goals")
        act(auth, goals_state["create"], {
            "project_id": "1",
            "title": "Greet the demo",
            "brief": "Add a greeting file to the demo repo end to end.",
            "branch_name": "goal/demo-run",
            "priority": "3",
        })
        goal1 = int([g["id"] for g in slot_public(auth, "/goals", "goals")["goals"]
                     if g["title"] == "Greet the demo"][0])
        check(goal_state(auth, goal1)["status"] == "draft", "the goal lands in draft")
        act(auth, goal_state(auth, goal1)["start"], {})
        check(wait_goal_status(auth, goal1, "planning", timeout=30) is not None,
              "starting the goal moves it to planning (agent dispatched)")
        run_task = db_query(
            "select id from tasks where goal_id = ? and kind = 'goal_run'", (goal1,))[0][0]
        check(wait_for(lambda: task_state(auth, run_task)["status"]
                       in ("assigned", "working", "blocked", "suspended") or False,
                       timeout=30, what="goal run dispatched") is not None,
              "the goal-run task reaches the daemon (assigned/working/blocked)")
        goal_clone = os.path.join(WORKSPACE, "goal-%d" % goal1, "demo")
        check(wait_for(lambda: os.path.isdir(os.path.join(goal_clone, ".git")) or False,
                       timeout=30, what="the goal clone") is not None,
              "the goal clone appears under the shared workspace root (goal-<id>)")
        branch = subprocess.run(["git", "-C", goal_clone, "branch", "--show-current"],
                                capture_output=True, text=True).stdout.strip()
        check(branch == "goal/demo-run", "the goal clone sits on the user-named goal branch")

        print("== pre-plan question -> one answer, one row ==")
        check(wait_for(lambda: any(b["block_type"] == "question" and not b.get("answered")
                                   for b in goal_state(auth, goal1).get("blocks", [])) or False,
                       timeout=90, what="the agent's question block") is not None,
              "the pre-plan question surfaces as a goal block")
        check(wait_task_status(auth, run_task, "blocked", timeout=30) is not None,
              "the goal-run task parks blocked on its own block row")
        question = [b for b in goal_state(auth, goal1)["blocks"]
                    if b["block_type"] == "question" and not b.get("answered")][0]
        check(question["options"] == ["Demo only", "Whole workspace"],
              "the question carries its options")
        check("Which scope should the greeting cover" in question["question"],
              "the ask_question question arrives verbatim")
        check("confirm the blast radius" in question["body"], "the question carries the markdown body")
        dupes = db_query(
            "select count(*) from task_blocks tb join tasks t on t.id = tb.task_id"
            " where t.goal_id = ? and t.kind = 'goal_run' and tb.type = 'question'",
            (goal1,))[0][0]
        check(dupes == 1, "one question row despite the park-unwinding round-end push")
        dash = slot_public(auth, "/", "dashboard")
        check(any(int(r.get("goal_id") or 0) == goal1 for r in dash["pinned"]),
              "the dashboard pile pins the goal block")
        act(auth, question["answer_action"], {"option": "Demo only"})
        answered_rows = db_query(
            "select answer from task_blocks tb join tasks t on t.id = tb.task_id"
            " where t.goal_id = ? and t.kind = 'goal_run' and tb.type = 'question'",
            (goal1,))
        check(answered_rows and answered_rows[0][0] == "Demo only",
              "the ONE answer lands on the goal-run task's single row")

        print("== plan review: reject -> revise -> approve ==")
        check(wait_goal_status(auth, goal1, "plan_review", timeout=90) is not None,
              "the plan submission moves the goal to plan_review")
        check(wait_for(lambda: goal_state(auth, goal1).get("plan_pending_block") not in
                       (None, False) or False, timeout=30, what="pending plan block") is not None,
              "the pending plan block renders on the goal page")
        pending = goal_state(auth, goal1)["plan_pending_block"]
        check("Plan v1" in pending["body"], "the plan block carries the submitted markdown")
        # Plan 16 S2/R4: the goal page posts a `feedback` reply / a
        # `verdict` marker — the server owns the stable wording the
        # daemon's resume matching reads.
        act(auth, pending["answer_action"], {"feedback": "also export the greeting"})
        check(wait_goal_status(auth, goal1, "planning", timeout=30) is not None,
              "rejecting with feedback returns the goal to planning")
        check(wait_for(lambda: goal_state(auth, goal1).get("plan_pending_block") not in
                       (None, False) and "Plan v2" in
                       goal_state(auth, goal1)["plan_pending_block"].get("body", "") or False,
                       timeout=90, what="revised plan") is not None,
              "the agent revises and re-submits incorporating the feedback")
        revised = goal_state(auth, goal1)
        check(len([r for r in revised["plan_revisions"] if r.get("answered")]) >= 1,
              "the rejected revision stays discoverable behind the expander")
        act(auth, revised["plan_pending_block"]["answer_action"], {"verdict": "accept"})
        check(wait_goal_status(auth, goal1, "active", timeout=30) is not None,
              "approving with the agreed wording activates the goal")

        print("== task fan-out via the real callbacks ==")
        worker_row = wait_for(lambda: db_query(
            "select id from tasks where goal_id = ? and kind = 'task'", (goal1,)) or False,
            timeout=90, what="the agent-created task")
        worker_task = int(worker_row[0][0])
        check(bool(worker_row), "the goal agent created its worker task through the callback")
        creator = db_query("select creator, mode from tasks where id = ?", (worker_task,))[0]
        check(creator == ("goal", "implement"),
              "the created task carries creator=goal and the requested mode")
        check(wait_task_status(auth, worker_task, "queued", timeout=30) is not None
              or task_state(auth, worker_task)["status"] in ("working", "done"),
              "the agent enqueued its task (creator-only rule honoured by construction)")
        ephemeral = "task/%d-%d" % (goal1, worker_task)
        check(wait_for(lambda: task_state(auth, worker_task)["status"]
                       in ("assigned", "working", "done", "failed") or False,
                       timeout=60, what="worker dispatched") is not None,
              "the dispatcher hands the worker to the freed saddle")
        check(wait_for(lambda: db_query(
            "select count(*) from sessions s where s.task_id = ? and s.outcome = 'parked'"
            " and s.ended_at < (select min(started_at) from sessions where task_id = ?)",
            (run_task, worker_task))[0][0] > 0 or db_query(
            "select suspended_at is not null from tasks where id = ?", (run_task,))[0][0] or False,
            timeout=30, what="the goal agent parks") is not None,
            "the goal agent parked suspended after fanning out (parked session / suspended_at)")
        # Plan 16 item 23: busy DERIVES from the open session (the
        # schema carries no work cache) — while the worker runs, the
        # card reads busy with the worker as its current task. The
        # open window can be sub-second (the local mock answers
        # instantly), so the card read shares the predicate iteration
        # with the open-session probe.
        # The card's busy derives from the session; the ledger's
        # state_change rows are the durable record of the working window
        # (a fast mock can open-and-close the session between the
        # harness's card polls, so accept either the live card reading
        # busy-with-the-worker or the recorded working transition).
        check(wait_for(lambda: bool(db_query(
            "select 1 from events where scope_id = ? and kind = 'state_change'"
            " and payload like '%\"to\":\"working\"%'", (worker_task,))[0:1]) or bool(db_query(
            "select 1 from sessions where task_id = ? and ended_at is null",
            (worker_task,))[0:1]) or False,
            timeout=90, interval=0.2, what="derived busy while the worker runs") is not None,
              "the saddle card derives busy from the worker's session (no cached row)")

        print("== the worker's ephemeral worktree ==")
        # The branch probe rides the ledger: the tool_call event carries
        # the git command, its tool_result event the branch on stdout.
        branch_rows = wait_for(lambda: db_query(
            "select payload from events where scope_type = 'task' and scope_id = ?"
            " and kind = 'tool_result' and payload like ?",
            (worker_task, "%" + ephemeral + "%")) or False,
            timeout=90, what="the worker's branch probe")
        check(branch_rows and ephemeral in (branch_rows[0][0] or ""),
              "the worker runs inside its ephemeral branch %s (tool_result stdout)" % ephemeral)
        branch_call = db_query(
            "select count(*) from events where scope_type = 'task' and scope_id = ?"
            " and kind = 'tool_call' and payload like '%git branch --show-current%'",
            (worker_task,))[0][0]
        check(branch_call == 1, "the branch probe's tool_call event sits in the ledger")
        worktree_dir = os.path.join(WORKSPACE, "task-%d" % worker_task)
        check(wait_for(lambda: task_state(auth, worker_task)["status"] == "done" or False,
                       timeout=120, what="worker done") is not None,
              "the worker completes with its report")
        detail = task_state(auth, worker_task)
        check("greeting.txt" in (detail.get("summary") or ""),
              "the worker summary carries its completion report")
        check(not os.path.exists(worktree_dir), "the worker's worktree auto-pruned on done")
        kept = subprocess.run(["git", "-C", goal_clone, "rev-parse", "--verify", ephemeral],
                              capture_output=True).returncode == 0
        merged_already = subprocess.run(
            ["git", "-C", goal_clone, "log", "--oneline", "-5"],
            capture_output=True, text=True).stdout
        check(kept or "greeting from the goal" in merged_already,
              "the ephemeral branch survives for the merge (or the merge already consumed it)")

        print("== coalesced resume -> merge -> push -> completion review ==")
        check(wait_for(lambda: any(
            e["kind"] == "task-done" and int(e["task_id"]) == worker_task
            for e in goal_state(auth, goal1).get("events", [])) or False,
            timeout=30, what="the task-done narrative row") is not None,
              "the worker's terminal transition lands in the derived goal narrative")
        # the goal agent's /status reports the branch push state while it
        # works; sample the moment the resumed agent is mid-round, then
        # cache it through the manual saddle refresh
        check(wait_for(lambda: (saddle_status() or {}).get("goal_branches") or False,
                       timeout=90, what="the resumed agent exposes goal_branches") is not None,
              "the daemon's /status exposes goal_branches while the goal agent works")
        saddles_state = slot_public(auth, "/saddles", "saddles")
        card = [s for s in saddles_state["saddles"] if s["name"] == SADDLE_NAME][0]
        act(auth, card["refresh"], {})
        check(wait_for(lambda: any(
            "demo" in (b.get("branch") or "") and isinstance(b.get("pushed"), bool)
            for b in goal_state(auth, goal1).get("push_branches", []))
            or False, timeout=60, what="cached push state") is not None,
              "the refresh probe caches the branch push state on the goal (live chips)")
        check(wait_goal_status(auth, goal1, "completion_review", timeout=120) is not None,
              "the resumed agent merges, pushes and enters completion review")
        state = goal_state(auth, goal1)
        completion = state["completion_pending_block"]
        # Plan 16 S1 (canonical-on-accept): pending, the report lives
        # ONLY in the block — goals.completion_report stays empty
        # until the accept verdict writes the winning body.
        check("greeting.txt added via one worker task" in (completion or {}).get("body", ""),
              "the completion report renders on the goal page (the pending block's body)")
        check(not (state.get("completion_report") or ""),
              "goals.completion_report stays clean before the verdict (canonical-on-accept)")
        merged = subprocess.run(
            ["git", "-C", goal_clone, "show", "goal/demo-run:greeting.txt"],
            capture_output=True, text=True).stdout
        check("hello-from-the-goal" in merged, "the merged task branch landed on the goal branch")
        deleted = subprocess.run(["git", "-C", goal_clone, "rev-parse", "--verify", ephemeral],
                                 capture_output=True).returncode != 0
        check(deleted, "the goal agent deleted the merged task branch")
        pushed = subprocess.run(["git", "-C", DEMO, "rev-parse", "--verify", "goal/demo-run"],
                                capture_output=True).returncode == 0
        check(pushed, "the goal branch was pushed to origin (push-before-review)")

        print("== completion accept ==")
        # Goal 21: the accept-time push gate is gone — the daemon already
        # refused the agent's completion attempt until every goal branch
        # was pushed (push-before-review, asserted above), so the branch
        # is on origin here and the verdict path needs no second check.
        check("greeting.txt added via one worker task" in (completion or {}).get("body", ""),
              "the run task's completion block carries the report")
        winning_body = (completion or {}).get("body", "")
        accept = act(auth, completion["answer_action"], {"verdict": "accept"})
        check(not any(d.get("type") == "notify" and d.get("kind") == "error" for d in accept),
              "the accept passes (no web-side push gate on the verdict)")
        check(wait_goal_status(auth, goal1, "complete", timeout=30) is not None,
              "accepting the verdict marker completes the goal (server-owned wording)")
        check(wait_task_status(auth, run_task, "done", timeout=30) is not None,
              "the goal-run task closes done with the goal")
        final = goal_state(auth, goal1)
        check(final.get("completion_report") == winning_body,
              "the accept canonises the winning block's body as goals.completion_report (S1)")

        print("== audit trail (the derived narrative + the unified ledger) ==")
        narrative = goal_state(auth, goal1)["events"]
        kinds = [e["kind"] for e in narrative]
        for expected in ("started", "block-raised", "plan-submitted", "plan-rejected",
                          "plan-approved", "task-created", "task-done",
                          "completion-submitted", "goal-approved"):
            check(expected in kinds, "the derived narrative records %s" % expected)
        check("system" not in kinds,
              "mechanical system rows never reach the user-facing narrative (U2)")
        check(all(k != "task-enqueued" for k in kinds),
              "timestamp-less bookkeeping never returns (task-enqueued is gone)")
        check(db_query(
            "select count(*) from sqlite_master where type = 'table'"
            " and name = 'goal_events'")[0][0] == 0,
            "the goal_events table is gone (nothing is written on ingest)")
        check(db_query(
            "select count(*) from sqlite_master where type = 'table'"
            " and name = 'session_commands'")[0][0] == 0,
            "the session_commands table is gone (the ledger is the record)")
        run_kinds = {row[0] for row in db_query(
            "select distinct kind from events where scope_type = 'task' and scope_id = ?",
            (run_task,))}
        check({"turn_text", "tool_call", "tool_result", "usage", "lifecycle"} <= run_kinds,
              "the goal agent's ledger carries the protocol kinds (%s)" % sorted(run_kinds))
        worker_kinds = {row[0] for row in db_query(
            "select distinct kind from events where scope_type = 'task' and scope_id = ?",
            (worker_task,))}
        check({"turn_text", "tool_call", "tool_result", "usage"} <= worker_kinds,
              "the worker's ledger carries the protocol kinds (%s)" % sorted(worker_kinds))
        daemon_kinds = {row[0] for row in db_query(
            "select distinct kind from events where session_id is not null")}
        check(daemon_kinds <= {"turn_text", "tool_call", "tool_result", "progress",
                               "comment", "lifecycle", "usage", "state_change", "yeehaw"},
              "every daemon-pushed kind is a §7.2 kind (%s)" % sorted(daemon_kinds))
        remark_rows = db_query(
            "select actor, payload from events where kind = 'comment'"
            " and session_id is not null and scope_id = ?", (worker_task,))
        check(len(remark_rows) == 1 and remark_rows[0][0] == "model"
              and "the mandate starts on the ephemeral branch" in (remark_rows[0][1] or ""),
              "the add_comment remark rides the ledger as a model-actor comment event")

        print("== goal 9 tool parity through both binaries ==")
        # yeehaw: the celebration rode the ledger as a yeehaw event with
        # the message intact (the task log renders it in the comments
        # bucket with the festive badge — pinned by test_task_log_js)
        whoop_rows = db_query(
            "select actor, payload from events where kind = 'yeehaw'"
            " and session_id is not null and scope_id = ?", (worker_task,))
        check(len(whoop_rows) == 1 and whoop_rows[0][0] == "model"
              and "genuinely fun task" in (whoop_rows[0][1] or ""),
              "the yeehaw celebration lands as a model-actor yeehaw event")
        # save_note: the note POSTed through the real /api/saddle/note
        # and persists on the demo project for future assignments
        note_rows = db_query(
            "select body from project_notes where project_id = 1"
            " and body like '%ephemeral branch is task/%'")
        check(len(note_rows) == 1
              and "hello-from-the-goal" in (note_rows[0][0] or ""),
              "the save_note tool's note persists on the project (real /api/saddle/note)")
        # query_llm: the side LLM answered and ONLY the answer reached
        # the model — the command's raw output never appears in any
        # tool result the conversation carries
        tool_turns_all = [text for role, text in capture_messages() if role == "tool"]
        check(any("usage: greet --style <flavour>" in turn for turn in tool_turns_all),
              "the query_llm side answer reached the model as its tool result")
        # the command string rides the result for context, but its
        # OUTPUT text (the printf'd body) never does — only the side
        # LLM saw that
        output_only = "PRESET=slow\\nTUNE=film usage"
        check(not any(output_only in turn for turn in tool_turns_all),
              "the hidden command's OUTPUT text never reached the model")
        check(not any("PRESET=slow" in (m or "") for m in
                      [t for t in tool_turns_all if "answer" not in t]),
              "no tool result leaks the command output")
        # the side call itself stays a two-message context (system +
        # user, no tool history): the MAIN conversation — every request
        # carrying the workspace-agent system prompt — never sees the
        # piped text as any turn
        arrivals = []
        if os.path.exists(CAPTURE_PATH):
            with open(CAPTURE_PATH) as handle:
                arrivals = [json.loads(line) for line in handle if line.strip()]
        main_turns = [message.get("content") or ""
                      for arrival in arrivals
                      if any(m.get("role") == "system"
                             and "workspace agent" in (m.get("content") or "")
                             for m in arrival.get("messages", []))
                      for message in arrival.get("messages", [])]
        # The tool RESULT legitimately echoes the command STRING (the
        # printf argv) for context; what must never appear is the
        # command's OUTPUT. The output-only marker is the joined body
        # the command printed (newlines joined, not shell-escaped).
        printed = "CRF=-18\nPRESET=slow\nTUNE=film usage: greet --style <flavour>"
        check(not any(printed in turn for turn in main_turns),
              "the command's printed output never injects into the main conversation")
        query_kinds = db_query(
            "select count(*) from events where kind = 'query'")[0][0]
        check(query_kinds == 0
              or db_query("select count(*) from events where kind = 'query'"
                          " and payload like '%Which usage line%'")[0][0] == query_kinds,
              "the QUERY audit entry (command + question) is ledger-visible when present")
        # list_goals: the goals wire answers through the registered
        # saddle's token with this goal's row and the detail flags
        reg_token = db_query("select token from saddles where name = ?",
                             (SADDLE_NAME,))[0][0]
        status, body = saddle_api("GET", "/api/saddle/goals?project=demo", reg_token)
        rows = body.get("goals", []) if status == 200 else []
        check(status == 200 and any(int(r["id"]) == goal1 for r in rows),
              "the goals wire lists this goal for the demo project (%s)" % status)
        status, body = saddle_api("GET", "/api/saddle/goals?goal_id=%d" % goal1, reg_token)
        detail = body.get("goal", {}) if status == 200 else {}
        check(status == 200 and detail.get("has_plan") is True
              and detail.get("has_report") in (True, False)
              and detail.get("branch") == "goal/demo-run",
              "the goals wire's goal_id detail carries branch + has_plan/has_report")
        narrative_remarks = [e["text"] for e in goal_state(auth, goal1)["events"]
                             if e["kind"] == "comment"]
        check(any(text.startswith("Probing done") for text in narrative_remarks),
              "the goal narrative renders the worker's remark as a comment row")
        plan_call = db_query(
            "select count(*) from events where scope_type = 'task' and scope_id = ?"
            " and kind = 'tool_call' and payload like '%submit_plan%'", (run_task,))[0][0]
        check(plan_call >= 2, "both plan submissions sit as tool_call events")
        usage_rows = db_query(
            "select payload from events where kind = 'usage' and scope_type = 'task'"
            " and scope_id = ? limit 1", (worker_task,))
        check(bool(usage_rows) and "prompt_tokens" in (usage_rows[0][0] or "")
              and "elapsed_ms" in (usage_rows[0][0] or ""),
              "usage events carry prompt/completion tokens and elapsed time")

        print("== the terminal watermark (C3) drives the resume ==")
        # The fresh-DB schema diff (the joint-QA gate): the removed
        # surfaces stay removed — the saddle work cache (item 23), the
        # /update-era columns (arch 19) and the legacy token table
        # (arch 24).
        saddle_cols = {row[1] for row in db_query("pragma table_info('saddles')")}
        check(not ({"current_task_id", "allocated_at"} & saddle_cols),
              "the saddles work-cache columns are gone (sessions are truth)")
        task_cols = {row[1] for row in db_query("pragma table_info('tasks')")}
        check("working_branch" not in task_cols,
              "tasks.working_branch is gone (only the deleted /update done wrote it)")
        check(db_query(
            "select count(*) from sqlite_master where type = 'table'"
            " and name = 'api_tokens'")[0][0] == 0,
            "the api_tokens table is gone (saddles.token is the single credential store)")
        watermark_row = db_query(
            "select last_terminal_watermark from goals where id = ?", (goal1,))
        worker_done_at = db_query(
            "select completed_at from tasks where id = ?", (worker_task,))[0][0]
        check(watermark_row[0][0] is not None and worker_done_at is not None
              and watermark_row[0][0] >= worker_done_at,
              "the goals row's terminal watermark advanced past the worker's transition"
              " (the resume consumed it)")

        print("== the harness bookkeeping stays ledger-internal (C2 + U2) ==")
        # Plan 16 S5 (U2): the mechanical rows (restores, park notices)
        # author NO user-facing comment rows — they persist as
        # system-authored comment events (the unified ledger) and never
        # render on a user-facing timeline.
        mechanical = db_query(
            "select count(*) from events where kind = 'comment' and actor = 'system'"
            " and (payload like '%session context restored%'"
            " or payload like '%parked until%')")[0][0]
        check(mechanical >= 1,
              "the harness bookkeeping (restore, park) persists as system comment events")
        thread = goal_state(auth, goal1)["comments"]
        check(not any("session context restored" in c["body"] or "parked until" in c["body"]
                      for c in thread),
              "mechanical rows never render on the goal thread (U2)")
        run_thread = task_state(auth, run_task)["comments"]
        check(not any("session context restored" in c["body"] or "parked until" in c["body"]
                      for c in run_thread),
              "the task page reads the same filtered thread (no mechanical rows)")
        check(not any(c.get("from_agent") and "[harness]" in c["body"]
                      for c in thread),
              "no harness-authored row renders as agent content")

        print("== parks close their sessions (D4) ==")
        outcomes = db_query(
            "select outcome, ended_at is not null from sessions where task_id in (?, ?)",
            (run_task, worker_task))
        check(bool(outcomes)
              and all(row[0] in ("parked", "done") and row[1] for row in outcomes),
              "every run/worker session closed exactly once — parks stamp outcome 'parked',"
              " none dangle open (%s)" % sorted({row[0] for row in outcomes}))

        print("== D-side through both binaries: tool results + loop warning ==")
        messages = capture_messages()
        tool_turns = [text for role, text in messages if role == "tool"]
        user_turns = [text for role, text in messages if role == "user"]
        check(any("probing-the-workspace" in turn for turn in tool_turns),
              "the probes' tool results reached the model as role=tool turns")
        check(any(ephemeral in turn for turn in tool_turns),
              "the branch probe's result (the ephemeral branch name) reached the model")
        check(any("you ran the exact same command" in turn and "you are in a loop" in turn
                  for turn in user_turns),
              "the loop-warning notice reached the model as its next user turn")
        check(any("Human selected" in turn for turn in user_turns),
              "the question's verdict wording reached the model on resume")

        print("== event log page + SSE ==")
        # The §7.4 JSON page: newest-first window of the run task's
        # ledger (id, session_id, seq, kind, actor, payload).
        status, page = saddle_api("GET", "/tasks/%d/events?limit=200" % run_task, None)
        page_events = page.get("events", []) if status == 200 else []
        check(status == 200 and len(page_events) >= 10,
              "the event log page serves the run's ledger (%d events)" % len(page_events))
        ids = [e["id"] for e in page_events]
        check(ids == sorted(ids, reverse=True), "the page is newest-first")
        check(any(e["kind"] == "tool_call" and e["payload"].get("tool") == "submit_plan"
                  for e in page_events),
              "turn cards render: submit_plan tool_call events in the page")
        check(any(e["kind"] == "tool_result" and e["payload"].get("tool") == "run_command"
                  and "stdout" in e["payload"].get("result", "") for e in page_events),
              "tool_result events carry the §7.1 result shape (stdout embedded)")
        check(any(e["kind"] == "turn_text" and "Reading the brief" in e["payload"].get("text", "")
                  for e in page_events),
              "the narrated turn renders as turn_text")
        check(all(e.get("seq") is not None and e["seq"] > 0 for e in page_events
                  if e["kind"] in ("turn_text", "tool_call", "tool_result", "usage")),
              "daemon events carry their per-session seq")
        state = goal_state(auth, goal1)
        check(any(r.get("terminal_url", "").endswith("/tasks/%d/log" % run_task)
                  for r in state.get("runs", [])),
              "the goal page's run list deep-links each session's event log")
        request = urllib.request.Request(WEB + "/tasks/%d/events/stream" % run_task)
        with urllib.request.urlopen(request, timeout=15) as stream:
            check(stream.headers.get("Content-Type", "").startswith("text/event-stream"),
                  "event stream is text/event-stream")
            streamed = []
            deadline = time.monotonic() + 10
            while time.monotonic() < deadline and len(streamed) < 120:
                line = stream.readline().decode().strip()
                if not line.startswith("data: ") or line == "data: ping":
                    continue
                try:
                    streamed.append(json.loads(line[6:]))
                except ValueError:
                    continue
                if len(streamed) >= len(page_events):
                    break
            check(len(streamed) >= 10,
                  "the stream replays the initial most-recent-30 window (%d events)" % len(streamed))
            check(any(e.get("kind") == "tool_call" for e in streamed)
                  and any(e.get("kind") == "tool_result" for e in streamed),
                  "the SSE replay carries tool_call/tool_result events from the daemon")

        pulse = directives("/goals/%d" % goal1, auth)
        check(any(d.get("type") == "subscribe" for d in pulse),
              "goal page subscribes to the pulse channel for live refresh")

        print("== push wire surface (§7.2 chunks + results states) ==")
        saddles_state = slot_public(auth, "/saddles", "saddles")
        act(auth, saddles_state["register"], {"name": "saddle-e2e-api", "host": "localhost:7199"})
        api_rows = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
                    if s["name"] == "saddle-e2e-api"]
        check(len(api_rows) == 1 and api_rows[0]["token"].startswith("saddle-"),
              "registration returns an API token")
        api_saddle = int(api_rows[0]["id"])
        api_token = api_rows[0]["token"]

        tasks_state = slot_public(auth, "/tasks", "tasks")
        act(auth, tasks_state["create"], {
            "project_id": "1",
            "title": "API surface probe",
            "description": "Chunks ingestion, exactly-once pushes",
            "priority": "6",
            "task_branch": "task/api-surface",
        })
        api_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                        if t["title"] == "API surface probe"][0])
        db = sqlite3.connect(DB_PATH)
        db.execute("insert into sessions (task_id, saddle_id, started_at)"
                   " values (?, ?, strftime('%s','now'))", (api_task, api_saddle))
        db.execute("update tasks set status = 'working' where id = ?", (api_task,))
        db.commit()
        api_session = db.execute("select max(id) from sessions").fetchone()[0]
        db.close()

        def push_state(state="working", block=None, summary=None, entries=None):
            """One state push in the daemon's /results shape."""
            payload = {"session_id": api_session, "state": state}
            if block is not None:
                payload["block"] = block
            if summary is not None:
                payload["summary"] = summary
            if entries is not None:
                payload["entries"] = entries
            return saddle_api("POST", "/api/saddle/results", api_token, payload)

        def push_events(events):
            """One §7.2 chunks batch ({task_id, session_id, events})."""
            return saddle_api("POST", "/api/saddle/chunks", api_token,
                              {"task_id": api_task, "session_id": api_session,
                               "events": events})

        status, _ = saddle_api("POST", "/api/saddle/results", None,
                               {"session_id": api_session, "state": "working"})
        check(status == 401, "results push without a token is rejected")
        status, _ = saddle_api("POST", "/api/saddle/results", "saddle-not-a-token",
                               {"session_id": api_session, "state": "working"})
        check(status == 401, "results push with an invalid token is rejected")
        status, _ = saddle_api("POST", "/api/saddle/chunks", None,
                               {"session_id": api_session, "events": []})
        check(status == 401, "chunks push without a token is rejected")
        status, _ = saddle_api("POST", "/api/saddle/chunks", api_token,
                               {"session_id": api_session, "events": [
                                   {"kind": "progress", "actor": "model", "ts": "now"}]})
        check(status == 400, "an event without a positive seq is refused (seq is mandatory)")

        # The poll-era ingest pair is GONE (arch 19): both routes 404
        # for any caller, authenticated or not — the subtask/plan
        # refusal arms (plan 12 removal) died with the channel.
        status, _ = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, api_token,
                               {"type": "status", "session_id": api_session,
                                "data": {"state": "working"}})
        check(status == 404, "POST /api/saddle/{id}/update is gone (404)")
        status, _ = saddle_api("GET", "/api/saddle/%d/poll" % api_saddle, api_token)
        check(status == 404, "GET /api/saddle/{id}/poll is gone (404)")
        status, _ = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, api_token,
                               {"type": "subtask", "session_id": api_session,
                                "data": {"title": "Ghost child"}})
        check(status == 404,
              "the /update subtask arm is gone with the channel (404, plan 12 removal)")
        status, _ = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, api_token,
                               {"type": "plan", "session_id": api_session,
                                "data": {"plan": "# nope"}})
        check(status == 404,
              "the /update plan arm is gone with the channel (404, plans are goal-only)")

        # The retired results-entries array is consumed silently: no
        # comment row, no projection — round entries live as ledger
        # events on the chunks channel now.
        status, body = push_state(entries=[{"id": 9001, "command": "saddle comment",
                                            "output": "Surveyed the module", "exit_code": 0,
                                            "type": "comment",
                                            "timestamp": "2026-09-13T00:00:00Z"}])
        check(status == 200 and body.get("ok"),
              "a results push carrying entries is accepted (consumed silently)")
        check(not any(c["body"] == "Surveyed the module"
                      for c in task_state(auth, api_task)["comments"]),
              "the retired entries array projects nothing (no comment row)")

        # §7.2 ingestion: a progress event lands in the ledger and the
        # task's progress view; the (session_id, seq) unique key makes
        # redelivery exactly-once.
        status, body = push_events([{"seq": 9000, "kind": "progress", "actor": "model",
                                     "ts": "2026-09-13T00:00:00Z",
                                     "payload": {"text": "Migrated the schemas"}}])
        check(status == 200 and body.get("stored") == 1,
              "the chunks push stores the progress event")
        status, body = push_events([{"seq": 9000, "kind": "progress", "actor": "model",
                                     "ts": "2026-09-13T00:00:01Z",
                                     "payload": {"text": "Migrated the schemas AGAIN"}}])
        check(status == 200 and body.get("stored") == 0,
              "the redelivered seq is consumed silently (UNIQUE(session_id, seq))")
        progress_view = task_state(auth, api_task)["progress"]
        check(len([p for p in progress_view
                   if p["summary"] in ("Migrated the schemas", "Migrated the schemas AGAIN")]) == 1
              and progress_view[0]["summary"] == "Migrated the schemas",
              "progress events project exactly once per seq")

        status, body = push_state(state="blocked", block={
            "question": "Which database?", "type": "question",
            "options": ["SQLite", "Postgres"], "body": "# Pick one"})
        check(status == 200 and body.get("ok"),
              "the blocked push raises the question through the results channel")
        parked = db_query("select outcome from sessions where id = ?", (api_session,))[0][0]
        check(parked == "parked",
              "the raise parked the round's session (outcome 'parked', D4)")
        status, body = push_state(state="blocked", block={
            "question": "Review the migration", "type": "review",
            "body": "Migration **ready** for review"})
        check(status == 200, "the review-type block push is accepted")

        detail = task_state(auth, api_task)
        check(detail["status"] == "review_requested", "the review block flips the task status")
        check([b["options"] for b in detail["blocks"]
               if b["block_type"] == "question"] == [["SQLite", "Postgres"]],
              "the pushed block carries its options")
        check(detail["progress"][0]["summary"] == "Migrated the schemas",
              "progress history newest-first")
        status, _ = push_state(state="working")
        check(status == 200 and task_state(auth, api_task)["status"] == "review_requested",
              "a park-unwinding working push never resurrects the parked round")

        question = [b for b in detail["blocks"] if b["block_type"] == "question"][0]
        act(auth, question["answer_action"], {"option": "SQLite"})
        check([b["answer"] for b in task_state(auth, api_task)["blocks"]
               if b["block_type"] == "question"] == ["SQLite"],
              "option answer stored verbatim (the resuming assignment formats it)")
        act(auth, detail["comment"],
            {"body": "Check the migration output", "quote": "ERROR: table demos exists"})
        check(any(c["body"] == "Check the migration output" and "ERROR" in (c.get("quote") or "")
                  and not c.get("from_agent")
                  for c in task_state(auth, api_task)["comments"]),
              "the quoted user comment lands on the task thread (delivery rides the"
              " assignment carrier / dispatcher pass — unit-pinned)")
        act(auth, detail["cancel"], {"task_id": str(api_task)})

        print("== stm-markdown ==")
        markdown_renderer_checks()

        print("== quiesce ==")
        check(wait_for(
            lambda: (lambda rows: all(int(s["current_task_id"] or 0) == 0
                                      and s["status"] == "idle"
                                      for s in rows))(
                [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
                 if s["name"] == SADDLE_NAME]) or False,
            timeout=30, what="saddle idle with no current task") is not None,
            "saddle derived idle with no running work once the goal finished")
    finally:
        if mock is not None:
            mock.terminate()

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


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