"""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 on the task page's session rows + the
    plan-17 event log page, 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 plant_live_worker(goal_id, title="Goalui live worker"):
        """Goal 27 refinements — the parking guard: the goal agent parks
        only while its goal owns a task that could still fire a
        terminal transition. The scenes that script it to park plant a
        BLOCKED live worker first (wakeable, never dispatched)."""
        conn = db()
        conn.execute("insert into tasks (project_id, goal_id, kind, title,"
                     " status, priority, mode, creator, created_at)"
                     " values (1, ?, 'task', ?, 'blocked', 5, 'implement',"
                     " 'goal', strftime('%s','now'))", (goal_id, title))
        conn.commit()
        conn.close()

    def open_session(task_id):
        """The task's newest session row (plan 16 D4: a parked round's
        row is closed 'parked' — still the record the CLI callbacks,
        the telemetry pushes and the page surfaces key on)."""
        conn = db()
        try:
            row = conn.execute(
                "select max(id) from sessions where task_id = ?",
                (task_id,)).fetchone()
            return int(row[0]) if row and row[0] 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 "badge-yeehaw" not in goal_page,
          "no yeehaw badge remains anywhere (the feature died with the entry stream)")

    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)
    # Goal 41 T6 ("actions surface"): a pending plan review LEADS the
    # acting column as its own card — before the status head, the
    # blocks and the (history-only) plan card.
    check(goal_page.index("Plan review") < goal_page.index('class="card goal-head')
          < goal_page.index("Goal agent blocks") < goal_page.index("<h3>Plan</h3>"),
          "a pending plan review leads the acting column (status, blocks, history after)")
    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"):
        check(marker in goal_page, "the remarks queue ships %s" % marker)
    # The queue ENGINE moved out of the inline page script into the
    # shared app.js (main.html's <script pstm-res-src="app.js">, the
    # theming commits): the page keeps the review markup and the
    # wiring, the engine keeps the machinery. The engine is asserted
    # on the SERVED bundle the page loads (the pages.py statum.js
    # pattern); main.html ships it ahead of the page's own script,
    # so the wiring below cannot run before the engine exists.
    status, app_js = http("GET", "/_statum/resource/app.js", timeout=15)
    check(status == 200 and "window.initPlanReviews" in app_js,
          "the served app.js ships the shared initPlanReviews engine")
    check("window.initPlanReviews" in goal_page,
          "the goal page references the shared engine (its review rides it)")
    for marker in ("data-plan-remove", "plan-popover", "remarksByBlock"):
        check(marker in app_js, "the remarks queue ships %s" % marker)
    check("(index + 1) + '. '" in app_js,
          "the rejection payload numbers every queued remark")
    check("rejectOpenByBlock" in app_js and "statum.afterEvaluate(syncPlanRemarks)" in app_js,
          "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: resume notices on the session rows + the event log page ==")
    plant_live_worker(goal)
    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")
    sessions = task_state(auth, run_task)["sessions"]
    ids = [int(s["id"]) for s in sessions]
    check(ids == sorted(ids, reverse=True),
          "the task page's session history lists newest-first")
    # 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()
    notices_by_id = {int(s["id"]): s.get("notice") or "" for s in sessions}
    check(set(i for i in ids if notices_by_id[i]) == set(resumed_ids),
          "the session rows' notice column rides exactly the notice-carrying sessions")
    check(any("Plan feedback received" in notices_by_id[i] for i in resumed_ids),
          "a resumed session's row shows the stored resume notice")
    log_state = slot_public(auth, "/tasks/%d/log" % run_task, "task_log")
    check(log_state.get("exists") is True and int(log_state["task_id"]) == run_task
          and log_state["task_url"] == "/tasks/%d" % run_task,
          "the event log page hydrates the run-task header")
    log_page = http_get_page(auth, "/tasks/%d/log" % run_task)
    # The SSE client ships as the compiled task_log.js resource (plan 17
    # b3+): the page references it and the resource carries the code.
    status, log_js = http("GET", "/_statum/resource/task_log.js", timeout=15)
    check(status == 200 and "task_log.js" in log_page
          and "EventSource" in log_js and "events/stream" in log_js
          and "data-log-feed" in log_page,
          "the log page subscribes to the ledger's SSE stream")
    check(goal_state(goal)["runs"][0]["terminal_url"] == "/tasks/%d/log" % run_task,
          "the goal runs table links the event log")

    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")
    conn = db()
    before_requeue = conn.execute(
        "select id, outcome from sessions where task_id = ?"
        " order by id desc limit 1", (run_task,)).fetchone()
    conn.close()
    check(tuple(before_requeue) == (open_session(run_task), "parked"),
          "the parked round's newest session row closed 'parked' (D4)")

    def parked_session_ended():
        # D4: the parked row is ALREADY closed at its park — the pin
        # keeps guarding the restore-on-dispatch successor shape (the
        # requeue may not resurrect the OLD row; a fresh round opens
        # its own).
        conn = db()
        try:
            return conn.execute(
                "select ended_at is not null from sessions where id = ?",
                (before_requeue[0],)).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 round's own row must stay closed (the
    # restore-on-dispatch successor opens a FRESH session row — the
    # check keys the old row, never the task's transient open window).
    check(wait_for(parked_session_ended, timeout=60,
                   what="parked session closed") is not None,
          "the requeue leaves the parked session row closed (fresh round, fresh row)")
    conn = db()
    conn.execute("update saddles set status = 'idle' where id = ?", (fake_id,))
    conn.commit()
    conn.close()
    fake.goal_script = [("suspend",)]
    plant_live_worker(goal, "Goalui live worker two")
    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",)]
    plant_live_worker(goal, "Goalui live worker three")
    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 + goal 41 T6: act|consult layout, rules, badges ==")
    goal_page = http_get_page(auth, "/goals/%d" % goal)
    # Goal 41 T6 (ux-notes §5): the state strip leads, then the anchor
    # sub-nav, then the Act|Consult two-column layout — the acting
    # surfaces (status, attention, plan review, comments) in one
    # column, the consulting ones (tasks, brief, projects, rules,
    # narrative, runs) in the other, stacking act-first on narrow
    # screens (DOM order IS the stack order).
    check(goal_page.index('class="goal-state-strip"')
          < goal_page.index('class="subnav"')
          < goal_page.index("data-goal-act")
          < goal_page.index("data-goal-consult"),
          "the goal page reads strip → sub-nav → act column → consult column")
    # mkpstm inlines pstm-fragment bodies at compile time, so the
    # include directive itself never serves; the adopted breadcrumb
    # (its ol/li markup) and the retired back anchor are the truth.
    check('class="breadcrumb"' in goal_page
          and 'aria-label="Breadcrumb"' in goal_page
          and "&larr; Goals" not in goal_page,
          "the shared breadcrumb replaces the hand-drawn back anchor")
    check(goal_state(goal)["crumbs"] == [
        {"href": "/", "label": "Dashboard"},
        {"href": "/goals", "label": "Goals"},
        {"href": None, "label": "#%d" % goal}],
        "the entrypoint publishes the Dashboard → Goals → #id crumb trail")
    act_markers = [
        ('id="goal-sec-plan-review"', "plan review (when pending)"),
        ('id="goal-sec-completion-review"', "completion review (when pending)"),
        ('class="card goal-head"', "status head"),
        ("Agent progress", "progress"),
        ("Goal agent blocks", "attention blocks"),
        ("<h3>Plan</h3>", "plan"),
        ("<h3>Comments</h3>", "comments"),
    ]
    consult_markers = [
        ("<h3>Tasks</h3>", "tasks"),
        ("Initial brief", "brief"),
        ("Involved projects", "projects"),
        ("<h3>Rules</h3>", "rules"),
        ("Supervision narrative", "narrative"),
        ("Goal agent runs", "runs"),
    ]
    for markers, column in ((act_markers, "act"), (consult_markers, "consult")):
        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 %s column reads %s (in order)" % (column,
              " → ".join(_ for _, __ in ((m, l) for m, l in markers))))
    act_at = goal_page.index("data-goal-act")
    consult_at = goal_page.index("data-goal-consult")
    check(all(act_at < goal_page.index(m) < consult_at
              for m, _ in act_markers),
          "every acting surface sits inside the act column")
    check(goal_page.index("data-goal-consult")
          < min(goal_page.index(m) for m, _ in consult_markers),
          "every consulting surface sits inside the consult column")
    # The strip: one glance — status badge, the open-item verdict, the
    # primary lifecycle button; the Status card keeps the detail.
    strip_at = goal_page.index("data-goal-state-strip")
    check(goal_page.index("'badge badge-' + goal.status", strip_at) > strip_at,
          "the strip leads with the status badge")
    for marker in ("goal.plan_pending_block", "goal.completion_pending",
                   "goal.open_blocks.length"):
        check(marker in goal_page[strip_at:strip_at + 2500],
              "the strip's verdict is computed from %s" % marker)
    check(goal_page.count("stm-action=\"goal.start\"") >= 2,
          "the primary lifecycle button rides the strip (and the status ladder)")
    check("goalHighlightSubnav" in goal_page
          and "statum.afterEvaluate(goalHighlightSubnav)" in goal_page,
          "the anchor sub-nav rebinds its highlight across statum re-renders")
    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 remark rides the ledger as a comment event (human
    # comments only thread) — pushed through the events channel while
    # parked, so the suspended state stays put.
    fake.push_events([{"kind": "comment", "actor": "model",
                       "payload": {"author": "agent",
                                   "body": "An agent aside."}}])
    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 remarks (the park notice, the
    # restore footnote) ride the ledger as harness comment events: the
    # event row STORES, but no comment row, narrative event or
    # dashboard timeline row renders for them.
    open_run = open_session(run_task)
    mechanical = [
        {"seq": 9501, "kind": "comment", "actor": "harness",
         "payload": {"author": "system",
                     "body": "goal agent parked until an owned task"
                             " completes or fails"}},
        {"seq": 9502, "kind": "comment", "actor": "harness",
         "payload": {"author": "system",
                     "body": "[harness] session context restored — task resumed"}},
    ]
    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
                              {"session_id": open_run, "events": mechanical})
    check(status == 200 and int(body.get("stored") or 0) == 2,
          "the mechanical harness remarks store on the ledger")
    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)
    # (Goal 27 refinements: the parking guard can hand the run a
    # refusal resume between rounds — the parked head line is the
    # settled state, not a transient working blip.)
    check(wait_for(lambda: goal_state(goal)["current_activity"]
                   == "parked awaiting its tasks' terminal transitions"
                   or False, timeout=60, what="run settles parked")
          is not None,
          "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 STATE redelivered under the run's
    # newest session (the entries channel is gone — the report rides
    # the summary) — the pending-report idempotency leaves the one
    # row alone.
    conn = db()
    burst_session = conn.execute(
        "select max(id) from sessions where task_id = ?",
        (run_task,)).fetchone()[0]
    conn.close()
    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                           {"session_id": burst_session, "state": "done",
                            "summary": "# Completion\n\nOne block only."})
    check(status == 200 and completion_block_count() == 1,
          "a redelivered done push 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);
    # the reject flow itself lives in the shared app.js engine
    # (app_js fetched in B2 above).
    # 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 app_js,
          "the reject flow posts raw feedback (the server wraps the prefix)")
    check("extraText.length === 0" in app_js,
          "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")


    # =====================================================================
    # Task 12 (goal half): the draft brief-card editor.
    # =====================================================================
    print("== task 12: the draft goal brief editor ==")
    # A fresh DRAFT goal (the fixtures above are all started): the
    # brief card carries the in-place editor with title/brief/priority
    # prefilled from state, goal.update saves through the sealed
    # action, and START removes the editor (the brief locks).
    act(auth, slot_public(auth, "/goals", "goals")["create"], {
        "project_id": "1", "title": "Editable draft goal",
        "brief": "The original brief.",
        "branch_name": "goal/editable-draft"})
    conn = db()
    edit_goal = conn.execute(
        "select id from goals where title = 'Editable draft goal'"
    ).fetchone()[0]
    conn.close()

    draft = goal_state(edit_goal)
    check("update" in draft,
          "a draft goal's state carries the update action (the editor's save)")
    goal_page = http_get_page(auth, "/goals/%d" % edit_goal)
    check('stm-if="goal.update != null"' in goal_page
          and 'data-lf-toggle>Edit</button>' in goal_page,
          "the brief card renders the draft-only Edit button")
    check('data-lf-src="goal.brief"' in goal_page
          and 'data-lf-src="goal.title"' in goal_page
          and 'data-lf-src="goal.priority"' in goal_page,
          "the editor prefills title, brief and priority from state"
          " (stm-attribute.value cannot carry a textarea)")

    act(auth, draft["update"], {"title": "Editable draft goal (edited)",
                                "brief": "The edited brief.",
                                "priority": "7"})
    saved = goal_state(edit_goal)
    check(saved["title"] == "Editable draft goal (edited)"
          and saved["brief"] == "The edited brief."
          and saved["priority"] == 7,
          "the draft edit round-trips title, brief and priority"
          " (draft priority is finally changeable)")

    # The draft-only guard is server-side: start the goal, then the
    # update action is gone from state (the editor with it).
    act(auth, saved["start"], {})
    check(wait_for(lambda: goal_state(edit_goal)["status"] != "draft"
                   or False, what="editable draft goal starts"),
          "the edited draft goal starts")
    check("update" not in goal_state(edit_goal),
          "a started goal's state carries no update action (the brief locked)")

    # The server-side guard: replaying the (still-valid) sealed
    # update action against the now-started goal must refuse — the
    # editor's disappearance is UI, the refusal is enforcement.
    replay = act(auth, draft["update"], {"brief": "Sneaked brief."})
    check(any(d.get("type") == "notify" and d.get("kind") == "error"
              for d in replay),
          "the draft-only guard refuses an edit replay against a started goal")
    check(goal_state(edit_goal)["brief"] == "The edited brief.",
          "the refused replay left the brief unchanged")

    print("== goal 4 t1: force-wake action stays unoffered unless parked ==")
    # The page template ships the gated button markup statically (the
    # stm-if hides it client-side), so absence is asserted on the SLOT
    # STATE: the action is not AUTHORED. The offered pair (parked run)
    # is covered by the goals bundle, which owns a parked run.
    check(goal_state(edit_goal).get("force_wake") is None,
          "a non-parked (planning) goal state offers no force-wake action")
    conn = db()
    any_goal = conn.execute(
        "select id from goals where status = 'draft' order by id limit 1").fetchone()
    plain = conn.execute(
        "select id from tasks where status != 'suspended'"
        " and kind = 'task' order by id limit 1").fetchone()
    conn.close()
    if any_goal:
        check(goal_state(int(any_goal[0])).get("force_wake") is None,
              "a draft goal state offers no force-wake action")
    if plain:
        check(task_state(auth, int(plain[0])).get("force_wake") is None,
              "a non-suspended task state offers no force-wake action")