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/project_mcp.rs b/apps/staged/src-tauri/src/project_mcp.rs
index ad806db4d..08255ab01 100644
--- a/apps/staged/src-tauri/src/project_mcp.rs
+++ b/apps/staged/src-tauri/src/project_mcp.rs
@@ -1695,6 +1695,7 @@ mod tests {
                 crate::session_commands::ContinueWaits::uniform(std::time::Duration::from_millis(
                     200,
                 )),
+                crate::session_commands::StoppedSessionPolicy::Resume,
             )
             .await
             .unwrap();
diff --git a/apps/staged/src-tauri/src/session_commands.rs b/apps/staged/src-tauri/src/session_commands.rs
index 215e7c6c2..be140cee9 100644
--- a/apps/staged/src-tauri/src/session_commands.rs
+++ b/apps/staged/src-tauri/src/session_commands.rs
@@ -1821,6 +1821,10 @@ pub(crate) enum ContinueOutcome {
     /// A stopped session that cannot be resumed: no agent conversation to
     /// load, or its worktree is gone. The caller should start a new session.
     NotResumable { reason: String },
+    /// The session is stopped and the caller asked for live delivery only
+    /// (`StoppedSessionPolicy::Refuse`): nothing was sent and the row was
+    /// not touched. The caller resumes it through its own path.
+    NotRunning { reason: String },
 }
 
 impl ContinueOutcome {
@@ -1833,14 +1837,15 @@ impl ContinueOutcome {
             ContinueOutcome::RejectedQueued => "rejected_queued",
             ContinueOutcome::Blocked { .. } => "blocked",
             ContinueOutcome::NotResumable { .. } => "not_resumable",
+            ContinueOutcome::NotRunning { .. } => "not_running",
         }
     }
 
     pub(crate) fn reason(&self) -> Option<&str> {
         match self {
-            ContinueOutcome::Blocked { reason } | ContinueOutcome::NotResumable { reason } => {
-                Some(reason)
-            }
+            ContinueOutcome::Blocked { reason }
+            | ContinueOutcome::NotResumable { reason }
+            | ContinueOutcome::NotRunning { reason } => Some(reason),
             _ => None,
         }
     }
@@ -1856,6 +1861,28 @@ impl ContinueOutcome {
             reason: reason.into(),
         }
     }
+
+    fn not_running(reason: impl Into<String>) -> Self {
+        ContinueOutcome::NotRunning {
+            reason: reason.into(),
+        }
+    }
+}
+
+/// What `route_continue_session` does with a session that is stopped
+/// (`completed`, `cancelled`, or `error`) — including one that stops while
+/// the request waits on its connection.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum StoppedSessionPolicy {
+    /// Claim it for a resume with the message as its next prompt, as the
+    /// `continue_repo_session` MCP tool does.
+    Resume,
+    /// Refuse with `ContinueOutcome::NotRunning` and leave the row alone.
+    /// The chat composer's "Send" uses this: its own resume path applies a
+    /// branch override and follow-up config that the claimed resume does
+    /// not, so a session that stopped between the composer's last poll and
+    /// the click must fall back to that path rather than be resumed here.
+    Refuse,
 }
 
 /// A stopped session `continue_resume_gate` claimed for a resume. Its row is
