integration: goal-era cross-binary pass — integration_test re-baselined, four ingestion faults fixed

integration_test.py now drives BOTH real binaries through the full
goal arc: create/start, pre-plan question (one answer resolving the
goal_blocks row AND the mirrored task_blocks row), plan
reject->revise->approve on the agreed verdict words, task fan-out via
the real /api/saddle/task callbacks, the worker's ephemeral worktree
(auto-pruned, branch kept), the coalesced terminal-event resume,
merge/branch-delete/push, completion review accept, the
/status goal_branches cache sampled by the manual refresh, SSE over
the run's raw terminal, the goal_events narrative and the legacy
ingest surface (plan/subtask update types now failing loudly).

Driving the real pair exposed four web faults, all fixed:
- raise_agent_block had no pending dedup: a redelivered blocked push
  stacked duplicate goal blocks
- a park-unwinding round-end push (blocked, entries only, no block
  payload) synthesised a fallback question row -- it now raises
  nothing; the payload-carrying transition already did
- the done ENTRY path completed tasks with an empty summary (it
  stripped the command instead of reading the entry's output), losing
  worker completion reports whenever the entry beat the transition
- resume_due counted ANY goal event: the agent's own
  task-created/task-enqueued narrative woke it early, before its
  tasks ran -- only task-done/task-failed (owned-task terminal
  events, per the plan) age the coalescing window now

src/Services/GoalService.vala

@@ -388,11 +388,18 @@ namespace ClankerSaddle.Services {
 
         /**
          * Records a goal-agent question/RFC block and parks the
-         * goal-run task on its mirrored task block.
+         * goal-run task on its mirrored task block. Idempotent on the
+         * pending question: a redelivered blocked push (outbox retry,
+         * the park-unwinding round-end push) must not stack duplicates.
          */
         public async GoalBlock raise_agent_block(int64 goal_id, string question,
                 string block_type, string[]? options = null, string? body = null) throws Error {
             var goal = yield require_goal(goal_id);
+            foreach (var existing in yield list_blocks(goal_id)) {
+                if (existing.question == question && existing.answer == null) {
+                    return existing;
+                }
+            }
             string options_json = null;
             if (options != null && options.length > 0) {
                 var array = new Json.Array();
@@ -682,12 +689,19 @@ namespace ClankerSaddle.Services {
                 return false;
             }
             DateTime? oldest = null;
+            // ONLY owned-task terminal events resume the parked agent —
+            // the agent's own task-created/task-enqueued narrative (and
+            // human comments) must never wake it; the first terminal
+            // event ages the window and bursts coalesce behind it.
             foreach (var event in yield db.query<GoalEvent>()
                     .where(expr("goal_id == $0 && id > $1",
                         elem<int64?>(run_task.goal_id),
                         elem<int64?>(((!)goal).last_seen_event_id)))
                     .order_by(expr("id"))
                     .materialise_async()) {
+                if (event.kind != "task-done" && event.kind != "task-failed") {
+                    continue;
+                }
                 if (event.created_at != null) {
                     oldest = ((!)event).created_at;
                     break;

src/Services/SaddleService.vala

@@ -1300,8 +1300,16 @@ namespace ClankerSaddle.Services {
                                 report.length > 0 ? report : "The goal agent submitted its completion report");
                             return true;
                         }
-                        yield tasks.complete_task(task_id,
-                            strip_quotes(strip_verb(command, "saddle done")), null, "done");
+                        {
+                            // the DONE entry's output IS the report
+                            // (the command is just "saddle done") —
+                            // falling back to it keeps the summary when
+                            // this entry path beats the transition push
+                            string report = output.length > 0 ? output
+                                : strip_quotes(strip_verb(command, "saddle done"));
+                            yield tasks.complete_task(task_id,
+                                report.length > 0 ? report : null, null, "done");
+                        }
                         return true;
                     case "failed":
                         string explanation = output.length > 0 ? output
@@ -1567,23 +1575,31 @@ namespace ClankerSaddle.Services {
                         // plan payload enters plan review, anything
                         // else raises a goal question/RFC. The mirror
                         // block on the run task parks it for the
-                        // verdict resume.
-                        string question = block_question_from_payload(
-                            block != null ? (!)block : new Json.Object());
+                        // verdict resume. A BLOCKED push WITHOUT a
+                        // block payload (the park-unwinding round-end
+                        // push, its trailing entries) raises nothing —
+                        // the payload-carrying transition already did,
+                        // and a synthesised fallback row would dupe it.
+                        if (block == null) {
+                            if (!(yield task_is_terminal(task_id))) {
+                                yield tasks.set_status(task_id, STATUS_BLOCKED);
+                            }
+                            yield park_if_current(saddle.id, task_id);
+                            break;
+                        }
+                        string question = block_question_from_payload((!)block);
                         if (question.length == 0) {
                             question = "The goal agent raised a question";
                         }
-                        if (block != null
-                                && block_string_or((!)block, "type", "block_type") == BLOCK_PLAN) {
+                        if (block_string_or((!)block, "type", "block_type") == BLOCK_PLAN) {
                             yield goal_service.submit_plan(goal_id,
                                 optional_string_member((!)block, "body") ?? question);
                         } else {
                             yield goal_service.raise_agent_block(goal_id, question,
-                                block_string_or(block != null ? (!)block : new Json.Object(),
-                                    "type", "block_type"),
-                                string_array_member(block != null ? (!)block : new Json.Object(),
-                                    block != null && block.has_member("options") ? "options" : "block_options"),
-                                optional_string_member(block != null ? (!)block : new Json.Object(), "body"));
+                                block_string_or((!)block, "type", "block_type"),
+                                string_array_member((!)block,
+                                    (!)block.has_member("options") ? "options" : "block_options"),
+                                optional_string_member((!)block, "body"));
                         }
                         if (!(yield task_is_terminal(task_id))) {
                             yield tasks.set_status(task_id, STATUS_BLOCKED);

test/integration_test.py

@@ -1,30 +1,47 @@
 #!/usr/bin/env python3
-"""End-to-end integration test: ClankerSaddle web app + saddle daemon (Part 2).
+"""End-to-end integration test: ClankerSaddle web app + saddle daemon.
 
 Drives the web app the way statum.js does (entrypoint state loads plus
 action POSTs carrying the auth slot and sealed private blobs) while a
-real saddle daemon works on a real git checkout under WORKSPACE/demo,
-powered by saddle/test/mock_llm.py with a deterministic script.
-
-Sections:
-1. explore task (bash + comment + done) and dependency gating (v1 loop)
-2. the Part 2 feature tour, driven by `saddle` commands the mock emits:
-   plan via heredoc, `saddle query` with piped stdin, progress updates,
-   a multiple-choice block (answered through the web UI), an RFC block
-   with a markdown body, a `saddle review` banner approved via the UI,
-   and a quoted user comment delivered to the daemon over /notify
-3. the ingest API surface: bearer-token auth (401/200), typed updates,
-   the poll endpoint's exactly-once handover
-4. SSE: terminal stream replay + live push (with piped query input)
-5. the stm-markdown renderer shipped in statum.js (via node)
-6. the audit trail in SQLite
-
-Prerequisites (see README.md "Integration testing"):
-- the web app running with SPRY_SEED_ADMIN_USERNAME=admin,
-  SPRY_SEED_ADMIN_PASSWORD=adminpass123, SADDLE_LLM_BASE_URL pointing at
-  the mock (below), SADDLE_PROJECTS_ROOT=$WORKSPACE
-- a saddle daemon reachable at SADDLE
-- a git repo with at least two commits at $WORKSPACE/demo
+REAL saddle daemon runs a REAL goal against a real git checkout under
+WORKSPACE, powered by saddle/test/mock_llm.py with a deterministic
+script — both binaries, the true cross-binary contract proof (the
+per-repo fake-peer suits live in web_stack_test.py / saddle e2e).
+
+The goal-era arc (plan 12):
+1. goal create -> start: the goal-run assignment reaches the daemon
+   over the ask/allocate channel with the full plan-12 payload (kind/
+   mode/creator/goal context/ephemeral branches for workers)
+2. pre-plan question block -> answered on the goal page -> the ONE
+   answer resolves the goal_blocks row AND the mirrored task_blocks
+   row, and the parked goal agent resumes with the verdict wording
+3. plan submit -> reject-with-feedback -> revision -> approval (the
+   agreed verdict words ride the resume)
+4. task fan-out through the real /api/saddle/task/create + enqueue
+   callbacks; the worker runs in its ephemeral worktree branch cut
+   from the goal clone, commits, completes — worktree auto-pruned,
+   branch kept
+5. the task-terminal event coalesces (~SADDLE_GOAL_RESUME_S) into the
+   goal agent's batched resume; it merges, deletes the task branch,
+   pushes the goal branch (push-before-review) and submits the
+   completion report
+6. completion review accept -> goal complete, run task done
+7. the goal agent's terminal sessions (SSE replay), the audit trail
+   (goal_events narrative, typed session_commands), the /status
+   goal_branches cache sampled by the manual saddle refresh
+8. the legacy ingest API surface (401s, typed updates, exactly-once
+   poll handover) and the stm-markdown renderer
+
+Prerequisites (see README.md "Integration testing"): the web app
+running with SPRY_SEED_ADMIN_USERNAME=admin,
+SPRY_SEED_ADMIN_PASSWORD=adminpass123, SADDLE_LLM_BASE_URL pointing
+at the mock, SADDLE_PROJECTS_ROOT=$WORKSPACE and a FRESH database
+(the harness asserts an empty tasks/goals state); a saddle daemon
+with SADDLE_WORKSPACE_ROOT=$WORKSPACE (the shared-volume deployment
+assumption — the goal clone anchors the worker worktrees); a git
+repo with at least two commits at $WORKSPACE/demo whose origin is
+pushable (the harness prepares demo + demo-origin.git when WORKSPACE
+is empty).
 
 Environment:
   WEB            web app base URL      (default http://localhost:8080)
@@ -64,6 +81,8 @@ MOCK_PORT = os.environ.get("MOCK_PORT", "9999")
 MOCK_CMD = os.environ.get("MOCK_CMD", "")
 DEMO = os.path.join(WORKSPACE, "demo")
 HOSTNAME = socket.gethostname().split(".")[0]
+PLAN_APPROVED = "Plan approved. Proceed with implementation."
+REVIEW_APPROVED = "Review approved. Continue with the task."
 
 passes = 0
 failures = 0
@@ -171,6 +190,15 @@ def wait_task_status(auth, task_id, status, timeout=90):
                     timeout=timeout, what="task %d -> %s" % (task_id, status))
 
 
+def goal_state(auth, goal_id):
+    return slot_public(auth, "/goals/%d" % goal_id, "goal")
+
+
+def wait_goal_status(auth, goal_id, status, timeout=90):
+    return wait_for(lambda: goal_state(auth, goal_id).get("status") == status or False,
+                    timeout=timeout, what="goal %d -> %s" % (goal_id, status))
+
+
 def saddle_api(method, path, token, payload=None):
     """Raw call against the saddle ingest API; returns (status, json)."""
     request = urllib.request.Request(WEB + path,
@@ -189,38 +217,85 @@ def saddle_api(method, path, token, payload=None):
             return error.code, {"raw": body}
 
 
+def db_query(sql, params=()):
+    db = sqlite3.connect(DB_PATH)
+    try:
+        return db.execute(sql, params).fetchall()
+    finally:
+        db.close()
+
+
+def prepare_workspace():
+    """Ensures $WORKSPACE/demo exists with a pushable origin (the goal
+    branch push needs a remote that accepts it). Reuses an existing
+    demo checkout; the DATABASE must still be fresh (checked in main)."""
+    if os.path.isdir(os.path.join(DEMO, ".git")):
+        return
+    os.makedirs(WORKSPACE, exist_ok=True)
+    origin = os.path.join(WORKSPACE, "demo-origin.git")
+    subprocess.run(["git", "init", "--bare", "-b", "main", origin], check=True,
+                   capture_output=True)
+    subprocess.run(["git", "clone", origin, DEMO], check=True, capture_output=True)
+    subprocess.run(["git", "-C", DEMO, "config", "user.email", "e2e@localhost"], check=True)
+    subprocess.run(["git", "-C", DEMO, "config", "user.name", "e2e"], check=True)
+    with open(os.path.join(DEMO, "hello.py"), "w") as handle:
+        handle.write("print('demo')\n")
+    subprocess.run(["git", "-C", DEMO, "add", "-A"], check=True, capture_output=True)
+    subprocess.run(["git", "-C", DEMO, "commit", "-m", "one"], check=True, capture_output=True)
+    with open(os.path.join(DEMO, "notes.md"), "w") as handle:
+        handle.write("# notes\n")
+    subprocess.run(["git", "-C", DEMO, "add", "-A"], check=True, capture_output=True)
+    subprocess.run(["git", "-C", DEMO, "commit", "-m", "two"], check=True, capture_output=True)
+    subprocess.run(["git", "-C", DEMO, "push", "origin", "main"], check=True, capture_output=True)
+
+
 def mock_argv():
     heredoc = "cat << 'SADDLE_EOF' | %s\n%%s\nSADDLE_EOF"
+    fanout = (
+        "# Fanning the plan out\n"
+        "out=$(cat << 'SADDLE_EOF' | saddle task create --title 'Write the greeting' --mode implement\n"
+        "Create greeting.txt containing hello-from-the-goal.\nSADDLE_EOF)\n"
+        "tid=$(echo \"$out\" | grep -oE '[0-9]+' | head -1)\n"
+        "saddle task enqueue \"$tid\"\n"
+        "saddle suspend"
+    )
     responses = [
-        "# Inspecting the checkout\nls -la %s" % DEMO,
-        "# Checking history\ngit -C %s log --oneline" % DEMO,
-        'saddle comment "Inspected the checkout"\nsaddle done "Explored the demo repository"',
-        "# Starting the follow-up\necho task-two-running",
-        'saddle done "Follow-up audit complete"',
-        # Part 2 feature tour (task 3)
-        "# First, write the plan\n" + (heredoc % "saddle plan") % (
-            "# Feature tour plan\n\n1. Query the host\n2. Ask a question\n"
-            "3. Propose an RFC\n4. Request review"),
-        "# Confirm the working branch\ngit branch --show-current",
-        "# Side question about this host\nhostname | saddle query \"What host is this?\"",
-        "# Report progress\nsaddle progress-update queried the host",
-        "# Need a decision\n" + (heredoc % 'saddle block question "Which database should we use?" "SQLite" "Postgres" "Flat files"') % (
-            "# Storage choice\n\nWe need somewhere to keep the audit trail.\n\n"
-            "- **SQLite** — zero admin\n- **Postgres** — serious scale\n- **Flat files** — yolo"),
-        "# Proposing the module layout\n" + (heredoc % "saddle block rfc") % (
-            "# Proposed layout\n\n```text\nsrc/\n  api/\n  services/\n```\n\n"
-            "The `api` layer stays thin; services own the domain."),
-        "# Ready for review\n" + (heredoc % "saddle review") % (
-            "## Review: feature tour\n\nEverything on the tour is wired up.\n\n"
-            "- plans render\n- blocks answer\n- reviews approve"),
-        "# Waiting for any human input before finishing\nsleep 12",
+        # R1 (fresh goal agent): investigate, then block on a scope question
+        "# Investigating before planning\nls -la\ngit log --oneline | head -3\n"
+        + (heredoc % 'saddle block question "Demo only" "Whole workspace"') % (
+            "Which scope should the greeting cover? The demo repo is small; confirm the blast radius."),
+        # R2 (plan revision after feedback): the rejected plan, re-submitted
+        (heredoc % "saddle plan") % (
+            "# Plan v2\n\n1. Add greeting.txt to the demo repo.\n2. Export it for reuse."),
+        # W1 (the worker task): verify the ephemeral branch, commit, done
+        "# Working the mandate\ngit branch --show-current\n"
+        "echo hello-from-the-goal > greeting.txt\ngit add -A\n"
+        "git commit -m 'greeting from the goal'\n"
+        + (heredoc % "saddle done") % (
+            "# Completion report\n\n- changed: greeting.txt\n"
+            "- verified: committed on the ephemeral branch"),
     ]
     argv = []
     for response in responses:
         argv += ["-r", response]
     argv += [
-        "--on-answer", "# Human decision received\nsaddle progress-update human input integrated",
-        "--on-comment", 'saddle comment "Got the human comment"\nsaddle done "Part 2 feature tour complete"',
+        # the question's verdict ("Human selected: Demo only") -> plan v1
+        "--on-answer", (heredoc % "saddle plan") % (
+            "# Plan v1\n\n1. Add greeting.txt to the demo repo."),
+        # the approval verdict -> task fan-out + suspend
+        "--on-plan-approved", fanout,
+        # the batched "Goal update" resume -> merge, prune, push, complete
+        "--answer-contains", "Goal update",
+        "--on-answer-seen",
+        "# Merging the finished task, then completing (the sleep holds the\n"
+        "# working state open so /status can sample the branch state)\n"
+        "sleep 4\n"
+        "git merge --no-edit $(git for-each-ref --format='%(refname:short)' refs/heads/task/*)\n"
+        "git branch -D $(git for-each-ref --format='%(refname:short)' refs/heads/task/*)\n"
+        "git push -u origin $(git branch --show-current)\n"
+        + (heredoc % "saddle done") % (
+            "# Goal completion report\n\n- achieved: greeting.txt added via one worker task\n"
+            "- verified: the goal branch is pushed to origin"),
         "--on-query", "The hostname is %s." % HOSTNAME,
     ]
     return argv
@@ -289,6 +364,12 @@ def markdown_renderer_checks():
 
 
 def main():
+    prepare_workspace()
+
+    if db_query("select count(*) from tasks")[0][0] or db_query("select count(*) from goals")[0][0]:
+        raise SystemExit("the web database is not fresh (tasks/goals exist) — "
+                         "wipe the DB and restart the app+daemon for a clean goal-era run")
+
     mock = start_mock()
     time.sleep(2.0)
 
@@ -306,47 +387,12 @@ def main():
         state = slot_public(auth, "/projects", "projects")
         act(auth, state["create"], {
             "name": "demo",
-            "git_url": "https://example.com/demo.git",
+            "git_url": DEMO,
             "default_branch": "main",
             "summary": "Integration test repository",
         })
         check(any(p["name"] == "demo" for p in slot_public(auth, "/projects", "projects")["projects"]),
-              "project demo created")
-
-        state = slot_public(auth, "/tasks", "tasks")
-        act(auth, state["create"], {
-            "project_id": "1",
-            "title": "Explore the demo repository",
-            "description": "List the checkout and inspect its git history.",
-            "priority": "3",
-        })
-        act(auth, state["create"], {
-            "project_id": "1",
-            "title": "Follow-up audit",
-            "description": "Runs after the explore task completes.",
-            "priority": "4",
-        })
-        act(auth, state["create"], {
-            "project_id": "1",
-            "title": "Part 2 feature tour",
-            "description": "Plan, query, block, RFC, review, comment — the works.",
-            "priority": "5",
-            "task_branch": "task/3-feature-tour",
-        })
-        titles = [t["title"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]]
-        check(titles == ["Explore the demo repository", "Follow-up audit", "Part 2 feature tour"],
-              "three tasks queued in priority order")
-
-        tour = task_state(auth, 3)
-        check(tour["branch_from"] == "main", "branch_from prefilled from the project default branch")
-        check(tour["task_branch"] == "task/3-feature-tour", "task_branch stored at creation")
-
-        detail = task_state(auth, 2)
-        act(auth, detail["dependency"], {"depends_on_task_id": "1"})
-        check(wait_for(lambda: any(int(d.get("id", 0)) == 1
-                                   for d in task_state(auth, 2).get("dependencies", [])) or False,
-                       timeout=5, what="dependency edge of task 2"),
-              "task 2 depends on task 1")
+              "project demo created (local git_url, pushable origin)")
 
         state = slot_public(auth, "/saddles", "saddles")
         act(auth, state["register"], {"name": SADDLE_NAME, "host": SADDLE_HOST})
@@ -355,225 +401,209 @@ def main():
         check(len(registered) == 1 and registered[0]["token"].startswith("saddle-"),
               "saddle registered at %s with an API token" % SADDLE_HOST)
 
-        print("== explore task ==")
-        left_queue = wait_for(
-            lambda: (lambda s: s if s != "queued" else False)(task_state(auth, 1).get("status")) or False,
-            timeout=15, what="task 1 to leave the queue")
-        check(left_queue is not None, "dispatcher picks task 1 up within 15s (status %s)" % left_queue)
-        check(wait_for(lambda: (saddle_status() or {}).get("state") in ("working", "blocked", "done") or False,
-                       timeout=15, what="saddle active") is not None,
-              "saddle daemon receives the task")
-        check(wait_task_status(auth, 1, "working", timeout=30) is not None,
-              "task 1 flips to working from daemon status")
-        check(wait_task_status(auth, 1, "done", timeout=90) is not None, "task 1 completes")
-
-        detail = task_state(auth, 1)
-        check("Explored the demo repository" in (detail.get("summary") or ""),
-              "task 1 carries the agent's summary")
-        check(any(c["author"] == "bot" and "Inspected the checkout" in c["body"]
-                  for c in detail.get("comments", [])),
-              "agent comment projected into task_comments")
-        check(detail.get("sessions") and detail["sessions"][0].get("outcome") == "done",
-              "session closed with outcome done")
-
-        print("== dependency gating ==")
-        check(wait_task_status(auth, 2, "done", timeout=90) is not None,
-              "task 2 (dependent) dispatches and completes after task 1")
-
-        print("== part 2 feature tour ==")
-        check(wait_task_status(auth, 3, "working", timeout=30) is not None,
-              "task 3 dispatches to the saddle")
-
-        # plan via `saddle plan` heredoc
-        check(wait_for(lambda: "# Feature tour plan" in (task_state(auth, 3).get("plan") or "") or False,
-                       timeout=60, what="task 3 plan") is not None,
-              "saddle plan markdown lands in the task plan slot")
-        db = sqlite3.connect(DB_PATH)
-        stored_plan = db.execute("select plan from tasks where id = 3").fetchone()[0] or ""
-        check("1. Query the host" in stored_plan and "4. Request review" in stored_plan,
-              "plan stored in the tasks table verbatim")
-        db.close()
-
-        # `saddle query` with piped stdin
-        check(wait_for(lambda: any("query" == row[0]
-                                   for row in query_commands(3)) or False,
-                       timeout=60, what="query entry") is not None,
-              "saddle query logged as a typed terminal entry")
-
-        # the workspace clone sits on the task's branch
-        db = sqlite3.connect(DB_PATH)
-        branch_rows = db.execute(
-            "select sc.output_text from session_commands sc join sessions s on s.id = sc.session_id"
-            " where s.task_id = 3 and sc.command_text = 'git branch --show-current'"
-        ).fetchall()
-        db.close()
-        check(branch_rows and (branch_rows[0][0] or "").strip() == "task/3-feature-tour",
-              "workspace clone checks out the task branch")
-        db = sqlite3.connect(DB_PATH)
-        query_rows = db.execute(
-            "select sc.command_text, sc.output_text, sc.piped_input from session_commands sc"
-            " join sessions s on s.id = sc.session_id where s.task_id = 3 and sc.type = 'query'"
-        ).fetchall()
-        db.close()
-        check(len(query_rows) == 1 and "What host is this?" in query_rows[0][0],
-              "query entry carries the question as its command")
-        check(query_rows and HOSTNAME in (query_rows[0][1] or ""),
-              "query entry carries the side-LLM answer as output")
-        check(query_rows and (query_rows[0][2] or "").strip(),
-              "query entry stores the piped stdin in session_commands.piped_input")
-
-        # progress updates
-        check(wait_for(lambda: any("queried the host" == p["summary"]
-                                   for p in task_state(auth, 3).get("progress", [])) or False,
-                       timeout=60, what="first progress update") is not None,
-              "saddle progress-update appears on the task page")
-        detail = task_state(auth, 3)
-        check(detail["progress"][0]["summary"] == "queried the host",
-              "latest progress update shown first")
-
-        # block question with options + markdown body
-        check(wait_task_status(auth, 3, "blocked", timeout=90) is not None,
-              "saddle block question flips the task to blocked")
-        detail = task_state(auth, 3)
-        open_q = [b for b in detail.get("blocks", [])
-                  if b["block_type"] == "question" and not b.get("answered")]
-        check(len(open_q) == 1 and "Which database" in open_q[0]["question"],
-              "block question surfaced in the web UI")
-        check(open_q and open_q[0]["options"] == ["SQLite", "Postgres", "Flat files"],
-              "block options stored on the block")
-        check(open_q and "# Storage choice" in open_q[0]["body"]
-              and "**SQLite**" in open_q[0]["body"],
-              "block markdown body stored on the block")
-        act(auth, open_q[0]["answer_action"], {"option": "Postgres"})
-        answered = [b for b in task_state(auth, 3)["blocks"]
-                    if b["block_type"] == "question"]
-        check(any(b.get("answer") == "Human selected: Postgres" for b in answered),
-              "option answer formatted for the agent")
-        # the resume window can be brief; the agent continuing past the
-        # answer is the durable signal (it raises the RFC next)
-        check(wait_for(lambda: any(b.get("block_type") == "rfc"
-                                   for b in task_state(auth, 3).get("blocks", [])) or False,
-                       timeout=90, what="agent continues after the answer") is not None,
-              "answered question unblocks the agent")
-
-        # RFC block with markdown body
-        check(wait_task_status(auth, 3, "blocked", timeout=90) is not None,
-              "saddle block rfc flips the task to blocked")
-        detail = task_state(auth, 3)
-        open_rfc = [b for b in detail.get("blocks", [])
-                    if b["block_type"] == "rfc" and not b.get("answered")]
-        check(len(open_rfc) == 1 and "# Proposed layout" in open_rfc[0]["body"],
-              "RFC markdown body stored on the block")
-        check(open_rfc and "```text" in open_rfc[0]["body"],
-              "RFC body keeps fenced code blocks intact")
-        act(auth, open_rfc[0]["answer_action"], {"answer": "Layout approved as proposed"})
-        check(any(b.get("answer") == "Layout approved as proposed"
-                  for b in task_state(auth, 3)["blocks"] if b["block_type"] == "rfc"),
-              "RFC reply recorded verbatim")
-        # the review banner following proves the agent resumed past the RFC
-        check(wait_for(lambda: task_state(auth, 3).get("review_pending") or False,
-                       timeout=90, what="agent resumes after the RFC") is not None,
-              "answered RFC unblocks the agent")
-
-        # review banner + approve
-        check(wait_task_status(auth, 3, "review_requested", timeout=90) is not None,
-              "saddle review flips the task to review_requested")
-        detail = task_state(auth, 3)
-        check(detail["review_pending"] and "Everything on the tour is wired up" in detail["review_summary"],
-              "review banner data on the task page")
-        check(detail["review_diff_url"].endswith(
-            "demo.git/compare/main...task/3-feature-tour"), "review diff link constructed")
-        dash = slot_public(auth, "/", "dashboard")
-        check(any(int(r["task_id"]) == 3 and r["block_type"] == "review"
-                  for r in dash["pinned"]),
-              "dashboard pins the pending review")
-        review_block = [b for b in detail["blocks"]
-                        if b["block_type"] == "review" and not b.get("answered")]
-        check(len(review_block) == 1, "review raised as an answerable block")
-        act(auth, review_block[0]["answer_action"], {"answer": "Human approved the review"})
-        check(any(b.get("answer") == "Human approved the review"
-                  for b in task_state(auth, 3)["blocks"] if b["block_type"] == "review"),
-              "review approval recorded")
-
-        # single formatting end to end: the daemon's ANSWER entry repeats the
-        # answer the way the agent read it
-        check(wait_task_status(auth, 3, "working", timeout=90) is not None,
-              "task 3 resumes working after the review approval")
-        db = sqlite3.connect(DB_PATH)
-        answer_rows = db.execute(
-            "select sc.command_text from session_commands sc join sessions s on s.id = sc.session_id"
-            " where s.task_id = 3 and sc.type = 'answer' order by sc.id").fetchall()
-        db.close()
-        check(any("Human selected: Postgres" == row[0] for row in answer_rows),
-              "daemon formats the option answer exactly once")
-
-        # quoted user comment while working (terminal highlight flow)
-        detail = task_state(auth, 3)
-        act(auth, detail["comment"], {
-            "body": "The sleep is fine, but flush sooner next time",
-            "quote": "sleep 12",
-            "return_to": "/tasks/3/terminal",
+        print("== goal create -> start ==")
+        goals_state = slot_public(auth, "/goals", "goals")
+        act(auth, goals_state["create"], {
+            "project_id": "1",
+            "title": "Greet the demo",
+            "brief": "Add a greeting file to the demo repo end to end.",
+            "branch_name": "goal/demo-run",
+            "priority": "3",
         })
-        check(wait_task_status(auth, 3, "done", timeout=90) is not None,
-              "user comment delivered via /notify; task 3 completes")
-        detail = task_state(auth, 3)
-        check("Part 2 feature tour complete" in (detail.get("summary") or ""),
-              "task 3 summary reflects the post-comment completion")
-        check(any(c["author"] == "user:admin" and c.get("quote") == "sleep 12"
-                  for c in detail.get("comments", [])),
-              "quoted user comment stored on the task")
-        check(any(c["author"] == "bot" and "Got the human comment" in c["body"]
-                  for c in detail.get("comments", [])),
-              "agent acknowledged the comment in-conversation")
-
-        db = sqlite3.connect(DB_PATH)
-        tour_types = {row[0] for row in db.execute(
-            "select sc.type from session_commands sc join sessions s on s.id = sc.session_id"
-            " where s.task_id = 3")}
-        notified = db.execute(
-            "select sc.output_text from session_commands sc join sessions s on s.id = sc.session_id"
-            " where s.task_id = 3 and sc.type = 'user-comment'").fetchall()
-        db.close()
-        check({"bash", "comment", "block", "answer", "plan", "progress", "review",
-               "query", "user-comment"} <= tour_types,
-              "terminal log carries every part 2 entry type (%s)" % sorted(tour_types))
-        check(any("sleep 12" in (row[0] or "") for row in notified),
-              "/notify delivered the comment with its quoted terminal text")
+        goal1 = int([g["id"] for g in slot_public(auth, "/goals", "goals")["goals"]
+                     if g["title"] == "Greet the demo"][0])
+        check(goal_state(auth, goal1)["status"] == "draft", "the goal lands in draft")
+        act(auth, goal_state(auth, goal1)["start"], {})
+        check(wait_goal_status(auth, goal1, "planning", timeout=30) is not None,
+              "starting the goal moves it to planning (agent dispatched)")
+        run_task = db_query(
+            "select id from tasks where goal_id = ? and kind = 'goal_run'", (goal1,))[0][0]
+        check(wait_for(lambda: task_state(auth, run_task)["status"]
+                       in ("assigned", "working", "blocked", "suspended") or False,
+                       timeout=30, what="goal run dispatched") is not None,
+              "the goal-run task reaches the daemon (assigned/working/blocked)")
+        goal_clone = os.path.join(WORKSPACE, "goal-%d" % goal1, "demo")
+        check(wait_for(lambda: os.path.isdir(os.path.join(goal_clone, ".git")) or False,
+                       timeout=30, what="the goal clone") is not None,
+              "the goal clone appears under the shared workspace root (goal-<id>)")
+        branch = subprocess.run(["git", "-C", goal_clone, "branch", "--show-current"],
+                                capture_output=True, text=True).stdout.strip()
+        check(branch == "goal/demo-run", "the goal clone sits on the user-named goal branch")
+
+        print("== pre-plan question -> one answer, dual rows ==")
+        check(wait_for(lambda: any(b["block_type"] == "question" and not b.get("answered")
+                                   for b in goal_state(auth, goal1).get("blocks", [])) or False,
+                       timeout=90, what="the agent's question block") is not None,
+              "the pre-plan question surfaces as a goal block")
+        check(wait_task_status(auth, run_task, "blocked", timeout=30) is not None,
+              "the goal-run task parks blocked on the mirrored block")
+        question = [b for b in goal_state(auth, goal1)["blocks"]
+                    if b["block_type"] == "question" and not b.get("answered")][0]
+        check(question["options"] == ["Demo only", "Whole workspace"],
+              "the question carries its options")
+        check("Which scope should the greeting cover" in question["question"],
+              "the question title derives from the body's first line")
+        check("confirm the blast radius" in question["body"], "the question carries the markdown body")
+        dupes = db_query(
+            "select count(*) from goal_blocks where goal_id = ? and type = 'question'", (goal1,))[0][0]
+        check(dupes == 1, "one question row despite the park-unwinding round-end push")
+        dash = slot_public(auth, "/", "dashboard")
+        check(any(int(r.get("goal_id") or 0) == goal1 for r in dash["pinned"]),
+              "the dashboard pile pins the goal block")
+        act(auth, question["answer_action"], {"option": "Demo only"})
+        answered_goal_rows = db_query(
+            "select answer from goal_blocks where goal_id = ? and type = 'question'",
+            (goal1,))
+        mirrored_rows = db_query(
+            "select answer from task_blocks tb join tasks t on t.id = tb.task_id"
+            " where t.goal_id = ? and t.kind = 'goal_run' and tb.type = 'question'",
+            (goal1,))
+        check(answered_goal_rows and answered_goal_rows[0][0] == "Demo only",
+              "the ONE answer lands on the goal_blocks row")
+        check(mirrored_rows and mirrored_rows[0][0] == "Demo only",
+              "the same answer resolves the mirrored task_blocks row")
+
+        print("== plan review: reject -> revise -> approve ==")
+        check(wait_goal_status(auth, goal1, "plan_review", timeout=90) is not None,
+              "the plan submission moves the goal to plan_review")
+        check(wait_for(lambda: goal_state(auth, goal1).get("plan_pending_block") not in
+                       (None, False) or False, timeout=30, what="pending plan block") is not None,
+              "the pending plan block renders on the goal page")
+        pending = goal_state(auth, goal1)["plan_pending_block"]
+        check("Plan v1" in pending["body"], "the plan block carries the submitted markdown")
+        act(auth, pending["answer_action"], {"answer": "Feedback: also export the greeting"})
+        check(wait_goal_status(auth, goal1, "planning", timeout=30) is not None,
+              "rejecting with feedback returns the goal to planning")
+        check(wait_for(lambda: goal_state(auth, goal1).get("plan_pending_block") not in
+                       (None, False) and "Plan v2" in
+                       goal_state(auth, goal1)["plan_pending_block"].get("body", "") or False,
+                       timeout=90, what="revised plan") is not None,
+              "the agent revises and re-submits incorporating the feedback")
+        revised = goal_state(auth, goal1)
+        check(len([r for r in revised["plan_revisions"] if r.get("answered")]) >= 1,
+              "the rejected revision stays discoverable behind the expander")
+        act(auth, revised["plan_pending_block"]["answer_action"], {"answer": PLAN_APPROVED})
+        check(wait_goal_status(auth, goal1, "active", timeout=30) is not None,
+              "approving with the agreed wording activates the goal")
+
+        print("== task fan-out via the real callbacks ==")
+        worker_row = wait_for(lambda: db_query(
+            "select id from tasks where goal_id = ? and kind = 'task'", (goal1,)) or False,
+            timeout=90, what="the agent-created task")
+        worker_task = int(worker_row[0][0])
+        check(bool(worker_row), "the goal agent created its worker task through the callback")
+        creator = db_query("select creator, mode from tasks where id = ?", (worker_task,))[0]
+        check(creator == ("goal", "implement"),
+              "the created task carries creator=goal and the requested mode")
+        check(wait_task_status(auth, worker_task, "queued", timeout=30) is not None
+              or task_state(auth, worker_task)["status"] in ("working", "done"),
+              "the agent enqueued its task (creator-only rule honoured by construction)")
+        ephemeral = "task/%d-%d" % (goal1, worker_task)
+        check(wait_for(lambda: task_state(auth, worker_task)["status"]
+                       in ("assigned", "working", "done", "failed") or False,
+                       timeout=60, what="worker dispatched") is not None,
+              "the dispatcher hands the worker to the freed saddle")
+        check(wait_for(lambda: db_query(
+            "select suspended_at is not null from tasks where id = ?", (run_task,))[0][0] or False,
+            timeout=30, what="the goal agent parks") is not None,
+            "the goal agent parked suspended after fanning out (suspended_at stamped)")
+
+        print("== the worker's ephemeral worktree ==")
+        branch_rows = wait_for(lambda: db_query(
+            "select sc.output_text from session_commands sc join sessions s"
+            " on s.id = sc.session_id where s.task_id = ?"
+            " and sc.command_text like '%git branch --show-current%'", (worker_task,)) or False,
+            timeout=90, what="the worker's branch probe")
+        check(branch_rows and ephemeral in (branch_rows[0][0] or ""),
+              "the worker runs inside its ephemeral branch %s" % ephemeral)
+        worktree_dir = os.path.join(WORKSPACE, "task-%d" % worker_task)
+        check(wait_for(lambda: task_state(auth, worker_task)["status"] == "done" or False,
+                       timeout=120, what="worker done") is not None,
+              "the worker completes with its report")
+        detail = task_state(auth, worker_task)
+        check("greeting.txt" in (detail.get("summary") or ""),
+              "the worker summary carries its completion report")
+        check(not os.path.exists(worktree_dir), "the worker's worktree auto-pruned on done")
+        kept = subprocess.run(["git", "-C", goal_clone, "rev-parse", "--verify", ephemeral],
+                              capture_output=True).returncode == 0
+        check(kept, "the ephemeral branch survives in the goal clone for the merge")
+
+        print("== coalesced resume -> merge -> push -> completion review ==")
+        check(wait_for(lambda: db_query(
+            "select 1 from goal_events where goal_id = ? and kind = 'task-done'",
+            (goal1,)) or False, timeout=30, what="the task-done event") is not None,
+              "the worker's terminal event lands in the goal narrative")
+        # the goal agent's /status reports the branch push state while it
+        # works; sample the moment the resumed agent is mid-round, then
+        # cache it through the manual saddle refresh
+        check(wait_for(lambda: (saddle_status() or {}).get("goal_branches") or False,
+                       timeout=90, what="the resumed agent exposes goal_branches") is not None,
+              "the daemon's /status exposes goal_branches while the goal agent works")
+        saddles_state = slot_public(auth, "/saddles", "saddles")
+        card = [s for s in saddles_state["saddles"] if s["name"] == SADDLE_NAME][0]
+        act(auth, card["refresh"], {})
+        check(wait_for(lambda: any("demo" in (b.get("branch") or "")
+                                   for b in goal_state(auth, goal1).get("push_branches", []))
+                       or False, timeout=60, what="cached push state") is not None,
+              "the refresh probe caches the branch push state on the goal")
+        check(wait_goal_status(auth, goal1, "completion_review", timeout=120) is not None,
+              "the resumed agent merges, pushes and enters completion review")
+        state = goal_state(auth, goal1)
+        check("greeting.txt added via one worker task" in state.get("completion_report", ""),
+              "the completion report renders on the goal page")
+        merged = subprocess.run(
+            ["git", "-C", goal_clone, "show", "goal/demo-run:greeting.txt"],
+            capture_output=True, text=True).stdout
+        check("hello-from-the-goal" in merged, "the merged task branch landed on the goal branch")
+        deleted = subprocess.run(["git", "-C", goal_clone, "rev-parse", "--verify", ephemeral],
+                                 capture_output=True).returncode != 0
+        check(deleted, "the goal agent deleted the merged task branch")
+        pushed = subprocess.run(["git", "-C", DEMO, "rev-parse", "--verify", "goal/demo-run"],
+                                capture_output=True).returncode == 0
+        check(pushed, "the goal branch was pushed to origin (push-before-review)")
+
+        print("== completion accept ==")
+        completion = state["completion_pending_block"]
+        check(completion and completion.get("body", "") == state.get("completion_report", ""),
+              "the completion block mirrors the report")
+        act(auth, completion["answer_action"], {"answer": REVIEW_APPROVED})
+        check(wait_goal_status(auth, goal1, "complete", timeout=30) is not None,
+              "accepting with the agreed wording completes the goal")
+        check(wait_task_status(auth, run_task, "done", timeout=30) is not None,
+              "the goal-run task closes done with the goal")
 
         print("== audit trail ==")
-        db = sqlite3.connect(DB_PATH)
-        rows = db.execute(
-            "select command_text, output_text, exit_code from session_commands order by id").fetchall()
-        commands = {row[0]: row for row in rows}
-        check(any("ls -la" in c and "hello.py" in (r[1] or "") for c, r in commands.items()),
-              "ls output captured in session_commands")
-        check(any("git" in c and "Initial commit" in (r[1] or "") and r[2] == 0
-                  for c, r in commands.items()),
-              "git log output captured with exit code 0")
-        per_session = db.execute(
-            "select session_id, count(*) from session_commands group by session_id order by session_id"
-        ).fetchall()
-        check(all(count > 0 for _, count in per_session) and len(per_session) == 3,
-              "each session carries its own commands (%s)" % per_session)
-        progress_rows = db.execute(
-            "select summary from task_progress where task_id = 3 order by id").fetchall()
-        check([row[0] for row in progress_rows] == ["queried the host", "human input integrated",
-                                                    "human input integrated"],
-              "task_progress keeps the full history in order")
-        db.close()
+        kinds = [row[0] for row in db_query(
+            "select kind from goal_events where goal_id = ? order by id", (goal1,))]
+        for expected in ("started", "block-raised", "plan-submitted", "plan-rejected",
+                         "plan-approved", "task-created", "task-enqueued", "task-done",
+                         "completion-submitted", "goal-approved"):
+            check(expected in kinds, "goal_events records %s" % expected)
+        run_types = {row[0] for row in db_query(
+            "select distinct sc.type from session_commands sc join sessions s"
+            " on s.id = sc.session_id where s.task_id = ?", (run_task,))}
+        check({"bash", "block", "plan"} <= run_types,
+              "the goal agent's terminal carries typed entries (%s)" % sorted(run_types))
+        worker_types = {row[0] for row in db_query(
+            "select distinct sc.type from session_commands sc join sessions s"
+            " on s.id = sc.session_id where s.task_id = ?", (worker_task,))}
+        check({"bash", "done"} <= worker_types,
+              "the worker's terminal carries typed entries (%s)" % sorted(worker_types))
 
         print("== SSE ==")
-        db = sqlite3.connect(DB_PATH)
-        tour_session = db.execute(
-            "select s.id from sessions s where s.task_id = 3 order by s.id desc limit 1"
-        ).fetchone()[0]
-        db.close()
-        request = urllib.request.Request(WEB + "/terminal-stream/%d" % tour_session)
+        run_session = db_query(
+            "select sc.session_id from session_commands sc join sessions s on s.id = sc.session_id"
+            " where s.task_id = ? and sc.type = 'plan' order by sc.id limit 1",
+            (run_task,))[0][0]
+        state = goal_state(auth, goal1)
+        check(any(r.get("terminal_url", "").endswith("/terminal?session_id=%d" % run_session)
+                  for r in state.get("runs", [])),
+              "the goal page's run list deep-links each session terminal")
+        request = urllib.request.Request(WEB + "/terminal-stream/%d" % run_session)
         with urllib.request.urlopen(request, timeout=15) as stream:
             check(stream.headers.get("Content-Type", "").startswith("text/event-stream"),
                   "terminal stream is text/event-stream")
             entries = []
-            saw_piped_input = False
             deadline = time.monotonic() + 10
             while time.monotonic() < deadline and len(entries) < 120:
                 if any(e.get("type") == "done" for e in entries):
@@ -581,36 +611,19 @@ def main():
                 line = stream.readline().decode().strip()
                 if not line.startswith("data: ") or line == "data: ping":
                     continue
-                entry = json.loads(line[6:])
-                entries.append(entry)
-                if entry.get("type") == "query" and (entry.get("input") or "").strip():
-                    saw_piped_input = True
-            check(len(entries) >= 12, "terminal stream replays the backlog (%d entries)" % len(entries))
-            check(saw_piped_input, "terminal stream carries the query's piped input")
-            check(any(e.get("type") == "review" for e in entries)
-                  and any(e.get("type") == "user-comment" for e in entries),
-                  "terminal stream carries typed part 2 events")
-            db = sqlite3.connect(DB_PATH)
-            db.execute("insert into session_commands (session_id, command_text, output_text,"
-                       " type, started_at) values (?, 'sse-live-check', 'ok', 'command',"
-                       " strftime('%s','now'))", (tour_session,))
-            db.commit()
-            db.close()
-            got_live = False
-            deadline = time.monotonic() + 6
-            while time.monotonic() < deadline and not got_live:
-                line = stream.readline().decode().strip()
-                if line.startswith("data: ") and "sse-live-check" in line:
-                    got_live = True
-            check(got_live, "terminal stream pushes new entries within seconds")
-
-        pulse = directives("/tasks/3", auth)
+                entries.append(json.loads(line[6:]))
+            check(len(entries) >= 4, "terminal stream replays the run's backlog (%d entries)" % len(entries))
+            # push-era sessions stream their raw PTY log (the typed
+            # commands live in session_commands) — the plan submission
+            # shows as its echoed heredoc in the raw replay
+            check(any("saddle plan" in json.dumps(e) for e in entries),
+                  "terminal stream replay carries the plan submission round")
+
+        pulse = directives("/goals/%d" % goal1, auth)
         check(any(d.get("type") == "subscribe" for d in pulse),
-              "task page subscribes to the pulse channel for live refresh")
-
-        print("== part 2 API surface ==")
-        db = sqlite3.connect(DB_PATH)
+              "goal page subscribes to the pulse channel for live refresh")
 
+        print("== legacy ingest API surface ==")
         saddles_state = slot_public(auth, "/saddles", "saddles")
         act(auth, saddles_state["register"], {"name": "saddle-e2e-api", "host": "localhost:7199"})
         api_rows = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
@@ -623,21 +636,14 @@ def main():
         tasks_state = slot_public(auth, "/tasks", "tasks")
         act(auth, tasks_state["create"], {
             "project_id": "1",
-            "title": "Part 2 API surface",
-            "description": "Typed blocks, plan, progress, review, quoted comments, SSE",
-            "plan": "# Initial plan",
+            "title": "API surface probe",
+            "description": "Typed updates, exactly-once poll",
             "priority": "6",
-            "task_branch": "task/9-api-surface",
+            "task_branch": "task/api-surface",
         })
         api_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
-                        if t["title"] == "Part 2 API surface"][0])
-        detail = task_state(auth, api_task)
-        check(detail["branch_from"] == "main",
-              "branch_from prefilled from the project default branch")
-        check(detail["task_branch"] == "task/9-api-surface", "task_branch stored")
-
-        # Dispatch bookkeeping without a daemon: the ingest API is under
-        # test, so the session and saddle/task state are staged directly.
+                        if t["title"] == "API surface probe"][0])
+        db = sqlite3.connect(DB_PATH)
         db.execute("insert into sessions (task_id, saddle_id, started_at)"
                    " values (?, ?, strftime('%s','now'))", (api_task, api_saddle))
         db.execute("update saddles set current_task_id = ?, status = 'idle' where id = ?",
@@ -645,6 +651,7 @@ def main():
         db.execute("update tasks set status = 'working' where id = ?", (api_task,))
         db.commit()
         api_session = db.execute("select max(id) from sessions").fetchone()[0]
+        db.close()
 
         status, _ = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, None,
                                {"type": "comment", "session_id": api_session, "data": {"body": "x"}})
@@ -652,8 +659,6 @@ def main():
         status, _ = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, "saddle-not-a-token",
                                {"type": "comment", "session_id": api_session, "data": {"body": "x"}})
         check(status == 401, "update with an invalid token is rejected")
-        status, _ = saddle_api("GET", "/api/saddle/%d/poll" % api_saddle, None)
-        check(status == 401, "poll without a token is rejected")
 
         status, body = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, api_token,
                                   {"type": "comment", "session_id": api_session,
@@ -673,10 +678,6 @@ def main():
                                    "type": "query",
                                    "input": "# notes\nmeaning of life"}})
         check(status == 200, "query output update accepted")
-        status, _ = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, api_token,
-                               {"type": "plan", "session_id": api_session,
-                                "data": {"plan": "# Plan\n- migrate schemas"}})
-        check(status == 200, "plan update accepted")
         status, _ = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, api_token,
                                {"type": "progress", "session_id": api_session,
                                 "data": {"summary": "Migrated the schemas"}})
