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

    def plant_live_worker(goal_id, title="Live supervision worker",
                          status="blocked"):
        """Goal 27 refinements — the parking guard: a goal agent may
        park only while its goal owns a task whose terminal transition
        could still fire (the park's wake source). Scenes that script
        the agent to park with every owned task already terminal plant
        one BLOCKED live worker first — wakeable (not terminal), never
        dispatched (nothing runs behind the scene's back), and swept
        by the goal's own cancel/fail cascade at teardown."""
        conn = db()
        conn.execute("insert into tasks (project_id, goal_id, kind, title,"
                     " status, priority, mode, creator, created_at)"
                     " values (1, ?, 'task', ?, ?, 5, 'implement', 'goal',"
                     " strftime('%s','now'))", (goal_id, title, status))
        planted = conn.execute("select max(id) from tasks").fetchone()[0]
        conn.commit()
        conn.close()
        return int(planted)

    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 class="nav-goal" href="/goals">' 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")
    # Goal 41 T10: the dashboard's feed is the work stream now; the
    # pinned goal cards keep their distinct styling through the
    # dynamic class expressions (item_type 'goal' -> goal-card +
    # badge-kind-goal at render time).
    check("'card goal-card'" in home_html
          and "'badge badge-kind-' + blockview.item_type" 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.
    # (The parking guard: the agent parks only with something live to
    # wait on — the blocked worker below is this park's wake source.)
    plant_live_worker(goal1)
    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.
    plant_live_worker(goal1, "Live supervision worker two")
    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)")
    plant_live_worker(goal1, "Live supervision worker three")
    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)")
    plant_live_worker(goal1, "Live supervision worker four")
    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")
    # Goal 41 T7: the run table's log links navigate SAME-TAB (a new
    # tab severed the back button; the log is one back-click away and
    # the task page embeds the same feed as its Log tab).
    check('stm-attribute.href="run.terminal_url"' in goal_page
          and 'target="_blank"' not in goal_page.split("run.terminal_url")[1][:200],
          "the run table's terminal links navigate in the same 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")
    plant_live_worker(goal1, "Live supervision worker five")
    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("== S6: repeated-failure signature in the resume notice ==")
    # The fail-loop shape (web-db health plan A1/A2): worker tasks
    # failing with IDENTICAL summary signatures since the park. The
    # notice must name the cluster and forbid re-cloning; distinct
    # signatures, or failures older than the park, must NOT cluster.
    check(wait_for(run_park_second_passed, timeout=30,
                   what="S6 park second passes") is not None,
          "the S6 park stamp is in a strictly older second")
    shared_sig = ("curl exited 22: JSON parse error: Invalid UTF-8 middle byte"
                  " 0x5c — the clamped plan slice cuts a multi-byte character"
                  " mid-character at byte 3000 of the shared goal plan")
    conn = db()
    for tail in (" (clone attempt one)", " (clone attempt two)"):
        conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
                     " priority, summary, completed_at, created_at) values"
                     " (1, ?, 'task', 'Flange clone', 'failed', 5, ?,"
                     " strftime('%s','now'), strftime('%s','now','-30 seconds'))",
                     (goal1, shared_sig + tail))
    clone_ids = [int(r[0]) for r in conn.execute(
        "select id from tasks where title = 'Flange clone'").fetchall()]
    conn.commit()
    conn.close()
    check(len(clone_ids) == 2,
          "the S6 fixture lands two same-signature failed workers")
    s6_before = len(fake.goal_assignments)
    check(wait_for(lambda: len(fake.goal_assignments) > s6_before or False,
                   timeout=60, what="S6 repeated-failure resume fires") is not None,
          "the repeated failures resume the parked agent")
    s6_notice = fake.goal_assignments[-1].get("resume_notice") or ""
    check("REPEATED FAILURE" in s6_notice
          and "do NOT recreate the same task again" in s6_notice
          and "change the approach, narrow the mandate, or split the scope"
          in s6_notice,
          "the notice carries the repeated-failure instruction")
    check(("#%d" % clone_ids[0]) in s6_notice
          and ("#%d" % clone_ids[1]) in s6_notice,
          "the repeated-failure section names both cluster members")
    check("Invalid UTF-8 middle byte 0x5c" in s6_notice,
          "the section names the shared signature's first line")
    plant_live_worker(goal1, "Live supervision worker six")
    check(wait_for(lambda: task_state(auth, run_task)["status"] == "suspended"
                   or False, timeout=60, what="S6 agent re-parks") is not None,
          "the S6 agent re-parks")
    # Negative round: one pre-park failure sharing the signature (the
    # watermark excludes it) plus one NEW failure with a distinct
    # signature — neither forms a ≥2 cluster, so no section renders.
    check(wait_for(run_park_second_passed, timeout=30,
                   what="S6 second park second passes") is not None,
          "the S6 negative-round park stamp is strictly older")
    conn = db()
    conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
                 " priority, summary, completed_at, created_at) values"
                 " (1, ?, 'task', 'Stale signature clone', 'failed', 5, ?,"
                 " strftime('%s','now','-120 seconds'),"
                 " strftime('%s','now','-150 seconds'))",
                 (goal1, shared_sig + " (stale pre-park clone)"))
    stale_id = conn.execute(
        "select id from tasks where title = 'Stale signature clone'"
    ).fetchone()[0]
    conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
                 " priority, summary, completed_at, created_at) values"
                 " (1, ?, 'task', 'Distinct failure', 'failed', 5,"
                 " 'tests failed: the flange widget lacks a widget prop',"
                 " strftime('%s','now'), strftime('%s','now','-30 seconds'))",
                 (goal1,))
    distinct_id = conn.execute(
        "select id from tasks where title = 'Distinct failure'"
    ).fetchone()[0]
    conn.commit()
    conn.close()
    s6_more = len(fake.goal_assignments)
    check(wait_for(lambda: len(fake.goal_assignments) > s6_more or False,
                   timeout=60, what="S6 negative-round resume fires") is not None,
          "the negative-round failures resume the agent")
    s6_next = fake.goal_assignments[-1].get("resume_notice") or ""
    check("REPEATED FAILURE" not in s6_next,
          "no repeated-failure section without a same-signature pair"
          " newer than the park")
    check(("#%d" % distinct_id) in s6_next and ("#%d" % stale_id) in s6_next,
          "the negative-round notice still reports both failures as statuses")
    plant_live_worker(goal1, "Live supervision worker seven")
    check(wait_for(lambda: task_state(auth, run_task)["status"] == "suspended"
                   or False, timeout=60, what="S6 negative agent re-parks")
          is not None,
          "the S6 negative round re-parks")

    print("== S5: the goal page's aggregate LLM usage stat line ==")
    # usage_summary_for_goal sums the usage events across every task
    # of the goal (the agent's run plus its workers); a goal with no
    # usage renders no line.
    conn = db()
    conn.execute("insert into goals (project_id, title, brief, branch_name,"
                 " status, created_at) values (1, 'Usage goal', 'Sum the usage.',"
                 " 'goal/usage', 'active', strftime('%s','now'))")
    usage_goal = conn.execute(
        "select id from goals where title = 'Usage goal'").fetchone()[0]
    for title, payloads in (
            ("Usage worker one", ({"prompt_tokens": 1000000, "completion_tokens": 5000,
                                   "elapsed_ms": 3600000},
                                  {"prompt_tokens": 500000, "completion_tokens": 2500,
                                   "elapsed_ms": 1800000})),
            ("Usage worker two", ({"prompt_tokens": 2500000, "completion_tokens": 12000,
                                   "elapsed_ms": 5400000},))):
        conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
                     " priority, created_at) values (1, ?, 'task', ?, 'done', 5,"
                     " strftime('%s','now'))", (usage_goal, title))
        worker = conn.execute(
            "select id from tasks where title = ?", (title,)).fetchone()[0]
        for payload in payloads:
            conn.execute(
                "insert into events (kind, actor, scope_type, scope_id, payload,"
                " created_at) values ('usage', 'harness', 'task', ?, ?,"
                " strftime('%s','now'))",
                (worker, json.dumps(payload)))
    conn.commit()
    conn.close()
    state = goal_state(usage_goal)
    check(state.get("usage_line") == "LLM: 3 calls · 4.0M in / 19.5K out tok · 3h 0m",
          "the goal state sums usage across its tasks (%r)" % state.get("usage_line"))
    goal_page = http_get_page(auth, "/goals/%d" % usage_goal)
    check('stm-if="goal.usage_line.length &gt; 0"' in goal_page
          and 'stm-text="goal.usage_line"' in goal_page,
          "the goal page renders the usage stat line in the status head")
    check(goal_state(goal_elsewhere).get("usage_line") == "",
          "a goal whose tasks recorded no usage carries no stat line")

    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).
    plant_live_worker(goal1, "Live supervision worker eight")
    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)")
    # Goal 21 (accept-gate removal): the accept-time push gate is GONE.
    # The push-before-review rule lives daemon-side, at the agent's
    # completion attempt — a goal agent parked in this review has no
    # open session, so the web's probe flags are necessarily stale and
    # an accept-time check could only misfire (error 88 wedged the
    # goal in completion_review forever). Regression: with the cached
    # flags still reading repob UNPUSHED, the accept must go straight
    # through — the flags feed only the goal page's advisory chips.
    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: (goal_state(goal1).get("push_branches") or [])
                   and any((b["branch"], b["pushed"]) == ("repob: goal/flange", False)
                           for b in goal_state(goal1)["push_branches"])
                   or False, timeout=30, what="unpushed flags land") is not None,
          "the chips still render the live (unpushed) 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 (the goal row never records it)")
    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,
          "the accept COMPLETES the goal while cached flags read unpushed (no gate)")
    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"})
    plant_live_worker(goal2, "Doomed live worker")
    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"})
    plant_live_worker(collapse, "Collapse live worker")
    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).
    # (Goal 27 refinements — the parking guard: the park needs a live
    # wake source now, so the blocked worker rides along; the never-
    # enqueued DRAFT below still produces no transition, which is the
    # stall the wake recovers.)
    stall_goal = create_goal({"project_id": "1", "title": "Stall goal",
                              "brief": "Parks with a never-enqueued draft.",
                              "branch_name": "goal/stall"})
    plant_live_worker(stall_goal, "Stall live worker")
    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")