fix(chat): hold a restored queue instead of showing it as work in flight

docs/system-specs/modules/session.md

@@ -3759,6 +3759,28 @@ so absence clears it.
   drains an idle slot on boot, so the user sends, edits or deletes them. This is
   the same rule `sendTurn.ts` follows for an indeterminate send: a prompt whose
   turn may have run un-persisted must never be auto-resent.
+  A restored card is **held** (`slot_queue_repository.is_held_queue_entry`) for
+  as long as it waits, not only at boot. The drain passes it by: a turn that
+  ends, a Resume and a fresh send all run ahead of it, and only the card's own
+  Send now (the idle Run-now path, which names it by `queue_id`) starts it.
+  Every reader that asks whether the queue will drain counts
+  `slot.queue_ready_depth` instead of `queue_depth`: Continue's
+  `slot_queue_pending` refusal, the completion sound's `continuing` flag and the
+  session-status `queued` state. The slot frame carries `queue_held_depth` and
+  each held queue entry `meta.held`, so the sidebar, board and command center do
+  not count a held card as work in flight and the card drops its waiting
+  animation. An interrupted chat with only held cards therefore offers Resume,
+  as its restart notice says.
+- **The dequeue is durable with the turn that consumed it.** The turn marker's
+  metadata merge (`_begin_local_turn_marker`) writes `queued_prompts` as the
+  queue stands after the drain popped the entry, in the same atomic write as
+  the opener copy, and credits `_queue_persisted_sig` under the lock. Written
+  apart, a death between them left the entry on disk twice: as the marker's
+  opener, which the restore re-appends as the interrupted turn's row, and in
+  `queued_prompts`, which the restore hands back as a card. The merge takes the
+  same stale-witness refusal as every queue writer: when another writer
+  committed a newer queue after the snapshot, the marker lands and the queue
+  key is left as that writer put it.
 - **Durability does not depend on the mutation site.** `_queue_persisted_sig`
   records what the last committed save wrote and `slot.queue_persist_pending`
   compares it against the live queue, so an in-place rewrite — a reorder, a

src/kiro_crew/builtin_skills/goal-conductor/SKILL.md

@@ -576,7 +576,8 @@ exactly like a plain send, so it is never dropped. Leave `steer` off for answers
 to a `question`, seeds, and anything that can wait for the turn boundary.
 A steer does not withdraw what is already queued: if `session_read_message`
 reports a non-zero `queue_depth`, those messages still run after the turn, so
-say in the steered message which earlier instruction it supersedes.
+say in the steered message which earlier instruction it supersedes. Messages
+it reports as held after a restart do not run unless the person sends them.
 
 ## When a conductor dispatched you
 

src/kiro_crew/context.py

@@ -1165,17 +1165,21 @@ def _replay_rows(
     exclude_last_n: int = 0,
     pending_messages: list[dict] | None = None,
     current_message: dict | None = None,
+    skip_row: Callable[[dict], bool] | None = None,
 ) -> list[dict]:
     """Tail of the chain under per-role quotas, in chronological order.
 
     Conversation rows get the full quota whatever the inject volume, which a
-    single bounded query cannot guarantee.
+    single bounded query cannot guarantee. *skip_row* drops rows from the
+    merged disk and live view before the quotas are counted.
     """
     messages = conversation_log.read_messages_chained(session_key) if conversation_log else []
     if pending_messages is not None or current_message is not None:
         messages = _merge_replay_rows(messages, pending_messages or [], current_message)
     elif exclude_last_n > 0:
         messages = messages[:-exclude_last_n]