@@ -685,41 +686,33 @@ def main():
                                {"type": "review", "session_id": api_session,
                                 "data": {"summary": "Migration **ready** for review"}})
         check(status == 200, "review update accepted")
+        status, body = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, api_token,
+                                  {"type": "plan", "session_id": api_session,
+                                   "data": {"plan": "# Plan\n- migrate schemas"}})
+        check(status == 400 and "gone" in (body.get("error") or ""),
+              "the task-plan update type fails loudly (plan 12 removal)")
+        status, body = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, api_token,
+                                  {"type": "subtask", "session_id": api_session, "data": {}})
+        check(status == 400 and "gone" in (body.get("error") or ""),
+              "the subtask update type fails loudly (plan 12 removal)")
 
         detail = task_state(auth, api_task)
         check(detail["status"] == "review_requested", "review flips the task status")
         check([b["options"] for b in detail["blocks"]
                if b["block_type"] == "question"] == [["SQLite", "Postgres"]],
               "question block carries its options")
-        check("- migrate schemas" in detail["plan"], "plan visible on the task page")
         check(detail["progress"][0]["summary"] == "Migrated the schemas",
               "progress history newest-first")
-        check(detail["review_pending"] and "ready" in detail["review_summary"],
-              "review banner data on the task page")
-        check(detail["review_diff_url"].endswith(
-            "demo.git/compare/main...task/9-api-surface"), "review diff link constructed")
-
-        dash = slot_public(auth, "/", "dashboard")
-        check(any(int(r["task_id"]) == api_task and r["block_type"] == "review"
-                  for r in dash["pinned"]),
-              "dashboard pins the pending review")
-        check(any(b["question"] == "Which database?" for b in dash["pinned"]),
-              "dashboard pins blocks needing answers")
-
-        piped = db.execute("select piped_input from session_commands where session_id = ?"
-                           " and type = 'query'", (api_session,)).fetchall()
-        check(piped and "meaning of life" in (piped[0][0] or ""),
-              "push API stores the query's piped input too")
 
         question = [b for b in detail["blocks"] if b["block_type"] == "question"][0]
         act(auth, question["answer_action"], {"option": "SQLite"})
         check([b["answer"] for b in task_state(auth, api_task)["blocks"]
-               if b["block_type"] == "question"] == ["Human selected: SQLite"],
-              "option answer formatted for the agent")
+               if b["block_type"] == "question"] == ["SQLite"],
+              "option answer stored verbatim (the resume notice formats it)")
         act(auth, detail["comment"],
             {"body": "Check the migration output", "quote": "ERROR: table demos exists"})
         status, body = saddle_api("GET", "/api/saddle/%d/poll" % api_saddle, api_token)
