feat(session-message): support urgent mid-turn-correction delivery

src/crates/assembly/core/src/agentic/coordination/scheduler.rs

@@ -556,6 +556,22 @@ impl DialogScheduler {
         }
     }
 
+    /// Returns the id of the turn currently being processed by the given session,
+    /// if any. This is the precise turn that `steer_dialog_turn` can target; callers
+    /// should query this first and fall back to a normal submission when idle.
+    pub fn current_processing_turn_id(&self, session_id: &str) -> Option<String> {
+        match self
+            .session_manager
+            .get_session(session_id)
+            .map(|s| s.state.clone())
+        {
+            Some(SessionState::Processing {
+                current_turn_id, ..
+            }) => Some(current_turn_id),
+            _ => None,
+        }
+    }
+
     /// Resume auto-continuation toward an active thread goal (after pause / blocked / usage limit).
     pub async fn deliver_thread_goal_resumed(
         &self,
@@ -4273,6 +4289,24 @@ mod tests {
             .expect("mark turn active");
     }
 
+    #[tokio::test]
+    async fn current_processing_turn_id_reports_running_turn_and_none_for_idle() {
+        let (scheduler, session_manager, _, root) = test_scheduler();
+        let session_id = "queries-running-session";
+        let turn_id = "running-turn";
+        mark_session_processing(&session_manager, &root, session_id, turn_id).await;
+
+        assert_eq!(
+            scheduler.current_processing_turn_id(session_id),
+            Some(turn_id.to_string())
+        );
+        // A session that is not running (or does not exist) reports None.
+        assert_eq!(
+            scheduler.current_processing_turn_id("no-such-session"),
+            None
+        );
+    }
+
     #[tokio::test]
     async fn steering_rejects_stale_processing_state_without_authoritative_active_turn() {
         let (scheduler, session_manager, _, root) = test_scheduler();

src/crates/assembly/core/src/agentic/tools/implementations/session_message_tool.rs

@@ -11,10 +11,12 @@ use crate::util::errors::{BitFunError, BitFunResult};
 use async_trait::async_trait;
 use bitfun_core_types::SessionExecutionTarget;
 use bitfun_runtime_ports::{
-    AgentDialogPrependedReminder, AgentDialogTurnRequest, AgentSessionCreateRequest,
-    AgentSessionListRequest, AgentSessionReplyRoute, AgentSessionSummary,
-    AgentSessionWorkspaceBinding, AgentSessionWorkspaceRequest,
+    AgentDialogPrependedReminder, AgentDialogSteerRequest, AgentDialogTurnPort,
+    AgentDialogTurnRequest, AgentSessionCreateRequest, AgentSessionListRequest,
+    AgentSessionReplyRoute, AgentSessionSummary, AgentSessionWorkspaceBinding,
+    AgentSessionWorkspaceRequest,
 };
+use log::{info, warn};
 use serde::Deserialize;
 use serde_json::{json, Value};
 use std::path::Path;
@@ -300,6 +302,45 @@ struct SessionMessageInput {
     session_name: Option<String>,
     message: String,
     agent_type: Option<SessionMessageAgentType>,
+    /// When true, deliver as an urgent mid-turn correction: if the target session
+    /// is currently processing, the message is injected into its running turn via
+    /// the UserSteering channel instead of starting a new turn. Falls back to
+    /// normal delivery when the target session is not processing.
+    #[serde(default)]
+    urgent: bool,
+}
+
+/// Delivery decision for an urgent message against a target session.
+#[derive(Debug, Clone, PartialEq)]
+enum UrgentDelivery {
+    /// Target session is processing a turn; steer into the running turn.
+    Steer { turn_id: String },
+    /// Target session is idle (or the turn ended); use normal submission.
+    NormalSubmit,
+}
+
+fn resolve_urgent_delivery(processing_turn_id: Option<String>) -> UrgentDelivery {
+    match processing_turn_id {
+        Some(turn_id) => UrgentDelivery::Steer { turn_id },
+        None => UrgentDelivery::NormalSubmit,
+    }
+}
+
+/// Dual-channel redundancy decision for urgent messages:
+/// only attempt the steering channel when the message is urgent AND the target
+/// session already exists (a brand-new session has no running turn to steer
+/// into) AND the dispatch does not carry a plan-todo binding (the steering
+/// channel carries no binding metadata, so a bound message falls back to the
+/// normal submission channel that preserves the binding and the reply route).
+/// Every other case uses the normal submission channel. When steering is
+/// attempted but rejected, the caller falls back to the normal channel, so one
+/// of the two channels always delivers the message.
+fn should_attempt_steering(
+    urgent: bool,
+    created_session_id: Option<&str>,
+    has_plan_todo_binding: bool,
+) -> bool {
+    urgent && created_session_id.is_none() && !has_plan_todo_binding
 }
 
 #[async_trait]
@@ -315,6 +356,7 @@ impl Tool for SessionMessageTool {
 Usage:
 - Create a new session and send: omit "session_id", and provide "workspace", "session_name", "agent_type", and "message".
 - Reusing an existing session: provide "session_id" and "message". You may omit "workspace"; the tool will resolve it from the target session when possible.
+- Urgent correction: set "urgent" to true to inject the message into the target session's running turn instead of waiting for a new turn. Requires "session_id".
 
 Allowed agent types when creating a session:
 - "agentic": Coding-focused agent for implementation, debugging, and code changes.
@@ -358,6 +400,10 @@ Allowed agent types when creating a session:
                     "type": "string",
                     "enum": ["agentic", "Plan", "Cowork", "DeepResearch"],
                     "description": "Required when session_id is omitted. Not allowed when sending to an existing session."
+                },
+                "urgent": {
+                    "type": "boolean",
+                    "description": "When true, deliver as an urgent mid-turn correction: if the target session is processing, inject into its running turn via the UserSteering channel; otherwise fall back to normal delivery. Requires session_id."
                 }
             },
             "required": ["message"],
@@ -556,7 +602,7 @@ Allowed agent types when creating a session:
             .ok_or_else(|| BitFunError::tool("scheduler not initialized".to_string()))?;
         let runtime = CoreServiceAgentRuntime::agent_runtime_with_dialog_turns(
             coordinator.clone(),
-            scheduler,
+            scheduler.clone(),
         )
         .map_err(BitFunError::tool)?;
 
@@ -691,32 +737,116 @@ Allowed agent types when creating a session:
         let (forwarded_message, prepended_messages) =
             self.format_forwarded_message(&params.message);
 
-        runtime
-            .submit_dialog_turn(AgentDialogTurnRequest {
-                session_id: target_session_id.clone(),
-                message: forwarded_message,
-                original_message: Some(params.message.clone()),
-                turn_id: None,
-                execution: Default::default(),
-                agent_type: target_agent_type.clone(),
-                workspace_path: Some(workspace_target.workspace_path.clone()),
-                remote_connection_id: workspace_target.remote_connection_id.clone(),
-                remote_ssh_host: workspace_target.remote_ssh_host.clone(),
-                policy: DialogSubmissionPolicy::for_source(DialogTriggerSource::AgentSession),
-                reply_route: Some(AgentSessionReplyRoute {
-                    source_session_id,
-                    source_workspace_path: source_workspace,
-                    source_remote_connection_id,
-                    source_remote_ssh_host,
-                }),
-                prepended_reminders: prepended_messages,
-                attachments: Vec::new(),
-                metadata: Self::forwarded_user_input_metadata(context),
-            })
-            .await
-            .map_err(|error| {
-                BitFunError::tool(CoreServiceAgentRuntime::runtime_error_message(error))
-            })?;
+        // Urgent delivery: when the target session is currently processing a turn,
+        // inject the message into that running turn via the UserSteering channel
+        // instead of starting a new turn. Honest fallback: when the target session
+        // is not processing, or the steering is rejected (the turn ended between
+        // the state query and the submit), deliver through the normal submission
+        // path so the message is never dropped.
+        let mut steering_turn_id: Option<String> = None;
+        // The base `SessionMessageInput` carries no plan-todo binding fields, so
+        // the steering gate never sees a binding and only narrows on `urgent` plus
+        // the "target session already exists" criterion.
+        let has_plan_todo_binding = false;
+        if should_attempt_steering(
+            params.urgent,
+            created_session_id.as_deref(),
+            has_plan_todo_binding,
+        ) {
+            match resolve_urgent_delivery(scheduler.current_processing_turn_id(&target_session_id))
+            {
+                UrgentDelivery::Steer { turn_id } => {
+                    // Reuse the native `AgentDialogSteerRequest` struct; the sender
+                    // identity reminder is baked into `content` (the native struct has
+                    // no `prepended_reminders` field) so the target agent still sees
+                    // it originates from another agent. `display_content` stays the
+                    // clean user-visible message.
+                    let mut steering_content = forwarded_message.clone();
+                    for reminder in &prepended_messages {
+                        steering_content = format!("{}\n{}", reminder.text, steering_content);
+                    }
+                    match scheduler
+                        .steer_dialog_turn(AgentDialogSteerRequest {
+                            session_id: target_session_id.clone(),
+                            turn_id: turn_id.clone(),
+                            content: steering_content,
+                            display_content: Some(params.message.clone()),
+                            attachments: Vec::new(),
+                            metadata: serde_json::Map::new(),
+                        })
+                        .await
+                    {
+                        Ok(_outcome) => {
+                            steering_turn_id = Some(turn_id.clone());
+                            info!(
+                                "Urgent SessionMessage steered into running turn: source_session_id={}, target_session_id={}, turn_id={}",
+                                source_session_id, target_session_id, turn_id
+                            );
+                        }
+                        Err(error) => {
+                            warn!(
+                                "Urgent SessionMessage steering rejected, falling back to normal submit: target_session_id={}, turn_id={}, error={}",
+                                target_session_id, turn_id, error
+                            );
+                        }
+                    }
+                }
+                UrgentDelivery::NormalSubmit => {}
+            }
+        }
+
+        if steering_turn_id.is_none() {
+            runtime
+                .submit_dialog_turn(AgentDialogTurnRequest {
+                    session_id: target_session_id.clone(),
+                    message: forwarded_message,
+                    original_message: Some(params.message.clone()),
+                    turn_id: None,
+                    execution: Default::default(),
+                    agent_type: target_agent_type.clone(),
+                    workspace_path: Some(workspace_target.workspace_path.clone()),
+                    remote_connection_id: workspace_target.remote_connection_id.clone(),
+                    remote_ssh_host: workspace_target.remote_ssh_host.clone(),
+                    policy: DialogSubmissionPolicy::for_source(DialogTriggerSource::AgentSession),
+                    reply_route: Some(AgentSessionReplyRoute {
+                        source_session_id,
+                        source_workspace_path: source_workspace,
+                        source_remote_connection_id,
+                        source_remote_ssh_host,
+                    }),
+                    prepended_reminders: prepended_messages,
+                    attachments: Vec::new(),
+                    metadata: Self::forwarded_user_input_metadata(context),
+                })
+                .await
+                .map_err(|error| {
+                    BitFunError::tool(CoreServiceAgentRuntime::runtime_error_message(error))
+                })?;
+        }
+
+        let urgent_fell_back =
+            params.urgent && steering_turn_id.is_none() && created_session_id.is_none();
+        let mut result_text = if let Some(steered_turn_id) = steering_turn_id.as_ref() {
+            format!(
+                "Urgent message injected into the running turn '{}' of session '{}' in workspace '{}' using agent type '{}'.",
+                steered_turn_id, target_session_id, workspace_target.workspace_path, target_agent_type
+            )
+        } else if let Some(created_session_id) = created_session_id.as_ref() {
+            format!(
+                "Created session '{}' and accepted the message in workspace '{}' using agent type '{}'.",
+                created_session_id, workspace_target.workspace_path, target_agent_type
+            )
+        } else {
+            format!(
+                "Message accepted for session '{}' in workspace '{}' using agent type '{}'.",
+                target_session_id, workspace_target.workspace_path, target_agent_type
+            )
+        };
+        if urgent_fell_back {
+            result_text.push_str(
+                " Steering into the running turn was not possible (the target session was idle, its turn had just ended, the queue was congested, or the message carries a plan-todo binding that the steering channel cannot carry), so the urgent message was delivered as a normal submission instead of a mid-turn correction.",
+            );
+        }
 
         Ok(vec![ToolResult::Result {
             data: json!({
@@ -725,18 +855,9 @@ Allowed agent types when creating a session:
                 "target_session_id": target_session_id.clone(),
                 "target_agent_type": target_agent_type.clone(),
                 "created_session_id": created_session_id.clone(),
+                "delivery": if steering_turn_id.is_some() { "steered" } else { "submitted" },
             }),
-            result_for_assistant: Some(if let Some(created_session_id) = created_session_id {
-                format!(
-                    "Created session '{}' and accepted the message in workspace '{}' using agent type '{}'.",
-                    created_session_id, workspace_target.workspace_path, target_agent_type
-                )
-            } else {
-                format!(
-                    "Message accepted for session '{}' in workspace '{}' using agent type '{}'.",
-                    target_session_id, workspace_target.workspace_path, target_agent_type
-                )
-            }),
+            result_for_assistant: Some(result_text),
             image_attachments: None,
         }])
     }