+    if skip_row is not None:
+        messages = [row for row in messages if not skip_row(row)]
     messages = _without_warm_task_brief_messages(messages)
     messages = [row for row in messages if not _replay.is_context_cap_notice(row)]
     return _replay._quota_tail(
@@ -1226,6 +1230,7 @@ def build_session_replay(
     pending_messages: list[dict] | None = None,
     current_message: dict | None = None,
     budget_chars: int = _REPLAY_BUDGET_CHARS,
+    skip_row: Callable[[dict], bool] | None = None,
 ) -> str | None:
     """Build session replay from KiroCrew's conversation_log.
 
@@ -1240,6 +1245,7 @@ def build_session_replay(
     With *pending_messages*, merge the disk and live window by delivery identity
     and exclude *current_message* explicitly before applying quotas or budgets.
     The legacy *exclude_last_n* applies only when no live snapshot is supplied.
+    *skip_row* drops rows from that merged view, disk copies included.
 
     *model_window* scales the replay budget to the active model's context
     window (the dashboard's primary history vehicle — it must shrink on a
@@ -1253,6 +1259,7 @@ def build_session_replay(
         exclude_last_n=exclude_last_n,
         pending_messages=pending_messages,
         current_message=current_message,
+        skip_row=skip_row,
     )
     if not messages:
         return None

src/kiro_crew/dashboard/chat_api/continue_turn.py

@@ -196,9 +196,12 @@ async def continue_slot_turn(
             raise SlotContinueRefusal("slot is running", "slot_running")
         if slot._stopping or slot._stop_state != "idle":
             raise SlotContinueRefusal("a stop is in progress", "slot_stopping")
-        if slot.queue_depth:
+        if slot.queue_ready_depth:
             # The runner is about to pick the thread back up on its own; adding a
-            # continuation would double-fire.
+            # continuation would double-fire. Held cards (restored after a
+            # restart) are not counted: nothing picks them up, so refusing on
+            # them left an interrupted chat with no way forward but a new
+            # message. The recovery runs ahead of them and they stay held.
             raise SlotContinueRefusal("queued messages pending", "slot_queue_pending")
         if any(not f.done() for f in slot._approval_futures.values()):
             raise SlotContinueRefusal("approval pending", "slot_approval_pending")

src/kiro_crew/dashboard/chat_api/resume.py

@@ -51,7 +51,7 @@
         logger,
         members_mod,
         note_crew_log_class,
-        queue_entry_view,
+        queue_entry_views,
         read_bounded_json,
         redact_credentials,
         redact_exfiltration_urls,
@@ -274,7 +274,7 @@ async def _live_slot_resume_payload(state, existing) -> dict:
         "ok": True,
         "key": existing.key,
         "messages": prepared,
-        "queue": [queue_entry_view(q) for q in existing._queue],
+        "queue": queue_entry_views(existing),
         "total": total,
         "has_more": next_before > 0,
         "next_before": next_before,
@@ -740,7 +740,7 @@ async def api_chat_slot_resume(request: web.Request) -> web.Response:
                 live_child=_live_child_instance(state, slot),
                 workspace=slot.workspace,
             ),
-            "queue": [queue_entry_view(q) for q in slot._queue],
+            "queue": queue_entry_views(slot),
             "total": total,
             "has_more": total > len(recent),
             "memory_mode": slot.memory_mode,

src/kiro_crew/dashboard/chat_api/slot_detail.py

@@ -18,6 +18,8 @@
         _FLUSH_SNAPSHOT_RETRIES,
         _TRANSIENT_ROLES,
         _UNOWED_WINDOW_ROLES,
+        HOLD_RELEASED_KEY,
+        RESTORED_QUEUE_KEY,
         SLOT_DETAIL_MAX_LIMIT,
         DashboardState,
         OversizedRecord,
@@ -37,6 +39,7 @@
         effective_session_key,
         folder_principal,
         history_corpus_unreadable,
+        is_held_queue_entry,
         logger,
         parse_cls_meta,
         queue_entry_view,
@@ -1285,6 +1288,9 @@ async def _read_page(view: SlotView) -> tuple[list[dict], int, bool, int]:
     # the loop keeps editing the live one; the view helper does the redaction.
     # The two origin stamps ride along because the view reads them to decide
     # whether an entry is shown as typed (`queue_entry_is_user_origin`).
+    # circular import: responsive_chat reaches the chat handlers at module level.
+    from kiro_crew.dashboard import responsive_chat
+
     queue_snapshot = [
         {
             "id": q["id"],
@@ -1293,6 +1299,12 @@ async def _read_page(view: SlotView) -> tuple[list[dict], int, bool, int]:
             "meta": dict(q.get("meta") or {}),
             "_directive_user_origin": q.get("_directive_user_origin", False),
             "_directive_channel_origin": q.get("_directive_channel_origin", False),
+            # Read by the view to mark a held card (`is_held_queue_entry`):
+            # restored, and not yet released by the person's Send now.
+            RESTORED_QUEUE_KEY: bool(q.get(RESTORED_QUEUE_KEY)),
+            HOLD_RELEASED_KEY: bool(q.get(HOLD_RELEASED_KEY)),
+            # Decided here, on the loop, against the live transcript.
+            "_row_written": is_held_queue_entry(q) and responsive_chat.row_already_written(slot, q),
         }
         for q in slot._queue
     ]
@@ -1381,7 +1393,9 @@ def _render(live_child: str) -> str:
         )
         tail = json.dumps(
             {
-                "queue": [queue_entry_view(q) for q in queue_snapshot],
+                "queue": [
+                    queue_entry_view(q, row_written=q["_row_written"]) for q in queue_snapshot
+                ],
                 "total": total,
                 **page_fields,
                 # Seeds the context meter on open. Turn-scoped WS frames alone

src/kiro_crew/dashboard/chat_api/turn_stop.py

@@ -22,16 +22,19 @@
         _ChatSlot,
         _history_key_for,
         _mark_permission_resolved,
+        _memory_preparation_pending,
         _orphan_in_current_turn,
         _slot_replaced_while_queued,
         _start_next_queued_turn,
         _unblock_pending_waits,
+        choose_for_send_now,
         compaction_in_flight,
         deny_app_slot_access,
         effective_session_key,
         logger,
         read_bounded_json,
         relay_archive_refusal,
+        release_held_queue_entry,
         sel,
         slot_not_found,
         stop_declined_armed,
@@ -950,12 +953,43 @@ async def _cancel_linked_turn() -> str:
             )
             if denied is not None:
                 return denied
+            # Lift a held card's hold before dispatch: with memory preparation
+            # pending the drain parks, and the parked drain that later starts it
+            # names no card. See `release_held_queue_entry`.
+            release_held_queue_entry(slot._queue, queue_id)
             started = await _start_next_queued_turn(
                 state,
                 slot,
                 allow_user_during_subagents=True,
                 required_queue_id=queue_id,
             )
+            # Parked, not refused: memory preparation is still running, so the
+            # drain left the card queued and preparation's completion runs it.
+            # Answered as accepted so the client keeps its pending latch until
+            # the card retires; a refusal would hand back controls for a card
+            # that is about to run anyway. The card is marked as the choice, so
+            # that drain starts this card and not the queue's head.
+            parked = (
+                not started
+                and _memory_preparation_pending(state)
+                and any(item.get("id") == queue_id for item in slot._queue)
+            )
+            if parked:
+                choose_for_send_now(slot._queue, queue_id)
+        if parked:
+            # The accepted command is what authorizes the deferred dispatch, so
+            # it is audited now: the later run may never happen.
+            sel().log_tool_invocation(
+                session_key=_history_key_for(name),
+                agent=getattr(slot, "agent", "") or "kirocrew",
+                source="dashboard",
+                tool_name="dashboard_interrupt",
+                tool_kind="command",
+                outcome="parked",
+                metadata={"slot": name, "queue_id": queue_id},
+            )
+            state.push_slots_update()
+            return web.json_response({"ok": True, "outcome": "parked"})
         if not started:
             return web.json_response(
                 {
@@ -1074,6 +1108,10 @@ async def _cancel_linked_turn() -> str:
         # here would overwrite it. Escalation keeps the same generation and a
         # cleared queue, so promotion there is a harmless no-op.
         slot.queue_promote_by_id(queue_id)
+        # A held card (restored after a restart) is one the drain passes by.
+        # The person just chose it, and the turn-end drain that will start it
+        # names no card, so the choice has to live on the entry itself.
+        release_held_queue_entry(slot._queue, queue_id)
 
     # A concurrent /stop can supersede our claim during the body await:
     # escalate it (soft_pending → killing), or escalate-settle-and-be-followed

src/kiro_crew/dashboard/chat_delivery.py

@@ -28,7 +28,11 @@
     _redact_for_display,
     _redact_meta,
 )
-from kiro_crew.dashboard.slot_queue_repository import ATTACHMENT_META_KEYS, warn_if_not_durable
+from kiro_crew.dashboard.slot_queue_repository import (
+    ATTACHMENT_META_KEYS,
+    is_held_queue_entry,
+    warn_if_not_durable,
+)
 from kiro_crew.history import HUMAN_TURN_META_KEY
 from kiro_crew.security import redact_credentials, redact_exfiltration_urls
 
@@ -1454,7 +1458,7 @@ def attachment_meta(user_meta: dict | None) -> dict[str, list[str]]:
     return {k: v for k, v in redacted.items() if isinstance(v, list)}
 
 
-def queue_entry_view(item: dict[str, Any]) -> dict[str, Any]:
+def queue_entry_view(item: dict[str, Any], *, row_written: bool = False) -> dict[str, Any]:
     """The wire form of one queue entry: ``id``, ``content``, and ``meta``
     holding its attachment lists when it carries any.
 
@@ -1508,4 +1512,28 @@ def queue_entry_view(item: dict[str, Any]) -> dict[str, Any]:
     label = (item.get("meta") or {}).get("appLabel")
     if isinstance(label, str) and label:
         view.setdefault("meta", {})["appLabel"] = label
+    # A held card (restored after a restart) waits for the person rather than
+    # the drain, so the card must not animate as if it is next to run. Omitted
+    # otherwise, like the other flags. See `slot_queue_repository.is_held_queue_entry`.
+    if is_held_queue_entry(item):
+        view.setdefault("meta", {})["held"] = True
+        # The caller knows whether the entry's user row is already in the
+        # transcript (a crewmate send writes it at send). Such an entry can be
+        # sent or discarded, not edited.
+        if row_written:
+            view["meta"]["rowWritten"] = True
     return view
+
+
+def queue_entry_views(slot: Any) -> list[dict[str, Any]]:
+    """``queue_entry_view`` for every entry of *slot*, with ``rowWritten`` decided."""
+    # circular import: responsive_chat imports this module.
+    from kiro_crew.dashboard import responsive_chat
+
+    return [
+        queue_entry_view(
+            q,
+            row_written=is_held_queue_entry(q) and responsive_chat.row_already_written(slot, q),
+        )
+        for q in slot._queue
+    ]

src/kiro_crew/dashboard/chat_handlers.py

@@ -320,6 +320,7 @@
     normalize_send_id,
     queue_entry_is_user_origin,
     queue_entry_view,
+    queue_entry_views,
     queue_for_next_turn,
     queued_text_for_display,
     quote_meta,
@@ -360,6 +361,7 @@
 )
 from kiro_crew.dashboard.chat_runner import (  # noqa: F401
     _context_usage_payload,
+    _memory_preparation_pending,
     _run_chat,
     _start_next_queued_turn,
     _sync_served_model,
@@ -469,7 +471,14 @@
     resolved_row_identity,
     stop_declined_armed,
 )
-from kiro_crew.dashboard.slot_queue_repository import warn_if_not_durable  # noqa: F401
+from kiro_crew.dashboard.slot_queue_repository import (  # noqa: F401
+    HOLD_RELEASED_KEY,
+    RESTORED_QUEUE_KEY,
+    choose_for_send_now,
+    is_held_queue_entry,
+    release_held_queue_entry,
+    warn_if_not_durable,
+)
 from kiro_crew.dashboard.slot_retention import (  # noqa: F401
     loop_slot_keys,
     select_idle_slot_keys,

src/kiro_crew/dashboard/chat_runner.py

@@ -181,6 +181,7 @@
 )
 from kiro_crew.dashboard.chat_turn.queue_drain import (
     _arm_queued_delivery_settlement,
+    _memory_preparation_pending,
     _start_next_queued_turn,
     drain_idle_parent_queue_for_stall,
     subagents_hold_user_messages,