From b51a83205093bb17645fec90dabdeb41f5e40f9a Mon Sep 17 00:00:00 2001 From: Matt Toohey Date: Sun, 1 Nov 2026 11:24:14 +1100 Subject: [PATCH 1/9] feat(staged): send chat messages to a running session now, beside queueing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The session and note chat composers could only queue a follow-up while a session ran. Project notes recently gained live delivery through the `continue_repo_session` MCP tool; this gives the composer the same path. Capability flag. A new `get_session_live_delivery` command (Tauri + web) reports `{ supported, reason }` for a session, computed by the backend from the same gate `route_continue_session` applies before delivering to a running row: the row is `running`, this process owns it (not another Staged instance sharing the database), and its ACP connection has handed out a live turn-control handle. Read without the startup wait, so a row still connecting is "queue only" with a reason. The chat pane asks on each 500 ms poll tick until the answer is yes and drops the flag whenever the status leaves `running`; nothing is duplicated client-side. Send-now command. `send_session_message_now` wraps the shared `continue_session_for_store` body — steer into the turn in flight, else interrupt and re-prompt, resume a stopped session, refuse a queued one — and returns the same `delivery` / `reason` spelling as the MCP tool. The web arm runs detached like `resume_session`, since a resume may be in flight when the client disconnects. UI. `composerSendControl` derives the submit control: a plain "Send" (start/resume) when the session is idle, a plain "Queue" when it runs but cannot take a message now, and a split button once the backend says it can — the primary segment performs the remembered mode (Queue / Send) and a dropdown (existing dropdown-menu component, accent button styling) switches it. Live "Send" success lets the poll pick up the persisted user row; `blocked` / `rejected_queued` / `not_resumable` surface the reason and restore the draft rather than silently re-queueing, because a steer the agent never acknowledged may already have landed. Not-running sessions. While a session is queued, completed, failed, or cancelled the composer renders exactly what it did before (single button); "Send" on a stopped session resumes it, which is what `continue_repo_session` does for those states. "Send" is never offered for a queued session, matching the tool's refusal. Persistence. The mode is a global preference, `composer-send-mode` in the existing Tauri-store-backed preferences, defaulting to `queue`; malformed values fall back to the default. Tests: live-delivery flag for connected / connecting / other-instance / not-running rows and the delivery reply shape (Rust); mode persistence, default, and malformed-value fallback across native and web store clients; the single-vs-split render decision in `composerSendControl`. Co-Authored-By: Claude Fable 5.1 Signed-off-by: Matt Toohey --- apps/staged/src-tauri/src/lib.rs | 2 + apps/staged/src-tauri/src/session_commands.rs | 240 ++++++++++++++++++ apps/staged/src-tauri/src/web_server.rs | 36 +++ apps/staged/src/lib/commands.ts | 27 ++ .../lib/features/sessions/ChatComposer.svelte | 114 +++++++-- .../features/sessions/SessionChatPane.svelte | 115 ++++++++- .../sessions/composerSendControl.test.ts | 57 +++++ .../features/sessions/composerSendControl.ts | 94 +++++++ .../features/settings/preferences.svelte.ts | 35 +++ .../lib/features/settings/preferences.test.ts | 27 ++ apps/staged/src/lib/types.ts | 27 ++ 11 files changed, 754 insertions(+), 20 deletions(-) create mode 100644 apps/staged/src/lib/features/sessions/composerSendControl.test.ts create mode 100644 apps/staged/src/lib/features/sessions/composerSendControl.ts diff --git a/apps/staged/src-tauri/src/lib.rs b/apps/staged/src-tauri/src/lib.rs index 5a14b7788..dc0598179 100644 --- a/apps/staged/src-tauri/src/lib.rs +++ b/apps/staged/src-tauri/src/lib.rs @@ -2341,6 +2341,8 @@ pub fn run() { session_commands::list_queued_session_messages, session_commands::delete_queued_session_message, session_commands::send_queued_session_message, + session_commands::send_session_message_now, + session_commands::get_session_live_delivery, session_commands::build_note_followup_message, session_commands::cancel_session, session_commands::stop_session_async_task, diff --git a/apps/staged/src-tauri/src/session_commands.rs b/apps/staged/src-tauri/src/session_commands.rs index 215e7c6c2..349007473 100644 --- a/apps/staged/src-tauri/src/session_commands.rs +++ b/apps/staged/src-tauri/src/session_commands.rs @@ -2333,6 +2333,123 @@ pub fn delete_queued_session_message( .map_err(|e| e.to_string()) } +/// Whether a message typed for `session_id` can reach its agent *now* (the +/// chat composer's "Send" mode) — exactly the gate `route_continue_session` +/// applies to a `running` row before delivering, read without waiting: +/// the row is `running`, this process owns it, and its ACP connection has +/// handed out a live turn-control handle. Anything else is "queue only", +/// with `reason` saying why, so the composer can hide the mode switch +/// rather than offer a send that would come back blocked. +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct SessionLiveDelivery { + pub supported: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub reason: Option, +} + +impl SessionLiveDelivery { + fn unsupported(reason: impl Into) -> Self { + Self { + supported: false, + reason: Some(reason.into()), + } + } +} + +pub(crate) fn live_delivery_for_session( + store: &Store, + registry: &session_runner::SessionRegistry, + session_id: &str, +) -> Result { + let session = store + .get_session(session_id) + .map_err(|e| e.to_string())? + .ok_or_else(|| format!("Session not found: {session_id}"))?; + if session.status != store::SessionStatus::Running { + return Ok(SessionLiveDelivery::unsupported(format!( + "the session is {}", + session.status.as_str() + ))); + } + if let Some(owner_pid) = session.owner_pid.filter(|pid| *pid != std::process::id()) { + return Ok(SessionLiveDelivery::unsupported(format!( + "the session is owned by another Staged instance (pid {owner_pid})" + ))); + } + match registry.turn_control_handle(session_id) { + Some(handle) if handle.is_connection_alive() => Ok(SessionLiveDelivery { + supported: true, + reason: None, + }), + _ => Ok(SessionLiveDelivery::unsupported( + "the session's agent has not connected yet", + )), + } +} + +#[tauri::command] +pub fn get_session_live_delivery( + store: tauri::State<'_, Mutex>>>, + registry: tauri::State<'_, Arc>, + session_id: String, +) -> Result { + let store = get_store(&store)?; + live_delivery_for_session(&store, ®istry, &session_id) +} + +/// The `send_session_message_now` reply: how the message reached (or failed +/// to reach) the agent, in the wire spelling the project MCP server's +/// `continue_repo_session` tool uses, with the reason when it did not. +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct SessionMessageDelivery { + pub delivery: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub reason: Option, +} + +impl From<&ContinueOutcome> for SessionMessageDelivery { + fn from(outcome: &ContinueOutcome) -> Self { + Self { + delivery: outcome.delivery().to_string(), + reason: outcome.reason().map(str::to_string), + } + } +} + +/// Deliver a chat message to a session now instead of queueing it: the +/// composer's "Send" mode. Shares `continue_session_for_store` with the +/// project MCP server's `continue_repo_session` tool, so a running session +/// is steered (or interrupted and re-prompted), a stopped one is resumed, +/// and a queued one is refused — the reply says which. +#[tauri::command] +#[allow(clippy::too_many_arguments)] +pub async fn send_session_message_now( + store: tauri::State<'_, Mutex>>>, + registry: tauri::State<'_, Arc>, + action_executor: tauri::State<'_, Arc>, + action_registry: tauri::State<'_, Arc>, + app_handle: tauri::AppHandle, + session_id: String, + content: String, + image_ids: Option>, +) -> Result { + let store = get_store(&store)?; + let outcome = continue_session_for_store( + store, + Arc::clone(®istry), + Arc::clone(&action_executor), + Arc::clone(&action_registry), + app_handle, + session_id, + content, + image_ids.unwrap_or_default(), + ) + .await?; + Ok(SessionMessageDelivery::from(&outcome)) +} + #[tauri::command] pub async fn send_queued_session_message( store: tauri::State<'_, Mutex>>>, @@ -8244,6 +8361,129 @@ mod tests { ); } + /// The composer's live-delivery flag is the running-row gate of + /// `route_continue_session` read without the wait: a `running` row this + /// process owns, with a live turn-control handle, can take a message now. + #[tokio::test(flavor = "current_thread")] + async fn live_delivery_is_supported_for_a_running_session_with_a_live_connection() { + let store = Arc::new(Store::in_memory().unwrap()); + let session = create_session_with_status(&store, "prompt", store::SessionStatus::Running); + let registry = test_registry(); + let _registration = registry.register_external(&session.id); + let (handle, _) = + acp_client::TurnControlHandle::scripted(Ok(acp_client::SteerOutcome::Injected), Ok(())); + registry.set_turn_control_handle(&session.id, handle); + + let capability = live_delivery_for_session(&store, ®istry, &session.id).unwrap(); + + assert_eq!( + capability, + SessionLiveDelivery { + supported: true, + reason: None + } + ); + } + + /// A `running` row still in its startup window (no connection yet) and + /// one whose connection has gone are both "queue only" — the flag never + /// waits for a handle the way the delivery itself does. + #[test] + fn live_delivery_is_unsupported_until_the_agent_connects() { + let store = Arc::new(Store::in_memory().unwrap()); + let session = create_session_with_status(&store, "prompt", store::SessionStatus::Running); + let registry = test_registry(); + let _registration = registry.register_external(&session.id); + + let starting = live_delivery_for_session(&store, ®istry, &session.id).unwrap(); + assert!(!starting.supported); + assert!( + starting + .reason + .as_deref() + .unwrap() + .contains("not connected"), + "{starting:?}" + ); + + registry.set_turn_control_handle(&session.id, acp_client::TurnControlHandle::detached()); + let dead = live_delivery_for_session(&store, ®istry, &session.id).unwrap(); + assert!(!dead.supported); + } + + /// Stopped and queued rows are not live: the composer keeps its plain + /// button for them (a stopped session's "Send" resumes it, as + /// `continue_session_for_store` would; a queued one cannot take a + /// message out of FIFO order). + #[test] + fn live_delivery_is_unsupported_for_rows_that_are_not_running() { + let store = Arc::new(Store::in_memory().unwrap()); + let registry = test_registry(); + + let queued = create_session_with_status(&store, "prompt", store::SessionStatus::Queued); + let capability = live_delivery_for_session(&store, ®istry, &queued.id).unwrap(); + assert!(!capability.supported); + assert_eq!(capability.reason.as_deref(), Some("the session is queued")); + + let completed = create_session_with_status(&store, "prompt", store::SessionStatus::Running); + store + .update_session_status( + &completed.id, + store::SessionStatus::Completed, + None, + Some(&store::CompletionReason::TurnComplete), + ) + .unwrap(); + let capability = live_delivery_for_session(&store, ®istry, &completed.id).unwrap(); + assert!(!capability.supported); + assert_eq!( + capability.reason.as_deref(), + Some("the session is completed") + ); + + assert!(live_delivery_for_session(&store, ®istry, "missing").is_err()); + } + + /// A `running` row another Staged instance owns has its connection in + /// that process; a handle here would never appear, so it is queue only. + #[tokio::test(flavor = "current_thread")] + async fn live_delivery_is_unsupported_for_a_session_owned_by_another_instance() { + let store = Arc::new(Store::in_memory().unwrap()); + let mut session = store::Session::new_running("prompt", Path::new("/tmp")); + let other_pid = std::process::id() + 1; + session.owner_pid = Some(other_pid); + store.create_session(&session).unwrap(); + let registry = test_registry(); + let _registration = registry.register_external(&session.id); + let (handle, _) = + acp_client::TurnControlHandle::scripted(Ok(acp_client::SteerOutcome::Injected), Ok(())); + registry.set_turn_control_handle(&session.id, handle); + + let capability = live_delivery_for_session(&store, ®istry, &session.id).unwrap(); + + assert!(!capability.supported); + let reason = capability.reason.unwrap(); + assert!(reason.contains("another Staged instance"), "{reason}"); + assert!(reason.contains(&other_pid.to_string()), "{reason}"); + } + + /// The wire reply of `send_session_message_now` spells the outcome the + /// way `continue_repo_session` does, reason included. + #[test] + fn session_message_delivery_mirrors_the_continue_outcome() { + let injected = SessionMessageDelivery::from(&ContinueOutcome::Injected); + assert_eq!(injected.delivery, "injected"); + assert_eq!(injected.reason, None); + + let blocked = SessionMessageDelivery::from(&ContinueOutcome::blocked("still starting")); + assert_eq!(blocked.delivery, "blocked"); + assert_eq!(blocked.reason.as_deref(), Some("still starting")); + + let rejected = SessionMessageDelivery::from(&ContinueOutcome::RejectedQueued); + assert_eq!(rejected.delivery, "rejected_queued"); + assert_eq!(rejected.reason, None); + } + /// A running session with a connection is steered first; the agent took /// the message into its turn, so nothing else is sent. #[tokio::test(flavor = "current_thread")] diff --git a/apps/staged/src-tauri/src/web_server.rs b/apps/staged/src-tauri/src/web_server.rs index cc0e419a5..0f3cc07c5 100644 --- a/apps/staged/src-tauri/src/web_server.rs +++ b/apps/staged/src-tauri/src/web_server.rs @@ -3679,6 +3679,42 @@ async fn dispatch(command: &str, args: Value, state: &WebAppState) -> Result { + let store = get_store(store_mutex)?; + let session_id: String = arg(&args, "sessionId")?; + let content: String = arg(&args, "content")?; + let image_ids: Option> = opt_arg(&args, "imageIds")?; + let app_handle = app_handle.clone(); + let session_registry = Arc::clone(session_registry); + let action_executor = Arc::clone(action_executor); + let action_registry = Arc::clone(action_registry); + // Detached like `resume_session`: a stopped session is resumed + // through the same launch, and a client that disconnects while + // that resume is mid-flight must not abandon it half-started. + let delivery = detached(async move { + let outcome = session_commands::continue_session_for_store( + store, + session_registry, + action_executor, + action_registry, + app_handle, + session_id, + content, + image_ids.unwrap_or_default(), + ) + .await?; + Ok(session_commands::SessionMessageDelivery::from(&outcome)) + }) + .await?; + Ok(serde_json::to_value(delivery).unwrap()) + } + "get_session_live_delivery" => { + let store = get_store(store_mutex)?; + let session_id: String = arg(&args, "sessionId")?; + let capability = + session_commands::live_delivery_for_session(&store, session_registry, &session_id)?; + Ok(serde_json::to_value(capability).unwrap()) + } "build_note_followup_message" => { let store = get_store(store_mutex)?; let session_id: String = arg(&args, "sessionId")?; diff --git a/apps/staged/src/lib/commands.ts b/apps/staged/src/lib/commands.ts index ca699d9b9..eded306b2 100644 --- a/apps/staged/src/lib/commands.ts +++ b/apps/staged/src/lib/commands.ts @@ -35,6 +35,8 @@ import type { SessionPollRequest, ToolCallBody, SessionBackgroundHoldSnapshot, + SessionLiveDelivery, + SessionMessageDelivery, QueuedSessionMessage, DiffFilesResponse, FileDiff, @@ -1153,6 +1155,31 @@ export function getSessionBackgroundHold( return invokeCommand('get_session_background_hold', { sessionId }); } +/** + * Whether the session can take a message right now (composer "Send" mode). + * Cheap: a registry lookup, no agent round trip. + */ +export function getSessionLiveDelivery(sessionId: string): Promise { + return invokeCommand('get_session_live_delivery', { sessionId }); +} + +/** + * Deliver a message to the session now instead of queueing it: steered into + * the running turn (or interrupting it) over the live connection, or resuming + * a stopped session — the same path as the `continue_repo_session` MCP tool. + */ +export function sendSessionMessageNow( + sessionId: string, + content: string, + imageIds?: string[] +): Promise { + return invokeCommand('send_session_message_now', { + sessionId, + content, + imageIds: imageIds ?? null, + }); +} + export function deleteSession(sessionId: string): Promise { return invokeCommand('delete_session', { sessionId }); } diff --git a/apps/staged/src/lib/features/sessions/ChatComposer.svelte b/apps/staged/src/lib/features/sessions/ChatComposer.svelte index 92a476020..d8afd452d 100644 --- a/apps/staged/src/lib/features/sessions/ChatComposer.svelte +++ b/apps/staged/src/lib/features/sessions/ChatComposer.svelte @@ -6,6 +6,12 @@ SessionChatPane, which backs both the session dialog and the note dialog's chat pane, so the composer looks and behaves identically everywhere: the text input gets its own full-width row with the controls beneath it. + + The submit control follows `composerSendControl`: a plain "Send" on an idle + session, a plain "Queue" on a running one — and, when the backend reports + the running session can take a message now (`canSendLive`), a split button + whose primary segment performs the remembered mode (Queue / Send) and whose + dropdown switches it. --> + +{#if message} + +{/if} + + diff --git a/apps/staged/src/lib/features/sessions/SessionChatPane.svelte b/apps/staged/src/lib/features/sessions/SessionChatPane.svelte index 795c82d76..ae8cee664 100644 --- a/apps/staged/src/lib/features/sessions/SessionChatPane.svelte +++ b/apps/staged/src/lib/features/sessions/SessionChatPane.svelte @@ -104,7 +104,8 @@ setComposerSendMode, } from '../settings/preferences.svelte'; import ChatComposer, { composerControlClass } from './ChatComposer.svelte'; - import { composerSendControl } from './composerSendControl'; + import ComposerError from './ComposerError.svelte'; + import { composerSendControl, composerSessionBlock } from './composerSendControl'; import { LIVE_DELIVERY_UNKNOWN, liveDeliveryKnowledge, @@ -602,6 +603,8 @@ async function handleAttachFiles(files: File[]) { if (!projectId) return; attachError = null; + // Attaching is a draft edit: the last failure line has been read. + sendError = null; const rejections: AttachRejection[] = []; for (const file of files) { const reason = await addImageFile(file); @@ -659,6 +662,8 @@ imagePreviews = new Map(imagePreviews); imagePreviews.delete(imageId); attachError = null; + // Removing an attachment is a draft edit too. + sendError = null; deleteImage(imageId).catch((err) => { console.error('Failed to delete image:', err); }); @@ -675,6 +680,7 @@ // Surface rejection reasons (signature, size) as paste does, reporting // the whole batch once so a later success cannot hide an earlier one. attachError = null; + sendError = null; const rejections: AttachRejection[] = []; for (const path of imagePaths) { try { @@ -1278,8 +1284,9 @@ * `error`: that state replaces the whole transcript with a centered card, * which hides the very conversation the user is trying to continue, and * (see `reopenMode`) would turn the next reopen into a full reload. - * Cleared when the next action starts, when the draft is edited, and when - * the pane moves to another session. + * Cleared when the next action starts, when the draft is edited (text, + * attachments), and when the pane moves to another session. Rendered by + * `ChatComposer`, or by the pipeline stop area that stands in for it. */ let sendError = $state(null); @@ -1294,14 +1301,18 @@ error = null; } - /** The submit control is disabled until the session row has loaded. */ - let composerLoading = $derived(loading || !session); + /** + * Why the composer has no session row to submit to (loading, or the load + * settled without one), or `null`. The submit control is disabled while + * set, with a title saying which. + */ + let composerBlock = $derived(composerSessionBlock({ loading, hasSession: session !== null })); async function handleSend() { - // Until the row has loaded there is nothing to send to; the control is - // disabled, and Cmd+Enter keeps the draft rather than taking it out of - // the box only to put it back. - if (composerLoading) return; + // Without a row there is nothing to send to; the control is disabled, and + // Cmd+Enter keeps the draft rather than taking it out of the box only to + // put it back. + if (composerBlock) return; // A send still in flight (a live send can take a few seconds: steer // acknowledgement, interrupt-and-reprompt, or a resume) owns the composer // until it settles; a second Cmd+Enter meanwhile must not clear the draft @@ -1512,6 +1523,7 @@ async function handleDeleteQueuedMessage(id: string) { if (queueActionIds.has(id)) return; queueActionIds = new Set(queueActionIds).add(id); + beginComposerAction(); try { await deleteQueuedSessionMessage(id); await refreshQueuedMessages(); @@ -1549,6 +1561,7 @@ async function handleCancel() { if (!session || cancelling) return; cancelling = true; + beginComposerAction(); try { await cancelSession(session.id); } catch (e) { @@ -2974,8 +2987,11 @@
{#if isPipelineRunning && grouped.length === 0} +
- + +