From 40b43c9b52d940ba165b31c236f099a11909f818 Mon Sep 17 00:00:00 2001 From: limityan Date: Fri, 17 Jul 2026 00:30:29 +0800 Subject: [PATCH] refactor(runtime): route CLI interaction responses through SDK --- .../agent-runtime-services-design.md | 13 +- docs/plans/core-decomposition-plan.md | 2 +- .../product-architecture-evolution-plan.md | 6 +- src/apps/cli/src/agent/core_adapter.rs | 72 ++++--- src/apps/cli/src/main.rs | 2 +- src/apps/cli/src/peer_host/commands/dialog.rs | 87 ++++++++- .../cli/src/peer_host/commands/session.rs | 2 +- src/apps/cli/src/root_handlers.rs | 10 +- src/apps/desktop/src/api/tool_api.rs | 2 +- .../src/agentic/coordination/coordinator.rs | 159 ++++++++++++++- .../assembly/core/src/product_runtime.rs | 22 --- .../core/src/service_agent_runtime.rs | 32 ++- .../execution/agent-runtime/src/runtime.rs | 102 ++++++++++ src/crates/execution/agent-runtime/src/sdk.rs | 38 +++- .../agent-runtime/src/user_questions.rs | 51 ++++- .../tests/interaction_response_contracts.rs | 184 ++++++++++++++++++ 16 files changed, 685 insertions(+), 99 deletions(-) create mode 100644 src/crates/execution/agent-runtime/tests/interaction_response_contracts.rs diff --git a/docs/architecture/agent-runtime-services-design.md b/docs/architecture/agent-runtime-services-design.md index 0083876222..c981d18766 100644 --- a/docs/architecture/agent-runtime-services-design.md +++ b/docs/architecture/agent-runtime-services-design.md @@ -697,8 +697,10 @@ pub struct HarnessExecutionContext { 当前 CLI 入口已使用类型化 `RuntimeServices` 构造 `ProductRuntimeParts`，并通过一个调用级上下文把 Agent Runtime SDK、Harness、能力注册、调用级权限和 Agentic 事件广播交给 TUI、Exec、Session、Usage 与 -交互模式下的 Peer Host。SDK 已承接会话创建/列举/删除、轮次提交和取消；SDK v1 尚未覆盖的恢复视图、 -消息、分支、用量、快照和工具确认由 `assembly/core` 的单一兼容门面转发。`doctor` 与 `health` 校验真实组装结果及必需注册完整性； +交互模式下的 Peer Host。SDK 已承接会话创建/列举/删除/恢复、类型化转录读取、轮次提交/取消，以及 CLI/TUI +的工具确认、拒绝和用户问题回答；Peer Host 当前只消费同一 SDK 的工具确认与拒绝。SDK v1 尚未覆盖的固定 ID +会话创建、模型更新、分支、用量、快照和 +持久化维护等操作仍由 `assembly/core` 的单一兼容门面转发。`doctor` 与 `health` 校验真实组装结果及必需注册完整性； Core 的 Network、Git 和 MCP Catalog 当前仍含兼容 marker，因此该诊断不等于对这些外部服务做实时探活。 该切换仍是 `product-full` 兼容组装，不是 owner 迁移。协调器、调度器、持久化、工具管线和 Agentic Event Queue @@ -777,8 +779,11 @@ ping 路由。未接入入口的 profile、枚举分支和单元测试仍不能 当前 Runtime SDK 已提供会话创建、列出、删除、恢复和类型化转录读取。`AgentSessionRestoreRequest/Result` 与 `AgentSessionRestorePort` 归 Agent Runtime SDK，以继续复用 Runtime owner 的完整 `SessionState`；类型化 `SessionTranscript` 归 `runtime-ports`。两者都由 `assembly/core` 注入真实 persistence owner，CLI/TUI 是当前恢复与 -转录消费方。`CoreAgentRuntimeCompatibility` 仍承载未迁移的持久化、分支、用量、快照和交互操作；不能据此把整个 -兼容门面一次性删除，也不能把这些操作提前声明为跨宿主稳定接口。 +转录消费方。CLI/TUI 的工具确认、拒绝和用户问题回答，以及 Peer Host 的工具确认与拒绝，通过类型化 +`AgentInteractionResponsePort` 回到 Core 的工具管线或用户输入 owner，不改变审批策略或交互所有权。 +`CoreAgentRuntimeCompatibility` 仍承载未迁移的 +固定 ID 会话创建、模型更新、持久化、分支、用量和快照等操作；不能据此把整个兼容门面一次性删除，也不能把这些 +操作提前声明为跨宿主稳定接口。 ### 4.3 Product Capability 设计 diff --git a/docs/plans/core-decomposition-plan.md b/docs/plans/core-decomposition-plan.md index dab828df78..0fa5d40f1d 100644 --- a/docs/plans/core-decomposition-plan.md +++ b/docs/plans/core-decomposition-plan.md @@ -71,7 +71,7 @@ Peer Host 的 Runtime 接入和跨 Relay/Desktop/Web 的协议切换保持独立 下一步按独立纵向切片推进： -1. 以真实调用方和行为等价测试补齐 SDK 端口，逐项缩小固定 ID、恢复视图、消息、分支、用量和工具确认兼容面。 +1. 以真实调用方和行为等价测试补齐 SDK 端口，逐项缩小固定 ID 会话创建、模型更新、分支、用量、快照和持久化维护兼容面。 2. 迁移 ACP 的会话/权限/事件投影，但保留 ACP stdio 生命周期在接口入口。 3. 继续拆分 TUI 副作用边界并补 package smoke，不以大规模重写替代现有回归保护。 diff --git a/docs/plans/product-architecture-evolution-plan.md b/docs/plans/product-architecture-evolution-plan.md index f1306f546b..db54295d23 100644 --- a/docs/plans/product-architecture-evolution-plan.md +++ b/docs/plans/product-architecture-evolution-plan.md @@ -98,8 +98,10 @@ HarmonyOS 手机 Remote App 不在该平台执行范围内。 ## 7. 工作流五：入口逐项迁移 -- CLI：只迁移 Runtime SDK 已有稳定调用方的 session/turn/cancel 等路径；恢复视图、消息、分支、用量和工具确认 - 在补齐端口与行为测试前继续由现有单一兼容路径转发。 +- CLI：会话创建/列举/删除/恢复、类型化转录和轮次提交/取消已由真实入口消费 Runtime SDK；TUI 通过 SDK 处理 + 工具确认、拒绝和用户问题回答，Exec 与交互模式下的 Peer Host 处理工具确认与拒绝。固定 ID 会话创建、模型 + 更新、分支、用量、快照和持久化维护等操作在补齐端口与行为测试前继续由 + 现有单一兼容路径转发。 - ACP：CLI 行为稳定后单独迁移会话、权限和事件投影；ACP stdio 生命周期留在接口入口。 - Desktop：按服务簇迁移，Tauri、窗口和 app-local 资源留在 Desktop。 - SDK/Server/Remote：只有真实独立调用方出现后才增加；枚举、空计划或测试替身不构成发布能力。 diff --git a/src/apps/cli/src/agent/core_adapter.rs b/src/apps/cli/src/agent/core_adapter.rs index 870c06f6c5..4dd718d72d 100644 --- a/src/apps/cli/src/agent/core_adapter.rs +++ b/src/apps/cli/src/agent/core_adapter.rs @@ -11,7 +11,8 @@ use tokio::sync::Mutex; use super::Agent; use bitfun_agent_runtime::sdk::{ AgentDialogTurnRequest, AgentRuntime, AgentSessionCreateRequest, AgentSessionDeleteRequest, - AgentSessionListRequest, AgentSessionRestoreRequest, AgentTurnCancellationRequest, + AgentSessionListRequest, AgentSessionRestoreRequest, AgentToolConfirmationRequest, + AgentToolRejectionRequest, AgentTurnCancellationRequest, AgentUserAnswersRequest, SessionTranscript, SessionTranscriptRequest, }; use bitfun_agent_runtime::user_questions::USER_INPUT_AVAILABLE_CONTEXT_KEY; @@ -101,7 +102,7 @@ impl CoreAgentAdapter { remote_ssh_host: None, }) .await - .map_err(|error| anyhow::anyhow!(error.to_string())) + .map_err(|error| anyhow::anyhow!(error.into_message())) } pub(crate) async fn list_sessions(&self) -> Result> { @@ -130,7 +131,7 @@ impl CoreAgentAdapter { remote_ssh_host: None, }) .await - .map_err(|error| anyhow::anyhow!(error.to_string()))?; + .map_err(|error| anyhow::anyhow!(error.into_message()))?; let mut session_id_guard = self.session_id.lock().await; let mut turn_id_guard = self.current_turn_id.lock().await; @@ -154,7 +155,7 @@ impl CoreAgentAdapter { remote_ssh_host: None, }) .await - .map_err(|error| anyhow::anyhow!(error.to_string())) + .map_err(|error| anyhow::anyhow!(error.into_message())) } pub(crate) async fn get_transcript(&self, session_id: &str) -> Result { @@ -164,7 +165,7 @@ impl CoreAgentAdapter { turn_id: None, }) .await - .map_err(|error| anyhow::anyhow!(error.to_string())) + .map_err(|error| anyhow::anyhow!(error.into_message())) } pub(crate) async fn update_session_model( @@ -292,14 +293,18 @@ impl CoreAgentAdapter { tracing::info!("Backend session restored: {}", session_id); Ok(()) } - Err(error) if Self::is_session_not_found_error(&error.to_string()) => { - tracing::warn!( - "Session is unavailable, recreating backend session: {}", - session_id - ); - self.recreate_session_with_id(session_id, agent_type).await + Err(error) => { + let message = error.into_message(); + if Self::is_session_not_found_error(&message) { + tracing::warn!( + "Session is unavailable, recreating backend session: {}", + session_id + ); + self.recreate_session_with_id(session_id, agent_type).await + } else { + Err(anyhow::anyhow!(message)) + } } - Err(error) => Err(anyhow::anyhow!(error.to_string())), } } @@ -349,7 +354,7 @@ impl Agent for CoreAgentAdapter { metadata: serde_json::Map::new(), }) .await - .map_err(|error| anyhow::anyhow!(error.to_string()))?; + .map_err(|error| anyhow::anyhow!(error.into_message()))?; let id = session.session_id.clone(); @@ -399,20 +404,21 @@ impl Agent for CoreAgentAdapter { let start_result = self.runtime.submit_dialog_turn(request.clone()).await; if let Err(err) = start_result { - if Self::is_session_not_found_error(&err.to_string()) { + let error_message = err.into_message(); + if Self::is_session_not_found_error(&error_message) { tracing::warn!( "Session missing when starting turn, attempting recovery and retry: session_id={}, error={}", session_id, - err + error_message ); self.ensure_backend_session_alive(&session_id, agent_type) .await?; self.runtime .submit_dialog_turn(request) .await - .map_err(|error| anyhow::anyhow!(error.to_string()))?; + .map_err(|error| anyhow::anyhow!(error.into_message()))?; } else { - return Err(anyhow::anyhow!(err.to_string())); + return Err(anyhow::anyhow!(error_message)); } } @@ -435,7 +441,7 @@ impl Agent for CoreAgentAdapter { wait_timeout_ms: None, }) .await - .map_err(|error| anyhow::anyhow!(error.to_string()))?; + .map_err(|error| anyhow::anyhow!(error.into_message()))?; let mut turn_id_guard = self.current_turn_id.lock().await; if turn_id_guard.as_deref() == Some(turn_id.as_str()) { @@ -460,7 +466,7 @@ impl Agent for CoreAgentAdapter { metadata: serde_json::Map::new(), }) .await - .map_err(|error| anyhow::anyhow!(error.to_string()))?; + .map_err(|error| anyhow::anyhow!(error.into_message()))?; let id = session.session_id.clone(); @@ -482,25 +488,35 @@ impl Agent for CoreAgentAdapter { updated_input: Option, ) -> Result<()> { tracing::info!("Confirming tool execution: {}", tool_id); - self.compatibility - .confirm_tool(tool_id, updated_input) + self.runtime + .confirm_tool(AgentToolConfirmationRequest { + tool_id: tool_id.to_string(), + updated_input, + }) .await - .map_err(|e| anyhow::anyhow!("Confirm tool failed: {}", e)) + .map_err(|e| anyhow::anyhow!("Confirm tool failed: {}", e.into_message())) } async fn reject_tool(&self, tool_id: &str, reason: String) -> Result<()> { tracing::info!("Rejecting tool execution: {}, reason: {}", tool_id, reason); - self.compatibility - .reject_tool(tool_id, reason) + self.runtime + .reject_tool(AgentToolRejectionRequest { + tool_id: tool_id.to_string(), + reason, + }) .await - .map_err(|e| anyhow::anyhow!("Reject tool failed: {}", e)) + .map_err(|e| anyhow::anyhow!("Reject tool failed: {}", e.into_message())) } async fn submit_user_answers(&self, tool_id: &str, answers: serde_json::Value) -> Result<()> { tracing::info!("Submitting user answers for tool: {}", tool_id); - self.compatibility - .submit_user_answers(tool_id, answers) - .map_err(|e| anyhow::anyhow!("Submit user answers failed: {}", e)) + self.runtime + .submit_user_answers(AgentUserAnswersRequest { + tool_id: tool_id.to_string(), + answers, + }) + .await + .map_err(|e| anyhow::anyhow!("Submit user answers failed: {}", e.into_message())) } } diff --git a/src/apps/cli/src/main.rs b/src/apps/cli/src/main.rs index 987f59a03e..e249a26e2f 100644 --- a/src/apps/cli/src/main.rs +++ b/src/apps/cli/src/main.rs @@ -954,7 +954,7 @@ async fn run_interactive_with_session( remote_ssh_host: None, }) .await - .map_err(|error| anyhow::anyhow!(error.to_string()))?; + .map_err(|error| anyhow::anyhow!(error.into_message()))?; let agent_type = sessions .iter() .find(|session| session.session_id == session_id) diff --git a/src/apps/cli/src/peer_host/commands/dialog.rs b/src/apps/cli/src/peer_host/commands/dialog.rs index b898a71e55..b847e8f4d9 100644 --- a/src/apps/cli/src/peer_host/commands/dialog.rs +++ b/src/apps/cli/src/peer_host/commands/dialog.rs @@ -2,6 +2,9 @@ use serde_json::{json, Value}; +use bitfun_agent_runtime::sdk::{ + AgentToolConfirmationRequest, AgentToolRejectionRequest, RuntimeError, +}; use bitfun_runtime_ports::{ AgentDialogTurnRequest, AgentSubmissionSource, AgentTurnCancellationRequest, DialogSubmissionPolicy, DialogTriggerSource, @@ -11,6 +14,17 @@ use crate::peer_host::args::{get_string, optional_string, request_value}; use crate::peer_host::control::{attached_controller_lease, is_controller_lease_current}; use crate::peer_host::state::{PeerHostState, PeerTurnKey}; +fn restore_confirmation_after_runtime_error( + turns: &crate::peer_host::state::PeerTurnTracker, + tool_id: String, + ownership: PeerTurnKey, + action: &str, + error: RuntimeError, +) -> String { + turns.restore_confirmation(tool_id, ownership); + format!("{action} tool failed: {}", error.into_message()) +} + fn peer_dialog_metadata(request: &Value) -> Result, String> { let mut metadata = match request.get("userMessageMetadata") { Some(Value::Object(metadata)) => metadata.clone(), @@ -82,7 +96,10 @@ pub(crate) async fn start_dialog_turn( .await; if let Err(error) = submit_result { state.turns.finish_turn(&turn); - return Err(format!("Failed to start dialog turn: {error}")); + return Err(format!( + "Failed to start dialog turn: {}", + error.into_message() + )); } if !state .turns @@ -101,6 +118,7 @@ pub(crate) async fn start_dialog_turn( }) .await; if let Err(error) = cancellation { + let error = error.into_message(); return Err(format!( "Peer continuity was lost after dialog submission and cancellation could not be confirmed: session_id={session_id}, turn_id={turn_id}, error={error}" )); @@ -135,7 +153,7 @@ pub(crate) async fn cancel_dialog_turn( wait_timeout_ms: Some(1_500), }) .await - .map_err(|e| format!("Failed to cancel dialog turn: {e}"))?; + .map_err(|error| format!("Failed to cancel dialog turn: {}", error.into_message()))?; Ok(json!({ "success": true })) } @@ -159,12 +177,20 @@ pub(crate) async fn confirm_tool_execution( } let updated_input = request.get("updatedInput").cloned(); if let Err(error) = state - .compatibility - .confirm_tool(&tool_id, updated_input) + .agent_runtime + .confirm_tool(AgentToolConfirmationRequest { + tool_id: tool_id.clone(), + updated_input, + }) .await { - state.turns.restore_confirmation(tool_id, ownership); - return Err(format!("Confirm tool failed: {error}")); + return Err(restore_confirmation_after_runtime_error( + &state.turns, + tool_id, + ownership, + "Confirm", + error, + )); } Ok(Value::Null) } @@ -188,9 +214,21 @@ pub(crate) async fn reject_tool_execution( return Err("Tool confirmation session or turn does not match its Peer owner".to_string()); } let reason = optional_string(request, "reason").unwrap_or_else(|| "User rejected".to_string()); - if let Err(error) = state.compatibility.reject_tool(&tool_id, reason).await { - state.turns.restore_confirmation(tool_id, ownership); - return Err(format!("Reject tool failed: {error}")); + if let Err(error) = state + .agent_runtime + .reject_tool(AgentToolRejectionRequest { + tool_id: tool_id.clone(), + reason, + }) + .await + { + return Err(restore_confirmation_after_runtime_error( + &state.turns, + tool_id, + ownership, + "Reject", + error, + )); } Ok(Value::Null) } @@ -199,7 +237,8 @@ pub(crate) async fn reject_tool_execution( mod tests { use serde_json::json; - use super::peer_dialog_metadata; + use super::{peer_dialog_metadata, restore_confirmation_after_runtime_error}; + use crate::peer_host::state::{PeerTurnKey, PeerTurnTracker}; #[test] fn peer_metadata_forces_confirmation_and_cannot_claim_acp_transport() { @@ -238,6 +277,34 @@ mod tests { assert_eq!(metadata.get("caller"), Some(&json!("desktop"))); } + #[test] + fn runtime_failure_restores_the_exact_peer_confirmation_claim() { + let turns = PeerTurnTracker::new(); + turns.mark_event_stream_ready(); + let owner = PeerTurnKey::new("session-1", "turn-1"); + turns.register_root(owner.clone()).expect("register turn"); + turns + .record_confirmation(&owner, "tool-1".to_string()) + .expect("record confirmation"); + let claimed = turns + .claim_confirmation("tool-1") + .expect("claim confirmation"); + + let message = restore_confirmation_after_runtime_error( + &turns, + "tool-1".to_string(), + claimed, + "Confirm", + bitfun_agent_runtime::sdk::RuntimeError::MissingInteractionResponsePort, + ); + + assert_eq!( + message, + "Confirm tool failed: agent interaction response port is not registered" + ); + assert_eq!(turns.claim_confirmation("tool-1"), Some(owner)); + } + #[test] fn peer_metadata_preserves_non_lineage_classification() { let metadata = peer_dialog_metadata(&json!({ diff --git a/src/apps/cli/src/peer_host/commands/session.rs b/src/apps/cli/src/peer_host/commands/session.rs index dcf3f16925..6bbcd3510c 100644 --- a/src/apps/cli/src/peer_host/commands/session.rs +++ b/src/apps/cli/src/peer_host/commands/session.rs @@ -270,7 +270,7 @@ pub(crate) async fn delete_session(state: &PeerHostState, args: &Value) -> Resul remote_ssh_host: optional_string(request, "remoteSshHost"), }) .await - .map_err(|error| format!("Failed to delete session: {error}"))?; + .map_err(|error| format!("Failed to delete session: {}", error.into_message()))?; Ok(Value::Null) } diff --git a/src/apps/cli/src/root_handlers.rs b/src/apps/cli/src/root_handlers.rs index 773425656d..260e233ac5 100644 --- a/src/apps/cli/src/root_handlers.rs +++ b/src/apps/cli/src/root_handlers.rs @@ -234,14 +234,16 @@ pub(crate) async fn handle_session_action( remote_connection_id: None, remote_ssh_host: None, }) - .await?; + .await + .map_err(|error| anyhow::anyhow!(error.into_message()))?; let transcript = runtime .agent_runtime() .read_session_transcript(SessionTranscriptRequest { session_id: session_id.clone(), turn_id: None, }) - .await?; + .await + .map_err(|error| anyhow::anyhow!(error.into_message()))?; println!("Session Details\n"); println!("Name: {}", restored.session.session_name); @@ -279,7 +281,7 @@ pub(crate) async fn handle_session_action( remote_ssh_host: None, }) .await - .map_err(|error| anyhow::anyhow!(error.to_string()))?; + .map_err(|error| anyhow::anyhow!(error.into_message()))?; println!("Deleted session from current project: {}", id); } @@ -346,7 +348,7 @@ async fn list_cli_sessions( remote_ssh_host: None, }) .await - .map_err(|error| anyhow::anyhow!(error.to_string())) + .map_err(|error| anyhow::anyhow!(error.into_message())) } pub(crate) fn handle_config_action(action: ConfigAction, config: &CliConfig) -> Result<()> { diff --git a/src/apps/desktop/src/api/tool_api.rs b/src/apps/desktop/src/api/tool_api.rs index 496045b460..d0f261e51e 100644 --- a/src/apps/desktop/src/api/tool_api.rs +++ b/src/apps/desktop/src/api/tool_api.rs @@ -440,7 +440,7 @@ pub async fn submit_user_answers( "Failed to send user answer: tool_id={}, error={}", tool_id, e ); - e + e.to_string() })?; Ok(()) diff --git a/src/crates/assembly/core/src/agentic/coordination/coordinator.rs b/src/crates/assembly/core/src/agentic/coordination/coordinator.rs index 8e86f22bb7..0664e19cc8 100644 --- a/src/crates/assembly/core/src/agentic/coordination/coordinator.rs +++ b/src/crates/assembly/core/src/agentic/coordination/coordinator.rs @@ -7651,6 +7651,27 @@ fn runtime_port_error_from_bitfun(error: BitFunError) -> bitfun_runtime_ports::P bitfun_runtime_ports::PortError::new(kind, message) } +fn runtime_port_error_preserving_message(error: BitFunError) -> bitfun_runtime_ports::PortError { + let message = error.to_string(); + let mut port_error = runtime_port_error_from_bitfun(error); + port_error.message = message; + port_error +} + +fn user_input_port_error( + error: bitfun_agent_runtime::user_questions::UserInputSendError, +) -> bitfun_runtime_ports::PortError { + let kind = match &error { + bitfun_agent_runtime::user_questions::UserInputSendError::MissingChannel { .. } => { + bitfun_runtime_ports::PortErrorKind::NotFound + } + bitfun_agent_runtime::user_questions::UserInputSendError::ChannelClosed { .. } => { + bitfun_runtime_ports::PortErrorKind::Cancelled + } + }; + bitfun_runtime_ports::PortError::new(kind, format!("Tool error: {error}")) +} + #[async_trait::async_trait] impl bitfun_runtime_ports::AgentSessionManagementPort for ConversationCoordinator { async fn list_sessions( @@ -7755,7 +7776,7 @@ impl bitfun_agent_runtime::sdk::AgentSessionRestorePort for ConversationCoordina &request.session_id, ) .await - .map_err(runtime_port_error_from_bitfun)?; + .map_err(runtime_port_error_preserving_message)?; Ok(bitfun_agent_runtime::sdk::AgentSessionRestoreResult { session: bitfun_runtime_ports::AgentSessionSummary { @@ -7771,6 +7792,36 @@ impl bitfun_agent_runtime::sdk::AgentSessionRestorePort for ConversationCoordina } } +#[async_trait::async_trait] +impl bitfun_agent_runtime::sdk::AgentInteractionResponsePort for ConversationCoordinator { + async fn confirm_tool( + &self, + request: bitfun_agent_runtime::sdk::AgentToolConfirmationRequest, + ) -> bitfun_runtime_ports::PortResult<()> { + self.confirm_tool(&request.tool_id, request.updated_input) + .await + .map_err(runtime_port_error_preserving_message) + } + + async fn reject_tool( + &self, + request: bitfun_agent_runtime::sdk::AgentToolRejectionRequest, + ) -> bitfun_runtime_ports::PortResult<()> { + self.reject_tool(&request.tool_id, request.reason) + .await + .map_err(runtime_port_error_preserving_message) + } + + async fn submit_user_answers( + &self, + request: bitfun_agent_runtime::sdk::AgentUserAnswersRequest, + ) -> bitfun_runtime_ports::PortResult<()> { + crate::agentic::tools::user_input_manager::get_user_input_manager() + .send_answer(&request.tool_id, request.answers) + .map_err(user_input_port_error) + } +} + #[async_trait::async_trait] impl bitfun_runtime_ports::AgentThreadGoalManagementPort for ConversationCoordinator { async fn get_thread_goal( @@ -7914,7 +7965,7 @@ impl bitfun_runtime_ports::SessionTranscriptReader for ConversationCoordinator { let messages = self .get_messages(&request.session_id) .await - .map_err(runtime_port_error_from_bitfun)?; + .map_err(runtime_port_error_preserving_message)?; let messages = messages .into_iter() @@ -8068,9 +8119,9 @@ mod tests { use super::{ background_subagent_delivery_metadata, merge_prepended_messages_for_turn, normalize_subagent_max_concurrency, resolve_agent_session_create_created_by, - resolve_agent_submission_turn_id, should_require_tool_confirmation, - turn_review_manifest_for_agent, validate_background_subagent_delivery, - ConversationCoordinator, SubagentExecutionRequest, + resolve_agent_submission_turn_id, runtime_port_error_preserving_message, + should_require_tool_confirmation, turn_review_manifest_for_agent, + validate_background_subagent_delivery, ConversationCoordinator, SubagentExecutionRequest, }; use crate::agentic::agents::{CustomSubagent, CustomSubagentKind, UserContextPolicy}; use crate::agentic::core::{ @@ -8103,6 +8154,104 @@ mod tests { use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; + + #[test] + fn migrated_runtime_ports_preserve_existing_core_error_messages() { + let error = runtime_port_error_preserving_message( + crate::util::errors::BitFunError::Validation("invalid session id".to_string()), + ); + + assert_eq!( + error.kind, + bitfun_runtime_ports::PortErrorKind::InvalidRequest + ); + assert_eq!(error.message, "Validation error: invalid session id"); + } + + #[tokio::test] + async fn interaction_response_port_uses_core_owners_and_typed_stale_errors() { + use bitfun_agent_runtime::sdk::{ + AgentInteractionResponsePort, AgentToolConfirmationRequest, AgentToolRejectionRequest, + AgentUserAnswersRequest, + }; + + let (coordinator, _) = test_coordinator(); + let answer_tool_id = format!("answer-{}", uuid::Uuid::new_v4()); + let (sender, receiver) = tokio::sync::oneshot::channel::< + bitfun_agent_runtime::user_questions::UserInputResponse, + >(); + crate::agentic::tools::user_input_manager::get_user_input_manager() + .register_channel(answer_tool_id.clone(), sender); + + AgentInteractionResponsePort::submit_user_answers( + &coordinator, + AgentUserAnswersRequest { + tool_id: answer_tool_id.clone(), + answers: serde_json::json!({ "0": "continue" }), + }, + ) + .await + .expect("deliver user answers through the Core-owned channel"); + assert_eq!( + receiver.await.expect("receive user answers").answers, + serde_json::json!({ "0": "continue" }) + ); + + let stale_answer = AgentInteractionResponsePort::submit_user_answers( + &coordinator, + AgentUserAnswersRequest { + tool_id: answer_tool_id.clone(), + answers: serde_json::json!({ "0": "continue" }), + }, + ) + .await + .expect_err("consumed answer channel must be reported as stale"); + assert_eq!( + stale_answer.kind, + bitfun_runtime_ports::PortErrorKind::NotFound + ); + assert_eq!( + stale_answer.message, + format!("Tool error: Waiting channel not found: {answer_tool_id}") + ); + + let missing_tool_id = format!("tool-{}", uuid::Uuid::new_v4()); + let confirmation = AgentInteractionResponsePort::confirm_tool( + &coordinator, + AgentToolConfirmationRequest { + tool_id: missing_tool_id.clone(), + updated_input: Some(serde_json::json!({ "path": "updated.txt" })), + }, + ) + .await + .expect_err("missing confirmation task"); + assert_eq!( + confirmation.kind, + bitfun_runtime_ports::PortErrorKind::NotFound + ); + assert_eq!( + confirmation.message, + format!("Not found: Tool task not found: {missing_tool_id}") + ); + + let rejection = AgentInteractionResponsePort::reject_tool( + &coordinator, + AgentToolRejectionRequest { + tool_id: missing_tool_id.clone(), + reason: "Use a read-only path".to_string(), + }, + ) + .await + .expect_err("missing rejection task"); + assert_eq!( + rejection.kind, + bitfun_runtime_ports::PortErrorKind::NotFound + ); + assert_eq!( + rejection.message, + format!("Not found: Tool task not found: {missing_tool_id}") + ); + } use tokio::sync::RwLock as TokioRwLock; fn test_coordinator() -> (ConversationCoordinator, Arc) { diff --git a/src/crates/assembly/core/src/product_runtime.rs b/src/crates/assembly/core/src/product_runtime.rs index e3bdb3d53d..48acdaa6d9 100644 --- a/src/crates/assembly/core/src/product_runtime.rs +++ b/src/crates/assembly/core/src/product_runtime.rs @@ -259,28 +259,6 @@ impl CoreAgentRuntimeCompatibility { .await } - pub async fn confirm_tool( - &self, - tool_id: &str, - updated_input: Option, - ) -> BitFunResult<()> { - self.coordinator.confirm_tool(tool_id, updated_input).await - } - - pub async fn reject_tool(&self, tool_id: &str, reason: String) -> BitFunResult<()> { - self.coordinator.reject_tool(tool_id, reason).await - } - - pub fn submit_user_answers( - &self, - tool_id: &str, - answers: serde_json::Value, - ) -> BitFunResult<()> { - crate::agentic::tools::user_input_manager::get_user_input_manager() - .send_answer(tool_id, answers) - .map_err(BitFunError::tool) - } - pub async fn branch_session_at_latest_turn( &self, workspace_path: &Path, diff --git a/src/crates/assembly/core/src/service_agent_runtime.rs b/src/crates/assembly/core/src/service_agent_runtime.rs index a7016b2a7c..ec98e6a702 100644 --- a/src/crates/assembly/core/src/service_agent_runtime.rs +++ b/src/crates/assembly/core/src/service_agent_runtime.rs @@ -6,7 +6,8 @@ //! implementations until a reviewed port/provider migration proves equivalence. use bitfun_agent_runtime::sdk::{ - AgentRuntime, AgentRuntimeBuilder, AgentSessionRestorePort, RuntimeError, + AgentInteractionResponsePort, AgentRuntime, AgentRuntimeBuilder, AgentSessionRestorePort, + RuntimeError, }; use bitfun_runtime_ports::{ AgentDialogTurnPort, AgentDialogTurnRequest, AgentInputAttachment, AgentLifecycleDeliveryPort, @@ -395,6 +396,7 @@ fn core_agent_runtime_builder( transcript_reader: Arc, thread_goal_management: Arc, cancellation: Arc, + interaction_response: Arc, ) -> AgentRuntimeBuilder { let agent_registry: Arc = crate::agentic::agents::get_agent_registry(); @@ -405,6 +407,7 @@ fn core_agent_runtime_builder( .with_session_transcript_reader(transcript_reader) .with_thread_goal_management_port(thread_goal_management) .with_cancellation_port(cancellation) + .with_interaction_response_port(interaction_response) .with_agent_registry(agent_registry) } @@ -782,7 +785,8 @@ impl CoreServiceAgentRuntime { let transcript_reader: Arc = coordinator.clone(); let thread_goal_management: Arc = coordinator.clone(); - let cancellation: Arc = coordinator; + let cancellation: Arc = coordinator.clone(); + let interaction_response: Arc = coordinator; core_agent_runtime_builder( submission, session_management, @@ -790,6 +794,7 @@ impl CoreServiceAgentRuntime { transcript_reader, thread_goal_management, cancellation, + interaction_response, ) .build() .map_err(|error| error.to_string()) @@ -806,7 +811,8 @@ impl CoreServiceAgentRuntime { let transcript_reader: Arc = coordinator.clone(); let thread_goal_management: Arc = coordinator.clone(); - let cancellation: Arc = coordinator; + let cancellation: Arc = coordinator.clone(); + let interaction_response: Arc = coordinator; let dialog_turn: Arc = scheduler.clone(); let lifecycle_delivery: Arc = scheduler; core_agent_runtime_builder( @@ -816,6 +822,7 @@ impl CoreServiceAgentRuntime { transcript_reader, thread_goal_management, cancellation, + interaction_response, ) .with_dialog_turn_port(dialog_turn) .with_lifecycle_delivery_port(lifecycle_delivery) @@ -834,7 +841,8 @@ impl CoreServiceAgentRuntime { let transcript_reader: Arc = coordinator.clone(); let thread_goal_management: Arc = coordinator.clone(); - let cancellation: Arc = coordinator; + let cancellation: Arc = coordinator.clone(); + let interaction_response: Arc = coordinator; let lifecycle_delivery: Arc = scheduler; core_agent_runtime_builder( submission, @@ -843,6 +851,7 @@ impl CoreServiceAgentRuntime { transcript_reader, thread_goal_management, cancellation, + interaction_response, ) .with_lifecycle_delivery_port(lifecycle_delivery) .build() @@ -859,7 +868,8 @@ impl CoreServiceAgentRuntime { let session_restore: Arc = coordinator.clone(); let transcript_reader: Arc = coordinator.clone(); - let thread_goal_management: Arc = coordinator; + let thread_goal_management: Arc = coordinator.clone(); + let interaction_response: Arc = coordinator; let cancellation: Arc = scheduler.clone(); let dialog_turn: Arc = scheduler.clone(); let lifecycle_delivery: Arc = scheduler; @@ -870,6 +880,7 @@ impl CoreServiceAgentRuntime { transcript_reader, thread_goal_management, cancellation, + interaction_response, ) .with_dialog_turn_port(dialog_turn) .with_lifecycle_delivery_port(lifecycle_delivery) @@ -889,7 +900,8 @@ impl CoreServiceAgentRuntime { let session_restore: Arc = coordinator.clone(); let transcript_reader: Arc = coordinator.clone(); - let thread_goal_management: Arc = coordinator; + let thread_goal_management: Arc = coordinator.clone(); + let interaction_response: Arc = coordinator; let cancellation: Arc = scheduler.clone(); let dialog_turn: Arc = scheduler.clone(); let lifecycle_delivery: Arc = scheduler; @@ -901,6 +913,7 @@ impl CoreServiceAgentRuntime { transcript_reader, thread_goal_management, cancellation, + interaction_response, ) .with_dialog_turn_port(dialog_turn) .with_lifecycle_delivery_port(lifecycle_delivery) @@ -919,10 +932,7 @@ impl CoreServiceAgentRuntime { } pub(crate) fn runtime_error_message(error: RuntimeError) -> String { - match error { - RuntimeError::Port(error) => error.message, - other => other.to_string(), - } + error.into_message() } } @@ -1509,6 +1519,7 @@ impl RemoteInteractionRuntimeHost for CoreRemoteInteractionRuntimeHost { fn answer_question(&self, tool_id: &str, answers: serde_json::Value) -> Result<(), String> { crate::agentic::tools::user_input_manager::get_user_input_manager() .send_answer(tool_id, answers) + .map_err(|error| error.to_string()) } } @@ -1580,6 +1591,7 @@ mod tests { fn assert_runtime_ports() where T: AgentSubmissionPort + + AgentInteractionResponsePort + AgentSessionManagementPort + AgentThreadGoalManagementPort + AgentTurnCancellationPort diff --git a/src/crates/execution/agent-runtime/src/runtime.rs b/src/crates/execution/agent-runtime/src/runtime.rs index 4d429f6342..080dc27163 100644 --- a/src/crates/execution/agent-runtime/src/runtime.rs +++ b/src/crates/execution/agent-runtime/src/runtime.rs @@ -50,12 +50,24 @@ pub enum RuntimeError { MissingSessionTranscriptReader, #[error("agent thread goal management port is not registered")] MissingThreadGoalManagementPort, + #[error("agent interaction response port is not registered")] + MissingInteractionResponsePort, #[error("runtime event sink is not registered")] MissingEventSink, #[error(transparent)] Port(#[from] PortError), } +impl RuntimeError { + /// Returns the provider message without prepending the structured port error kind. + pub fn into_message(self) -> String { + match self { + Self::Port(error) => error.message, + other => other.to_string(), + } + } +} + #[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)] #[serde(rename_all = "camelCase")] pub struct AgentSessionRestoreRequest { @@ -82,6 +94,41 @@ pub trait AgentSessionRestorePort: Send + Sync { ) -> PortResult; } +#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +/// Confirms a pending tool call, optionally replacing its input before execution. +pub struct AgentToolConfirmationRequest { + pub tool_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub updated_input: Option, +} + +#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +/// Rejects a pending tool call with the user's reason. +pub struct AgentToolRejectionRequest { + pub tool_id: String, + pub reason: String, +} + +#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "camelCase")] +/// Delivers answers to a pending user-question tool call. +pub struct AgentUserAnswersRequest { + pub tool_id: String, + pub answers: serde_json::Value, +} + +#[async_trait::async_trait] +/// Routes product responses to the existing tool and user-input owners. +/// +/// Implementations do not own approval policy or interaction lifecycle state. +pub trait AgentInteractionResponsePort: Send + Sync { + async fn confirm_tool(&self, request: AgentToolConfirmationRequest) -> PortResult<()>; + async fn reject_tool(&self, request: AgentToolRejectionRequest) -> PortResult<()>; + async fn submit_user_answers(&self, request: AgentUserAnswersRequest) -> PortResult<()>; +} + #[derive(Clone, Default)] pub struct AgentEventStream { events: Arc>>, @@ -140,6 +187,7 @@ pub struct AgentRuntime { dialog_turn: Option>, lifecycle_delivery: Option>, cancellation: Option>, + interaction_response: Option>, services: Option, event_stream: Option, tool_registry: Option>, @@ -202,6 +250,13 @@ impl std::fmt::Debug for AgentRuntime { .as_ref() .map(|_| ""), ) + .field( + "interaction_response", + &self + .interaction_response + .as_ref() + .map(|_| ""), + ) .field( "services", &self.services.as_ref().map(|_| ""), @@ -254,6 +309,7 @@ pub struct AgentRuntimeBuilder { dialog_turn: Option>, lifecycle_delivery: Option>, cancellation: Option>, + interaction_response: Option>, services: Option, event_stream: Option, tool_registry: Option>, @@ -320,6 +376,14 @@ impl AgentRuntimeBuilder { self } + pub fn with_interaction_response_port( + mut self, + port: Arc, + ) -> Self { + self.interaction_response = Some(port); + self + } + pub fn with_services(mut self, services: RuntimeServices) -> Self { self.services = Some(services); self @@ -365,6 +429,7 @@ impl AgentRuntimeBuilder { dialog_turn, lifecycle_delivery, cancellation, + interaction_response, services, event_stream, tool_registry, @@ -387,6 +452,7 @@ impl AgentRuntimeBuilder { dialog_turn, lifecycle_delivery, cancellation, + interaction_response, services, event_stream, tool_registry, @@ -506,6 +572,42 @@ impl AgentRuntime { .unwrap_or_default() } + pub async fn confirm_tool( + &self, + request: AgentToolConfirmationRequest, + ) -> Result<(), RuntimeError> { + self.interaction_response + .as_ref() + .ok_or(RuntimeError::MissingInteractionResponsePort)? + .confirm_tool(request) + .await?; + Ok(()) + } + + pub async fn reject_tool( + &self, + request: AgentToolRejectionRequest, + ) -> Result<(), RuntimeError> { + self.interaction_response + .as_ref() + .ok_or(RuntimeError::MissingInteractionResponsePort)? + .reject_tool(request) + .await?; + Ok(()) + } + + pub async fn submit_user_answers( + &self, + request: AgentUserAnswersRequest, + ) -> Result<(), RuntimeError> { + self.interaction_response + .as_ref() + .ok_or(RuntimeError::MissingInteractionResponsePort)? + .submit_user_answers(request) + .await?; + Ok(()) + } + pub fn harness_provider_ids(&self) -> Vec<&str> { self.harness_registry .as_ref() diff --git a/src/crates/execution/agent-runtime/src/sdk.rs b/src/crates/execution/agent-runtime/src/sdk.rs index d8557eab16..6f21e077c8 100644 --- a/src/crates/execution/agent-runtime/src/sdk.rs +++ b/src/crates/execution/agent-runtime/src/sdk.rs @@ -40,10 +40,11 @@ pub use crate::post_call_hooks::{ RuntimeHookRegistryBuildError, }; pub use crate::runtime::{ - AgentEventStream, AgentRunHandle, AgentRunRequest, AgentSessionRestorePort, - AgentSessionRestoreRequest, AgentSessionRestoreResult, RuntimeAgentRegistry, - RuntimeAgentRegistryQuery, RuntimeBuildError, RuntimeError, RuntimeToolRegistry, - SessionSelector, + AgentEventStream, AgentInteractionResponsePort, AgentRunHandle, AgentRunRequest, + AgentSessionRestorePort, AgentSessionRestoreRequest, AgentSessionRestoreResult, + AgentToolConfirmationRequest, AgentToolRejectionRequest, AgentUserAnswersRequest, + RuntimeAgentRegistry, RuntimeAgentRegistryQuery, RuntimeBuildError, RuntimeError, + RuntimeToolRegistry, SessionSelector, }; pub use crate::session_state::{session_state_label_for_state, ProcessingPhase, SessionState}; pub use bitfun_agent_tools::{ToolRegistry, ToolRegistryItem}; @@ -151,6 +152,14 @@ impl AgentRuntimeBuilder { self } + pub fn with_interaction_response_port( + mut self, + port: Arc, + ) -> Self { + self.inner = self.inner.with_interaction_response_port(port); + self + } + pub fn with_services(mut self, services: RuntimeServices) -> Self { self.inner = self.inner.with_services(services); self @@ -312,6 +321,27 @@ impl AgentRuntime { self.inner.cancel_turn(request).await } + pub async fn confirm_tool( + &self, + request: AgentToolConfirmationRequest, + ) -> Result<(), RuntimeError> { + self.inner.confirm_tool(request).await + } + + pub async fn reject_tool( + &self, + request: AgentToolRejectionRequest, + ) -> Result<(), RuntimeError> { + self.inner.reject_tool(request).await + } + + pub async fn submit_user_answers( + &self, + request: AgentUserAnswersRequest, + ) -> Result<(), RuntimeError> { + self.inner.submit_user_answers(request).await + } + pub async fn publish_event(&self, event: RuntimeEventEnvelope) -> Result<(), RuntimeError> { self.inner.publish_event(event).await } diff --git a/src/crates/execution/agent-runtime/src/user_questions.rs b/src/crates/execution/agent-runtime/src/user_questions.rs index e3f195f2fa..818048e709 100644 --- a/src/crates/execution/agent-runtime/src/user_questions.rs +++ b/src/crates/execution/agent-runtime/src/user_questions.rs @@ -38,6 +38,14 @@ pub struct UserInputResponse { pub answers: Value, } +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +pub enum UserInputSendError { + #[error("Waiting channel not found: {tool_id}")] + MissingChannel { tool_id: String }, + #[error("Channel closed, cannot send answer: {tool_id}")] + ChannelClosed { tool_id: String }, +} + pub struct UserInputManager { channels: Arc>>, } @@ -60,20 +68,24 @@ impl UserInputManager { self.channels.insert(tool_id, sender); } - pub fn send_answer(&self, tool_id: &str, answers: Value) -> Result<(), String> { + pub fn send_answer(&self, tool_id: &str, answers: Value) -> Result<(), UserInputSendError> { info!("Sending user answer: tool_id={}", tool_id); if let Some((_, sender)) = self.channels.remove(tool_id) { let response = UserInputResponse { answers }; sender .send(response) - .map_err(|_| format!("Channel closed, cannot send answer: {}", tool_id))?; + .map_err(|_| UserInputSendError::ChannelClosed { + tool_id: tool_id.to_string(), + })?; debug!("Answer sent: tool_id={}", tool_id); Ok(()) } else { - let error_msg = format!("Waiting channel not found: {}", tool_id); - warn!("{}", error_msg); - Err(error_msg) + let error = UserInputSendError::MissingChannel { + tool_id: tool_id.to_string(), + }; + warn!("{}", error); + Err(error) } } @@ -249,7 +261,7 @@ fn format_result_for_assistant(questions: &[Question], answers: &Value) -> Strin #[cfg(test)] mod tests { - use super::{UserInputManager, UserInputResponse}; + use super::{UserInputManager, UserInputResponse, UserInputSendError}; use serde_json::json; #[tokio::test] @@ -280,6 +292,33 @@ mod tests { assert!(!manager.cancel("tool-1")); } + #[tokio::test] + async fn user_input_manager_distinguishes_missing_and_closed_channels() { + let manager = UserInputManager::new(); + let missing = manager + .send_answer("missing-tool", json!({ "0": "yes" })) + .expect_err("missing channel"); + assert_eq!( + missing, + UserInputSendError::MissingChannel { + tool_id: "missing-tool".to_string(), + } + ); + + let (sender, receiver) = tokio::sync::oneshot::channel::(); + manager.register_channel("closed-tool".to_string(), sender); + drop(receiver); + let closed = manager + .send_answer("closed-tool", json!({ "0": "yes" })) + .expect_err("closed channel"); + assert_eq!( + closed, + UserInputSendError::ChannelClosed { + tool_id: "closed-tool".to_string(), + } + ); + } + #[test] fn user_input_manager_reports_pending_tool_ids() { let manager = UserInputManager::new(); diff --git a/src/crates/execution/agent-runtime/tests/interaction_response_contracts.rs b/src/crates/execution/agent-runtime/tests/interaction_response_contracts.rs new file mode 100644 index 0000000000..a484c392e7 --- /dev/null +++ b/src/crates/execution/agent-runtime/tests/interaction_response_contracts.rs @@ -0,0 +1,184 @@ +use std::sync::{Arc, Mutex}; + +use async_trait::async_trait; +use bitfun_agent_runtime::sdk::{ + AgentInteractionResponsePort, AgentRuntimeBuilder, AgentSubmissionPort, AgentSubmissionRequest, + AgentSubmissionResult, AgentToolConfirmationRequest, AgentToolRejectionRequest, + AgentUserAnswersRequest, PortError, PortResult, RuntimeError, +}; +use bitfun_runtime_ports::PortErrorKind; +use serde_json::json; + +#[derive(Debug, Default)] +struct FakeSubmissionPort; + +#[async_trait] +impl AgentSubmissionPort for FakeSubmissionPort { + async fn create_session( + &self, + _request: bitfun_agent_runtime::sdk::AgentSessionCreateRequest, + ) -> PortResult { + unreachable!("interaction response contracts do not create sessions") + } + + async fn submit_message( + &self, + _request: AgentSubmissionRequest, + ) -> PortResult { + unreachable!("interaction response contracts do not submit messages") + } + + async fn resolve_session_agent_type(&self, _session_id: &str) -> PortResult> { + Ok(None) + } +} + +#[derive(Debug, Clone, PartialEq)] +enum RecordedResponse { + Confirm(AgentToolConfirmationRequest), + Reject(AgentToolRejectionRequest), + Answers(AgentUserAnswersRequest), +} + +#[derive(Debug, Default)] +struct RecordingInteractionResponsePort { + responses: Mutex>, +} + +#[async_trait] +impl AgentInteractionResponsePort for RecordingInteractionResponsePort { + async fn confirm_tool(&self, request: AgentToolConfirmationRequest) -> PortResult<()> { + self.responses + .lock() + .unwrap() + .push(RecordedResponse::Confirm(request)); + Ok(()) + } + + async fn reject_tool(&self, request: AgentToolRejectionRequest) -> PortResult<()> { + self.responses + .lock() + .unwrap() + .push(RecordedResponse::Reject(request)); + Ok(()) + } + + async fn submit_user_answers(&self, request: AgentUserAnswersRequest) -> PortResult<()> { + self.responses + .lock() + .unwrap() + .push(RecordedResponse::Answers(request)); + Ok(()) + } +} + +#[tokio::test] +async fn sdk_forwards_typed_interaction_responses_without_losing_payloads() { + let responses = Arc::new(RecordingInteractionResponsePort::default()); + let runtime = AgentRuntimeBuilder::new() + .with_submission_port(Arc::new(FakeSubmissionPort)) + .with_interaction_response_port(responses.clone()) + .build() + .expect("runtime with interaction response port"); + + let confirmation = AgentToolConfirmationRequest { + tool_id: "tool-1".to_string(), + updated_input: Some(json!({ "path": "updated.txt" })), + }; + let rejection = AgentToolRejectionRequest { + tool_id: "tool-2".to_string(), + reason: "Use the read-only path".to_string(), + }; + let answers = AgentUserAnswersRequest { + tool_id: "tool-3".to_string(), + answers: json!({ "choice": "continue", "notes": ["keep history"] }), + }; + + runtime + .confirm_tool(confirmation.clone()) + .await + .expect("confirm tool"); + runtime + .reject_tool(rejection.clone()) + .await + .expect("reject tool"); + runtime + .submit_user_answers(answers.clone()) + .await + .expect("submit user answers"); + + assert_eq!( + *responses.responses.lock().unwrap(), + vec![ + RecordedResponse::Confirm(confirmation), + RecordedResponse::Reject(rejection), + RecordedResponse::Answers(answers), + ] + ); +} + +#[tokio::test] +async fn sdk_reports_a_missing_interaction_response_port() { + let runtime = AgentRuntimeBuilder::new() + .with_submission_port(Arc::new(FakeSubmissionPort)) + .build() + .expect("runtime without optional interaction response port"); + + let error = runtime + .confirm_tool(AgentToolConfirmationRequest { + tool_id: "tool-1".to_string(), + updated_input: None, + }) + .await + .expect_err("missing port must be explicit"); + + assert_eq!(error, RuntimeError::MissingInteractionResponsePort); +} + +#[test] +fn interaction_response_requests_keep_camel_case_wire_fields() { + assert_eq!( + serde_json::to_value(AgentToolConfirmationRequest { + tool_id: "tool-1".to_string(), + updated_input: Some(json!({ "command": "safe" })), + }) + .expect("serialize confirmation request"), + json!({ + "toolId": "tool-1", + "updatedInput": { "command": "safe" }, + }) + ); + assert_eq!( + serde_json::to_value(AgentToolRejectionRequest { + tool_id: "tool-2".to_string(), + reason: "User rejected".to_string(), + }) + .expect("serialize rejection request"), + json!({ + "toolId": "tool-2", + "reason": "User rejected", + }) + ); + assert_eq!( + serde_json::to_value(AgentUserAnswersRequest { + tool_id: "tool-3".to_string(), + answers: json!({ "choice": "continue" }), + }) + .expect("serialize user answers request"), + json!({ + "toolId": "tool-3", + "answers": { "choice": "continue" }, + }) + ); +} + +#[test] +fn runtime_error_message_keeps_provider_text_without_port_kind_prefix() { + let message = RuntimeError::Port(PortError::new( + PortErrorKind::Backend, + "Tool error: question channel closed", + )) + .into_message(); + + assert_eq!(message, "Tool error: question channel closed"); +}