B1-B6: goalui test shard — plan 14 workstream B coverage

A new parallel shard (its own app/daemon/DB) drives one goal through
the supervision surface end to end: the /tasks/<goal-run-id> redirect
(navigate directive; the slot still hydrates) against a worker page
that renders normally; the no-notify contract and the full status-head
field set; yeehaw rows-only. The plan-review remarks queue ships its
markers (queue under the plan body, numbering, popover,
pulse-refresh persistence) and the numbered rejection payload carries
the remarks verbatim in order. The dual-row mirror-sync: a goal-page
answer closes both rows and the pile entry, a mirror-side answer
closes the goal row, historical one-sided rows don't pin.
sessions.resume_notice: fresh rows empty, resumed rows populated, the
dropdown newest-first with [resumed] on exactly the notice-carrying
sessions, session 1's opening-brief panel, the empty state, the 200px
stylesheet. Requeue-to-resume on a dead saddle (session closed, task
queued, fresh re-dispatch with no notice) and Retry supervision
(goal revived, run re-offered, narrative event). The B5 section order
assertion, the effective-rules listing (primary + co-developed, not
reference), per-row capability badges, and the B6 footnote row weight
with full-weight kind badges. 78 checks; suite 678 -> 756.

test/web_stack_test.py

@@ -6264,9 +6264,448 @@ def bundle_replay(ctx):
     # Drain: nothing queued remains for the shard's shutdown.
     check(queued_count() == 0, "the replay shard drains its queue")
 
+def bundle_goalui(ctx):
+    """Plan 14 workstream B: 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, B3 — the
+    dual-row mirror-sync (either side closes the pair; historical
+    one-sided rows don't pin), B4 — sessions.resume_notice + the
+    terminal dropdown/resume panel, B5 — page order, effective rules
+    and per-task capability badges, B6 — narrative footnote weights.
+    """
+    auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token
+
+    def goal_state(goal_id):
+        return slot_public(auth, "/goals/%d" % goal_id, "goal")
+
+    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)")
+
+    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("== B3: either side of the dual row closes the pair ==")
+    check("Which datastore?" in pile_questions(),
+          "the open goal question pins onto the dashboard pile")
+    mirror_rows = [b for b in task_state(auth, run_task)["blocks"]
+                   if b["question"] == "Which datastore?"]
+    check(len(mirror_rows) == 1, "the question mirrors once onto the run task")
+    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()
+    rows = conn.execute(
+        "select (select count(*) from goal_blocks where question = 'Which datastore?'"
+        "        and answer is null),"
+        "       (select count(*) from task_blocks where question = 'Which datastore?'"
+        "        and answer is null)").fetchone()
+    conn.close()
+    check(rows[0] == 0 and rows[1] == 0,
+          "answering from the goal page closes BOTH rows (goal + mirror)")
+    check("Which datastore?" not in pile_questions(),
+          "the answered pair no longer pins the dashboard pile")
+
+    # The mirror-side answer path (a stale deep link firing the task
+    # action API): the goal row must close with it.
+    conn = db()
+    conn.execute("insert into goal_blocks (goal_id, type, question, created_at)"
+                 " values (?, 'question', 'Mirror probe question?',"
+                 " strftime('%s','now'))", (goal,))
+    conn.execute("insert into task_blocks (task_id, type, question, created_at)"
+                 " values (?, 'question', 'Mirror probe question?',"
+                 " strftime('%s','now'))", (run_task,))
+    conn.commit()
+    conn.close()
+    mirror_block = [b for b in task_state(auth, run_task)["blocks"]
+                    if b["question"] == "Mirror probe question?" and not b.get("answered")][0]
+    act(auth, mirror_block["answer_action"], {"answer": "Answered via the mirror"})
+    conn = db()
+    answered = conn.execute(
+        "select count(*) from goal_blocks"
+        " where question = 'Mirror probe question?' and answer is not null"
+    ).fetchone()[0]
+    conn.close()
+    check(answered == 1,
+          "an answer on the mirror closes the goal row too (the API path)")
+
+    # Historical one-sided rows (the observed reference-DB shape):
+    # goal row open, its mirror already answered — the pile must not
+    # show it forever.
+    conn = db()
+    conn.execute("insert into goal_blocks (goal_id, type, question, created_at)"
+                 " values (?, 'question', 'Legacy one-sided?',"
+                 " strftime('%s','now'))", (goal,))
+    conn.execute("insert into task_blocks (task_id, type, question, answer,"
+                 " answered_at, created_at)"
+                 " values (?, 'question', 'Legacy one-sided?', 'Resolved long ago',"
+                 " strftime('%s','now'), strftime('%s','now'))", (run_task,))
+    conn.commit()
+    conn.close()
+    check("Legacy one-sided?" not in pile_questions(),
+          "a historical one-sided row does not render in the pile")
+
+    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 B3 mirror 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()
+    state = goal_state(goal)
+    check(state.get("requeue_run") 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)
+    act(auth, state["requeue_run"], {})
+    check(task_state(auth, run_task)["status"] == "queued",
+          "the requeue returns the run to the queue")
+    check(open_session(run_task) is 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")
+    state = goal_state(goal)
+    check(state.get("retry_run") is not None,
+          "a failed goal offers Retry supervision")
+    fake.goal_script = [("suspend",)]
+    act(auth, state["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"] == "retry" for e in state["events"]),
+          "the narrative records the supervision retry")
+
+    print("== B5: order, effective rules, capability badges ==")
+    goal_page = http_get_page(auth, "/goals/%d" % goal)
+    markers = [
+        ('class="card goal-head"', "status head"),
+        ("Agent progress", "progress"),
+        ("Current activity", "activity"),
+        ("Goal agent blocks", "attention blocks"),
+        ("<h3>Tasks</h3>", "tasks"),
+        ("Initial brief", "brief"),
+        ("Involved projects", "projects"),
+        ("<h3>Rules</h3>", "rules"),
+        ("<h3>Plan</h3>", "plan"),
+        ("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 → attention → tasks → brief → rules → plan →"
+          " narrative → comments → runs (in order)")
+
+    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")
+
+
 BUNDLES_ORDER = ("pages", "dispatch", "subtasks", "lineage",
                  "allocation", "lifecycle", "answers", "push", "goals",
-                 "replay")
+                 "replay", "goalui")
 
 
 def run_bundle(name):