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

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 works on a real git checkout under WORKSPACE/demo,
powered by saddle/test/mock_llm.py with a deterministic script.

Sections:
1. explore task (bash + comment + done) and dependency gating (v1 loop)
2. the Part 2 feature tour, driven by `saddle` commands the mock emits:
   plan via heredoc, `saddle query` with piped stdin, progress updates,
   a multiple-choice block (answered through the web UI), an RFC block
   with a markdown body, a `saddle review` banner approved via the UI,
   and a quoted user comment delivered to the daemon over /notify
3. the ingest API surface: bearer-token auth (401/200), typed updates,
   the poll endpoint's exactly-once handover
4. SSE: terminal stream replay + live push (with piped query input)
5. the stm-markdown renderer shipped in statum.js (via node)
6. the audit trail in SQLite

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 (below), SADDLE_PROJECTS_ROOT=$WORKSPACE
- a saddle daemon reachable at SADDLE
- a git repo with at least two commits at $WORKSPACE/demo

Environment:
  WEB            web app base URL      (default http://localhost:8080)
  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

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:8080")
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", "")
DEMO = os.path.join(WORKSPACE, "demo")
HOSTNAME = socket.gethostname().split(".")[0]

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 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 mock_argv():
    heredoc = "cat << 'SADDLE_EOF' | %s\n%%s\nSADDLE_EOF"
    responses = [
        "# Inspecting the checkout\nls -la %s" % DEMO,
        "# Checking history\ngit -C %s log --oneline" % DEMO,
        'saddle comment "Inspected the checkout"\nsaddle done "Explored the demo repository"',
        "# Starting the follow-up\necho task-two-running",
        'saddle done "Follow-up audit complete"',
        # Part 2 feature tour (task 3)
        "# First, write the plan\n" + (heredoc % "saddle plan") % (
            "# Feature tour plan\n\n1. Query the host\n2. Ask a question\n"
            "3. Propose an RFC\n4. Request review"),
        "# Confirm the working branch\ngit branch --show-current",
        "# Side question about this host\nhostname | saddle query \"What host is this?\"",
        "# Report progress\nsaddle progress-update queried the host",
        "# Need a decision\n" + (heredoc % 'saddle block question "Which database should we use?" "SQLite" "Postgres" "Flat files"') % (
            "# Storage choice\n\nWe need somewhere to keep the audit trail.\n\n"
            "- **SQLite** — zero admin\n- **Postgres** — serious scale\n- **Flat files** — yolo"),
        "# Proposing the module layout\n" + (heredoc % "saddle block rfc") % (
            "# Proposed layout\n\n```text\nsrc/\n  api/\n  services/\n```\n\n"
            "The `api` layer stays thin; services own the domain."),
        "# Ready for review\n" + (heredoc % "saddle review") % (
            "## Review: feature tour\n\nEverything on the tour is wired up.\n\n"
            "- plans render\n- blocks answer\n- reviews approve"),
        "# Waiting for any human input before finishing\nsleep 12",
    ]
    argv = []
    for response in responses:
        argv += ["-r", response]
    argv += [
        "--on-answer", "# Human decision received\nsaddle progress-update human input integrated",
        "--on-comment", 'saddle comment "Got the human comment"\nsaddle done "Part 2 feature tour complete"',
        "--on-query", "The hostname is %s." % HOSTNAME,
    ]
    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]

    try:
        parts = [extract(name, body)
                 for name in ("safeMarkdownUrl", "mdLinks", "mdInline", "renderMarkdown")]
    except ValueError:
        check(False, "statum.js carries the markdown renderer functions")
        return
    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():
    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": "https://example.com/demo.git",
            "default_branch": "main",
            "summary": "Integration test repository",
        })
        check(any(p["name"] == "demo" for p in slot_public(auth, "/projects", "projects")["projects"]),
              "project demo created")

        state = slot_public(auth, "/tasks", "tasks")
        act(auth, state["create"], {
            "project_id": "1",
            "title": "Explore the demo repository",
            "description": "List the checkout and inspect its git history.",
            "priority": "3",
        })
        act(auth, state["create"], {
            "project_id": "1",
            "title": "Follow-up audit",
            "description": "Runs after the explore task completes.",
            "priority": "4",
        })
        act(auth, state["create"], {
            "project_id": "1",
            "title": "Part 2 feature tour",
            "description": "Plan, query, block, RFC, review, comment — the works.",
            "priority": "5",
            "task_branch": "task/3-feature-tour",
        })
        titles = [t["title"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]]
        check(titles == ["Explore the demo repository", "Follow-up audit", "Part 2 feature tour"],
              "three tasks queued in priority order")

        tour = task_state(auth, 3)
        check(tour["branch_from"] == "main", "branch_from prefilled from the project default branch")
        check(tour["task_branch"] == "task/3-feature-tour", "task_branch stored at creation")

        detail = task_state(auth, 2)
        act(auth, detail["dependency"], {"depends_on_task_id": "1"})
        check(wait_for(lambda: any(int(d.get("id", 0)) == 1
                                   for d in task_state(auth, 2).get("dependencies", [])) or False,
                       timeout=5, what="dependency edge of task 2"),
              "task 2 depends on task 1")

        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("== explore task ==")
        left_queue = wait_for(
            lambda: (lambda s: s if s != "queued" else False)(task_state(auth, 1).get("status")) or False,
            timeout=15, what="task 1 to leave the queue")
        check(left_queue is not None, "dispatcher picks task 1 up within 15s (status %s)" % left_queue)
        check(wait_for(lambda: (saddle_status() or {}).get("state") in ("working", "blocked", "done") or False,
                       timeout=15, what="saddle active") is not None,
              "saddle daemon receives the task")
        check(wait_task_status(auth, 1, "working", timeout=30) is not None,
              "task 1 flips to working from daemon status")
        check(wait_task_status(auth, 1, "done", timeout=90) is not None, "task 1 completes")

        detail = task_state(auth, 1)
        check("Explored the demo repository" in (detail.get("summary") or ""),
              "task 1 carries the agent's summary")
        check(any(c["author"] == "bot" and "Inspected the checkout" in c["body"]
                  for c in detail.get("comments", [])),
              "agent comment projected into task_comments")
        check(detail.get("sessions") and detail["sessions"][0].get("outcome") == "done",
              "session closed with outcome done")

        print("== dependency gating ==")
        check(wait_task_status(auth, 2, "done", timeout=90) is not None,
              "task 2 (dependent) dispatches and completes after task 1")

        print("== part 2 feature tour ==")
        check(wait_task_status(auth, 3, "working", timeout=30) is not None,
              "task 3 dispatches to the saddle")

        # plan via `saddle plan` heredoc
        check(wait_for(lambda: "# Feature tour plan" in (task_state(auth, 3).get("plan") or "") or False,
                       timeout=60, what="task 3 plan") is not None,
              "saddle plan markdown lands in the task plan slot")
        db = sqlite3.connect(DB_PATH)
        stored_plan = db.execute("select plan from tasks where id = 3").fetchone()[0] or ""
        check("1. Query the host" in stored_plan and "4. Request review" in stored_plan,
              "plan stored in the tasks table verbatim")
        db.close()

        # `saddle query` with piped stdin
        check(wait_for(lambda: any("query" == row[0]
                                   for row in query_commands(3)) or False,
                       timeout=60, what="query entry") is not None,
              "saddle query logged as a typed terminal entry")

        # the workspace clone sits on the task's branch
        db = sqlite3.connect(DB_PATH)
        branch_rows = db.execute(
            "select sc.output_text from session_commands sc join sessions s on s.id = sc.session_id"
            " where s.task_id = 3 and sc.command_text = 'git branch --show-current'"
        ).fetchall()
        db.close()
        check(branch_rows and (branch_rows[0][0] or "").strip() == "task/3-feature-tour",
              "workspace clone checks out the task branch")
        db = sqlite3.connect(DB_PATH)
        query_rows = db.execute(
            "select sc.command_text, sc.output_text, sc.piped_input from session_commands sc"
            " join sessions s on s.id = sc.session_id where s.task_id = 3 and sc.type = 'query'"
        ).fetchall()
        db.close()
        check(len(query_rows) == 1 and "What host is this?" in query_rows[0][0],
              "query entry carries the question as its command")
        check(query_rows and HOSTNAME in (query_rows[0][1] or ""),
              "query entry carries the side-LLM answer as output")
        check(query_rows and (query_rows[0][2] or "").strip(),
              "query entry stores the piped stdin in session_commands.piped_input")

        # progress updates
        check(wait_for(lambda: any("queried the host" == p["summary"]
                                   for p in task_state(auth, 3).get("progress", [])) or False,
                       timeout=60, what="first progress update") is not None,
              "saddle progress-update appears on the task page")
        detail = task_state(auth, 3)
        check(detail["progress"][0]["summary"] == "queried the host",
              "latest progress update shown first")

        # block question with options + markdown body
        check(wait_task_status(auth, 3, "blocked", timeout=90) is not None,
              "saddle block question flips the task to blocked")
        detail = task_state(auth, 3)
        open_q = [b for b in detail.get("blocks", [])
                  if b["block_type"] == "question" and not b.get("answered")]
        check(len(open_q) == 1 and "Which database" in open_q[0]["question"],
              "block question surfaced in the web UI")
        check(open_q and open_q[0]["options"] == ["SQLite", "Postgres", "Flat files"],
              "block options stored on the block")
        check(open_q and "# Storage choice" in open_q[0]["body"]
              and "**SQLite**" in open_q[0]["body"],
              "block markdown body stored on the block")
        act(auth, open_q[0]["answer_action"], {"option": "Postgres"})
        answered = [b for b in task_state(auth, 3)["blocks"]
                    if b["block_type"] == "question"]
        check(any(b.get("answer") == "Human selected: Postgres" for b in answered),
              "option answer formatted for the agent")
        # the resume window can be brief; the agent continuing past the
        # answer is the durable signal (it raises the RFC next)
        check(wait_for(lambda: any(b.get("block_type") == "rfc"
                                   for b in task_state(auth, 3).get("blocks", [])) or False,
                       timeout=90, what="agent continues after the answer") is not None,
              "answered question unblocks the agent")

        # RFC block with markdown body
        check(wait_task_status(auth, 3, "blocked", timeout=90) is not None,
              "saddle block rfc flips the task to blocked")
        detail = task_state(auth, 3)
        open_rfc = [b for b in detail.get("blocks", [])
                    if b["block_type"] == "rfc" and not b.get("answered")]
        check(len(open_rfc) == 1 and "# Proposed layout" in open_rfc[0]["body"],
              "RFC markdown body stored on the block")
        check(open_rfc and "```text" in open_rfc[0]["body"],
              "RFC body keeps fenced code blocks intact")
        act(auth, open_rfc[0]["answer_action"], {"answer": "Layout approved as proposed"})
        check(any(b.get("answer") == "Layout approved as proposed"
                  for b in task_state(auth, 3)["blocks"] if b["block_type"] == "rfc"),
              "RFC reply recorded verbatim")
        # the review banner following proves the agent resumed past the RFC
        check(wait_for(lambda: task_state(auth, 3).get("review_pending") or False,
                       timeout=90, what="agent resumes after the RFC") is not None,
              "answered RFC unblocks the agent")

        # review banner + approve
        check(wait_task_status(auth, 3, "review_requested", timeout=90) is not None,
              "saddle review flips the task to review_requested")
        detail = task_state(auth, 3)
        check(detail["review_pending"] and "Everything on the tour is wired up" in detail["review_summary"],
              "review banner data on the task page")
        check(detail["review_diff_url"].endswith(
            "demo.git/compare/main...task/3-feature-tour"), "review diff link constructed")
        dash = slot_public(auth, "/", "dashboard")
        check(any(int(r["task_id"]) == 3 for r in dash["reviews"]),
              "dashboard lists the pending review")
        review_block = [b for b in detail["blocks"]
                        if b["block_type"] == "review" and not b.get("answered")]
        check(len(review_block) == 1, "review raised as an answerable block")
        act(auth, review_block[0]["answer_action"], {"answer": "Human approved the review"})
        check(any(b.get("answer") == "Human approved the review"
                  for b in task_state(auth, 3)["blocks"] if b["block_type"] == "review"),
              "review approval recorded")

        # single formatting end to end: the daemon's ANSWER entry repeats the
        # answer the way the agent read it
        check(wait_task_status(auth, 3, "working", timeout=90) is not None,
              "task 3 resumes working after the review approval")
        db = sqlite3.connect(DB_PATH)
        answer_rows = db.execute(
            "select sc.command_text from session_commands sc join sessions s on s.id = sc.session_id"
            " where s.task_id = 3 and sc.type = 'answer' order by sc.id").fetchall()
        db.close()
        check(any("Human selected: Postgres" == row[0] for row in answer_rows),
              "daemon formats the option answer exactly once")

        # quoted user comment while working (terminal highlight flow)
        detail = task_state(auth, 3)
        act(auth, detail["comment"], {
            "body": "The sleep is fine, but flush sooner next time",
            "quote": "sleep 12",
            "return_to": "/tasks/3/terminal",
        })
        check(wait_task_status(auth, 3, "done", timeout=90) is not None,
              "user comment delivered via /notify; task 3 completes")
        detail = task_state(auth, 3)
        check("Part 2 feature tour complete" in (detail.get("summary") or ""),
              "task 3 summary reflects the post-comment completion")
        check(any(c["author"] == "user:admin" and c.get("quote") == "sleep 12"
                  for c in detail.get("comments", [])),
              "quoted user comment stored on the task")
        check(any(c["author"] == "bot" and "Got the human comment" in c["body"]
                  for c in detail.get("comments", [])),
              "agent acknowledged the comment in-conversation")

        db = sqlite3.connect(DB_PATH)
        tour_types = {row[0] for row in db.execute(
            "select sc.type from session_commands sc join sessions s on s.id = sc.session_id"
            " where s.task_id = 3")}
        notified = db.execute(
            "select sc.output_text from session_commands sc join sessions s on s.id = sc.session_id"
            " where s.task_id = 3 and sc.type = 'user-comment'").fetchall()
        db.close()
        check({"bash", "comment", "block", "answer", "plan", "progress", "review",
               "query", "user-comment"} <= tour_types,
              "terminal log carries every part 2 entry type (%s)" % sorted(tour_types))
        check(any("sleep 12" in (row[0] or "") for row in notified),
              "/notify delivered the comment with its quoted terminal text")

        print("== audit trail ==")
        db = sqlite3.connect(DB_PATH)
        rows = db.execute(
            "select command_text, output_text, exit_code from session_commands order by id").fetchall()
        commands = {row[0]: row for row in rows}
        check(any("ls -la" in c and "hello.py" in (r[1] or "") for c, r in commands.items()),
              "ls output captured in session_commands")
        check(any("git" in c and "Initial commit" in (r[1] or "") and r[2] == 0
                  for c, r in commands.items()),
              "git log output captured with exit code 0")
        per_session = db.execute(
            "select session_id, count(*) from session_commands group by session_id order by session_id"
        ).fetchall()
        check(all(count > 0 for _, count in per_session) and len(per_session) == 3,
              "each session carries its own commands (%s)" % per_session)
        progress_rows = db.execute(
            "select summary from task_progress where task_id = 3 order by id").fetchall()
        check([row[0] for row in progress_rows] == ["queried the host", "human input integrated",
                                                    "human input integrated"],
              "task_progress keeps the full history in order")
        db.close()

        print("== SSE ==")
        db = sqlite3.connect(DB_PATH)
        tour_session = db.execute(
            "select s.id from sessions s where s.task_id = 3 order by s.id desc limit 1"
        ).fetchone()[0]
        db.close()
        request = urllib.request.Request(WEB + "/terminal-stream/%d" % tour_session)
        with urllib.request.urlopen(request, timeout=15) as stream:
            check(stream.headers.get("Content-Type", "").startswith("text/event-stream"),
                  "terminal stream is text/event-stream")
            entries = []
            saw_piped_input = False
            deadline = time.monotonic() + 10
            while time.monotonic() < deadline and len(entries) < 120:
                if any(e.get("type") == "done" for e in entries):
                    break
                line = stream.readline().decode().strip()
                if not line.startswith("data: ") or line == "data: ping":
                    continue
                entry = json.loads(line[6:])
                entries.append(entry)
                if entry.get("type") == "query" and (entry.get("input") or "").strip():
                    saw_piped_input = True
            check(len(entries) >= 12, "terminal stream replays the backlog (%d entries)" % len(entries))
            check(saw_piped_input, "terminal stream carries the query's piped input")
            check(any(e.get("type") == "review" for e in entries)
                  and any(e.get("type") == "user-comment" for e in entries),
                  "terminal stream carries typed part 2 events")
            db = sqlite3.connect(DB_PATH)
            db.execute("insert into session_commands (session_id, command_text, output_text,"
                       " type, started_at) values (?, 'sse-live-check', 'ok', 'command',"
                       " strftime('%s','now'))", (tour_session,))
            db.commit()
            db.close()
            got_live = False
            deadline = time.monotonic() + 6
            while time.monotonic() < deadline and not got_live:
                line = stream.readline().decode().strip()
                if line.startswith("data: ") and "sse-live-check" in line:
                    got_live = True
            check(got_live, "terminal stream pushes new entries within seconds")

        pulse = directives("/tasks/3", auth)
        check(any(d.get("type") == "subscribe" for d in pulse),
              "task page subscribes to the pulse channel for live refresh")

        print("== part 2 API surface ==")
        db = sqlite3.connect(DB_PATH)

        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": "Part 2 API surface",
            "description": "Typed blocks, plan, progress, review, quoted comments, SSE",
            "plan": "# Initial plan",
            "priority": "6",
            "task_branch": "task/9-api-surface",
        })
        api_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                        if t["title"] == "Part 2 API surface"][0])
        detail = task_state(auth, api_task)
        check(detail["branch_from"] == "main",
              "branch_from prefilled from the project default branch")
        check(detail["task_branch"] == "task/9-api-surface", "task_branch stored")

        # Dispatch bookkeeping without a daemon: the ingest API is under
        # test, so the session and saddle/task state are staged directly.
        db.execute("insert into sessions (task_id, saddle_id, started_at)"
                   " values (?, ?, strftime('%s','now'))", (api_task, api_saddle))
        db.execute("update saddles set current_task_id = ?, status = 'idle' where id = ?",
                   (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]

        status, _ = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, None,
                               {"type": "comment", "session_id": api_session, "data": {"body": "x"}})
        check(status == 401, "update without a token is rejected")
        status, _ = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, "saddle-not-a-token",
                               {"type": "comment", "session_id": api_session, "data": {"body": "x"}})
        check(status == 401, "update with an invalid token is rejected")
        status, _ = saddle_api("GET", "/api/saddle/%d/poll" % api_saddle, None)
        check(status == 401, "poll without a token is rejected")

        status, body = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, api_token,
                                  {"type": "comment", "session_id": api_session,
                                   "data": {"body": "Surveyed the module"}})
        check(status == 200 and body.get("ok"), "comment update accepted with token")
        status, body = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, api_token,
                                  {"type": "block", "session_id": api_session, "data": {
                                      "question": "Which database?",
                                      "type": "question",
                                      "options": ["SQLite", "Postgres"],
                                      "body": "# Pick one"}})
        check(status == 200 and body.get("block_id"), "typed block returns block_id")
        status, _ = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, api_token,
                               {"type": "output", "session_id": api_session, "data": {
                                   "command": "cat notes.md | saddle query 'meaning?'",
                                   "output": "42",
                                   "type": "query",
                                   "input": "# notes\nmeaning of life"}})
        check(status == 200, "query output update accepted")
        status, _ = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, api_token,
                               {"type": "plan", "session_id": api_session,
                                "data": {"plan": "# Plan\n- migrate schemas"}})
        check(status == 200, "plan update accepted")
        status, _ = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, api_token,
                               {"type": "progress", "session_id": api_session,
                                "data": {"summary": "Migrated the schemas"}})
        check(status == 200, "progress update accepted")
        status, _ = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, api_token,
                               {"type": "review", "session_id": api_session,
                                "data": {"summary": "Migration **ready** for review"}})
        check(status == 200, "review update accepted")

        detail = task_state(auth, api_task)
        check(detail["status"] == "review_requested", "review flips the task status")
        check([b["options"] for b in detail["blocks"]
               if b["block_type"] == "question"] == [["SQLite", "Postgres"]],
              "question block carries its options")
        check("- migrate schemas" in detail["plan"], "plan visible on the task page")
        check(detail["progress"][0]["summary"] == "Migrated the schemas",
              "progress history newest-first")
        check(detail["review_pending"] and "ready" in detail["review_summary"],
              "review banner data on the task page")
        check(detail["review_diff_url"].endswith(
            "demo.git/compare/main...task/9-api-surface"), "review diff link constructed")

        dash = slot_public(auth, "/", "dashboard")
        check(any(int(r["task_id"]) == api_task for r in dash["reviews"]),
              "dashboard lists the pending review")
        check(any(b["question"] == "Which database?" for b in dash["open_blocks"]),
              "dashboard lists blocks needing answers")

        piped = db.execute("select piped_input from session_commands where session_id = ?"
                           " and type = 'query'", (api_session,)).fetchall()
        check(piped and "meaning of life" in (piped[0][0] or ""),
              "push API stores the query's piped input too")

        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"] == ["Human selected: SQLite"],
              "option answer formatted for the agent")
        act(auth, detail["comment"],
            {"body": "Check the migration output", "quote": "ERROR: table demos exists"})
        status, body = saddle_api("GET", "/api/saddle/%d/poll" % api_saddle, api_token)
        check(any(a["answer"] == "Human selected: SQLite" for a in body.get("answers", [])),
              "poll hands over the block answer")
        check(any(c["body"] == "Check the migration output" and "ERROR" in c["quote"]
                  for c in body.get("comments", [])),
              "poll hands over the quoted user comment")
        status, body = saddle_api("GET", "/api/saddle/%d/poll" % api_saddle, api_token)
        check(body.get("answers") == [] and body.get("comments") == [],
              "poll delivers answers and comments exactly once")

        entry_types = {row[0] for row in db.execute(
            "select type from session_commands where session_id = ?", (api_session,))}
        check({"comment", "block", "plan", "progress", "review", "user-comment"} <= entry_types,
              "conversation events logged as typed terminal entries")

        act(auth, detail["cancel"], {"task_id": str(api_task)})
        db.close()

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

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

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


def query_commands(task_id):
    db = sqlite3.connect(DB_PATH)
    try:
        return db.execute(
            "select sc.type from session_commands sc join sessions s on s.id = sc.session_id"
            " where s.task_id = ? and sc.type = 'query'", (task_id,)).fetchall()
    finally:
        db.close()


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