clanker-saddle/test/bundles/goalui.py

"""The goalui 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


def run(ctx):
    """Plan 14 workstream B + plan 15 C1: the goal supervision surface.

    B1 — goal-run control parity (Retry supervision / Requeue-to-resume
    on the status head, the /tasks/<goal-run-id> redirect, no
    notify-me toggle), B2 — the plan-review remarks queue, B4 —
    sessions.resume_notice + the terminal dropdown/resume panel, B5 —
    page order, effective rules and per-task capability badges, B6 —
    narrative footnote weights. C1 — the single-row block world: the
    goal's blocks ARE the run task's rows (answered from every
    surface through one path; no dual rows, no stragglers, at most
    one review-kind block per verdict path). C4 — the goal page
    batch: one muted head activity line, plan between attention and
    tasks, answered blocks behind an expander, footnote comments.
    C5 — the dashboard deep-link closure: goal pile cards link the
    goal page, every surface settles the single row, and no
    goal-side store exists to hold stragglers.
    """
    auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token

    def run_task_of(goal_id):
        conn = db()
        try:
            return int(conn.execute(
                "select id from tasks where goal_id = ? and kind = 'goal_run'",
                (goal_id,)).fetchone()[0])
        finally:
            conn.close()

    def wait_run(status, timeout=60, what="run status"):
        return wait_for(lambda: task_state(auth, run_task_of(goal))["status"] == status
                        or False, timeout=timeout, what=what)

    def pile_questions():
        return [b.get("question") for b in
                slot_public(auth, "/", "dashboard")["pinned"]
                if b.get("item_type") == "goal"]

    def open_session(task_id):
        conn = db()
        try:
            row = conn.execute(
                "select id from sessions where task_id = ? and ended_at is null",
                (task_id,)).fetchone()
            return int(row[0]) if row else None
        finally:
            conn.close()

    print("== B1: the goal-run task page redirects to the goal ==")
    fake.goal_script = [("plan", "# Remarks plan\n\n## Steps\n\n1. First move")]
    act(auth, slot_public(auth, "/goals", "goals")["create"], {
        "project_id": "1", "title": "Goalui goal",
        "brief": "Drive the plan-14 goal surface end to end.",
        "branch_name": "goal/goalui"})
    conn = db()
    goal = conn.execute("select id from goals where title = 'Goalui goal'"
                        ).fetchone()[0]
    conn.close()
    act(auth, goal_state(goal)["start"], {})
    wait_run("blocked", what="goalui goal agent parks on its plan")  # plan review parks blocked
    run_task = run_task_of(goal)

    frames = directives("/tasks/%d" % run_task, auth)
    navigations = [d for d in frames if d.get("type") == "navigate"]
    check(any(d.get("uri") == "/goals/%d" % goal for d in navigations),
          "the goal-run task page redirects to /goals/<id> (navigate directive)")
    check(slot_public(auth, "/tasks/%d" % run_task, "task")["id"] == run_task,
          "the redirect still hydrates the task slot (sealed refresh + API readers)")

    status, page = http("GET", "/tasks/%d" % run_task, headers={
        "X-Statum-Slot": "key=%s; as_at=%s" % (auth["key"], auth["as_at"])}, timeout=15)
    check(status == 200, "the compiled goal-run page serves")

    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1", "title": "Goalui worker", "priority": "4",
        "description": "A plain worker under no goal."})
    conn = db()
    worker = conn.execute("select id from tasks where title = 'Goalui worker'"
                          ).fetchone()[0]
    conn.close()
    check(not [d for d in directives("/tasks/%d" % worker, auth)
               if d.get("type") == "navigate"],
          "worker task pages render normally (no redirect)")

    state = goal_state(goal)
    check("toggle_notify" not in state and state.get("notify") is None,
          "no notify-me toggle on the goal (goals notify unconditionally)")
    goal_page = http_get_page(auth, "/goals/%d" % goal)
    check("Notify me" not in goal_page and "toggle-notify" not in goal_page,
          "the goal page template ships no notify control")
    check(state["branch_name"] == "goal/goalui" and state["base_branch"] == "main"
          and isinstance(state["push_branches"], list)
          and state["priority"] == 3
          and len(state["started_display"]) > 0
          and len(state["agent_saddle"]) > 0,
          "the status head carries the full field set (branch pair, chips list,"
          " priority, timestamps, agent saddle)")
    check("Retry supervision" in goal_page and "Requeue to resume" in goal_page,
          "the status head ships the run-control parity buttons")
    check("status-corner" not in goal_page and "goal complete" not in goal_page
          and goal_page.count("badge-yeehaw") == 1,
          "yeehaw is rows-only (no goal-head corner badge; the row badge remains)")

    check(goal_state(goal)["current_activity"] == "parked awaiting the plan review",
          "the head line reads 'parked awaiting the plan review' while one is open")
    goal_page = http_get_page(auth, "/goals/%d" % goal)
    check(goal_page.index("Goal agent blocks") < goal_page.index("<h3>Plan</h3>")
          < goal_page.index("<h3>Tasks</h3>"),
          "the plan section sits between attention and tasks")
    check("data-plan-review" in goal_page and "Approved plan &amp; revisions" in goal_page,
          "an open review renders expanded beside the collapsed approved-plan expander")

    print("== B2: the plan-review remarks queue ==")
    check(goal_state(goal)["status"] == "plan_review"
          and goal_state(goal)["plan_pending_block"] is not None,
          "the goal sits in plan review with a pending block")
    goal_page = http_get_page(auth, "/goals/%d" % goal)
    review_at = goal_page.index("data-plan-review")
    check(review_at > 0, "the plan review region is marked for the remarks queue")
    check(goal_page.index('data-block-id="goal.plan_pending_block.id"') > 0,
          "the queue keys on the pending goal block id")
    body_at = goal_page.index('data-plan-body')
    remarks_at = goal_page.index("data-plan-remarks")
    check(body_at < remarks_at,
          "the remarks queue renders at the bottom of the plan body")
    for marker in ("data-plan-accept", "data-plan-toggle", "data-plan-submit",
                   "data-plan-remove", "plan-popover", "remarksByBlock"):
        check(marker in goal_page, "the remarks queue ships %s" % marker)
    check("(index + 1) + '. '" in goal_page,
          "the rejection payload numbers every queued remark")
    check("rejectOpenByBlock" in goal_page and "statum.afterEvaluate(syncPlanRemarks)" in goal_page,
          "the queue survives pulse refreshes (maps outside statum re-renders)")

    # The rejection payload exactly as the client composes it: the
    # stable verdict wording, then each remark NUMBERED in queue order
    # with its quote prefixed like a comment.
    payload = ("Plan feedback received:\n\n"
               "> ## Steps\n\n1. First move\n\n"
               "1. Add a rollback step\n\n"
               "2. Name the owner of the migration")
    act(auth, goal_state(goal)["plan_pending_block"]["answer_action"],
        {"answer": payload})
    state = goal_state(goal)
    check(state["status"] == "planning",
          "the remarks rejection returns the agent to planning")
    rejected = [r for r in state["plan_revisions"] if r.get("answered")]
    check(any(r["answer"] == payload for r in rejected),
          "the rejection carries the remarks verbatim, in order")
    check(payload.index("1. Add a rollback step") < payload.index("2. Name the owner"),
          "the numbering is queue-ordered")

    print("== B4: resume_notice lands on the resumed session row ==")
    fake.goal_script = [("question", "Which datastore?", ["SQLite", "Redis"])]
    check(wait_for(lambda: goal_state(goal)["status"] == "planning" or False,
                   timeout=30, what="rejection observed") is not None,
          "the agent revises")
    # The parked plan agent resumes with the verdict; the next round
    # raises the scripted question.
    check(wait_for(lambda: any(b["question"] == "Which datastore?"
                               and not b.get("answered")
                               for b in goal_state(goal)["blocks"]) or False,
                   timeout=60, what="question raised") is not None,
          "the resumed agent raises its question")
    conn = db()
    notices = dict(conn.execute(
        "select id, resume_notice from sessions where task_id = ?", (run_task,)
    ).fetchall())
    conn.close()
    resumed_ids = [i for i, text in notices.items() if text]
    fresh_ids = [i for i, text in notices.items() if not text]
    check(len(resumed_ids) >= 1 and any("Plan feedback received" in notices[i]
                                        for i in resumed_ids),
          "the resumed session stores the notice it was allocated")
    check(min(notices) in fresh_ids,
          "session 1 (the fresh opening round) carries no resume notice")

    print("== C1: the single run-task row IS the goal's block ==")
    check("Which datastore?" in pile_questions(),
          "the open goal question pins onto the dashboard pile")
    goal_rows = [b for b in goal_state(goal)["blocks"]
                 if b["question"] == "Which datastore?"]
    run_rows = [b for b in task_state(auth, run_task)["blocks"]
                if b["question"] == "Which datastore?"]
    check(len(run_rows) == 1 and len(goal_rows) == 1
          and int(run_rows[0]["id"]) == int(goal_rows[0]["id"]),
          "the goal page and the run task read the SAME single row")
    pile_card = [b for b in slot_public(auth, "/", "dashboard")["pinned"]
                 if b.get("question") == "Which datastore?"][0]
    check(pile_card.get("item_type") == "goal"
          and int(pile_card.get("goal_id") or 0) == goal
          and int(pile_card.get("id") or 0) == int(run_rows[0]["id"])
          and pile_card.get("goal_title") == "Goalui goal",
          "the pile card is the run-task row styled as a goal item")
    act(auth, [b for b in goal_state(goal)["blocks"]
               if b["question"] == "Which datastore?" and not b.get("answered")]
        [0]["answer_action"], {"option": "SQLite"})
    conn = db()
    unanswered = conn.execute(
        "select count(*) from task_blocks where question = 'Which datastore?'"
        " and answer is null").fetchone()[0]
    conn.close()
    check(unanswered == 0, "answering from the goal page closes the single row")
    check("Which datastore?" not in pile_questions(),
          "the answered row no longer pins the dashboard pile")

    # The task-page deep link (the AnswerBlockAction API surface)
    # answers the SAME row through the one goal path — there is no
    # second row to sweep and no stale copy can survive.
    conn = db()
    conn.execute("insert into task_blocks (task_id, type, question, created_at)"
                 " values (?, 'question', 'Deep link probe?',"
                 " strftime('%s','now'))", (run_task,))
    conn.commit()
    conn.close()
    check("Deep link probe?" in pile_questions(),
          "the fixture row pins like any goal question")
    deep_block = [b for b in task_state(auth, run_task)["blocks"]
                  if b["question"] == "Deep link probe?" and not b.get("answered")][0]
    act(auth, deep_block["answer_action"], {"answer": "Answered via the deep link"})
    conn = db()
    answered_row = conn.execute(
        "select answer from task_blocks where question = 'Deep link probe?'"
    ).fetchone()[0]
    tables = {r[0] for r in conn.execute(
        "select name from sqlite_master where type = 'table'").fetchall()}
    conn.close()
    check(answered_row == "Answered via the deep link",
          "the task-page deep link answers the same single row (the API path)")
    check("Deep link probe?" not in pile_questions(),
          "the deep-link answer closes the pile card too")
    check("goal_blocks" not in tables,
          "the goal_blocks table is gone (run-task blocks are canonical)")

    print("== C5: deep-link closure on the single-row world ==")
    # The pile card's goal item deep-links the goal page (the answer
    # surface); answering from either surface settled THE row above.
    home_markup = http_get_page(auth, "/")
    check("'/goals/' + blockview.goal_id" in home_markup,
          "the pile card deep-links the goal page for goal items")
    conn = db()
    dupes = conn.execute(
        "select count(*) from (select question from task_blocks where task_id = ?"
        " group by question having count(*) > 1)", (run_task,)).fetchone()[0]
    stragglers = conn.execute(
        "select count(*) from task_blocks where task_id = ? and answer is null",
        (run_task,)).fetchone()[0]
    goal_side_tables = [r[0] for r in conn.execute(
        "select name from sqlite_master where type = 'table'"
        " and name in ('goal_blocks','goal_comments','goal_events')").fetchall()]
    conn.close()
    check(dupes == 0 and stragglers == 0,
          "one row per question on the run task, none left unanswered")
    check(not goal_side_tables,
          "no goal-side store exists — historical stragglers are impossible")

    print("== B4: the terminal session dropdown + resume-info panel ==")
    fake.goal_script = [("suspend",)]
    check(wait_for(lambda: task_state(auth, run_task)["status"] == "suspended"
                   or False, timeout=60, what="agent parks suspended") is not None,
          "the agent parks suspended for the requeue scenario")
    terminal = slot_public(auth, "/tasks/%d/terminal" % run_task, "terminal")
    ids = [int(s["id"]) for s in terminal["sessions"]]
    check(ids == sorted(ids, reverse=True),
          "the session dropdown lists newest-first")
    check(int(terminal["session_id"]) == ids[0],
          "the terminal still defaults to the latest session")
    labels = {int(s["id"]): s["label"] for s in terminal["sessions"]}
    check(all(label.startswith("Session #") for label in labels.values()),
          "dropdown labels read 'Session #N — started <time>'")
    # Recompute at the check point: the C1 answers above each
    # resumed the agent too, so the earlier snapshot would go stale.
    conn = db()
    resumed_ids = [int(i) for (i, text) in conn.execute(
        "select id, resume_notice from sessions where task_id = ?",
        (run_task,)).fetchall() if text]
    conn.close()
    resumed_markers = [i for i in ids if "[resumed]" in labels[i]]
    check(set(resumed_markers) == set(resumed_ids),
          "the [resumed] marker rides exactly the notice-carrying sessions")
    notices_by_id = {int(s["id"]): s.get("notice") or "" for s in terminal["sessions"]}
    check("Opening brief" in notices_by_id[min(ids)]
          and "Drive the plan-14 goal surface" in notices_by_id[min(ids)],
          "session 1's panel shows the opening-brief framing (from the goal brief)")
    check(any("Plan feedback received" in (notices_by_id[i] or "") for i in resumed_ids),
          "a resumed session's panel shows the stored resume notice")
    terminal_page = http_get_page(auth, "/tasks/%d/terminal" % run_task)
    check('id="session-select"' in terminal_page
          and "window.terminalPickSession" in terminal_page,
          "the session picker is an options dropdown")
    check('stm-markdown="window.terminalSessionNotice(terminal)"' in terminal_page,
          "the panel renders the selected session's notice as markdown")
    check("No opening brief or resume notice recorded" in terminal_page,
          "the panel ships its graceful empty state")
    css = open(os.path.join(REPO, "src/pages/app.css")).read()
    check("resume-notice-panel" in css and "max-height: 200px" in css,
          "the panel is a 200px vertical-scroll strip (stylesheet)")

    print("== B1: requeue-to-resume from the goal head ==")
    conn = db()
    conn.execute("update saddles set status = 'offline' where id = ?", (fake_id,))
    conn.commit()
    conn.close()

    def requeue_offered():
        # Act on the SAME state that satisfied the wait: a second read
        # once raced the scripted daemon's ask reviving the saddle row,
        # and the conditionally-authored action vanished between the
        # two reads (KeyError on act).
        offered = goal_state(goal)
        return offered if offered.get("requeue_run") is not None else False
    state = wait_for(requeue_offered, timeout=30, what="requeue offered")
    check(state is not None,
          "a parked run on a dead saddle offers Requeue-to-resume")
    check(state.get("retry_run") is None,
          "an active goal offers no Retry supervision")
    open_before = open_session(run_task)

    def parked_session_ended():
        if open_before is None:
            return True
        conn = db()
        try:
            return conn.execute(
                "select ended_at is not null from sessions where id = ?",
                (open_before,)).fetchone()[0] == 1
        finally:
            conn.close()
    act(auth, state["requeue_run"], {})
    check(task_state(auth, run_task)["status"] == "queued",
          "the requeue returns the run to the queue")
    # The PARKED session itself must end (the restore-on-dispatch
    # successor may open immediately — the check keys the old row,
    # not the task's transient open-session window).
    check(wait_for(parked_session_ended, timeout=60,
                   what="parked session closed") is not None,
          "the requeue closed the parked session (restore-on-dispatch)")
    conn = db()
    conn.execute("update saddles set status = 'idle' where id = ?", (fake_id,))
    conn.commit()
    conn.close()
    fake.goal_script = [("suspend",)]
    check(wait_for(lambda: task_state(auth, run_task)["status"] == "suspended"
                   or False, timeout=60, what="requeued run re-dispatches") is not None,
          "the requeued run re-dispatches and re-parks")
    fresh_session = open_session(run_task)
    conn = db()
    fresh_notice = conn.execute(
        "select resume_notice from sessions where id = ?", (fresh_session,)
    ).fetchone()[0]
    conn.close()
    check(fresh_notice is None,
          "a requeue's fresh session carries no resume notice (empty panel state)")

    print("== B1: Retry supervision revives a failed goal ==")
    fake.push_results("failed", summary="the supervision collapsed (scripted)")
    check(wait_for(lambda: goal_state(goal)["status"] == "failed" or False,
                   timeout=30, what="goal fails") is not None,
          "the run's failure fails the goal")
    check(wait_for(lambda: goal_state(goal).get("retry_run") is not None
                   or False, timeout=30, what="retry offered") is not None,
          "a failed goal offers Retry supervision")
    fake.goal_script = [("suspend",)]
    act(auth, goal_state(goal)["retry_run"], {})
    state = goal_state(goal)
    check(state["status"] == "active",
          "the retry revives supervision (goal back to active)")
    check(task_state(auth, run_task)["status"] in ("queued", "assigned", "working",
                                                   "suspended", "blocked"),
          "the retry re-offered the run task")
    check(wait_for(lambda: task_state(auth, run_task)["status"] == "suspended"
                   or False, timeout=60, what="retried supervision parks") is not None,
          "the retried supervision dispatches and parks again")
    state = goal_state(goal)
    check(any(e["kind"] == "system" and "Supervision retried" in e["text"]
              for e in state["events"]),
          "the narrative records the supervision retry (a system row)")

    print("== B5 + C4: order, effective rules, capability badges ==")
    goal_page = http_get_page(auth, "/goals/%d" % goal)
    markers = [
        ('class="card goal-head"', "status head"),
        ("Agent progress", "progress"),
        ("Goal agent blocks", "attention blocks"),
        ("<h3>Plan</h3>", "plan"),
        ("<h3>Tasks</h3>", "tasks"),
        ("Initial brief", "brief"),
        ("Involved projects", "projects"),
        ("<h3>Rules</h3>", "rules"),
        ("Supervision narrative", "narrative"),
        ("<h3>Comments</h3>", "comments"),
        ("Goal agent runs", "runs"),
    ]
    positions = []
    for marker, _ in markers:
        check(marker in goal_page, "the goal page ships the %s section" % _)
        positions.append(goal_page.index(marker))
    check(positions == sorted(positions),
          "the page reads status → progress → attention → plan → tasks → brief →"
          " rules → narrative → comments → runs (in order)")
    check("goal.current_activity" in goal_page and "Current activity" not in goal_page,
          "the activity lives as ONE muted head line (the card is gone)")
    check(".goal-page .card" in open(os.path.join(REPO, "src/pages/app.css")).read(),
          "the stylesheet restores the goal page's card rhythm (padding regression)")

    rules_state = slot_public(auth, "/rules", "rules")
    act(auth, rules_state["create"], {"name": "goalui-primary", "body": "Primary rule"})
    act(auth, rules_state["create"], {"name": "goalui-codev", "body": "Co-dev rule"})
    act(auth, rules_state["create"], {"name": "goalui-ref", "body": "Reference rule"})
    conn = db()
    rule_ids = {name: int(conn.execute("select id from rules where name = ?",
                                       (name,)).fetchone()[0])
                for name in ("goalui-primary", "goalui-codev", "goalui-ref")}
    conn.execute("insert into rule_projects (rule_id, project_id, created_at)"
                 " values (?, 1, strftime('%s','now'))", (rule_ids["goalui-primary"],))
    conn.execute("insert into projects (name, git_url, default_branch, created_at)"
                 " values ('goalui-codev-project', 'https://example.com/g.git',"
                 " 'main', strftime('%s','now'))")
    codev_project = conn.execute(
        "select id from projects where name = 'goalui-codev-project'").fetchone()[0]
    conn.execute("insert into projects (name, git_url, default_branch, created_at)"
                 " values ('goalui-ref-project', 'https://example.com/r.git',"
                 " 'main', strftime('%s','now'))")
    ref_project = conn.execute(
        "select id from projects where name = 'goalui-ref-project'").fetchone()[0]
    conn.execute("insert into rule_projects (rule_id, project_id, created_at)"
                 " values (?, ?, strftime('%s','now'))",
                 (rule_ids["goalui-codev"], codev_project))
    conn.execute("insert into rule_projects (rule_id, project_id, created_at)"
                 " values (?, ?, strftime('%s','now'))",
                 (rule_ids["goalui-ref"], ref_project))
    conn.commit()
    conn.close()
    act(auth, goal_state(goal)["add_link"], {"linked_project_id": str(codev_project),
                                             "relationship": "co-developed"})
    act(auth, goal_state(goal)["add_link"], {"linked_project_id": str(ref_project),
                                             "relationship": "reference"})
    rule_names = [r["name"] for r in goal_state(goal)["rules"]]
    check(rule_names == ["goalui-primary", "goalui-codev"],
          "the Rules section lists the goal's effective set (primary then"
          " co-developed; reference contributes nothing)")

    act(auth, slot_public(auth, "/saddles", "saddles")["create_capability"],
        {"name": "goalui-capability"})
    conn = db()
    capability = conn.execute(
        "select id from capabilities where name = 'goalui-capability'"
    ).fetchone()[0]
    # The early redirect-fixture worker joins the goal (a fanned
    # worker's shape) so its row carries the badge too.
    conn.execute("update tasks set goal_id = ? where id = ?", (goal, worker))
    conn.execute("insert into task_capabilities (task_id, capability_id)"
                 " values (?, ?)", (worker, capability))
    conn.commit()
    conn.close()
    goal_worker = {"id": worker, "caps": None}
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1", "title": "Goalui fanned worker", "priority": "4",
        "description": "A worker the agent fanned out."})
    conn = db()
    fanned = conn.execute(
        "select id from tasks where title = 'Goalui fanned worker'").fetchone()[0]
    conn.execute("update tasks set goal_id = ? where id = ?", (goal, fanned))
    conn.execute("insert into task_capabilities (task_id, capability_id)"
                 " values (?, ?)", (fanned, capability))
    conn.commit()
    conn.close()
    rows = {int(t["id"]): t for t in goal_state(goal)["goal_tasks"]}
    check(rows[fanned].get("capabilities") == ["goalui-capability"]
          and rows[worker].get("capabilities") == ["goalui-capability"],
          "task rows carry their required capability tags")
    goal_page = http_get_page(auth, "/goals/%d" % goal)
    check('stm-for-capability-in="goal_task.capabilities"' in goal_page
          and "badge-capability" in goal_page,
          "the task rows render capability badges")

    print("== B6: narrative weights — events full, comments footnote ==")
    # An AGENT comment projects the narrative's comment event (human
    # comments only thread) — pushed through the entries channel while
    # parked, so the suspended state stays put.
    fake.push_results("suspended", entries=[{
        "id": 9001, "command": "saddle comment", "output": "An agent aside.",
        "exit_code": 0, "type": "comment", "timestamp": "2026-09-12T00:00:00Z"}])
    check(wait_for(lambda: any(e["kind"] == "comment"
                               for e in goal_state(goal)["events"]) or False,
                   timeout=15, what="comment event lands") is not None,
          "an agent comment lands as a narrative event")
    goal_page = http_get_page(auth, "/goals/%d" % goal)
    check("stm-attribute.class=\"'goal-event' + (event.kind == 'comment'"
          " ? ' footnote' : '')\"" in goal_page.replace("\n", " ").replace("  ", " ")
          or "'goal-event' + (event.kind == 'comment' ? ' footnote' : '')" in goal_page,
          "the WHOLE comment event row carries .footnote (badge included)")
    check("'badge badge-goal-' + event.kind" in goal_page,
          "non-comment events keep their full-weight kind badges")
    css = open(os.path.join(REPO, "src/pages/app.css")).read()
    check(".goal-event.footnote" in css and ".footnote {" in css,
          "the stylesheet ships the footnote row weight beside the utility")

    print("== C2: system rows render distinctly on the ONE thread ==")
    run_comments = task_state(auth, run_task)["comments"]
    system_rows = [c for c in run_comments if c.get("is_system")]
    check(system_rows and all(c["author"] == "system" for c in system_rows)
          and not any(c.get("from_agent") for c in system_rows),
          "the harness bookkeeping rows are authored 'system' (never agent)")
    goal_thread = goal_state(goal)["comments"]
    check(any(c.get("is_system") for c in goal_thread),
          "the goal page reads the same thread (the system rows included)")
    css = open(os.path.join(REPO, "src/pages/app.css")).read()
    check(".comment.system" in css and ".system-label" in css,
          "the stylesheet ships the distinct system-event styling")
    page = http_get_page(auth, "/goals/%d" % goal)
    check("comment.is_system" in page and "system-label" in page,
          "the goal page renders system rows with the labelled muted style")
    task_page = http_get_page(auth, "/tasks/%d" % worker)
    check("comment.is_system" in task_page and "system-label" in task_page,
          "the task page renders system rows with the same style")

    print("== S5 U2: mechanical rows filtered from user-facing timelines ==")
    # The daemon's mechanical system entries (the park notice, the
    # restore footnote) ride the wire typed 'system': the raw terminal
    # row STORES (session_commands), but no comment row, narrative
    # event or dashboard timeline row is authored for them.
    open_run = open_session(run_task)
    mechanical = [
        {"id": 9501, "command": "saddle suspend",
         "output": "goal agent parked until an owned task completes or"
                   " fails; resume via /start resume_notice",
         "exit_code": 0, "type": "system", "timestamp": "2026-09-12T00:00:00Z"},
        {"id": 9502, "command": "harness resume",
         "output": "[harness] session context restored — task resumed",
         "exit_code": 0, "type": "system", "timestamp": "2026-09-12T00:00:01Z"},
    ]
    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                           {"session_id": open_run, "state": "suspended",
                            "entries": mechanical})
    check(status == 200, "the mechanical system entries push cleanly")
    conn = db()
    raw_rows = conn.execute(
        "select count(*) from session_commands where session_id = ?"
        " and type = 'system'", (open_run,)).fetchone()[0]
    conn.close()
    check(raw_rows >= 2,
          "the raw terminal keeps the mechanical rows (typed entries store)")
    check(all("parked until" not in c["body"]
              and "session context restored" not in c["body"]
              for c in task_state(auth, run_task)["comments"]),
          "no mechanical comment row lands on the thread (U2)")
    check(all("parked until" not in e["text"]
              and "session context restored" not in e["text"]
              for e in goal_state(goal)["events"]),
          "no mechanical narrative event lands on the goal page (U2)")
    dash_events = slot_public(auth, "/", "dashboard")["activity"]
    check(all("parked until" not in e["text"]
              and "session context restored" not in e["text"]
              for e in dash_events),
          "no mechanical row lands on the dashboard timeline (U2)")

    print("== C4: head line, answered-blocks expander, footnote comments ==")
    state = goal_state(goal)
    check(state["current_activity"] == "parked awaiting its tasks' terminal transitions",
          "a suspended run folds into the one muted head line")
    check(len(state.get("answered_blocks") or []) >= 2
          and len(state.get("open_blocks") or []) == 0,
          "the historic answered blocks expose as their own vector (open ones empty)")
    goal_page = http_get_page(auth, "/goals/%d" % goal)
    check("Answered blocks (" in goal_page and "comment-meta" in goal_page,
          "the answered blocks and the footnote comment meta ship in the markup")
    check("timeless note for a future reader" in goal_page,
          "the composer hint states the timeless-comment rule")
    css = open(os.path.join(REPO, "src/pages/app.css")).read()
    check(".goal-page .card { margin-bottom" in css and ".comment-meta" in css,
          "the stylesheet ships the card rhythm and the comment footnote meta")

    print("== C1: one review block per verdict path (F6) ==")
    # The daemon's honest round pushes the ENTRY and the STATE
    # together — the done pair must raise exactly ONE completion
    # block, and a redelivered pair nothing more.
    fake.goal_script = [("done", "# Completion\n\nOne block only.")]
    conn = db()
    conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
                 " priority, completed_at, created_at) values (1, ?, 'task',"
                 " 'Warmup', 'done', 5,"
                 " strftime('%s','now'),"
                 " strftime('%s','now','-30 seconds'))", (goal,))
    conn.commit()
    conn.close()
    check(wait_for(lambda: goal_state(goal)["status"] == "completion_review" or False,
                   timeout=60, what="goalui completion review opens") is not None,
          "the scripted done round enters completion review")
    def completion_block_count():
        conn = db()
        try:
            return conn.execute(
                "select count(*) from task_blocks where task_id = ?"
                " and type = 'completion'", (run_task,)).fetchone()[0]
        finally:
            conn.close()
    check(completion_block_count() == 1,
          "the done state raises exactly ONE completion block")
    # Plan 16 S5 (U1): the verdict-shaped block renders the rich review
    # form — explicit accept/reject affordances with the free-text
    # feedback path, never a free-text-only verdict.
    goal_page = http_get_page(auth, "/goals/%d" % goal)
    check("Accept goal" in goal_page and "Reject with feedback" in goal_page
          and 'name="feedback"' in goal_page,
          "the completion review presents explicit accept/reject + free-text feedback (U1)")
    # The burst shape: the done ENTRY and the done STATE in one push
    # (fresh entry ids), then the whole pair redelivered — the
    # pending-report idempotency leaves the one row alone.
    conn = db()
    burst_session = conn.execute(
        "select id from sessions where task_id = ? and ended_at is null",
        (run_task,)).fetchone()[0]
    conn.close()
    done_pair = [{"id": 9401, "command": "saddle done",
                  "output": "# Completion\n\nOne block only.", "exit_code": 0,
                  "type": "done", "timestamp": "2026-09-12T00:00:00Z"}]
    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                           {"session_id": burst_session, "state": "done",
                            "entries": done_pair,
                            "summary": "# Completion\n\nOne block only."})
    check(status == 200 and completion_block_count() == 1,
          "the done entry + state pair raises nothing more (one block, ever)")
    state = goal_state(goal)
    act(auth, state["completion_pending_block"]["answer_action"],
        {"verdict": "accept"})
    check(wait_for(lambda: goal_state(goal)["status"] == "complete" or False,
                   timeout=30, what="goalui goal completes") is not None,
          "accepting the completion completes the goal")
    check(completion_block_count() == 1,
          "the accept path leaves exactly one review-kind block behind")
    conn = db()
    stragglers = conn.execute(
        "select count(*) from task_blocks where task_id = ? and answer is null",
        (run_task,)).fetchone()[0]
    conn.close()
    check(stragglers == 0,
          "no unanswered straggler rows survive the completion")

    print("== S2: server-owned verdicts + the container-resolved reject ==")
    # A fresh goal drives the verdict paths the goal page's forms now
    # use: verdict markers for accept, raw `feedback` for reject, and
    # the block resolution living in the review CONTAINER (not
    # goal.plan_pending_block — the diagnosed completion-review bug).
    fake.goal_script = [("plan", "# S2 plan\n\nOne step")]
    act(auth, slot_public(auth, "/goals", "goals")["create"], {
        "project_id": "1", "title": "S2 verdict goal",
        "brief": "Covers the plan-16 S2 verdict paths.",
        "branch_name": "goal/s2-verdicts"})
    conn = db()
    s2_goal = int(conn.execute(
        "select id from goals where title = 'S2 verdict goal'").fetchone()[0])
    conn.close()
    act(auth, goal_state(s2_goal)["start"], {})
    check(wait_for(lambda: goal_state(s2_goal)["status"] == "plan_review" or False,
                   timeout=90, what="S2 goal enters plan review") is not None,
          "the S2 goal enters plan review")
    s2_state = goal_state(s2_goal)
    s2_pending = s2_state["plan_pending_block"]

    # (1) A remark-less, text-less rejection is refused: the action
    # errors and the block keeps waiting.
    refusal = act(auth, s2_pending["answer_action"], {"feedback": "   "})
    check(any(d.get("type") == "notify" and d.get("kind") == "error"
              and "answer is required" in (d.get("message") or "").lower()
              for d in refusal),
          "a remark-less, text-less rejection is refused with an error")
    conn = db()
    s2_still_open = conn.execute(
        "select answer from task_blocks where id = ?",
        (s2_pending["id"],)).fetchone()[0]
    conn.close()
    check(s2_still_open is None,
          "the refused rejection leaves the block unanswered")

    # (3) The goal page template ships no verdict constants and its
    # script resolves the review block from ANY [data-plan-review]
    # container via the open-blocks list (the completion-review fix).
    # Fetched while the review is open so the accept form renders.
    s2_page = http_get_page(auth, "/goals/%d" % s2_goal)
    check("Plan approved. Proceed with implementation." not in s2_page
          and "Review approved. Continue with the task." not in s2_page,
          "the goal page ships no verdict constants (server-owned wording)")
    check('name="verdict" value="accept"' in s2_page,
          "the goal page's accept forms post the verdict marker")
    check("resolveReviewBlock" in s2_page and "goal.open_blocks" in s2_page,
          "the reject flow resolves its block from the open-blocks list (container-keyed)")
    check("statum.act(block.answer_action, { feedback: feedback })" in s2_page,
          "the reject flow posts raw feedback (the server wraps the prefix)")
    check("extraText.length === 0" in s2_page,
          "a remark-less rejection with textarea content is allowed (S2)")

    # (2) The goal page's reject rides `feedback` (the client composes
    # the numbered-remarks body) and the SERVER wraps it with the plan
    # prefix — byte-identical to the historical client-built answer.
    act(auth, s2_pending["answer_action"],
        {"feedback": "> ## One step\n\n1. Tighten it\n\n2. Name the owner"})
    conn = db()
    s2_wrapped = conn.execute(
        "select answer from task_blocks where id = ?",
        (s2_pending["id"],)).fetchone()[0]
    conn.close()
    check(s2_wrapped
          == "Plan feedback received:\n\n"
             "> ## One step\n\n1. Tighten it\n\n2. Name the owner",
          "the plan rejection wraps the client-shaped body with the stable prefix (%r)"
          % s2_wrapped)
    check(goal_state(s2_goal)["status"] == "planning",
          "the feedback rejection returns the agent to planning")