From fb7d9935b8cc1f2f2b5d7f4e3ad061b81495a9c9 Mon Sep 17 00:00:00 2001 From: user Date: Mon, 24 Aug 2026 15:52:59 +0800 Subject: [PATCH 1/3] feat(session-message): add urgent delivery type layer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add the urgent mid-turn-correction type layer to the SessionMessage tool so a message can target a session's running turn instead of starting a new turn: - `SessionMessageInput.urgent: bool` (`#[serde(default)]`, backward compatible) - `UrgentDelivery` enum (`Steer{ turn_id }` / `NormalSubmit`) - `resolve_urgent_delivery` and `should_attempt_steering` pure functions - `urgent` in `input_schema` + a "Urgent correction" usage line in `description` - 8 pure/parse test contracts (the batch-item contract requires the batch framework, which is a separate feature and out of scope for this base) Test: - cargo check -p bitfun-core --features agent-runtime --jobs 4 (0 errors, 0 new warnings; only pre-existing `fork_session_for_plugin`) - cargo test -p bitfun-core --features agent-runtime --jobs 4 urgent (8 passed) AI: 已测 --- .../implementations/session_message_tool.rs | 116 ++++++++++++++++++ 1 file changed, 116 insertions(+) diff --git a/src/crates/assembly/core/src/agentic/tools/implementations/session_message_tool.rs b/src/crates/assembly/core/src/agentic/tools/implementations/session_message_tool.rs index a94aee7a0b..6f4b6ff5ee 100644 --- a/src/crates/assembly/core/src/agentic/tools/implementations/session_message_tool.rs +++ b/src/crates/assembly/core/src/agentic/tools/implementations/session_message_tool.rs @@ -300,6 +300,49 @@ struct SessionMessageInput { session_name: Option, message: String, agent_type: Option, + /// 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)] + #[allow(dead_code)] + urgent: bool, +} + +/// Delivery decision for an urgent message against a target session. +#[derive(Debug, Clone, PartialEq)] +#[allow(dead_code)] +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, +} + +#[allow(dead_code)] +fn resolve_urgent_delivery(processing_turn_id: Option) -> 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. +#[allow(dead_code)] +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 +358,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 +402,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"], @@ -1076,4 +1124,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)); + } } From 5934f5dfdd1f0e932c80244f66057b162fc5446e Mon Sep 17 00:00:00 2001 From: user Date: Mon, 24 Aug 2026 15:57:08 +0800 Subject: [PATCH 2/3] feat(scheduler): expose current processing turn id MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add `DialogScheduler::current_processing_turn_id(session_id)` — a thin read-only query over the native `SessionState::Processing { current_turn_id }` state. This is the precise turn that `steer_dialog_turn` can target; callers query it first and fall back to a normal submission when the session is idle. Test: - cargo check -p bitfun-core --features agent-runtime --jobs 4 (0 errors, 0 new warnings; only pre-existing `fork_session_for_plugin`) - cargo test -p bitfun-core --features agent-runtime --jobs 4 current_processing_turn_id (1 passed) AI: 已测 --- .../src/agentic/coordination/scheduler.rs | 34 +++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/src/crates/assembly/core/src/agentic/coordination/scheduler.rs b/src/crates/assembly/core/src/agentic/coordination/scheduler.rs index 120d16bbee..6b6979921f 100644 --- a/src/crates/assembly/core/src/agentic/coordination/scheduler.rs +++ b/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 { + 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(); From 4a89f2b0fd9aa1906d7f63df4370e75e9d9212f0 Mon Sep 17 00:00:00 2001 From: user Date: Mon, 24 Aug 2026 16:05:23 +0800 Subject: [PATCH 3/3] feat(session-message): steer urgent messages into running turns MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Integrate urgent delivery behavior into SessionMessage dispatch. When `urgent: true` and the target session is processing a turn, the message is steered into that running turn via the native `steer_dialog_turn` channel, reusing the native `AgentDialogSteerRequest` struct (with the sender identity reminder baked into `content`). When the target is idle, or steering is rejected, the message falls back to the normal `submit_dialog_turn` submission so it is never dropped. Adds the honest fallback note to the result text, the three-state result message (steered / created / accepted), and a machine-readable `delivery` field ("steered" / "submitted"). Removes the transient `#[allow(dead_code)]` suppressions now that the type layer is consumed. Test: - cargo check -p bitfun-core --features agent-runtime --jobs 4 (0 errors, 0 new warnings; only pre-existing `fork_session_for_plugin`) - cargo test -p bitfun-core --features agent-runtime --jobs 4 --lib (1430 passed, 0 failed) - cargo clippy -p bitfun-core --features agent-runtime --all-targets (0 new warnings) - cargo fmt -p bitfun-core (clean) AI: 已测 --- .../implementations/session_message_tool.rs | 163 +++++++++++++----- 1 file changed, 118 insertions(+), 45 deletions(-) diff --git a/src/crates/assembly/core/src/agentic/tools/implementations/session_message_tool.rs b/src/crates/assembly/core/src/agentic/tools/implementations/session_message_tool.rs index 6f4b6ff5ee..9dee354998 100644 --- a/src/crates/assembly/core/src/agentic/tools/implementations/session_message_tool.rs +++ b/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; @@ -305,13 +307,11 @@ struct SessionMessageInput { /// the UserSteering channel instead of starting a new turn. Falls back to /// normal delivery when the target session is not processing. #[serde(default)] - #[allow(dead_code)] urgent: bool, } /// Delivery decision for an urgent message against a target session. #[derive(Debug, Clone, PartialEq)] -#[allow(dead_code)] enum UrgentDelivery { /// Target session is processing a turn; steer into the running turn. Steer { turn_id: String }, @@ -319,7 +319,6 @@ enum UrgentDelivery { NormalSubmit, } -#[allow(dead_code)] fn resolve_urgent_delivery(processing_turn_id: Option) -> UrgentDelivery { match processing_turn_id { Some(turn_id) => UrgentDelivery::Steer { turn_id }, @@ -336,7 +335,6 @@ fn resolve_urgent_delivery(processing_turn_id: Option) -> UrgentDelivery /// 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. -#[allow(dead_code)] fn should_attempt_steering( urgent: bool, created_session_id: Option<&str>, @@ -604,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)?; @@ -739,32 +737,116 @@ Allowed agent types when creating a session: let (forwarded_message, prepended_messages) = self.format_forwarded_message(¶ms.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 = 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!({ @@ -773,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, }]) }