clanker-saddle/test/bundles/goals.py

"""The goals bundle (moved verbatim from the pre-split web_stack_test.py;
see test/harness.py for the shared machinery and the runner)."""
from harness import *  # noqa: F401,F403 — the shared wire helpers


# =============================================================================
# Bundle: goals — the plan-12 goal lifecycle: creation with branch-name
# uniqueness, the start → planning → plan review → active flow with
# revisions, the goal agent's task fan-out through the task CLI
# callbacks (ownership + creator-only enqueue), the event-driven
# coalesced resume, pause/resume queueing, the completion review with
# branch push state, the cancel cascade, and the human tasks'
# auto-enqueue-after predecessor.
# =============================================================================

def run(ctx):
    auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token
    app = ctx.app

    def create_goal(fields):
        act(auth, slot_public(auth, "/goals", "goals")["create"], fields)
        rows = [g for g in slot_public(auth, "/goals", "goals")["goals"]
                if g["title"] == fields["title"]]
        return int(rows[0]["id"])

    print("== plan 12: goal creation, validation, lifecycle actions ==")
    goal1 = create_goal({"project_id": "1", "title": "Ship the flange",
                         "brief": "Make the flange module production-grade.",
                         "branch_name": "goal/flange", "priority": "3"})
    state = goal_state(goal1)
    check(state["status"] == "draft" and state["display"] == "draft"
          and state["branch_name"] == "goal/flange"
          and state["base_branch"] == "main",
          "a goal lands in draft with its branch pair")
    check(state.get("start") is not None and state.get("cancel") is not None
          and state.get("pause") is None,
          "draft offers start + cancel only")
    check(state["run_task_id"] == 0 and state["comments"] == [],
          "a draft goal has no thread yet (the run task owns it)")
    draft_page = http_get_page(auth, "/goals/%d" % goal1)
    check('stm-if="goal.run_task_id > 0" stm-action="goal.comment"' in draft_page
          and "comments open with its run" in draft_page,
          "the draft goal page gates the composer on the run (muted note until then)")
    # Branch-name uniqueness: same name, same involved project → refuse.
    duplicate_directives = act(auth, slot_public(auth, "/goals", "goals")["create"], {
        "project_id": "1", "title": "Clash", "brief": "b", "branch_name": "goal/flange"})
    check(any(d.get("type") == "notify" and d.get("kind") == "error"
              and "goal/flange" in (d.get("message") or "")
              for d in duplicate_directives),
          "a live goal's branch name cannot be reused in an involved project (%s)"
          % [d.get("message") for d in duplicate_directives if d.get("type") == "notify"])
    conn = db()
    conn.execute("insert into projects (name, git_url, default_branch, created_at)"
                 " values ('other-repo', 'https://example.com/other.git', 'main',"
                 " strftime('%s','now'))")
    conn.commit()
    conn.close()
    goal_elsewhere = create_goal({"project_id": "2", "title": "Elsewhere",
                                  "brief": "b", "branch_name": "goal/flange"})
    check(goal_elsewhere > 0,
          "the same branch name in a disjoint project is fine (one per repo)")
    # Brief validation.
    brief_directives = act(auth, slot_public(auth, "/goals", "goals")["create"], {
        "project_id": "1", "title": "No brief", "brief": "  ", "branch_name": "goal/x1"})
    check(any(d.get("type") == "notify" and d.get("kind") == "error"
              and "brief" in (d.get("message") or "").lower()
              for d in brief_directives),
          "an empty brief rejects")

    # Start: planning + the goal-run internal task.
    act(auth, state["start"], {})
    state = goal_state(goal1)
    check(state["status"] == "planning" and state.get("start") is None,
          "start moves the goal to planning")
    conn = db()
    run_row = conn.execute(
        "select id, kind, mode, creator, goal_id, priority, status from tasks"
        " where goal_id = ? and kind = 'goal_run'", (goal1,)).fetchone()
    conn.close()
    check(run_row is not None and run_row[1] == "goal_run"
          and run_row[2] == "orchestrate" and run_row[3] == "goal"
          and run_row[4] == goal1 and run_row[5] == 3
          and run_row[6] in ("queued", "assigned", "working"),
          "start creates the goal-run task (orchestrate, priority 3, creator goal,"
          " queued)")
    run_task = int(run_row[0])
    check(goal_state(goal1)["run_task_id"] == run_task,
          "the goal state carries the run task id")
    check(any(e["kind"] == "started" for e in goal_state(goal1)["events"]),
          "the narrative records the start")

    print("== feedback: goal-agent runs are hidden from every task listing ==")
    # The internal run task never lists anywhere tasks are browsed —
    # only the goal page exposes it (worker tasks stay listed). The
    # plan script is armed BEFORE the first dispatch so this section's
    # waits never consume the round the plan-review flow below needs.
    fake.goal_script = [("plan", "# Plan v1\n\n1. Add tests\n2. Ship")]
    check(wait_for(lambda: task_state(auth, run_task)["status"]
                   in ("assigned", "working") or False,
                   timeout=60, what="run task dispatches") is not None,
          "the run task reaches its saddle")
    queue_rows = slot_public(auth, "/tasks", "tasks")["tasks"]
    check(all(int(t["id"]) != run_task for t in queue_rows),
          "the tasks queue never lists the goal-agent run task")
    check(all(t["kind"] != "goal_run" for t in queue_rows),
          "no goal_run rows leak into the queue at all")
    project_state = slot_public(auth, "/projects/1", "project")
    check(all(int(t["id"]) != run_task for t in project_state["tasks"]),
          "the project page's task list excludes the run task")
    dashboard = slot_public(auth, "/", "dashboard")
    check(all(int(a["id"]) != run_task for a in dashboard["active_tasks"]),
          "the dashboard's active tasks exclude the run task")
    check(all(int(i["task_id"]) != run_task for i in dashboard["activity"]),
          "the activity timeline never references the run task")
    check(wait_for(lambda: int(next(
              (s for s in slot_public(auth, "/", "dashboard")["saddles"]
               if s["name"] == "fake-saddle"), {})
          .get("current_goal_id") or 0) == goal1 or False,
          timeout=30, what="saddle card links the goal") is not None,
          "a saddle hosting the goal agent reads as supervising the goal")
    fake_dash_card = next(s for s in slot_public(auth, "/", "dashboard")["saddles"]
                          if s["name"] == "fake-saddle")
    check(fake_dash_card["current_goal_title"] == "Ship the flange",
          "the supervision link carries the goal title")

    print("== plan 12: the goal agent plans, plan review, revisions ==")
    # (The plan script was armed before the run's first dispatch — see
    # the listing-hiding section above — so the first round submits it.)
    check(wait_for(lambda: goal_state(goal1)["status"] == "plan_review" or False,
                   timeout=120, what="goal agent submits its plan") is not None,
          "the goal agent's plan push enters plan review")
    state = goal_state(goal1)
    check(task_state(auth, run_task)["status"] == "blocked",
          "the goal-run task parks on the plan review")
    # U3's plan-review counterpart: the run-task card never reads as a
    # bare "blocked" while a review-shaped block parks it.
    check(task_state(auth, run_task)["display"] == "awaiting your review",
          "the run-task card reads 'awaiting your review' during the plan review (U3)")
    pending = state.get("plan_pending_block")
    check(pending and "# Plan v1" in (pending.get("body") or ""),
          "the pending plan review carries the submitted markdown")
    check(any(b["block_type"] == "plan" and b["is_plan_review"]
              for b in state["blocks"]),
          "the goal blocks list carries the pending plan block")

    print("== feedback: goal visual identity ==")
    # A goal never reads as a task at a glance: the nav carries the
    # plain Goals entry (5eaab31 dropped the nav-goal accent), the
    # list page is iconed, and the dashboard pile's goal cards carry
    # their own accent styling.
    home_html = http_get_page(auth, "/")
    check('<a href="/goals">Goals</a>' in home_html,
          "the nav carries the Goals entry beside Tasks")
    check(any(p["item_type"] == "goal" for p in
              slot_public(auth, "/", "dashboard")["pinned"]),
          "the goal's pending block pins onto the dashboard pile")
    check("'card goal-card'" in home_html and 'badge-kind-goal' in home_html,
          "dashboard goal cards carry the distinct goal styling")
    goals_html = http_get_page(auth, "/goals")
    check("&#127919;" in goals_html and "goal-icon" in goals_html,
          "the goals list page carries the goal icon")
    goal_page_html = http_get_page(auth, "/goals/%d" % goal1)
    check('class="card goal-head"' in goal_page_html and "goal-page" in goal_page_html,
          "the goal page carries the goal-accented identity chrome")
    # Reject with feedback: the agent revises.
    fake.goal_script = [("plan", "# Plan v2\n\n1. Add tests\n2. Add docs\n3. Ship")]
    reject_resume_index = len(fake.goal_assignments)
    act(auth, pending["answer_action"], {"answer": "Add documentation please"})
    check(wait_for(lambda: goal_state(goal1)["status"] == "plan_review" or False,
                   timeout=60, what="revised plan enters review") is not None,
          "feedback sends the agent back to planning; the revision re-enters review")
    # The rejection resume carried NO context_update (nothing in the
    # instruction context had changed).
    check(all("context_update" not in p
              for p in fake.goal_assignments[reject_resume_index:]),
          "a resume without context changes carries no context_update")
    state = goal_state(goal1)
    revisions = state["plan_revisions"]
    # Newest-first since the ordering sweep (73dcec9): index 1 is the
    # older rejected revision, index 0 the pending re-submission.
    check(len(revisions) == 2
          and revisions[1]["answered"] and "Add documentation please" in revisions[1]["answer"]
          and not revisions[0]["answered"],
          "rejected revisions stay as answered history (historical-blocks data)")
    # Plan 16 S1 pin: goals.plan is accept-gated — the pending revision
    # lives only in its block; nothing lands on the goal before accept.
    conn = db()
    plan_column = conn.execute(
        "select plan from goals where id = ?", (goal1,)).fetchone()[0]
    conn.close()
    check((plan_column is None or plan_column == "")
          and (state["plan"] or "") == "",
          "goals.plan stays empty while the revision is only pending (accept-gated, S1)")
    # Approve: goal.plan = v2, status active, agent resumes and parks.
    # Plan 16 S2: the accept posts the verdict MARKER — the server
    # resolves the goal-plan kind and owns the constant.
    fake.goal_script = [("suspend",)]
    pending = state["plan_pending_block"]
    act(auth, pending["answer_action"], {"verdict": "accept"})
    check(wait_for(lambda: goal_state(goal1)["status"] == "active" or False,
                   timeout=60, what="approved plan activates the goal") is not None,
          "approving the plan activates the goal")
    state = goal_state(goal1)
    check(state["plan"] == "# Plan v2\n\n1. Add tests\n2. Add docs\n3. Ship",
          "the approved plan markdown is stored on the goal (the accept verdict's write)")
    # The approval resume carries the plan delta as a context_update
    # turn (the prompt-cache-friendly replacement for the daemon's
    # old system-prompt refresh).
    check(wait_for(lambda: len(fake.goal_assignments) > 0 and
                   "context_update" in fake.goal_assignments[-1] or False,
                   timeout=30, what="approval resume carries context_update") is not None,
          "the plan-approval resume carries a context_update turn")
    approval_update = fake.goal_assignments[-1]["context_update"]
    check("Plan approved" in approval_update and "# Plan v2" in approval_update,
          "the context_update summarises the approved plan markdown")
    check(wait_for(lambda: task_state(auth, run_task)["status"] == "suspended"
                   or False, timeout=60, what="goal agent parks after approval") is not None,
          "the resumed goal agent parks (saddle suspend)")

    print("== plan 12: task fan-out via the task CLI callbacks ==")
    # The goal agent (parked on the plan review — D4: its session row
    # closed 'parked', still the record the CLI keys on) creates,
    # enqueues and deletes its tasks through /api/saddle/task/*.
    conn = db()
    agent_session = conn.execute(
        "select max(id) from sessions where task_id = ?",
        (run_task,)).fetchone()[0]
    conn.close()
    status, body = saddle_api("POST", "/api/saddle/task/create", fake_token, {
        "session_id": agent_session, "title": "Flange tests",
        "mandate": "Write the test suite for the flange module.",
        "mode": "test"})
    check(status == 200 and body.get("ok") is True,
          "task create under the goal succeeds (%s)" % body.get("error"))
    fan_task = int(body["task_id"])
    detail = task_state(auth, fan_task)
    check(detail["status"] == "draft" and detail["mode"] == "test"
          and detail["creator"] == "goal" and detail.get("goal_id") == goal1,
          "the created task is a draft goal task with its mode and creator")
    check(detail["task_branch"] == "task/%d-%d" % (goal1, fan_task)
          and detail["branch_from"] == "goal/flange",
          "goal worker tasks get ephemeral branches cut from the goal branch")
    check(any(e["kind"] == "task-created" and int(e["task_id"]) == fan_task
              for e in goal_state(goal1)["events"]),
          "the narrative records the agent's task creation")

    # Draft until the agent enqueues; humans may not (creator-only).
    check(detail.get("enqueue") is None,
          "the human Enqueue action is not offered for goal tasks (creator-only)")
    status, body = saddle_api("POST", "/api/saddle/task/enqueue", fake_token, {
        "session_id": agent_session, "task_id": fan_task})
    check(status == 200 and task_state(auth, fan_task)["status"] == "queued",
          "task enqueue by the goal agent queues the draft")
    # Ownership + shape errors, verbatim for the CLI.
    status, body = saddle_api("POST", "/api/saddle/task/enqueue", fake_token, {
        "session_id": agent_session, "task_id": 999999})
    check(status == 400 and "No task exists" in (body.get("error") or ""),
          "enqueue of a missing task errors verbatim")
    create_queued(auth, {"project_id": "1", "title": "Innocent bystander", "priority": "4"})
    bystander = task_id_of("Innocent bystander")
    status, body = saddle_api("POST", "/api/saddle/task/delete", fake_token, {
        "session_id": agent_session, "task_id": bystander})
    check(status == 400 and "does not belong to your goal" in (body.get("error") or ""),
          "the goal agent cannot touch tasks outside its goal")
    status, body = saddle_api("POST", "/api/saddle/task/create", fake_token, {
        "session_id": agent_session, "title": "Bad mode", "mandate": "m", "mode": "party"})
    check(status == 400 and "valid modes" in (body.get("error") or ""),
          "an unknown mode rejects with the valid list")
    status, body = saddle_api("POST", "/api/saddle/task/create", fake_token, {
        "session_id": agent_session, "title": "Bad tag", "mandate": "m",
        "capabilities": ["warp-drive"]})
    check(status == 400 and "warp-drive" in (body.get("error") or "")
          and "valid tags" in (body.get("error") or ""),
          "an unknown capability tag rejects naming the valid set")
    # A plain task's session is not a goal-agent identity.
    status, body = saddle_api("POST", "/api/saddle/task/create", fake_token,
                              {"session_id": 999999, "title": "x", "mandate": "m"})
    check(status == 400 and "session" in (body.get("error") or ""),
          "an unknown session_id errors verbatim")
    # Delete: draft + owned.
    status, body = saddle_api("POST", "/api/saddle/task/create", fake_token, {
        "session_id": agent_session, "title": "Scratch task", "mandate": "m"})
    scratch = int(body["task_id"])
    status, body = saddle_api("POST", "/api/saddle/task/delete", fake_token, {
        "session_id": agent_session, "task_id": scratch})
    check(status == 200 and not any(
        t["id"] == scratch for t in slot_public(auth, "/tasks", "tasks")["tasks"]),
        "task delete removes the owned draft")
    status, body = saddle_api("POST", "/api/saddle/task/delete", fake_token, {
        "session_id": agent_session, "task_id": fan_task})
    check(status == 400 and "draft" in (body.get("error") or ""),
          "deleting a queued task refuses (draft only)")

    print("== plan 12: goal context in the worker assignment + coalesced resume ==")
    # The queued fan-out task dispatches to the daemon with the goal
    # reference + ephemeral branch; its completion schedules the parked
    # goal agent's resume (>= SADDLE_GOAL_RESUME_S after the event).
    check(wait_for(lambda: any(int(p.get("task", {}).get("id") or 0) == fan_task
                               for p in fake.start_payloads) or False,
                   timeout=60, what="fan-out task dispatches") is not None,
          "the goal worker task dispatches")
    fan_payload = [p for p in fake.start_payloads
                   if int(p.get("task", {}).get("id") or 0) == fan_task][-1]
    goal_ctx = fan_payload.get("goal") or {}
    check(goal_ctx.get("id") == goal1 and goal_ctx.get("branch") == "goal/flange"
          and goal_ctx.get("base_branch") == "main"
          and "flange module" in goal_ctx.get("brief", "")
          and "# Plan v2" in goal_ctx.get("plan", ""),
          "the worker assignment carries the goal context (read-only brief + plan)")
    check(fan_payload.get("ephemeral_branch") == "task/%d-%d" % (goal1, fan_task)
          and fan_payload.get("kind") == "task" and fan_payload.get("mode") == "test"
          and fan_payload.get("creator") == "goal",
          "the worker assignment carries the ephemeral branch + identity triplet")
    check(wait_task_done(auth, fan_task, timeout=60) is not None,
          "the goal worker task completes")
    check(any(e["kind"] == "task-done" and int(e["task_id"]) == fan_task
              for e in goal_state(goal1)["events"]),
          "the narrative records the task's completion")
    goal_runs_before = len(fake.goal_assignments)
    check(wait_for(lambda: len(fake.goal_assignments) > goal_runs_before or False,
                   timeout=30, what="parked goal agent resumes") is not None,
          "the task-terminal event resumes the parked goal agent")
    resume_payload = fake.goal_assignments[-1]
    notice = resume_payload.get("resume_notice") or ""
    check("Task statuses" in notice and "Flange tests" in notice
          and "done" in notice,
          "the resume notice batches ALL plan task statuses")
    check("New completion reports" in notice and "fake daemon finished" in notice,
          "the resume notice carries the new completion report")
    check("context_update" not in resume_payload,
          "a later resume with no context changes omits context_update entirely")
    check(resume_payload.get("kind") == "goal_run"
          and resume_payload.get("mode") == "orchestrate",
          "the resumed goal agent carries its identity triplet")
    # The resumed agent parked again (script): two events at once must
    # coalesce into ONE resume within the window.
    check(wait_for(lambda: task_state(auth, run_task)["status"] == "suspended"
                   or False, timeout=60, what="goal agent re-parks") is not None,
          "the resumed goal agent parks again")

    print("== plan 16 S3: delta resume notices — one-line statuses, bodies once ==")
    # A MULTI-LINE completion lands while the agent is parked. The next
    # notice lists every status ONE-LINE (first line of the summary
    # only) and embeds each NEW completion's full body exactly once —
    # no statuses-section/reports-section double embedding. (Timestamps
    # are second-granular: each fixture waits out the park's second so
    # completed_at is strictly newer than suspended_at.)
    def run_park_second_passed():
        conn = db()
        try:
            row = conn.execute("select suspended_at from tasks where id = ?",
                               (run_task,)).fetchone()
            if not row or not row[0]:
                return False
            return time.time() > float(row[0]) + 1.05
        finally:
            conn.close()

    check(wait_for(run_park_second_passed, timeout=30,
                   what="S3 park second passes") is not None,
          "the park stamp is in a strictly older second")
    conn = db()
    conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
                 " priority, summary, completed_at, created_at) values (1, ?, 'task',"
                 " 'Multiline report', 'done', 5,"
                 " 'Headline: the flange calibrates\n\nBody: the full report with all"
                 " the detail lines.',"
                 " strftime('%s','now'),"
                 " strftime('%s','now','-30 seconds'))", (goal1,))
    conn.commit()
    conn.close()
    s3_before = len(fake.goal_assignments)
    check(wait_for(lambda: len(fake.goal_assignments) > s3_before or False,
                   timeout=60, what="S3 delta resume fires") is not None,
          "the multiline completion resumes the parked agent")
    s3_notice = fake.goal_assignments[-1].get("resume_notice") or ""
    s3_statuses = s3_notice.split("## Task statuses", 1)[1] \
                        .split("## New completion reports", 1)[0]
    check("Headline: the flange calibrates" in s3_statuses
          and "the full report with all the detail lines" not in s3_statuses,
          "the statuses section carries ONE LINE per task (the summary's first line)")
    check(s3_notice.count("the full report with all the detail lines.") == 1
          and "### #" in s3_notice,
          "each new completion's full body embeds EXACTLY ONCE (no double embedding)")
    check(wait_for(lambda: task_state(auth, run_task)["status"] == "suspended"
                   or False, timeout=60, what="S3 agent re-parks") is not None,
          "the S3 agent re-parks")
    # Size constant per event, not cumulative: the NEXT resume (armed
    # with a fresh single-line completion) reports only the NEW body —
    # the consumed multiline report never re-embeds.
    check(wait_for(run_park_second_passed, timeout=30,
                   what="S3 second park second passes") is not None,
          "the second park stamp is in a strictly older second")
    conn = db()
    conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
                 " priority, summary, completed_at, created_at) values (1, ?, 'task',"
                 " 'Fresh completion', 'done', 5, 'fresh one-liner',"
                 " strftime('%s','now'),"
                 " strftime('%s','now','-30 seconds'))", (goal1,))
    conn.commit()
    conn.close()
    s3_more = len(fake.goal_assignments)
    check(wait_for(lambda: len(fake.goal_assignments) > s3_more or False,
                   timeout=60, what="S3 second delta resume") is not None,
          "the fresh completion resumes the agent again")
    s3_next = fake.goal_assignments[-1].get("resume_notice") or ""
    check("the full report with all the detail lines." not in s3_next
          and "### #" in s3_next and "fresh one-liner" in s3_next,
          "the consumed completion's body never re-embeds (notice size constant per event)")
    check(wait_for(lambda: task_state(auth, run_task)["status"] == "suspended"
                   or False, timeout=60, what="S3 agent parks a third time") is not None,
          "the S3 agent parks for the pause section")

    print("== feedback: goal page parity — progress feed, current activity, runs ==")
    # The goal agent's runs feed the page's live surfaces: progress
    # lines (green latest-first feed), the parked round's newest
    # activity line and its hosting saddle. Parked rounds keep their
    # session record (closed 'parked', D4) — the page reads the
    # newest row, so fixtures land as rows without disturbing the
    # scripted daemon.
    conn = db()
    open_run_session = conn.execute(
        "select max(id) from sessions where task_id = ?",
        (run_task,)).fetchone()[0]
    conn.execute(
        "insert into events (session_id, seq, kind, actor, scope_type, scope_id,"
        " payload, created_at) values (?, 8001, 'progress', 'model', 'task', ?,"
        " ?, strftime('%s','now'))",
        (open_run_session, run_task, json.dumps({"text": "Overseeing the flange fan-out"})))
    conn.execute(
        "insert into events (session_id, seq, kind, actor, scope_type, scope_id,"
        " payload, created_at) values (?, 8002, 'tool_call', 'model', 'task', ?,"
        " ?, strftime('%s','now'))",
        (open_run_session, run_task,
         json.dumps({"tool": "run_command",
                     "args": {"command": "git -C demo log --oneline -5"}})))
    conn.commit()
    conn.close()
    # A manual refresh records the daemon's health snapshot (the
    # parked daemon answers /status with its suspended state).
    act(auth, saddle_card(auth, "fake-saddle")["refresh"], {})
    state = goal_state(goal1)
    check(state["progress"] and state["progress"][0]["summary"]
          == "Overseeing the flange fan-out",
          "the goal page carries the agent's live progress feed")
    check(state["current_command"] == "run_command: git -C demo log --oneline -5"
          and state["current_command_running"] is True,
          "the current activity shows the open round's running command")
    check(state["agent_saddle"] == "fake-saddle"
          and "fake-saddle" in state["saddle_health"],
          "the assigned agent saddle and its health display")
    check(state["runs"] and state["runs"][0]["saddle_name"] == "fake-saddle"
          and state["runs"][0]["terminal_url"] == "/tasks/%d/log" % run_task,
          "the run table carries host saddles and the event log link")
    goal_page = http_get_page(auth, "/goals/%d" % goal1)
    check('stm-markdown="goal.brief"' in goal_page
          and 'stm-markdown="goal.plan"' in goal_page
          and 'stm-markdown="comment.body"' in goal_page
          and 'stm-markdown="event.text"' in goal_page
          and 'stm-markdown="block.body"' in goal_page,
          "every long-form goal body renders through stm-markdown")
    check('stm-attribute.href="run.terminal_url" target="_blank" rel="noopener"' in goal_page,
          "the run table's terminal links open in a new tab")
    check("progress-latest" in goal_page and "Agent progress" in goal_page,
          "the green progress feed renders on the goal page")
    # Plan revision history keeps its verdicts (approved + rejected).
    revisions = goal_state(goal1)["plan_revisions"]
    # Newest-first since the ordering sweep (73dcec9): index 0 is the
    # approved revision, index 1 the older rejected one.
    check(revisions[1]["answered"] is True and revisions[1]["approved"] is False
          and revisions[0]["approved"] is True,
          "plan revision history badges approved and rejected verdicts")

    print("== plan 12: pause freezes supervision; events coalesce on resume ==")
    state = goal_state(goal1)
    act(auth, state["pause"], {})
    check(goal_state(goal1)["status"] == "paused"
          and goal_state(goal1).get("resume") is not None,
          "an active goal pauses and offers resume")
    # Terminal events under a paused goal queue without waking anybody.
    # (The real completion path — complete_task — records the goal's
    # task-done narrative event; the resume machinery keys on exactly
    # those terminal kinds, never on the agent's own enqueue narrative.)
    conn = db()
    conn.execute("insert into tasks (project_id, goal_id, kind, title, status, priority,"
                 " summary, completed_at, created_at) values (1, ?, 'task',"
                 " 'Quiet task', 'done', 5, 'finished while paused',"
                 " strftime('%s','now'), strftime('%s','now'))", (goal1,))
    quiet_id = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.commit()
    conn.close()
    paused_runs = len(fake.goal_assignments)
    check(quiet_window(quiet_seconds(),
                       lambda: len(fake.goal_assignments) == paused_runs,
                       "paused goal does not resume"),
          "a paused goal's events queue without scheduling a resume")
    state = goal_state(goal1)
    act(auth, state["resume"], {})
    check(goal_state(goal1)["status"] == "active",
          "resume returns the goal to active")
    check(wait_for(lambda: len(fake.goal_assignments) > paused_runs or False,
                   timeout=30, what="queued events flush on resume") is not None,
          "the queued events coalesce into the post-resume wake")
    check("finished while paused" in (fake.goal_assignments[-1].get("resume_notice") or ""),
          "the flushed notice reports the task that finished while paused")
    check(wait_for(lambda: task_state(auth, run_task)["status"] == "suspended"
                   or False, timeout=60, what="goal agent parks again") is not None,
          "the goal agent parks again after the flush")
    # The watermark consumed the flushed transition: with nothing new
    # under the goal, no further resume fires (parked stays parked).
    flushed_assignments = len(fake.goal_assignments)
    check(quiet_window(quiet_seconds(),
                       lambda: len(fake.goal_assignments) == flushed_assignments,
                       "consumed watermark never re-fires"),
          "the consumed watermark never re-fires a resume")

    print("== plan 12: completion review, push state, accept ==")
    # The daemon exposes goal-branch push state on /status; the web
    # caches it on the goal when it probes the goal agent's saddle.
    def status_extra():
        return None
    fake.status_extra = {"goal_branches": [
        {"branch": "goal/flange", "pushed": True},
        {"branch": "goal/other-repo", "pushed": False}]}
    fake_card = saddle_card(auth, "fake-saddle")
    act(auth, fake_card["refresh"], {})
    check(wait_for(lambda: goal_state(goal1).get("push_branches") or False,
                   timeout=30, what="push state lands on the goal") is not None,
          "the daemon's /status goal_branches exposure caches on the goal")
    branches = goal_state(goal1)["push_branches"]
    check({(b["branch"], b["pushed"]) for b in branches}
          == {("goal/flange", True), ("goal/other-repo", False)},
          "the push state renders per branch")
    # the daemon's real shape carries repo too — multi-repo rows stay
    # distinguishable as "repo: branch" labels
    fake.status_extra = {"goal_branches": [
        {"repo": "repoa", "branch": "goal/flange", "pushed": True},
        {"repo": "repob", "branch": "goal/flange", "pushed": False}]}
    act(auth, fake_card["refresh"], {})
    check(wait_for(lambda: {b["branch"] for b in goal_state(goal1).get("push_branches") or []}
                   >= {"repoa: goal/flange"}, timeout=30,
                   what="repo-qualified push state lands") is not None,
          "repo-qualified goal_branches rows render as 'repo: branch'")
    fake.status_extra = None

    # `saddle done` in goal context: the completion review.
    fake.goal_script = [("done", "# Completion\n\nThe flange ships.")]
    # Wake the parked agent: a terminal transition newer than the
    # watermark (the pause-flush consumed everything up to it, so arm
    # one directly with a fresh row).
    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'))", (goal1,))
    conn.commit()
    conn.close()
    check(wait_for(lambda: goal_state(goal1)["status"] == "completion_review" or False,
                   timeout=60, what="goal agent submits its completion") is not None,
          "`saddle done` in goal context enters completion review")
    state = goal_state(goal1)
    # Plan 16 S1 (canonical-on-accept): while the review is PENDING the
    # goal row carries NO report — the document lives only in its
    # block; completion_report is written solely by the accept verdict.
    conn = db()
    pending_report_column = conn.execute(
        "select completion_report from goals where id = ?", (goal1,)).fetchone()[0]
    conn.close()
    check((pending_report_column is None or pending_report_column == "")
          and (state["completion_report"] or "") == ""
          and "# Completion" in (state["completion_pending_block"].get("body") or "")
          and state["completion_pending"] is True
          and state["completion_pending_block"] is not None
          and state["completion_pending_block"] is not False,
          "the pending report lives ONLY in its block (goal row clean, S1)")
    check(task_state(auth, run_task)["status"] == "blocked",
          "the goal-run task parks on the completion review")
    # Plan 16 S5 (U3): label reconciliation — the goal awaiting its
    # completion review reads "awaiting your review" and the run-task
    # status card agrees (never a bare "blocked" for a review park).
    check(goal_state(goal1)["display"] == "awaiting your review"
          and task_state(auth, run_task)["display"] == "awaiting your review",
          "the goal and the run-task card agree: 'awaiting your review' (U3)")
    check(state["compare_url"].endswith("/compare/main...goal/flange"),
          "the completion review links the compare view")
    # Reject first: feedback returns the goal to active. Plan 16 S2:
    # the goal page's reject posts raw `feedback` — the SERVER wraps
    # it with the completion-review prefix (byte-identical to the
    # historical client-built answer).
    fake.goal_script = [("suspend",)]
    act(auth, state["completion_pending_block"]["answer_action"],
        {"feedback": "Not done — the docs are missing"})
    check(wait_for(lambda: goal_state(goal1)["status"] == "active" or False,
                   timeout=30, what="rejected completion reactivates") is not None,
          "rejecting the completion returns the goal to active")
    conn = db()
    rejected_answer = conn.execute(
        "select answer from task_blocks where task_id = ?"
        " and type = 'completion' order by id desc", (run_task,)).fetchone()[0]
    conn.close()
    check(rejected_answer
          == "Review feedback received:\n\nNot done — the docs are missing",
          "the completion rejection wraps feedback with the stable prefix (%r)"
          % rejected_answer)
    conn = db()
    rejected_report = conn.execute(
        "select completion_report from goals where id = ?", (goal1,)).fetchone()[0]
    conn.close()
    check(rejected_report is None or rejected_report == "",
          "reject/revise leaves the goal row clean (no report leaked, S1)")
    # Round 2: submit again and accept.
    fake.goal_script = [("done", "# Completion v2\n\nThe flange ships, with docs.")]
    conn = db()
    conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
                 " priority, completed_at, created_at) values (1, ?, 'task',"
                 " 'Warmup2', 'done', 5,"
                 " strftime('%s','now'),"
                 " strftime('%s','now','-30 seconds'))", (goal1,))
    conn.commit()
    conn.close()
    check(wait_for(lambda: goal_state(goal1)["status"] == "completion_review" or False,
                   timeout=60, what="resubmitted completion") is not None,
          "the agent resubmits its completion after feedback")
    state = goal_state(goal1)
    # Regression (plan-15 followup): the daemon's done-park ALSO pushes a
    # review-kind block payload (question "Goal completion review", the
    # real daemon's wording) after the done entry raised the completion
    # row — the two paths must converge on ONE row, not twin reviews.
    fake.push_results("blocked", block={
        "question": "Goal completion review", "type": "review",
        "body": "# Completion v2\n\nThe flange ships, with docs.",
        "options": None})
    conn = db()
    rows = conn.execute(
        "select count(*) from task_blocks where task_id = ?"
        " and answer is null and type in ('completion','review')",
        (run_task,)).fetchone()[0]
    conn.close()
    check(rows == 1,
          "the daemon's review-kind completion push adopts the pending row (no twin)")
    state = goal_state(goal1)
    review_family_in_open = [b for b in state.get("open_blocks", [])
                             if b.get("is_completion_review")]
    check(state["completion_pending"] is True and len(review_family_in_open) == 1,
          "the pending completion review renders ONCE — as the rich attention block")
    fake.pending_block = None
    # Raise-side single-row invariant (the render guard and the
    # twin-answer sweep are gone — fresh DBs cannot hold pre-dedupe
    # twins, and 4884259's raise-side cross-dedupe keeps it that way):
    # a REDELIVERED review-kind completion push adopts the same pending
    # row instead of stacking a twin.
    fake.push_results("blocked", block={
        "question": "Goal completion review", "type": "review",
        "body": "# Completion v2\n\nThe flange ships, with docs.",
        "options": None})
    fake.pending_block = None
    conn = db()
    rows = conn.execute(
        "select count(*) from task_blocks where task_id = ?"
        " and answer is null and type in ('completion','review')",
        (run_task,)).fetchone()[0]
    conn.close()
    check(rows == 1,
          "a redelivered review-kind push adopts the pending row (raise-side single-row invariant)")
    # Plan 15 C4: the completion-accept hard gate — the live probe
    # flags still report repob's branch unpushed, so the operator's
    # accept REFUSES with origin behind.
    refusal = action_notify(act(auth, state["completion_pending_block"]["answer_action"],
                                {"answer": "Review approved. Continue with the task."}))
    check(goal_state(goal1)["status"] == "completion_review",
          "the accept REFUSES while origin is behind (hard gate, live flags)")
    # One parser, one label: the gate's refusal names the SAME repo-
    # qualified unpushed row the chips render — both surfaces read
    # GoalService.parse_goal_branches (the gate-vs-chips divergence is
    # structurally gone).
    branches = {(b["branch"], b["pushed"]) for b in goal_state(goal1)["push_branches"]}
    check("repob: goal/flange" in (refusal or "")
          and ("repob: goal/flange", False) in branches
          and ("repoa: goal/flange", True) in branches,
          "the refusal and the chips read ONE parser (same labels, same flags)")
    # Push the branch daemon-side and probe again — the flags flip
    # live (no goal-row write), and the accept goes through.
    fake.status_extra = {"goal_branches": [
        {"repo": "repoa", "branch": "goal/flange", "pushed": True},
        {"repo": "repob", "branch": "goal/flange", "pushed": True}]}
    act(auth, fake_card["refresh"], {})
    check(wait_for(lambda: (goal_state(goal1).get("push_branches") or [])
                   and all(b["pushed"] for b in goal_state(goal1)["push_branches"])
                   or False, timeout=30, what="pushed flags land") is not None,
          "the push chips derive from the LIVE probe flags")
    conn = db()
    no_push_column = conn.execute(
        "select count(*) from pragma_table_info('goals')"
        " where name = 'branch_push_state'").fetchone()[0] == 0
    conn.close()
    check(no_push_column,
          "no stored push snapshot exists (F9: the goal row never records it)")
    state = goal_state(goal1)
    # The passing accept posts the verdict MARKER (the goal page's
    # accept form shape, plan 16 S2) — the server resolves the
    # completion kind and stores the stable approval wording.
    act(auth, state["completion_pending_block"]["answer_action"],
        {"verdict": "accept"})
    check(wait_for(lambda: goal_state(goal1)["status"] == "complete" or False,
                   timeout=30, what="accepted completion completes") is not None,
          "accepting the completion completes the goal once origin is current")
    state = goal_state(goal1)
    conn = db()
    accepted_report = conn.execute(
        "select completion_report from goals where id = ?", (goal1,)).fetchone()[0]
    conn.close()
    check(state["terminal"] is True and state.get("cancel") is None
          and accepted_report == "# Completion v2\n\nThe flange ships, with docs."
          and "# Completion v2" in state["completion_report"],
          "a complete goal is terminal with the WINNING block's body as its report (S1)")
    check(wait_for(lambda: task_state(auth, run_task)["status"] == "done" or False,
                   timeout=30, what="goal run task closes") is not None,
          "approving the completion closes the goal-run task done")
    conn = db()
    twins_left = conn.execute(
        "select count(*) from task_blocks where task_id = ?"
        " and answer is null and type in ('completion','review')",
        (run_task,)).fetchone()[0]
    conn.close()
    check(twins_left == 0,
          "the verdict answered the pending completion-family row (single-row invariant)")
    # Plan 16 D4 (web half): the verdict path closes what it closes —
    # after the accept NO session row of the run task dangles open,
    # and every parked round carries outcome 'parked' (the daemon's
    # file-log park brackets name these same rows).
    conn = db()
    open_after_accept = conn.execute(
        "select count(*) from sessions where task_id = ? and ended_at is null",
        (run_task,)).fetchone()[0]
    parked_rounds = conn.execute(
        "select count(*) from sessions where task_id = ? and outcome = 'parked'",
        (run_task,)).fetchone()[0]
    conn.close()
    check(open_after_accept == 0,
          "the accept leaves no dangling open session (D4: every park closed its row)")
    check(parked_rounds >= 1,
          "the run's parked rounds all closed 'parked' (D4 outcome rename)")

    print("== plan 15 C3: the narrative derives from the recorded shape ==")
    # Content equivalence with the old projection on this run's exact
    # shape: every kind the goal_events projection produced, keyed the
    # same way, plus the watermark and no-ingest guarantees.
    narrative = goal_state(goal1)["events"]
    kinds = [e["kind"] for e in narrative]
    for expected in ("created", "started", "plan-submitted", "plan-rejected",
                     "plan-approved", "task-created", "task-done",
                     "completion-submitted", "completion-rejected", "goal-approved"):
        check(expected in kinds,
              "the derived narrative carries the %s row" % expected)
    check(all(k != "task-enqueued" and k != "resumed" and k != "retry" for k in kinds),
          "timestamp-less bookkeeping kinds never return")
    stamps = [e["at_display"] for e in narrative if e.get("at_display")]
    check(stamps == sorted(stamps, reverse=True),
          "the narrative reads newest first")
    conn = db()
    no_event_table = conn.execute(
        "select count(*) from sqlite_master where type = 'table'"
        " and name = 'goal_events'").fetchone()[0] == 0
    conn.close()
    check(no_event_table,
          "the goal_events table is gone (no narrative rows on ingest)")

    print("== plan 12: cancel cascades to the goal's tasks ==")
    goal2 = create_goal({"project_id": "1", "title": "Doomed goal",
                         "brief": "This goal will be cancelled.",
                         "branch_name": "goal/doomed"})
    fake.goal_script = [("suspend",)]
    act(auth, goal_state(goal2)["start"], {})
    conn = db()
    run2 = conn.execute("select id from tasks where goal_id = ? and kind = 'goal_run'",
                        (goal2,)).fetchone()[0]
    conn.close()
    check(wait_for(lambda: task_state(auth, run2)["status"] == "suspended" or False,
                   timeout=60, what="doomed goal agent parks") is not None,
          "the doomed goal's agent parks")
    conn = db()
    agent_session2 = conn.execute(
        "select max(id) from sessions where task_id = ?",
        (run2,)).fetchone()[0]
    conn.close()
    status, body = saddle_api("POST", "/api/saddle/task/create", fake_token, {
        "session_id": agent_session2, "title": "Doomed draft", "mandate": "m"})
    doomed_draft = int(body["task_id"])
    doomed_queued_task = None
    status, body = saddle_api("POST", "/api/saddle/task/create", fake_token, {
        "session_id": agent_session2, "title": "Doomed runner", "mandate": "m"})
    doomed_runner = int(body["task_id"])
    saddle_api("POST", "/api/saddle/task/enqueue", fake_token,
               {"session_id": agent_session2, "task_id": doomed_runner})
    act(auth, goal_state(goal2)["cancel"], {})
    check(goal_state(goal2)["status"] == "cancelled",
          "the goal cancels from active (parked-agent) state")
    check(task_state(auth, run2)["status"] == "failed",
          "the cascade cancels the goal agent's run task")
    check(wait_for(lambda: task_state(auth, doomed_runner)["status"] == "failed"
                   or False, timeout=30, what="cascade cancels the queued task") is not None,
          "the cascade cancels the goal's queued task")
    check(not any(t["id"] == doomed_draft
                  for t in slot_public(auth, "/tasks", "tasks")["tasks"]),
          "the cascade deletes the goal's draft tasks")
    # A cancelled goal frees its branch name.
    reuse = create_goal({"project_id": "1", "title": "Reused name",
                         "brief": "b", "branch_name": "goal/doomed"})
    check(reuse > goal2, "a cancelled goal's branch name is reusable")

    print("== plan 13 B3: goal failure cascades to non-terminal tasks ==")
    # A goal holding a draft, a queued task and a working task: the
    # agent's failure cascades like a cancellation — drafts deleted (no
    # orphan stays dispatchable), queued/working cancelled — and the
    # narrative records the cascade.
    collapse = create_goal({"project_id": "1", "title": "Collapse goal",
                            "brief": "Fails with tasks in flight.",
                            "branch_name": "goal/collapse"})
    fake.goal_script = [("suspend",)]
    act(auth, goal_state(collapse)["start"], {})
    conn = db()
    run3 = conn.execute("select id from tasks where goal_id = ? and kind = 'goal_run'",
                        (collapse,)).fetchone()[0]
    conn.close()
    check(wait_for(lambda: task_state(auth, run3)["status"] == "suspended" or False,
                   timeout=60, what="collapse goal agent parks") is not None,
          "the collapse goal's agent parks first")
    conn = db()
    run3_session = conn.execute(
        "select max(id) from sessions where task_id = ?",
        (run3,)).fetchone()[0]
    conn.close()

    def agent_task(title):
        status, body = saddle_api("POST", "/api/saddle/task/create", fake_token, {
            "session_id": run3_session, "title": title, "mandate": "m"})
        task_id = int(body["task_id"])
        saddle_api("POST", "/api/saddle/task/enqueue", fake_token,
                   {"session_id": run3_session, "task_id": task_id})
        return task_id

    status, body = saddle_api("POST", "/api/saddle/task/create", fake_token, {
        "session_id": run3_session, "title": "Collapse draft", "mandate": "m"})
    collapse_draft = int(body["task_id"])  # stays draft (never enqueued)
    collapse_queued = agent_task("Collapse queued")
    collapse_worker = agent_task("Collapse worker")
    # Both queued tasks race for the one saddle; freeze the daemon the
    # moment the first dispatches so one stays genuinely WORKING (and
    # the other queued) when the goal fails.
    fake.hold_state = True
    def collapse_working():
        for task_id in (collapse_queued, collapse_worker):
            if task_state(auth, task_id)["status"] == "working":
                return task_id
        return False
    running_one = wait_for(collapse_working, timeout=60, what="a collapse task works")
    check(running_one is not None,
          "a collapse task reaches working before the failure")
    queued_one = collapse_worker if running_one == collapse_queued else collapse_queued

    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                           {"session_id": run3_session, "state": "failed",
                            "summary": "the foundation crumbled"})
    check(status == 200 and goal_state(collapse)["status"] == "failed",
          "the agent's failure fails the goal")
    check(task_state(auth, run3)["status"] == "failed",
          "the failure fails the goal agent's run task")
    check(wait_for(lambda: task_state(auth, queued_one)["status"] == "failed"
                   and task_state(auth, running_one)["status"] == "failed" or False,
                   timeout=30, what="failure cascade settles") is not None,
          "the failure cascade cancels the queued and working tasks")
    check(not any(t["id"] == collapse_draft
                  for t in slot_public(auth, "/tasks", "tasks")["tasks"]),
          "the failure cascade deletes the goal's draft tasks")
    state = goal_state(collapse)
    check(any(e["kind"] == "cascade" for e in state["events"]),
          "the narrative records the failure cascade")
    check(state["draft_tasks"] == 0 and state["queued_tasks"] == 0,
          "nothing under the failed goal stays dispatchable (no orphan drafts)")
    check(len([e for e in state["events"] if e["kind"] == "goal-failed"]) == 1,
          "the failing goal records exactly one goal-failed event")
    # Plan 16 S1: the FAILURE report keeps its goal-side home (failure
    # is terminal — no pending duplication exists, so unlike the
    # completion report it is written at failure time).
    conn = db()
    failure_report = conn.execute(
        "select completion_report from goals where id = ?", (collapse,)).fetchone()[0]
    conn.close()
    check(failure_report == "the foundation crumbled",
          "fail_goal's failure report keeps its goal-side home (S1)")
    # Plan 16 S4: the narrative renders the task-failed rows from the
    # terminal summary's lead — the run thread carries NO duplicate
    # agent comment of the goal failure text.
    failed_rows = [e for e in state["events"] if e["kind"] == "task-failed"]
    check(failed_rows
          and all(": Cancelled by user" in e["text"] for e in failed_rows),
          "task-failed narrative rows render the summary lead (S4: summary is the terminal cache)")
    run_comments = [c for c in task_state(auth, run3)["comments"]
                    if c["author"] == "agent"]
    check(not any("the foundation crumbled" in c["body"] for c in run_comments),
          "the goal failure text never duplicates as an agent comment (S4)")
    fake.hold_state = False

    print("== plan 12: auto-enqueue-after for human tasks ==")
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1", "title": "Predecessor task", "priority": "4"})
    predecessor = task_id_of("Predecessor task")
    enqueue_task(auth, predecessor)
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1", "title": "Successor task", "priority": "4",
        "enqueue_after_task_id": str(predecessor)})
    successor = task_id_of("Successor task")
    check(task_state(auth, successor)["status"] == "draft"
          and task_state(auth, successor).get("enqueue_after_task_id") == predecessor,
          "a human task can carry an auto-enqueue predecessor")
    check(wait_task_done(auth, predecessor, timeout=60) is not None,
          "the predecessor completes")
    check(wait_for(lambda: task_state(auth, successor)["status"] in
                   ("queued", "assigned", "working") or False,
                   timeout=30, what="successor auto-enqueues") is not None,
          "the done predecessor auto-enqueues its dependent")
    # Cycle validation refuses a predecessor chain that loops.
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1", "title": "Cycle A", "priority": "4"})
    cycle_a = task_id_of("Cycle A")
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1", "title": "Cycle B", "priority": "4",
        "enqueue_after_task_id": str(cycle_a)})
    cycle_b = task_id_of("Cycle B")
    cycle_directives = act(auth, task_state(auth, cycle_a)["update"], {
        "title": "Cycle A", "description": "x", "priority": "4",
        "enqueue_after_task_id": str(cycle_b)})
    check(any(d.get("type") == "notify" and d.get("kind") == "error"
              and "cycle" in (d.get("message") or "").lower()
              for d in cycle_directives),
          "a predecessor cycle rejects with a naming error")
    # A failed predecessor leaves the dependent in draft.
    fake.fail_next = True
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1", "title": "Failing predecessor", "priority": "1"})
    failing_pred = task_id_of("Failing predecessor")
    enqueue_task(auth, failing_pred)
    act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
        "project_id": "1", "title": "Stranded successor", "priority": "4",
        "enqueue_after_task_id": str(failing_pred)})
    stranded = task_id_of("Stranded successor")
    check(wait_for(lambda: task_state(auth, failing_pred)["status"] == "failed"
                   or False, timeout=60, what="failing predecessor fails") is not None,
          "the scripted-failure predecessor fails")
    check(quiet_window(quiet_seconds(),
                       lambda: task_state(auth, stranded)["status"] == "draft",
                       "failed predecessor leaves the draft"),
          "a failed predecessor leaves its dependent in draft")

    # Successor cleanup: let the queued successor finish so the
    # shard's app shuts down with an empty queue.
    check(wait_task_done(auth, successor, timeout=90) is not None,
          "the auto-enqueued successor completes")

    print("== feedback: context_update carries rule/note deltas on resume ==")
    # A standalone task parks on a question; while it is parked a rule
    # is attached to the project and a note lands; the resuming
    # assignment carries both as the context_update markdown turn.
    probe_id = create_queued(auth, {"project_id": "1", "title": "Context probe",
                                    "priority": "4"})
    check(wait_for(lambda: fake.running_task == probe_id
                   and task_state(auth, probe_id)["status"] == "working" or False,
                   timeout=60, what="probe task runs") is not None,
          "the probe task dispatches")
    fake.set_state("blocked")
    fake.push_results("blocked", block={"type": "question",
                                        "question": "Which database dialect?"})
    check(wait_for(lambda: task_state(auth, probe_id)["status"] == "blocked" or False,
                   timeout=60, what="probe task blocks") is not None,
          "the probe task parks on its question")
    detail = task_state(auth, probe_id)
    open_block = [b for b in detail["blocks"] if not b["answered"]][0]
    # Backdate the parked round's start: stamps are second-resolution
    # epochs, and the delta compares strictly-later — the park must
    # span a real interval like an operator's would.
    conn = db()
    conn.execute("update sessions set started_at = strftime('%s','now') - 60"
                 " where id = (select min(id) from sessions where task_id = ?)",
                 (probe_id,))
    conn.commit()
    conn.close()
    # While parked: a new rule attached to the project + a new note.
    rules_state = slot_public(auth, "/rules", "rules")
    act(auth, rules_state["create"], {
        "name": "Nightly gate", "body": "Every PR must pass the nightly suite before merge."})
    conn = db()
    rule_id = conn.execute("select id from rules where name = 'Nightly gate'").fetchone()[0]
    conn.close()
    act(auth, slot_public(auth, "/projects/1", "project")["attach_rule"],
        {"rule_id": str(rule_id)})
    conn = db()
    conn.execute("insert into project_notes (project_id, body, created_at)"
                 " values (1, 'The demo repo moved its CI to nightly builds.',"
                 " strftime('%s','now'))")
    conn.commit()
    check(conn.execute("select count(*) from rule_projects where rule_id = ?",
                       (rule_id,)).fetchone()[0] == 1,
          "the rule attaches to the project while the task is parked")
    conn.close()
    probe_assignments = len([p for p in fake.start_payloads
                             if int(p.get("task", {}).get("id") or 0) == probe_id])
    act(auth, open_block["answer_action"], {"answer": "postgres"})
    check(wait_for(lambda: len([p for p in fake.start_payloads
                                if int(p.get("task", {}).get("id") or 0) == probe_id])
                   > probe_assignments or False, timeout=60,
                   what="probe task resumes") is not None,
          "answering the block resumes the probe task")
    probe_resume = [p for p in fake.start_payloads
                    if int(p.get("task", {}).get("id") or 0) == probe_id][-1]
    delta = probe_resume.get("context_update") or ""
    check("Rules changed" in delta and "Nightly gate" in delta
          and "nightly suite" in delta,
          "the resume's context_update summarises the newly attached rule")
    check("New project notes" in delta and "nightly builds" in delta,
          "the resume's context_update summarises the new project note")
    check("Plan approved" not in delta,
          "standalone resumes carry no plan section")
    check(wait_task_done(auth, probe_id, timeout=60) is not None,
          "the probe task completes after its resume")

    print("== goal 4 t1: force wake — the human resume for a parked agent ==")
    # The goal-2 stall class: the parked run has NO pending wake (a
    # draft task never enqueued → no terminal transition → parked
    # forever). Force wake is the human recovery lever. A fresh goal
    # owns the scenario (goal1's run completed with its goal).
    stall_goal = create_goal({"project_id": "1", "title": "Stall goal",
                              "brief": "Parks with a never-enqueued draft.",
                              "branch_name": "goal/stall"})
    fake.goal_script = [("plan", "# Stall plan\n\n1. Wait for the stall")]
    act(auth, goal_state(stall_goal)["start"], {})
    check(wait_for(lambda: goal_state(stall_goal)["status"] == "plan_review" or False,
                   timeout=60, what="stall plan enters review") is not None,
          "the stall goal reaches plan review")
    act(auth, goal_state(stall_goal)["plan_pending_block"]["answer_action"],
        {"verdict": "accept"})
    check(wait_for(lambda: goal_state(stall_goal)["status"] == "active" or False,
                   timeout=30, what="stall goal activates") is not None,
          "the stall goal activates")
    fake.goal_script = [("suspend",)]
    conn = db()
    run_task = conn.execute(
        "select id from tasks where goal_id = ? and kind = 'goal_run'",
        (stall_goal,)).fetchone()[0]
    conn.close()
    check(wait_for(lambda: task_state(auth, run_task)["status"] == "suspended"
                   or False, timeout=60, what="agent parks for force wake") is not None,
          "the goal agent parks before the force-wake scenario")
    # The stall fixture: an agent-created draft that was never enqueued
    # (typo'd enqueue) — statuses must name it as NOT enqueued.
    conn = db()
    agent_session = conn.execute(
        "select max(id) from sessions where task_id = ?", (run_task,)).fetchone()[0]
    conn.close()
    status, body = saddle_api("POST", "/api/saddle/task/create", fake_token, {
        "session_id": agent_session, "title": "QA sweep",
        "mandate": "Run the QA sweep.", "mode": "test"})
    check(status == 200 and body.get("ok") is True,
          "the stall fixture draft task creates (%s)" % body.get("error"))
    stall_draft = int(body["task_id"])
    goal_page = http_get_page(auth, "/goals/%d" % stall_goal)
    check('stm-action="goal.force_wake"' in goal_page,
          "the goal page authors the Force wake button while the run is parked")
    check('stm-if="goal.force_wake != null"' in goal_page,
          "the Force wake button is gated on the action being offered")
    # No wake without the press: parked stays parked.
    parked_runs = len(fake.goal_assignments)
    check(quiet_window(quiet_seconds(),
                       lambda: len(fake.goal_assignments) == parked_runs,
                       "no press, no wake"),
          "without a press the parked agent never wakes")
    state = goal_state(stall_goal)
    check(state.get("force_wake") is not None,
          "the goal state exposes the offered force-wake action")
    act(auth, state["force_wake"], {})
    check(wait_for(lambda: len(fake.goal_assignments) > parked_runs or False,
                   timeout=30, what="force-woke agent resumes") is not None,
          "the force-woke agent resumes immediately (no coalesce wait)")
    woke_notice = fake.goal_assignments[-1].get("resume_notice") or ""
    check("force-woke this goal's supervision" in woke_notice
          and "no pending wake" in woke_notice,
          "the force-wake resume leads with the explicit wake line")
    check(("draft (NOT enqueued" in woke_notice)
          and ("saddle task enqueue %d" % stall_draft) in woke_notice,
          "the statuses list names the never-enqueued draft with the enqueue fix")
    check("Task statuses" in woke_notice,
          "the force-wake notice still carries the standard batched report")
    # The flag is consumed: re-park (script), no second wake fires.
    check(wait_for(lambda: task_state(auth, run_task)["status"] == "suspended"
                   or False, timeout=60, what="woken agent re-parks") is not None,
          "the woken agent parks again (script)")
    woke_runs = len(fake.goal_assignments)
    check(quiet_window(quiet_seconds(),
                       lambda: len(fake.goal_assignments) == woke_runs,
                       "consumed flag never re-fires"),
          "the consumed flag does not re-wake (one wake per press)")
    # Guards: a paused goal refuses the wake (Resume is its lever).
    state = goal_state(stall_goal)
    act(auth, state["pause"], {})
    conn = db()
    paused_flag = conn.execute("select force_wake from goals where id = ?",
                               (stall_goal,)).fetchone()
    conn.close()
    check(paused_flag is not None and paused_flag[0] in (0, False),
          "pausing does not set the flag")
    state = goal_state(stall_goal)
    check(state.get("force_wake") is None,
          "a paused goal does not author the Force wake action")
    act(auth, state["resume"], {})
    check(wait_for(lambda: goal_state(stall_goal)["status"] == "active" or False,
                   timeout=30, what="goal reactivates") is not None,
          "the goal reactivates for the guard checks")
    # Non-suspended run refuses: run is parked again (script parked it
    # above), wake it, let it park, then a second press while parked is
    # fine — the 400 path needs a NON-suspended run. Start there: wake
    # once more and press while it is working.
    state = goal_state(stall_goal)
    act(auth, state["force_wake"], {})
    check(wait_for(lambda: len(fake.goal_assignments) > woke_runs or False,
                   timeout=30, what="second wake fires") is not None,
          "a second press wakes again (re-press is allowed)")
    # While it is mid-script (not yet parked), the action is hidden.
    check(goal_state(stall_goal).get("force_wake") is None
          or task_state(auth, run_task)["status"] != "suspended",
          "the wake action hides while the run is not parked")
    check(wait_for(lambda: task_state(auth, run_task)["status"] == "suspended"
                   or False, timeout=60, what="run parks for cleanup") is not None,
          "the run parks again")