r23: D4 web half — parks close their session rows 'parked'

Sessions become the honest record of a round's lifecycle, aligned with
the daemon's file-log brackets (its c22942c): every park — a raised
block (add_block's guarded flip, entry-driven raises included), a
suspend (set_parked), a park-unwinding blocked push — closes the
round's open sessions with the new OUTCOME_PARKED ('parked', not the
removed 'resumed': R2 deferred the constant until a writer existed;
the park path is the writer). The verdict arrives as a resuming
allocation on a FRESH row (the claim's parked path keys on the task's
newest session, open or closed, closing any straggler open row), so no
raise-side path can leave a dangling open session (the friction run's
session-11 shape) and the verdict paths inherit closure — the accept's
release finds closed rows. A 'working' state push only advances an
assigned (or legacy never-allocated queued) task: a parked task keeps
its park, never resurrected as working out from under its review
banner. Answering a block flips to working only when the task NEVER
ran (a parked round has session rows and stays parked for the affinity
resume). The goal page's current-activity panel and its
Requeue-to-resume affordance read the newest session, open or 'parked'
— a parked round keeps its hosting-saddle line and its dead-saddle
requeue.

Tests: answers' resume pins flip to the 'parked' outcome and pin the
close at park time; the offline-keep-parked pins key the parked record
(outcome) instead of open-ness; parked-agent session lookups
(allocation/goals/goalui/replay/lineage/subtasks) take the newest row;
goalui's requeue pins the parked row staying closed; goals pins the
accept leaving zero dangling open sessions and every parked round
'parked'. 855 checks (was 851).

src/Services/AllocationService.vala

