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

    print("== B6b: create-form branch fields (goal 16) ==")
    # The queue form carries both branch rows; they post through
    # tasks.create into the stored task (empty optional -> null,
    # branch_from prefill from the project's default branch).
    tasks_page = http_get_page(auth, "/tasks")
    check('name="task_branch"' in tasks_page and "Task branch name" in tasks_page,
          "the create form renders the Task branch name input")
    check('name="branch_from"' in tasks_page and "Branch from" in tasks_page,
          "the create form renders the Branch from input")
    check("default: the project's default branch" in tasks_page,
          "the Branch from label hints at the project default")
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1",
        "title": "Branch fields task",
        "priority": "5",
        "task_branch": "task/999-branch-fields",
        "branch_from": "release/2026",
    })
    detail = task_state(auth, task_id_of("Branch fields task"))
    check(detail["task_branch"] == "task/999-branch-fields"
          and detail["branch_from"] == "release/2026",
          "posted branch fields store on the created task")
    # Both optional: empty posts leave task_branch null and prefill
    # branch_from from the project's default branch (demo -> main).
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1",
        "title": "Default branch fields task",
        "priority": "5",
    })
    detail = task_state(auth, task_id_of("Default branch fields task"))
    check((detail.get("task_branch") is None
           or detail.get("task_branch") == "")
          and detail.get("branch_from") == "main",
          "empty posts leave task_branch null and prefill branch_from"
          " from the project default")
    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")

    print("== goal 9: the yeehaw celebration rides the ledger + stream ==")
    # The daemon's yeehaw tool emits a `yeehaw` ledger event (actor
    # model, payload {message}); it must store, stream and leave the
    # page wiring intact for the client renderer (the comments bucket
    # + .yeehaw-message span are pinned by test_task_log_js.py).
    whoop = "Finally cracked the SSE replay - every byte landed."
    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token, {
        "session_id": session_id,
        "events": [{"seq": 9200, "kind": "yeehaw", "actor": "model",
                    "payload": {"message": whoop}}]})
    check(status == 200 and int(body.get("stored") or 0) == 1,
          "a yeehaw event stores on the session ledger (%s)" % body.get("error"))
    request = urllib.request.Request(WEB + "/tasks/%d/events/stream" % plan_task)
    with urllib.request.urlopen(request, timeout=15) as stream:
        yeehaw_event = None
        deadline = time.monotonic() + 8
        while time.monotonic() < deadline and yeehaw_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") == "yeehaw":
                yeehaw_event = event
        check(yeehaw_event is not None
              and yeehaw_event.get("actor") == "model"
              and yeehaw_event.get("payload", {}).get("message") == whoop,
              "the yeehaw event streams with its message intact")
    conn = db()
    stored_kind = conn.execute(
        "select kind, actor from events where session_id = ? and seq = 9200",
        (session_id,)).fetchone()
    conn.close()
    check(stored_kind is not None and stored_kind[0] == "yeehaw"
          and stored_kind[1] == "model",
          "the ledger row keeps the yeehaw kind + model actor")
    status, page = http("GET", "/tasks/%d/log" % plan_task, timeout=15)
    check(status == 200 and "data-log-feed" in page
          and "yeehaw" not in page.split("<script>", 1)[0],
          "the log page still renders with the yeehaw renderer shipped in its script")

    # 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 "with-sidebar act-consult" in page
          and page.index("<h3>Progress</h3>") < page.index("status-head"),
          "the progress feed renders above the status card")
    check(page.index('data-subnav-section="task-blocks"') < page.index("status-head"),
          "the blocks section leads the acting column above the status card")
    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)")
    # Goal 41 T5: the edit surface is the in-place Edit task card
    # (no expander) in the acting column, below the progress feed.
    check(history_page.index("Show history") < history_page.index(">Edit task"),
          "the Edit task card sits under the progress history expander")
    check("edit-expander" not in history_page,
          "the old edit expander is gone (in-place editor instead)")
    check(history_page.index('data-subnav-section="task-blocks"')
          < history_page.index('data-subnav-section="task-comments"'),
          "the blocks section precedes comments in the acting column")

    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("data-subnav-section=\"task-blocks\"") < page.index("completion-report")
          < page.index("status-head"),
          "the completion-report card sits below the blocks, above the status card")
    card = page[page.index("completion-report"):page.index("status-head")]
    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("goal-strip") < queue_page.index("<table")
          and "filter-goal" in queue_page,
          "the filter controls (status chips + goal select) 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 form 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.
    # (Goal 27 refinements — the parking guard: the later scripted park
    # needs a live wake source under the goal; plant one before the
    # approval resumes the agent.)
    conn = db()
    conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
                 " priority, mode, creator, created_at) values (1, ?, 'task',"
                 " 'Dash live worker', 'blocked', 5, 'implement', 'goal',"
                 " strftime('%s','now'))", (goal3,))
    conn.commit()
    conn.close()
    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")
    # Goal 41 T7 (ux-notes structural bet 3): the event log is the
    # task page's second tab, not a new-tab navigation - the toggle is
    # a <label> for the log radio, the shared shell rides the panel,
    # and no log link on the page opens a new tab any more.
    check('id="task-view-log"' in task_page_html and 'class="tabs task-tabs"' in task_page_html,
          "the task page mounts the Overview|Log tabs")
    check(re.search(r'<label class="tab-link" for="task-view-log">Event log</label>',
                    task_page_html) is not None,
          "the task page's event log control toggles the log tab in place")
    check(re.search(r"/log'\s*\"\s*target=\"_blank\"", task_page_html) is None,
          "the task page no longer opens the event log in a new tab")
    check('class="log-shell"' in task_page_html,
          "the task page embeds the shared log shell (compiled in from the partial)")
    task_log_state = slot_public(auth, "/tasks/%d" % dash_run, "task_log")
    check(task_log_state.get("exists") is True
          and int(task_log_state["task_id"]) == dash_run,
          "the task page state also publishes the embedded log slot")
    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)
    # The log page's client JS ships as a compiled resource (plan 17
    # b3+), so the SSE subscription is asserted on the served resource
    # the page references, not the page HTML itself.
    status, log_js = http("GET", "/_statum/resource/task_log.js", timeout=15)
    check(status == 200 and "task_log.js" in log_html
          and "EventSource" in log_js and "events/stream" in log_js,
          "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-markdown="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\(--spry-fg-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, "/")
    # Goal 41 T10: the dashboard's feed is the work stream now; the
    # comment-as-footnote contract lives on in its log-highlight rows
    # (the compiled page escapes && as &amp;&amp; in the attribute).
    check("item.kind == 'log_highlight' &amp;&amp; item.event_kind"
          " == 'comment' ? 'footnote'" in home_html,
          "the dashboard stream 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 button 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")
    # Goal 41 T5: the edit surface is the in-place Edit task card —
    # short properties only (no mandate field).
    edit_card_at = task_page.index('>Edit task')
    edit_card_end = task_page.index('</section>', edit_card_at)
    edit_html = task_page[edit_card_at:edit_card_end]
    check('name="description"' not in edit_html,
          "the Edit task card carries no mandate field")
    check('data-lf-toggle' in edit_html and 'data-lf-edit' in edit_html,
          "the Edit task card is the in-place editor (view + toggle + form)")
    check(task_page.count('data-long-form data-field="task"') == 2,
          "two in-place cards exist: the mandate and the Edit task card")

    # 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)
    # Goal 41 T5: two in-place cards exist (Edit task, Mandate); the
    # draft-gate assertions address the MANDATE card.
    card_at = active_page.find('>Mandate')
    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")

    # =====================================================================
    # Goal 8 task 38: the schema is UPGRADEABLE now — M0002 is the
    # incremental migration that un-flattened it. On this bundle's
    # fresh boot both migrations ran: assert serial 2 is recorded and
    # every object M0002 owns exists, so a future in-place edit of
    # M0001 (the regression this goal shipped against) is caught by
    # the serial staying 1 here while the schema drifts.
    # P1 (M0003) extends the chain: serial 3 records the event ledger
    # reader indexes.
    print("== migrations: incremental schema (M0002, M0003 recorded) ==")
    conn = db()
    serials = conn.execute(
        "select serial from __migrations where namespace = 'clanker-saddle'"
        " order by serial").fetchall()
    event_indexes = {r[0] for r in conn.execute(
        "select name from sqlite_master"
        " where type = 'index' and tbl_name = 'events'").fetchall()}
    conn.close()
    check([s[0] for s in serials] == [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11],
          "the clanker-saddle namespace carries exactly serials 1 through 11"
          " (a fresh boot ran M0001 then the M0002 no-op then M0003,"
          " M0004 understandings, M0005 discoveries, M0006 advisor,"
          " M0007 the parking guard, M0008 answer-wake, M0009 the"
          " turn-cap park, M0010 the model config and M0011 the user"
          " preferences)")
    check(len(serials) == 11,
          "no duplicate migration rows (each guarded up() applied once)")
    check({"idx_events_scope", "idx_events_terminal"} <= event_indexes,
          "M0003's event reader indexes exist (fresh boot)")

    conn = db()
    tables = {r[0] for r in conn.execute(
        "select name from sqlite_master where type = 'table'")}
    saddle_cols = {r[1] for r in conn.execute(
        "pragma table_info('saddles')")}
    conn.close()
    check({"notes", "stacks", "stack_projects"} <= tables,
          "M0002's tables (notes, stacks, stack_projects) exist")
    check({"prompt_note", "workspace_root"} <= saddle_cols,
          "M0002's saddles columns (prompt_note, workspace_root) exist")

    print("== S5: the task page's LLM usage stat line ==")
    # usage_summary aggregates the seeded usage events (the daemon's
    # prompt/completion/elapsed payload) into one muted stat line on
    # the status card; zero-usage tasks carry no line at all.
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority,"
                 " created_at) values (1, 'Usage ledger task', 'working', 5,"
                 " strftime('%s','now'))")
    usage_task = conn.execute("select max(id) from tasks").fetchone()[0]
    for payload in ({"prompt_tokens": 2084, "completion_tokens": 66,
                     "elapsed_ms": 8519},
                    {"prompt_tokens": 2458, "completion_tokens": 148,
                     "elapsed_ms": 6032},
                    {"prompt_tokens": 4519, "completion_tokens": 243,
                     "elapsed_ms": 10544}):
        conn.execute(
            "insert into events (kind, actor, scope_type, scope_id, payload,"
            " created_at) values ('usage', 'harness', 'task', ?, ?,"
            " strftime('%s','now'))",
            (usage_task, json.dumps(payload)))
    conn.execute("insert into tasks (project_id, title, status, priority,"
                 " created_at) values (1, 'Usage magnitude task', 'working', 5,"
                 " strftime('%s','now'))")
    magnitude_task = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.execute(
        "insert into events (kind, actor, scope_type, scope_id, payload,"
        " created_at) values ('usage', 'harness', 'task', ?, ?,"
        " strftime('%s','now'))",
        (magnitude_task, json.dumps({"prompt_tokens": 45200000,
                                     "completion_tokens": 1100000,
                                     "elapsed_ms": 11520000})))
    conn.commit()
    conn.close()
    detail = task_state(auth, usage_task)
    check(detail.get("usage_line") == "LLM: 3 calls · 9,061 in / 457 out tok · 25s",
          "the task state aggregates the seeded usage events (calls, tokens,"
          " wall time): %r" % detail.get("usage_line"))
    check(task_state(auth, magnitude_task).get("usage_line")
          == "LLM: 1 calls · 45.2M in / 1.1M out tok · 3h 12m",
          "token magnitudes and durations format human-friendly"
          " (%r)" % task_state(auth, magnitude_task).get("usage_line"))
    task_page = http_get_page(auth, "/tasks/%d" % usage_task)
    check('stm-if="task.usage_line.length &gt; 0"' in task_page
          and 'stm-text="task.usage_line"' in task_page,
          "the task page renders the usage stat line (gated on non-empty)")
    check(task_state(auth, draft_display).get("usage_line") == "",
          "a task with no usage events carries no stat line")