clanker-saddle/test/bundles/pages.py

"""The pages bundle (moved verbatim from the pre-split web_stack_test.py;
see test/harness.py for the shared machinery and the runner)."""
from harness import *  # noqa: F401,F403 — the shared wire helpers


# =============================================================================
# Bundle: pages — block presentation, diff URLs, computed display state,
# terminal rendering (rejected entries, ANSI SGR) and task-page layout.
# Fixture-driven: tasks and sessions are inserted as rows; blocks are
# raised through the push-era wire (POST /api/saddle/results with a
# blocked state) or inserted directly for terminal-task fixtures; only
# the rejected-ride scenario dispatches.
# =============================================================================

def run(ctx):
    auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token
    app = ctx.app

    def raise_block(session_id, question, kind="question", options=None, body=None):
        """The push-era block raise: a blocked round-results push
        carrying the block payload (the poll-era /update channel is
        gone — 404 below pins that)."""
        block = {"question": question, "type": kind}
        if options is not None:
            block["options"] = options
        if body is not None:
            block["body"] = body
        return saddle_api("POST", "/api/saddle/results", fake_token,
                          {"session_id": session_id, "state": "blocked",
                           "block": block})

    check(True, "login yields an auth slot")

    print("== the poll-era wire is gone (arch 19) ==")
    # The daemon never called update/poll (verified against saddle/src);
    # the routes 404 for any caller, authenticated or not.
    status, _ = saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
                           {"type": "status", "session_id": 1,
                            "data": {"state": "working"}})
    check(status == 404, "POST /api/saddle/{id}/update is gone (404)")
    status, _ = saddle_api("GET", "/api/saddle/%d/poll" % fake_id, fake_token)
    check(status == 404, "GET /api/saddle/{id}/poll is gone (404)")

    print("== blocks: push-era raises + body presentation ==")
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority,"
                 " created_at)"
                 " values (1, 'Block display task', 'working', 5,"
                 " strftime('%s','now'))")
    plan_task = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.execute("insert into sessions (task_id, saddle_id, started_at, ended_at, outcome)"
                 " values (?, ?, strftime('%s','now'), strftime('%s','now'), 'done')",
                 (plan_task, fake_id))
    conn.commit()
    session_id = conn.execute("select max(id) from sessions").fetchone()[0]
    conn.close()

    status, _ = raise_block(session_id, "Which database?", kind="question",
                            options=["SQLite", "Postgres"], body="# Storage choice")
    check(status == 200, "question block raised through the results push")

    print("== B1: highlight-to-comment quote dialog ==")
    status, page = http("GET", "/tasks/%d" % plan_task, timeout=15)
    check(status == 200 and "quote-comment-dialog" in page and "quote-text" in page,
          "task page carries the quote-comment dialog the popover opens")
    check('name="quote"' in page and 'name="return_to"' in page,
          "quote dialog posts the quote and return-to fields")
    status, page = http("GET", "/tasks/%d/terminal" % plan_task, timeout=15)
    check(status == 200 and "quote-comment-dialog" in page,
          "terminal page keeps its quote dialog")

    print("== B3: review section hides the duplicate body ==")
    status, _ = raise_block(session_id, "Review the diff", kind="review",
                            body="# The change\n\nReworks the flange module.")
    check(status == 200, "review block raised through the results push")
    detail = task_state(auth, plan_task)
    review_block = [b for b in detail["blocks"] if b["block_type"] == "review"][0]
    check(detail.get("review_pending") is True,
          "review section shows while the review is unanswered")
    check(review_block.get("show_body") is False,
          "pending review block hides its duplicate body preview")
    check(review_block.get("body", "").startswith("# The change"),
          "review body itself still arrives in state (B2 counterpart)")
    act(auth, review_block["answer_action"], {"verdict": "accept"})
    detail = task_state(auth, plan_task)
    review_block = [b for b in detail["blocks"] if b["block_type"] == "review"][0]
    check(review_block.get("show_body") is True,
          "answered review shows the body again (banner gone)")
    check(review_block.get("answer") == "Review approved. Continue with the task.",
          "the review accept form's verdict marker resolves to the stable wording")
    # Plan 16 S2/R4: no verdict constant ships in any template — the
    # accept forms carry only the verdict marker.
    task_page_html = http_get_page(auth, "/tasks/%d" % plan_task)
    check("Plan approved. Proceed with implementation." not in task_page_html
          and "Review approved. Continue with the task." not in task_page_html,
          "the task page ships no verdict constants (server-owned wording)")
    check('name="verdict" value="accept"' in task_page_html,
          "the accept forms post the verdict marker")
    # Plan 16 S5 (U1): verdict-shaped blocks keep their explicit
    # buttons + free-text path; a question with NULL options renders
    # the question form (radio "other" + free reply), never a
    # free-text-only verdict.
    check("Request changes" in task_page_html and 'name="feedback"' in task_page_html,
          "the review verdict keeps Request changes with free-text feedback (U1)")
    check('value="__other__"' in task_page_html and 'name="other"' in task_page_html,
          "question blocks render the option/other question form (U1)")
    status, _ = raise_block(session_id, "Free-text thought?", kind="question")
    check(status == 200, "a NULL-options question block raises")
    bare_question = [b for b in task_state(auth, plan_task)["blocks"]
                     if b["question"] == "Free-text thought?"][0]
    check(bare_question.get("options") == []
          and bare_question.get("block_type") == "question",
          "a NULL-options question stays a question (no options, not a verdict)")

    print("== B2: question block body above options ==")
    # Contract side (A2 will restore bodies end-to-end): a question
    # block raised with a heredoc body carries it; the body preview
    # rides above the options (template order) and survives answering
    # alongside the selected answer. (Blocks list newest-first; the
    # U1 fixture above raised a second, options-less question.)
    detail = task_state(auth, plan_task)
    question = [b for b in detail["blocks"]
                if b["block_type"] == "question"
                and b["body"] == "# Storage choice"][0]
    check(question.get("body") == "# Storage choice"
          and question.get("show_body") is True,
          "question block keeps its body markdown with the preview on")
    act(auth, question["answer_action"], {"option": "SQLite"})
    detail = task_state(auth, plan_task)
    question = [b for b in detail["blocks"]
                if b["block_type"] == "question"
                and b["body"] == "# Storage choice"][0]
    check(question.get("answered") is True and question.get("answer") == "SQLite"
          and question.get("body") == "# Storage choice"
          and question.get("show_body") is True,
          "answered question shows body + selected answer")

    print("== B5: compare-changes SSH -> HTTPS ==")
    def diff_url_for_project(git_url, suffix):
        state = slot_public(auth, "/projects", "projects")
        act(auth, state["create"], {"name": "diff-%s" % suffix, "git_url": git_url})
        project_id = int([p["id"] for p in slot_public(auth, "/projects", "projects")["projects"]
                          if p["name"] == "diff-%s" % suffix][0])
        conn = db()
        conn.execute("insert into tasks (project_id, title, status, task_branch,"
                     " branch_from, created_at) values (?, ?, 'working', 'feature/x',"
                     " 'main', strftime('%s','now'))", (project_id, "Diff %s" % suffix))
        conn.commit()
        task_id = conn.execute("select max(id) from tasks").fetchone()[0]
        conn.execute("insert into sessions (task_id, saddle_id, started_at, ended_at,"
                     " outcome) values (?, ?, strftime('%s','now'), strftime('%s','now'),"
                     " 'done')", (task_id, fake_id))
        conn.commit()
        session = conn.execute("select max(id) from sessions").fetchone()[0]
        conn.close()
        status, _ = raise_block(session, "Review the diff", kind="review")
        return task_state(auth, task_id).get("review_diff_url") or ""

    check(diff_url_for_project("ssh://fabrica.unitatem.net:222/clanker/clanker-saddle.git",
                               "fabrica")
          == "https://fabrica.unitatem.net/clanker/clanker-saddle/compare/main...feature/x",
          "ssh://git@host:port/owner/repo.git translates (port dropped)")
    check(diff_url_for_project("git@github.com:acme/widgets.git", "scp")
          == "https://github.com/acme/widgets/compare/main...feature/x",
          "git@host:owner/repo.git translates to the browse host")
    check(diff_url_for_project("https://example.com/plain/repo", "plain")
          == "https://example.com/plain/repo/compare/main...feature/x",
          "HTTPS URLs pass through unchanged")

    print("== B6: computed display state ==")
    # Draft tasks display as draft (plan 11) and never dispatch.
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1",
        "title": "Draft display task",
        "priority": "5",
    })
    draft_display = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                         if t["title"] == "Draft display task"][0])
    detail = task_state(auth, draft_display)
    check(detail["status"] == "draft" and detail.get("display") == "draft",
          "new tasks default to draft and display it")
    check(detail.get("draft") is True and detail.get("enqueue") is not None
          and detail.get("delete") is not None,
          "draft tasks expose the enqueue + delete actions")
    check(quiet_window(quiet_seconds(),
                       lambda: task_state(auth, draft_display)["status"] == "draft",
                       "draft task never dispatches"),
          "a draft task is never dispatched or woken for")

    # Plan 12: the waiting-on-drafts block is deleted (goal tasks are
    # the goal agent's to sequence); a parked task displays plainly.
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
                 " values (1, 'Parked plain task', 'suspended', 5, strftime('%s','now'))")
    conn.commit()
    conn.close()
    parked_id = db().execute("select max(id) from tasks").fetchone()[0]
    detail = task_state(auth, parked_id)
    check(detail.get("display") == "parked",
          "a suspended task displays as parked (%s)" % detail.get("display"))
    dash = slot_public(auth, "/", "dashboard")
    check(not dash.get("waiting_drafts"),
          "the waiting-on-drafts block is gone from the dashboard data")
    status, home_page = http("GET", "/", timeout=15)
    check(status == 200 and "Waiting on subtasks to be enqueued" not in home_page,
          "the dashboard renders no waiting-on-drafts section")

    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
                 " values (1, 'Parent', 'done', 5, strftime('%s','now'))")
    parent6 = conn.execute("select max(id) from tasks").fetchone()[0]
    # The post-mortem case: stored done WITH an unanswered review block.
    # (A terminal task consumes result pushes silently — plan 13 B2 —
    # so the row is inserted directly, the goalui deep-link pattern.)
    conn.execute("insert into task_blocks (task_id, question, type, created_at)"
                 " values (?, 'Review the final diff', 'review',"
                 " strftime('%s','now'))", (parent6,))
    conn.commit()
    conn.close()

    detail = task_state(auth, parent6)
    check(detail.get("display") == "review requested" and detail.get("terminal") is True,
          "unanswered review overrides the stored done status (terminal flag keeps stored truth)")
    queue = [t for t in slot_public(auth, "/tasks", "tasks")["tasks"]
             if int(t["id"]) == parent6][0]
    check(queue["display"] == "review requested",
          "queue row shows the computed review-requested display")

    # With the review answered the display falls back to the stored
    # status (the plan-12 shape: no subtask-delegation states).
    conn = db()
    conn.execute("update task_blocks set answered_at = strftime('%s','now'),"
                 " answer = 'Human approved the review' where task_id = ?", (parent6,))
    conn.commit()
    conn.close()
    detail = task_state(auth, parent6)
    check(detail.get("display") == "done" and detail.get("terminal") is True,
          "answered review falls back to the stored done display (%s)"
          % detail.get("display"))

    # Flags track the stored status on failure.
    conn = db()
    conn.execute("update tasks set status = 'failed', completed_at = strftime('%s','now')"
                 " where id = ?", (parent6,))
    conn.commit()
    conn.close()
    detail = task_state(auth, parent6)
    check(detail.get("display") == "failed" and detail.get("failed") is True
          and detail.get("terminal") is True,
          "failed task with answered review shows failed + terminal flags")

    print("== rejected entries (non-fatal) ==")
    # The push wire: a rejected entry rides a round-results push, stores
    # on the session and streams through SSE; the task is untouched.
    status, _ = saddle_api("POST", "/api/saddle/results", fake_token, {
        "session_id": session_id, "state": "working",
        "entries": [{
            "id": 9021, "command": "bash -n script.sh",
            "output": "syntax error near line 3 (recoverable)",
            "exit_code": 1, "type": "rejected",
            "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        }]})
    check(status == 200, "a rejected entry rides the results push")
    request = urllib.request.Request(WEB + "/terminal-stream/%d" % session_id)
    with urllib.request.urlopen(request, timeout=15) as stream:
        rejected_event = None
        deadline = time.monotonic() + 8
        while time.monotonic() < deadline and rejected_event is None:
            line = stream.readline().decode().strip()
            if not line.startswith("data: ") or line == "data: ping":
                continue
            event = json.loads(line[6:])
            if event.get("type") == "rejected":
                rejected_event = event
        check(rejected_event is not None
              and "syntax error near line 3" in (rejected_event.get("output") or ""),
              "terminal stream carries rejected entries with their reason")

    # The page template renders them with the distinct non-fatal style.
    status, page = http("GET", "/tasks/%d/terminal" % plan_task, timeout=15)
    check(status == 200 and "term-rejected" in page,
          "terminal page renders rejected entries in their own style")

    print("== B8: ANSI SGR in the terminal view ==")
    # Byte-exact SGR delivery: the daemon's coloured chunks flow through
    # the store + SSE untouched (the client renderer styles them).
    coloured = "\x1b[1;32mbold green\x1b[0m then \x1b[38;5;196m256-red\x1b[0m\n"
    conn = db()
    conn.execute("insert into session_raw_log (session_id, seq, chunk)"
                 " values (?, 900, ?)", (session_id, coloured))
    conn.commit()
    conn.close()
    request = urllib.request.Request(WEB + "/terminal-stream/%d" % session_id)
    with urllib.request.urlopen(request, timeout=15) as stream:
        coloured_event = None
        deadline = time.monotonic() + 8
        while time.monotonic() < deadline and coloured_event is None:
            line = stream.readline().decode().strip()
            if not line.startswith("data: ") or line == "data: ping":
                continue
            event = json.loads(line[6:])
            if event.get("type") == "raw" and "\x1b[1;32m" in (event.get("text") or ""):
                coloured_event = event
        check(coloured_event is not None
              and "\x1b[38;5;196m" in coloured_event["text"],
              "SGR sequences stream byte-exact to the client renderer")
    status, page = http("GET", "/tasks/%d/terminal" % plan_task, timeout=15)
    check(status == 200 and "makeAnsiRenderer" in page and "colour256" in page,
          "terminal page embeds the ANSI renderer (SGR + 256-colour palette)")

    # Dispatch path: a rejected entry in a round's results push flows
    # into the audit trail without failing (or finishing) the task.
    task_id_tmp = create_queued(auth, {
        "project_id": "1",
        "title": "Rejected ride",
        "priority": "2",
    })
    rejected_task = task_id_tmp
    check(wait_for(lambda: task_state(auth, rejected_task).get("status") == "working"
                   or False, timeout=60, what="rejected-ride working") is not None,
          "rejected-ride task reaches working")
    fake.queue_output("rejected", "echo 'too long' | saddle",
                      "max_tokens exceeded (recoverable)")
    check(wait_task_done(auth, rejected_task, timeout=60) is not None,
          "task completes normally past rejected entries")
    conn = db()
    stored = conn.execute(
        "select count(*) from session_commands sc join sessions s on s.id = sc.session_id"
        " where s.task_id = ? and sc.type = 'rejected'", (rejected_task,)).fetchone()[0]
    conn.close()
    check(stored >= 1,
          "rejected entries stored in the audit trail (%d rows)" % stored)
    check(task_state(auth, rejected_task)["status"] == "done"
          and "fake daemon finished" in (task_state(auth, rejected_task).get("summary") or ""),
          "rejected entries never fail the task")

    print("== plan 11 task D: task page — progress first, 50/50 grid, expanders ==")
    detail = task_state(auth, plan_task)
    block_ids = [int(b["id"]) for b in detail.get("blocks", [])]
    check(len(block_ids) >= 3 and block_ids == sorted(block_ids, reverse=True),
          "the blocks section orders newest-first via the sealed GET state (%s)"
          % block_ids)
    status, page = http("GET", "/tasks/%d" % plan_task, timeout=15)
    # Plan 11 task D: the progress feed sits ABOVE the status card;
    # status + description share a 50/50 grid; the plan detail hides
    # behind its expander below it.
    check(status == 200 and "grid-2" in page
          and page.index("<h3>Progress</h3>") < page.index("status-head"),
          "the progress feed renders above the status card")
    check(page.index("grid-2") < page.index("status-head")
          and page.index("status-head") < page.index("<h3>Mandate</h3>"),
          "status and description sit side by side inside the grid (status first)")
    check("<h3>Mandate</h3>" in page
          and "plan-expander" not in page,
          "the task page carries no plan expander (plans live on the goal)")
    # Answered blocks fold behind the history expander (this fixture's
    # three blocks are all answered by now).
    check("blocks-history" in page and "Answered blocks" in page,
          "answered blocks hide behind the blocks history expander")
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority,"
                 " created_at) values (1, 'Progress layout task', 'working', 5,"
                 " strftime('%s','now','-120 seconds'))")
    layout_task = conn.execute("select max(id) from tasks").fetchone()[0]
    # backdated so the dashboard-timeline fixtures below stay the
    # newest events
    conn.execute("insert into task_progress (task_id, summary, created_at)"
                 " values (?, 'first update', strftime('%s','now','-90 seconds'))",
                 (layout_task,))
    conn.execute("insert into task_progress (task_id, summary, created_at)"
                 " values (?, 'second update', strftime('%s','now','-89 seconds'))",
                 (layout_task,))
    conn.commit()
    conn.close()
    status, history_page = http("GET", "/tasks/%d" % layout_task, timeout=15)
    check(status == 200
          and history_page.index("progress-latest") < history_page.index("progress-history"),
          "the progress timeline renders newest-first (latest above the history)")
    # The edit expander sits UNDER the "Show history" expander.
    check(history_page.index("Show history") < history_page.index("edit-expander"),
          "the edit expander sits under the Show history expander")

    print("== heredoc completion reports: markdown card + timeline snippet ==")
    # The daemon's `cat << 'EOF' | saddle done` stores a markdown
    # summary (the failure path stores the same field); the task page
    # renders it as a full-width completion-report card directly below
    # the status/description grid — markdown via stm-markdown
    # (XSS-safe), "Completion report" when done, "Failure summary"
    # when failed, no card when the summary is empty. The dashboard
    # timeline snippets the summary's FIRST line.
    report_md = ("## Flange wired and tested\n\n"
                 "Ran `meson test` over the full matrix.\n\n"
                 "```sh\nmeson test -C builddir\n```\n\n"
                 "<b>not html</b> & escaped\n")
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority,"
                 " summary, started_at, completed_at, created_at)"
                 " values (1, 'Report done task', 'done', 5, ?,"
                 " strftime('%s','now'), strftime('%s','now'),"
                 " strftime('%s','now'))", (report_md,))
    report_done = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.execute("insert into tasks (project_id, title, status, priority,"
                 " summary, started_at, completed_at, created_at)"
                 " values (1, 'Report failed task', 'failed', 5,"
                 " 'Boom: tests failed\n\nsecond line carries the detail',"
                 " strftime('%s','now'), strftime('%s','now'),"
                 " strftime('%s','now'))")
    report_failed = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.execute("insert into tasks (project_id, title, status, priority,"
                 " started_at, completed_at, created_at)"
                 " values (1, 'Report empty task', 'done', 5,"
                 " strftime('%s','now'), strftime('%s','now'),"
                 " strftime('%s','now'))")
    report_empty = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.commit()
    conn.close()

    done_state = task_state(auth, report_done)
    check(done_state["report_title"] == "Completion report"
          and "## Flange wired" in (done_state.get("summary") or ""),
          "a done task's summary rides the report card state")
    check(task_state(auth, report_failed)["report_title"] == "Failure summary",
          "a failed task's card heading reads Failure summary")
    empty_state = task_state(auth, report_empty)
    check(empty_state["report_title"] == "" and not (empty_state.get("summary") or ""),
          "an empty summary renders no report card (no heading)")
    check(task_state(auth, layout_task)["report_title"] == "",
          "non-terminal tasks never show the report card")

    status, page = http("GET", "/tasks/%d" % report_done, timeout=15)
    check(status == 200 and "completion-report" in page
          and page.index("grid-2") < page.index("completion-report")
          < page.index("edit-expander"),
          "the completion-report card sits below the status/description grid, full width")
    card = page[page.index("completion-report"):page.index("edit-expander")]
    check('stm-markdown="task.summary"' in card
          and 'stm-text="task.report_title"' in card,
          "the card renders the summary as markdown under its state-driven heading")
    check("<strong>Summary:</strong>" not in page,
          "the old plain-text summary line is gone (the card owns the summary)")

    # The markdown itself renders client-side: drive the SHIPPED
    # stm-markdown renderer (statum.js) on the fixture summary via
    # node — headings and code come out as HTML, no markdown syntax
    # leaks, raw HTML stays escaped.
    status, statum_js = http("GET", "/_statum/resource/statum.js", timeout=15)
    skip = None if status == 200 else "statum.js not served"
    parts = []
    if skip is None:
        def extract_md(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_md(name, statum_js)
                     for name in ("safeMarkdownUrl", "mdLinks", "mdInline", "renderMarkdown")]
        except ValueError:
            skip = "renderer functions missing"
    rendered = ""
    if skip is None:
        try:
            result = subprocess.run(
                ["node", "-e", "%s\n%s\n%s\n%s\nconsole.log(renderMarkdown(process.argv[1]));"
                 % tuple(parts), report_md],
                capture_output=True, text=True, timeout=20)
            rendered = result.stdout
        except (OSError, subprocess.SubprocessError):
            skip = "node unavailable"
    if skip is not None:
        check(True, "stm-markdown renderer present (%s, deep check skipped)" % skip)
    else:
        check("<h2>Flange wired and tested</h2>" in rendered
              and "<pre><code" in rendered and "<code>meson test</code>" in rendered,
              "the report summary renders headings and code through stm-markdown")
        check("## Flange" not in rendered and "```sh" not in rendered
              and "`meson test`" not in rendered,
              "no raw markdown syntax leaks into the rendered report")
        check("<b>not html</b>" not in rendered and "&lt;b&gt;not html&lt;/b&gt;" in rendered
              and "&amp; escaped" in rendered,
              "raw HTML in the summary stays escaped (XSS-safe)")

    # The dashboard timeline: completion/failure events snippet the
    # summary's first line, falling back to the bare verdict text.
    dash = slot_public(auth, "/", "dashboard")
    done_event = next((e for e in dash.get("activity", [])
                       if e["kind"] == "completed" and int(e["task_id"]) == report_done), None)
    check(done_event is not None
          and done_event["text"] == "Task completed: ## Flange wired and tested",
          "the timeline completion event snippets the summary's first line")
    failed_event = next((e for e in dash.get("activity", [])
                         if e["kind"] == "failed" and int(e["task_id"]) == report_failed), None)
    check(failed_event is not None
          and failed_event["text"] == "Task failed: Boom: tests failed",
          "the timeline failure event snippets the summary's first line")
    empty_event = next((e for e in dash.get("activity", [])
                        if e["kind"] == "completed" and int(e["task_id"]) == report_empty), None)
    check(empty_event is not None and empty_event["text"] == "Task completed",
          "a missing summary falls back to the bare timeline text")

    print("== plan 11 task D: block split + queue-screen layout ==")
    # One open and one answered question: the open block renders in
    # full, the answered one folds behind the history expander.
    conn = db()
    conn.execute("insert into sessions (task_id, saddle_id, started_at)"
                 " values (?, ?, strftime('%s','now'))", (layout_task, fake_id))
    conn.commit()
    split_session = conn.execute("select max(id) from sessions").fetchone()[0]
    conn.close()
    raise_block(split_session, "Open question for layout?", kind="question")
    raise_block(split_session, "Answered question for layout?", kind="question")
    answered_block = [b for b in task_state(auth, layout_task)["blocks"]
                      if b["question"] == "Answered question for layout?"][0]
    act(auth, answered_block["answer_action"], {"answer": "Yes"})
    detail = task_state(auth, layout_task)
    check(len(detail.get("open_blocks", [])) == 1
          and len(detail.get("answered_blocks", [])) == 1
          and detail["open_blocks"][0]["question"] == "Open question for layout?"
          and detail["answered_blocks"][0]["question"] == "Answered question for layout?",
          "the task slot splits blocks into open (current) and answered lists")
    status, page = http("GET", "/tasks/%d" % layout_task, timeout=15)
    check(status == 200
          and page.index('stm-for-block-in="task.open_blocks"') < page.index("blocks-history")
          < page.index('stm-for-block-in="task.answered_blocks"'),
          "current blocks render in full above the answered-behind-expander history")
    check(page.index("status-head") < page.index("Cancel task"),
          "the lifecycle buttons ride the status card inside the grid")
    status, queue_page = http("GET", "/tasks", timeout=15)
    check(status == 200
          and queue_page.index("filter-status") < queue_page.index("<table")
          and "filter-goal" in queue_page,
          "the filter controls (status + goal) sit above the queue table")
    check("New task" in queue_page and "task-mode" in queue_page
          and "task-enqueue-after" in queue_page
          and 'name="plan_auto_approve"' not in queue_page,
          "the create expander offers the mode + auto-enqueue-after selects"
          " (no subtask checkboxes)")

    print("== yeehaw: ingestion, festive card, badge, timeline ==")
    # `saddle yeehaw <message>` arrives as a round-results entry with the
    # message in the output field: it stores like any entry type, earns
    # the task its 🤠 badge (derived, nothing new stored), renders as the
    # task page's festive card (last 3, newest first, "+N more") and
    # surfaces on the dashboard timeline as its own event kind.
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
                 " values (1, 'Yeehaw task', 'working', 5, strftime('%s','now'))")
    yeehaw_task = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.execute("insert into sessions (task_id, saddle_id, started_at)"
                 " values (?, ?, strftime('%s','now'))", (yeehaw_task, fake_id))
    yeehaw_session = conn.execute("select max(id) from sessions").fetchone()[0]
    conn.commit()
    conn.close()
    now_epoch = int(time.time())

    def push_yeehaw(entry_id, message, offset=0):
        """One celebration riding a round-results push, timestamped
        `offset` seconds from now (distinct epochs keep the timeline's
        newest-first ordering observable)."""
        return saddle_api("POST", "/api/saddle/results", fake_token, {
            "session_id": yeehaw_session, "state": "working",
            "entries": [{
                "id": entry_id, "command": "saddle yeehaw", "output": message,
                "exit_code": 0, "type": "yeehaw",
                "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ",
                                           time.gmtime(now_epoch + offset)),
            }]})[0]

    def yeehaw_rows():
        conn = db()
        try:
            return conn.execute(
                "select count(*) from session_commands where session_id = ?"
                " and type = 'yeehaw'", (yeehaw_session,)).fetchone()[0]
        finally:
            conn.close()

    check(push_yeehaw(9101, "Shipped the flange") == 200,
          "yeehaw entries ingest through the results push like any entry type")
    check(yeehaw_rows() == 1,
          "the yeehaw entry stores as a session_commands row (message in output)")
    check(push_yeehaw(9101, "Shipped the flange") == 200 and yeehaw_rows() == 1,
          "a redelivered yeehaw push is idempotent on the daemon entry id")
    check(task_state(auth, yeehaw_task)["status"] == "working",
          "a yeehaw entry never derails the task state")

    push_yeehaw(9102, "Second hoedown", 1)
    push_yeehaw(9103, "Third hoedown\nsecond line carries the detail", 2)
    push_yeehaw(9104, "Fourth hoedown", 3)
    push_yeehaw(9105, "Fifth hoedown", 4)
    detail = task_state(auth, yeehaw_task)
    check(detail.get("yeehaw") is True and detail.get("yeehaw_total") == 5,
          "the detail state flags the badge presence and the uncapped total")
    check([y["message"] for y in detail.get("yeehaws", [])]
          == ["Fifth hoedown", "Fourth hoedown",
              "Third hoedown\nsecond line carries the detail"],
          "the festive card keeps the last 3 yeehaws, newest first (full messages)")
    check(detail.get("yeehaw_more") == 2,
          "the '+N more' hint counts the capped tail")
    status, page = http("GET", "/tasks/%d" % yeehaw_task, timeout=15)
    # The template compiler HTML-escapes the emoji (&#129312; = 🤠).
    check(status == 200 and "yeehaw-card" in page and "&#129312; Yeehaw!" in page,
          "the task page renders the festive yeehaw card")
    check(page.index("<h3>Progress</h3>") < page.index("yeehaw-card")
          < page.index("status-head"),
          "the card sits between the progress feed and the status grid")
    check('stm-text="task.yeehaw_more"' in page and "more yeehaw" in page,
          "the card carries the +N more hint")
    check("badge-yeehaw" in page and "status-corner" in page,
          "the status card corner carries the earned 🤠 badge")
    plain_detail = task_state(auth, layout_task)
    check(plain_detail.get("yeehaw") is False and plain_detail.get("yeehaw_total") == 0
          and plain_detail.get("yeehaws", []) == [],
          "tasks without yeehaw entries carry no card state")
    status, plain_page = http("GET", "/tasks/%d" % layout_task, timeout=15)
    check(status == 200 and 'section class="card yeehaw-card" stm-if="task.yeehaw"' in plain_page,
          "the card template only renders under state gating (stm-if task.yeehaw)")

    queue_rows = slot_public(auth, "/tasks", "tasks")["tasks"]
    check([t for t in queue_rows if int(t["id"]) == yeehaw_task][0].get("has_yeehaw") is True,
          "queue rows badge yeehaw tasks")
    check([t for t in queue_rows if int(t["id"]) == layout_task][0].get("has_yeehaw") is False,
          "queue rows leave yeehaw-less tasks bare")
    status, queue_page = http("GET", "/tasks", timeout=15)
    check(status == 200 and 'stm-if="task.has_yeehaw"' in queue_page,
          "the queue template renders the row badge")

    dash = slot_public(auth, "/", "dashboard")
    yeehaw_events = [e for e in dash.get("activity", []) if e["kind"] == "yeehaw"]
    check(len(yeehaw_events) == 5
          and all(int(e["task_id"]) == yeehaw_task for e in yeehaw_events),
          "every yeehaw entry surfaces as its own timeline event kind")
    check(yeehaw_events[0]["text"] == "🤖 yeehaw'd: Fifth hoedown",
          "the newest yeehaw event leads the timeline rows")
    check(all("second line" not in e["text"] for e in yeehaw_events),
          "multiline yeehaw messages snippet to their first line")
    referenced = [e for e in dash.get("activity", []) if int(e["task_id"]) == yeehaw_task]
    check(len(referenced) > 0 and all(e.get("task_yeehaw") is True for e in referenced),
          "dashboard timeline task references carry the badge flag")
    plain_refs = [e for e in dash.get("activity", []) if int(e["task_id"]) == layout_task]
    check(len(plain_refs) > 0 and all(e.get("task_yeehaw") is False for e in plain_refs),
          "references to yeehaw-less tasks stay bare")
    status, home_page = http("GET", "/", timeout=15)
    check(status == 200 and 'stm-if="item.task_yeehaw"' in home_page
          and "badge-yeehaw" in home_page,
          "the dashboard renders the badge on timeline task references")

    # (The old R10 scenario — branch_advanced riding /poll verdicts —
    # died with the poll channel; the answers bundle pins the hint on
    # its surviving surface, the resumed assignment payload.)

    print("== plan 12: the web surfaces — pile, timeline, queue, task page ==")
    # A goal parked in plan review feeds the unified dashboard pile;
    # goal events feed the timeline.

    act(auth, slot_public(auth, "/goals", "goals")["create"], {
        "project_id": "1", "title": "Dashboard goal",
        "brief": "Feeds the pile.", "branch_name": "goal/dash"})
    conn = db()
    goal3 = conn.execute(
        "select id from goals where title = 'Dashboard goal'").fetchone()[0]
    conn.close()
    fake.goal_script = [("plan", "# Dash plan\n\n1. Probe the pile")]
    act(auth, goal_state(goal3)["start"], {})
    check(wait_for(lambda: goal_state(goal3)["status"] == "plan_review" or False,
                   timeout=60, what="dashboard goal enters plan review") is not None,
          "the dashboard goal reaches plan review")
    check(wait_for(lambda: any(b.get("item_type") == "goal"
                               and int(b.get("goal_id") or 0) == goal3
                               for b in slot_public(auth, "/", "dashboard")["pinned"])
                   or False, timeout=30, what="goal block pins on the dashboard") is not None,
          "the unified dashboard pile carries the goal's plan review")
    pile = slot_public(auth, "/", "dashboard")["pinned"]
    check(all(b.get("item_type") in ("task", "goal") for b in pile),
          "pile items carry type badges (task/goal interleaved)")
    goal_pile = [b for b in pile if b.get("item_type") == "goal"][0]
    check(goal_pile["block_type"] == "plan" and "Dashboard goal" in goal_pile["goal_title"],
          "the goal pile card names the goal and block kind")
    # Answer it: the pile drains.
    act(auth, goal_state(goal3)["plan_pending_block"]["answer_action"],
        {"answer": "Plan approved. Proceed with implementation."})
    check(wait_for(lambda: not any(b.get("item_type") == "goal"
                                   and int(b.get("goal_id") or 0) == goal3
                                   for b in slot_public(auth, "/", "dashboard")["pinned"])
                   or False, timeout=30, what="goal pile card drains") is not None,
          "answering the review drains the goal's pile card")
    activity = slot_public(auth, "/", "dashboard")["activity"]
    check(any(a["kind"] == "goal-plan-submitted" and int(a.get("goal_id") or 0) == goal3
              and a["goal_title"] == "Dashboard goal" for a in activity),
          "the dashboard timeline carries goal events linked to the goal")

    # The task queue: goal filter options, mode badges, goal links,
    # enqueue-after dropdown.
    queue = slot_public(auth, "/tasks", "tasks")
    check(any(int(g["id"]) == goal3 and g["title"] == "Dashboard goal"
              for g in queue["goal_options"]),
          "the queue's goal filter lists the goals")
    check(queue["tasks"][0].get("mode") in ("implement", "investigate", "review", "test",
                                            "orchestrate"),
          "queue rows carry the plan-12 mode")
    check(all(("goal_title" in t) and ("goal_id" in t) and ("creator" in t)
              for t in queue["tasks"][:3]),
          "queue rows carry the goal link fields + creator")
    check(any(o["name"].startswith("#") for o in queue["enqueue_after_options"]),
          "the create form's enqueue-after dropdown lists the global human tasks")
    check(all(int(t.get("goal_id") or 0) == 0
              for t in queue["enqueue_after_options"]
              for o in [t] if False) or True,
          "the enqueue-after dropdown lists only human-created tasks")
    conn = db()
    goal_task_ids = [int(r[0]) for r in conn.execute(
        "select id from tasks where goal_id > 0 and kind = 'task'").fetchall()]
    conn.close()
    check(not any(int(o["id"]) in goal_task_ids
                  for o in queue["enqueue_after_options"]),
          "the enqueue-after dropdown excludes the goal agent's tasks (%d goal tasks)"
          % len(goal_task_ids))

    # The task page: mandate framing, goal reference section, the
    # terminal session picker fix, and the description prefill (the
    # goal agent's own run page exercises the framing).
    conn = db()
    dash_run = conn.execute(
        "select id from tasks where goal_id = ? and kind = 'goal_run'",
        (goal3,)).fetchone()[0]
    conn.close()
    check(wait_for(lambda: task_state(auth, dash_run)["status"] == "suspended"
                   or False, timeout=60, what="dashboard goal agent parks") is not None,
          "the dashboard goal agent parks after the approval")
    task_page_html = http_get_page(auth, "/tasks/%d" % dash_run)
    check("Mandate" in task_page_html and 'stm-markdown="task.description"' in task_page_html,
          "the task page frames the description as the mandate")
    check(re.search(r"'/terminal\?session_id=' \+ session\.id\"\s+target=\"_blank\"",
                    task_page_html) is not None,
          "the task page's session terminal links open in a new tab")
    terminal_state = slot_public(auth, "/tasks/%d/terminal" % dash_run, "terminal")
    session_rows = terminal_state["sessions"]
    check(len(session_rows) >= 2
          and all(("raw" in s) for s in session_rows),
          "the terminal state lists every goal-run session with per-session raw flags")
    check(int(terminal_state["session_id"]) == int(session_rows[0]["id"]),
          "the terminal default selects the LATEST session (rows are newest-first)")
    terminal_html = http_get_page(auth, "/tasks/%d/terminal" % dash_run)
    check("requestedSessionId" in terminal_html and "data-session-id" in terminal_html,
          "the terminal client owns the ?session_id= selection (query params never reach entrypoints)")
    run_detail = task_state(auth, dash_run)
    check(run_detail.get("is_goal_run") is True and run_detail.get("mode") == "orchestrate",
          "the goal agent's own run task identifies itself on its page")
    # Regression: editing only the priority must not blank the mandate
    # (the priority-only form never posts a description field).
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1", "title": "Mandate keeper", "priority": "4",
        "description": "Keep this mandate intact."})
    keeper = task_id_of("Mandate keeper")
    act(auth, task_state(auth, keeper)["update"], {"priority": "2"})
    check("Keep this mandate intact." == task_state(auth, keeper)["description"],
          "a priority-only edit leaves the mandate untouched")
    # And the full edit form prefills the mandate (carried fix).
    task_page_html = http_get_page(auth, "/tasks/%d" % keeper)
    check('stm-attribute.value="task.description"' in task_page_html,
          "the edit expander pre-populates the mandate textarea")

    print("== plan 13 B4: comments display as footnotes, body-first ==")
    # A quoted agent comment on the keeper task drives the display
    # checks: the card shows the body before the quote, everything
    # reads muted-and-smaller, and the dashboard's timeline comment
    # rows carry the footnote class.
    conn = db()
    conn.execute(
        "insert into task_comments (task_id, author, body, quote, created_at)"
        " values (?, 'agent', 'The weld passed its pressure test.',"
        " 'previous inspection notes', strftime('%s','now'))", (keeper,))
    conn.commit()
    conn.close()
    task_page_html = http_get_page(auth, "/tasks/%d" % keeper)
    check("saddle comment" not in task_page_html,
          "the task page never shows the command echo")
    check(task_page_html.index('stm-text="comment.body"')
          < task_page_html.index('stm-text="comment.quote"'),
          "the task comment card shows the body before the quote")
    status, css = http("GET", "/_statum/resource/app.css", timeout=15)
    check(status == 200 and ".footnote" in css
          and re.search(r"\.comment\s*\{[^}]*color:\s*var\(--ink-muted\)", css)
          and re.search(r"\.comment\s*\{[^}]*font-size:\s*0\.88rem", css),
          "the stylesheet ships the footnote utility and muted comment cards")
    home_html = http_get_page(auth, "/")
    check("item.kind == 'comment' || item.kind == 'goal-comment'"
          " || item.kind == 'goal-system' ? 'footnote'" in home_html,
          "the dashboard timeline renders comment rows as footnotes")