@@ -13,7 +13,9 @@ namespace ClankerSaddle.Services {
      * (dependencies met, {@link affine_saddle_id} 0) or a parked task
      * ready to resume — blocked/review-requested with its newest block
      * answered, or suspended with a due goal event behind it
-     * parked — pinned to the saddle holding its open session.
+     * parked — pinned to the saddle holding its session record
+     * (plan 16 D4: parked rounds close 'parked'; the row still names
+     * the workspace's saddle).
      */
     public class OfferableTask : Object {
         public TaskEntity task { get; set; }
@@ -113,7 +115,7 @@ namespace ClankerSaddle.Services {
          * Every offerable task (see {@link OfferableTask}) — the
          * dispatcher's wake computation walks this list: parked tasks
          * target their affine saddle first, fresh tasks any capable
-         * sleeping one. A parked task without an open session has no
+         * sleeping one. A parked task without a session record has no
          * workspace to resume and is not offerable.
          */
         public async ImmutableLot<OfferableTask> offerable_tasks() throws Error {
@@ -175,11 +177,11 @@ namespace ClankerSaddle.Services {
 
         /**
          * A parked offerable bound to its session's saddle, or null when
-         * the task has no open (parked) session — nothing to resume.
+         * the task has no session record — nothing to resume.
          */
         private async OfferableTask? parked_offerable(TaskEntity task, bool suspended_resume)
-                throws Error {
-            var session = yield tasks.open_session_for_task(task.id);
+                 throws Error {
+            var session = yield tasks.newest_session_for_task(task.id);
             if (session == null) {
                 return null;
             }
@@ -228,26 +230,30 @@ namespace ClankerSaddle.Services {
                             elem<int64?>(candidate.task.id), elem<int64?>(saddle.id)))
                         .order_by(expr("id"))
                         .materialise();
-                    Session? newest_open = null;
-                    foreach (var open in session) {
-                        if (open.ended_at == null) {
-                            newest_open = open;
-                        }
+                    Session? newest = null;
+                    foreach (var prior in session) {
+                        newest = prior;
                     }
-                    if (newest_open == null) {
-                        // The parked session closed underneath us (a
-                        // concurrent requeue); nothing to resume onto.
+                    if (newest == null) {
+                        // The parked workspace's session record is gone
+                        // (a concurrent teardown); nothing to resume
+                        // onto.
                         connection.execute("ROLLBACK");
                         return null;
                     }
-                    // Plan 11: the resuming assignment carries a FRESH
-                    // session id — the parked row closes with outcome
-                    // 'resumed' (the park record) and the daemon keys
-                    // every telemetry push of the resumed round on the
-                    // new row.
-                    ((!)newest_open).ended_at = new DateTime.now_local();
-                    ((!)newest_open).outcome = OUTCOME_RESUMED;
-                    db.update<Session>((!)newest_open);
+                    if (((!)newest).ended_at == null) {
+                        // Plan 16 D4: a straggler open row (a raise
+                        // racing this claim) closes 'parked' like
+                        // every park — the record never dangles.
+                        ((!)newest).ended_at = new DateTime.now_local();
+                        ((!)newest).outcome = OUTCOME_PARKED;
+                        db.update<Session>((!)newest);
+                    }
+                    // Plan 11 + D4: the resuming assignment carries a
+                    // FRESH session id — the parked row closed with
+                    // outcome 'parked' at its park (the park record)
+                    // and the daemon keys every telemetry push of the
+                    // resumed round on the new row.
                     var resumed = new Session() {
                         task_id = candidate.task.id,
                         saddle_id = saddle.id,

src/Services/SaddleService.vala

@@ -20,14 +20,16 @@ namespace ClankerSaddle.Services {
     public const string SADDLE_OFFLINE = "offline";
 
     /**
-     * Session outcomes (the `sessions.outcome` values the ONE release
-     * path and the allocation claim write): `stopped` (a
-     * cancelled/requeued/unreachable release), `resumed` (a parked
-     * workspace handed to its successor session), `done`/`failed`
-     * (the terminal pushes).
+     * Session outcomes (the `sessions.outcome` values the park, the
+     * ONE release path and the terminal pushes write): `parked` (a
+     * round that parked on a block or a suspend — the row closes at
+     * park time and the workspace awaits its resuming allocation;
+     * plan 16 D4, aligned with the daemon's file-log brackets),
+     * `stopped` (a cancelled/requeued/unreachable release),
+     * `done`/`failed` (the terminal pushes).
      */
     public const string OUTCOME_STOPPED = "stopped";
-    public const string OUTCOME_RESUMED = "resumed";
+    public const string OUTCOME_PARKED = "parked";
     public const string OUTCOME_DONE = "done";
     public const string OUTCOME_FAILED = "failed";
 
@@ -1499,8 +1501,10 @@ namespace ClankerSaddle.Services {
          * blocks ride the run task — no mirrored second row exists)
          * and parks the saddle; `suspended`
          * parks the same way — for a goal agent the resume is driven
-         * by the goal's task-terminal events; and `working` flips the
-         * task to working (never downgrading a pending review).
+         * by the goal's task-terminal events; and `working` flips an
+         * ASSIGNED task to working (a parked task keeps its park —
+         * the verdict arrives as a resuming allocation, never as a
+         * state push; plan 16 D4).
          *
          * Plan 13 B2: a push against a task that is already terminal —
          * or a goal-context push against a finished goal — is an
@@ -1526,7 +1530,9 @@ namespace ClankerSaddle.Services {
                         // Goal context: `saddle done` submits the
                         // completion report and enters completion
                         // review — the run parks on the mirrored
-                        // block, it does not complete.
+                        // block (submit_completion's raise already
+                        // closed the session 'parked'), it does not
+                        // complete.
                         yield goal_service.submit_completion(goal_id,
                             summary != null && summary.length > 0 ? summary
                                 : "The goal agent submitted its completion report");
@@ -1558,6 +1564,7 @@ namespace ClankerSaddle.Services {
                         // and a synthesised fallback row would dupe it.
                         if (block == null) {
                             yield tasks.set_status(task_id, STATUS_BLOCKED);
+                            yield tasks.park_open_sessions(task_id);
                             yield park_if_current(saddle.id, task_id);
                             break;
                         }
@@ -1605,11 +1612,30 @@ namespace ClankerSaddle.Services {
                     yield park_if_current(saddle.id, task_id);
                     break;
                 case "working":
-                    // A pending review keeps its banner: started_at (and
-                    // the first-start push) only stamp for work that
-                    // never allocated.
-                    var current = yield tasks.set_status(task_id, STATUS_WORKING);
-                    if (current != null && ((!)current).status != STATUS_REVIEW_REQUESTED
+                    // A working push only advances a task that is
+                    // actually running — assigned (the claim's flip)
+                    // or a legacy never-allocated queued row. A
+                    // PARKED task (blocked / review / suspended, its
+                    // session closed at the raise — plan 16 D4) keeps
+                    // its park: the verdict arrives as a resuming
+                    // allocation, never as a state push, so a
+                    // park-unwinding or late round push can never
+                    // resurrect the workspace as "working" out from
+                    // under its review banner.
+                    var advance = connection.create_command(
+                        "UPDATE tasks SET status = 'working' WHERE id = :id"
+                        + " AND status IN ('assigned', 'queued')");
+                    advance.with_parameter<Invercargill.Element>("id",
+                        elem<int64?>(task_id));
+                    yield advance.execute_non_query_async();
+                    var current = yield db.query<TaskEntity>()
+                        .where(expr("id == $0", elem<int64?>(task_id)))
+                        .first_async();
+                    // started_at (and the first-start push) only stamp
+                    // for work that never allocated; the guarded
+                    // advance above means the re-read task is working
+                    // exactly when the flip landed or it already ran.
+                    if (current != null && ((!)current).status == STATUS_WORKING
                             && ((!)current).started_at == null) {
                         ((!)current).started_at = new DateTime.now_local();
                         yield db.update_async<TaskEntity>((!)current);

src/Services/TaskService.vala

@@ -917,6 +917,13 @@ namespace ClankerSaddle.Services {
                 guard.with_parameter<Invercargill.Element>("id",
                     elem<int64?>(task_id));
                 yield guard.execute_non_query_async();
+                // Plan 16 D4: a raised block parks the round — the
+                // session row CLOSES 'parked' right here (every raise
+                // path funnels through this flip, entry-driven raises
+                // included, so no parked task ever keeps a dangling
+                // open session). The workspace awaits its resuming
+                // allocation on a fresh row.
+                yield park_open_sessions(task_id);
             }
             yield pulse.bump(task_id, "block raised");
             // Web Push (plan 11 task E): every raised block notifies all
@@ -941,10 +948,12 @@ namespace ClankerSaddle.Services {
         }
 
         /**
-         * Records a human's answer to a block. With an open session the
-         * task stays 'blocked'/'review_requested' — the dispatcher resumes
-         * the paused workspace and hands the answer over; without one the
-         * task flips straight back to 'working'.
+         * Records a human's answer to a block. A parked round (its
+         * session closed 'parked' at the raise — plan 16 D4) stays
+         * 'blocked'/'review_requested': the dispatcher resumes the
+         * paused workspace and hands the answer over. Only a block on
+         * a task that NEVER RAN (no session rows at all) flips
+         * straight back to 'working'.
          */
         public async void answer_block(int64 block_id, string answer,
                 string? answered_by = null) throws Error {
@@ -967,7 +976,8 @@ namespace ClankerSaddle.Services {
             if (task != null && (((!)task).status == STATUS_BLOCKED
                     || ((!)task).status == STATUS_REVIEW_REQUESTED)) {
                 var session = yield open_session_for_task(((!)block).task_id);
-                if (session == null) {
+                if (session == null
+                        && (yield newest_session_for_task(((!)block).task_id)) == null) {
                     ((!)task).status = STATUS_WORKING;
                     yield db.update_async<TaskEntity>((!)task);
                 }
@@ -1188,6 +1198,9 @@ namespace ClankerSaddle.Services {
          * `suspended_at` stamp, stamped once per parking window so the
          * resume trigger only sees goal events that landed after the
          * suspension. No-op when already parked (the stamp survives).
+         * Plan 16 D4: parking closes the round's open sessions
+         * ('parked') — the pause record the resume machinery keys on
+         * is the newest session row, open or closed.
          */
         public async void set_parked(int64 task_id) throws Error {
             var task = yield db.query<TaskEntity>()
@@ -1199,6 +1212,7 @@ namespace ClankerSaddle.Services {
             ((!)task).status = STATUS_SUSPENDED;
             ((!)task).suspended_at = new DateTime.now_local();
             yield db.update_async<TaskEntity>((!)task);
+            yield park_open_sessions(task_id);
             yield pulse.bump(task_id, "task suspended");
         }
 
@@ -1598,6 +1612,44 @@ namespace ClankerSaddle.Services {
             return null;
         }
 
+        /**
+         * The task's newest session row, open or closed (plan 16 D4:
+         * a parked round's row is closed 'parked' — it is still the
+         * pause record the affinity resume, the session-scoped task
+         * CLI and the goal page's run surfaces key on), or null when
+         * the task never ran.
+         */
+        public async Session? newest_session_for_task(int64 task_id) throws Error {
+            Session? newest = null;
+            foreach (var session in yield db.query<Session>()
+                    .where(expr("task_id == $0", elem<int64?>(task_id)))
+                    .order_by(expr("id"))
+                    .materialise_async()) {
+                newest = session;
+            }
+            return newest;
+        }
+
+        /**
+         * Closes the task's open sessions with outcome 'parked'
+         * (plan 16 D4 — the ONE park close every raise and suspend
+         * path runs; idempotent, nothing to do for rows already
+         * closed). The workspace stays on its saddle awaiting the
+         * resuming allocation that opens the successor row.
+         */
+        public async void park_open_sessions(int64 task_id) throws Error {
+            foreach (var session in yield db.query<Session>()
+                    .where(expr("task_id == $0", elem<int64?>(task_id)))
+                    .materialise_async()) {
+                if (session.ended_at != null) {
+                    continue;
+                }
+                session.ended_at = new DateTime.now_local();
+                session.outcome = OUTCOME_PARKED;
+                yield db.update_async<Session>(session);
+            }
+        }
+
         /**
          * The ONE release path (plan 16 R2, arch item 13): every path
          * that takes a task off its saddle runs through here. Closes

src/entrypoints/GoalDetailEntrypoint.vala

@@ -710,11 +710,15 @@ namespace ClankerSaddle {
                         || run_status == STATUS_REVIEW_REQUESTED
                         || run_status == STATUS_SUSPENDED;
                     if (!pub.terminal && stale_or_parked) {
-                        var open_run = yield ctx.tasks.open_session_for_task(pub.run_task_id);
-                        if (open_run != null) {
+                        // Plan 16 D4: a parked round's session is
+                        // closed 'parked' — the newest record (open or
+                        // parked) still names the workspace's saddle.
+                        var run_session = yield ctx.tasks.newest_session_for_task(pub.run_task_id);
+                        if (run_session != null && (((!)run_session).ended_at == null
+                                || ((!)run_session).outcome == OUTCOME_PARKED)) {
                             var host = yield ctx.db.query<Saddle>()
                                 .where(expr("id == $0",
-                                    elem<int64?>(((!)open_run).saddle_id)))
+                                    elem<int64?>(((!)run_session).saddle_id)))
                                 .first_async();
                             if (host != null && ((!)host).status == SADDLE_OFFLINE) {
                                 pub.requeue_run = ((!)ctx.actions)
@@ -787,10 +791,17 @@ namespace ClankerSaddle {
                         terminal_url = @"/tasks/$(pub.run_task_id)/terminal?session_id=$(session.id)"
                     });
                 }
-                // Current activity (task-page parity): the open round's
+                // Current activity (task-page parity): the live round's
                 // hosting saddle, its health and the newest recorded
-                // activity line.
-                var open = yield ctx.tasks.open_session_for_task(pub.run_task_id);
+                // activity line. Plan 16 D4: a PARKED round keeps its
+                // panel — its session row closed 'parked' at the park,
+                // and the newest record (open or parked) is still the
+                // round the page is describing.
+                var open = yield ctx.tasks.newest_session_for_task(pub.run_task_id);
+                if (open != null && (((!)open).ended_at != null
+                        && ((!)open).outcome != OUTCOME_PARKED)) {
+                    open = null;
+                }
                 if (open != null) {
                     open_session_saddle = ((!)open).saddle_id;
                     if (ctx.saddles != null) {

test/bundles/allocation.py

@@ -219,7 +219,7 @@ def run(ctx):
         "select id from tasks where goal_id = ? and kind = 'goal_run'",
         (affinity_goal,)).fetchone()[0]
     agent_session = conn.execute(
-        "select id from sessions where task_id = ? and ended_at is null",
+        "select max(id) from sessions where task_id = ?",
         (affinity_parent,)).fetchone()[0]
     conn.close()
     status, resume_payload = fake2.ask_now()
@@ -230,7 +230,8 @@ def run(ctx):
           "the affinity goal agent parks on racer-two")
     conn = db()
     affinity_session = conn.execute(
-        "select saddle_id from sessions where task_id = ? and ended_at is null",
+        "select saddle_id from sessions where task_id = ?"
+        " order by id desc limit 1",
         (affinity_parent,)).fetchone()[0]
     conn.close()
     check(affinity_session == int(racer2_card["id"]),
@@ -431,7 +432,7 @@ def run(ctx):
           "the tag goal agent parks")
     conn = db()
     tag_session = conn.execute(
-        "select id from sessions where task_id = ? and ended_at is null",
+        "select max(id) from sessions where task_id = ?",
         (tag_run,)).fetchone()[0]
     conn.close()
     conn = db()

test/bundles/answers.py

@@ -70,6 +70,14 @@ def run(ctx):
           "the task parks as blocked")
     check(saddle_card(auth, "resume-sleeper")["status"] == "idle",
           "the parked session's saddle row reads idle")
+    conn = db()
+    parked_row = conn.execute(
+        "select outcome, ended_at is not null from sessions where id = ?",
+        (sleeping_session,)).fetchone()
+    conn.close()
+    check(tuple(parked_row) == ("parked", 1),
+          "the park closes its session row 'parked' at park time (D4: the row"
+          " aligns with the daemon's file-log park bracket)")
     ask_status, _ = sleeper.ask_now()
     check(ask_status == 204,
           "the parked daemon's ask returns 204 — it sleeps until woken")
@@ -87,8 +95,8 @@ def run(ctx):
           "answering the block wakes the sleeping affine saddle")
     # Answering = allocating (plan 11): the resume assignment carries
     # the verdict as resume_notice (stable wording) AND a FRESH
-    # session id — the parked row closed, the daemon keys its resumed
-    # pushes on the new one.
+    # session id — the parked row closed 'parked' at its park (D4),
+    # the daemon keys its resumed pushes on the new one.
     check(wait_for(lambda: any(
         int(p.get("task", {}).get("id") or 0) == sleeping_task
         and p.get("session_id") != sleeping_session
@@ -113,10 +121,10 @@ def run(ctx):
         "select id, outcome, ended_at is not null from sessions where task_id = ?"
         " order by id", (sleeping_task,)).fetchall()
     conn.close()
-    check(len(rows) == 2 and rows[0][1] == "resumed" and rows[0][2]
+    check(len(rows) == 2 and rows[0][1] == "parked" and rows[0][2]
           and rows[1][1] == "done" and rows[1][2],
-          "parked session closed 'resumed'; the resumed round ran on a fresh row to done (%s)"
-          % (rows,))
+          "parked session closed 'parked'; the resumed round ran on a fresh row"
+          " to done (%s)" % (rows,))
     retire_daemon(sleeper, "resume-sleeper")
 
     print("== phase 2: answer resume — the parked daemon is woken (POST /answer retired) ==")
@@ -210,10 +218,10 @@ def run(ctx):
         "select id, outcome from sessions where task_id = ? order by id",
         (parked_task,)).fetchall()
     conn.close()
-    check(row is not None and row[0] == "resumed"
-          and len(session_rows) == 2 and session_rows[0] == (parked_session, "resumed"),
-          "the parked session closed 'resumed' and the resumed round kept its own row (%s)"
-          % (session_rows,))
+    check(row is not None and row[0] == "parked"
+          and len(session_rows) == 2 and session_rows[0] == (parked_session, "parked"),
+          "the parked session closed 'parked' (at its park, D4) and the resumed round"
+          " kept its own row (%s)" % (session_rows,))
     retire_daemon(parked, "resume-parked")
 
     print("== plan 11: plan acceptance — accept / reject / auto-approve ==")
@@ -296,13 +304,12 @@ def run(ctx):
 
     print("== phase 2: offline requeue keeps parked-with-answered tasks ==")
     # A daemon dying while its parked task carries an answered
-    # verdict: going offline must not requeue the task or close its
-    # session — the resume machinery (delivery once the daemon
-    # returns, or the affinity allocation on its next ask) needs
-    # both. The question park leaves the row idle; the review park
-    # keeps it busy@task (`saddle review` blocks the round without
-    # a parking state push), exercising the unreachable path's
-    # requeue decision both ways.
+    # verdict: going offline must not requeue the task — the resume
+    # machinery (delivery once the daemon returns, or the affinity
+    # allocation on its next ask) needs the parked record (the
+    # session row closed 'parked' at its park, D4) and the task left
+    # parked. Both park kinds (question and review) exercise the
+    # unreachable path's requeue decision.
     stranded = answerflow_daemon()
     register_answerflow(stranded, "resume-stranded")
     stranded_task = create_answerflow_task("Stranded answered task")
@@ -329,12 +336,13 @@ def run(ctx):
                        "stranded task not requeued"),
           "the parked-with-answered task is not requeued by the offline pass")
     conn = db()
-    open_session = conn.execute(
-        "select count(*) from sessions where task_id = ? and ended_at is null",
-        (stranded_task,)).fetchone()[0]
+    stranded_rows = conn.execute(
+        "select outcome, ended_at is not null from sessions where task_id = ?"
+        " order by id desc limit 1", (stranded_task,)).fetchone()
     conn.close()
-    check(open_session == 1,
-          "the parked-with-answered task keeps its open session for the resume")
+    check(tuple(stranded_rows) == ("parked", 1),
+          "the parked-with-answered task keeps its parked session record ('parked')"
+          " for the resume")
 
     reviewer = answerflow_daemon(ask_on_terminal=False)
     register_answerflow(reviewer, "resume-reviewer")
@@ -370,12 +378,13 @@ def run(ctx):
                        "review-parked task not requeued"),
           "the review-parked task with an answered verdict is never requeued")
     conn = db()
-    open_session = conn.execute(
-        "select count(*) from sessions where task_id = ? and ended_at is null",
-        (review_task,)).fetchone()[0]
+    review_rows = conn.execute(
+        "select outcome, ended_at is not null from sessions where task_id = ?"
+        " order by id desc limit 1", (review_task,)).fetchone()
     conn.close()
-    check(open_session == 1,
-          "the review-parked task keeps its open session for the resume")
+    check(tuple(review_rows) == ("parked", 1),
+          "the review-parked task keeps its parked session record ('parked')"
+          " for the resume")
 
     print("== phase 2: staleness — one probe decides ==")
     # A silent-but-alive runner (long think, nothing to flush): after

test/bundles/goals.py

@@ -215,11 +215,12 @@ def run(ctx):
           "the resumed goal agent parks (saddle suspend)")
 
     print("== plan 12: task fan-out via the task CLI callbacks ==")
-    # The goal agent (parked, open session on its saddle) creates,
+    # 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 id from sessions where task_id = ? and ended_at is null",
+        "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, {
@@ -409,13 +410,14 @@ def run(ctx):
 
     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 open round's newest
-    # activity line and its hosting saddle. Parked runs keep their
-    # open session, so fixtures land as rows without disturbing the
+    # 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 id from sessions where task_id = ? and ended_at is null",
+        "select max(id) from sessions where task_id = ?",
         (run_task,)).fetchone()[0]
     conn.execute(
         "insert into task_progress (task_id, summary, created_at)"
@@ -714,6 +716,22 @@ def run(ctx):
     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
@@ -754,7 +772,7 @@ def run(ctx):
           "the doomed goal's agent parks")
     conn = db()
     agent_session2 = conn.execute(
-        "select id from sessions where task_id = ? and ended_at is null",
+        "select max(id) from sessions where task_id = ?",
         (run2,)).fetchone()[0]
     conn.close()
     status, body = saddle_api("POST", "/api/saddle/task/create", fake_token, {
@@ -801,7 +819,7 @@ def run(ctx):
           "the collapse goal's agent parks first")
     conn = db()
     run3_session = conn.execute(
-        "select id from sessions where task_id = ? and ended_at is null",
+        "select max(id) from sessions where task_id = ?",
         (run3,)).fetchone()[0]
     conn.close()
 

test/bundles/goalui.py

@@ -42,12 +42,15 @@ def run(ctx):
                 if b.get("item_type") == "goal"]
 
     def open_session(task_id):
+        """The task's newest session row (plan 16 D4: a parked round's
+        row is closed 'parked' — still the record the CLI callbacks,
+        the telemetry pushes and the page surfaces key on)."""
         conn = db()
         try:
             row = conn.execute(
-                "select id from sessions where task_id = ? and ended_at is null",
+                "select max(id) from sessions where task_id = ?",
                 (task_id,)).fetchone()
-            return int(row[0]) if row else None
+            return int(row[0]) if row and row[0] else None
         finally:
             conn.close()
 
@@ -318,27 +321,35 @@ def run(ctx):
           "a parked run on a dead saddle offers Requeue-to-resume")
     check(state.get("retry_run") is None,
           "an active goal offers no Retry supervision")
-    open_before = open_session(run_task)
+    conn = db()
+    before_requeue = conn.execute(
+        "select id, outcome from sessions where task_id = ?"
+        " order by id desc limit 1", (run_task,)).fetchone()
+    conn.close()
+    check(tuple(before_requeue) == (open_session(run_task), "parked"),
+          "the parked round's newest session row closed 'parked' (D4)")
 
     def parked_session_ended():
-        if open_before is None:
-            return True
+        # D4: the parked row is ALREADY closed at its park — the pin
+        # keeps guarding the restore-on-dispatch successor shape (the
+        # requeue may not resurrect the OLD row; a fresh round opens
+        # its own).
         conn = db()
         try:
             return conn.execute(
                 "select ended_at is not null from sessions where id = ?",
-                (open_before,)).fetchone()[0] == 1
+                (before_requeue[0],)).fetchone()[0] == 1
         finally:
             conn.close()
     act(auth, state["requeue_run"], {})
     check(task_state(auth, run_task)["status"] == "queued",
           "the requeue returns the run to the queue")
-    # The PARKED 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).
+    # The parked round's own row must stay closed (the
+    # restore-on-dispatch successor opens a FRESH session row — the
+    # check keys the old row, never the task's transient open window).
     check(wait_for(parked_session_ended, timeout=60,
                    what="parked session closed") is not None,
-          "the requeue closed the parked session (restore-on-dispatch)")
+          "the requeue leaves the parked session row closed (fresh round, fresh row)")
     conn = db()
     conn.execute("update saddles set status = 'idle' where id = ?", (fake_id,))
     conn.commit()
@@ -613,7 +624,7 @@ def run(ctx):
     # pending-report idempotency leaves the one row alone.
     conn = db()
     burst_session = conn.execute(
-        "select id from sessions where task_id = ? and ended_at is null",
+        "select max(id) from sessions where task_id = ?",
         (run_task,)).fetchone()[0]
     conn.close()
     done_pair = [{"id": 9401, "command": "saddle done",

test/bundles/lineage.py

@@ -40,13 +40,15 @@ def run(ctx):
     auto_parent = conn.execute(
         "select id from tasks where goal_id = ? and kind = 'goal_run'",
         (goal,)).fetchone()[0]
-    agent_session = conn.execute(
-        "select id from sessions where task_id = ? and ended_at is null",
-        (auto_parent,)).fetchone()[0]
     conn.close()
     check(wait_for(lambda: task_state(auth, auto_parent)["status"] == "suspended"
                    or False, timeout=60, what="lineage goal agent parks") is not None,
           "the lineage goal agent parks")
+    conn = db()
+    agent_session = conn.execute(
+        "select max(id) from sessions where task_id = ?",
+        (auto_parent,)).fetchone()[0]
+    conn.close()
 
     # Two workers under the goal: one free, one held by the gpu tag.
     status, body = saddle_api("POST", "/api/saddle/task/create", r4_token, {

test/bundles/replay.py

@@ -225,7 +225,7 @@ def run(ctx):
           "the burst goal's agent parks first")
     conn = db()
     run_session = conn.execute(
-        "select id from sessions where task_id = ? and ended_at is null",
+        "select max(id) from sessions where task_id = ?",
         (run_task,)).fetchone()[0]
     conn.close()
 
@@ -290,7 +290,7 @@ def run(ctx):
           "the comment goal's agent parks")
     conn = db()
     comment_session = conn.execute(
-        "select id from sessions where task_id = ? and ended_at is null",
+        "select max(id) from sessions where task_id = ?",
         (comment_run,)).fetchone()[0]
     conn.close()
 

test/bundles/subtasks.py

@@ -76,7 +76,7 @@ def run(ctx):
     # the verdict channel (/poll) with it.
     conn = db()
     plain_session = conn.execute(
-        "select id from sessions where task_id = ? and ended_at is null",
+        "select max(id) from sessions where task_id = ?",
         (plain,)).fetchone()[0]
     conn.close()
     status, body = saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
@@ -121,13 +121,15 @@ def run(ctx):
     run_task = conn.execute(
         "select id from tasks where goal_id = ? and kind = 'goal_run'",
         (goal,)).fetchone()[0]
-    agent_session = conn.execute(
-        "select id from sessions where task_id = ? and ended_at is null",
-        (run_task,)).fetchone()[0]
     conn.close()
     check(wait_for(lambda: task_state(auth, run_task)["status"] == "suspended"
                    or False, timeout=60, what="sequencing agent parks") is not None,
           "the sequencing goal agent parks")
+    conn = db()
+    agent_session = conn.execute(
+        "select max(id) from sessions where task_id = ?",
+        (run_task,)).fetchone()[0]
+    conn.close()
 
     # Create with capabilities: attached until assignment (the hold
     # keeps the task queued), mandate + mode stored, ephemeral branch.