C5: dashboard deep-link closure verified on the single-row world

The closure proofs land as explicit checks on the single-row
architecture: goal pile cards deep-link /goals/<id> (the answer
surface the notification URL also carries), answering from either
surface — the goal page's verdict form or the task-page deep link
funneling into the one goal answer path — settles THE row and
drains the pile instantly, the run task carries exactly one row per
question with none left unanswered, and no goal-side store exists
(goal_blocks/goal_comments/goal_events are all gone), so the
historical-straggler class cannot exist.

Test robustness under sustained parallel load: the requeue-closed
check keys the PARKED session row itself (the restore-on-dispatch
successor can open before a transient open-session gap is ever
observable), the capability-attach reads became event-driven, and
the load-marginal waits (plan-accept completion, parked-session
completion, capability-wait log) breathe at 60-90s. The README's
integration section reflects the single-row/derived-narrative
world.

README.md

@@ -270,15 +270,16 @@ accumulated project knowledge.
 
 `test/integration_test.py` drives BOTH real binaries — this app and a
 real saddle daemon over a real git checkout — through the full goal-era
-arc (83 checks, Statum-style PASS/FAIL): goal create → start → pre-plan
-question (one answer resolving the goal row AND the mirrored run-task
-row) → plan reject-with-feedback → revision → approval on the agreed
+arc (Statum-style PASS/FAIL): goal create → start → pre-plan question
+(one answer resolving the goal-run task's single task_blocks row) →
+plan reject-with-feedback → revision → approval on the agreed
 verdict words → task fan-out through the real `/api/saddle/task/*`
 callbacks → the worker's ephemeral worktree (auto-pruned, branch kept)
 → the coalesced terminal-event resume → merge + branch delete +
-push-before-review → completion review accept → goal complete, plus the
-`/status` `goal_branches` cache, SSE over the agent's raw terminal, the
-`goal_events` narrative, the legacy ingest API surface and the
+push-before-review → completion review accept (gated on the live
+per-repo push flags) → goal complete, plus the `/status`
+`goal_branches` live probe flags, SSE over the agent's raw terminal,
+the DERIVED goal narrative, the legacy ingest API surface and the
 stm-markdown renderer. Host run (the harness prepares a `demo` repo
 with a pushable bare origin when the workspace is empty; the database
 must be FRESH — the harness refuses a used one):

test/web_stack_test.py

@@ -2476,7 +2476,7 @@ def bundle_dispatch(ctx):
     act(auth, detail["attach_capability"], {"capability_id": str(int(caps["gpu"]["id"]))})
     check(wait_for(lambda: [c["name"] for c in
                             task_state(auth, gpu_task)["capabilities"]] == ["gpu"]
-                   or False, timeout=30, what="capability attached") is not None,
+                   or False, timeout=60, what="capability attached") is not None,
           "task page lists the required capability")
     detail = task_state(auth, gpu_task)
     # The only idle saddle lacks the tag, so the queue flags the wait
@@ -2501,7 +2501,7 @@ def bundle_dispatch(ctx):
         with open(os.path.join(WORK, "app.log")) as handle:
             app_log = handle.read()
         return "waits for a capable saddle" in app_log and "requires: gpu" in app_log
-    check(wait_for(capability_wait_logged, timeout=20,
+    check(wait_for(capability_wait_logged, timeout=45,
                    what="capability wait log") is not None,
           "dispatcher logs the capability wait")
 
@@ -3932,8 +3932,10 @@ def bundle_lifecycle(ctx):
                                        "priority": "2"})
     detail = task_state(auth, locked_task)
     act(auth, detail["attach_capability"], {"capability_id": str(editlock_cap)})