-        check(any(a["answer"] == "Human selected: SQLite" for a in body.get("answers", [])),
+        check(any(a["answer"] == "SQLite" for a in body.get("answers", [])),
               "poll hands over the block answer")
         check(any(c["body"] == "Check the migration output" and "ERROR" in c["quote"]
                   for c in body.get("comments", [])),
@@ -727,24 +720,19 @@ def main():
         status, body = saddle_api("GET", "/api/saddle/%d/poll" % api_saddle, api_token)
         check(body.get("answers") == [] and body.get("comments") == [],
               "poll delivers answers and comments exactly once")
-
-        entry_types = {row[0] for row in db.execute(
-            "select type from session_commands where session_id = ?", (api_session,))}
-        check({"comment", "block", "plan", "progress", "review", "user-comment"} <= entry_types,
-              "conversation events logged as typed terminal entries")
-
         act(auth, detail["cancel"], {"task_id": str(api_task)})
-        db.close()
 
         print("== stm-markdown ==")
         markdown_renderer_checks()
 
         print("== quiesce ==")
         check(wait_for(
-            lambda: all(int(s["current_task_id"] or 0) == 0 and s["status"] == "idle"
-                        for s in slot_public(auth, "/saddles", "saddles")["saddles"]) or False,
+            lambda: (lambda rows: all(int(s["current_task_id"] or 0) == 0 and s["status"] == "idle"
+                                      for s in rows))(
+                [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
+                 if s["name"] == SADDLE_NAME]) or False,
             timeout=30, what="saddle idle with no current task") is not None,
-            "saddle freed and idle once every task finished")
+            "saddle freed and idle once the goal finished")
     finally:
         if mock is not None:
             mock.terminate()
@@ -753,15 +741,5 @@ def main():
     return 1 if failures else 0
 
 
-def query_commands(task_id):
-    db = sqlite3.connect(DB_PATH)
-    try:
-        return db.execute(
-            "select sc.type from session_commands sc join sessions s on s.id = sc.session_id"
-            " where s.task_id = ? and sc.type = 'query'", (task_id,)).fetchall()
-    finally:
-        db.close()
-
-
 if __name__ == "__main__":
     sys.exit(main())

test/web_stack_test.py

@@ -5411,11 +5411,18 @@ def bundle_goals(ctx):
           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','+1 second'), strftime('%s','now'))", (goal1,))
+    quiet_id = conn.execute("select max(id) from tasks").fetchone()[0]
+    conn.execute("insert into goal_events (goal_id, kind, text, task_id, created_at)"
+                 " values (?, 'task-done', 'Task #? Quiet task — done: finished while"
+                 " paused', ?, strftime('%s','now','+1 second'))", (goal1, quiet_id))
     conn.commit()
     conn.close()
     paused_runs = len(fake.goal_assignments)