#!/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 consolidated goal-era arc (plan 15, re-baselined for C+D):
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 commentary-only round and a trailing-comment round feed VERBATIM
   (comment-policy-free inputs, plan 15 D1) before 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 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's DEGENERATE first response is
   validator-caught (plan 15 D2 — never executed, one receipt, the
   session continues); the clean resend runs in its ephemeral
   worktree branch cut from the goal clone, commits, completes —
   worktree auto-pruned, branch kept
5. the terminal transition (the goals row's last_terminal_watermark,
   plan 15 C3 — goal_events is gone) 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 — the live per-repo push flags gate it
   (plan 15 C4): the accept REFUSES while origin is behind, passes
   once the branch is re-pushed and the probe re-cached -> 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
   harness-authored bookkeeping riding the RAW terminal only), the
   /status goal_branches cache sampled by the manual saddle refresh
8. the push wire surface (401s on the id-less routes, typed entries
   and block raises riding POST /api/saddle/results with per-entry
   exactly-once dedup, 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, 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_user_turns():
    """Every user turn the mock received, oldest first (the mock's
    --capture file: one JSON line per request). The D-side receipts
    (round outputs) ride these turns back to the model."""
    turns = []
    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") == "user":
                        turns.append(message.get("content") or "")
    return turns


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): commentary-ONLY (plan 15 D1) — fed
        # verbatim, receipt [0 command lines] plus the factual
        # no-commands line, conversation continues
        "# Reading the brief before touching anything\n"
        "# the demo repo is tiny; the greeting belongs in greeting.txt\n"
        "# nothing worth running yet — probing comes next",
        # R2: investigate with TRAILING comments after the commands
        # (comments-after-commands feeds exactly like leading ones),
        # then block on a scope question
        "# Investigating before planning\nls -la\ngit log --oneline | head -3\n"
        "# two commits and one file — the blast radius is greeting.txt alone\n"
        + (heredoc % 'saddle block question "Demo only" "Whole workspace"') % (
            "Which scope should the greeting cover? The demo repo is small; confirm the blast radius."),
        # W1 (the worker's first round): a DEGENERATE repetition (plan
        # 15 D2) — the validator catches it before feeding; nothing
        # executes, one rejection receipt, the round is consumed
        "# Starting the mandate\n" + "echo probing-the-workspace\n" * 12,
        # W2 (the clean resend): 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 plan-feedback verdict -> the revision (the export step
        # incorporates the harness's feedback)
        "--on-plan-feedback", (heredoc % "saddle plan") % (
            "# Plan v2\n\n1. Add greeting.txt to the demo repo.\n2. Export it for reuse."),
        # 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,
        # 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 D-side receipts 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]

    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 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 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")
        # 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 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)")
        # 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.
        check(wait_for(lambda: bool(db_query(
            "select 1 from sessions where task_id = ? and ended_at is null",
            (worker_task,))[0:1]) and (lambda card: card is not None
                                       and card["status"] == "busy"
                                       and int(card["current_task_id"] or 0) == worker_task)(
            next((s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
                  if s["name"] == SADDLE_NAME), None)) 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 open session (no cached row)")

        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(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("== accept refuses while origin is behind (the live push gate) ==")
        # Rewind origin's goal branch to the pre-merge base and refresh
        # the daemon probe: the live per-repo flags flip to unpushed and
        # the completion accept must REFUSE (plan 15 C4 / F9) until the
        # branch is pushed again — enforcing push-after-each-merge at
        # the human gate, read from probe state (never a snapshot).
        completion = state["completion_pending_block"]
        subprocess.run(["git", "-C", goal_clone, "push", "-f", "origin",
                        "main:refs/heads/goal/demo-run"], check=True, capture_output=True)
        subprocess.run(["git", "-C", goal_clone, "fetch", "origin"],
                       check=True, capture_output=True)
        act(auth, card["refresh"], {})
        check(wait_for(lambda: any(
            "demo" in (b.get("branch") or "") and b.get("pushed") is False
            for b in goal_state(auth, goal1).get("push_branches", [])) or False,
            timeout=60, what="unpushed chip") is not None,
              "the rewound origin flips the live push chip to unpushed")
        refusal = act(auth, completion["answer_action"], {"verdict": "accept"})
        check(any(d.get("type") == "notify" and d.get("kind") == "error"
                  and "Origin is behind" in (d.get("message") or "")
                  for d in refusal),
              "the accept refuses with a clear message while origin is behind")
        check(goal_state(auth, goal1)["status"] == "completion_review",
              "the refused accept leaves the goal in completion review")
        check(goal_state(auth, goal1)["completion_pending_block"]
              not in (None, False),
              "the refused accept leaves the completion block unanswered")
        subprocess.run(["git", "-C", goal_clone, "push", "origin", "goal/demo-run"],
                       check=True, capture_output=True)
        act(auth, card["refresh"], {})
        check(wait_for(lambda: any(
            "demo" in (b.get("branch") or "") and b.get("pushed") is True
            for b in goal_state(auth, goal1).get("push_branches", [])) or False,
            timeout=60, what="pushed chip restored") is not None,
              "re-pushing the branch restores the live push chip (after push)")
        completion = goal_state(auth, goal1)["completion_pending_block"]

        print("== completion accept ==")
        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 once the push flags read pushed")
        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) ==")
        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: raw log only)")
        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("== 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 rides the RAW terminal (C2 + U2) ==")
        # Plan 16 S5 (U2): the mechanical rows (restores, park
        # notices) author NO comment rows — they persist as typed
        # 'system' session_commands (the raw terminal log) and never
        # render on a user-facing timeline.
        raw_system = db_query(
            "select sc.command_text, sc.output_text from session_commands sc"
            " join sessions s on s.id = sc.session_id where s.task_id = ?"
            " and sc.type = 'system'", (run_task,))
        check(any("session context restored" in (row[1] or "") for row in raw_system)
              and any("parked until an owned task" in (row[1] or "") for row in raw_system),
              "the harness bookkeeping (restore, park) persists in the raw terminal log")
        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: commentary + degenerate ==")
        run_bash = db_query(
            "select sc.command_text from session_commands sc join sessions s"
            " on s.id = sc.session_id where s.task_id = ? and sc.type = 'bash'",
            (run_task,))
        check(any("nothing worth running yet" in (row[0] or "")
                  and "Reading the brief" in (row[0] or "") for row in run_bash),
              "the narration-only response was fed verbatim (comments are style, not policy)")
        check(any("git log --oneline" in (row[0] or "")
                  and "blast radius is greeting.txt alone" in (row[0] or "")
                  for row in run_bash),
              "trailing comments after commands rode the same round (no rejection)")
        rejected = db_query(
            "select sc.command_text, sc.output_text from session_commands sc"
            " join sessions s on s.id = sc.session_id"
            " where sc.type = 'rejected' and s.task_id in (?, ?)",
            (run_task, worker_task))
        check(len(rejected) == 1 and rejected[0][0] == "degenerate response"
              and "degenerate signature" in (rejected[0][1] or "")
              and "echo probing-the-workspace" in (rejected[0][1] or "")
              and "Resend the intended action only" in (rejected[0][1] or ""),
              "exactly ONE rejected entry — the degenerate repetition, caught with its receipt")
        worker_bash = db_query(
            "select sc.command_text from session_commands sc join sessions s"
            " on s.id = sc.session_id where s.task_id = ? and sc.type = 'bash'",
            (worker_task,))
        check(not any("probing-the-workspace" in (row[0] or "") for row in worker_bash),
              "the degenerate script never executed (validator-caught before feeding)")
        turns = capture_user_turns()
        check(any(turn.startswith("[0 command lines executed")
                  and "no commands ran — narrate in the same response as your next command" in turn
                  for turn in turns),
              "the narration round's receipt reached the model (no-commands line)")
        check(any("degenerate signature" in turn and "Resend the intended action only" in turn
                  for turn in turns),
              "the degenerate rejection receipt reached the model as its next turn")

        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("== push wire surface (the poll-era endpoints deleted) ==")
        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 entries, 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(entry_id, entry_type, state="working", command="", output="",
                 block=None, **extra):
            """One round-results push in the daemon's shape (the
            push-era wire the real daemon speaks — the poll-era
            /update channel is gone, 404-pinned below)."""
            payload = {"session_id": api_session, "state": state}
            if entry_type is not None:
                entry = {"id": entry_id, "command": command, "output": output,
                         "exit_code": 0, "type": entry_type,
                         "timestamp": "2026-09-13T00:00:00Z"}
                entry.update(extra)
                payload["entries"] = [entry]
            if block is not None:
                payload["block"] = block
            return saddle_api("POST", "/api/saddle/results", api_token, payload)

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

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

        status, body = push(9001, "comment", command="saddle comment",
                            output="Surveyed the module")
        check(status == 200 and body.get("ok"), "comment entry accepted on the results push")
        check(any(c.get("from_agent") and c["body"] == "Surveyed the module"
                  for c in task_state(auth, api_task)["comments"]),
              "the comment entry authors an agent row (output-first)")
        status, body = push(None, None, 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(9002, "query", command="cat notes.md | saddle query 'meaning?'",
                            output="42", input="# notes\nmeaning of life")
        check(status == 200, "query entry accepted on the results push")
        query_rows = db_query(
            "select sc.type, sc.piped_input from session_commands sc"
            " where sc.session_id = ? and sc.command_text like '%saddle query%'",
            (api_session,))
        check(query_rows and query_rows[0][0] == "query"
              and query_rows[0][1] == "# notes\nmeaning of life",
              "the query entry persists typed with its piped input")
        status, body = push(9003, "progress", command="saddle progress-update",
                            output="Migrated the schemas")
        check(status == 200, "progress entry accepted on the results push")
        status, body = push(9003, "progress", command="saddle progress-update",
                            output="Migrated the schemas AGAIN")
        check(status == 200, "the redelivered entry id is consumed silently (200, no projection)")
        check(len([p for p in task_state(auth, api_task)["progress"]
                   if p["summary"] in ("Migrated the schemas", "Migrated the schemas AGAIN")])
              == 1 and task_state(auth, api_task)["progress"][0]["summary"] == "Migrated the schemas",
              "entries project exactly once per daemon-global id (push-era exactly-once)")
        status, body = push(9004, "review", command="saddle review",
                            output="Migration **ready** for review")
        check(status == 200, "review entry accepted on the results push")

        detail = task_state(auth, api_task)
        check(detail["status"] == "review_requested", "the review entry 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(None, None, 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())