@@ -1076,4 +1197,72 @@ mod tests {
             Some("workspace is required when session_id is omitted")
         );
     }
+
+    #[test]
+    fn session_message_input_defaults_urgent_to_false_for_backward_compat() {
+        let input: SessionMessageInput = serde_json::from_value(json!({
+            "session_id": "worker_1",
+            "message": "hello",
+        }))
+        .expect("legacy payload without urgent must parse");
+
+        assert!(!input.urgent);
+    }
+
+    #[test]
+    fn session_message_input_parses_urgent_flag() {
+        let input: SessionMessageInput = serde_json::from_value(json!({
+            "session_id": "worker_1",
+            "message": "stop what you are doing and correct this",
+            "urgent": true,
+        }))
+        .expect("payload with urgent must parse");
+
+        assert!(input.urgent);
+    }
+
+    #[test]
+    fn urgent_delivery_steers_into_a_processing_turn() {
+        assert_eq!(
+            resolve_urgent_delivery(Some("turn-7".to_string())),
+            UrgentDelivery::Steer {
+                turn_id: "turn-7".to_string()
+            }
+        );
+    }
+
+    #[test]
+    fn urgent_delivery_falls_back_to_normal_submit_for_idle_session() {
+        assert_eq!(resolve_urgent_delivery(None), UrgentDelivery::NormalSubmit);
+    }
+
+    #[test]
+    fn urgent_message_to_existing_session_attempts_steering_channel() {
+        assert!(should_attempt_steering(true, None, false));
+    }
+
+    #[test]
+    fn urgent_message_to_new_session_uses_normal_channel_only() {
+        assert!(!should_attempt_steering(true, Some("new-session-1"), false));
+    }
+
+    #[test]
+    fn urgent_message_with_plan_todo_binding_uses_normal_channel_only() {
+        // The steering channel carries no plan-todo binding metadata, so a
+        // bound dispatch must fall back to the normal submission channel that
+        // preserves the binding and the reply route.
+        assert!(!should_attempt_steering(true, None, true));
+        assert!(!should_attempt_steering(true, Some("new-session-1"), true));
+    }
+
+    #[test]
+    fn non_urgent_message_never_attempts_steering_channel() {
+        assert!(!should_attempt_steering(false, None, false));
+        assert!(!should_attempt_steering(
+            false,
+            Some("new-session-1"),
+            false
+        ));
+        assert!(!should_attempt_steering(false, None, true));
+    }
 }