From 7e1bc1ff4a28d1b5a65643776397a6d6e1d0cd5b Mon Sep 17 00:00:00 2001 From: limityan Date: Sun, 2 Aug 2026 21:15:20 +0800 Subject: [PATCH] feat(runtime): unify active-turn steering across clients --- .../agent-runtime-services-design.md | 7 +- .../rules/source/forbidden-rules.mjs | 2 +- scripts/core-boundaries/self-test.mjs | 5 +- src/apps/cli/src/agent/runtime_client.rs | 95 +++++++- src/apps/cli/src/chat_state.rs | 95 +++++++- src/apps/cli/src/dispatch/worker.rs | 39 ++-- src/apps/cli/src/modes/chat/commands.rs | 68 +++++- src/apps/cli/src/modes/chat/run.rs | 23 ++ src/apps/cli/src/modes/chat/sessions.rs | 1 + src/apps/cli/src/modes/chat/tests.rs | 78 ++++++- src/apps/cli/src/modes/chat/transcript.rs | 16 +- src/apps/cli/src/shared_runtime.rs | 33 +++ src/apps/cli/src/ui/chat/render.rs | 42 ++++ src/apps/desktop/src/api/agentic_api.rs | 44 +++- .../adapters/agent-runtime-ipc/AGENTS-CN.md | 2 +- .../adapters/agent-runtime-ipc/AGENTS.md | 3 +- .../agent-runtime-ipc/src/operation.rs | 51 ++++- .../agent-runtime-ipc/src/protocol.rs | 2 +- .../adapters/agent-runtime-ipc/src/server.rs | 40 ++++ .../src/tests/protocol_contracts.rs | 44 +++- .../src/tests/shared_controller.rs | 125 ++++++++++ .../src/agentic/coordination/scheduler.rs | 165 +++++++++++++- .../assembly/core/src/product_runtime.rs | 18 +- .../core/src/service_agent_runtime.rs | 7 + src/crates/contracts/runtime-ports/src/lib.rs | 67 +++++- .../agent-runtime/examples/sdk_minimal.rs | 2 +- .../execution/agent-runtime/src/runtime.rs | 215 ++++++++++++++++-- src/crates/execution/agent-runtime/src/sdk.rs | 19 +- .../agent-runtime/tests/sdk_smoke.rs | 2 +- .../components/PendingQueuePanel.tsx | 41 ++-- .../PendingQueueModule.test.ts | 96 ++++++++ .../flow-chat-manager/PendingQueueModule.ts | 30 +++ src/web-ui/src/locales/en-US/flow-chat.json | 1 + src/web-ui/src/locales/zh-CN/flow-chat.json | 1 + src/web-ui/src/locales/zh-TW/flow-chat.json | 1 + 35 files changed, 1332 insertions(+), 148 deletions(-) create mode 100644 src/web-ui/src/flow_chat/services/flow-chat-manager/PendingQueueModule.test.ts diff --git a/docs/architecture/agent-runtime-services-design.md b/docs/architecture/agent-runtime-services-design.md index 7c566b4bf1..75f9770ff0 100644 --- a/docs/architecture/agent-runtime-services-design.md +++ b/docs/architecture/agent-runtime-services-design.md @@ -68,7 +68,7 @@ Agent Runtime API 的逻辑归属与物理部署分离：相同归属模块可 私有 SDK Host 或目标机器 Runtime 中。任何 Rust 部署都只管理自己进程树内的服务与 Node/Bun Plugin Host；不能因为多个 GUI/TUI/Remote Client 连接就复制 Runtime 状态模块，或按 Client/Workspace 创建 Plugin Host。 -Rust Runtime SDK 以 `AGENT_RUNTIME_SDK_API_VERSION` 标记兼容边界。当前接口版本为 v3 preview： +Rust Runtime SDK 以 `AGENT_RUNTIME_SDK_API_VERSION` 标记兼容边界。当前接口版本为 v4 preview： 小版本更新允许增加可选 builder hook、有默认实现的端口方法或注册表查询能力，但不得向外部可用 Rust 结构体字面量（struct literal）构造的 DTO 直接增加字段，也不得改变既有端口语义、错误分类、session / turn 标识含义或 默认 feature 依赖。任何需要调用方改写现有嵌入代码的变更，必须提升接口版本并提供兼容迁移路径。 @@ -80,6 +80,9 @@ v2 的迁移只涉及 Rust 错误名词治理：调用方把 v3 为 `AgentDialogTurnRequest` 增加来源无关的 `execution` 事实。现有 Rust struct literal 调用方迁移时增加 `execution: AgentDialogTurnExecution::Standard`（或 `Default::default()`）；旧 wire payload 缺省为标准执行。 +v4 将活动 Turn 的文本 steer 纳入 `AgentDialogTurnPort`，复用同一个 Runtime owner 和精确 Session/Turn +身份校验；默认端口实现仍返回 `NotAvailable`，未选择该能力的 provider 不需要建立第二套 queue 或 transport。 + 只要外部调用方仍必须导入 `bitfun-core`、启用 `product-full`、持有具体服务管理器、读取产品命令 注册表、理解 ACP/内部端口或依赖全局可变状态，公开 SDK 发布边界就不成立。公开 SDK 的完整 术语、能力等价和版本要求以 [`agent-sdk-product-architecture.md`](agent-sdk-product-architecture.md) 为准。 @@ -423,7 +426,7 @@ impl AgentRuntime { 该 Rust 接口是内部产品入口复用的当前形态，不是公开 Python/TypeScript SDK 的目标 API。它必须只接收 已组装的类型化部件，不负责创建 文件系统、终端、MCP、AI 客户端、Remote 提供方或产品命令。 -当前 v3 preview 接口以 message / attachment / metadata 和默认标准执行目标作为最小输入形态；若把 +当前 v4 preview 接口以 message / attachment / metadata、默认标准执行目标和活动 Turn 文本 steer 作为最小输入形态；若把 model-round cancellation token、结构化 AgentInput 或更复杂的事件游标纳入公开 SDK， 必须分别评审 Rust Runtime SDK、SDK Host protocol 和公开 SDK API 的版本，并保留旧路径兼容。 diff --git a/scripts/core-boundaries/rules/source/forbidden-rules.mjs b/scripts/core-boundaries/rules/source/forbidden-rules.mjs index 3da0ae7f35..bde743a6c5 100644 --- a/scripts/core-boundaries/rules/source/forbidden-rules.mjs +++ b/scripts/core-boundaries/rules/source/forbidden-rules.mjs @@ -6,7 +6,7 @@ export const forbiddenContentRules = [ reason: 'agent-runtime-ipc operation scope is frozen to the reviewed Shared TUI slice', patterns: [ { - regex: /^\s+(?!(?:Health|ListSessions|CreateSession|RestoreSession|DeleteSession|ForkSession|RenameSession|UpdateSessionMode|UpdateSessionModel|ReloadSessionContext|CompactSession|UndoSession|RedoSession|SearchWorkspaceReferences|WorkspaceReferencesForMessage|WorkspaceDiff|SubmitTurn|RunUserShellCommand|CancelTurn|PendingPermissions|RespondPermission|SubmitUserAnswers|Unit|Sessions|SessionCreated|SessionRestored|SessionForked|SessionReverted|WorkspaceReferenceSearch|WorkspaceReferences|TurnAccepted|TurnCancelled|None|CurrentController|AttachExisting|UncontrolledTarget|Self|RuntimeIpcSessionRequirement|RuntimeIpcOperationRules|RuntimeSessionForkRequest|AgentContextReloadRequest|AgentDialogTurnRequest|AgentMessageWorkspaceReferencesRequest|AgentSessionCompactionRequest|AgentSessionCreateRequest|AgentSessionCreateResult|AgentSessionListRequest|AgentSessionModeUpdateRequest|AgentSessionModelUpdateRequest|AgentSessionRevertRequest|AgentSessionRevertResult|AgentSessionSummary|AgentTurnCancellationRequest|AgentTurnCancellationResult|AgentUserShellCommandRequest|AgentWorkspaceReference|AgentWorkspaceReferenceSearchRequest|AgentWorkspaceReferenceSearchResult|SessionTranscript|WorkspaceDiffSnapshot)\b)[A-Z][A-Za-z0-9_]*\b/, + regex: /^\s+(?!(?:Health|ListSessions|CreateSession|RestoreSession|DeleteSession|ForkSession|RenameSession|UpdateSessionMode|UpdateSessionModel|ReloadSessionContext|CompactSession|UndoSession|RedoSession|SearchWorkspaceReferences|WorkspaceReferencesForMessage|WorkspaceDiff|SubmitTurn|SteerTurn|RunUserShellCommand|CancelTurn|PendingPermissions|RespondPermission|SubmitUserAnswers|Unit|Sessions|SessionCreated|SessionRestored|SessionForked|SessionReverted|WorkspaceReferenceSearch|WorkspaceReferences|TurnAccepted|TurnSteered|TurnCancelled|None|CurrentController|AttachExisting|UncontrolledTarget|Self|RuntimeIpcSessionRequirement|RuntimeIpcOperationRules|RuntimeSessionForkRequest|AgentContextReloadRequest|AgentDialogSteerRequest|AgentDialogTurnRequest|AgentMessageWorkspaceReferencesRequest|AgentSessionCompactionRequest|AgentSessionCreateRequest|AgentSessionCreateResult|AgentSessionListRequest|AgentSessionModeUpdateRequest|AgentSessionModelUpdateRequest|AgentSessionRevertRequest|AgentSessionRevertResult|AgentSessionSummary|AgentTurnCancellationRequest|AgentTurnCancellationResult|AgentUserShellCommandRequest|AgentWorkspaceReference|AgentWorkspaceReferenceSearchRequest|AgentWorkspaceReferenceSearchResult|SessionTranscript|WorkspaceDiffSnapshot)\b)[A-Z][A-Za-z0-9_]*\b/, message: 'agent-runtime-ipc may not add archive, replay, observer, general controller-transfer, or other operations beyond the reviewed Shared TUI slice', }, diff --git a/scripts/core-boundaries/self-test.mjs b/scripts/core-boundaries/self-test.mjs index d42b312630..e23cbf4e15 100644 --- a/scripts/core-boundaries/self-test.mjs +++ b/scripts/core-boundaries/self-test.mjs @@ -5086,10 +5086,13 @@ async fn release_baseline_claim(release: BaselineClaimRelease) -> Result<(), Dis runtimeIpcOperationPattern.test(' WorkspaceDiff {') || runtimeIpcOperationPattern.test(' WorkspaceDiffSnapshot,') || runtimeIpcOperationPattern.test(' SubmitTurn {') || + runtimeIpcOperationPattern.test(' SteerTurn {') || + runtimeIpcOperationPattern.test(' AgentDialogSteerRequest {') || runtimeIpcOperationPattern.test(' RunUserShellCommand {') || runtimeIpcOperationPattern.test(' AgentUserShellCommandRequest {') || runtimeIpcOperationPattern.test(' SessionForked {') || - runtimeIpcOperationPattern.test(' SessionReverted {') + runtimeIpcOperationPattern.test(' SessionReverted {') || + runtimeIpcOperationPattern.test(' TurnSteered {') ) { throw new Error('agent-runtime-ipc operation guard must preserve the Shared TUI operation budget'); } diff --git a/src/apps/cli/src/agent/runtime_client.rs b/src/apps/cli/src/agent/runtime_client.rs index 5ea308e226..c403561674 100644 --- a/src/apps/cli/src/agent/runtime_client.rs +++ b/src/apps/cli/src/agent/runtime_client.rs @@ -12,18 +12,19 @@ use std::sync::{Arc, RwLock}; use tokio::sync::{broadcast, Mutex}; use bitfun_agent_runtime::sdk::{ - AgentDialogTurnExecution, AgentDialogTurnRequest, AgentEventReceiver, AgentInputAttachment, - AgentLocalCommandTurnRecordRequest, AgentMessageWorkspaceReferencesRequest, AgentRuntime, - AgentSessionCompactionRequest, AgentSessionCreateRequest, AgentSessionDeleteRequest, - AgentSessionForkBeforeTurnRequest, AgentSessionForkRequest, AgentSessionForkResult, - AgentSessionListRequest, AgentSessionModeUpdateRequest, AgentSessionModelUpdateRequest, - AgentSessionRenameRequest, AgentSessionRestoreRequest, AgentSessionRevertRequest, - AgentSessionRevertResult, AgentSessionUsageRequest, AgentTurnCancellationRequest, - AgentTurnSettlementRequest, AgentUserAnswersRequest, AgentUserShellCommandRequest, - AgentWorkspaceReference, AgentWorkspaceReferenceSearchRequest, - AgentWorkspaceReferenceSearchResult, PermissionReply, PermissionRequest, - PermissionRequestEventReceiver, PortError, PortErrorKind, RuntimeError, SessionTranscript, - SessionTranscriptRequest, SessionUsageReport, WorkspaceDiffSnapshot, + AgentDialogSteerRequest, AgentDialogTurnExecution, AgentDialogTurnRequest, AgentEventReceiver, + AgentInputAttachment, AgentLocalCommandTurnRecordRequest, + AgentMessageWorkspaceReferencesRequest, AgentRuntime, AgentSessionCompactionRequest, + AgentSessionCreateRequest, AgentSessionDeleteRequest, AgentSessionForkBeforeTurnRequest, + AgentSessionForkRequest, AgentSessionForkResult, AgentSessionListRequest, + AgentSessionModeUpdateRequest, AgentSessionModelUpdateRequest, AgentSessionRenameRequest, + AgentSessionRestoreRequest, AgentSessionRevertRequest, AgentSessionRevertResult, + AgentSessionUsageRequest, AgentTurnCancellationRequest, AgentTurnSettlementRequest, + AgentUserAnswersRequest, AgentUserShellCommandRequest, AgentWorkspaceReference, + AgentWorkspaceReferenceSearchRequest, AgentWorkspaceReferenceSearchResult, DialogSteerOutcome, + PermissionReply, PermissionRequest, PermissionRequestEventReceiver, PortError, PortErrorKind, + RuntimeError, SessionTranscript, SessionTranscriptRequest, SessionUsageReport, + WorkspaceDiffSnapshot, }; use bitfun_agent_runtime_ipc::{ RuntimeIpcClient, RuntimeIpcClientError, RuntimeIpcClientEvent, RuntimeIpcErrorCode, @@ -1319,6 +1320,55 @@ impl CliAgentRuntimeClient { submission } + pub(crate) async fn steer_current_turn( + &self, + content: String, + display_content: Option, + ) -> Result { + if content.trim().is_empty() { + return Err(anyhow::anyhow!("Steering content cannot be empty")); + } + let session_id = self + .session_id + .lock() + .await + .clone() + .ok_or_else(|| anyhow::anyhow!("No active session is available for steering"))?; + let turn_id = self + .current_turn_id + .lock() + .await + .clone() + .ok_or_else(|| anyhow::anyhow!("No active turn is available for steering"))?; + let request = AgentDialogSteerRequest { + session_id: session_id.clone(), + turn_id: turn_id.clone(), + content, + display_content, + }; + + match &self.backend { + CliAgentRuntimeBackend::Embedded(runtime) => match runtime + .steer_dialog_turn(request) + .await + .map_err(|error| anyhow::anyhow!(error.into_message()))? + { + DialogSteerOutcome::Buffered { steering_id, .. } => Ok(steering_id), + }, + CliAgentRuntimeBackend::Shared(client) => match client + .request(RuntimeIpcOperation::SteerTurn { request }) + .await? + { + RuntimeIpcOperationResult::TurnSteered { + session_id: steered_session, + turn_id: steered_turn, + steering_id, + } if steered_session == session_id && steered_turn == turn_id => Ok(steering_id), + _ => Err(unexpected_shared_result("steer_turn")), + }, + } + } + pub(crate) async fn run_user_shell_command( &self, command: String, @@ -1966,6 +2016,27 @@ mod tests { assert!(!compact.contains("serde_json::from_value")); } + #[test] + fn steering_uses_the_existing_runtime_contract_in_both_deployments() { + let source = include_str!("runtime_client.rs").replace("\r\n", "\n"); + let steering = source + .split_once("pub(crate) async fn steer_current_turn(") + .expect("steering method") + .1 + .split_once("pub(crate) async fn run_user_shell_command(") + .expect("steering method boundary") + .0; + + assert!(steering.contains("AgentDialogSteerRequest")); + assert!(steering.contains("CliAgentRuntimeBackend::Embedded(runtime)")); + assert!(steering.contains(".steer_dialog_turn(request)")); + assert!(steering.contains("CliAgentRuntimeBackend::Shared(client)")); + assert!(steering.contains("RuntimeIpcOperation::SteerTurn { request }")); + assert!(steering.contains("RuntimeIpcOperationResult::TurnSteered")); + assert!(!steering.contains("RuntimeIpcOperation::SubmitTurn")); + assert!(!steering.contains("Uuid::new_v4")); + } + #[test] fn image_attachments_use_the_runtime_contract_and_fail_before_shared_ipc() { let source = include_str!("runtime_client.rs").replace("\r\n", "\n"); diff --git a/src/apps/cli/src/chat_state.rs b/src/apps/cli/src/chat_state.rs index ae93619dbc..c5e1073aad 100644 --- a/src/apps/cli/src/chat_state.rs +++ b/src/apps/cli/src/chat_state.rs @@ -155,6 +155,12 @@ pub(crate) enum FlowItem { Text { content: String, is_streaming: bool }, /// AI thinking/reasoning block Thinking { content: String }, + /// User steering injected between model-round flow items. + UserSteering { + steering_id: String, + content: String, + is_pending: bool, + }, /// Tool call block Tool { tool_state: ToolDisplayState }, } @@ -307,7 +313,9 @@ fn visible_message_text(message: &ChatMessage) -> String { .iter() .filter_map(|item| match item { FlowItem::Text { content, .. } => Some(content.as_str()), - FlowItem::Thinking { .. } | FlowItem::Tool { .. } => None, + FlowItem::Thinking { .. } | FlowItem::UserSteering { .. } | FlowItem::Tool { .. } => { + None + } }) .collect::>() .join("\n") @@ -896,6 +904,50 @@ impl ChatState { self.rebuild_streaming_message(); } + /// Add an optimistic steering item or upgrade it when the runtime emits + /// the authoritative injection event. Returns true only for a new item. + pub(crate) fn handle_user_steering( + &mut self, + steering_id: &str, + content: &str, + is_pending: bool, + ) -> bool { + if !self.is_processing || self.current_turn_id.is_none() { + return false; + } + if let Some(existing) = self.current_flow_items.iter_mut().find(|item| { + matches!( + item, + FlowItem::UserSteering { + steering_id: existing_id, + .. + } if existing_id == steering_id + ) + }) { + if let FlowItem::UserSteering { + content: existing_content, + is_pending: existing_pending, + .. + } = existing + { + *existing_content = content.to_string(); + if !is_pending { + *existing_pending = false; + } + } + self.rebuild_streaming_message(); + return false; + } + + self.current_flow_items.push(FlowItem::UserSteering { + steering_id: steering_id.to_string(), + content: content.to_string(), + is_pending, + }); + self.rebuild_streaming_message(); + true + } + /// Handle a tool event. /// New tools are appended to current_flow_items in chronological order. /// Existing tools are updated in-place via tool_index for O(1) lookup. @@ -1829,6 +1881,47 @@ mod tests { assert_create_plan_item(&state.current_flow_items[0]); } + #[test] + fn user_steering_is_deduplicated_and_preserves_stream_order() { + let mut state = ChatState::new( + "session-1".to_string(), + "Session".to_string(), + "agentic".to_string(), + None, + ); + state.handle_turn_started("turn-1", "Start the task"); + state.handle_text_chunk("Before steering"); + + assert!(state.handle_user_steering("steer-1", "Also check tests", true)); + assert!(!state.handle_user_steering("steer-1", "Also check tests", false)); + state.handle_text_chunk("After steering"); + + assert!(matches!( + state.current_flow_items.as_slice(), + [ + FlowItem::Text { content: before, .. }, + FlowItem::UserSteering { + steering_id, + content, + is_pending: false, + }, + FlowItem::Text { content: after, .. }, + ] if before == "Before steering" + && steering_id == "steer-1" + && content == "Also check tests" + && after == "After steering" + )); + assert_eq!( + state.current_flow_items.len(), + state + .messages + .last() + .expect("assistant message") + .flow_items + .len() + ); + } + #[test] fn deferred_history_projects_effective_view_without_mutating_wire_message() { let wire_input = deferred_input(); diff --git a/src/apps/cli/src/dispatch/worker.rs b/src/apps/cli/src/dispatch/worker.rs index 740e6f90e9..41ca3d6a2a 100644 --- a/src/apps/cli/src/dispatch/worker.rs +++ b/src/apps/cli/src/dispatch/worker.rs @@ -4,9 +4,9 @@ use std::time::Duration; use anyhow::{anyhow, bail, Context, Result}; use bitfun_agent_runtime::sdk::{ - AgentDialogTurnRequest, AgentSessionCreateRequest, AgentSessionRestoreRequest, - AgentTurnCancellationRequest, AgentTurnSettlementRequest, PermissionReply, - PermissionReplySource, PermissionRequest, PermissionRequestEvent, + AgentDialogSteerRequest, AgentDialogTurnRequest, AgentSessionCreateRequest, + AgentSessionRestoreRequest, AgentTurnCancellationRequest, AgentTurnSettlementRequest, + PermissionReply, PermissionReplySource, PermissionRequest, PermissionRequestEvent, }; use bitfun_events::{project_agentic_frontend_event, AgenticEvent}; use bitfun_runtime_ports::{ @@ -364,7 +364,6 @@ async fn run_inner(store: &DispatchStore, job_id: &str) -> Result<()> { store, job_id, &agent_runtime, - &compatibility, &job.request.session_id, &turn_id, ).await? { @@ -447,7 +446,6 @@ async fn process_mailboxes( store: &DispatchStore, job_id: &str, runtime: &bitfun_agent_runtime::sdk::AgentRuntime, - compatibility: &bitfun_core::product_runtime::CoreAgentRuntimeCompatibility, session_id: &str, turn_id: &str, ) -> Result)>> { @@ -483,15 +481,15 @@ async fn process_mailboxes( } for request in store.list_pending_append_messages(job_id)? { - compatibility - .submit_steering( - session_id.to_string(), - turn_id.to_string(), - request.content.clone(), - request.display_content.clone(), - ) + runtime + .steer_dialog_turn(AgentDialogSteerRequest { + session_id: session_id.to_string(), + turn_id: turn_id.to_string(), + content: request.content.clone(), + display_content: request.display_content.clone(), + }) .await - .map_err(anyhow::Error::msg) + .map_err(|error| anyhow!(error.into_message())) .with_context(|| { format!( "append message {} to running dispatch turn", @@ -674,6 +672,21 @@ fn terminal_outcome( mod tests { use super::*; + #[test] + fn dispatch_append_reuses_the_runtime_steering_port() { + let source = include_str!("worker.rs").replace("\r\n", "\n"); + let mailboxes = source + .split_once("async fn process_mailboxes(") + .expect("mailbox processor") + .1 + .split_once("async fn cancel_turn(") + .expect("mailbox processor boundary") + .0; + + assert!(mailboxes.contains(".steer_dialog_turn(AgentDialogSteerRequest")); + assert!(!mailboxes.contains("compatibility.submit_steering")); + } + #[test] fn terminal_events_map_to_persistent_job_states() { let completed = AgenticEvent::DialogTurnCompleted { diff --git a/src/apps/cli/src/modes/chat/commands.rs b/src/apps/cli/src/modes/chat/commands.rs index 13047de375..53881ced4b 100644 --- a/src/apps/cli/src/modes/chat/commands.rs +++ b/src/apps/cli/src/modes/chat/commands.rs @@ -2,6 +2,22 @@ fn session_update_blocks_typed_submission(pending_for_current_session: bool, inp pending_for_current_session && !input.trim().starts_with('/') } +fn steering_unsupported_reason( + draft: &crate::ui::composer::ComposerDraft, +) -> Option<&'static str> { + if draft.has_images() { + return Some( + "Images cannot steer an active turn yet. Wait for it to finish to send this draft.", + ); + } + if !draft.workspace_references.is_empty() { + return Some( + "Workspace references cannot steer an active turn yet. Wait for it to finish to send this draft.", + ); + } + None +} + fn parse_reload_target( arguments: &str, ) -> std::result::Result { @@ -1594,13 +1610,18 @@ impl ChatMode { if let Some(input) = chat_view.send_input() { return self.handle_command(&input.text, chat_view, chat_state, rt_handle); } + } else if shell_mode && !trimmed.is_empty() { + chat_view.set_status(Some( + "Currently processing. Wait for the turn to finish or interrupt it." + .to_string(), + )); } else if !trimmed.is_empty() { - chat_view.set_status(Some(if shell_mode { - "Currently processing. Wait for the turn to finish or interrupt it.".to_string() - } else { - "Currently processing. Type a /command, or use the interrupt shortcut." - .to_string() - })); + let draft = chat_view.draft_snapshot(); + if let Some(reason) = steering_unsupported_reason(&draft) { + chat_view.set_status(Some(reason.to_string())); + } else if let Some(draft) = chat_view.send_input() { + self.steer_draft_to_agent(draft, chat_view, chat_state, rt_handle); + } } return Ok(None); } @@ -1619,6 +1640,41 @@ impl ChatMode { Ok(None) } + fn steer_draft_to_agent( + &mut self, + draft: crate::ui::composer::ComposerDraft, + chat_view: &mut ChatView, + chat_state: &mut ChatState, + rt_handle: &tokio::runtime::Handle, + ) { + let agent = self.agent.clone(); + let result = tokio::task::block_in_place(|| { + rt_handle.block_on(agent.steer_current_turn( + draft.text.clone(), + Some(draft.text.clone()), + )) + }); + match result { + Ok(steering_id) => { + tracing::info!( + "Steering submitted: turn_id={:?}, steering_id={}", + chat_state.current_turn_id(), + steering_id + ); + chat_view.remember_submitted_draft(&chat_state.core_session_id, &draft); + chat_state.handle_user_steering(&steering_id, &draft.text, true); + chat_view.invalidate_lines_cache(); + let display_name = agent_display_name(&self.agent_type); + chat_view.set_status(Some(format!("{} is thinking...", display_name))); + } + Err(error) => { + tracing::error!("Failed to steer active turn: {error}"); + chat_view.set_status(Some(format!("Error: {error}"))); + chat_view.set_draft(draft); + } + } + } + fn send_shell_command( &mut self, draft: crate::ui::composer::ComposerDraft, diff --git a/src/apps/cli/src/modes/chat/run.rs b/src/apps/cli/src/modes/chat/run.rs index d6fb2a9821..b98bd6f1f6 100644 --- a/src/apps/cli/src/modes/chat/run.rs +++ b/src/apps/cli/src/modes/chat/run.rs @@ -966,6 +966,29 @@ impl ChatMode { needs_redraw = true; } + AgenticEvent::UserSteeringInjected { + turn_id, + steering_id, + display_content, + .. + } => { + if chat_state.current_turn_id() == Some(turn_id.as_str()) { + chat_state.handle_user_steering( + steering_id, + display_content, + false, + ); + chat_view.invalidate_lines_cache(); + needs_redraw = true; + } else { + tracing::debug!( + "Ignoring UserSteeringInjected for non-active turn: active={:?}, event={}", + chat_state.current_turn_id(), + turn_id + ); + } + } + AgenticEvent::ContextCompressionStarted { .. } | AgenticEvent::ContextCompressionCompleted { .. } | AgenticEvent::ContextCompressionFailed { .. } => { diff --git a/src/apps/cli/src/modes/chat/sessions.rs b/src/apps/cli/src/modes/chat/sessions.rs index 15ccb924fd..627c13a2f2 100644 --- a/src/apps/cli/src/modes/chat/sessions.rs +++ b/src/apps/cli/src/modes/chat/sessions.rs @@ -305,6 +305,7 @@ impl ChatMode { } if chat_state.is_processing { chat_state.add_system_message("Already processing, please wait.".to_string()); + chat_view.set_draft(draft); return; } diff --git a/src/apps/cli/src/modes/chat/tests.rs b/src/apps/cli/src/modes/chat/tests.rs index 865faefba8..767ef7c6ec 100644 --- a/src/apps/cli/src/modes/chat/tests.rs +++ b/src/apps/cli/src/modes/chat/tests.rs @@ -27,9 +27,10 @@ mod tests { session_command_help_note, session_delete_allowed, session_delete_feedback, session_update_allowed, session_update_blocks_typed_submission, session_update_completion_should_exit, shared_session_change_is_blocked, - terminal_event_allowed_while_local_effect_pending, CommandRoute, ExternalAgentReviewAction, - ExternalControlUiAction, ExternalSourceConflictPreferences, ExternalToolReviewAction, - HookManagementAction, SessionUpdateApplyOutcome, SHARED_TUI_CHAT_STATUS, + steering_unsupported_reason, terminal_event_allowed_while_local_effect_pending, + CommandRoute, ExternalAgentReviewAction, ExternalControlUiAction, + ExternalSourceConflictPreferences, ExternalToolReviewAction, HookManagementAction, + SessionUpdateApplyOutcome, SHARED_TUI_CHAT_STATUS, }; use crate::actions::{ action_conflict_behavior_version, ActionHandler, ActionState, ResolvedKeymap, @@ -2592,4 +2593,75 @@ mod tests { assert!(config.contains("can read and save its settings")); assert!(!config.contains("requested model is not available")); } + + #[test] + fn busy_chat_submission_steers_without_inventing_a_command_or_losing_the_draft() { + let source = include_str!("commands.rs").replace("\r\n", "\n"); + let submission = source + .split_once("fn submit_input(") + .expect("submit input") + .1 + .split_once("fn send_shell_command(") + .expect("submit input boundary") + .0; + let steering = source + .split_once("fn steer_draft_to_agent(") + .expect("steering submission") + .1 + .split_once("fn send_shell_command(") + .expect("steering submission boundary") + .0; + + assert!(submission.contains("if chat_state.is_processing")); + assert!(submission.contains("steering_unsupported_reason")); + assert!(submission.contains("self.steer_draft_to_agent")); + assert!(!submission.contains("/steer")); + assert!(steering.contains("agent.steer_current_turn")); + assert!(steering.contains("chat_state.handle_user_steering")); + assert!(steering.contains("chat_view.set_draft(draft)")); + } + + #[test] + fn active_turn_steering_accepts_text_and_rejects_rich_drafts() { + let plain = crate::ui::composer::ComposerDraft::from_text("check tests"); + assert_eq!(steering_unsupported_reason(&plain), None); + + let mut referenced = plain.clone(); + referenced.workspace_references.push( + bitfun_agent_runtime::sdk::AgentWorkspaceReference { + path: "src/lib.rs".to_string(), + kind: bitfun_agent_runtime::sdk::AgentWorkspaceReferenceKind::File, + start_line: None, + end_line: None, + source: bitfun_agent_runtime::sdk::AgentWorkspaceReferenceSourceRange { + start: 0, + end: 11, + value: "@src/lib.rs".to_string(), + }, + }, + ); + assert!(steering_unsupported_reason(&referenced) + .expect("workspace reference rejection") + .contains("Workspace references")); + + let mut imaged = plain; + imaged + .image_attachments + .push(crate::ui::composer::ComposerImageAttachment { + image: crate::ui::composer::ComposerImage::new( + "image-1", + "image.png", + "image/png", + std::sync::Arc::<[u8]>::from([1, 2, 3]), + ), + source: crate::ui::composer::ComposerSourceRange { + start: 0, + end: 9, + value: "[Image 1]".to_string(), + }, + }); + assert!(steering_unsupported_reason(&imaged) + .expect("image rejection") + .contains("Images")); + } } diff --git a/src/apps/cli/src/modes/chat/transcript.rs b/src/apps/cli/src/modes/chat/transcript.rs index 6d36c833ce..dd3f82414d 100644 --- a/src/apps/cli/src/modes/chat/transcript.rs +++ b/src/apps/cli/src/modes/chat/transcript.rs @@ -93,6 +93,14 @@ pub(super) fn render_session_markdown( { push_block(&mut body, &format!("_Thinking:_\n\n{content}")); } + FlowItem::UserSteering { + content, + is_pending, + .. + } if !content.is_empty() => { + let status = if *is_pending { " (pending)" } else { "" }; + push_block(&mut body, &format!("> **You steered{status}:** {content}")); + } FlowItem::Tool { tool_state } => { let paired = tool_results.get(tool_state.tool_id.as_str()).copied(); push_block( @@ -100,7 +108,9 @@ pub(super) fn render_session_markdown( &render_tool(tool_state, paired, options.include_tool_details), ); } - FlowItem::Text { .. } | FlowItem::Thinking { .. } => {} + FlowItem::Text { .. } + | FlowItem::Thinking { .. } + | FlowItem::UserSteering { .. } => {} } } if body.is_empty() { @@ -129,7 +139,9 @@ fn collect_tool_results(state: &ChatState) -> HashMap<&str, &ToolDisplayState> { .flat_map(|message| message.flow_items.iter()) .filter_map(|item| match item { FlowItem::Tool { tool_state } => Some((tool_state.tool_id.as_str(), tool_state)), - FlowItem::Text { .. } | FlowItem::Thinking { .. } => None, + FlowItem::Text { .. } | FlowItem::Thinking { .. } | FlowItem::UserSteering { .. } => { + None + } }) .collect() } diff --git a/src/apps/cli/src/shared_runtime.rs b/src/apps/cli/src/shared_runtime.rs index 65b2b18ba5..88e2d81388 100644 --- a/src/apps/cli/src/shared_runtime.rs +++ b/src/apps/cli/src/shared_runtime.rs @@ -405,6 +405,22 @@ impl RuntimeIpcRequestHandler for SharedRuntimeHandler { turn_id, }) } + RuntimeIpcOperation::SteerTurn { request } => self + .runtime + .steer_dialog_turn(request) + .await + .map(|outcome| match outcome { + bitfun_agent_runtime::sdk::DialogSteerOutcome::Buffered { + session_id, + turn_id, + steering_id, + } => RuntimeIpcOperationResult::TurnSteered { + session_id, + turn_id, + steering_id, + }, + }) + .map_err(runtime_ipc_error), RuntimeIpcOperation::RunUserShellCommand { request } => self .runtime .run_user_shell_command(request) @@ -1156,6 +1172,23 @@ mod tests { use std::time::Duration; use tokio::sync::{watch, Notify}; + #[test] + fn shared_handler_steering_delegates_to_the_runtime_sdk() { + let source = include_str!("shared_runtime.rs").replace("\r\n", "\n"); + let execute = source + .split_once("impl RuntimeIpcRequestHandler for SharedRuntimeHandler") + .expect("shared handler") + .1 + .split_once("fn subscribe_events(") + .expect("shared handler boundary") + .0; + + assert!(execute.contains("RuntimeIpcOperation::SteerTurn { request }")); + assert!(execute.contains(".steer_dialog_turn(request)")); + assert!(execute.contains("RuntimeIpcOperationResult::TurnSteered")); + assert!(!execute.contains("submit_steering")); + } + #[derive(Default)] struct RecordingSessionPort { delete_requests: Mutex>, diff --git a/src/apps/cli/src/ui/chat/render.rs b/src/apps/cli/src/ui/chat/render.rs index fb61bd27f3..3926c82b58 100644 --- a/src/apps/cli/src/ui/chat/render.rs +++ b/src/apps/cli/src/ui/chat/render.rs @@ -704,6 +704,48 @@ impl ChatView { plain_lines.push(String::new()); } + FlowItem::UserSteering { + content, + is_pending, + .. + } => { + close_user_bubble( + &mut items, + &mut plain_lines, + &mut user_bubble_open, + user_bg_style, + user_border_style, + ); + let label = if *is_pending { + "You steered (pending)" + } else { + "You steered" + }; + let prefix = format!(" {label}: "); + let content_width = available_width + .saturating_sub(prefix.width().min(u16::MAX as usize) as u16) + as usize; + let mut first_line = true; + for line in content.lines() { + for wrapped in wrap_hard_display_width(line, content_width.max(1)) { + let line_prefix = if first_line { + first_line = false; + prefix.clone() + } else { + " ".to_string() + }; + plain_lines.push(format!("{line_prefix}{wrapped}")); + items.push(ListItem::new(Line::from(vec![ + Span::styled( + line_prefix, + self.theme.style(StyleKind::Primary), + ), + Span::raw(wrapped), + ]))); + } + } + } + FlowItem::Tool { tool_state } => { close_user_bubble( &mut items, diff --git a/src/apps/desktop/src/api/agentic_api.rs b/src/apps/desktop/src/api/agentic_api.rs index ee4a46bf82..0408ec030c 100644 --- a/src/apps/desktop/src/api/agentic_api.rs +++ b/src/apps/desktop/src/api/agentic_api.rs @@ -16,10 +16,10 @@ use crate::runtime::{ use crate::startup_trace::DesktopStartupTrace; use bitfun_agent_runtime::deep_review::sanitize_focused_review_public_metadata; use bitfun_agent_runtime::sdk::{ - AgentDialogTurnExecution, AgentDialogTurnRequest, AgentInputAttachment, - AgentSessionCreateResult, AgentSessionModelUpdateRequest, AgentSubmissionSource, - AgentTurnCancellationRequest, PermissionAuditRecord, PermissionGrant, PermissionGrantKey, - PermissionReply, PermissionRequest, + AgentDialogSteerRequest, AgentDialogTurnExecution, AgentDialogTurnRequest, + AgentInputAttachment, AgentSessionCreateResult, AgentSessionModelUpdateRequest, + AgentSubmissionSource, AgentTurnCancellationRequest, DialogSteerOutcome, PermissionAuditRecord, + PermissionGrant, PermissionGrantKey, PermissionReply, PermissionRequest, }; use bitfun_core::agentic::agents::AgentSource; use bitfun_core::agentic::coordination::{ @@ -2505,7 +2505,7 @@ pub async fn cancel_dialog_turn( #[tauri::command] pub async fn steer_dialog_turn( - scheduler: State<'_, Arc>, + runtime: State<'_, DesktopRuntimeContext>, request: SteerDialogTurnRequest, ) -> Result { let SteerDialogTurnRequest { @@ -2520,15 +2520,19 @@ pub async fn steer_dialog_turn( return Err("Steering content cannot be empty".to_string()); } - let outcome = scheduler - .submit_steering(session_id, dialog_turn_id, content, display_content) + let outcome = runtime + .agent_runtime() + .steer_dialog_turn(AgentDialogSteerRequest { + session_id, + turn_id: dialog_turn_id, + content, + display_content, + }) .await - .map_err(|e| format!("Failed to steer dialog turn: {}", e))?; + .map_err(|error| format!("Failed to steer dialog turn: {}", error.into_message()))?; let steering_id = match outcome { - bitfun_core::agentic::coordination::DialogSteerOutcome::Buffered { - steering_id, .. - } => steering_id, + DialogSteerOutcome::Buffered { steering_id, .. } => steering_id, }; Ok(SteerDialogTurnResponse { @@ -3248,6 +3252,24 @@ mod tests { use bitfun_product_domains::tool_permissions::{PermissionEffect, PermissionRule}; use serde_json::json; + #[test] + fn desktop_steering_uses_the_same_agent_runtime_port_as_other_surfaces() { + let source = include_str!("agentic_api.rs").replace("\r\n", "\n"); + let steering = source + .split_once("pub async fn steer_dialog_turn(") + .expect("steering command") + .1 + .split_once("pub async fn control_deep_review_queue(") + .expect("steering command boundary") + .0; + + assert!(steering.contains("State<'_, DesktopRuntimeContext>")); + assert!(steering.contains(".agent_runtime()")); + assert!(steering.contains(".steer_dialog_turn(AgentDialogSteerRequest")); + assert!(!steering.contains("State<'_, Arc>")); + assert!(!steering.contains(".submit_steering(")); + } + #[test] fn unknown_title_outcomes_reach_the_frontend_with_a_stable_code() { assert_eq!( diff --git a/src/crates/adapters/agent-runtime-ipc/AGENTS-CN.md b/src/crates/adapters/agent-runtime-ipc/AGENTS-CN.md index fe35071ef8..3ea157886d 100644 --- a/src/crates/adapters/agent-runtime-ipc/AGENTS-CN.md +++ b/src/crates/adapters/agent-runtime-ipc/AGENTS-CN.md @@ -15,7 +15,7 @@ ## 边界 - 只导出 CLI adapter 实际使用的 workspace-private API，且 crate 不得发布，也不得把 wire 作为 SDK 合同。 -- 封闭 operation 范围为 Health、Session list/create/restore/delete/fork（restore/fork 结果包含 transcript）、当前 Session rename、Agent mode/model update、手动 context compaction、Session undo/redo、current-controller 限定的只读工作区引用搜索/持久化引用读取，以及不取得 Session lease 的 Runtime 绑定工作区只读 diff；此外还包括声明式上下文 reload、Turn submit/用户显式 Shell 执行/cancel、pending/respond Permission 和 UserInput answers。delete 只允许作用于未被任何 Client 控制的空闲 Session。fork 要求当前 controller 且 Session 空闲：可以复制到最新持久化 Turn，也可以停在显式选中 Turn 之前；只有包含新 Session 与 transcript 的成功结果完成编码后，Server 才能把连接 lease 从源 Session 原子切换到 fork。手动 compaction 要求当前 controller 且 Session 空闲；Client 在准入前提供精确 Turn ID，使超时或断连 cleanup 可以取消同一个 owned task；Core 开始原子 context commit 后，晚到取消不能暴露错误的空闲状态。用户显式 Shell 执行同样要求当前 controller、Session 空闲和调用方提供的 Turn ID；它只委托给窄 Runtime port，并复用正常 ToolPipeline、权限、工作区路由、持久化与取消 owner，不是通用 Tool 或进程执行 wire。undo/redo 要求当前 controller，但可在活动 Turn 中进入，因为取消、drain 与回退写入顺序由 Core 统一负责；成功结果携带权威 transcript，并清除连接侧活动 Turn 投影。该能力只支持本地工作区，不暴露通用 checkpoint 协议。上下文 reload 可在活动 Turn 中执行，不改写该 Turn，并通过缓存保护保证下一条消息重新读取已失效的 instructions。断连 cleanup 属于内部生命周期，不是 detach operation。模型目录和默认值仍是 wire 之外的产品配置；禁止顺带加入 archive、replay、observer、通用 controller transfer、Tool/MCP/Hook 管理或其他产品配置。 +- 封闭 operation 范围为 Health、Session list/create/restore/delete/fork（restore/fork 结果包含 transcript）、当前 Session rename、Agent mode/model update、手动 context compaction、Session undo/redo、current-controller 限定的只读工作区引用搜索/持久化引用读取，以及不取得 Session lease 的 Runtime 绑定工作区只读 diff；此外还包括声明式上下文 reload、Turn submit/steer、用户显式 Shell 执行/cancel、pending/respond Permission 和 UserInput answers。delete 只允许作用于未被任何 Client 控制的空闲 Session。fork 要求当前 controller 且 Session 空闲：可以复制到最新持久化 Turn，也可以停在显式选中 Turn 之前；只有包含新 Session 与 transcript 的成功结果完成编码后，Server 才能把连接 lease 从源 Session 原子切换到 fork。手动 compaction 要求当前 controller 且 Session 空闲；Client 在准入前提供精确 Turn ID，使超时或断连 cleanup 可以取消同一个 owned task；Core 开始原子 context commit 后，晚到取消不能暴露错误的空闲状态。用户显式 Shell 执行同样要求当前 controller、Session 空闲和调用方提供的 Turn ID；它只委托给窄 Runtime port，并复用正常 ToolPipeline、权限、工作区路由、持久化与取消 owner，不是通用 Tool 或进程执行 wire。steer 要求当前 controller、活动 Turn 以及调用方提供的精确 Session/Turn ID；它只委托给共享 Runtime owner，拒绝过期投影，不创建第二个 Turn 或 queue owner。undo/redo 要求当前 controller，但可在活动 Turn 中进入，因为取消、drain 与回退写入顺序由 Core 统一负责；成功结果携带权威 transcript，并清除连接侧活动 Turn 投影。该能力只支持本地工作区，不暴露通用 checkpoint 协议。上下文 reload 可在活动 Turn 中执行，不改写该 Turn，并通过缓存保护保证下一条消息重新读取已失效的 instructions。断连 cleanup 属于内部生命周期，不是 detach operation。模型目录和默认值仍是 wire 之外的产品配置；禁止顺带加入 archive、replay、observer、通用 controller transfer、Tool/MCP/Hook 管理或其他产品配置。 - 可以复用稳定 Event、Product Domain 和 Runtime Port DTO。禁止依赖 `bitfun-core`、Agent Runtime 实现、SDK Host、services、Tauri、terminal、tool runtime 或远程 transport。 - 只使用 Windows Named Pipe 或 Unix Domain Socket；禁止 TCP、HTTP、WebSocket、浏览器访问或远程 fallback。 - 这是本机同用户隔离，不是沙箱。未来产品 composition 必须提供当前用户私有 runtime 目录。 diff --git a/src/crates/adapters/agent-runtime-ipc/AGENTS.md b/src/crates/adapters/agent-runtime-ipc/AGENTS.md index b55d5e5047..6c2a36fa78 100644 --- a/src/crates/adapters/agent-runtime-ipc/AGENTS.md +++ b/src/crates/adapters/agent-runtime-ipc/AGENTS.md @@ -23,10 +23,11 @@ session controller leases, event delivery, connection bounds, and cleanup. It is - Export only the exact workspace-private API needed by the CLI adapter. Do not publish this crate or expose its wire as an SDK contract. - The closed operation budget is Health, Session list/create/restore/delete/fork (including transcript on restore/fork), current-Session rename, Agent mode/model update, manual context compaction, Session undo/redo, current-controller read-only workspace-reference search/persisted-reference lookup, and a read-only diff of the Runtime-bound workspace that does not acquire a Session lease, - declarative context reload, Turn submit/user-authored Shell execution/cancel, pending/respond Permission, and UserInput answers. Delete is limited to an idle Session not controlled by any client. + declarative context reload, Turn submit/steer/user-authored Shell execution/cancel, pending/respond Permission, and UserInput answers. Delete is limited to an idle Session not controlled by any client. Fork is a current-controller, idle-only operation. It either copies through the latest persisted Turn or stops immediately before an explicitly selected Turn. The encoded success result carries the authoritative new Session and transcript; only then may the server atomically switch the connection lease from the source Session to the fork. Manual compaction is a current-controller, idle-only Turn operation. The client supplies its exact Turn ID before admission so timeout or disconnect cleanup can cancel the same owned task; once Core begins the atomic context commit, a late cancellation does not expose a false idle state. User-authored Shell execution is a current-controller, idle-only Turn operation with a caller-supplied Turn ID. It delegates to the narrow Runtime port and normal ToolPipeline, permission, workspace-routing, persistence, and cancellation owners; it is not a generic Tool or process-execution wire. + Steering is a current-controller, active-Turn-only operation with caller-supplied Session and Turn IDs. It delegates to the shared Runtime owner, rejects stale projections, and does not create a second Turn or queue owner. Context reload may run during an active Turn, does not rewrite that Turn, and guards the cache so the next message reads invalidated instructions. Undo/redo is a current-controller operation that may enter during an active Turn because Core owns cancel-and-drain before mutation. Its success response carries the authoritative transcript and clears the connection's active-Turn projection. It is local-workspace only and does not expose a generic checkpoint protocol. Disconnect cleanup is internal lifecycle, not a detach operation. diff --git a/src/crates/adapters/agent-runtime-ipc/src/operation.rs b/src/crates/adapters/agent-runtime-ipc/src/operation.rs index 60068f0ec6..1d41dfffd8 100644 --- a/src/crates/adapters/agent-runtime-ipc/src/operation.rs +++ b/src/crates/adapters/agent-runtime-ipc/src/operation.rs @@ -1,12 +1,13 @@ use bitfun_product_domains::tool_permissions::{PermissionReply, PermissionRequest}; use bitfun_runtime_ports::{ - AgentContextReloadRequest, AgentDialogTurnRequest, AgentMessageWorkspaceReferencesRequest, - AgentSessionCompactionRequest, AgentSessionCreateRequest, AgentSessionCreateResult, - AgentSessionListRequest, AgentSessionModeUpdateRequest, AgentSessionModelUpdateRequest, - AgentSessionRevertRequest, AgentSessionRevertResult, AgentSessionSummary, - AgentTurnCancellationRequest, AgentTurnCancellationResult, AgentUserShellCommandRequest, - AgentWorkspaceReference, AgentWorkspaceReferenceSearchRequest, - AgentWorkspaceReferenceSearchResult, SessionTranscript, WorkspaceDiffSnapshot, + AgentContextReloadRequest, AgentDialogSteerRequest, AgentDialogTurnRequest, + AgentMessageWorkspaceReferencesRequest, AgentSessionCompactionRequest, + AgentSessionCreateRequest, AgentSessionCreateResult, AgentSessionListRequest, + AgentSessionModeUpdateRequest, AgentSessionModelUpdateRequest, AgentSessionRevertRequest, + AgentSessionRevertResult, AgentSessionSummary, AgentTurnCancellationRequest, + AgentTurnCancellationResult, AgentUserShellCommandRequest, AgentWorkspaceReference, + AgentWorkspaceReferenceSearchRequest, AgentWorkspaceReferenceSearchResult, SessionTranscript, + WorkspaceDiffSnapshot, }; use serde::{Deserialize, Serialize}; @@ -97,6 +98,9 @@ pub enum RuntimeIpcOperation { SubmitTurn { request: AgentDialogTurnRequest, }, + SteerTurn { + request: AgentDialogSteerRequest, + }, RunUserShellCommand { request: AgentUserShellCommandRequest, }, @@ -132,6 +136,7 @@ impl RuntimeIpcOperation { Self::SearchWorkspaceReferences { request } => Some(&request.session_id), Self::WorkspaceReferencesForMessage { request } => Some(&request.session_id), Self::SubmitTurn { request } => Some(&request.session_id), + Self::SteerTurn { request } => Some(&request.session_id), Self::RunUserShellCommand { request } => Some(&request.session_id), Self::CancelTurn { request } => Some(&request.session_id), Self::PendingPermissions { session_id } @@ -175,6 +180,7 @@ impl RuntimeIpcOperation { Self::ReloadSessionContext { .. } | Self::UndoSession { .. } | Self::RedoSession { .. } + | Self::SteerTurn { .. } | Self::CancelTurn { .. } | Self::RespondPermission { .. } | Self::SubmitUserAnswers { .. } => { @@ -257,6 +263,11 @@ pub enum RuntimeIpcOperationResult { session_id: String, turn_id: String, }, + TurnSteered { + session_id: String, + turn_id: String, + steering_id: String, + }, TurnCancelled { cancellation: AgentTurnCancellationResult, }, @@ -277,7 +288,9 @@ pub enum RuntimeIpcOperationResult { #[cfg(test)] mod tests { use super::{RuntimeIpcOperation, RuntimeIpcSessionRequirement, RuntimeSessionRestoreRequest}; - use bitfun_runtime_ports::{AgentContextReloadRequest, AgentContextReloadTarget}; + use bitfun_runtime_ports::{ + AgentContextReloadRequest, AgentContextReloadTarget, AgentDialogSteerRequest, + }; #[test] fn delete_rules_are_fail_closed_for_shared_session_selection() { @@ -314,6 +327,28 @@ mod tests { assert!(rules.side_effecting); } + #[test] + fn steer_rules_require_the_current_controller_but_allow_an_active_turn() { + let operation = RuntimeIpcOperation::SteerTurn { + request: AgentDialogSteerRequest { + session_id: "session-1".to_string(), + turn_id: "turn-1".to_string(), + content: "check tests".to_string(), + display_content: None, + }, + }; + let rules = operation.rules(); + + assert_eq!(operation.session_id(), Some("session-1")); + assert_eq!( + rules.session_requirement, + RuntimeIpcSessionRequirement::CurrentController + ); + assert!(!rules.requires_idle); + assert!(!rules.serializes_session_selection); + assert!(rules.side_effecting); + } + #[test] fn restore_and_pending_permission_rules_preserve_existing_behavior() { let restore = RuntimeIpcOperation::RestoreSession { diff --git a/src/crates/adapters/agent-runtime-ipc/src/protocol.rs b/src/crates/adapters/agent-runtime-ipc/src/protocol.rs index 140da42209..26813e9af1 100644 --- a/src/crates/adapters/agent-runtime-ipc/src/protocol.rs +++ b/src/crates/adapters/agent-runtime-ipc/src/protocol.rs @@ -5,7 +5,7 @@ use crate::{RuntimeIpcOperation, RuntimeIpcOperationResult}; use bitfun_events::AgenticEventEnvelope; use bitfun_product_domains::tool_permissions::PermissionRequestEvent; -pub const PROTOCOL_VERSION: u32 = 12; +pub const PROTOCOL_VERSION: u32 = 13; #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(tag = "type", rename_all = "snake_case", deny_unknown_fields)] diff --git a/src/crates/adapters/agent-runtime-ipc/src/server.rs b/src/crates/adapters/agent-runtime-ipc/src/server.rs index 4a1183e447..c9ce4d9062 100644 --- a/src/crates/adapters/agent-runtime-ipc/src/server.rs +++ b/src/crates/adapters/agent-runtime-ipc/src/server.rs @@ -424,6 +424,19 @@ async fn run_initialized_connection( .await?; continue; } + if let RuntimeIpcOperation::SteerTurn { request } = &operation { + if active_turn_id.as_deref() != Some(request.turn_id.as_str()) { + send_error( + stream, + config.request_timeout, + Some(request_id), + RuntimeIpcErrorCode::SessionInUse, + "steering requires the connection's exact active turn", + ) + .await?; + continue; + } + } // Serialize attachment so a newly visible Session cannot be claimed // before its generated ID returns to the creating connection, or @@ -475,6 +488,12 @@ async fn run_initialized_connection( } _ => None, }; + let steering_target = match &operation { + RuntimeIpcOperation::SteerTurn { request } => { + Some((request.session_id.clone(), request.turn_id.clone())) + } + _ => None, + }; let side_effecting = rules.side_effecting; let result = tokio::time::timeout(config.request_timeout, handler.execute(operation)).await; @@ -635,6 +654,27 @@ async fn run_initialized_connection( return Err(RuntimeIpcServerError::Disconnected); } } + if let Some((expected_session_id, expected_turn_id)) = steering_target.as_ref() { + let identity_matches = matches!( + result, + RuntimeIpcOperationResult::TurnSteered { + session_id, + turn_id, + .. + } if session_id == expected_session_id && turn_id == expected_turn_id + ); + if !identity_matches { + send_error( + stream, + config.request_timeout, + Some(request_id), + RuntimeIpcErrorCode::Internal, + "runtime returned an invalid result for the steering operation", + ) + .await?; + return Err(RuntimeIpcServerError::Disconnected); + } + } if let Err(error) = timeout_write_serialized(config.request_timeout, stream, &response_bytes).await { diff --git a/src/crates/adapters/agent-runtime-ipc/src/tests/protocol_contracts.rs b/src/crates/adapters/agent-runtime-ipc/src/tests/protocol_contracts.rs index f068440ddc..35e259348c 100644 --- a/src/crates/adapters/agent-runtime-ipc/src/tests/protocol_contracts.rs +++ b/src/crates/adapters/agent-runtime-ipc/src/tests/protocol_contracts.rs @@ -8,8 +8,8 @@ use crate::{ use bitfun_product_domains::tool_permissions::PermissionReply; use bitfun_runtime_ports::{ - AgentContextReloadRequest, AgentContextReloadTarget, AgentDialogTurnRequest, - AgentMessageWorkspaceReferencesRequest, AgentSessionCompactionRequest, + AgentContextReloadRequest, AgentContextReloadTarget, AgentDialogSteerRequest, + AgentDialogTurnRequest, AgentMessageWorkspaceReferencesRequest, AgentSessionCompactionRequest, AgentSessionModeUpdateRequest, AgentSessionModelUpdateRequest, AgentSessionRevertRequest, AgentSubmissionSource, AgentWorkspaceReferenceSearchRequest, DialogSubmissionPolicy, WorkspaceDiffContent, WorkspaceDiffFile, WorkspaceDiffFileStatus, WorkspaceDiffSnapshot, @@ -71,6 +71,42 @@ fn protocol_round_trips_reviewed_permission_and_user_input_operations() { } } +#[test] +fn protocol_round_trips_exact_turn_steering_without_replacing_turn_admission() { + assert_eq!(PROTOCOL_VERSION, 13); + let operation = RuntimeIpcOperation::SteerTurn { + request: AgentDialogSteerRequest { + session_id: "session-1".to_string(), + turn_id: "turn-1".to_string(), + content: "check tests".to_string(), + display_content: Some("Check tests".to_string()), + }, + }; + let result = RuntimeIpcOperationResult::TurnSteered { + session_id: "session-1".to_string(), + turn_id: "turn-1".to_string(), + steering_id: "steer-1".to_string(), + }; + + let operation_json = serde_json::to_value(&operation).expect("serialize steer operation"); + let result_json = serde_json::to_value(&result).expect("serialize steer result"); + + assert_eq!(operation_json["operation"], "steer_turn"); + assert_eq!(operation_json["request"]["turnId"], "turn-1"); + assert_eq!(result_json["result"], "turn_steered"); + assert_eq!(result_json["steeringId"], "steer-1"); + assert_eq!( + serde_json::from_value::(operation_json) + .expect("deserialize steer operation"), + operation + ); + assert_eq!( + serde_json::from_value::(result_json) + .expect("deserialize steer result"), + result + ); +} + #[test] fn protocol_round_trips_read_only_workspace_reference_operations() { let operations = vec![ @@ -108,7 +144,7 @@ fn protocol_round_trips_read_only_workspace_reference_operations() { #[test] fn protocol_round_trips_workspace_diff_as_a_read_only_workspace_operation() { - assert_eq!(PROTOCOL_VERSION, 12); + assert_eq!(PROTOCOL_VERSION, 13); let operation = RuntimeIpcOperation::WorkspaceDiff; let encoded = serde_json::to_value(&operation).expect("serialize workspace diff operation"); @@ -229,7 +265,7 @@ fn protocol_round_trips_the_reviewed_session_model_operation() { #[test] fn protocol_round_trips_the_current_session_rename_operation() { - assert_eq!(PROTOCOL_VERSION, 12); + assert_eq!(PROTOCOL_VERSION, 13); let operation = RuntimeIpcOperation::RenameSession { request: RuntimeSessionRenameRequest { diff --git a/src/crates/adapters/agent-runtime-ipc/src/tests/shared_controller.rs b/src/crates/adapters/agent-runtime-ipc/src/tests/shared_controller.rs index 9c92ec9e98..b7ba071603 100644 --- a/src/crates/adapters/agent-runtime-ipc/src/tests/shared_controller.rs +++ b/src/crates/adapters/agent-runtime-ipc/src/tests/shared_controller.rs @@ -90,6 +90,7 @@ struct FakeHandler { rename_delay: Option, delete_delay: Option, submit_delay: Option, + invalid_steer_result: bool, settle_cancel: bool, events: broadcast::Sender, available: watch::Sender, @@ -114,6 +115,7 @@ impl Default for FakeHandler { rename_delay: None, delete_delay: None, submit_delay: None, + invalid_steer_result: false, settle_cancel: true, events, available, @@ -223,6 +225,16 @@ impl RuntimeIpcRequestHandler for FakeHandler { turn_id: request.turn_id.expect("test turn id"), }) } + RuntimeIpcOperation::SteerTurn { request } => { + if self.invalid_steer_result { + return Ok(RuntimeIpcOperationResult::Unit); + } + Ok(RuntimeIpcOperationResult::TurnSteered { + session_id: request.session_id, + turn_id: request.turn_id, + steering_id: "steer-fixture".to_string(), + }) + } RuntimeIpcOperation::RunUserShellCommand { request } => { if let Some(delay) = self.submit_delay { tokio::time::sleep(delay).await; @@ -556,6 +568,17 @@ fn submit_operation(workspace: &Path, session_id: &str, turn_id: &str) -> Runtim } } +fn steer_operation(session_id: &str, turn_id: &str) -> RuntimeIpcOperation { + RuntimeIpcOperation::SteerTurn { + request: bitfun_runtime_ports::AgentDialogSteerRequest { + session_id: session_id.to_string(), + turn_id: turn_id.to_string(), + content: "check tests".to_string(), + display_content: None, + }, + } +} + fn shell_operation(session_id: &str, turn_id: &str) -> RuntimeIpcOperation { RuntimeIpcOperation::RunUserShellCommand { request: AgentUserShellCommandRequest { @@ -885,6 +908,108 @@ async fn one_connection_rejects_a_second_turn_until_the_first_finishes() { server.finish().await; } +#[tokio::test] +async fn steering_requires_and_preserves_the_connections_exact_active_turn() { + let handler = Arc::new(FakeHandler::default()); + let server = TestServer::start(server_config(), handler.clone()).await; + let mut client = server.connect("steering-controller").await; + expect_response( + &mut client, + 2, + restore_operation(server.workspace.path(), "session-a"), + ) + .await; + expect_response( + &mut client, + 3, + submit_operation(server.workspace.path(), "session-a", "turn-a"), + ) + .await; + expect_response(&mut client, 4, steer_operation("session-a", "turn-a")).await; + expect_error( + &mut client, + 5, + steer_operation("session-a", "turn-b"), + RuntimeIpcErrorCode::SessionInUse, + ) + .await; + expect_error( + &mut client, + 6, + submit_operation(server.workspace.path(), "session-a", "turn-b"), + RuntimeIpcErrorCode::SessionInUse, + ) + .await; + + let calls = handler.calls.lock().unwrap().clone(); + assert_eq!( + calls + .iter() + .filter(|operation| matches!(operation, RuntimeIpcOperation::SteerTurn { .. })) + .count(), + 1, + "a mismatched steer must be rejected before reaching the runtime" + ); + + drop(client); + wait_for_calls(&handler, |calls| { + calls.iter().any(|call| { + matches!( + call, + RuntimeIpcOperation::CancelTurn { request } + if request.session_id == "session-a" + && request.turn_id.as_deref() == Some("turn-a") + ) + }) + }) + .await; + server.finish().await; +} + +#[tokio::test] +async fn steering_rejects_an_invalid_runtime_result_and_closes_the_connection() { + let handler = Arc::new(FakeHandler { + invalid_steer_result: true, + ..FakeHandler::default() + }); + let server = TestServer::start(server_config(), handler.clone()).await; + let mut client = server.connect("invalid-steering-result").await; + expect_response( + &mut client, + 2, + restore_operation(server.workspace.path(), "session-a"), + ) + .await; + expect_response( + &mut client, + 3, + submit_operation(server.workspace.path(), "session-a", "turn-a"), + ) + .await; + expect_error( + &mut client, + 4, + steer_operation("session-a", "turn-a"), + RuntimeIpcErrorCode::Internal, + ) + .await; + + assert!(read_frame(&mut client).await.is_err()); + wait_for_calls(&handler, |calls| { + calls.iter().any(|call| { + matches!( + call, + RuntimeIpcOperation::CancelTurn { request } + if request.session_id == "session-a" + && request.turn_id.as_deref() == Some("turn-a") + ) + }) + }) + .await; + drop(client); + server.finish().await; +} + #[tokio::test] async fn manual_compaction_owns_the_supplied_turn_until_disconnect_cancels_it() { let handler = Arc::new(FakeHandler::default()); diff --git a/src/crates/assembly/core/src/agentic/coordination/scheduler.rs b/src/crates/assembly/core/src/agentic/coordination/scheduler.rs index 1354bc8786..58995c37fc 100644 --- a/src/crates/assembly/core/src/agentic/coordination/scheduler.rs +++ b/src/crates/assembly/core/src/agentic/coordination/scheduler.rs @@ -57,12 +57,12 @@ use bitfun_agent_runtime::scheduler::{ }; use bitfun_runtime_ports::{ resolve_dialog_submit_queue_action, AgentBackgroundResultRequest, AgentDialogPrependedReminder, - AgentDialogTurnExecution, AgentDialogTurnPort, AgentDialogTurnRequest, AgentInputAttachment, - AgentLifecycleDeliveryPort, AgentThreadGoalDeliveryKind, AgentThreadGoalDeliveryRequest, - AgentTurnCancellationPort, AgentTurnCancellationRequest, AgentTurnCancellationResult, - DialogSessionStateFact, DialogSubmitQueueAction, DialogSubmitQueueFacts, PortError, - PortErrorKind, PortResult, RoundInjection, RoundInjectionKind, SessionStoragePathRequest, - SessionStorePort, + AgentDialogSteerRequest, AgentDialogTurnExecution, AgentDialogTurnPort, AgentDialogTurnRequest, + AgentInputAttachment, AgentLifecycleDeliveryPort, AgentThreadGoalDeliveryKind, + AgentThreadGoalDeliveryRequest, AgentTurnCancellationPort, AgentTurnCancellationRequest, + AgentTurnCancellationResult, DialogSessionStateFact, DialogSubmitQueueAction, + DialogSubmitQueueFacts, PortError, PortErrorKind, PortResult, RoundInjection, + RoundInjectionKind, SessionStoragePathRequest, SessionStorePort, }; pub use bitfun_runtime_ports::{ AgentSessionReplyRoute, DialogQueuePriority, DialogSteerOutcome, DialogSubmissionPolicy, @@ -451,14 +451,19 @@ impl DialogScheduler { /// can inject it at the next model-round boundary. Errors: /// /// - Session is not currently `Processing` the requested `turn_id` (the targeted turn - /// already finished or never existed). Caller should fall back to `submit`. - pub async fn submit_steering( + /// already finished or never existed). Callers must preserve the user's input so it + /// can be submitted explicitly after authoritative state is observed. + async fn buffer_steering( &self, session_id: String, turn_id: String, content: String, display_content: Option, ) -> Result { + if content.trim().is_empty() { + return Err("Steering content cannot be empty".to_string()); + } + let _operation_guard = self.lock_session_operation(&session_id).await; let active_turn_id = match self .session_manager .get_session(&session_id) @@ -466,7 +471,12 @@ impl DialogScheduler { { Some(SessionState::Processing { current_turn_id, .. - }) => Some(current_turn_id), + }) if self + .active_turns + .matches_turn(&session_id, ¤t_turn_id) => + { + Some(current_turn_id) + } _ => None, }; @@ -482,7 +492,7 @@ impl DialogScheduler { ) { DialogSteeringAction::Reject { error } => { warn!( - "submit_steering rejected: target turn is not running: session_id={}, turn_id={}", + "Steering rejected: target turn is not running: session_id={}, turn_id={}", session_id, turn_id ); Err(error) @@ -2325,6 +2335,31 @@ impl AgentDialogTurnPort for DialogScheduler { self.submit_agent_dialog_turn_with_busy_policy(request, false) .await } + + async fn steer_dialog_turn( + &self, + request: AgentDialogSteerRequest, + ) -> PortResult { + let empty_content = request.content.trim().is_empty(); + DialogScheduler::buffer_steering( + self, + request.session_id, + request.turn_id, + request.content, + request.display_content, + ) + .await + .map_err(|error| { + PortError::new( + if empty_content { + PortErrorKind::InvalidRequest + } else { + PortErrorKind::SessionInUse + }, + error, + ) + }) + } } #[async_trait::async_trait] @@ -3406,6 +3441,116 @@ mod tests { ) } + async fn mark_session_processing( + session_manager: &SessionManager, + root: &tempfile::TempDir, + session_id: &str, + turn_id: &str, + ) { + let workspace = root.path().join(format!("workspace-{session_id}")); + std::fs::create_dir_all(&workspace).expect("workspace"); + session_manager + .create_session_with_id( + Some(session_id.to_string()), + "Steering".to_string(), + "agentic".to_string(), + SessionConfig { + workspace_path: Some(workspace.to_string_lossy().into_owned()), + ..Default::default() + }, + ) + .await + .expect("create session"); + session_manager + .update_session_state( + session_id, + SessionState::Processing { + current_turn_id: turn_id.to_string(), + phase: ProcessingPhase::Thinking, + }, + ) + .await + .expect("mark turn active"); + } + + #[tokio::test] + async fn steering_rejects_stale_processing_state_without_authoritative_active_turn() { + let (scheduler, session_manager, _, root) = test_scheduler(); + let session_id = "stale-steering-session"; + let turn_id = "stale-turn"; + mark_session_processing(&session_manager, &root, session_id, turn_id).await; + + let error = scheduler + .buffer_steering( + session_id.to_string(), + turn_id.to_string(), + "check tests".to_string(), + None, + ) + .await + .expect_err("stale processing state must not accept steering"); + + assert!(error.contains("no longer running"), "{error}"); + assert!(scheduler + .round_injection_monitor() + .take_pending(session_id, turn_id) + .is_empty()); + } + + #[tokio::test] + async fn steering_rejects_empty_content_as_an_invalid_request() { + let (scheduler, _, _, _) = test_scheduler(); + + let error = AgentDialogTurnPort::steer_dialog_turn( + scheduler.as_ref(), + AgentDialogSteerRequest { + session_id: "session-1".to_string(), + turn_id: "turn-1".to_string(), + content: " ".to_string(), + display_content: None, + }, + ) + .await + .expect_err("empty steering must fail"); + + assert_eq!(error.kind, PortErrorKind::InvalidRequest); + } + + #[tokio::test] + async fn steering_serializes_with_other_operations_for_the_same_session() { + let (scheduler, session_manager, _, root) = test_scheduler(); + let session_id = "locked-steering-session"; + let turn_id = "active-turn"; + mark_session_processing(&session_manager, &root, session_id, turn_id).await; + scheduler + .active_turns + .insert(session_id, desktop_active_turn(turn_id)); + + let operation_guard = scheduler.lock_session_operation(session_id).await; + let steering_scheduler = scheduler.clone(); + let steering = tokio::spawn(async move { + steering_scheduler + .buffer_steering( + session_id.to_string(), + turn_id.to_string(), + "check tests".to_string(), + None, + ) + .await + }); + tokio::task::yield_now().await; + + assert!( + !steering.is_finished(), + "steering must wait for the session operation lock" + ); + drop(operation_guard); + steering + .await + .expect("steering task") + .expect("steering outcome"); + } + #[tokio::test] async fn explicit_cancel_cannot_cross_session_by_reusing_a_turn_id() { let (scheduler, _, _, _root) = test_scheduler(); diff --git a/src/crates/assembly/core/src/product_runtime.rs b/src/crates/assembly/core/src/product_runtime.rs index fd5d4b9ff6..3d4798e523 100644 --- a/src/crates/assembly/core/src/product_runtime.rs +++ b/src/crates/assembly/core/src/product_runtime.rs @@ -31,7 +31,7 @@ use bitfun_services_core::permission_store::ProjectPermissionSqliteStore; use bitfun_services_core::session::SessionBranchBoundary; use crate::agentic::coordination::{ - ConversationCoordinator, DialogScheduler, DialogSteerOutcome, SessionMaintenancePermit, + ConversationCoordinator, DialogScheduler, SessionMaintenancePermit, }; use crate::agentic::core::Session; use crate::agentic::events::EventQueue; @@ -600,22 +600,6 @@ impl CoreAgentRuntimeCompatibility { } } - /// Buffer a user steering message into a currently running turn. - /// - /// Detached dispatch and compatibility hosts use this narrow facade so - /// they do not reach through the public Runtime SDK into scheduler state. - pub async fn submit_steering( - &self, - session_id: String, - turn_id: String, - content: String, - display_content: Option, - ) -> Result { - self.scheduler - .submit_steering(session_id, turn_id, content, display_content) - .await - } - /// Start a manual context compaction as a caller-identified turn. /// /// Detached dispatch supplies its own turn id so the compaction's diff --git a/src/crates/assembly/core/src/service_agent_runtime.rs b/src/crates/assembly/core/src/service_agent_runtime.rs index 8fd16a13ef..bc20243363 100644 --- a/src/crates/assembly/core/src/service_agent_runtime.rs +++ b/src/crates/assembly/core/src/service_agent_runtime.rs @@ -582,6 +582,13 @@ impl AgentDialogTurnPort for RejectBusyAgentDialogTurnPort { .submit_agent_dialog_turn_reject_if_busy(request) .await } + + async fn steer_dialog_turn( + &self, + request: bitfun_runtime_ports::AgentDialogSteerRequest, + ) -> bitfun_runtime_ports::PortResult { + AgentDialogTurnPort::steer_dialog_turn(self.0.as_ref(), request).await + } } #[async_trait::async_trait] diff --git a/src/crates/contracts/runtime-ports/src/lib.rs b/src/crates/contracts/runtime-ports/src/lib.rs index 11576585b0..4983c97f81 100644 --- a/src/crates/contracts/runtime-ports/src/lib.rs +++ b/src/crates/contracts/runtime-ports/src/lib.rs @@ -1565,6 +1565,17 @@ pub struct AgentDialogTurnRequest { pub metadata: serde_json::Map, } +/// Text-only steering request for one exact running dialog turn. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct AgentDialogSteerRequest { + pub session_id: String, + pub turn_id: String, + pub content: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub display_content: Option, +} + impl AgentDialogTurnExecution { pub fn is_standard(&self) -> bool { matches!(self, Self::Standard) @@ -1764,7 +1775,13 @@ pub struct AgentSessionReplyRoute { } /// Outcome for steering a message into an already-running dialog turn. -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde( + tag = "kind", + rename_all = "snake_case", + rename_all_fields = "camelCase", + deny_unknown_fields +)] pub enum DialogSteerOutcome { /// Steering was buffered for the running turn and will be consumed at the /// next model-round boundary. @@ -2414,6 +2431,16 @@ pub trait AgentDialogTurnPort: Send + Sync { &self, request: AgentDialogTurnRequest, ) -> PortResult; + + async fn steer_dialog_turn( + &self, + _request: AgentDialogSteerRequest, + ) -> PortResult { + Err(PortError::new( + PortErrorKind::NotAvailable, + "dialog turn steering is not supported by this provider", + )) + } } #[async_trait::async_trait] @@ -3621,6 +3648,44 @@ mod tests { assert_eq!(json["attachments"][0]["kind"], "remote_image"); } + #[test] + fn agent_dialog_steer_contract_round_trips_exact_turn_identity() { + let request = AgentDialogSteerRequest { + session_id: "session_1".to_string(), + turn_id: "turn_1".to_string(), + content: "Please also check the tests".to_string(), + display_content: Some("Also check tests".to_string()), + }; + let outcome = DialogSteerOutcome::Buffered { + session_id: "session_1".to_string(), + turn_id: "turn_1".to_string(), + steering_id: "steer_1".to_string(), + }; + + let request_json = serde_json::to_value(&request).expect("serialize steer request"); + let outcome_json = serde_json::to_value(&outcome).expect("serialize steer outcome"); + + assert_eq!(request_json["sessionId"], "session_1"); + assert_eq!(request_json["turnId"], "turn_1"); + assert_eq!(request_json["content"], "Please also check the tests"); + assert_eq!(request_json["displayContent"], "Also check tests"); + assert_eq!(outcome_json["kind"], "buffered"); + assert_eq!(outcome_json["sessionId"], "session_1"); + assert_eq!(outcome_json["turnId"], "turn_1"); + assert_eq!(outcome_json["steeringId"], "steer_1"); + + assert_eq!( + serde_json::from_value::(request_json) + .expect("deserialize steer request"), + request + ); + assert_eq!( + serde_json::from_value::(outcome_json) + .expect("deserialize steer outcome"), + outcome + ); + } + #[test] fn agent_background_result_request_serializes_lifecycle_contract() { let mut metadata = serde_json::Map::new(); diff --git a/src/crates/execution/agent-runtime/examples/sdk_minimal.rs b/src/crates/execution/agent-runtime/examples/sdk_minimal.rs index d9c338d61b..001f826cb1 100644 --- a/src/crates/execution/agent-runtime/examples/sdk_minimal.rs +++ b/src/crates/execution/agent-runtime/examples/sdk_minimal.rs @@ -49,7 +49,7 @@ impl AgentSubmissionPort for ExampleAgentProvider { #[tokio::main] async fn main() -> Result<(), Box> { let compatibility = AgentRuntimeSdkCompatibility::current(); - assert_eq!(compatibility.api_version, 3); + assert_eq!(compatibility.api_version, 4); let provider = Arc::new(ExampleAgentProvider::default()); let events = AgentEventStream::new(); diff --git a/src/crates/execution/agent-runtime/src/runtime.rs b/src/crates/execution/agent-runtime/src/runtime.rs index 7218e3d1bd..a7a3b339c5 100644 --- a/src/crates/execution/agent-runtime/src/runtime.rs +++ b/src/crates/execution/agent-runtime/src/runtime.rs @@ -10,29 +10,30 @@ use std::sync::{Arc, Mutex}; use bitfun_agent_tools::{ToolRegistry, ToolRegistryItem}; use bitfun_harness::HarnessRegistry; use bitfun_runtime_ports::{ - AgentBackgroundResultRequest, AgentDialogTurnPort, AgentDialogTurnRequest, - AgentInputAttachment, AgentLifecycleDeliveryPort, AgentLocalCommandTurnPort, - AgentLocalCommandTurnRecordRequest, AgentMessageWorkspaceReferencesRequest, - AgentSessionArchiveRequest, AgentSessionArchiveStateRequest, AgentSessionClosePort, - AgentSessionCompactionPort, AgentSessionCompactionRequest, AgentSessionCompactionResult, - AgentSessionCreateRequest, AgentSessionCreateResult, AgentSessionDeleteRequest, - AgentSessionForkAtTurnRequest, AgentSessionForkBeforeTurnRequest, AgentSessionForkPort, - AgentSessionForkRequest, AgentSessionForkResult, AgentSessionListRequest, - AgentSessionManagementPort, AgentSessionModePort, AgentSessionModeUpdateRequest, - AgentSessionModelPort, AgentSessionModelUpdateRequest, AgentSessionRenameRequest, - AgentSessionRevertPort, AgentSessionRevertRequest, AgentSessionRevertResult, - AgentSessionSummary, AgentSessionUsagePort, AgentSessionUsageRequest, - AgentSessionWorkspaceBinding, AgentSessionWorkspaceRequest, AgentSubmissionPort, - AgentSubmissionRequest, AgentSubmissionResult, AgentSubmissionSource, - AgentThreadGoalCreateRequest, AgentThreadGoalDeliveryRequest, AgentThreadGoalGetRequest, - AgentThreadGoalManagementPort, AgentThreadGoalUpdateStatusRequest, - AgentTransientSessionDiscardRequest, AgentTurnCancellationPort, AgentTurnCancellationRequest, - AgentTurnCancellationResult, AgentTurnSettlementPort, AgentTurnSettlementRequest, - AgentUserShellCommandPort, AgentUserShellCommandRequest, AgentUserShellCommandResult, - AgentWorkspaceReference, AgentWorkspaceReferencePort, AgentWorkspaceReferenceSearchRequest, - AgentWorkspaceReferenceSearchResult, DialogSubmitOutcome, PermissionAuditRecord, - PermissionGrant, PermissionGrantKey, PluginRuntimeBinding, PortError, PortErrorKind, - PortResult, RuntimeEventEnvelope, SessionTranscript, SessionTranscriptReader, + AgentBackgroundResultRequest, AgentDialogSteerRequest, AgentDialogTurnPort, + AgentDialogTurnRequest, AgentInputAttachment, AgentLifecycleDeliveryPort, + AgentLocalCommandTurnPort, AgentLocalCommandTurnRecordRequest, + AgentMessageWorkspaceReferencesRequest, AgentSessionArchiveRequest, + AgentSessionArchiveStateRequest, AgentSessionClosePort, AgentSessionCompactionPort, + AgentSessionCompactionRequest, AgentSessionCompactionResult, AgentSessionCreateRequest, + AgentSessionCreateResult, AgentSessionDeleteRequest, AgentSessionForkAtTurnRequest, + AgentSessionForkBeforeTurnRequest, AgentSessionForkPort, AgentSessionForkRequest, + AgentSessionForkResult, AgentSessionListRequest, AgentSessionManagementPort, + AgentSessionModePort, AgentSessionModeUpdateRequest, AgentSessionModelPort, + AgentSessionModelUpdateRequest, AgentSessionRenameRequest, AgentSessionRevertPort, + AgentSessionRevertRequest, AgentSessionRevertResult, AgentSessionSummary, + AgentSessionUsagePort, AgentSessionUsageRequest, AgentSessionWorkspaceBinding, + AgentSessionWorkspaceRequest, AgentSubmissionPort, AgentSubmissionRequest, + AgentSubmissionResult, AgentSubmissionSource, AgentThreadGoalCreateRequest, + AgentThreadGoalDeliveryRequest, AgentThreadGoalGetRequest, AgentThreadGoalManagementPort, + AgentThreadGoalUpdateStatusRequest, AgentTransientSessionDiscardRequest, + AgentTurnCancellationPort, AgentTurnCancellationRequest, AgentTurnCancellationResult, + AgentTurnSettlementPort, AgentTurnSettlementRequest, AgentUserShellCommandPort, + AgentUserShellCommandRequest, AgentUserShellCommandResult, AgentWorkspaceReference, + AgentWorkspaceReferencePort, AgentWorkspaceReferenceSearchRequest, + AgentWorkspaceReferenceSearchResult, DialogSteerOutcome, DialogSubmitOutcome, + PermissionAuditRecord, PermissionGrant, PermissionGrantKey, PluginRuntimeBinding, PortError, + PortErrorKind, PortResult, RuntimeEventEnvelope, SessionTranscript, SessionTranscriptReader, SessionTranscriptRequest, ThreadGoal, WorkspaceDiffSnapshot, }; use bitfun_runtime_services::RuntimeServices; @@ -1404,6 +1405,38 @@ impl AgentRuntime { Ok(outcome) } + pub async fn steer_dialog_turn( + &self, + request: AgentDialogSteerRequest, + ) -> Result { + let requested_session_id = request.session_id.clone(); + let requested_turn_id = request.turn_id.clone(); + let dialog_turn = self + .dialog_turn + .as_ref() + .ok_or(RuntimeError::MissingDialogTurnPort)?; + let outcome = dialog_turn + .steer_dialog_turn(request) + .await + .map_err(RuntimeError::from)?; + let DialogSteerOutcome::Buffered { + session_id, + turn_id, + .. + } = &outcome; + if session_id != &requested_session_id || turn_id != &requested_turn_id { + return Err(PortError::new( + PortErrorKind::Backend, + format!( + "agent dialog provider returned session_id '{}' and turn_id '{}' for requested session_id '{}' and turn_id '{}'", + session_id, turn_id, requested_session_id, requested_turn_id + ), + ) + .into()); + } + Ok(outcome) + } + pub async fn deliver_background_result( &self, request: AgentBackgroundResultRequest, @@ -3057,6 +3090,142 @@ mod tests { assert!(error.into_message().contains("requested-session")); } + #[tokio::test] + async fn steer_dialog_turn_requires_registered_dialog_turn_port() { + let runtime = AgentRuntimeBuilder::new() + .with_submission_port(Arc::new(FakeAgentRuntimePorts::default())) + .build() + .expect("runtime"); + + let error = runtime + .steer_dialog_turn(bitfun_runtime_ports::AgentDialogSteerRequest { + session_id: "session_1".to_string(), + turn_id: "turn_1".to_string(), + content: "check tests".to_string(), + display_content: None, + }) + .await + .expect_err("steering without a dialog-turn provider must fail"); + + assert_eq!(error, RuntimeError::MissingDialogTurnPort); + } + + #[tokio::test] + async fn steer_dialog_turn_delegates_and_validates_exact_turn_identity() { + #[derive(Debug, Default)] + struct RecordingSteerPort { + requests: Mutex>, + } + + #[async_trait::async_trait] + impl bitfun_runtime_ports::AgentDialogTurnPort for RecordingSteerPort { + async fn submit_dialog_turn( + &self, + request: AgentDialogTurnRequest, + ) -> PortResult { + Ok(DialogSubmitOutcome::Started { + session_id: request.session_id, + turn_id: request.turn_id.unwrap_or_else(|| "generated".to_string()), + }) + } + + async fn steer_dialog_turn( + &self, + request: bitfun_runtime_ports::AgentDialogSteerRequest, + ) -> PortResult { + self.requests.lock().unwrap().push(request.clone()); + Ok(bitfun_runtime_ports::DialogSteerOutcome::Buffered { + session_id: request.session_id, + turn_id: request.turn_id, + steering_id: "steer_1".to_string(), + }) + } + } + + let port = Arc::new(RecordingSteerPort::default()); + let runtime = AgentRuntimeBuilder::new() + .with_submission_port(Arc::new(FakeAgentRuntimePorts::default())) + .with_dialog_turn_port(port.clone()) + .build() + .expect("runtime"); + let request = bitfun_runtime_ports::AgentDialogSteerRequest { + session_id: "session_1".to_string(), + turn_id: "turn_1".to_string(), + content: "check tests".to_string(), + display_content: Some("Check tests".to_string()), + }; + + let result = runtime + .steer_dialog_turn(request.clone()) + .await + .expect("steer dialog turn"); + + assert_eq!(port.requests.lock().unwrap().as_slice(), &[request]); + assert_eq!( + result, + bitfun_runtime_ports::DialogSteerOutcome::Buffered { + session_id: "session_1".to_string(), + turn_id: "turn_1".to_string(), + steering_id: "steer_1".to_string(), + } + ); + } + + #[tokio::test] + async fn steer_dialog_turn_rejects_provider_turn_identity_mismatch() { + #[derive(Debug)] + struct MismatchedSteerPort; + + #[async_trait::async_trait] + impl bitfun_runtime_ports::AgentDialogTurnPort for MismatchedSteerPort { + async fn submit_dialog_turn( + &self, + request: AgentDialogTurnRequest, + ) -> PortResult { + Ok(DialogSubmitOutcome::Started { + session_id: request.session_id, + turn_id: request.turn_id.unwrap_or_else(|| "generated".to_string()), + }) + } + + async fn steer_dialog_turn( + &self, + request: bitfun_runtime_ports::AgentDialogSteerRequest, + ) -> PortResult { + Ok(bitfun_runtime_ports::DialogSteerOutcome::Buffered { + session_id: request.session_id, + turn_id: "different-turn".to_string(), + steering_id: "steer_1".to_string(), + }) + } + } + + let runtime = AgentRuntimeBuilder::new() + .with_submission_port(Arc::new(FakeAgentRuntimePorts::default())) + .with_dialog_turn_port(Arc::new(MismatchedSteerPort)) + .build() + .expect("runtime"); + + let error = runtime + .steer_dialog_turn(bitfun_runtime_ports::AgentDialogSteerRequest { + session_id: "session_1".to_string(), + turn_id: "turn_1".to_string(), + content: "check tests".to_string(), + display_content: None, + }) + .await + .expect_err("provider turn mismatch must fail closed"); + + assert!(matches!( + error, + RuntimeError::Port(PortError { + kind: PortErrorKind::Backend, + .. + }) + )); + assert!(error.into_message().contains("different-turn")); + } + #[tokio::test] async fn deliver_background_result_requires_registered_lifecycle_port() { let ports = Arc::new(FakeAgentRuntimePorts::default()); diff --git a/src/crates/execution/agent-runtime/src/sdk.rs b/src/crates/execution/agent-runtime/src/sdk.rs index f160225979..8c8908ffa2 100644 --- a/src/crates/execution/agent-runtime/src/sdk.rs +++ b/src/crates/execution/agent-runtime/src/sdk.rs @@ -8,7 +8,7 @@ use std::sync::Arc; -pub const AGENT_RUNTIME_SDK_API_VERSION: u32 = 3; +pub const AGENT_RUNTIME_SDK_API_VERSION: u32 = 4; #[derive(Debug, Clone, Copy, PartialEq, Eq)] #[non_exhaustive] @@ -58,8 +58,8 @@ pub use bitfun_harness::{ HarnessRegistry, HarnessWorkflow, }; pub use bitfun_runtime_ports::{ - AgentBackgroundResultRequest, AgentDialogTurnExecution, AgentDialogTurnPort, - AgentDialogTurnRequest, AgentInputAttachment, AgentLifecycleDeliveryPort, + AgentBackgroundResultRequest, AgentDialogSteerRequest, AgentDialogTurnExecution, + AgentDialogTurnPort, AgentDialogTurnRequest, AgentInputAttachment, AgentLifecycleDeliveryPort, AgentLocalCommandTurnPort, AgentLocalCommandTurnRecordRequest, AgentMessageWorkspaceReferencesRequest, AgentSessionArchiveRequest, AgentSessionArchiveStateRequest, AgentSessionClosePort, AgentSessionCompactionPort, @@ -81,9 +81,9 @@ pub use bitfun_runtime_ports::{ AgentWorkspaceReference, AgentWorkspaceReferenceKind, AgentWorkspaceReferencePort, AgentWorkspaceReferenceSearchEntry, AgentWorkspaceReferenceSearchRequest, AgentWorkspaceReferenceSearchResult, AgentWorkspaceReferenceSourceRange, ClockPort, - DialogSubmissionPolicy, DialogSubmitOutcome, FileSystemPort, GitPort, McpCatalogPort, - NetworkPort, PermissionAuditRecord, PermissionDelegationContext, PermissionGrant, - PermissionGrantKey, PermissionReply, PermissionReplySource, PermissionRequest, + DialogSteerOutcome, DialogSubmissionPolicy, DialogSubmitOutcome, FileSystemPort, GitPort, + McpCatalogPort, NetworkPort, PermissionAuditRecord, PermissionDelegationContext, + PermissionGrant, PermissionGrantKey, PermissionReply, PermissionReplySource, PermissionRequest, PermissionRequestEvent, PermissionRequestSource, PermissionRequestSourceKind, PortError, PortErrorKind, PortResult, RemoteAssistantWorkspaceFacts, RemoteCapabilityPort, RemoteConnectionPort, RemoteProjectionPort, RemoteRecentWorkspaceFacts, RemoteWorkspaceFacts, @@ -615,6 +615,13 @@ impl AgentRuntime { self.inner.submit_dialog_turn(request).await } + pub async fn steer_dialog_turn( + &self, + request: AgentDialogSteerRequest, + ) -> Result { + self.inner.steer_dialog_turn(request).await + } + pub async fn deliver_background_result( &self, request: AgentBackgroundResultRequest, diff --git a/src/crates/execution/agent-runtime/tests/sdk_smoke.rs b/src/crates/execution/agent-runtime/tests/sdk_smoke.rs index 090fc73a5a..0272e6633b 100644 --- a/src/crates/execution/agent-runtime/tests/sdk_smoke.rs +++ b/src/crates/execution/agent-runtime/tests/sdk_smoke.rs @@ -53,7 +53,7 @@ struct FakeSessionClosePort { fn sdk_facade_exposes_versioned_preview_compatibility_contract() { let compatibility = AgentRuntimeSdkCompatibility::current(); - assert_eq!(compatibility.api_version, 3); + assert_eq!(compatibility.api_version, 4); assert_eq!(compatibility.crate_version, env!("CARGO_PKG_VERSION")); assert_eq!(compatibility.stability, AgentRuntimeSdkStability::Preview); } diff --git a/src/web-ui/src/flow_chat/components/PendingQueuePanel.tsx b/src/web-ui/src/flow_chat/components/PendingQueuePanel.tsx index 71ec2267b0..e7189b1423 100644 --- a/src/web-ui/src/flow_chat/components/PendingQueuePanel.tsx +++ b/src/web-ui/src/flow_chat/components/PendingQueuePanel.tsx @@ -27,7 +27,10 @@ import { Tooltip, IconButton } from '@/component-library'; import { agentAPI } from '@/infrastructure/api/service-api/AgentAPI'; import { stateMachineManager } from '../state-machine'; import { FlowChatStore } from '../store/FlowChatStore'; -import { pendingQueueManager } from '../services/flow-chat-manager/PendingQueueModule'; +import { + pendingQueueManager, + queuedMessageHasUnsupportedSteeringPayload, +} from '../services/flow-chat-manager/PendingQueueModule'; import { FlowChatManager } from '../services/FlowChatManager'; import { insertSteeringItemIfAbsent } from '../services/flow-chat-manager/EventHandlerModule'; import { notificationService } from '../../shared/notification-system'; @@ -138,29 +141,12 @@ export function PendingQueuePanel({ sessionId, className }: PendingQueuePanelPro itemId: item.id, }); try { - // Move this specific item to the head, then trigger drain. - const allItems = pendingQueueManager.list(sessionId); - if (allItems.length > 1 && allItems[0]?.id !== item.id) { - pendingQueueManager.clear(sessionId); - pendingQueueManager.enqueue({ + if (!pendingQueueManager.promoteForExplicitDrain(sessionId, item.id)) { + log.warn('Send now fallback item is no longer queued', { sessionId, - content: item.content, - displayMessage: item.displayMessage, - agentType: item.agentType, - imageContexts: item.imageContexts, - imageDisplayData: item.imageDisplayData, + itemId: item.id, }); - for (const other of allItems) { - if (other.id === item.id) continue; - pendingQueueManager.enqueue({ - sessionId, - content: other.content, - displayMessage: other.displayMessage, - agentType: other.agentType, - imageContexts: other.imageContexts, - imageDisplayData: other.imageDisplayData, - }); - } + return; } await FlowChatManager.getInstance().drainPendingQueueForSession(sessionId); } catch (err) { @@ -170,6 +156,17 @@ export function PendingQueuePanel({ sessionId, className }: PendingQueuePanelPro return; } + if (queuedMessageHasUnsupportedSteeringPayload(item)) { + log.info('Send now kept queued because steering is text-only', { + sessionId, + itemId: item.id, + }); + notificationService.warning(t('pendingQueue.errors.richContentUnsupported'), { + duration: 4000, + }); + return; + } + pendingQueueManager.setStatus(sessionId, item.id, 'sending_now'); try { const resp = await agentAPI.steerDialogTurn({ diff --git a/src/web-ui/src/flow_chat/services/flow-chat-manager/PendingQueueModule.test.ts b/src/web-ui/src/flow_chat/services/flow-chat-manager/PendingQueueModule.test.ts new file mode 100644 index 0000000000..a571e1e79c --- /dev/null +++ b/src/web-ui/src/flow_chat/services/flow-chat-manager/PendingQueueModule.test.ts @@ -0,0 +1,96 @@ +// @vitest-environment jsdom + +import { afterEach, describe, expect, it } from 'vitest'; +import { + pendingQueueManager, + queuedMessageHasUnsupportedSteeringPayload, +} from './PendingQueueModule'; + +const sessions: string[] = []; + +function testSession(): string { + const sessionId = `pending-queue-test-${sessions.length}`; + sessions.push(sessionId); + return sessionId; +} + +afterEach(() => { + for (const sessionId of sessions.splice(0)) { + pendingQueueManager.clear(sessionId); + } +}); + +describe('PendingQueueModule', () => { + it('promotes an existing item for explicit drain without rebuilding or losing its payload', () => { + const sessionId = testSession(); + pendingQueueManager.enqueue({ sessionId, content: 'first' }); + const target = pendingQueueManager.enqueue({ + sessionId, + content: 'second', + displayMessage: 'Second display', + agentType: 'agentic', + imageContexts: [{ id: 'image-1' }], + imageDisplayData: [{ id: 'image-1', name: 'clip.png' }], + userMessageMetadata: { sessionReferences: [{ sessionId: 'source' }] }, + retryCount: 2, + initialStatus: 'failed', + }); + pendingQueueManager.enqueue({ sessionId, content: 'third' }); + const payload = { + id: target.id, + content: target.content, + displayMessage: target.displayMessage, + agentType: target.agentType, + imageContexts: structuredClone(target.imageContexts), + imageDisplayData: structuredClone(target.imageDisplayData), + userMessageMetadata: structuredClone(target.userMessageMetadata), + timestamp: target.timestamp, + }; + + expect(pendingQueueManager.promoteForExplicitDrain(sessionId, target.id)).toBe(true); + + const items = pendingQueueManager.list(sessionId); + expect(items.map(item => item.content)).toEqual(['second', 'first', 'third']); + expect(items[0]).toBe(target); + expect(items[0]).toMatchObject(payload); + expect(items[0].status).toBe('queued'); + expect(items[0].retryCount).toBe(0); + }); + + it('rejects only payloads that the text-only steering contract would flatten', () => { + const plain = { + id: 'plain', + sessionId: 'session-1', + content: 'plain text', + timestamp: 1, + status: 'queued' as const, + retryCount: 0, + }; + + expect(queuedMessageHasUnsupportedSteeringPayload(plain)).toBe(false); + expect( + queuedMessageHasUnsupportedSteeringPayload({ + ...plain, + imageContexts: [{ id: 'image-1' }], + }), + ).toBe(true); + expect( + queuedMessageHasUnsupportedSteeringPayload({ + ...plain, + userMessageMetadata: { deepReviewRunManifest: { requestId: 'review-1' } }, + }), + ).toBe(true); + expect( + queuedMessageHasUnsupportedSteeringPayload({ + ...plain, + userMessageMetadata: { sessionReferences: [{ sessionId: 'source' }] }, + }), + ).toBe(true); + expect( + queuedMessageHasUnsupportedSteeringPayload({ + ...plain, + userMessageMetadata: { composerPresentation: { parts: [] } }, + }), + ).toBe(true); + }); +}); diff --git a/src/web-ui/src/flow_chat/services/flow-chat-manager/PendingQueueModule.ts b/src/web-ui/src/flow_chat/services/flow-chat-manager/PendingQueueModule.ts index f5e48478f2..9cf9ea6428 100644 --- a/src/web-ui/src/flow_chat/services/flow-chat-manager/PendingQueueModule.ts +++ b/src/web-ui/src/flow_chat/services/flow-chat-manager/PendingQueueModule.ts @@ -43,6 +43,18 @@ export interface EnqueueInput { export type PendingQueueListener = (sessionId: string, items: QueuedMessage[]) => void; +/** + * The core steering contract is text-only. Keep payloads that need structured + * attachment/reference metadata queued for the regular turn submission path. + */ +export function queuedMessageHasUnsupportedSteeringPayload(item: QueuedMessage): boolean { + if ((item.imageContexts?.length ?? 0) > 0 || (item.imageDisplayData?.length ?? 0) > 0) { + return true; + } + const metadata = item.userMessageMetadata; + return metadata != null && Object.keys(metadata).length > 0; +} + class PendingQueueManager { private static _instance: PendingQueueManager | null = null; private queues = new Map(); @@ -165,6 +177,24 @@ class PendingQueueManager { return true; } + /** Confirm a queued item for immediate idle-session drain without rebuilding its payload. */ + promoteForExplicitDrain(sessionId: string, id: string): boolean { + const items = this.queues.get(sessionId); + if (!items) return false; + const index = items.findIndex(item => item.id === id); + if (index === -1) return false; + const item = items[index]; + if (index > 0) { + items.splice(index, 1); + items.unshift(item); + } + item.status = 'queued'; + item.retryCount = 0; + this.persist(sessionId); + this.notify(sessionId); + return true; + } + setStatus(sessionId: string, id: string, status: QueuedMessage['status']): void { const items = this.queues.get(sessionId); if (!items) return; diff --git a/src/web-ui/src/locales/en-US/flow-chat.json b/src/web-ui/src/locales/en-US/flow-chat.json index 37c4d19eb4..1cb1b8a404 100644 --- a/src/web-ui/src/locales/en-US/flow-chat.json +++ b/src/web-ui/src/locales/en-US/flow-chat.json @@ -2486,6 +2486,7 @@ }, "errors": { "emptyContent": "Message content cannot be empty", + "richContentUnsupported": "Messages with attachments or structured metadata can only be sent after the active turn finishes", "sendNowFailed": "Send now failed; please try again" } }, diff --git a/src/web-ui/src/locales/zh-CN/flow-chat.json b/src/web-ui/src/locales/zh-CN/flow-chat.json index 35d39997e9..59300c2681 100644 --- a/src/web-ui/src/locales/zh-CN/flow-chat.json +++ b/src/web-ui/src/locales/zh-CN/flow-chat.json @@ -2486,6 +2486,7 @@ }, "errors": { "emptyContent": "消息内容不能为空", + "richContentUnsupported": "包含附件或结构化信息的消息需要等待当前对话完成后发送", "sendNowFailed": "立即发送失败，请稍后重试" } }, diff --git a/src/web-ui/src/locales/zh-TW/flow-chat.json b/src/web-ui/src/locales/zh-TW/flow-chat.json index 7c775a51da..9096abc952 100644 --- a/src/web-ui/src/locales/zh-TW/flow-chat.json +++ b/src/web-ui/src/locales/zh-TW/flow-chat.json @@ -2486,6 +2486,7 @@ }, "errors": { "emptyContent": "訊息內容不能為空", + "richContentUnsupported": "包含附件或結構化資訊的訊息需要等待目前對話完成後傳送", "sendNowFailed": "立即發送失敗，請稍後重試" } },