@@ -1892,8 +1919,9 @@ pub(crate) enum ContinueRoute {
 
 /// Deliver `message` to a session now, whatever state it is in.
 ///
-/// The Tauri/web `send_session_message_now` command and the project MCP
-/// server's `continue_repo_session` tool share this body. A `running`
+/// The body of the project MCP server's `continue_repo_session` tool (the
+/// Tauri/web `send_session_message_now` command shares the routing but not
+/// the resume; see `send_session_message_now_for_store`). A `running`
 /// session is reached over its live ACP connection — steering first,
 /// interrupt-and-reprompt when the agent refuses — so the row never leaves
 /// `running` and the runner never sees a turn boundary; a stopped session is
@@ -1922,6 +1950,7 @@ pub(crate) async fn continue_session_for_store(
         &message,
         &image_ids,
         ContinueWaits::DEFAULT,
+        StoppedSessionPolicy::Resume,
     )
     .await?;
     let claim = match route {
@@ -1943,6 +1972,40 @@ pub(crate) async fn continue_session_for_store(
     .map(|()| ContinueOutcome::Resumed)
 }
 
+/// Deliver `message` to a `running` session now, and only then: the body of
+/// the Tauri/web `send_session_message_now` command. The live paths are the
+/// ones `continue_session_for_store` takes (steer, else interrupt and
+/// re-prompt) and a `queued` session is refused the same way; a stopped
+/// session comes back `NotRunning` instead of being resumed, so the composer
+/// can resume it through `resume_session` with its branch override and
+/// follow-up config. Needs no app handle, since nothing here launches.
+pub(crate) async fn send_session_message_now_for_store(
+    store: &Arc<Store>,
+    registry: &Arc<session_runner::SessionRegistry>,
+    session_id: &str,
+    message: &str,
+    image_ids: &[String],
+) -> Result<ContinueOutcome, String> {
+    match route_continue_session(
+        store,
+        registry,
+        session_id,
+        message,
+        image_ids,
+        ContinueWaits::DEFAULT,
+        StoppedSessionPolicy::Refuse,
+    )
+    .await?
+    {
+        ContinueRoute::Delivered(outcome) => Ok(outcome),
+        ContinueRoute::Resume(claim) => Err(format!(
+            "send_session_message_now: session {} was claimed for a resume under the refuse \
+             policy",
+            claim.session.id
+        )),
+    }
+}
+
 /// Finish the resume of a session `continue_resume_gate` claimed: the same
 /// preparation and launch as `resume_session_for_store`, without the claim
 /// it already made. No `branch_id` override and no config selection: the
@@ -1997,7 +2060,9 @@ pub(crate) async fn resume_claimed_session_for_store(
 /// preparation and launch, which need the app handle. Loops because the
 /// status can move under the request: a `running` session whose connection
 /// ends while this waits for its handle is routed again on the status it
-/// ended with. Each wait takes its own budget from `waits`.
+/// ended with. Each wait takes its own budget from `waits`; `stopped` says
+/// whether a session found stopped (at the start or after such a re-route)
+/// is claimed for a resume or refused.
 pub(crate) async fn route_continue_session(
     store: &Arc<Store>,
     registry: &Arc<session_runner::SessionRegistry>,
@@ -2005,6 +2070,7 @@ pub(crate) async fn route_continue_session(
     message: &str,
     image_ids: &[String],
     waits: ContinueWaits,
+    stopped: StoppedSessionPolicy,
 ) -> Result<ContinueRoute, String> {
     // A handle whose connection went away under this request; the wait below
     // skips it so only a re-connect's handle (or a status change) ends it.
@@ -2086,6 +2152,11 @@ pub(crate) async fn route_continue_session(
             store::SessionStatus::Completed
             | store::SessionStatus::Cancelled
             | store::SessionStatus::Error => {
+                if stopped == StoppedSessionPolicy::Refuse {
+                    return Ok(ContinueRoute::Delivered(ContinueOutcome::not_running(
+                        format!("the session is {}", session.status.as_str()),
+                    )));
+                }
                 return Ok(match continue_resume_gate(store, registry, session)? {
                     Ok(claim) => ContinueRoute::Resume(Box::new(claim)),
                     Err(refused) => ContinueRoute::Delivered(refused),
@@ -2333,6 +2404,141 @@ 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.
+///
+/// `retryable` tells the composer whether a "no" is worth asking about
+/// again while the row stays `running`: it is for a connection still coming
+/// up (or re-connecting), and is not for a row another Staged instance owns
+/// or one that is not running, where only a status change can alter the
+/// answer. It is always `false` when `supported` is true.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SessionLiveDelivery {
+    pub supported: bool,
+    pub retryable: bool,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub reason: Option<String>,
+}
+
+impl SessionLiveDelivery {
+    const SUPPORTED: Self = Self {
+        supported: true,
+        retryable: false,
+        reason: None,
+    };
+
+    /// Queue only for this run: asking again changes nothing until the
+    /// status moves.
+    fn unsupported(reason: impl Into<String>) -> Self {
+        Self {
+            supported: false,
+            retryable: false,
+            reason: Some(reason.into()),
+        }
+    }
+
+    /// Queue only for now: the row is `running` and ours, so a later ask
+    /// may find the connection up.
+    fn not_yet(reason: impl Into<String>) -> Self {
+        Self {
+            supported: false,
+            retryable: true,
+            reason: Some(reason.into()),
+        }
+    }
+}
+
+pub(crate) fn live_delivery_for_session(
+    store: &Store,
+    registry: &session_runner::SessionRegistry,
+    session_id: &str,
+) -> Result<SessionLiveDelivery, String> {
+    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),
+        _ => Ok(SessionLiveDelivery::not_yet(
+            "the session's agent has not connected yet",
+        )),
+    }
+}
+
+#[tauri::command]
+pub fn get_session_live_delivery(
+    store: tauri::State<'_, Mutex<Option<Arc<Store>>>>,
+    registry: tauri::State<'_, Arc<session_runner::SessionRegistry>>,
+    session_id: String,
+) -> Result<SessionLiveDelivery, String> {
+    let store = get_store(&store)?;
+    live_delivery_for_session(&store, &registry, &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<String>,
+}
+
+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 running session now instead of queueing it:
+/// the composer's "Send" mode. Takes the live paths of the project MCP
+/// server's `continue_repo_session` tool — steer, else interrupt and
+/// re-prompt — and refuses a queued session the same way; unlike the tool it
+/// does not resume a stopped session (`not_running`), leaving that to the
+/// composer's `resume_session`, which honors the branch override and
+/// follow-up config. The reply says which.
+#[tauri::command]
+pub async fn send_session_message_now(
+    store: tauri::State<'_, Mutex<Option<Arc<Store>>>>,
+    registry: tauri::State<'_, Arc<session_runner::SessionRegistry>>,
+    session_id: String,
+    content: String,
+    image_ids: Option<Vec<String>>,
+) -> Result<SessionMessageDelivery, String> {
+    let store = get_store(&store)?;
+    let outcome = send_session_message_now_for_store(
+        &store,
+        &registry,
+        &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<Option<Arc<Store>>>>,
@@ -8191,6 +8397,27 @@ mod tests {
             message,
             image_ids,
             ContinueWaits::uniform(Duration::from_millis(300)),
+            StoppedSessionPolicy::Resume,
+        )
+        .await
+        .unwrap()
+    }
+
+    /// `send_session_message_now`'s body, with the test wait budget.
+    async fn send_now_through_shared_body(
+        registry: &Arc<session_runner::SessionRegistry>,
+        store: &Arc<Store>,
+        session_id: &str,
+        message: &str,
+    ) -> ContinueRoute {
+        route_continue_session(
+            store,
+            registry,
+            session_id,
+            message,
+            &[],
+            ContinueWaits::uniform(Duration::from_millis(300)),
+            StoppedSessionPolicy::Refuse,
         )
         .await
         .unwrap()
@@ -8244,6 +8471,216 @@ 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, &registry, &session.id).unwrap();
+
+        assert_eq!(
+            capability,
+            SessionLiveDelivery {
+                supported: true,
+                retryable: false,
+                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. Both answers are
+    /// retryable: the row is ours and running, so a (re-)connect can still
+    /// hand out a handle.
+    #[test]
+    fn live_delivery_is_unsupported_but_retryable_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, &registry, &session.id).unwrap();
+        assert!(!starting.supported);
+        assert!(starting.retryable, "{starting:?}");
+        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, &registry, &session.id).unwrap();
+        assert!(!dead.supported);
+        assert!(dead.retryable, "{dead:?}");
+    }
+
+    /// Stopped and queued rows are not live: the composer keeps its plain
+    /// button for them (a stopped session's "Send" resumes it through
+    /// `resume_session`; a queued one cannot take a message out of FIFO
+    /// order). Not retryable: only a status change can alter the answer.
+    #[test]
+    fn live_delivery_is_unsupported_and_not_retryable_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, &registry, &queued.id).unwrap();
+        assert!(!capability.supported);
+        assert!(!capability.retryable, "{capability:?}");
+        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, &registry, &completed.id).unwrap();
+        assert!(!capability.supported);
+        assert!(!capability.retryable, "{capability:?}");
+        assert_eq!(
+            capability.reason.as_deref(),
+            Some("the session is completed")
+        );
+
+        assert!(live_delivery_for_session(&store, &registry, "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
+    /// for the whole run — not retryable.
+    #[tokio::test(flavor = "current_thread")]
+    async fn live_delivery_is_unsupported_and_not_retryable_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, &registry, &session.id).unwrap();
+
+        assert!(!capability.supported);
+        assert!(!capability.retryable, "{capability:?}");
+        let reason = capability.reason.unwrap();
+        assert!(reason.contains("another Staged instance"), "{reason}");
+        assert!(reason.contains(&other_pid.to_string()), "{reason}");
+    }
+
+    /// `send_session_message_now` never resumes: a stopped session that the
+    /// MCP tool would claim comes back `not_running` with the row untouched,
+    /// so the composer can resume it through `resume_session` with its branch
+    /// override and follow-up config.
+    #[tokio::test(flavor = "current_thread")]
+    async fn send_now_refuses_a_stopped_session_instead_of_resuming_it() {
+        let (store, branch) = setup_branch_store_with_workdir();
+        let session = completed_commit_session_with_agent_id(&store, &branch.id);
+        let registry = test_registry();
+
+        let refused =
+            delivered(send_now_through_shared_body(&registry, &store, &session.id, "x").await);
+        assert_eq!(
+            refused,
+            ContinueOutcome::not_running("the session is completed")
+        );
+        assert_eq!(
+            store.get_session(&session.id).unwrap().unwrap().status,
+            store::SessionStatus::Completed
+        );
+
+        // The same session is still claimed under the tool's policy.
+        match continue_through_shared_body(&registry, &store, &session.id, "x").await {
+            ContinueRoute::Resume(claim) => assert_eq!(claim.session.id, session.id),
+            other => panic!("expected a resume, got {other:?}"),
+        }
+    }
+
+    /// The refusal also covers a session that stops while the request waits
+    /// for its connection: it is re-routed on the new status and refused
+    /// rather than claimed.
+    #[tokio::test(flavor = "current_thread")]
+    async fn send_now_refuses_a_session_that_stops_during_the_wait() {
+        let (store, branch) = setup_branch_store_with_workdir();
+        let session =
+            create_branch_commit_session(&store, &branch.id, store::SessionStatus::Running);
+        let registry = test_registry();
+        let _registration = registry.register_external(&session.id);
+        let settle = {
+            let store = Arc::clone(&store);
+            let session_id = session.id.clone();
+            tokio::spawn(async move {
+                tokio::time::sleep(Duration::from_millis(50)).await;
+                store
+                    .update_session_status(
+                        &session_id,
+                        store::SessionStatus::Cancelled,
+                        None,
+                        Some(&store::CompletionReason::Interrupted),
+                    )
+                    .unwrap();
+                store.set_agent_session_id(&session_id, "acp-1").unwrap();
+            })
+        };
+
+        let outcome =
+            delivered(send_now_through_shared_body(&registry, &store, &session.id, "x").await);
+
+        settle.await.unwrap();
+        assert_eq!(
+            outcome,
+            ContinueOutcome::not_running("the session is cancelled")
+        );
+        assert_eq!(
+            store.get_session(&session.id).unwrap().unwrap().status,
+            store::SessionStatus::Cancelled
+        );
+    }
+
+    /// 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);
+
+        let not_running =
+            SessionMessageDelivery::from(&ContinueOutcome::not_running("the session is completed"));
+        assert_eq!(not_running.delivery, "not_running");
+        assert_eq!(
+            not_running.reason.as_deref(),
+            Some("the session is completed")
+        );
+    }
+
     /// 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")]
@@ -8637,6 +9074,7 @@ mod tests {
                     "x",
                     &[],
                     ContinueWaits::uniform(Duration::from_millis(300)),
+                    StoppedSessionPolicy::Resume,
                 )
                 .await
                 .unwrap()
@@ -8692,6 +9130,7 @@ mod tests {
             "x",
             &[],
             ContinueWaits::uniform(Duration::from_secs(5)),
+            StoppedSessionPolicy::Resume,
         )
         .await
         .unwrap();
@@ -8729,6 +9168,7 @@ mod tests {
                 remote_startup: Duration::from_millis(500),
                 settle: Duration::from_millis(100),
             },
+            StoppedSessionPolicy::Resume,
         )
         .await
         .unwrap();
@@ -8863,6 +9303,9 @@ mod tests {
         let not_resumable = ContinueOutcome::not_resumable("gone");
         assert_eq!(not_resumable.delivery(), "not_resumable");
         assert_eq!(not_resumable.reason(), Some("gone"));
+        let not_running = ContinueOutcome::not_running("the session is completed");
+        assert_eq!(not_running.delivery(), "not_running");
+        assert_eq!(not_running.reason(), Some("the session is completed"));
         assert_eq!(ContinueOutcome::Injected.reason(), None);
     }
 
diff --git a/apps/staged/src-tauri/src/web_server.rs b/apps/staged/src-tauri/src/web_server.rs
index cc0e419a5..a9754506a 100644
--- a/apps/staged/src-tauri/src/web_server.rs
+++ b/apps/staged/src-tauri/src/web_server.rs
@@ -3679,6 +3679,36 @@ async fn dispatch(command: &str, args: Value, state: &WebAppState) -> Result<Val
             .await?;
             Ok(Value::Null)
         }
+        "send_session_message_now" => {
+            let store = get_store(store_mutex)?;
+            let session_id: String = arg(&args, "sessionId")?;
+            let content: String = arg(&args, "content")?;
+            let image_ids: Option<Vec<String>> = opt_arg(&args, "imageIds")?;
+            let session_registry = Arc::clone(session_registry);
+            // Detached: interrupt-and-reprompt is two agent round trips, and
+            // a client that disconnects between them must not leave the turn
+            // cancelled with the re-prompt never sent.
+            let delivery = detached(async move {
+                let outcome = session_commands::send_session_message_now_for_store(
+                    &store,
+                    &session_registry,
+                    &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..a0c9a513b 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,34 @@ 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<SessionLiveDelivery> {
+  return invokeCommand('get_session_live_delivery', { sessionId });
+}
+
+/**
+ * Deliver a message to the running session now instead of queueing it:
+ * steered into the running turn (or interrupting it) over the live
+ * connection, as the `continue_repo_session` MCP tool does. Unlike the tool
+ * it does not resume a stopped session: that comes back `not_running`, and
+ * the caller resumes through `resumeSession` so the branch override and
+ * follow-up config apply.
+ */
+export function sendSessionMessageNow(
+  sessionId: string,
+  content: string,
+  imageIds?: string[]
+): Promise<SessionMessageDelivery> {
+  return invokeCommand('send_session_message_now', {
+    sessionId,
+    content,
+    imageIds: imageIds ?? null,
+  });
+}
+
 export function deleteSession(sessionId: string): Promise<void> {
   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..1d9d5dc2a 100644
--- a/apps/staged/src/lib/features/sessions/ChatComposer.svelte
+++ b/apps/staged/src/lib/features/sessions/ChatComposer.svelte
@@ -6,6 +6,19 @@
   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. While the pane has no session row to submit to
+  (`sessionBlock`: still loading, or the load settled without one) the
+  control is disabled with a title saying which, but the draft can be typed.
+
+  Failures of the composer's own actions (`sendError`: a refused send, a
+  queue change that failed) are shown on a `ComposerError` line inside the
+  composer, next to the attachment error, rather than in the pane's
+  transcript slot — the transcript stays readable behind them.
 -->
 <script lang="ts" module>
   // min-h instead of h so the vertical model/effort label can grow the button.
@@ -25,10 +38,22 @@
   import type { Snippet } from 'svelte';
   import X from '@lucide/svelte/icons/x';
   import Paperclip from '@lucide/svelte/icons/paperclip';
+  import ChevronDown from '@lucide/svelte/icons/chevron-down';
   import Spinner from '../../shared/Spinner.svelte';
-  import { Button } from '$lib/components/ui/button';
+  import { Button, buttonVariants } from '$lib/components/ui/button';
+  import * as DropdownMenu from '$lib/components/ui/dropdown-menu';
+  import { cn } from '$lib/components/utils';
+  import {
+    COMPOSER_SEND_MODE_OPTIONS,
+    composerSendControl,
+    composerSubmitBlock,
+    composerSubmitBlockTitle,
+    type ComposerSendMode,
+    type ComposerSessionBlock,
+  } from './composerSendControl';
   import { ATTACHMENT_ACCEPT } from './attachErrors';
   import AttachmentThumb from './AttachmentThumb.svelte';
+  import ComposerError from './ComposerError.svelte';
   import type { AttachmentPreview } from './attachmentPreview';
   import HashtagInput from './HashtagInput.svelte';
   import type { HashtagItem } from '../../types';
@@ -54,7 +79,29 @@
     attachError?: string | null;
     /** Session is running — attach/remove disabled, send queues instead. */
     isLive?: boolean;
+    /**
+     * The running session can take a message now (backend-reported), so the
+     * submit control offers "Send" beside "Queue". Ignored unless `isLive`.
+     */
+    canSendLive?: boolean;
+    /** The remembered submit mode for live-capable sessions. */
+    sendMode?: ComposerSendMode;
+    /** The user picked a mode from the split button's menu. */
+    onSendModeChange?: (mode: ComposerSendMode) => void;
     sending?: boolean;
+    /**
+     * Why there is no session row to submit to — still loading, or the load
+     * settled without one — or `null` once the row is in. While set, submit
+     * is disabled with a title saying which (and Cmd+Enter is the parent's
+     * no-op) while the draft can still be typed.
+     */
+    sessionBlock?: ComposerSessionBlock | null;
+    /**
+     * Why the last composer action failed (send refused, queue change
+     * failed), shown inline until the parent clears it — on the next
+     * successful action or when the draft is edited.
+     */
+    sendError?: string | null;
     onSend?: () => void;
     /** Agent config picker rendered at the start of the control row. */
     configPicker?: Snippet;
@@ -74,13 +121,29 @@
     onRemoveImage,
     attachError = null,
     isLive = false,
+    canSendLive = false,
+    sendMode = 'queue',
+    onSendModeChange,
     sending = false,
+    sessionBlock = null,
+    sendError = null,
     onSend,
     configPicker,
   }: Props = $props();
 
   let fileInputEl = $state<HTMLInputElement>();
 
+  let sendControl = $derived(composerSendControl({ isLive, canSendLive, mode: sendMode }));
+  let submitBlock = $derived(
+    composerSubmitBlock({ sessionBlock, sending, hasText: value.trim().length > 0 })
+  );
+  let sendDisabled = $derived(submitBlock !== null);
+  let sendTitle = $derived(composerSubmitBlockTitle(submitBlock) ?? sendControl.title);
+
+  function handleSendModeChange(value: string) {
+    if (value === 'queue' || value === 'send') onSendModeChange?.(value);
+  }
+
   function openImagePicker() {
     fileInputEl?.click();
   }
@@ -115,9 +178,10 @@
       {/each}
     </div>
   {/if}
-  {#if canAttachImages && attachError}
-    <p class="attach-error" role="alert">{attachError}</p>
+  {#if canAttachImages}
+    <ComposerError message={attachError} />
   {/if}
+  <ComposerError message={sendError} />
   <HashtagInput
     bind:textareaEl
     bind:value
@@ -152,26 +216,66 @@
         </Button>
       </span>
     {/if}
-    <span class="inline-flex composer-send" title={isLive ? 'Queue message' : 'Send message'}>
-      <Button
-        variant="accent"
-        class="h-auto min-h-9 gap-1.5 px-4 py-1 text-sm max-[640px]:min-h-11"
-        onclick={onSend}
-        disabled={sending || !value.trim()}
-      >
-        {#if sending}
-          <Spinner size={14} />
-          Sending…
-        {:else if isLive}
-          Queue
-        {:else}
-          Send
-        {/if}
-      </Button>
-    </span>
+    {#if sendControl.split}
+      <span class="inline-flex composer-send composer-send-split" title={sendTitle}>
+        <Button
+          variant="accent"
+          class="composer-send-primary h-auto min-h-9 gap-1.5 rounded-r-none px-4 py-1 text-sm max-[640px]:min-h-11"
+          onclick={onSend}
+          disabled={sendDisabled}
+        >
+          {@render sendLabel()}
+        </Button>
+        <DropdownMenu.Root>
+          <DropdownMenu.Trigger
+            class={cn(
+              buttonVariants({ variant: 'accent' }),
+              'composer-send-menu-trigger -ml-px h-auto min-h-9 w-8 rounded-l-none px-0 max-[640px]:min-h-11'
+            )}
+            title="Change how messages are delivered"
+            aria-label="Change how messages are delivered"
+            disabled={sending || sessionBlock !== null}
+          >
+            <ChevronDown size={14} />
+          </DropdownMenu.Trigger>
+          <DropdownMenu.Content align="end" side="top" sideOffset={4} class="min-w-[220px]">
+            <DropdownMenu.RadioGroup value={sendMode} onValueChange={handleSendModeChange}>
+              {#each COMPOSER_SEND_MODE_OPTIONS as option (option.mode)}
+                <DropdownMenu.RadioItem value={option.mode}>
+                  <span class="flex min-w-0 flex-col">
+                    <span>{option.label}</span>
+                    <span class="truncate text-xs text-muted-foreground">{option.description}</span>
+                  </span>
+                </DropdownMenu.RadioItem>
+              {/each}
+            </DropdownMenu.RadioGroup>
+          </DropdownMenu.Content>
+        </DropdownMenu.Root>
+      </span>
+    {:else}
+      <span class="inline-flex composer-send" title={sendTitle}>
+        <Button
+          variant="accent"
+          class="h-auto min-h-9 gap-1.5 px-4 py-1 text-sm max-[640px]:min-h-11"
+          onclick={onSend}
+          disabled={sendDisabled}
+        >
+          {@render sendLabel()}
+        </Button>
+      </span>
+    {/if}
   </div>
 </div>
 
+{#snippet sendLabel()}
+  {#if sending}
+    <Spinner size={14} />
+    Sending…
+  {:else}
+    {sendControl.label}
+  {/if}
+{/snippet}
+
 <style>
   .chat-composer {
     display: flex;
@@ -196,12 +300,6 @@
     flex-wrap: wrap;
   }
 
-  .attach-error {
-    margin: 0;
-    font-size: var(--size-sm);
-    color: var(--ui-danger);
-  }
-
   .reply-image-thumb {
     position: relative;
     width: 48px;
@@ -259,4 +357,32 @@
   .composer-controls .composer-send {
     margin-left: auto;
   }
+
+  /* A hairline between the split button's two accent segments, exactly the
+     height of the fills. Every button paints a 1px transparent border with
+     bg-clip-padding, so its fill is inset 1px from the border box on all
+     sides. The trigger's -ml-px overlaps the primary's transparent right
+     border column; with no left border of its own the trigger's fill paints
+     over that column and the two fills meet with no seam. The divider is then
+     drawn on top as a 1px ::before spanning the trigger's padding box (the
+     containing block of an absolutely positioned child), which is the fill's
+     extent, so it neither overshoots the fills nor notches them. A tinted
+     border-left would instead span the border box and mitre into the
+     transparent top and bottom borders, poking 1px past the fills at each
+     end. The colour mixes --bg-primary into the accent so the line is an
+     opaque tint of the fill rather than a wash over the chrome behind it. */
+  .composer-send-split :global(.composer-send-menu-trigger) {
+    position: relative;
+    border-left-width: 0;
+  }
+
+  .composer-send-split :global(.composer-send-menu-trigger::before) {
+    content: '';
+    position: absolute;
+    top: 0;
+    bottom: 0;
+    left: 0;
+    width: 1px;
+    background: color-mix(in srgb, var(--bg-primary) 35%, var(--ui-accent));
+  }
 </style>
diff --git a/apps/staged/src/lib/features/sessions/ComposerError.svelte b/apps/staged/src/lib/features/sessions/ComposerError.svelte
new file mode 100644
index 000000000..1c0ae038b
--- /dev/null
+++ b/apps/staged/src/lib/features/sessions/ComposerError.svelte
@@ -0,0 +1,33 @@
+<!--
+  ComposerError.svelte — One failure line for the chat composer area.
+
+  Shows why the last composer action failed (a refused send, a queue change,
+  an attachment the pane rejected) in the composer's own danger style, as a
+  live region so it is announced when it appears. Rendered by ChatComposer
+  for the attachment and send errors, and by SessionChatPane in the
+  pipeline-stop area that replaces the composer while a workflow runs with
+  no transcript yet, so a failed Stop is visible there too.
+
+  Renders nothing when `message` is null. The parent clears the message when
+  the next action starts, so a repeated identical failure remounts the line
+  and is announced again.
+-->
+<script lang="ts">
+  interface Props {
+    message: string | null;
+  }
+
+  let { message }: Props = $props();
+</script>
+
+{#if message}
+  <p class="composer-error" role="alert">{message}</p>
+{/if}
+
+<style>
+  .composer-error {
+    margin: 0;
+    font-size: var(--size-sm);
+    color: var(--ui-danger);
+  }
+</style>
diff --git a/apps/staged/src/lib/features/sessions/SessionChatPane.svelte b/apps/staged/src/lib/features/sessions/SessionChatPane.svelte
index acae36cc2..ae8cee664 100644
--- a/apps/staged/src/lib/features/sessions/SessionChatPane.svelte
+++ b/apps/staged/src/lib/features/sessions/SessionChatPane.svelte
@@ -13,6 +13,11 @@
     a labeled Stop button; once messages exist, Stop lives on the Thinking row
     and the composer queues follow-ups as usual
   - Message queue: sending while running persists a follow-up for backend drain
+  - Live send: when the backend reports the running session can take a message
+    now (`get_session_live_delivery`), the composer offers "Send" beside
+    "Queue"; "Send" goes through `send_session_message_now`, the live path the
+    `continue_repo_session` MCP tool uses (steer, else interrupt-and-reprompt);
+    a session that stopped meanwhile falls back to the idle resume path
   - Copy button on every message
   - Tool calls show name + args preview; expand to see output
   - Fixed-size modal with proper scrolling
@@ -49,6 +54,7 @@
     Session,
     SessionBackgroundHoldPayload,
     SessionMessage,
+    SessionStatus,
     QueuedSessionMessage,
     HashtagItem,
     ProjectRepo,
@@ -65,6 +71,7 @@
     getSession,
     getSessionAcpMetadataMessages,
     getSessionBackgroundHold,
+    getSessionLiveDelivery,
     getLatestSessionMessages,
     getOlderSessionMessages,
     handleExternalLinkClick,
@@ -74,6 +81,7 @@
     queueSessionMessage,
     resumeSession,
     sendQueuedSessionMessage,
+    sendSessionMessageNow,
     stopSessionAsyncTask,
     type AcpConfigDiscovery,
     type AcpConfigSelector,
@@ -90,8 +98,20 @@
     defaultSelectorValue,
     selectorHasValue,
   } from '../agents/acpConfigSelection';
-  import { setAcpConfigPref } from '../settings/preferences.svelte';
+  import {
+    preferences,
+    setAcpConfigPref,
+    setComposerSendMode,
+  } from '../settings/preferences.svelte';
   import ChatComposer, { composerControlClass } from './ChatComposer.svelte';
+  import ComposerError from './ComposerError.svelte';
+  import { composerSendControl, composerSessionBlock } from './composerSendControl';
+  import {
+    LIVE_DELIVERY_UNKNOWN,
+    liveDeliveryKnowledge,
+    shouldAskLiveDelivery,
+    type LiveDeliveryKnowledge,
+  } from './liveDelivery';
   import {
     backgroundHoldTaskRows,
     isTaskHeld,
@@ -270,6 +290,16 @@
   let closed = false;
   /** Latest background hold reported for this session; null when not holding. */
   let backgroundHold = $state<SessionBackgroundHold | null>(null);
+  /**
+   * What the backend has said about this running session taking a message
+   * now (owned by this process, agent connection up). Asked on each poll tick
+   * until the answer is settled: a yes, or a no that cannot change during this
+   * run (another Staged instance owns the row). A retryable no — the
+   * connection comes up a few seconds into a run — is asked again. Forgotten
+   * whenever the status leaves `running` or the pane changes session.
+   */
+  let liveDelivery = $state<LiveDeliveryKnowledge>(LIVE_DELIVERY_UNKNOWN);
+  let liveDeliveryInFlight = false;
   /**
    * Counts hold reports received by event, so the mount-time snapshot can tell
    * whether a newer report landed while it was in flight.
@@ -327,6 +357,10 @@
   let followupDiscoveryRun = 0;
 
   let isLive = $derived(session?.status === 'running');
+  let canSendLive = $derived(isLive && liveDelivery.supported);
+  let sendControl = $derived(
+    composerSendControl({ isLive, canSendLive, mode: preferences.composerSendMode })
+  );
   // Holding is a sub-state of running, so the session stays live and only the
   // activity row changes: the wait plus its live task count, and a Stop that
   // says what it stops.
@@ -569,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);
@@ -626,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);
     });
@@ -642,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 {
@@ -783,6 +822,7 @@
       } else {
         // A terminal status ends any wait: the agent has been torn down.
         clearBackgroundHold();
+        liveDelivery = LIVE_DELIVERY_UNKNOWN;
         finishPolling();
       }
     });
@@ -802,6 +842,7 @@
     if (!isActive || !id) return;
 
     backgroundHold = null;
+    liveDelivery = LIVE_DELIVERY_UNKNOWN;
     stoppingTaskIds = new Set();
     taskStopNotices = new Map();
     const unlisten = listenToEvent<SessionBackgroundHoldPayload>(
@@ -915,6 +956,8 @@
       hasOlderMessages = false;
       acpMetadataMessages = [];
       queuedMessages = [];
+      // A failure line is about the session that was shown when it happened.
+      sendError = null;
     }
     loading = true;
     error = null;
@@ -1093,6 +1136,7 @@
       if (JSON.stringify(result.queuedMessages) !== JSON.stringify(queuedMessages)) {
         queuedMessages = result.queuedMessages;
       }
+      refreshLiveDelivery(s?.status ?? session?.status);
 
       applyAcpMetadataUpdates(result.acpMetadataMessages);
       const merged = mergeSinceMessages(messages, result.messages, request.sinceId);
@@ -1146,6 +1190,35 @@
     acpMetadataMessages = [...byId.values()].sort((a, b) => a.id - b.id);
   }
 
+  /**
+   * Learn whether the running session can take a message now. One cheap
+   * registry lookup per poll tick while the answer is a retryable no; nothing
+   * once it is settled (see `liveDelivery`). A status other than `running`
+   * forgets the answer at once, so the next run asks afresh.
+   */
+  function refreshLiveDelivery(status: SessionStatus | undefined) {
+    if (status !== 'running') {
+      liveDelivery = LIVE_DELIVERY_UNKNOWN;
+      return;
+    }
+    if (!shouldAskLiveDelivery(status, liveDelivery) || liveDeliveryInFlight || !sessionId) {
+      return;
+    }
+    const id = sessionId;
+    liveDeliveryInFlight = true;
+    getSessionLiveDelivery(id)
+      .then((capability) => {
+        if (closed || sessionId !== id || session?.status !== 'running') return;
+        liveDelivery = liveDeliveryKnowledge(capability);
+      })
+      .catch(() => {
+        // Best-effort: the next tick asks again.
+      })
+      .finally(() => {
+        liveDeliveryInFlight = false;
+      });
+  }
+
   async function refreshQueuedMessages() {
     if (!sessionId || closed) return;
     try {
@@ -1205,8 +1278,46 @@
 
   let sending = $state(false);
   let noteFollowupSending = $state(false);
+  /**
+   * Why the last composer action failed (a refused or failed send, a queue
+   * change or cancel that failed). Shown on a line inside the composer, not in
+   * `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 (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<string | null>(null);
+
+  /**
+   * A composer action is starting: drop the previous action's failure line,
+   * and a stale load error with it — the action's outcome is what the pane
+   * should show now, and the poll a successful one starts refreshes the
+   * transcript anyway.
+   */
+  function beginComposerAction() {
+    sendError = null;
+    error = null;
+  }
+
+  /**
+   * 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() {
+    // 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
+    // typed since, so the guard comes before anything is taken out of the box.
+    if (sending) return;
     const text = inputText.trim();
     if (!text) return;
     inputText = '';
@@ -1217,39 +1328,148 @@
     // Reset textarea height after clearing (oninput won't fire for programmatic changes)
     tick().then(() => autoResize());
 
+    if (sendControl.action === 'send-now') {
+      await sendMessageNow(text, imageIdsToSend);
+      return;
+    }
+
     if (isLive) {
+      beginComposerAction();
       try {
         await queueSessionMessage(sessionId, text, imageIdsToSend, branchId ?? null);
         await refreshQueuedMessages();
       } catch (e) {
-        error = `Failed to queue: ${e instanceof Error ? e.message : String(e)}`;
+        sendError = `Failed to queue: ${e instanceof Error ? e.message : String(e)}`;
         inputText = text;
         replyImageIds = imageIdsToSend ?? [];
       }
       return;
     }
 
-    // Session is idle — send immediately
-    await sendMessage(text, imageIdsToSend);
+    // Session is idle — send immediately. `sendMessage` reports the failure
+    // itself; the draft goes back in the box so it can be retried.
+    if (!(await sendMessage(text, imageIdsToSend))) {
+      inputText = text;
+      replyImageIds = imageIdsToSend ?? [];
+    }
+  }
+
+  /**
+   * Deliver a message to the running session now — the composer's "Send"
+   * mode. The backend steers it into the turn in flight, or interrupts the
+   * turn and re-prompts when the agent cannot steer. It never resumes: a
+   * session that stopped between the last poll and the click comes back
+   * `not_running`, and the message goes through `sendMessage` instead — the
+   * idle path, which applies this pane's branch override and follow-up
+   * config (the note chat pane's branch would otherwise be lost). Anything
+   * not delivered (`blocked`, `rejected_queued`, `not_resumable`, or a kind
+   * this pane does not know) is reported and the draft restored — never
+   * re-queued behind the user's back, since a blocked steer may already have
+   * landed.
+   */
+  async function sendMessageNow(text: string, imageIds?: string[]) {
+    if (!session || sending) return;
+    const id = session.id;
+    sending = true;
+    beginComposerAction();
+    try {
+      const result = await sendSessionMessageNow(id, text, imageIds);
+      if (session?.id !== id) return;
+      switch (result.delivery) {
+        case 'injected':
+        case 'interrupted_and_reprompted':
+        case 'started_new_turn':
+          // The user row is persisted by the delivery itself; the poll picks
+          // it up.
+          startPolling();
+          scrollToBottom();
+          return;
+        case 'resumed':
+          // The command does not resume (see `not_running`); the kind belongs
+          // to the shared MCP spelling. Should it ever arrive, the row is
+          // running again on a fresh connection.
+          session = { ...session, status: 'running' };
+          clearBackgroundHold();
+          liveDelivery = LIVE_DELIVERY_UNKNOWN;
+          startPolling();
+          scrollToBottom();
+          return;
+        case 'not_running':
+          // Stopped since the last poll: resume it the way the idle composer
+          // does, with the branch override and follow-up config. `sending` is
+          // already held here, so the lock is passed through. A resume that
+          // fails (worktree gone, or the queue drain re-ran the session first
+          // so the row is already running) has set `sendError`; fall through
+          // to the draft restore below without overwriting it.
+          liveDelivery = LIVE_DELIVERY_UNKNOWN;
+          if (await sendMessage(text, imageIds, true)) return;
+          break;
+        case 'rejected_queued':
+          sendError = 'The session has not started yet; queue the message instead.';
+          break;
+        case 'blocked':
+          // The connection may have gone under the request: ask again before
+          // offering "Send" next time.
+          liveDelivery = LIVE_DELIVERY_UNKNOWN;
+          sendError = `Not delivered: ${result.reason ?? result.delivery}`;
+          break;
+        case 'not_resumable':
+          // Defensive: the command refuses a stopped session before the
+          // resume gate (`not_running`), so this kind only reaches the MCP
+          // tool. Kept because the shared union still spells it.
+          sendError = `Not delivered: ${result.reason ?? result.delivery}`;
+          break;
+        default: {
+          // A delivery kind the backend added that this hand-maintained union
+          // does not know: say so rather than restoring the draft in silence.
+          const unknownKind: never = result.delivery;
+          sendError = `Not delivered: ${String(unknownKind)}`;
+          break;
+        }
+      }
+      inputText = text;
+      replyImageIds = imageIds ?? [];
+    } catch (e) {
+      sendError = `Failed to send: ${e instanceof Error ? e.message : String(e)}`;
+      inputText = text;
+      replyImageIds = imageIds ?? [];
+    } finally {
+      sending = false;
+    }
   }
 
-  /** Actually send a message to the backend and start the agent. */
+  /**
+   * Actually send a message to the backend and start the agent. Resolves
+   * `true` once the backend accepted the message (even if the pane has moved
+   * to another session since), `false` when nothing was sent, and the caller
+   * decides whether to restore its draft. `sendError` says why on `false` in
+   * every case but two: when nothing could be targeted (defensive — the
+   * composer is disabled until the row has loaded, and a targeted send checks
+   * its target before calling), and when another send already holds
+   * `sending`, since the spinner already explains the wait.
+   */
   async function sendMessage(
     text: string,
     imageIds?: string[],
     sendingLocked = false,
     target: SendMessageTarget | null = null
-  ) {
+  ): Promise<boolean> {
     const targetSessionId = target?.sessionId ?? session?.id;
     const targetBranchId = target ? target.branchId : (branchId ?? null);
-    if (!targetSessionId || session?.id !== targetSessionId || (!sendingLocked && sending)) return;
+    if (!targetSessionId || session?.id !== targetSessionId) {
+      // The row is not loaded, or the pane moved on before a targeted send
+      // ran. Nothing is said: a message here would be about a session this
+      // pane no longer shows.
+      return false;
+    }
+    if (!sendingLocked && sending) return false;
     if (!sendingLocked) sending = true;
-    error = null;
+    beginComposerAction();
     const acpConfigSelection: AcpConfigSelection | undefined =
       followupAcpConfigSelection ?? undefined;
     try {
       await resumeSession(targetSessionId, text, imageIds, targetBranchId, acpConfigSelection);
-      if (session?.id !== targetSessionId) return;
+      if (session?.id !== targetSessionId) return true;
       // Backend sets status to running and emits an event.
       // Force an immediate poll to pick up the new user message + status.
       session = {
@@ -1259,15 +1479,19 @@
       };
       // A new turn is starting: any hold reported belonged to the previous
       // one, so drop it rather than letting this local flip back to `running`
-      // re-render the old wait in place of "Thinking…".
+      // re-render the old wait in place of "Thinking…". Likewise whatever was
+      // known about live delivery was about the previous run.
       clearBackgroundHold();
+      liveDelivery = LIVE_DELIVERY_UNKNOWN;
       modelSpecificFollowupConfig = null;
       modelSpecificFollowupModelValue = null;
       modelSpecificFollowupSessionId = null;
       startPolling();
       scrollToBottom();
+      return true;
     } catch (e) {
-      error = `Failed to send: ${e instanceof Error ? e.message : String(e)}`;
+      sendError = `Failed to send: ${e instanceof Error ? e.message : String(e)}`;
+      return false;
     } finally {
       if (!sendingLocked) sending = false;
     }
@@ -1279,7 +1503,7 @@
     const hasParsedNote = noteInfo.hasParsedNote;
     sending = true;
     noteFollowupSending = true;
-    error = null;
+    beginComposerAction();
     try {
       const prompt = await buildNoteFollowupMessage(
         target.sessionId,
@@ -1289,7 +1513,7 @@
       if (session?.id !== target.sessionId || (branchId ?? null) !== target.branchId) return;
       await sendMessage(prompt, undefined, true, target);
     } catch (e) {
-      error = `Failed to send: ${e instanceof Error ? e.message : String(e)}`;
+      sendError = `Failed to send: ${e instanceof Error ? e.message : String(e)}`;
     } finally {
       noteFollowupSending = false;
       sending = false;
@@ -1299,11 +1523,12 @@
   async function handleDeleteQueuedMessage(id: string) {
     if (queueActionIds.has(id)) return;
     queueActionIds = new Set(queueActionIds).add(id);
+    beginComposerAction();
     try {
       await deleteQueuedSessionMessage(id);
       await refreshQueuedMessages();
     } catch (e) {
-      error = `Failed to remove queued message: ${e instanceof Error ? e.message : String(e)}`;
+      sendError = `Failed to remove queued message: ${e instanceof Error ? e.message : String(e)}`;
     } finally {
       const next = new Set(queueActionIds);
       next.delete(id);
@@ -1314,7 +1539,7 @@
   async function handleSendQueuedMessage(id: string) {
     if (isLive || sending || queueActionIds.has(id)) return;
     queueActionIds = new Set(queueActionIds).add(id);
-    error = null;
+    beginComposerAction();
     try {
       await sendQueuedSessionMessage(id);
       await refreshQueuedMessages();
@@ -1324,7 +1549,7 @@
       startPolling();
       scrollToBottom();
     } catch (e) {
-      error = `Failed to send queued message: ${e instanceof Error ? e.message : String(e)}`;
+      sendError = `Failed to send queued message: ${e instanceof Error ? e.message : String(e)}`;
       await refreshQueuedMessages();
     } finally {
       const next = new Set(queueActionIds);
@@ -1336,10 +1561,11 @@
   async function handleCancel() {
     if (!session || cancelling) return;
     cancelling = true;
+    beginComposerAction();
     try {
       await cancelSession(session.id);
     } catch (e) {
-      error = `Failed to cancel: ${e instanceof Error ? e.message : String(e)}`;
+      sendError = `Failed to cancel: ${e instanceof Error ? e.message : String(e)}`;
     } finally {
       cancelling = false;
     }
@@ -1422,6 +1648,12 @@
     }
   }
 
+  /** The draft changed: the last failure line has been read, so drop it. */
+  function handleInput() {
+    sendError = null;
+    autoResize();
+  }
+
   function autoResize() {
     if (!inputEl) return;
     inputEl.style.height = 'auto';
@@ -2755,8 +2987,11 @@
   <!-- Input area with queued messages and image previews -->
   <div class="input-wrapper">
     {#if isPipelineRunning && grouped.length === 0}
+      <!-- Stands in for the composer, so it shows the composer's failure
+           line too: a failed Stop has nowhere else to appear. -->
       <div class="pipeline-stop-area">
-        <span class="inline-flex" title="Stop workflow">
+        <ComposerError message={sendError} />
+        <span class="inline-flex pipeline-stop-button" title="Stop workflow">
           <Button
             variant="destructive"
             size="sm"
@@ -2844,10 +3079,10 @@
       <ChatComposer
         bind:textareaEl={inputEl}
         bind:value={inputText}
-        placeholder={isLive ? 'Type to queue a follow-up…' : 'Send a message…'}
+        placeholder={sendControl.placeholder}
         {hashtagItems}
         onkeydown={handleInputKeydown}
-        oninput={autoResize}
+        oninput={handleInput}
         {canAttachImages}
         attachedImageIds={replyImageIds}
         {imagePreviews}
@@ -2855,7 +3090,12 @@
         onRemoveImage={removeReplyImage}
         {attachError}
         {isLive}
+        {canSendLive}
+        sendMode={preferences.composerSendMode}
+        onSendModeChange={setComposerSendMode}
         {sending}
+        sessionBlock={composerBlock}
+        {sendError}
         onSend={handleSend}
       >
         {#snippet configPicker()}
@@ -3460,15 +3700,22 @@
 
   /* ----- Pipeline stop area ---------------------------------------------- */
 
+  /* Same chrome as .chat-composer: the error line (if any) above, the
+     button right-aligned beneath it. */
   .pipeline-stop-area {
     display: flex;
-    justify-content: flex-end;
+    flex-direction: column;
+    gap: 8px;
     padding: 10px 14px;
     border-top: 1px solid var(--border-subtle);
     background: var(--bg-chrome);
     flex-shrink: 0;
   }
 
+  .pipeline-stop-button {
+    align-self: flex-end;
+  }
+
   /* ======================================================================= */
   /* Markdown content (assistant messages)                                   */
   /* ======================================================================= */
diff --git a/apps/staged/src/lib/features/sessions/composerSendControl.test.ts b/apps/staged/src/lib/features/sessions/composerSendControl.test.ts
new file mode 100644
index 000000000..fe12e810c
--- /dev/null
+++ b/apps/staged/src/lib/features/sessions/composerSendControl.test.ts
@@ -0,0 +1,127 @@
+import { describe, expect, it } from 'vitest';
+import {
+  COMPOSER_LOADING_TITLE,
+  COMPOSER_SEND_MODE_OPTIONS,
+  COMPOSER_UNAVAILABLE_TITLE,
+  composerSendControl,
+  composerSessionBlock,
+  composerSubmitBlock,
+  composerSubmitBlockTitle,
+  normalizeComposerSendMode,
+} from './composerSendControl';
+
+describe('composerSendControl', () => {
+  it('shows a single Send button that starts a turn when the session is not running', () => {
+    for (const mode of ['queue', 'send'] as const) {
+      const control = composerSendControl({ isLive: false, canSendLive: true, mode });
+      expect(control).toMatchObject({ split: false, action: 'start', label: 'Send' });
+    }
+    expect(composerSendControl({ isLive: false, canSendLive: false, mode: 'queue' })).toMatchObject(
+      { split: false, action: 'start', label: 'Send' }
+    );
+  });
+
+  it('shows a single Queue button when the running session cannot take a message now', () => {
+    for (const mode of ['queue', 'send'] as const) {
+      const control = composerSendControl({ isLive: true, canSendLive: false, mode });
+      expect(control).toMatchObject({ split: false, action: 'queue', label: 'Queue' });
+    }
+  });
+
+  it('splits the button when live delivery is supported, with the primary following the mode', () => {
+    expect(composerSendControl({ isLive: true, canSendLive: true, mode: 'queue' })).toMatchObject({
+      split: true,
+      action: 'queue',
+      label: 'Queue',
+    });
+    expect(composerSendControl({ isLive: true, canSendLive: true, mode: 'send' })).toMatchObject({
+      split: true,
+      action: 'send-now',
+      label: 'Send',
+    });
+  });
+
+  it('gives each state a distinct placeholder', () => {
+    const start = composerSendControl({ isLive: false, canSendLive: false, mode: 'queue' });
+    const queue = composerSendControl({ isLive: true, canSendLive: true, mode: 'queue' });
+    const sendNow = composerSendControl({ isLive: true, canSendLive: true, mode: 'send' });
+    expect(new Set([start.placeholder, queue.placeholder, sendNow.placeholder]).size).toBe(3);
+  });
+
+  it('normalizes stored modes, defaulting to queue', () => {
+    expect(normalizeComposerSendMode('send')).toBe('send');
+    expect(normalizeComposerSendMode('queue')).toBe('queue');
+    expect(normalizeComposerSendMode(undefined)).toBe('queue');
+    expect(normalizeComposerSendMode('later')).toBe('queue');
+    expect(normalizeComposerSendMode({ mode: 'send' })).toBe('queue');
+  });
+
+  it('offers exactly the two modes in the dropdown', () => {
+    expect(COMPOSER_SEND_MODE_OPTIONS.map((o) => o.mode)).toEqual(['queue', 'send']);
+  });
+});
+
+describe('composerSessionBlock', () => {
+  it('reports loading while the fetch is in flight, with or without a stale row', () => {
+    expect(composerSessionBlock({ loading: true, hasSession: false })).toBe('loading');
+    expect(composerSessionBlock({ loading: true, hasSession: true })).toBe('loading');
+  });
+
+  it('reports unavailable once the fetch settled without a row', () => {
+    expect(composerSessionBlock({ loading: false, hasSession: false })).toBe('unavailable');
+  });
+
+  it('reports no block once the row is in', () => {
+    expect(composerSessionBlock({ loading: false, hasSession: true })).toBeNull();
+  });
+});
+
+describe('composerSubmitBlock', () => {
+  it('disables submit while the session row is loading, whatever the draft holds', () => {
+    const sessionBlock = 'loading';
+    expect(composerSubmitBlock({ sessionBlock, sending: false, hasText: true })).toBe('loading');
+    expect(composerSubmitBlock({ sessionBlock, sending: false, hasText: false })).toBe('loading');
+    // Loading explains the block better than a spinner that is not spinning.
+    expect(composerSubmitBlock({ sessionBlock, sending: true, hasText: true })).toBe('loading');
+  });
+
+  it('disables submit once the load settled without a row', () => {
+    const sessionBlock = 'unavailable';
+    expect(composerSubmitBlock({ sessionBlock, sending: false, hasText: true })).toBe(
+      'unavailable'
+    );
+    expect(composerSubmitBlock({ sessionBlock, sending: true, hasText: false })).toBe(
+      'unavailable'
+    );
+  });
+
+  it('disables submit while a send is in flight', () => {
+    expect(composerSubmitBlock({ sessionBlock: null, sending: true, hasText: true })).toBe(
+      'sending'
+    );
+    expect(composerSubmitBlock({ sessionBlock: null, sending: true, hasText: false })).toBe(
+      'sending'
+    );
+  });
+
+  it('disables submit on an empty draft and otherwise allows it', () => {
+    expect(composerSubmitBlock({ sessionBlock: null, sending: false, hasText: false })).toBe(
+      'empty'
+    );
+    expect(composerSubmitBlock({ sessionBlock: null, sending: false, hasText: true })).toBeNull();
+  });
+});
+
+describe('composerSubmitBlockTitle', () => {
+  it('titles the session blocks differently so a settled failure reads as such', () => {
+    expect(composerSubmitBlockTitle('loading')).toBe(COMPOSER_LOADING_TITLE);
+    expect(composerSubmitBlockTitle('unavailable')).toBe(COMPOSER_UNAVAILABLE_TITLE);
+    expect(COMPOSER_LOADING_TITLE).not.toBe(COMPOSER_UNAVAILABLE_TITLE);
+  });
+
+  it('leaves the control title alone for blocks the control already explains', () => {
+    expect(composerSubmitBlockTitle('sending')).toBeNull();
+    expect(composerSubmitBlockTitle('empty')).toBeNull();
+    expect(composerSubmitBlockTitle(null)).toBeNull();
+  });
+});
diff --git a/apps/staged/src/lib/features/sessions/composerSendControl.ts b/apps/staged/src/lib/features/sessions/composerSendControl.ts
new file mode 100644
index 000000000..6473a4a81
--- /dev/null
+++ b/apps/staged/src/lib/features/sessions/composerSendControl.ts
@@ -0,0 +1,178 @@
+/**
+ * composerSendControl.ts — What the chat composer's submit control shows and
+ * does, derived from the session's state and the user's chosen mode.
+ *
+ * A session that is not running has one button: "Send", which starts (or
+ * resumes) a turn. A running session queues follow-ups for after the turn —
+ * unless the backend reports the session can take a message now (see
+ * `SessionLiveDelivery`), in which case the button splits: the primary
+ * segment performs the remembered mode (queue or live send) and a dropdown
+ * switches between the two. The mode is a global preference
+ * (`preferences.composerSendMode`), remembered across sessions and repos.
+ */
+
+/** How the composer submits a message while the session is running. */
+export type ComposerSendMode = 'queue' | 'send';
+
+export const DEFAULT_COMPOSER_SEND_MODE: ComposerSendMode = 'queue';
+
+/** A stored mode, or the default when the value is missing or malformed. */
+export function normalizeComposerSendMode(value: unknown): ComposerSendMode {
+  return value === 'queue' || value === 'send' ? value : DEFAULT_COMPOSER_SEND_MODE;
+}
+
+/** What pressing the primary segment does. */
+export type ComposerSendAction =
+  /** Start or resume a turn on a session that is not running. */
+  | 'start'
+  /** Persist a follow-up for after the running turn ends. */
+  | 'queue'
+  /** Deliver to the running session now (steer / interrupt-and-reprompt). */
+  | 'send-now';
+
+export interface ComposerSendControl {
+  /** Whether the mode dropdown is rendered beside the primary segment. */
+  split: boolean;
+  action: ComposerSendAction;
+  /** Primary segment label. */
+  label: string;
+  /** Hover title for the primary segment. */
+  title: string;
+  /** Placeholder for the empty input. */
+  placeholder: string;
+}
+
+export interface ComposerSendControlInput {
+  /** The session is running. */
+  isLive: boolean;
+  /** The backend reports the running session can take a message now. */
+  canSendLive: boolean;
+  /** The remembered mode (only consulted when both of the above hold). */
+  mode: ComposerSendMode;
+}
+
+export function composerSendControl({
+  isLive,
+  canSendLive,
+  mode,
+}: ComposerSendControlInput): ComposerSendControl {
+  if (!isLive) {
+    return {
+      split: false,
+      action: 'start',
+      label: 'Send',
+      title: 'Send message',
+      placeholder: 'Send a message…',
+    };
+  }
+  if (!canSendLive || mode === 'queue') {
+    return {
+      split: canSendLive,
+      action: 'queue',
+      label: 'Queue',
+      title: 'Queue message for after this turn',
+      placeholder: 'Type to queue a follow-up…',
+    };
+  }
+  return {
+    split: true,
+    action: 'send-now',
+    label: 'Send',
+    title: 'Send to the running agent now',
+    placeholder: 'Send a message to the running agent…',
+  };
+}
+
+/**
+ * Why the pane has no session row to submit to, or `null` once it has one.
+ * The draft can be typed in either state.
+ */
+export type ComposerSessionBlock =
+  /** The row is still being fetched. */
+  | 'loading'
+  /** The fetch settled without a row: it failed, or the session is gone. */
+  | 'unavailable';
+
+export interface ComposerSessionBlockInput {
+  /** The pane's load is in flight. */
+  loading: boolean;
+  /** The pane holds a session row. */
+  hasSession: boolean;
+}
+
+/**
+ * The pane's session state as the composer sees it. Loading wins over a
+ * missing row because a reopen keeps `hasSession` false until the fetch
+ * settles, and only a settled fetch with no row means the session is gone.
+ */
+export function composerSessionBlock({
+  loading,
+  hasSession,
+}: ComposerSessionBlockInput): ComposerSessionBlock | null {
+  if (loading) return 'loading';
+  if (!hasSession) return 'unavailable';
+  return null;
+}
+
+/** Why the submit control is disabled, or `null` when it can be pressed. */
+export type ComposerSubmitBlock =
+  | ComposerSessionBlock
+  /** A send is in flight and owns the composer until it settles. */
+  | 'sending'
+  /** Nothing to send. */
+  | 'empty';
+
+export interface ComposerSubmitBlockInput {
+  sessionBlock: ComposerSessionBlock | null;
+  sending: boolean;
+  /** The draft has non-whitespace text. */
+  hasText: boolean;
+}
+
+/**
+ * Whether the composer's submit control is disabled, and why. A session
+ * block wins because it is the only kind the control does not already
+ * explain (the spinner covers `sending`, the empty box covers `empty`), so
+ * it gets a hover title of its own: see `composerSubmitBlockTitle`.
+ */
+export function composerSubmitBlock({
+  sessionBlock,
+  sending,
+  hasText,
+}: ComposerSubmitBlockInput): ComposerSubmitBlock | null {
+  if (sessionBlock) return sessionBlock;
+  if (sending) return 'sending';
+  if (!hasText) return 'empty';
+  return null;
+}
+
+/** Hover title for the submit control while the session row is loading. */
+export const COMPOSER_LOADING_TITLE = 'Loading session…';
+
+/** Hover title for the submit control once the load settled without a row. */
+export const COMPOSER_UNAVAILABLE_TITLE = 'Session unavailable';
+
+/**
+ * The hover title a submit block earns on its own, or `null` when the
+ * control's regular title (Send / Queue) should stand.
+ */
+export function composerSubmitBlockTitle(block: ComposerSubmitBlock | null): string | null {
+  switch (block) {
+    case 'loading':
+      return COMPOSER_LOADING_TITLE;
+    case 'unavailable':
+      return COMPOSER_UNAVAILABLE_TITLE;
+    default:
+      return null;
+  }
+}
+
+/** Menu entries for the mode dropdown, in display order. */
+export const COMPOSER_SEND_MODE_OPTIONS: readonly {
+  mode: ComposerSendMode;
+  label: string;
+  description: string;
+}[] = [
+  { mode: 'queue', label: 'Queue', description: 'Deliver after the current turn finishes' },
+  { mode: 'send', label: 'Send', description: 'Deliver to the running agent now' },
+];
diff --git a/apps/staged/src/lib/features/sessions/liveDelivery.test.ts b/apps/staged/src/lib/features/sessions/liveDelivery.test.ts
new file mode 100644
index 000000000..c1d48a033
--- /dev/null
+++ b/apps/staged/src/lib/features/sessions/liveDelivery.test.ts
@@ -0,0 +1,47 @@
+import { describe, expect, it } from 'vitest';
+import {
+  LIVE_DELIVERY_UNKNOWN,
+  liveDeliveryKnowledge,
+  shouldAskLiveDelivery,
+} from './liveDelivery';
+
+describe('liveDelivery', () => {
+  it('asks while the session is running and nothing is settled', () => {
+    expect(shouldAskLiveDelivery('running', LIVE_DELIVERY_UNKNOWN)).toBe(true);
+    expect(shouldAskLiveDelivery('running', { supported: false, settled: false })).toBe(true);
+  });
+
+  it('never asks about a session that is not running', () => {
+    for (const status of ['queued', 'completed', 'cancelled', 'error', undefined] as const) {
+      expect(shouldAskLiveDelivery(status, LIVE_DELIVERY_UNKNOWN)).toBe(false);
+    }
+  });
+
+  it('stops asking once the backend says yes', () => {
+    const knowledge = liveDeliveryKnowledge({ supported: true, retryable: false });
+    expect(knowledge).toEqual({ supported: true, settled: true });
+    expect(shouldAskLiveDelivery('running', knowledge)).toBe(false);
+  });
+
+  it('keeps asking after a retryable no (the agent has not connected yet)', () => {
+    const knowledge = liveDeliveryKnowledge({
+      supported: false,
+      retryable: true,
+      reason: "the session's agent has not connected yet",
+    });
+    expect(knowledge).toEqual({ supported: false, settled: false });
+    expect(shouldAskLiveDelivery('running', knowledge)).toBe(true);
+  });
+
+  it('stops asking after a non-retryable no (another instance owns the session)', () => {
+    const knowledge = liveDeliveryKnowledge({
+      supported: false,
+      retryable: false,
+      reason: 'the session is owned by another Staged instance (pid 42)',
+    });
+    expect(knowledge).toEqual({ supported: false, settled: true });
+    expect(shouldAskLiveDelivery('running', knowledge)).toBe(false);
+    // A status change resets to unknown, which asks again.
+    expect(shouldAskLiveDelivery('running', LIVE_DELIVERY_UNKNOWN)).toBe(true);
+  });
+});
diff --git a/apps/staged/src/lib/features/sessions/liveDelivery.ts b/apps/staged/src/lib/features/sessions/liveDelivery.ts
new file mode 100644
index 000000000..2d7fbe1bc
--- /dev/null
+++ b/apps/staged/src/lib/features/sessions/liveDelivery.ts
@@ -0,0 +1,43 @@
+/**
+ * liveDelivery.ts — What the chat pane remembers about whether a running
+ * session can take a message now, and when it asks the backend again.
+ *
+ * The backend answers `get_session_live_delivery` from a registry lookup, so
+ * asking is cheap, but asking at every poll tick for a whole run is pointless
+ * when the answer cannot change: a session another Staged instance owns never
+ * gets a connection in this process. The backend marks such a "no" as not
+ * `retryable`; a "no" while the connection is still coming up is. The pane
+ * asks until it has a yes or a non-retryable no, and forgets everything when
+ * the status moves or the pane shows a different session.
+ */
+import type { SessionLiveDelivery, SessionStatus } from '../../types';
+
+export interface LiveDeliveryKnowledge {
+  /** The backend said the running session can take a message now. */
+  supported: boolean;
+  /**
+   * Nothing to ask until the status moves: a yes, or a no that cannot change
+   * during this run. False means the next poll tick asks (again).
+   */
+  settled: boolean;
+}
+
+/** Nothing known yet: the next tick while running asks. */
+export const LIVE_DELIVERY_UNKNOWN: LiveDeliveryKnowledge = Object.freeze({
+  supported: false,
+  settled: false,
+});
+
+/** Whether this poll tick should ask the backend. */
+export function shouldAskLiveDelivery(
+  status: SessionStatus | undefined,
+  knowledge: LiveDeliveryKnowledge
+): boolean {
+  return status === 'running' && !knowledge.settled;
+}
+
+/** What to remember from the backend's answer. */
+export function liveDeliveryKnowledge(capability: SessionLiveDelivery): LiveDeliveryKnowledge {
+  if (capability.supported) return { supported: true, settled: true };
+  return { supported: false, settled: !capability.retryable };
+}
diff --git a/apps/staged/src/lib/features/settings/preferences.svelte.ts b/apps/staged/src/lib/features/settings/preferences.svelte.ts
index be5392ce1..ed1f70fdb 100644
--- a/apps/staged/src/lib/features/settings/preferences.svelte.ts
+++ b/apps/staged/src/lib/features/settings/preferences.svelte.ts
@@ -34,9 +34,15 @@ import {
   normalizeProjectStatusOptions,
   type ProjectStatusOption,
 } from '../projects/projectStatusDisplay';
+import {
+  DEFAULT_COMPOSER_SEND_MODE,
+  normalizeComposerSendMode,
+  type ComposerSendMode,
+} from '../sessions/composerSendControl';
 
 // Re-export so picker code can import the type alongside the actions.
 export type { AcpConfigPref, AcpConfigPrefPatch };
+export type { ComposerSendMode };
 
 /**
  * The agent/model/effort the `generate_pikchr` diagram sub-session runs under,
@@ -84,6 +90,12 @@ const DIAGRAM_SUBSESSION_CONFIG_STORE_KEY = 'diagram-subsession-config';
 const BRANCH_PREFIX_STORE_KEY = 'branch-prefix';
 /** The statuses a project can be manually set to (frontend-only). */
 const PROJECT_STATUS_OPTIONS_STORE_KEY = 'project-status-options';
+/**
+ * How the chat composer submits to a running session that can take a message
+ * now: queue it for after the turn, or send it live. Global, not per session
+ * or repo (frontend-only).
+ */
+const COMPOSER_SEND_MODE_STORE_KEY = 'composer-send-mode';
 /** Maximum number of recent agents to remember. */
 const RECENT_AGENTS_MAX = 10;
 
@@ -191,6 +203,11 @@ export const preferences = $state({
   projectStatusOptions: DEFAULT_PROJECT_STATUS_OPTIONS.map((o) => ({
     ...o,
   })) as ProjectStatusOption[],
+  /**
+   * The chat composer's last-used submit mode for running sessions whose
+   * agent can take a message now. Remembered globally; defaults to queueing.
+   */
+  composerSendMode: DEFAULT_COMPOSER_SEND_MODE as ComposerSendMode,
   /** Whether all preferences have been loaded from storage */
   loaded: false,
 });
@@ -266,30 +283,67 @@ function ensureSystemModeListener() {
 // Initialization
 // =============================================================================
 
-let unlistenStatusOptions: (() => void) | undefined;
-let statusOptionsRevision = 0;
+/**
+ * A preference every client (Tauri window or web tab) keeps in step: any
+ * client's write, this one's included, arrives as a store-change event and
+ * goes through `apply`, which must normalize missing or malformed values.
+ */
+function syncedPreference(key: string, apply: (value: unknown) => void) {
+  let unlisten: (() => void) | undefined;
+  let revision = 0;
 
-function applyProjectStatusOptions(value: unknown): void {
-  statusOptionsRevision += 1;
-  preferences.projectStatusOptions =
-    normalizeProjectStatusOptions(value) ?? DEFAULT_PROJECT_STATUS_OPTIONS.map((o) => ({ ...o }));
-}
+  function receive(value: unknown): void {
+    revision += 1;
+    apply(value);
+  }
 
-async function refreshProjectStatusOptions(): Promise<void> {
-  const revision = ++statusOptionsRevision;
-  try {
-    const value = await getStoreValue<unknown>(PROJECT_STATUS_OPTIONS_STORE_KEY);
-    // A notification, local edit, or newer refresh wins over an older snapshot.
-    if (revision === statusOptionsRevision) applyProjectStatusOptions(value);
-  } catch (error) {
-    console.error('[Preferences] Failed to refresh project status options:', error);
+  async function refresh(): Promise<void> {
+    const current = ++revision;
+    try {
+      const value = await getStoreValue<unknown>(key);
+      // A notification, local edit, or newer refresh wins over an older snapshot.
+      if (current === revision) apply(value);
+    } catch (error) {
+      console.error(`[Preferences] Failed to refresh ${key}:`, error);
+    }
   }
+
+  return {
+    /** Subscribe (once) and load the current value. */
+    start(): Promise<void> {
+      unlisten ??= listenToStoreKeyChange(key, receive, {
+        // Close the initial registration gap and recover missed web events on reconnect.
+        onEstablished: () => void refresh(),
+      });
+      return refresh();
+    },
+    /** Persist a value the caller has already applied locally. */
+    save(value: unknown): void {
+      revision += 1;
+      setStoreValue(key, value);
+    },
+    dispose(): void {
+      unlisten?.();
+      revision += 1;
+    },
+  };
 }
 
+const projectStatusOptionsSync = syncedPreference(PROJECT_STATUS_OPTIONS_STORE_KEY, (value) => {
+  preferences.projectStatusOptions =
+    normalizeProjectStatusOptions(value) ?? DEFAULT_PROJECT_STATUS_OPTIONS.map((o) => ({ ...o }));
+});
+
+// Global, so a mode picked in one window must reach the others: a stale `send`
+// would deliver into (or interrupt) a running turn the user meant to queue behind.
+const composerSendModeSync = syncedPreference(COMPOSER_SEND_MODE_STORE_KEY, (value) => {
+  preferences.composerSendMode = normalizeComposerSendMode(value);
+});
+
 if (import.meta.hot) {
   import.meta.hot.dispose(() => {
-    unlistenStatusOptions?.();
-    statusOptionsRevision += 1;
+    projectStatusOptionsSync.dispose();
+    composerSendModeSync.dispose();
   });
 }
 
@@ -300,17 +354,9 @@ if (import.meta.hot) {
 export async function initPreferences(): Promise<void> {
   await initPersistentStore();
 
-  unlistenStatusOptions ??= listenToStoreKeyChange(
-    PROJECT_STATUS_OPTIONS_STORE_KEY,
-    applyProjectStatusOptions,
-    {
-      // Close the initial registration gap and recover missed web events on reconnect.
-      onEstablished: () => void refreshProjectStatusOptions(),
-    }
-  );
-  // Also load without waiting for the web event socket to connect. Statuses
+  // Loads without waiting for the web event socket to connect. Statuses
   // belong to the first project paint, ahead of the slower diff-theme loading.
-  await refreshProjectStatusOptions();
+  await projectStatusOptionsSync.start();
 
   const [savedSize, savedMode] = await Promise.all([
     getStoreValue<number>(SIZE_STORE_KEY),
@@ -404,6 +450,9 @@ export async function initPreferences(): Promise<void> {
   if (typeof savedBranchPrefix === 'string') {
     preferences.branchPrefix = savedBranchPrefix;
   }
+
+  // Load and follow the composer's last-used submit mode (malformed values fall back to queue)
+  await composerSendModeSync.start();
 }
 
 // =============================================================================
@@ -455,15 +504,27 @@ export function setBranchPrefix(prefix: string): void {
   setStoreValue(BRANCH_PREFIX_STORE_KEY, prefix.trim());
 }
 
+// =============================================================================
+// Composer Send Mode Actions
+// =============================================================================
+
+/**
+ * Remember how the chat composer submits to a running session: `queue` (after
+ * the turn) or `send` (deliver now). One global setting for every session.
+ */
+export function setComposerSendMode(mode: ComposerSendMode): void {
+  preferences.composerSendMode = mode;
+  composerSendModeSync.save(mode);
+}
+
 // =============================================================================
 // Project Status Option Actions
 // =============================================================================
 
 /** Replace the list of statuses a project can be set to. */
 export function setProjectStatusOptions(options: ProjectStatusOption[]): void {
-  statusOptionsRevision += 1;
   preferences.projectStatusOptions = options;
-  setStoreValue(PROJECT_STATUS_OPTIONS_STORE_KEY, options);
+  projectStatusOptionsSync.save(options);
 }
 
 /** Restore the built-in project statuses. */
diff --git a/apps/staged/src/lib/features/settings/preferences.test.ts b/apps/staged/src/lib/features/settings/preferences.test.ts
index 8c09118bb..0b6bffe36 100644
--- a/apps/staged/src/lib/features/settings/preferences.test.ts
+++ b/apps/staged/src/lib/features/settings/preferences.test.ts
@@ -144,7 +144,8 @@ describe.each([true, false])('project status synchronization (native=%s)', (nati
     const client = await newClient(native);
     await client.initPreferences();
     await client.initPreferences();
-    expect(listeners).toHaveLength(1);
+    // One listener per synced key (project statuses first, then composer send mode).
+    expect(listeners).toHaveLength(2);
 
     values.set(key, [custom]);
     listeners[0].options?.onEstablished?.();
@@ -199,3 +200,82 @@ describe.each([true, false])('project status synchronization (native=%s)', (nati
     expect(client.preferences.projectStatusOptions).toEqual([]);
   });
 });
+
+describe.each([true, false])('composer send mode (native=%s)', (native) => {
+  const modeKey = 'composer-send-mode';
+
+  it('defaults to queue and persists the chosen mode globally', async () => {
+    const client = await newClient(native);
+    await client.initPreferences();
+    expect(client.preferences.composerSendMode).toBe('queue');
+
+    client.setComposerSendMode('send');
+    expect(client.preferences.composerSendMode).toBe('send');
+    expect(values.get(modeKey)).toBe('send');
+
+    // A fresh client reads the remembered mode back.
+    const reopened = await newClient(native);
+    await reopened.initPreferences();
+    expect(reopened.preferences.composerSendMode).toBe('send');
+  });
+
+  it('falls back to queue for a malformed stored value', async () => {
+    values.set(modeKey, 'later');
+    const client = await newClient(native);
+    await client.initPreferences();
+    expect(client.preferences.composerSendMode).toBe('queue');
+    expect(set).not.toHaveBeenCalledWith(modeKey, expect.anything());
+  });
+
+  it('follows a mode picked in another client without echoing it back', async () => {
+    const a = await newClient(native);
+    await a.initPreferences();
+    const b = await newClient(!native);
+    await b.initPreferences();
+
+    a.setComposerSendMode('send');
+    expect(b.preferences.composerSendMode).toBe('send');
+    b.setComposerSendMode('queue');
+    expect(a.preferences.composerSendMode).toBe('queue');
+    expect(set).toHaveBeenCalledTimes(2);
+  });
+
+  it('falls back to queue when another client deletes or corrupts the mode', async () => {
+    const client = await newClient(native);
+    await client.initPreferences();
+    emit('send', { key: modeKey });
+    emit(null, { key: modeKey, exists: false });
+    expect(client.preferences.composerSendMode).toBe('queue');
+    emit('send', { key: modeKey });
+    emit('later', { key: modeKey });
+    expect(client.preferences.composerSendMode).toBe('queue');
+    expect(set).not.toHaveBeenCalled();
+  });
+
+  it('refreshes on reconnection to recover a missed change', async () => {
+    const client = await newClient(native);
+    await client.initPreferences();
+    values.set(modeKey, 'send');
+    listeners[1].options?.onEstablished?.();
+    await vi.waitFor(() => expect(client.preferences.composerSendMode).toBe('send'));
+  });
+
+  it('keeps a mode picked while the initial read is in flight', async () => {
+    let resolve!: (value: unknown) => void;
+    get.mockImplementation(async (changedKey: string) =>
+      changedKey === modeKey
+        ? new Promise((done) => (resolve = done))
+        : structuredClone(values.get(changedKey))
+    );
+    const client = await newClient(native);
+    const initialized = client.initPreferences();
+    await vi.waitFor(() => expect(get).toHaveBeenCalledWith(modeKey));
+
+    // The local write's event has not arrived yet either.
+    set.mockResolvedValueOnce(undefined);
+    client.setComposerSendMode('send');
+    resolve('queue');
+    await initialized;
+    expect(client.preferences.composerSendMode).toBe('send');
+  });
+});
diff --git a/apps/staged/src/lib/types.ts b/apps/staged/src/lib/types.ts
index ae252bc5c..59cca3065 100644
--- a/apps/staged/src/lib/types.ts
+++ b/apps/staged/src/lib/types.ts
@@ -673,6 +673,50 @@ export interface SessionPoll {
   acpMetadataMessages: SessionMessage[];
 }
 
+/**
+ * Whether a message typed for a session can reach its agent now (the chat
+ * composer's "Send" mode) instead of waiting in the queue. Computed by the
+ * backend from the same gate `continue_repo_session` applies before delivering
+ * to a running session; `reason` says why not.
+ */
+export interface SessionLiveDelivery {
+  supported: boolean;
+  /**
+   * When `supported` is false, whether asking again while the session stays
+   * running could get a yes: true while the agent's connection is still coming
+   * up, false when another Staged instance owns the session or it is not
+   * running. Always false when `supported` is true.
+   */
+  retryable: boolean;
+  reason?: string;
+}
+
+/**
+ * How `send_session_message_now` delivered a message, or why it did not.
+ * Mirrors the Rust `ContinueOutcome::delivery()` spellings, which the
+ * `continue_repo_session` MCP tool shares; `resumed` and `not_resumable` are
+ * only reached through the tool, since the command never resumes a stopped
+ * session (it answers `not_running` instead).
+ */
+export type SessionMessageDeliveryKind =
+  | 'injected'
+  | 'interrupted_and_reprompted'
+  | 'started_new_turn'
+  | 'resumed'
+  | 'rejected_queued'
+  | 'blocked'
+  | 'not_resumable'
+  | 'not_running';
+
+export interface SessionMessageDelivery {
+  delivery: SessionMessageDeliveryKind;
+  /**
+   * Set when the message was not delivered (`blocked`, `not_resumable`,
+   * `not_running`).
+   */
+  reason?: string;
+}
+
 // =============================================================================
 // Branch sessions
 // =============================================================================