#!/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
script — both binaries, the true cross-binary contract proof (the
per-repo fake-peer suits live in web_stack_test.py / saddle e2e).

The goal-era arc (plan 12):
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. pre-plan question block -> answered on the goal page -> the ONE
   answer resolves the goal-run task's single task_blocks row
   row, and the parked goal agent resumes with the verdict wording
3. plan submit -> reject-with-feedback -> revision -> approval (the
   agreed verdict words ride the resume)
4. task fan-out through the real /api/saddle/task/create + enqueue
   callbacks; the worker runs in its ephemeral worktree branch cut
   from the goal clone, commits, completes — worktree auto-pruned,
   branch kept
5. the task-terminal event 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) and submits the
   completion report
6. completion review accept -> goal complete, run task done
7. the goal agent's terminal sessions (SSE replay), the audit trail
   (the derived goal narrative, typed session_commands), the /status
   goal_branches cache sampled by the manual saddle refresh
8. the legacy ingest API surface (401s, typed updates, exactly-once
   poll handover) 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, 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: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]
PLAN_APPROVED = "Plan approved. Proceed with implementation."
REVIEW_APPROVED = "Review approved. Continue with the task."

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)
    try:
        return db.execute(sql, params).fetchall()
    finally:
        db.close()


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 mock_argv():
    heredoc = "cat << 'SADDLE_EOF' | %s\n%%s\nSADDLE_EOF"
    fanout = (
        "# Fanning the plan out\n"
        "out=$(cat << 'SADDLE_EOF' | saddle task create --title 'Write the greeting' --mode implement\n"
        "Create greeting.txt containing hello-from-the-goal.\nSADDLE_EOF)\n"
        "tid=$(echo \"$out\" | grep -oE '[0-9]+' | head -1)\n"
        "saddle task enqueue \"$tid\"\n"
        "saddle suspend"
    )
    responses = [
        # R1 (fresh goal agent): investigate, then block on a scope question
        "# Investigating before planning\nls -la\ngit log --oneline | head -3\n"
        + (heredoc % 'saddle block question "Demo only" "Whole workspace"') % (
            "Which scope should the greeting cover? The demo repo is small; confirm the blast radius."),
        # R2 (plan revision after feedback): the rejected plan, re-submitted
        (heredoc % "saddle plan") % (
            "# Plan v2\n\n1. Add greeting.txt to the demo repo.\n2. Export it for reuse."),
        # W1 (the worker task): verify the ephemeral branch, commit, done
        "# Working the mandate\ngit branch --show-current\n"
        "echo hello-from-the-goal > greeting.txt\ngit add -A\n"
        "git commit -m 'greeting from the goal'\n"
        + (heredoc % "saddle done") % (
            "# Completion report\n\n- changed: greeting.txt\n"
            "- verified: committed on the ephemeral branch"),
    ]
    argv = []
    for response in responses:
        argv += ["-r", response]
    argv += [
        # the question's verdict ("Human selected: Demo only") -> plan v1
        "--on-answer", (heredoc % "saddle plan") % (
            "# Plan v1\n\n1. Add greeting.txt to the demo repo."),
        # the approval verdict -> task fan-out + suspend
        "--on-plan-approved", fanout,
        # the batched "Goal update" resume -> merge, prune, push, complete
        "--answer-contains", "Goal update",
        "--on-answer-seen",
        "# Merging the finished task, then completing (the sleep holds the\n"
        "# working state open so /status can sample the branch state)\n"
        "sleep 4\n"
        "git merge --no-edit $(git for-each-ref --format='%(refname:short)' refs/heads/task/*)\n"
        "git branch -D $(git for-each-ref --format='%(refname:short)' refs/heads/task/*)\n"
        "git push -u origin $(git branch --show-current)\n"
        + (heredoc % "saddle done") % (
            "# Goal completion report\n\n- achieved: greeting.txt added via one worker task\n"
            "- verified: the goal branch is pushed to origin"),
        "--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():
    prepare_workspace()

    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 question title derives from the body's first line")
        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")
        act(auth, pending["answer_action"], {"answer": "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"], {"answer": PLAN_APPROVED})
        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 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 (suspended_at stamped)")

        print("== the worker's ephemeral worktree ==")
        branch_rows = wait_for(lambda: db_query(
            "select sc.output_text from session_commands sc join sessions s"
            " on s.id = sc.session_id where s.task_id = ?"
            " and sc.command_text like '%git branch --show-current%'", (worker_task,)) 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" % ephemeral)
        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
        check(kept, "the ephemeral branch survives in the goal clone for the merge")

        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(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 "")
                                   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")
        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)
        check("greeting.txt added via one worker task" in state.get("completion_report", ""),
              "the completion report renders on the goal page")
        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 ==")
        completion = state["completion_pending_block"]
        check(completion and completion.get("body", "") == state.get("completion_report", ""),
              "the run task's completion block carries the report")
        act(auth, completion["answer_action"], {"answer": REVIEW_APPROVED})
        check(wait_goal_status(auth, goal1, "complete", timeout=30) is not None,
              "accepting with the agreed wording completes the goal")
        check(wait_task_status(auth, run_task, "done", timeout=30) is not None,
              "the goal-run task closes done with the goal")

        print("== audit trail (the derived narrative) ==")
        narrative = goal_state(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(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)")
        run_types = {row[0] for row in db_query(
            "select distinct sc.type from session_commands sc join sessions s"
            " on s.id = sc.session_id where s.task_id = ?", (run_task,))}
        check({"bash", "block", "plan"} <= run_types,
              "the goal agent's terminal carries typed entries (%s)" % sorted(run_types))
        worker_types = {row[0] for row in db_query(
            "select distinct sc.type from session_commands sc join sessions s"
            " on s.id = sc.session_id where s.task_id = ?", (worker_task,))}
        check({"bash", "done"} <= worker_types,
              "the worker's terminal carries typed entries (%s)" % sorted(worker_types))

        print("== SSE ==")
        run_session = db_query(
            "select sc.session_id from session_commands sc join sessions s on s.id = sc.session_id"
            " where s.task_id = ? and sc.type = 'plan' order by sc.id limit 1",
            (run_task,))[0][0]
        state = goal_state(auth, goal1)
        check(any(r.get("terminal_url", "").endswith("/terminal?session_id=%d" % run_session)
                  for r in state.get("runs", [])),
              "the goal page's run list deep-links each session terminal")
        request = urllib.request.Request(WEB + "/terminal-stream/%d" % run_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 = []
            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
                entries.append(json.loads(line[6:]))
            check(len(entries) >= 4, "terminal stream replays the run's backlog (%d entries)" % len(entries))
            # push-era sessions stream their raw PTY log (the typed
            # commands live in session_commands) — the plan submission
            # shows as its echoed heredoc in the raw replay
            check(any("saddle plan" in json.dumps(e) for e in entries),
                  "terminal stream replay carries the plan submission round")

        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("== legacy ingest API surface ==")
        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": "Typed updates, exactly-once poll",
            "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 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]
        db.close()

        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, 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": "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")
        status, body = 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 == 400 and "gone" in (body.get("error") or ""),
              "the task-plan update type fails loudly (plan 12 removal)")
        status, body = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, api_token,
                                  {"type": "subtask", "session_id": api_session, "data": {}})
        check(status == 400 and "gone" in (body.get("error") or ""),
              "the subtask update type fails loudly (plan 12 removal)")

        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(detail["progress"][0]["summary"] == "Migrated the schemas",
              "progress history newest-first")

        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 resume notice formats it)")
        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"] == "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")
        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 freed and idle 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())