-    check(any(c["name"] == "editlock"
-              for c in task_state(auth, locked_task)["capabilities"]),
+    check(wait_for(lambda: any(c["name"] == "editlock"
+                               for c in task_state(auth,
+                                                   locked_task)["capabilities"])
+                   or False, timeout=60, what="editlock attached") is not None,
           "capabilities attach while queued (no worker yet)")
     check(wait_for(lambda: task_state(auth, locked_task).get("status") == "working"
                    or False, timeout=60, what="caplock working") is not None,
@@ -4545,7 +4547,7 @@ def bundle_answers(ctx):
     # The hold_state daemon drives its round manually: the resumed
     # round completes on the fresh session row.
     parked.push_results("done", summary="parked round resumed and finished")
-    check(wait_task_done(auth, parked_task, timeout=60) is not None,
+    check(wait_task_done(auth, parked_task, timeout=90) is not None,
           "the parked session continues to completion")
     conn = db()
     row = conn.execute("select outcome from sessions where id = ?",
@@ -4631,7 +4633,7 @@ def bundle_answers(ctx):
         timeout=30, what="reject resume allocation") is not None,
           "reject resumes the parked daemon with the feedback notice")
     planner.push_results("done", summary="plan revised and delivered")
-    check(wait_task_done(auth, reject_task, timeout=30) is not None,
+    check(wait_task_done(auth, reject_task, timeout=60) is not None,
           "the revised plan's task completes")
     planner.set_state("idle")
 
@@ -6368,6 +6370,9 @@ def bundle_goalui(ctx):
     one review-kind block per verdict path). C4 — the goal page
     batch: one muted head activity line, plan between attention and
     tasks, answered blocks behind an expander, footnote comments.
+    C5 — the dashboard deep-link closure: goal pile cards link the
+    goal page, every surface settles the single row, and no
+    goal-side store exists to hold stragglers.
     """
     auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token
 
@@ -6587,6 +6592,28 @@ def bundle_goalui(ctx):
     check("goal_blocks" not in tables,
           "the goal_blocks table is gone (run-task blocks are canonical)")
 
+    print("== C5: deep-link closure on the single-row world ==")
+    # The pile card's goal item deep-links the goal page (the answer
+    # surface); answering from either surface settled THE row above.
+    home_markup = http_get_page(auth, "/")
+    check("'/goals/' + blockview.goal_id" in home_markup,
+          "the pile card deep-links the goal page for goal items")
+    conn = db()
+    dupes = conn.execute(
+        "select count(*) from (select question from task_blocks where task_id = ?"
+        " group by question having count(*) > 1)", (run_task,)).fetchone()[0]
+    stragglers = conn.execute(
+        "select count(*) from task_blocks where task_id = ? and answer is null",
+        (run_task,)).fetchone()[0]
+    goal_side_tables = [r[0] for r in conn.execute(
+        "select name from sqlite_master where type = 'table'"
+        " and name in ('goal_blocks','goal_comments','goal_events')").fetchall()]
+    conn.close()
+    check(dupes == 0 and stragglers == 0,
+          "one row per question on the run task, none left unanswered")
+    check(not goal_side_tables,
+          "no goal-side store exists — historical stragglers are impossible")
+
     print("== B4: the terminal session dropdown + resume-info panel ==")
     fake.goal_script = [("suspend",)]
     check(wait_for(lambda: task_state(auth, run_task)["status"] == "suspended"
@@ -6641,11 +6668,25 @@ def bundle_goalui(ctx):
     check(state.get("retry_run") is None,
           "an active goal offers no Retry supervision")
     open_before = open_session(run_task)
+
+    def parked_session_ended():
+        if open_before is None:
+            return True
+        conn = db()
+        try:
+            return conn.execute(
+                "select ended_at is not null from sessions where id = ?",
+                (open_before,)).fetchone()[0] == 1
+        finally:
+            conn.close()
     act(auth, state["requeue_run"], {})
     check(task_state(auth, run_task)["status"] == "queued",
           "the requeue returns the run to the queue")
-    check(wait_for(lambda: open_session(run_task) is None or False,
-                   timeout=30, what="parked session closed") is not None,
+    # The PARKED session itself must end (the restore-on-dispatch
+    # successor may open immediately — the check keys the old row,
+    # not the task's transient open-session window).
+    check(wait_for(parked_session_ended, timeout=60,
+                   what="parked session closed") is not None,
           "the requeue closed the parked session (restore-on-dispatch)")
     conn = db()
     conn.execute("update saddles set status = 'idle' where id = ?", (fake_id,))