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/log" % plan_task, timeout=15)
    check(status == 200 and "data-log-feed" in page and "data-log-more" in page,
          "the event log page renders the feed and the load-more control")

    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, body = saddle_api("POST", "/api/saddle/chunks", fake_token, {
        "session_id": session_id,
        "events": [{"seq": 9021, "kind": "tool_result", "actor": "harness",
                    "payload": {"command": "bash -n script.sh",
                                "stderr": "syntax error near line 3 (recoverable)",
                                "exit_code": 1}}]})
    check(status == 200 and int(body.get("stored") or 0) == 1,
          "a failed tool result rides the events push")
    request = urllib.request.Request(WEB + "/tasks/%d/events/stream" % plan_task)
    with urllib.request.urlopen(request, timeout=15) as stream:
        result_event = None
        deadline = time.monotonic() + 8
        while time.monotonic() < deadline and result_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("kind") == "tool_result"
                    and "syntax error near line 3" in
                    (event.get("payload", {}).get("stderr") or "")):
                result_event = event
        check(result_event is not None,
              "the event stream carries tool results with their reason")

    # The page template renders the feed and the filters.
    status, page = http("GET", "/tasks/%d/log" % plan_task, timeout=15)
    check(status == 200 and "data-log-filter" in page and "data-log-search" in page,
          "the event log page renders filters and search")

    print("== B8: byte-exact payload delivery over the event stream ==")
    # Byte-exact delivery: ANSI-coloured turn text flows through the
    # ledger + SSE untouched (the client renderer styles them).
    coloured = "\x1b[1;32mbold green\x1b[0m then \x1b[38;5;196m256-red\x1b[0m\n"
    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token, {
        "session_id": session_id,
        "events": [{"seq": 9100, "kind": "turn_text",
                    "payload": {"turn": 9, "text": coloured}}]})
    check(status == 200 and int(body.get("stored") or 0) == 1,
          "the coloured turn text stores as one event")
    request = urllib.request.Request(WEB + "/tasks/%d/events/stream" % plan_task)
    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("kind") == "turn_text"
                    and "\x1b[1;32m" in (event.get("payload", {}).get("text") or "")):
                coloured_event = event
        check(coloured_event is not None
              and "\x1b[38;5;196m" in coloured_event["payload"]["text"],
              "payload bytes stream byte-exact to the client renderer")

    # 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")
    stored = [row for row in event_rows(rejected_task, ["command_output"])
              if "max_tokens exceeded" in row[1]]
    check(len(stored) >= 1,
          "command_output events stored in the audit trail (%d rows)" % len(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('data-long-form data-field="task"'),
          "status and description sit side by side inside the grid (status first)")
    check("Mandate" 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
    for offset, line in ((90, "first update"), (89, "second update")):
        payload = json.dumps({"text": line})
        conn.execute(
            "insert into events (kind, actor, scope_type, scope_id, payload,"
            " created_at) values ('progress', 'model', 'task', ?, ?,"
            " strftime('%s','now', ?))",
            (layout_task, payload, "-%d seconds" % offset))
    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]
    # The terminal transitions ride the ledger (state_change events
    # carrying the summary the timeline snippets).
    for task_id, to, summary in (
            (report_done, "done", report_md),
            (report_failed, "failed", "Boom: tests failed\n\nsecond line carries the detail"),
            (report_empty, "done", "")):
        payload = {"from": "working", "to": to,
                   "text": "Task completed" if to == "done" else "Task failed"}
        if summary:
            payload["summary"] = summary
        conn.execute(
            "insert into events (kind, actor, scope_type, scope_id, payload,"
            " created_at) values ('state_change', 'system', 'task', ?, ?,"
            " strftime('%s','now'))",
            (task_id, json.dumps(payload)))
    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("== 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"'/tasks/' \+ task\.id \+ '/log'\"\s+target=\"_blank\"",
                    task_page_html) is not None,
          "the task page's event log links open in a new tab")
    log_state = slot_public(auth, "/tasks/%d/log" % dash_run, "task_log")
    check(log_state.get("exists") is True and int(log_state["task_id"]) == dash_run,
          "the event log page state names the run task")
    log_html = http_get_page(auth, "/tasks/%d/log" % dash_run)
    check("EventSource" in log_html and "events/stream" in log_html,
          "the log page subscribes to the task's SSE stream")
    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")
    # Task 12: the mandate's edit surface is the in-place card
    # editor (the expander no longer carries the field); the editor
    # prefills from state, not stm-attribute.value (a content
    # attribute, which textareas ignore).
    task_page_html = http_get_page(auth, "/tasks/%d" % keeper)
    check('data-lf-src="task.description"' in task_page_html,
          "the mandate card editor prefills the textarea from state")

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

    print("== list ordering: newest first everywhere ==")
    # The ordering sweep: every table leads with the most recent row.
    # Fixtures land old-then-new; the first rendered data row must be
    # the newest task on the queue, and the create expander sits above
    # the table (verified on the compiled page markup).
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
                 " values (1, 'Order probe old', 'queued', 5, strftime('%s','now') - 100)")
    conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
                 " values (1, 'Order probe new', 'queued', 5, strftime('%s','now'))")
    conn.commit()
    conn.close()
    rows = [t for t in slot_public(auth, "/tasks", "tasks")["tasks"]
            if t["title"].startswith("Order probe")]
    check(len(rows) == 2 and rows[0]["title"] == "Order probe new",
          "the tasks table lists newest first (created desc, newest row leads)")
    status, queue_page = http("GET", "/tasks", timeout=15)
    check(status == 200
          and queue_page.index("New task") < queue_page.index("<table"),
          "the New task expander sits above the tasks table")
    # Comments/blocks threads: newest first since the sweep. The
    # comment list leads with the most recently added one.
    conn = db()
    conn.execute(
        "insert into task_comments (task_id, author, body, created_at)"
        " values (?, 'agent', 'Order probe old remark',"
        " strftime('%s','now') - 100)", (keeper,))
    conn.execute(
        "insert into task_comments (task_id, author, body, created_at)"
        " values (?, 'agent', 'Order probe new remark',"
        " strftime('%s','now'))", (keeper,))
    conn.commit()
    conn.close()
    bodies = [c["body"] for c in task_state(auth, keeper)["comments"]
              if c["body"].startswith("Order probe")]
    check(len(bodies) == 2 and bodies[0] == "Order probe new remark",
          "the task comment thread leads with the newest remark")


    # =====================================================================
    # Task 12 (task half): the mandate-card in-place editor.
    # =====================================================================
    print("== task 12: the mandate card editor ==")
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1", "title": "Mandate editor task", "priority": "4",
        "description": "# The original mandate\n\nWith a body."})
    conn = db()
    mandate_task = conn.execute(
        "select id from tasks where title = 'Mandate editor task'"
    ).fetchone()[0]
    conn.close()

    detail = task_state(auth, mandate_task)
    check(detail.get("editable_all") is True and detail.get("draft") is True,
          "the mandate editor task is a draft (every field editable)")
    task_page = http_get_page(auth, "/tasks/%d" % mandate_task)
    card_at = task_page.index('data-long-form data-field="task"')
    edit_at = task_page.index('data-lf-toggle>Edit</button>', card_at)
    area_at = task_page.index('data-lf-src="task.description"', card_at)
    check(card_at >= 0 and edit_at > card_at and area_at > card_at,
          "the mandate card carries the in-place Edit button and editor")
    check('name="description" rows="8"' in task_page,
          "the editor's textarea is the wide in-place one")
    # The expander no longer carries the mandate field (short
    # properties only) — and no other description textarea exists.
    expander_at = task_page.index('class="edit-expander"')
    expander_end = task_page.index('</details>', expander_at)
    expander_html = task_page[expander_at:expander_end]
    check('name="description"' not in expander_html,
          "the edit expander no longer carries the mandate field")

    # The PREFILL contract: the editor reads task.description from
    # state at open (JS-applied — stm-attribute.value is a no-op on
    # textarea). The compiled page therefore carries the state path,
    # and the save is a description-only update.
    check(task_page.count('data-lf-src="task.description"') == 1,
          "exactly one mandate editor textarea exists, keyed to state")
    act(auth, detail["update"], {"task_id": str(mandate_task),
                                 "description": "# The edited mandate\n\nSecond."})
    saved = task_state(auth, mandate_task)
    check(saved["description"] == "# The edited mandate\n\nSecond."
          and saved["title"] == "Mandate editor task",
          "the description-only save round-trips and leaves the title alone")

    # Editability rules unchanged: an ACTIVE task keeps the read-only
    # mandate (no Edit button, no editor textarea) while priority
    # remains editable in the expander.
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority,"
                 " created_at) values (1, 'Active mandate task', 'working', 5,"
                 " strftime('%s','now'))")
    conn.commit()
    active_task = conn.execute(
        "select id from tasks where title = 'Active mandate task'"
    ).fetchone()[0]
    conn.close()
    active = task_state(auth, active_task)
    check(active.get("editable_all") is False
          and active.get("editable_priority") is True,
          "an active task edits priority only")
    # The compiled page is state-independent markup: statum gates the
    # editor through stm-if, which HIDES (never removes) the branch.
    # What the raw page can promise: the Edit button and the editor
    # form are BOTH inside task.editable_all-gated branches (statum
    # hides them at render), while the rendered mandate view carries
    # no gate at all.
    active_page = http_get_page(auth, "/tasks/%d" % active_task)
    card_at = active_page.find('data-long-form data-field="task"')
    card_end = active_page.find('</section>', card_at)
    card_html = active_page[card_at:card_end]
    toggle_at = card_html.find('data-lf-toggle')
    editor_at = card_html.find('data-lf-edit')
    check(toggle_at < 0 or 'stm-if="task.editable_all"' in card_html,
          "an active task's Edit button is draft-gated (stm-if hides it)")
    check(editor_at < 0 or card_html.rfind('stm-if="task.editable_all"', 0, editor_at) > 0,
          "an active task's editor form sits inside the draft-gated branch")