diff --git a/docs/system-specs/modules/session.md b/docs/system-specs/modules/session.md
index 5371fb0c5a4..39368d76e30 100644
--- a/docs/system-specs/modules/session.md
+++ b/docs/system-specs/modules/session.md
@@ -3773,6 +3773,18 @@ 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.
- **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
diff --git a/src/kiro_crew/context.py b/src/kiro_crew/context.py
index 65fa802849d..53aea036050 100644
--- a/src/kiro_crew/context.py
+++ b/src/kiro_crew/context.py
@@ -1167,17 +1167,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(
@@ -1228,6 +1232,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.
@@ -1242,6 +1247,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
@@ -1255,6 +1261,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
diff --git a/src/kiro_crew/dashboard/chat_api/continue_turn.py b/src/kiro_crew/dashboard/chat_api/continue_turn.py
index d461de8e5a4..c1c1d4135ae 100644
--- a/src/kiro_crew/dashboard/chat_api/continue_turn.py
+++ b/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")
diff --git a/src/kiro_crew/dashboard/chat_api/resume.py b/src/kiro_crew/dashboard/chat_api/resume.py
index ca8093eccf7..e9e864e6ab6 100644
--- a/src/kiro_crew/dashboard/chat_api/resume.py
+++ b/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,
diff --git a/src/kiro_crew/dashboard/chat_api/slot_detail.py b/src/kiro_crew/dashboard/chat_api/slot_detail.py
index 6b2ba8c7a0a..620b8e01edc 100644
--- a/src/kiro_crew/dashboard/chat_api/slot_detail.py
+++ b/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
diff --git a/src/kiro_crew/dashboard/chat_api/turn_stop.py b/src/kiro_crew/dashboard/chat_api/turn_stop.py
index 9c4d2b079b9..376e85cc101 100644
--- a/src/kiro_crew/dashboard/chat_api/turn_stop.py
+++ b/src/kiro_crew/dashboard/chat_api/turn_stop.py
@@ -22,6 +22,7 @@
_ChatSlot,
_history_key_for,
_mark_permission_resolved,
+ _memory_preparation_pending,
_orphan_in_current_turn,
_slot_replaced_while_queued,
_start_next_queued_turn,
@@ -29,9 +30,11 @@
compaction_in_flight,
deny_app_slot_access,
effective_session_key,
+ is_held_queue_entry,
logger,
read_bounded_json,
relay_archive_refusal,
+ release_held_queue_entry,
sel,
slot_not_found,
stop_declined_armed,
@@ -950,6 +953,20 @@ async def _cancel_linked_turn() -> str:
)
if denied is not None:
return denied
+ # A held card (restored after a restart) that the drain would park
+ # here never runs: the drain that ends the parking names no card and
+ # passes held ones by. Refuse it as what it is, so the person sends
+ # it again once memory is ready.
+ if _memory_preparation_pending(state) and any(
+ item.get("id") == queue_id and is_held_queue_entry(item) for item in slot._queue
+ ):
+ return web.json_response(
+ {
+ "error": "memory is still being prepared; send it again when it is ready",
+ "code": "memory_preparing",
+ },
+ status=409,
+ )
started = await _start_next_queued_turn(
state,
slot,
@@ -1074,6 +1091,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
diff --git a/src/kiro_crew/dashboard/chat_delivery.py b/src/kiro_crew/dashboard/chat_delivery.py
index bbddf9ff1fc..b0f9accfdad 100644
--- a/src/kiro_crew/dashboard/chat_delivery.py
+++ b/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
+ ]
diff --git a/src/kiro_crew/dashboard/chat_handlers.py b/src/kiro_crew/dashboard/chat_handlers.py
index 89dafd94122..2d016280403 100644
--- a/src/kiro_crew/dashboard/chat_handlers.py
+++ b/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,13 @@
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,
+ 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,
diff --git a/src/kiro_crew/dashboard/chat_runner.py b/src/kiro_crew/dashboard/chat_runner.py
index d37514b6d14..71ef1c0814c 100644
--- a/src/kiro_crew/dashboard/chat_runner.py
+++ b/src/kiro_crew/dashboard/chat_runner.py
@@ -114,6 +114,7 @@
from kiro_crew.dashboard import directive_queue
from kiro_crew.dashboard import git_repo_touches as _git_repo_touches
from kiro_crew.dashboard.chat_delivery import COMMANDS_OFF_META_KEY # noqa: F401
+from kiro_crew.dashboard.chat_delivery import start_queue_persist # noqa: F401
from kiro_crew.dashboard.chat_delivery import TURN_ACTOR_META_KEY as _TURN_ACTOR_META_KEY
from kiro_crew.dashboard.chat_persistence import save_slot_off_loop # noqa: F401
from kiro_crew.dashboard.chat_summary import generate_session_summary
@@ -181,6 +182,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,
diff --git a/src/kiro_crew/dashboard/chat_turn/prompt_assembly.py b/src/kiro_crew/dashboard/chat_turn/prompt_assembly.py
index e1b4c8c8ae7..6224a9df8d1 100644
--- a/src/kiro_crew/dashboard/chat_turn/prompt_assembly.py
+++ b/src/kiro_crew/dashboard/chat_turn/prompt_assembly.py
@@ -34,6 +34,7 @@
user_text_span,
)
from kiro_crew.dashboard.handlers import _prompt_read_within_root
+from kiro_crew.dashboard.slot_queue_repository import is_held_queue_entry
from kiro_crew.dashboard.state import TURN_OPENING_INJECT_KINDS
from kiro_crew.executors import run_in_embed_pool
from kiro_crew.hooks import FileTooLargeError, safe_read_file_bytes_nolink
@@ -44,6 +45,8 @@
)
if TYPE_CHECKING:
+ from collections.abc import Callable
+
from kiro_crew.context import ContextBuilder
from kiro_crew.dashboard.chat_runner import (
_STARTUP_QUERY_PREWARM_SECS,
@@ -82,6 +85,42 @@
from kiro_crew.dashboard.chat_turn.turn_locals import _ChatTurn
+def _unsent_send_filter(slot: _ChatSlot) -> Callable[[dict], bool] | None:
+ """Whether a row is the user row of a send that is not part of the cut turn.
+
+ A crewmate send writes its user row at send time (``write_row_at_send``),
+ tagged with its queue entry's id. When the restart hands that entry back
+ held, the row stays in the transcript but was never sent again. Without
+ this filter a Resume's replay or interrupted-turn preamble takes it, the
+ newest user row, as the request to continue.
+ ``None`` when no row can match.
+ """
+ # circular import: responsive_chat reaches the chat runner at module level.
+ from kiro_crew.dashboard.responsive_chat import QUEUED_ROW_META_KEY
+
+ skipped = {item.get("id") for item in slot._queue if is_held_queue_entry(item)}
+ if not skipped:
+ return None
+
+ def _unsent(row: dict) -> bool:
+ meta = row.get("meta")
+ return (
+ row.get("role") == "user"
+ and isinstance(meta, dict)
+ and meta.get(QUEUED_ROW_META_KEY) in skipped
+ )
+
+ return _unsent
+
+
+def _rows_without_held_sends(slot: _ChatSlot) -> list[dict]:
+ """The slot's rows minus those :func:`_unsent_send_filter` matches."""
+ unsent = _unsent_send_filter(slot)
+ if unsent is None:
+ return list(slot.messages)
+ return [row for row in slot.messages if not unsent(row)]
+
+
async def _session_replay_history(
state: DashboardState,
context_builder: ContextBuilder,
@@ -137,7 +176,9 @@ async def _session_replay_history(
_budget = {"budget_chars": CONTEXT_CAP_TAIL_CHARS} if _trim_seed else {}
# Merge the disk transcript and a frozen live-window tail
# before one budget pass. Exclude this request by identity,
- # whether or not the periodic flush has persisted it yet.
+ # whether or not the periodic flush has persisted it yet, and
+ # the rows of held sends, from disk and live alike.
+ _unsent = _unsent_send_filter(slot)
compressed = (
await asyncio.to_thread(
build_session_replay,
@@ -146,6 +187,7 @@ async def _session_replay_history(
pending_messages=list(slot.messages),
current_message=_current_replay_message,
model_window=window_for_provider_client(client),
+ skip_row=_unsent,
**_budget,
)
or ""
@@ -668,7 +710,7 @@ async def _turn_build_prompt(turn: _ChatTurn) -> None:
)
preamble = build_interrupted_turn_preamble(
- list(slot.messages),
+ _rows_without_held_sends(slot),
current=_current_replay_message,
opener_inject_kinds=TURN_OPENING_INJECT_KINDS,
)
diff --git a/src/kiro_crew/dashboard/chat_turn/queue_drain.py b/src/kiro_crew/dashboard/chat_turn/queue_drain.py
index 84688d5c577..5c193bb12a4 100644
--- a/src/kiro_crew/dashboard/chat_turn/queue_drain.py
+++ b/src/kiro_crew/dashboard/chat_turn/queue_drain.py
@@ -35,7 +35,7 @@
REFUSAL_SELF_WAKE_META_KEY,
REFUSAL_WAKE_LOOP_ID_META_KEY,
)
-from kiro_crew.dashboard.slot_queue_repository import RESTORED_QUEUE_KEY
+from kiro_crew.dashboard.slot_queue_repository import RESTORED_QUEUE_KEY, is_held_queue_entry
from kiro_crew.dashboard.state import CRON_NOTIFY_RE, SUBAGENT_COMPLETION_PREFIXES
from kiro_crew.history import HUMAN_TURN_META_KEY
@@ -59,6 +59,7 @@
redact_exfiltration_urls,
shutdown_event,
spawn_guarded_turn,
+ start_queue_persist,
)
from kiro_crew.dashboard.chat_turn.recovery import (
_drop_revoked_replays,
@@ -232,6 +233,22 @@ def settle_drained_subagent_notices(state: Any, slot: Any, notices: list[str]) -
)
+def _next_dispatchable_entry(slot: _ChatSlot, required_queue_id: str | None) -> dict | None:
+ """The entry the drain would start: the named card, else the first one not held.
+
+ ``None`` means the drain has nothing it may start: a queue holding only
+ restored cards waits for Send now, an edit or a delete
+ (:func:`is_held_queue_entry`).
+ """
+ for item in slot._queue:
+ if required_queue_id:
+ if item.get("id") == required_queue_id:
+ return item
+ elif not is_held_queue_entry(item):
+ return item
+ return None
+
+
def _memory_preparation_pending(state: DashboardState) -> bool:
"""Whether the gateway's shared memory preparation is still running.
@@ -458,7 +475,9 @@ async def _start_next_queued_turn(
# first would consume input that never runs, so leave the queue as it is
# (cards stay visible, order unchanged) and let the preparation task's
# completion drain it (`_arm_parked_queue_drain`).
- if slot._queue and _memory_preparation_pending(state):
+ if _next_dispatchable_entry(
+ slot, required_queue_id
+ ) is not None and _memory_preparation_pending(state):
_arm_parked_queue_drain(state, slot, notice=True)
return False
@@ -488,8 +507,11 @@ async def _start_next_queued_turn(
# Withheld when the next queued item carries a structural origin tag -- a cron
# notification or a runner-injected recovery prompt -- for the same reason
# _finish_queue_cycle withholds from synthesis: a note is owed to the next
- # USER turn, and that cycle's own flush delivers it afterwards.
- if not (slot._queue and slot._queue[0].get("kind")):
+ # USER turn, and that cycle's own flush delivers it afterwards. "Next" is the
+ # entry this drain would start, not the queue head: a held card at the head
+ # is passed by, and the entry behind it decides.
+ next_entry = _next_dispatchable_entry(slot, required_queue_id)
+ if not (next_entry is not None and next_entry.get("kind")):
try:
slot.flush_deferred_notes()
except Exception:
@@ -503,7 +525,10 @@ async def _start_next_queued_turn(
exc_info=True,
)
- if not slot._queue:
+ # Held cards (restored from disk, see `is_held_queue_entry`) are not this
+ # drain's to start: a queue of only those is an empty queue here. A card the
+ # person named (`required_queue_id`) is the one way such an entry runs.
+ if _next_dispatchable_entry(slot, required_queue_id) is None:
return False
# The promise-only recovery continuation is queue_insert(0)'d at
@@ -550,6 +575,16 @@ async def _start_next_queued_turn(
if await _drop_revoked_replays(state, slot):
return False
+ # Again, right before the dequeue: the purges above may have dropped the
+ # only entry the drain owned, and every dequeue below reads from the head.
+ if _next_dispatchable_entry(slot, required_queue_id) is None:
+ return False
+ if not required_queue_id and is_held_queue_entry(slot._queue[0]):
+ # Bring the first entry the drain owns to the head. The held cards keep
+ # their own order behind it.
+ ready = next(item for item in slot._queue if not is_held_queue_entry(item))
+ slot.queue_promote_by_id(ready["id"])
+
try:
merge = KiroCrewConfig.load().dashboard.merge_queued_messages
except Exception:
@@ -987,6 +1022,13 @@ async def _start_next_queued_turn(
# Persist it on the row so auto-titling and refresh rebase both exclude it.
_drained_meta[RESTORED_TURN_META_KEY] = True
current_row = responsive_chat.reusable_row(slot, consumed, next_msg)
+ if current_row is not None:
+ # The entry's row was written when it was sent, so the queue on disk is
+ # now the one record that it has not run. Left to the periodic flush, a
+ # restart inside that interval hands it back as a held card, and Resume
+ # then skips its row (`_unsent_send_filter`) and continues an older
+ # request. Start the queue write now, as an accepted send does.
+ start_queue_persist(state, slot)
if current_row is None:
if is_automation_steer:
current_row = append_and_surface(
diff --git a/src/kiro_crew/dashboard/chat_turn/steer_queue.py b/src/kiro_crew/dashboard/chat_turn/steer_queue.py
index 9ab7af24497..e774100a5b6 100644
--- a/src/kiro_crew/dashboard/chat_turn/steer_queue.py
+++ b/src/kiro_crew/dashboard/chat_turn/steer_queue.py
@@ -27,6 +27,7 @@
is_system_injection_item,
)
from kiro_crew.dashboard.chat_utils_injected_rows import AUTOMATION_STEER_META
+from kiro_crew.dashboard.slot_queue_repository import is_held_queue_entry
if TYPE_CHECKING:
from kiro_crew.dashboard.chat_runner import (
@@ -467,8 +468,18 @@ def _queue_entry_is_orchestration(item: dict) -> bool:
have their intervention silently ignored while the announced action dispatches
anyway. A user message carries no enqueue tag, so it correctly counts as a
user follow-up and aborts the pending recovery; real cron / sub-agent events
- are tagged at their injection sites and stay excluded."""
- return is_synthetic_payload_item(item) or is_system_injection_item(item)
+ are tagged at their injection sites and stay excluded.
+
+ A HELD entry (restored after a restart, `is_held_queue_entry`) is excluded
+ too. The person said it to a process that is gone, and it waits for them to
+ send it; it is not this process's user intervening. Counted, it would keep
+ every recovery off for as long as the card sits, which is until the person
+ acts on it. Send now lifts the hold, and from then on it counts again."""
+ return (
+ is_synthetic_payload_item(item)
+ or is_system_injection_item(item)
+ or is_held_queue_entry(item)
+ )
#: Which turn actor an enqueue-time queue ``kind`` names. The tag is stamped by
diff --git a/src/kiro_crew/dashboard/chat_utils.py b/src/kiro_crew/dashboard/chat_utils.py
index 52a6f022b9e..d964c182432 100644
--- a/src/kiro_crew/dashboard/chat_utils.py
+++ b/src/kiro_crew/dashboard/chat_utils.py
@@ -816,6 +816,7 @@ def _validate_tool_name(tool_name: str, *, is_shell: bool = False, canonical_nam
)
from kiro_crew.dashboard.slot_queue_repository import ( # noqa: F401
ATTACHMENT_META_KEYS,
+ is_held_queue_entry,
)
from kiro_crew.dashboard.state import ( # noqa: F401
BUSY_RECOVERY_PREFIX,
diff --git a/src/kiro_crew/dashboard/chat_utils_completion.py b/src/kiro_crew/dashboard/chat_utils_completion.py
index 020a5621a6b..15b21a020ab 100644
--- a/src/kiro_crew/dashboard/chat_utils_completion.py
+++ b/src/kiro_crew/dashboard/chat_utils_completion.py
@@ -265,7 +265,7 @@ async def chat_done_payload(
continuing = bool(
continuing
or slot._pending_synthesis
- or (slot.queue_depth and not queue_held and not slot._last_turn_auth_required)
+ or (slot.queue_ready_depth and not queue_held and not slot._last_turn_auth_required)
or await subagents_attached_async(
state, slot, effective_session_key(slot), "completion_sound"
)
diff --git a/src/kiro_crew/dashboard/chat_utils_injected_rows.py b/src/kiro_crew/dashboard/chat_utils_injected_rows.py
index 556977e0f36..54225777a4c 100644
--- a/src/kiro_crew/dashboard/chat_utils_injected_rows.py
+++ b/src/kiro_crew/dashboard/chat_utils_injected_rows.py
@@ -15,7 +15,7 @@
from enum import Enum
from typing import Any
-from kiro_crew.dashboard.slot_queue_repository import ATTACHMENT_META_KEYS
+from kiro_crew.dashboard.slot_queue_repository import ATTACHMENT_META_KEYS, is_held_queue_entry
from kiro_crew.dashboard.state import (
BUSY_RECOVERY_PREFIX,
COMPACTION_RECOVERY_PREFIX,
@@ -563,7 +563,9 @@ def _dequeue_next_system_message(slot) -> tuple:
drained item, or ``(None, [])`` when only held (user) messages remain queued.
"""
for i, item in enumerate(slot._queue):
- if releases_past_subagent_hold(item):
+ # A held card (restored after a restart) runs only on its own Send now,
+ # never because this scan reached it: a restored steer keeps its mark.
+ if releases_past_subagent_hold(item) and not is_held_queue_entry(item):
popped = slot.queue_pop(i)
return popped["content"], [popped]
return None, []
diff --git a/src/kiro_crew/dashboard/handlers/members_thread_turns.py b/src/kiro_crew/dashboard/handlers/members_thread_turns.py
index b61880a534c..cc7fed3f02a 100644
--- a/src/kiro_crew/dashboard/handlers/members_thread_turns.py
+++ b/src/kiro_crew/dashboard/handlers/members_thread_turns.py
@@ -87,7 +87,7 @@ def _turn_busy(state: DashboardState, slot: Any, key: str) -> bool:
def _work_pending(slot: Any) -> bool:
"""Queued messages or unconsumed steers: work the runner starts on its own."""
- return bool(slot.queue_depth) or bool(getattr(slot, "_pending_steers", None))
+ return bool(slot.queue_ready_depth) or bool(getattr(slot, "_pending_steers", None))
def _queue_pending(name: str) -> web.Response:
diff --git a/src/kiro_crew/dashboard/session_control_parts/session_status.py b/src/kiro_crew/dashboard/session_control_parts/session_status.py
index 1b9e69ea0fa..5c4c2bb32c6 100644
--- a/src/kiro_crew/dashboard/session_control_parts/session_status.py
+++ b/src/kiro_crew/dashboard/session_control_parts/session_status.py
@@ -601,9 +601,16 @@ async def created_session_status(
{
"target": key,
"title": _bounded_status_title(slot.display_title),
- "status": "working" if running else ("queued" if queue_depth else "idle"),
+ # "queued" promises the runner will start it; held cards (restored
+ # after a restart) wait for the person, so they read as idle.
+ "status": (
+ "working" if running else ("queued" if slot.queue_ready_depth else "idle")
+ ),
"running": running,
"queue_depth": queue_depth,
+ # Restored cards no drain will start; agents read "queued" as
+ # "will run", so they get the split.
+ "queue_held_depth": slot.queue_held_depth,
"process": process_state(state, slot),
**_pending_approval_fields(state, slot),
"source": "+".join(sources),
@@ -790,6 +797,7 @@ def read_messages(
# "nothing happening".
**({"streaming": True} if durable_end < len(raw_window) else {}),
"queue_depth": len(slot._queue),
+ "queue_held_depth": slot.queue_held_depth,
# A turn parked on a tool approval is still ``running``; this says it is
# waiting on a person rather than working.
**_pending_approval_fields(state, slot),
diff --git a/src/kiro_crew/dashboard/slot_projection.py b/src/kiro_crew/dashboard/slot_projection.py
index 520b225d957..09d8480799f 100644
--- a/src/kiro_crew/dashboard/slot_projection.py
+++ b/src/kiro_crew/dashboard/slot_projection.py
@@ -483,6 +483,10 @@ def to_dict(
# the backend's answer cannot disagree.
"stop_declined": stop_declined_armed(slot),
"queue_depth": slot.queue_depth,
+ # How many of those cards no drain will start: restored after a
+ # restart, they wait for Send now, an edit or a delete. Readers that
+ # ask "is work coming?" subtract it; see `is_held_queue_entry`.
+ "queue_held_depth": slot.queue_held_depth,
"stopping": slot._stopping,
"pending_approval": pending_approval,
"pending_approval_info": pending_approval_info,
diff --git a/src/kiro_crew/dashboard/slot_queue_repository.py b/src/kiro_crew/dashboard/slot_queue_repository.py
index 4426988d032..014d8faa211 100644
--- a/src/kiro_crew/dashboard/slot_queue_repository.py
+++ b/src/kiro_crew/dashboard/slot_queue_repository.py
@@ -81,6 +81,49 @@
#: consumer asks instead of trying to tell the two apart from the actor alone.
RESTORED_QUEUE_KEY = "_restored_from_disk"
+
+def is_held_queue_entry(item: Any) -> bool:
+ """Whether *item* waits for an explicit action instead of the drain.
+
+ A restored entry is HELD: no process owns it, so nothing runs it on its
+ own. It is handed back as a card, and only the person moves it: Send now
+ on that card (the idle Run-now path names it by id), an edit, or a delete.
+ A turn that ends, a Resume and a fresh send all pass it by. The reason is
+ the one ``sendTurn.ts`` gives for an indeterminate send: the process that
+ accepted these words is gone, and a prompt is never re-sent for the person.
+
+ Every reader that asks "will the runner pick this queue up?" asks this
+ instead of ``queue_depth``. The drain skips held entries, Continue does
+ not count them as pending, and the slot projection reports them apart
+ (``queue_held_depth``). A held queue is then never shown as work in flight.
+ """
+ return (
+ isinstance(item, dict)
+ and bool(item.get(RESTORED_QUEUE_KEY))
+ and not item.get(HOLD_RELEASED_KEY)
+ )
+
+
+#: Marks a held entry the person chose with Send now. Process-local, like
+#: :data:`RESTORED_QUEUE_KEY`, and set beside it rather than in its place: the
+#: entry's provenance is still unknown (the drain keeps reading the restored key
+#: for that), only the hold is lifted. Send now does not always dispatch at
+#: once: on a running slot it promotes the card and stops the turn, and with
+#: memory preparation pending the drain parks. The release is what lets the
+#: deferred drain that follows start the chosen card. A restart drops the key
+#: with the rest of the entry's process state, so the card comes back held.
+HOLD_RELEASED_KEY = "_hold_released"
+
+
+def release_held_queue_entry(queue: list[dict[str, Any]], queue_id: str) -> bool:
+ """Lift the hold on the entry *queue_id* names; whether one was held."""
+ for item in queue:
+ if item.get("id") == queue_id and is_held_queue_entry(item):
+ item[HOLD_RELEASED_KEY] = True
+ return True
+ return False
+
+
_DURABLE_QUEUE_KEYS: tuple[str, ...] = (
"id",
"content",
diff --git a/src/kiro_crew/dashboard/state_parts/chat_slot.py b/src/kiro_crew/dashboard/state_parts/chat_slot.py
index 0d4871af40a..075b07a85a8 100644
--- a/src/kiro_crew/dashboard/state_parts/chat_slot.py
+++ b/src/kiro_crew/dashboard/state_parts/chat_slot.py
@@ -15,6 +15,7 @@
SlotQueueRepository,
durable_queue_entries,
durable_queue_view,
+ is_held_queue_entry,
queue_persist_signature,
)
from kiro_crew.dashboard.state_parts import slot_ladder as _slot_ladder
@@ -875,6 +876,25 @@ def queue_depth(self) -> int:
"""Number of prompts currently queued behind the active turn."""
return len(self._queue)
+ @property
+ def queue_held_depth(self) -> int:
+ """Queued entries no drain will start: restored cards waiting for the person.
+
+ See :func:`slot_queue_repository.is_held_queue_entry`.
+ """
+ return sum(1 for item in self._queue if is_held_queue_entry(item))
+
+ @property
+ def queue_ready_depth(self) -> int:
+ """Queued entries the runner starts on its own when the slot frees.
+
+ The number every "will this queue drain?" decision reads: Continue's
+ pending-queue refusal, the completion sound's ``continuing`` flag, the
+ session-status ``queued`` state. ``queue_depth`` still counts every
+ card the person can see.
+ """
+ return self.queue_depth - self.queue_held_depth
+
@property
def model_withheld(self) -> bool | None:
"""Whether the live session withholds this slot's pinned model.
diff --git a/src/kiro_crew/mcp_dashboard.py b/src/kiro_crew/mcp_dashboard.py
index 9f65cd3cb23..e5231238aff 100644
--- a/src/kiro_crew/mcp_dashboard.py
+++ b/src/kiro_crew/mcp_dashboard.py
@@ -798,6 +798,7 @@ def _board_tools() -> tuple[Tool, ...]:
_closed_at_display,
_create_refusal_remedy,
_pending_approval_phrase,
+ _queue_phrase,
_render_session_summary,
_resolve_folder_for_new_session,
_run_session_adopt,
diff --git a/src/kiro_crew/mcp_dashboard_session_runs.py b/src/kiro_crew/mcp_dashboard_session_runs.py
index 00e7bcf3e25..dd3ac0140a9 100644
--- a/src/kiro_crew/mcp_dashboard_session_runs.py
+++ b/src/kiro_crew/mcp_dashboard_session_runs.py
@@ -812,8 +812,7 @@ def _run_session_status(args: dict[str, Any], ctx: ToolContext) -> str:
)
continue
title = str(row.get("title", ""))
- depth = int(row.get("queue_depth", 0) or 0)
- queued = f", {depth} queued" if depth else ""
+ queued = _queue_phrase(row)
if status == "unknown":
# Its OWN mark, not the glyph map's default. Falling through to that
# default draws idle's sleep glyph -- "open and doing nothing", the
@@ -894,6 +893,18 @@ def _run_session_release(args: dict[str, Any], ctx: ToolContext) -> str:
)
+def _queue_phrase(resp: dict[str, Any], unit: str = "") -> str:
+ """ ", N queued", counting only the entries that will run.
+
+ A held entry (restored after a restart) runs only when the person presses
+ Send now on its card, and "queued" tells an agent the message still runs
+ after the turn, so held entries are left out of the count.
+ """
+ depth = int(resp.get("queue_depth", 0) or 0)
+ ready = depth - min(depth, int(resp.get("queue_held_depth", 0) or 0))
+ return f", {ready}{unit} queued" if ready else ""
+
+
def _pending_approval_phrase(resp: dict[str, Any]) -> str:
"""The state words for a session parked on a tool approval."""
tool = str(resp.get("pending_approval_tool") or "")
@@ -912,9 +923,7 @@ def _run_session_read_message(args: dict[str, Any], ctx: ToolContext) -> str:
state_line = "still working" if resp.get("running") else "idle"
if resp.get("pending_approval"):
state_line = _pending_approval_phrase(resp)
- queued = resp.get("queue_depth", 0)
- if queued:
- state_line += f", {queued} message(s) queued"
+ state_line += _queue_phrase(resp, unit=" message(s)")
if resp.get("model"):
state_line += f", model {redact(str(resp['model']))}"
if resp.get("pending_model"):
diff --git a/test/test_chat_mode_security_send.py b/test/test_chat_mode_security_send.py
index 1fbdaa0aad3..4fd0bfd7fb3 100644
--- a/test/test_chat_mode_security_send.py
+++ b/test/test_chat_mode_security_send.py
@@ -1014,6 +1014,21 @@ async def _drain_once(state, slot, **kwargs) -> dict:
return slot.messages[-1]
+async def _send_now_once(state, slot, **kwargs) -> dict:
+ """Run the head card the way a person's Send now does: named by its id.
+
+ A restored entry is held (`is_held_queue_entry`), so this is the one path
+ that still turns its text into a row, and the one these rules must cover.
+ """
+ return await _drain_once(
+ state,
+ slot,
+ allow_user_during_subagents=True,
+ required_queue_id=slot._queue[0]["id"],
+ **kwargs,
+ )
+
+
@pytest.mark.asyncio
async def test_a_queued_mcp_app_message_is_an_app_inject_whatever_its_text(state) -> None:
slot = state.get_or_create_slot("s1", origin=SlotOrigin.USER)
@@ -1051,7 +1066,7 @@ async def test_a_restored_queue_entry_is_never_a_cron_notification(state) -> Non
slot._queue = sanitize_restored_queue(durable_queue_entries(live._queue))
assert slot._queue, "a plain queued send must survive the restart"
- row = await _drain_once(state, slot)
+ row = await _send_now_once(state, slot)
assert row["role"] == "user"
assert "injectKind" not in (row.get("meta") or {})
@@ -1092,7 +1107,7 @@ async def test_restored_recovery_identity_cannot_continue_a_live_stream(state) -
)
slot._queue[0][RESTORED_QUEUE_KEY] = True
- await _drain_once(state, slot, predecessor_actor="app", predecessor_turn_id="turn-a")
+ await _send_now_once(state, slot, predecessor_actor="app", predecessor_turn_id="turn-a")
assert slot.drain()[0]["cls"] == TURN_END_WIRE_CLS
@@ -1105,7 +1120,7 @@ async def test_restored_app_turn_titles_only_an_app_owned_slot(state, owned: boo
slot = _user_slot(state)
slot._queue = sanitize_restored_queue(durable_queue_entries(live._queue))
assert TURN_ACTOR_META_KEY not in slot._queue[0].get("meta", {})
- row = await _drain_once(state, slot)
+ row = await _send_now_once(state, slot)
# Exercise the titler's owner exemption independently of queue admission:
# an app-bound restore may itself be held by containment revalidation.
slot._app = "crew-keyboard" if owned else ""
@@ -1142,20 +1157,21 @@ async def test_restored_and_fresh_queue_rows_never_merge(state, restored_first:
config = KiroCrewConfig()
config.dashboard.merge_queued_messages = True
with patch.object(KiroCrewConfig, "load", return_value=config):
- first = await _drain_once(state, slot)
+ # The drain takes the fresh entry wherever the held one sits, alone.
+ fresh = await _drain_once(state, slot)
slot.append("assistant", "first reply")
assert slot.queue_depth == 1
- second = await _drain_once(state, slot)
+ # The held card runs only when the person sends it.
+ restored_row = await _send_now_once(state, slot)
slot.append("assistant", "second reply")
assert slot.queue_depth == 0
- fresh, restored_row = (second, first) if restored_first else (first, second)
assert fresh["content"] == "fresh user request"
assert restored_row["content"] == "restored app request"
assert _counts_as_user_turn(slot, fresh)
assert not _counts_as_user_turn(slot, restored_row)
assert [row["content"] for row in _titling_messages(slot)] == [
"fresh user request",
- "second reply" if restored_first else "first reply",
+ "first reply",
]
# Keep the count above the fixed-milestone floor, so counting the restored
# row incorrectly changes the rebase result instead of hiding under it.
diff --git a/test/test_chat_slot_facade_contract.py b/test/test_chat_slot_facade_contract.py
index 92186246f28..c8f18854c37 100644
--- a/test/test_chat_slot_facade_contract.py
+++ b/test/test_chat_slot_facade_contract.py
@@ -54,6 +54,7 @@
"preparing",
"stop_declined",
"queue_depth",
+ "queue_held_depth",
"stopping",
"pending_approval",
"pending_approval_info",
diff --git a/test/test_dashboard_chat_handlers_coverage.py b/test/test_dashboard_chat_handlers_coverage.py
index a40e9a94859..ed219de280a 100644
--- a/test/test_dashboard_chat_handlers_coverage.py
+++ b/test/test_dashboard_chat_handlers_coverage.py
@@ -880,6 +880,31 @@ async def test_cancel_removes_the_item_and_broadcasts(self, _sel):
assert payload["queue_id"] == qid
state.push_slots_update.assert_called_once()
+ @pytest.mark.asyncio
+ async def test_a_held_send_already_in_the_conversation_is_not_cancelled_either(self, _sel):
+ # Restored after a restart (held) with its row written at send: the
+ # row is the message, so Cancel is refused like any written send and
+ # Send now is the way out. The card stays, held.
+ from kiro_crew.dashboard.responsive_chat import QUEUED_ROW_META_KEY
+ from kiro_crew.dashboard.slot_queue_repository import (
+ RESTORED_QUEUE_KEY,
+ is_held_queue_entry,
+ )
+
+ slot = _ChatSlot("s1")
+ qid = slot.queue_append("orig")
+ slot._queue[0][RESTORED_QUEUE_KEY] = True
+ slot.append("user", "orig", "msg msg-u", meta={QUEUED_ROW_META_KEY: qid})
+ state = _state(slot)
+
+ status, body = await self._delete(state, "s1", qid)
+
+ assert status == 409
+ assert body["code"] == "queue_item_already_sent"
+ assert [q["id"] for q in slot._queue] == [qid]
+ assert is_held_queue_entry(slot._queue[0])
+ state.broadcast_ws.assert_not_called()
+
@pytest.mark.asyncio
async def test_a_send_already_in_the_conversation_is_not_cancelled(self, _sel):
# Its row is in the transcript already: dropping only the entry would
diff --git a/test/test_gateway_appkit_endpoints_note.py b/test/test_gateway_appkit_endpoints_note.py
index fb03b467900..1fc9c539e67 100644
--- a/test/test_gateway_appkit_endpoints_note.py
+++ b/test/test_gateway_appkit_endpoints_note.py
@@ -1259,9 +1259,11 @@ def test_every_turn_start_seam_flushes_above_its_own_row(self):
assert queued.index("flush_deferred_notes") < queued.index("_dequeue_next")
# ...but guarded, so a queued item carrying a structural origin tag (a
# cron notification, a runner-injected recovery prompt) is not fed the
- # note; _finish_queue_cycle delivers it after that turn instead.
- assert 'slot._queue[0].get("kind")' in queued
- assert queued.index('slot._queue[0].get("kind")') < queued.index("flush_deferred_notes")
+ # note; _finish_queue_cycle delivers it after that turn instead. The
+ # guard reads the entry the drain would start, which passes a held
+ # (restored) card at the head by.
+ assert 'next_entry.get("kind")' in queued
+ assert queued.index('next_entry.get("kind")') < queued.index("flush_deferred_notes")
def test_every_queue_drain_seam_flushes_before_starting_the_successor(self):
"""A NEW successor-dispatch path that skips the flush must turn this red.
diff --git a/test/test_restart_held_queue.py b/test/test_restart_held_queue.py
new file mode 100644
index 00000000000..60c2050ddb4
--- /dev/null
+++ b/test/test_restart_held_queue.py
@@ -0,0 +1,578 @@
+"""A queue restored after a restart is HELD: no owner runs it, and nothing says it will.
+
+Handed back as cards (``docs/system-specs/modules/session.md``, "Restored entries
+are handed back as queue CARDS, not dispatched"), a restored entry moves only
+when the person acts on it: Send now on the card, an edit, or a delete. These
+tests pin each reader that could read such a queue as work about to run:
+
+- the drain passes held entries by (a turn ending, a fresh send, a Resume);
+- Continue does not refuse on them, so an interrupted chat keeps its Resume;
+- the slot projection and the session status report them apart;
+- the dequeue is durable in the same write as the turn marker, so a death
+ between the two never leaves the entry both as the interrupted opener and as
+ a restored card.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from aiohttp.test_utils import TestClient, TestServer
+from chat_runner_coverage_helpers import _slot, _state
+from chat_test_helpers import _make_state
+from test_chat_slot_continue import _make_app, _mock_state
+
+from kiro_crew.dashboard import chat_runner
+from kiro_crew.dashboard.chat_delivery import queue_entry_view
+from kiro_crew.dashboard.chat_persistence import _rehydrate_slot_from_history, _save_slot_to_history
+from kiro_crew.dashboard.chat_turn.steer_queue import _has_user_queued_followup
+from kiro_crew.dashboard.chat_utils import (
+ CRON_NOTIFICATION_KIND,
+ STEER_INTENT_META,
+ SYNTHETIC_RECOVERY_KIND,
+)
+from kiro_crew.dashboard.slot_queue_repository import (
+ RESTORED_QUEUE_KEY,
+ is_held_queue_entry,
+ release_held_queue_entry,
+)
+from kiro_crew.dashboard.state import _ChatSlot
+
+_RESTART_KIND = "gateway_restart_interruption"
+_RESTART_TEXT = (
+ "This turn was interrupted when the app restarted. "
+ "Review the partial output above, then resume to continue."
+)
+
+
+def _hold(slot: _ChatSlot, text: str) -> str:
+ """Queue *text* as a restore would hand it back: a plain prompt, held."""
+ qid = slot.queue_append(text)
+ slot._queue[-1][RESTORED_QUEUE_KEY] = True
+ return qid
+
+
+def _interrupted(slot: _ChatSlot) -> None:
+ slot.append("user", "first request", "msg msg-u")
+ slot.append("assistant", "partial", "msg msg-a")
+ slot.append("error", _RESTART_TEXT, "msg msg-err", meta={"kind": _RESTART_KIND})
+
+
+def _restart_rows(messages: list[dict]) -> list[dict]:
+ return [m for m in messages if (m.get("meta") or {}).get("kind") == _RESTART_KIND]
+
+
+# ── what the slot reports ──────────────────────────────────────────────────
+
+
+class TestHeldQueueProjection:
+ def test_held_entries_are_counted_apart_from_ready_ones(self):
+ slot = _ChatSlot("s")
+ _hold(slot, "restored one")
+ slot.queue_append("sent after the restart")
+
+ assert slot.queue_depth == 2
+ assert slot.queue_held_depth == 1
+ assert slot.queue_ready_depth == 1
+
+ def test_slot_frame_carries_the_held_count(self):
+ slot = _ChatSlot("s")
+ _interrupted(slot)
+ _hold(slot, "restored one")
+
+ frame = slot.to_dict()
+
+ assert frame["running"] is False
+ assert frame["interrupted"] is True
+ assert frame["queue_depth"] == 1
+ assert frame["queue_held_depth"] == 1
+
+ def test_a_released_card_is_no_longer_shown_held(self):
+ slot = _ChatSlot("s")
+ h1 = _hold(slot, "restored one")
+ release_held_queue_entry(slot._queue, h1)
+
+ assert "held" not in (queue_entry_view(slot._queue[0]).get("meta") or {})
+ assert slot.queue_held_depth == 0
+
+ def test_a_held_entry_with_its_row_written_is_marked_so(self):
+ """Its row is in the transcript, so the card offers Send now and Discard, no Edit."""
+ from kiro_crew.dashboard.chat_delivery import queue_entry_views
+ from kiro_crew.dashboard.responsive_chat import QUEUED_ROW_META_KEY
+
+ slot = _ChatSlot("s")
+ written = _hold(slot, "sent from a crewmate chat")
+ slot.append(
+ "user", "sent from a crewmate chat", "msg msg-u", meta={QUEUED_ROW_META_KEY: written}
+ )
+ _hold(slot, "restored plain")
+
+ first, second = queue_entry_views(slot)
+
+ assert first["meta"] == {"held": True, "rowWritten": True}
+ assert second["meta"] == {"held": True}
+
+ def test_queue_entry_view_marks_only_held_entries(self):
+ slot = _ChatSlot("s")
+ _hold(slot, "restored one")
+ slot.queue_append("live one")
+
+ held, live = (queue_entry_view(item) for item in slot._queue)
+
+ assert held["meta"]["held"] is True
+ assert "held" not in (live.get("meta") or {})
+
+ def test_a_restore_holds_every_entry_it_hands_back(self, tmp_path):
+ state = _make_state(tmp_path)
+ slot = state.get_or_create_slot("chat-1")
+ _interrupted(slot)
+ slot.queue_append("queued before the restart")
+ assert _save_slot_to_history(state, slot, force=True)
+ state._slots.pop("chat-1", None)
+
+ restored = _rehydrate_slot_from_history(state, "chat-1")
+
+ assert restored is not None
+ assert [item["content"] for item in restored._queue] == ["queued before the restart"]
+ assert all(is_held_queue_entry(item) for item in restored._queue)
+ assert restored.to_dict()["queue_held_depth"] == 1
+
+
+# ── the drain passes held entries by ───────────────────────────────────────
+
+
+class TestDrainSkipsHeldEntries:
+ @pytest.mark.asyncio
+ async def test_a_queue_of_only_held_entries_starts_nothing(self, tmp_path):
+ state, slot = _state(tmp_path), _slot()
+ state.subagents = None
+ qid = _hold(slot, "restored one")
+
+ with patch.object(chat_runner, "spawn_guarded_turn", return_value=MagicMock()) as spawn:
+ started = await chat_runner._start_next_queued_turn(state, slot)
+
+ assert started is False
+ spawn.assert_not_called()
+ assert [item["id"] for item in slot._queue] == [qid]
+
+ @pytest.mark.asyncio
+ async def test_a_ready_entry_runs_and_the_held_ones_keep_their_order(self, tmp_path):
+ state, slot = _state(tmp_path), _slot()
+ state.subagents = None
+ h1 = _hold(slot, "restored one")
+ h2 = _hold(slot, "restored two")
+ slot.queue_append("sent after the restart")
+
+ with (
+ patch.object(chat_runner.KiroCrewConfig, "load") as load,
+ patch.object(chat_runner, "spawn_guarded_turn", return_value=MagicMock()) as spawn,
+ patch.object(
+ chat_runner, "_run_chat", new=MagicMock(return_value=MagicMock())
+ ) as run_chat,
+ ):
+ load.return_value.dashboard.merge_queued_messages = True
+ started = await chat_runner._start_next_queued_turn(state, slot)
+
+ assert started is True
+ assert spawn.call_count == 1
+ # Alone, never merged with the held cards behind it.
+ assert run_chat.call_args.args[2] == "sent after the restart"
+ assert [item["id"] for item in slot._queue] == [h1, h2]
+
+ @pytest.mark.asyncio
+ async def test_a_held_steer_is_not_released_by_the_sub_agent_hold(self, tmp_path):
+ """A restored steer keeps its mark; the hold's system scan must pass it by."""
+ state, slot = _state(tmp_path), _slot()
+ state.subagents = MagicMock(running_agents_for=MagicMock(return_value=["agent-1"]))
+ h1 = _hold(slot, "restored steer")
+ slot._queue[0].setdefault("meta", {})[STEER_INTENT_META] = True
+ fresh = slot.queue_append("sent while sub-agents run")
+
+ with patch.object(chat_runner, "spawn_guarded_turn", return_value=MagicMock()) as spawn:
+ started = await chat_runner._start_next_queued_turn(state, slot)
+
+ assert started is False
+ spawn.assert_not_called()
+ # Both still queued: the fresh entry was brought ahead, the held one waits.
+ assert [item["id"] for item in slot._queue] == [fresh, h1]
+ assert is_held_queue_entry(slot._queue[1])
+
+ @pytest.mark.asyncio
+ async def test_a_note_is_not_flushed_into_a_cron_turn_behind_a_held_card(self, tmp_path):
+ """The deferred note waits for a user turn; the held head is not that turn."""
+ state, slot = _state(tmp_path), _slot()
+ state.subagents = None
+ _hold(slot, "restored one")
+ slot.queue_append("cron fired", kind=CRON_NOTIFICATION_KIND)
+
+ with (
+ patch.object(_ChatSlot, "flush_deferred_notes") as flush,
+ patch.object(chat_runner.KiroCrewConfig, "load") as load,
+ patch.object(chat_runner, "spawn_guarded_turn", return_value=MagicMock()),
+ patch.object(chat_runner, "_run_chat", new=MagicMock(return_value=MagicMock())),
+ ):
+ load.return_value.dashboard.merge_queued_messages = False
+ assert await chat_runner._start_next_queued_turn(state, slot) is True
+
+ flush.assert_not_called()
+
+ @pytest.mark.asyncio
+ async def test_send_now_on_a_held_card_runs_exactly_that_card(self, tmp_path):
+ state, slot = _state(tmp_path), _slot()
+ state.subagents = None
+ h1 = _hold(slot, "restored one")
+ h2 = _hold(slot, "restored two")
+
+ with (
+ patch.object(chat_runner.KiroCrewConfig, "load") as load,
+ patch.object(chat_runner, "spawn_guarded_turn", return_value=MagicMock()),
+ patch.object(
+ chat_runner, "_run_chat", new=MagicMock(return_value=MagicMock())
+ ) as run_chat,
+ ):
+ load.return_value.dashboard.merge_queued_messages = False
+ started = await chat_runner._start_next_queued_turn(
+ state, slot, allow_user_during_subagents=True, required_queue_id=h2
+ )
+
+ assert started is True
+ assert run_chat.call_args.args[2] == "restored two"
+ assert [item["id"] for item in slot._queue] == [h1]
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize("row_written", [True, False])
+ async def test_running_a_written_send_writes_the_queue_at_once(self, tmp_path, row_written):
+ """Its row is already on disk, so only the queue can say it has not run."""
+ from kiro_crew.dashboard.responsive_chat import QUEUED_ROW_META_KEY
+
+ state, slot = _state(tmp_path), _slot()
+ state.subagents = None
+ qid = _hold(slot, "sent from a crewmate chat")
+ if row_written:
+ slot.append(
+ "user", "sent from a crewmate chat", "msg msg-u", meta={QUEUED_ROW_META_KEY: qid}
+ )
+
+ with (
+ patch.object(chat_runner.KiroCrewConfig, "load") as load,
+ patch.object(chat_runner, "spawn_guarded_turn", return_value=MagicMock()),
+ patch.object(chat_runner, "_run_chat", new=MagicMock(return_value=MagicMock())),
+ patch.object(chat_runner, "start_queue_persist") as persist,
+ ):
+ load.return_value.dashboard.merge_queued_messages = False
+ started = await chat_runner._start_next_queued_turn(
+ state, slot, allow_user_during_subagents=True, required_queue_id=qid
+ )
+
+ assert started is True
+ assert slot._queue == []
+ assert persist.called is row_written
+
+ @pytest.mark.asyncio
+ async def test_a_released_card_parked_on_memory_preparation_runs_when_it_ends(self, tmp_path):
+ """Idle Send now while memory prepares: the parked drain names no card."""
+ state, slot = _state(tmp_path), _slot()
+ state.subagents = None
+ h1 = _hold(slot, "restored one")
+ release_held_queue_entry(slot._queue, h1)
+ pending = MagicMock()
+ pending.done = MagicMock(return_value=False)
+ state.memory_startup_task = pending
+
+ parked = await chat_runner._start_next_queued_turn(state, slot, required_queue_id=h1)
+ assert parked is False
+ pending.add_done_callback.assert_called_once()
+
+ state.memory_startup_task = None
+ with (
+ patch.object(chat_runner.KiroCrewConfig, "load") as load,
+ patch.object(chat_runner, "spawn_guarded_turn", return_value=MagicMock()),
+ patch.object(
+ chat_runner, "_run_chat", new=MagicMock(return_value=MagicMock())
+ ) as run_chat,
+ ):
+ load.return_value.dashboard.merge_queued_messages = False
+ started = await chat_runner._start_next_queued_turn(state, slot)
+
+ assert started is True
+ assert run_chat.call_args.args[2] == "restored one"
+
+ @pytest.mark.asyncio
+ async def test_held_entries_do_not_park_on_memory_preparation(self, tmp_path):
+ """Only work the drain owns parks; a held card gets no "will run" notice."""
+ state, slot = _state(tmp_path), _slot()
+ state.subagents = None
+ _hold(slot, "restored one")
+ pending = MagicMock()
+ pending.done = MagicMock(return_value=False)
+ state.memory_startup_task = pending
+
+ started = await chat_runner._start_next_queued_turn(state, slot)
+
+ assert started is False
+ assert not any(m.get("role") == "notice" for m in slot.messages)
+ pending.add_done_callback.assert_not_called()
+
+ @pytest.mark.asyncio
+ async def test_a_released_card_runs_on_the_next_drain_that_names_no_card(self, tmp_path):
+ """Send now on a running slot promotes and releases; the turn-end drain runs it."""
+ state, slot = _state(tmp_path), _slot()
+ state.subagents = None
+ h1 = _hold(slot, "restored one")
+ slot.queue_append("sent after the restart")
+ # What /interrupt does for a running slot: promote, then lift the hold.
+ assert slot.queue_promote_by_id(h1)
+ assert release_held_queue_entry(slot._queue, h1)
+
+ with (
+ patch.object(chat_runner.KiroCrewConfig, "load") as load,
+ patch.object(chat_runner, "spawn_guarded_turn", return_value=MagicMock()),
+ patch.object(
+ chat_runner, "_run_chat", new=MagicMock(return_value=MagicMock())
+ ) as run_chat,
+ ):
+ load.return_value.dashboard.merge_queued_messages = False
+ started = await chat_runner._start_next_queued_turn(state, slot)
+
+ assert started is True
+ assert run_chat.call_args.args[2] == "restored one"
+
+ @pytest.mark.asyncio
+ async def test_a_purge_that_empties_the_ready_work_never_falls_through_to_a_held_card(
+ self, tmp_path
+ ):
+ state, slot = _state(tmp_path), _slot()
+ state.subagents = None
+ slot.queue_append("a replay the purge will revoke")
+ h1 = _hold(slot, "restored one")
+
+ async def _revoke_the_head(_state, s):
+ s._queue.pop(0)
+ return False # the queue is not empty, so the drain carries on
+
+ with (
+ patch.object(chat_runner, "_drop_revoked_replays", side_effect=_revoke_the_head),
+ patch.object(chat_runner, "spawn_guarded_turn", return_value=MagicMock()) as spawn,
+ ):
+ started = await chat_runner._start_next_queued_turn(state, slot)
+
+ assert started is False
+ spawn.assert_not_called()
+ assert [item["id"] for item in slot._queue] == [h1]
+
+
+class TestHeldCardsAreNotUserIntervention:
+ def test_held_cards_do_not_cancel_recovery_until_the_person_sends_one(self):
+ slot = _ChatSlot("s")
+ h1 = _hold(slot, "restored one")
+ assert _has_user_queued_followup(slot) is False
+
+ release_held_queue_entry(slot._queue, h1)
+
+ assert _has_user_queued_followup(slot) is True
+
+ def test_a_live_send_still_counts(self):
+ slot = _ChatSlot("s")
+ _hold(slot, "restored one")
+ slot.queue_append("sent after the restart")
+ assert _has_user_queued_followup(slot) is True
+
+
+# ── Continue counts only what the runner will pick up ─────────────────────
+
+
+class TestContinueWithHeldQueue:
+ @pytest.mark.asyncio
+ async def test_resume_runs_ahead_of_held_cards(self, _patched):
+ slot = _ChatSlot("s")
+ _interrupted(slot)
+ h1 = _hold(slot, "restored one")
+ state = _mock_state(slot)
+
+ async with TestClient(TestServer(_make_app(state))) as client:
+ resp = await client.post("/api/chat/slots/s/continue")
+ assert resp.status == 200
+
+ assert slot._queue[0]["kind"] == SYNTHETIC_RECOVERY_KIND
+ assert [item["id"] for item in slot._queue[1:]] == [h1]
+ assert is_held_queue_entry(slot._queue[1])
+ _patched.assert_awaited_once()
+
+ @pytest.mark.asyncio
+ async def test_a_second_press_before_the_recovery_starts_is_refused(self, monkeypatch):
+ """Idempotent: the inserted recovery is ready work, so it blocks a duplicate."""
+ mock_sel = MagicMock()
+ parked = AsyncMock(return_value=False)
+ with (
+ patch("kiro_crew.dashboard.chat_handlers.sel", return_value=mock_sel),
+ patch("kiro_crew.dashboard.chat_handlers._start_next_queued_turn", parked),
+ ):
+ slot = _ChatSlot("s")
+ _interrupted(slot)
+ _hold(slot, "restored one")
+ state = _mock_state(slot)
+ async with TestClient(TestServer(_make_app(state))) as client:
+ first = await client.post("/api/chat/slots/s/continue")
+ second = await client.post("/api/chat/slots/s/continue")
+ assert first.status == 200
+ assert second.status == 409
+ assert (await second.json())["code"] == "slot_queue_pending"
+
+ assert sum(1 for item in slot._queue if item.get("kind") == SYNTHETIC_RECOVERY_KIND) == 1
+ assert slot.queue_held_depth == 1
+
+ @pytest.mark.asyncio
+ async def test_a_ready_entry_still_refuses(self, _patched):
+ slot = _ChatSlot("s")
+ _interrupted(slot)
+ _hold(slot, "restored one")
+ slot.queue_append("sent after the restart")
+ state = _mock_state(slot)
+
+ async with TestClient(TestServer(_make_app(state))) as client:
+ resp = await client.post("/api/chat/slots/s/continue")
+ assert resp.status == 409
+ assert (await resp.json())["code"] == "slot_queue_pending"
+
+ assert slot.queue_depth == 2
+
+
+class TestResumeNeverContinuesAHeldSend:
+ def test_the_interrupted_preamble_skips_a_held_sends_written_row(self):
+ """A crewmate send's row is in the transcript, but it is held, not the cut turn."""
+ from kiro_crew.context import build_interrupted_turn_preamble
+ from kiro_crew.dashboard.chat_turn.prompt_assembly import _rows_without_held_sends
+ from kiro_crew.dashboard.responsive_chat import QUEUED_ROW_META_KEY
+ from kiro_crew.dashboard.state import TURN_OPENING_INJECT_KINDS
+
+ slot = _ChatSlot("s")
+ slot.append("user", "the request the restart cut", "msg msg-u")
+ slot.append("assistant", "partial", "msg msg-a")
+ held = _hold(slot, "sent while it ran")
+ slot.append("user", "sent while it ran", "msg msg-u", meta={QUEUED_ROW_META_KEY: held})
+ slot.append("error", _RESTART_TEXT, "msg msg-err", meta={"kind": _RESTART_KIND})
+
+ rows = _rows_without_held_sends(slot)
+ preamble = build_interrupted_turn_preamble(
+ rows, current=None, opener_inject_kinds=TURN_OPENING_INJECT_KINDS
+ )
+
+ assert "the request the restart cut" in preamble
+ assert "sent while it ran" not in preamble
+ # The transcript itself keeps the row.
+ assert any(m.get("content") == "sent while it ran" for m in slot.messages)
+
+ def test_the_cold_replay_skips_held_rows_on_disk_and_live(self, tmp_path):
+ """A cold-start Resume replays the transcript, which carries the held row on disk."""
+ from kiro_crew.context import build_session_replay
+ from kiro_crew.dashboard.chat_turn.prompt_assembly import _unsent_send_filter
+ from kiro_crew.dashboard.responsive_chat import QUEUED_ROW_META_KEY
+ from kiro_crew.history import ConversationLog
+
+ log = ConversationLog(base_dir=tmp_path)
+ slot = _ChatSlot("s")
+ slot.append("user", "the request the restart cut", "msg msg-u")
+ slot.append("assistant", "partial answer", "msg msg-a")
+ held = _hold(slot, "held while it ran")
+ slot.append("user", "held while it ran", "msg msg-u", meta={QUEUED_ROW_META_KEY: held})
+ log.rewrite_session("dashboard:s", [dict(row) for row in slot.messages])
+
+ unsent = _unsent_send_filter(slot)
+ assert unsent is not None
+ for pending in ([], list(slot.messages)):
+ replay = build_session_replay(
+ log, "dashboard:s", pending_messages=pending, current_message=None, skip_row=unsent
+ )
+ assert replay is not None
+ assert "the request the restart cut" in replay
+ assert "held while it ran" not in replay
+
+ def test_the_filter_is_off_when_nothing_is_held(self):
+ from kiro_crew.dashboard.chat_turn.prompt_assembly import _unsent_send_filter
+
+ assert _unsent_send_filter(_ChatSlot("s")) is None
+
+
+class TestAgentsAreToldHeldIsNotQueued:
+ def test_the_session_tools_count_only_entries_that_will_run(self):
+ from kiro_crew.mcp_dashboard_session_runs import _queue_phrase
+
+ assert _queue_phrase({"queue_depth": 2, "queue_held_depth": 2}) == ""
+ assert _queue_phrase({"queue_depth": 3, "queue_held_depth": 1}, unit=" message(s)") == (
+ ", 2 message(s) queued"
+ )
+ # An older gateway sends no held count: every entry reads as queued.
+ assert _queue_phrase({"queue_depth": 1}) == ", 1 queued"
+
+
+class TestSendNowWhileMemoryPrepares:
+ @pytest.mark.asyncio
+ async def test_send_now_on_a_held_card_is_refused_as_memory_preparing(
+ self, _patch_interrupt_sel
+ ):
+ """The drain would park it and never run it, so the route says why and keeps it held."""
+ from test_chat_slot_interrupt import _make_app as _interrupt_app
+ from test_chat_slot_interrupt import _mock_state as _interrupt_state
+
+ slot = _ChatSlot("test")
+ qid = _hold(slot, "restored one")
+ state = _interrupt_state(slot)
+ pending = MagicMock()
+ pending.done = MagicMock(return_value=False)
+ state.memory_startup_task = pending
+ start = AsyncMock(return_value=False)
+
+ with patch("kiro_crew.dashboard.chat_handlers._start_next_queued_turn", new=start):
+ async with TestClient(TestServer(_interrupt_app(state))) as client:
+ resp = await client.post("/api/chat/slots/test/interrupt", json={"queue_id": qid})
+ data = await resp.json()
+
+ assert resp.status == 409
+ assert data["code"] == "memory_preparing"
+ start.assert_not_called()
+ assert is_held_queue_entry(slot._queue[0])
+
+
+@pytest.fixture
+def _patch_interrupt_sel():
+ with patch("kiro_crew.dashboard.chat_handlers.sel", return_value=MagicMock()):
+ yield
+
+
+# ── repeated recovery is stable ────────────────────────────────────────────
+
+
+def test_repeated_restarts_keep_one_notice_and_the_same_held_cards(tmp_path):
+ state = _make_state(tmp_path)
+ slot = state.get_or_create_slot("chat-1")
+ slot.append("user", "first request", "msg msg-u")
+ slot._turn_in_flight_generation = 3
+ slot._turn_in_flight_prompt = None
+ slot.queue_append("queued one")
+ slot.queue_append("queued two")
+ assert _save_slot_to_history(state, slot, force=True)
+ state._slots.pop("chat-1", None)
+
+ first = _rehydrate_slot_from_history(state, "chat-1")
+ assert first is not None
+ assert _save_slot_to_history(state, first, force=True)
+ state._slots.pop("chat-1", None)
+ second = _rehydrate_slot_from_history(state, "chat-1")
+
+ assert second is not None
+ assert len(_restart_rows(second.messages)) <= 1
+ assert [item["content"] for item in second._queue] == ["queued one", "queued two"]
+ assert second.queue_held_depth == 2
+ assert second.queue_ready_depth == 0
+
+
+@pytest.fixture
+def _patched():
+ mock_sel = MagicMock()
+ started = AsyncMock(return_value=True)
+ with (
+ patch("kiro_crew.dashboard.chat_handlers.sel", return_value=mock_sel),
+ patch("kiro_crew.dashboard.chat_handlers._start_next_queued_turn", started),
+ ):
+ yield started
diff --git a/test/test_state_parts_facade.py b/test/test_state_parts_facade.py
index 5808629c589..25987f5ed9c 100644
--- a/test/test_state_parts_facade.py
+++ b/test/test_state_parts_facade.py
@@ -273,6 +273,8 @@
"False) -> 'str'",
),
"queue_depth": ("property", False, False, "(self) -> 'int'"),
+ "queue_held_depth": ("property", False, False, "(self) -> 'int'"),
+ "queue_ready_depth": ("property", False, False, "(self) -> 'int'"),
"queue_edit_by_id": (
"function",
False,
diff --git a/website/src/components/ChatPane.tsx b/website/src/components/ChatPane.tsx
index 728364e4b79..badc6cc63b3 100644
--- a/website/src/components/ChatPane.tsx
+++ b/website/src/components/ChatPane.tsx
@@ -303,8 +303,17 @@ export default function ChatPane({
const composerBusy = busy || awaitingReply
// Queued rows are drawn in place from the slot's full row list, so a parked
// message sits where it was sent (`withQueuedAsBubbles`).
+ // A held message (restored after a restart) is the exception: it is not
+ // waiting for the crewmate, nothing sends it, so it is not drawn as a sent
+ // bubble but as a Held card the person can edit or take back.
+ const crewmateHeld = useMemo(
+ () => (crewmate ? queuedMessages.filter(m => m.meta?.held === true) : []),
+ [crewmate, queuedMessages],
+ )
const messages = useMemo(
- () => (crewmate ? filterCrewmateChat(withQueuedAsBubbles(allMessages, queuedMessages)) : paneMessages),
+ () => (crewmate
+ ? filterCrewmateChat(withQueuedAsBubbles(allMessages, queuedMessages.filter(m => m.meta?.held !== true)))
+ : paneMessages),
[crewmate, allMessages, paneMessages, queuedMessages],
)
// The host's greeting is drawn while it is still the newest word: a reply
@@ -920,6 +929,13 @@ export default function ChatPane({
{!crewmate && queuedMessages.length > 0 && (
)}
+ {/* A crewmate's held messages (restored after a restart) are the one
+ queue a crewmate chat shows as cards, and the one place it offers
+ Send now: nothing else will send them. Edit and Cancel are withheld
+ when the gateway already wrote the message's row (rowWritten). */}
+ {crewmate && crewmateHeld.length > 0 && (
+
+ )}
{/* The pending ask_question card shares the message column's gutter and
--mc-content-width clamp, so it lines up with the messages above it
diff --git a/website/src/components/QueueStack.tsx b/website/src/components/QueueStack.tsx
index 1d9bb602aba..dd0b8043629 100644
--- a/website/src/components/QueueStack.tsx
+++ b/website/src/components/QueueStack.tsx
@@ -20,9 +20,21 @@ import { usePasteTokens } from '../chat-core/composer/pasteEditor'
* queue card is noise at scale — they collapse into one progress line
* (SubagentDeliveryProgress) instead of the interactive QueueStack. */
export function isSystemDelivery(m: ChatMessage): boolean {
+ // A held entry (restored after a restart) is the person's own words whatever
+ // its text opens with: the gateway restores only plain prompts. It must stay
+ // an actionable card on every surface, since only the person moves it.
+ if (isHeldQueued(m)) return false
return hasSubagentCompletionPrefix(m.content || '')
}
+/** A queued entry the gateway restored after a restart and holds for the
+ * person (`is_held_queue_entry`): no drain starts it, so every surface that
+ * draws queue cards must draw it with Send now. Read from `meta.held`, the
+ * server's structural flag, never from the text. */
+export function isHeldQueued(m: ChatMessage): boolean {
+ return m.meta?.held === true
+}
+
/** A queued entry that must NOT render as an interactive (edit/cancel) user
* card. Two families qualify, both machine orchestration rather than user
* speech:
@@ -35,6 +47,7 @@ export function isSystemDelivery(m: ChatMessage): boolean {
* filtered out of the QueueStack (sub-agent deliveries are still counted for
* the progress line via isSystemDelivery). */
export function isNonInteractiveQueued(m: ChatMessage): boolean {
+ if (isHeldQueued(m)) return false
return isSystemDelivery(m) || parseRecoveryMessage(m.content || '') !== null
}
@@ -635,6 +648,9 @@ function QueueStackInner({ messages, onCancel, onInterrupt, onEdit, onEditDraft,
) : null
+ // Restored after a restart: nothing will send it on its own, so the
+ // card says so in words and drops the "next up" hourglass motion.
+ const isHeld = m.meta?.held === true
return (
// The motion box only places the card (peek offset, scale, layer);
@@ -679,7 +695,17 @@ function QueueStackInner({ messages, onCancel, onInterrupt, onEdit, onEditDraft,
>
{i + 1}
- {isFrontCollapsed && (
+ {isHeld ? (
+
+ {i18nT('components.queueStack.held')}
+
+ ) : isFrontCollapsed && (
@@ -756,7 +782,7 @@ function QueueStackInner({ messages, onCancel, onInterrupt, onEdit, onEditDraft,
)}
- {onEdit && showActions && !isAppEntry && (
+ {onEdit && showActions && !isAppEntry && m.meta?.rowWritten !== true && (
)}
- {onCancel && showActions && (
+ {/* A held card whose row is already in the transcript has
+ no Cancel: the gateway refuses it (409
+ `queue_item_already_sent`), and Send now is its way out. */}
+ {onCancel && showActions && m.meta?.rowWritten !== true && (