diff --git a/channels-src/feishu/Cargo.lock b/channels-src/feishu/Cargo.lock index 60f68fcc..4e95f3fe 100644 --- a/channels-src/feishu/Cargo.lock +++ b/channels-src/feishu/Cargo.lock @@ -44,6 +44,7 @@ version = "0.1.0" dependencies = [ "serde", "serde_json", + "subtle", "wit-bindgen", ] @@ -208,6 +209,12 @@ dependencies = [ "smallvec", ] +[[package]] +name = "subtle" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" + [[package]] name = "syn" version = "2.0.117" diff --git a/channels-src/feishu/Cargo.toml b/channels-src/feishu/Cargo.toml index 53b9357d..95762410 100644 --- a/channels-src/feishu/Cargo.toml +++ b/channels-src/feishu/Cargo.toml @@ -15,6 +15,7 @@ wit-bindgen = "0.36" # Serialization serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" +subtle = "2.6" # Exclude from parent workspace (this is a standalone WASM component) diff --git a/channels-src/feishu/feishu.capabilities.json b/channels-src/feishu/feishu.capabilities.json index a228cc4e..cf344d74 100644 --- a/channels-src/feishu/feishu.capabilities.json +++ b/channels-src/feishu/feishu.capabilities.json @@ -27,7 +27,7 @@ { "name": "feishu_verification_token", "prompt": "Enter your Feishu/Lark Verification Token (from Event Subscription webhook settings)", - "optional": true + "optional": false } ], "setup_url": "https://open.feishu.cn/app" @@ -63,13 +63,15 @@ }, "webhook": { "secret_header": "X-Feishu-Verification-Token", - "secret_name": "feishu_verification_token" + "secret_name": "feishu_verification_token", + "managed_by_host": false } } }, "config": { "app_id": null, "app_secret": null, + "verification_token": null, "api_base": "https://open.feishu.cn", "owner_id": null, "dm_policy": "pairing", diff --git a/channels-src/feishu/src/lib.rs b/channels-src/feishu/src/lib.rs index 62440d2c..647f5fa5 100644 --- a/channels-src/feishu/src/lib.rs +++ b/channels-src/feishu/src/lib.rs @@ -23,7 +23,8 @@ //! - App credentials (app_id, app_secret) are injected by the host into //! the config JSON during startup for token exchange //! - Bearer token for API calls is obtained via token exchange and cached -//! - Verification token validated by host for webhook requests +//! - Webhook requests must be authenticated by the host or by a matching +//! Feishu verification token in the request body // Generate bindings from the WIT file wit_bindgen::generate!({ @@ -32,6 +33,7 @@ wit_bindgen::generate!({ }); use serde::{Deserialize, Serialize}; +use subtle::ConstantTimeEq; // Re-export generated types use exports::near::agent::channel::{ @@ -50,6 +52,7 @@ const ALLOW_FROM_PATH: &str = "allow_from"; const API_BASE_PATH: &str = "api_base"; const APP_ID_PATH: &str = "app_id"; const APP_SECRET_PATH: &str = "app_secret"; +const VERIFICATION_TOKEN_PATH: &str = "verification_token"; const TOKEN_PATH: &str = "tenant_access_token"; const TOKEN_EXPIRY_PATH: &str = "token_expiry"; @@ -102,6 +105,10 @@ struct FeishuEventHeader { /// Tenant key. #[serde(default)] tenant_key: Option, + + /// Verification token for v2 event payloads. + #[serde(default)] + token: Option, } /// Message receive event payload (im.message.receive_v1). @@ -251,6 +258,9 @@ struct FeishuConfig { /// Feishu App Secret (for token exchange). app_secret: Option, + /// Feishu Event Subscription verification token. + verification_token: Option, + /// API base URL. Defaults to "https://open.feishu.cn" (use /// "https://open.larksuite.com" for Lark international). #[serde(default = "default_api_base")] @@ -300,6 +310,9 @@ impl Guest for FeishuChannel { if let Some(ref app_secret) = config.app_secret { let _ = channel_host::workspace_write(APP_SECRET_PATH, app_secret); } + if let Some(ref verification_token) = config.verification_token { + let _ = channel_host::workspace_write(VERIFICATION_TOKEN_PATH, verification_token); + } if let Some(owner_id) = &config.owner_id { let _ = channel_host::workspace_write(OWNER_ID_PATH, owner_id); @@ -376,6 +389,23 @@ impl Guest for FeishuChannel { } }; + let configured_token = + channel_host::workspace_read(VERIFICATION_TOKEN_PATH).filter(|token| !token.is_empty()); + if !is_authenticated_webhook( + req.secret_validated, + configured_token.as_deref(), + request_verification_token(&event), + ) { + channel_host::log( + channel_host::LogLevel::Warn, + "Rejecting unauthenticated Feishu webhook request", + ); + return json_response( + 401, + serde_json::json!({"error": "Webhook authentication failed"}), + ); + } + // Handle URL verification challenge (initial webhook setup). if event.event_type.as_deref() == Some("url_verification") { if let Some(challenge) = &event.challenge { @@ -839,6 +869,31 @@ fn json_response(status: u16, body: serde_json::Value) -> OutgoingHttpResponse { } } +fn is_authenticated_webhook( + secret_validated: bool, + configured_token: Option<&str>, + request_token: Option<&str>, +) -> bool { + if secret_validated { + return true; + } + + match (configured_token, request_token) { + (Some(expected), Some(provided)) => { + bool::from(expected.as_bytes().ct_eq(provided.as_bytes())) + } + _ => false, + } +} + +fn request_verification_token(event: &FeishuEvent) -> Option<&str> { + event + .header + .as_ref() + .and_then(|header| header.token.as_deref()) + .or(event.token.as_deref()) +} + #[cfg(test)] mod tests { use super::*; @@ -862,7 +917,10 @@ mod tests { fn parse_token_response_rejects_missing_token() { let json = r#"{"code": 0, "msg": "ok", "expire": 7200}"#; let result: Result = serde_json::from_str(json); - assert!(result.is_err(), "should fail when tenant_access_token is missing"); + assert!( + result.is_err(), + "should fail when tenant_access_token is missing" + ); } #[test] @@ -894,4 +952,64 @@ mod tests { assert_eq!(resp.code, 10003); assert!(resp.tenant_access_token.is_empty()); } + + #[test] + fn webhook_auth_requires_host_auth_or_matching_verification_token() { + assert!( + !is_authenticated_webhook(false, None, Some("token")), + "requests without any configured verification mechanism must be rejected" + ); + assert!( + !is_authenticated_webhook(false, Some("expected"), None), + "requests missing the Feishu token must be rejected when host auth did not pass" + ); + assert!( + !is_authenticated_webhook(false, Some("expected"), Some("wrong")), + "requests with the wrong Feishu token must be rejected" + ); + assert!( + is_authenticated_webhook(false, Some("expected"), Some("expected")), + "matching Feishu verification token should authenticate the request" + ); + assert!( + is_authenticated_webhook(true, None, None), + "host-authenticated requests should still be accepted" + ); + assert!( + is_authenticated_webhook(true, Some("expected"), Some("wrong")), + "host authentication should take precedence over body token checks" + ); + } + + #[test] + fn request_verification_token_prefers_v2_header_token() { + let event: FeishuEvent = serde_json::from_str( + r#"{ + "schema": "2.0", + "header": { + "event_id": "evt_123", + "event_type": "im.message.receive_v1", + "token": "header-token" + }, + "event": {} + }"#, + ) + .unwrap(); + + assert_eq!(request_verification_token(&event), Some("header-token")); + } + + #[test] + fn request_verification_token_falls_back_to_top_level_token() { + let event: FeishuEvent = serde_json::from_str( + r#"{ + "type": "url_verification", + "challenge": "abc", + "token": "top-level-token" + }"#, + ) + .unwrap(); + + assert_eq!(request_verification_token(&event), Some("top-level-token")); + } } diff --git a/src/agent/dispatcher.rs b/src/agent/dispatcher.rs index a5f9cd6f..35f76ed1 100644 --- a/src/agent/dispatcher.rs +++ b/src/agent/dispatcher.rs @@ -562,10 +562,6 @@ impl<'a> LoopDelegate for ChatDelegate<'a> { // Walk tool_calls checking approval and hooks. Classify // each tool as Rejected (by hook) or Runnable. Stop at the // first tool that needs approval. - enum PreflightOutcome { - Rejected(String), - Runnable, - } let mut preflight: Vec<(crate::llm::ToolCall, PreflightOutcome)> = Vec::new(); let mut runnable: Vec<(usize, crate::llm::ToolCall)> = Vec::new(); let mut approval_needed: Option<( @@ -818,17 +814,21 @@ impl<'a> LoopDelegate for ChatDelegate<'a> { for (pf_idx, (tc, outcome)) in preflight.into_iter().enumerate() { match outcome { PreflightOutcome::Rejected(error_msg) => { + let (result_content, tool_message) = preflight_rejection_tool_message( + self.agent.safety(), + &tc.name, + &tc.id, + &error_msg, + ); { let mut sess = self.session.lock().await; if let Some(thread) = sess.threads.get_mut(&self.thread_id) && let Some(turn) = thread.last_turn_mut() { - turn.record_tool_error_for(&tc.id, error_msg.clone()); + turn.record_tool_error_for(&tc.id, result_content.clone()); } } - reason_ctx - .messages - .push(ChatMessage::tool_result(&tc.id, &tc.name, error_msg)); + reason_ctx.messages.push(tool_message); } PreflightOutcome::Runnable => { let tool_result = exec_results[pf_idx].take().unwrap_or_else(|| { @@ -936,18 +936,13 @@ impl<'a> LoopDelegate for ChatDelegate<'a> { .insert(tc.id.clone(), output.clone()); } - // Sanitize and add tool result to context let is_tool_error = tool_result.is_err(); - let result_content = match tool_result { - Ok(output) => { - let sanitized = - self.agent.safety().sanitize_tool_output(&tc.name, &output); - self.agent - .safety() - .wrap_for_llm(&tc.name, &sanitized.content) - } - Err(e) => format!("Tool '{}' failed: {}", tc.name, e), - }; + let (result_content, tool_message) = crate::tools::execute::process_tool_result( + self.agent.safety(), + &tc.name, + &tc.id, + &tool_result, + ); // Record sanitized result in thread (identity-based matching). { @@ -966,11 +961,7 @@ impl<'a> LoopDelegate for ChatDelegate<'a> { } } - reason_ctx.messages.push(ChatMessage::tool_result( - &tc.id, - &tc.name, - result_content, - )); + reason_ctx.messages.push(tool_message); } } } @@ -1076,6 +1067,21 @@ pub(super) fn check_auth_required( Some((name, instructions)) } +enum PreflightOutcome { + Rejected(String), + Runnable, +} + +fn preflight_rejection_tool_message( + safety: &crate::safety::SafetyLayer, + tool_name: &str, + tool_call_id: &str, + error_msg: &str, +) -> (String, ChatMessage) { + let result: Result = Err(error_msg); + crate::tools::execute::process_tool_result(safety, tool_name, tool_call_id, &result) +} + /// Build a contextual thinking message based on tool names. /// /// Instead of a generic "Executing 2 tool(s)..." this returns messages like @@ -2509,15 +2515,19 @@ mod tests { #[test] fn test_tool_error_format_includes_tool_name() { - // Regression test for issue #487: tool errors sent to the LLM should - // include the tool name so the model can reason about which tool failed - // and try alternatives. let tool_name = "http"; let err = crate::error::ToolError::ExecutionFailed { name: tool_name.to_string(), reason: "connection refused".to_string(), }; - let formatted = format!("Tool '{}' failed: {}", tool_name, err); + let safety = crate::safety::SafetyLayer::new(&crate::config::SafetyConfig { + max_output_length: 1000, + injection_check_enabled: true, + }); + let result: Result = Err(err); + let (formatted, message) = + crate::tools::execute::process_tool_result(&safety, tool_name, "call_1", &result); + assert!( formatted.contains("Tool 'http' failed:"), "Error should identify the tool by name, got: {formatted}" @@ -2526,6 +2536,11 @@ mod tests { formatted.contains("connection refused"), "Error should include the underlying reason, got: {formatted}" ); + assert!( + formatted.contains("tool_output"), + "Error should be wrapped before entering LLM context, got: {formatted}" + ); + assert_eq!(message.content, formatted); } #[test] @@ -2617,4 +2632,21 @@ mod tests { assert!(result_msg.contains("approval")); assert!(result_msg.contains("DM")); } + + #[test] + fn test_preflight_rejection_tool_message_is_wrapped() { + let safety = crate::safety::SafetyLayer::new(&crate::config::SafetyConfig { + max_output_length: 1000, + injection_check_enabled: true, + }); + let rejection = "requires approval override"; + + let (content, message) = + super::preflight_rejection_tool_message(&safety, "shell", "call_1", rejection); + + assert!(content.contains("tool_output")); + assert!(content.contains("Tool 'shell' failed:")); + assert!(!content.contains("\n")); + assert_eq!(message.content, content); + } } diff --git a/src/agent/thread_ops.rs b/src/agent/thread_ops.rs index a5288f68..af0bd67f 100644 --- a/src/agent/thread_ops.rs +++ b/src/agent/thread_ops.rs @@ -1907,7 +1907,10 @@ fn rebuild_chat_messages_from_db( let name = c["name"].as_str().unwrap_or("unknown").to_string(); let content = if let Some(err) = c.get("error").and_then(|v| v.as_str()) { - format!("Error: {}", err) + // Both wrapped (new) and legacy (plain) errors pass + // through as-is. Legacy errors are already descriptive + // (e.g. "Tool 'http' failed: timeout"), so no prefix needed. + err.to_string() } else if let Some(res) = c.get("result").and_then(|v| v.as_str()) { res.to_string() } else if let Some(preview) = @@ -1993,13 +1996,38 @@ mod tests { assert_eq!(result[3].role, crate::llm::Role::Tool); assert_eq!(result[3].tool_call_id, Some("call_1".to_string())); - assert!(result[3].content.contains("Error: timeout")); + assert!(result[3].content.contains("timeout")); // final assistant assert_eq!(result[4].role, crate::llm::Role::Assistant); assert_eq!(result[4].content, "I found some results."); } + #[test] + fn test_rebuild_chat_messages_preserves_wrapped_tool_error() { + let wrapped_error = + "\nTool 'http' failed: timeout\n"; + let tool_json = serde_json::json!([ + { + "name": "http", + "call_id": "call_1", + "parameters": {"url": "https://example.com"}, + "error": wrapped_error + } + ]); + let messages = vec![ + make_db_msg("user", "Fetch example"), + make_db_msg("tool_calls", &tool_json.to_string()), + ]; + + let result = rebuild_chat_messages_from_db(&messages); + + assert_eq!(result.len(), 3); + assert_eq!(result[2].role, crate::llm::Role::Tool); + assert_eq!(result[2].tool_call_id, Some("call_1".to_string())); + assert_eq!(result[2].content, wrapped_error); + } + #[test] fn test_rebuild_chat_messages_legacy_tool_calls_skipped() { // Legacy format: no call_id field diff --git a/src/channels/relay/client.rs b/src/channels/relay/client.rs index b67f2c5e..1bc60a56 100644 --- a/src/channels/relay/client.rs +++ b/src/channels/relay/client.rs @@ -123,7 +123,7 @@ impl RelayClient { /// for validating the callback — no URLs. pub async fn initiate_oauth(&self, state_nonce: Option<&str>) -> Result { let url = format!("{}/oauth/slack/auth", self.base_url); - tracing::debug!(relay_url = %url, "RelayClient::initiate_oauth: sending request"); + tracing::trace!(relay_url = %url, "RelayClient::initiate_oauth: sending request"); let mut query: Vec<(&str, &str)> = vec![]; if let Some(nonce) = state_nonce { query.push(("state_nonce", nonce)); @@ -143,7 +143,7 @@ impl RelayClient { ); RelayError::Network(e.to_string()) })?; - tracing::debug!( + tracing::trace!( relay_url = %url, status = %resp.status(), "RelayClient::initiate_oauth: received response" @@ -239,7 +239,7 @@ impl RelayClient { body: serde_json::Value, ) -> Result { let url = format!("{}/proxy/{}/{}", self.base_url, provider, method); - tracing::debug!( + tracing::trace!( relay_url = %url, provider = %provider, method = %method, @@ -289,7 +289,7 @@ impl RelayClient { /// extension manager so subsequent calls to `relay_signing_secret()` use it. pub async fn get_signing_secret(&self, team_id: &str) -> Result, RelayError> { let url = format!("{}/relay/signing-secret", self.base_url); - tracing::debug!( + tracing::trace!( relay_url = %url, "RelayClient::get_signing_secret: fetching signing secret" ); @@ -323,7 +323,7 @@ impl RelayClient { message: body, }); } - tracing::debug!( + tracing::trace!( relay_url = %url, "RelayClient::get_signing_secret: received successful response" ); diff --git a/src/channels/wasm/loader.rs b/src/channels/wasm/loader.rs index 6329428f..41ecfac1 100644 --- a/src/channels/wasm/loader.rs +++ b/src/channels/wasm/loader.rs @@ -317,6 +317,14 @@ impl LoadedChannel { .map(|f| f.webhook_secret_name()) .unwrap_or_else(|| format!("{}_webhook_secret", self.channel.channel_name())) } + + /// Whether the host should enforce generic webhook-secret validation. + pub fn webhook_secret_managed_by_host(&self) -> bool { + self.capabilities_file + .as_ref() + .map(|f| f.webhook_secret_managed_by_host()) + .unwrap_or(true) + } } /// Results from loading multiple channels. diff --git a/src/channels/wasm/schema.rs b/src/channels/wasm/schema.rs index b5081426..f2ca6674 100644 --- a/src/channels/wasm/schema.rs +++ b/src/channels/wasm/schema.rs @@ -185,6 +185,19 @@ impl ChannelCapabilitiesFile { .and_then(|w| w.secret_name.clone()) .unwrap_or_else(|| format!("{}_webhook_secret", self.name)) } + + /// Whether the host should enforce generic webhook-secret validation. + /// + /// Defaults to true. Channels can opt out when they validate the shared + /// secret themselves using provider-specific request body fields. + pub fn webhook_secret_managed_by_host(&self) -> bool { + self.capabilities + .channel + .as_ref() + .and_then(|c| c.webhook.as_ref()) + .and_then(|w| w.managed_by_host) + .unwrap_or(true) + } } /// Schema for channel capabilities. @@ -302,6 +315,14 @@ pub struct WebhookSchema { /// Secret name in secrets store for HMAC-SHA256 signing (Slack-style). #[serde(default)] pub hmac_secret_name: Option, + + /// Whether the host/router should enforce generic webhook-secret + /// validation before the channel sees the request. + /// + /// Default: true. Set to false when the provider sends the shared secret + /// in a provider-specific request field rather than the configured header. + #[serde(default)] + pub managed_by_host: Option, } /// Setup configuration schema. @@ -611,6 +632,25 @@ mod tests { Some("X-Telegram-Bot-Api-Secret-Token") ); assert_eq!(file.webhook_secret_name(), "telegram_webhook_secret"); + assert!(file.webhook_secret_managed_by_host()); + } + + #[test] + fn test_webhook_schema_can_disable_host_managed_secret_validation() { + let json = r#"{ + "name": "feishu", + "capabilities": { + "channel": { + "webhook": { + "secret_name": "feishu_verification_token", + "managed_by_host": false + } + } + } + }"#; + + let file = ChannelCapabilitiesFile::from_json(json).unwrap(); + assert!(!file.webhook_secret_managed_by_host()); } #[test] diff --git a/src/channels/wasm/setup.rs b/src/channels/wasm/setup.rs index 7f0bb8fb..84df615f 100644 --- a/src/channels/wasm/setup.rs +++ b/src/channels/wasm/setup.rs @@ -139,13 +139,18 @@ async fn register_channel( }; let secret_header = loaded.webhook_secret_header().map(|s| s.to_string()); + let host_webhook_secret = if loaded.webhook_secret_managed_by_host() { + webhook_secret.clone() + } else { + None + }; let webhook_path = format!("/webhook/{}", channel_name); let endpoints = vec![RegisteredEndpoint { channel_name: channel_name.clone(), path: webhook_path, methods: vec!["POST".to_string()], - require_secret: webhook_secret.is_some(), + require_secret: host_webhook_secret.is_some(), }]; let channel_arc = Arc::new(loaded.channel.with_owner_actor_id(owner_actor_id.clone())); @@ -205,7 +210,7 @@ async fn register_channel( tracing::info!( channel = %channel_name, - has_webhook_secret = webhook_secret.is_some(), + has_webhook_secret = host_webhook_secret.is_some(), secret_header = ?secret_header, "Registering channel with router" ); @@ -214,7 +219,7 @@ async fn register_channel( .register( Arc::clone(&channel_arc), endpoints, - webhook_secret.clone(), + host_webhook_secret.clone(), secret_header, ) .await; @@ -392,8 +397,9 @@ pub async fn inject_channel_credentials( /// placeholders in URLs and headers, so this function fills config fields /// that map to secret names. /// -/// Mapping: for a channel named "feishu", secrets `feishu_app_id` and -/// `feishu_app_secret` are injected as config keys `app_id` and `app_secret`. +/// Mapping: for a channel named "feishu", secrets `feishu_app_id`, +/// `feishu_app_secret`, and `feishu_verification_token` are injected as config +/// keys `app_id`, `app_secret`, and `verification_token`. async fn inject_channel_secrets_into_config( channel_name: &str, secrets_store: &Option>, @@ -404,6 +410,7 @@ async fn inject_channel_secrets_into_config( "feishu" => &[ ("app_id", "feishu_app_id"), ("app_secret", "feishu_app_secret"), + ("verification_token", "feishu_verification_token"), ], _ => return, }; diff --git a/src/channels/web/handlers/chat.rs b/src/channels/web/handlers/chat.rs index bc4e3dbc..d1580f5c 100644 --- a/src/channels/web/handlers/chat.rs +++ b/src/channels/web/handlers/chat.rs @@ -15,7 +15,9 @@ use crate::channels::IncomingMessage; use crate::channels::web::auth::AuthenticatedUser; use crate::channels::web::server::GatewayState; use crate::channels::web::types::*; -use crate::channels::web::util::{build_turns_from_db_messages, truncate_preview}; +use crate::channels::web::util::{ + build_turns_from_db_messages, tool_error_for_display, truncate_preview, +}; pub async fn chat_send_handler( State(state): State>, @@ -397,7 +399,7 @@ pub async fn chat_history_handler( }; truncate_preview(&s, 500) }), - error: tc.error.clone(), + error: tc.error.as_deref().map(tool_error_for_display), rationale: tc.rationale.clone(), }) .collect(), @@ -533,7 +535,7 @@ pub async fn chat_threads_handler( // Fallback: in-memory only (no assistant thread without DB) let sess = session.lock().await; let mut sorted_threads: Vec<_> = sess.threads.values().collect(); - sorted_threads.sort_by(|a, b| b.updated_at.cmp(&a.updated_at)); + sorted_threads.sort_by_key(|t| std::cmp::Reverse(t.updated_at)); let threads: Vec = sorted_threads .into_iter() .map(|t| ThreadInfo { diff --git a/src/channels/web/mod.rs b/src/channels/web/mod.rs index fb17383d..6ff11915 100644 --- a/src/channels/web/mod.rs +++ b/src/channels/web/mod.rs @@ -18,6 +18,7 @@ pub mod auth; pub(crate) mod handlers; pub mod log_layer; pub mod openai_compat; +pub mod responses_api; pub mod server; pub mod sse; pub mod types; diff --git a/src/channels/web/responses_api.rs b/src/channels/web/responses_api.rs new file mode 100644 index 00000000..70850472 --- /dev/null +++ b/src/channels/web/responses_api.rs @@ -0,0 +1,1411 @@ +//! OpenAI Responses API (`POST /v1/responses`, `GET /v1/responses/{id}`). +//! +//! Unlike the Chat Completions proxy (`openai_compat.rs`) which is a raw LLM +//! passthrough, this module routes requests through the full agent loop — +//! giving callers access to tools, memory, safety, and server-side +//! conversation state via a standard OpenAI-compatible interface. + +use std::convert::Infallible; +use std::sync::Arc; +use std::time::Duration; + +use axum::{ + Json, + extract::{Path, State}, + http::StatusCode, + response::{ + IntoResponse, Response, + sse::{Event, KeepAlive, Sse}, + }, +}; +use futures::Stream; +use serde::{Deserialize, Serialize}; +use tokio_stream::StreamExt; +use uuid::Uuid; + +use crate::channels::IncomingMessage; +use crate::channels::web::types::AppEvent; + +use super::server::GatewayState; + +// --------------------------------------------------------------------------- +// Constants +// --------------------------------------------------------------------------- + +/// Maximum time to wait for the agent to finish a turn (non-streaming). +const RESPONSE_TIMEOUT: Duration = Duration::from_secs(120); + +/// Prefix for response IDs. +const RESP_PREFIX: &str = "resp_"; + +/// Length of a UUID in simple (no-hyphen) hex form. +const UUID_HEX_LEN: usize = 32; + +// --------------------------------------------------------------------------- +// Request types +// --------------------------------------------------------------------------- + +#[derive(Debug, Deserialize)] +pub struct ResponsesRequest { + #[serde(default = "default_model")] + pub model: String, + pub input: ResponsesInput, + #[serde(default)] + pub instructions: Option, + #[serde(default)] + pub previous_response_id: Option, + #[serde(default)] + pub stream: Option, + #[serde(default)] + pub temperature: Option, + #[serde(default)] + pub max_output_tokens: Option, + #[serde(default)] + pub tools: Option>, + #[serde(default)] + pub tool_choice: Option, +} + +fn default_model() -> String { + "default".to_string() +} + +#[derive(Debug, Deserialize)] +#[serde(untagged)] +pub enum ResponsesInput { + Text(String), + Messages(Vec), +} + +#[derive(Debug, Clone, Deserialize)] +pub struct ResponsesInputMessage { + pub role: String, + pub content: String, +} + +#[derive(Debug, Deserialize)] +pub struct ResponsesTool { + #[serde(rename = "type")] + pub tool_type: String, + #[serde(default)] + pub name: Option, + #[serde(default)] + pub description: Option, + #[serde(default)] + pub parameters: Option, +} + +// --------------------------------------------------------------------------- +// Response types +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone, Serialize)] +pub struct ResponseObject { + pub id: String, + pub object: &'static str, + pub created_at: i64, + pub model: String, + pub status: ResponseStatus, + pub output: Vec, + pub usage: ResponseUsage, + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, +} + +#[derive(Debug, Clone, Serialize)] +pub struct ResponseError { + pub message: String, + pub code: Option, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum ResponseStatus { + InProgress, + Completed, + Failed, +} + +#[derive(Debug, Clone, Serialize)] +#[serde(tag = "type")] +pub enum ResponseOutputItem { + #[serde(rename = "message")] + Message { + id: String, + role: String, + content: Vec, + }, + #[serde(rename = "function_call")] + FunctionCall { + id: String, + call_id: String, + name: String, + arguments: String, + }, + #[serde(rename = "function_call_output")] + FunctionCallOutput { + id: String, + call_id: String, + output: String, + }, +} + +#[derive(Debug, Clone, Serialize)] +#[serde(tag = "type")] +pub enum MessageContent { + #[serde(rename = "output_text")] + OutputText { text: String }, +} + +#[derive(Debug, Clone, Serialize, Default)] +pub struct ResponseUsage { + pub input_tokens: u64, + pub output_tokens: u64, + pub total_tokens: u64, +} + +// --------------------------------------------------------------------------- +// Streaming event types +// --------------------------------------------------------------------------- + +/// Server-sent events emitted during a streaming response. +/// +/// Each variant serialises with `"type": "response.xxx"` matching the OpenAI +/// Responses API wire format. +#[derive(Debug, Serialize)] +#[serde(tag = "type")] +pub enum ResponseStreamEvent { + #[serde(rename = "response.created")] + ResponseCreated { response: ResponseObject }, + + #[serde(rename = "response.in_progress")] + ResponseInProgress { response: ResponseObject }, + + #[serde(rename = "response.output_item.added")] + OutputItemAdded { + output_index: usize, + item: ResponseOutputItem, + }, + + #[serde(rename = "response.output_text.delta")] + OutputTextDelta { + output_index: usize, + content_index: usize, + delta: String, + }, + + #[serde(rename = "response.output_item.done")] + OutputItemDone { + output_index: usize, + item: ResponseOutputItem, + }, + + #[serde(rename = "response.completed")] + ResponseCompleted { response: ResponseObject }, + + #[serde(rename = "response.failed")] + ResponseFailed { response: ResponseObject }, +} + +// --------------------------------------------------------------------------- +// Error types +// --------------------------------------------------------------------------- + +#[derive(Debug, Serialize)] +pub struct ResponsesApiError { + pub error: ResponsesApiErrorDetail, +} + +#[derive(Debug, Serialize)] +pub struct ResponsesApiErrorDetail { + pub message: String, + #[serde(rename = "type")] + pub error_type: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub code: Option, +} + +type ApiError = (StatusCode, Json); + +fn api_error(status: StatusCode, message: impl Into, error_type: &str) -> ApiError { + ( + status, + Json(ResponsesApiError { + error: ResponsesApiErrorDetail { + message: message.into(), + error_type: error_type.to_string(), + code: None, + }, + }), + ) +} + +// --------------------------------------------------------------------------- +// ID encoding/decoding +// --------------------------------------------------------------------------- + +/// Encode a response ID: `resp_{response_uuid_hex}{thread_uuid_hex}`. +/// +/// Each POST generates a unique `response_uuid` so that response IDs differ +/// across turns even when the underlying thread (conversation) is the same. +fn encode_response_id(response_uuid: &Uuid, thread_uuid: &Uuid) -> String { + format!( + "{}{}{}", + RESP_PREFIX, + response_uuid.simple(), + thread_uuid.simple() + ) +} + +/// Decode a response ID back to `(response_uuid, thread_uuid)`. +fn decode_response_id(id: &str) -> Result<(Uuid, Uuid), String> { + let hex = id + .strip_prefix(RESP_PREFIX) + .ok_or_else(|| format!("response ID must start with '{RESP_PREFIX}'"))?; + if hex.len() != UUID_HEX_LEN * 2 { + return Err(format!( + "response ID must contain exactly {} hex characters after prefix", + UUID_HEX_LEN * 2 + )); + } + let (resp_hex, thread_hex) = hex.split_at(UUID_HEX_LEN); + let response_uuid = + Uuid::parse_str(resp_hex).map_err(|e| format!("invalid response UUID: {e}"))?; + let thread_uuid = + Uuid::parse_str(thread_hex).map_err(|e| format!("invalid thread UUID: {e}"))?; + Ok((response_uuid, thread_uuid)) +} + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +fn unix_timestamp() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() as i64 +} + +fn make_item_id() -> String { + format!("item_{}", Uuid::new_v4().simple()) +} + +/// Extract the user message text from the input. +fn extract_user_content(input: &ResponsesInput) -> Result { + match input { + ResponsesInput::Text(s) => { + if s.is_empty() { + Err("input must not be empty".to_string()) + } else { + Ok(s.clone()) + } + } + ResponsesInput::Messages(msgs) => { + // Find the last user message. + let last_user = msgs + .iter() + .rev() + .find(|m| m.role == "user") + .ok_or("input messages must contain at least one user message")?; + if last_user.content.is_empty() { + Err("user message content must not be empty".to_string()) + } else { + Ok(last_user.content.clone()) + } + } + } +} + +/// Check whether an `AppEvent` belongs to the target thread. +fn event_matches_thread(event: &AppEvent, target: &str) -> bool { + match event { + AppEvent::Response { thread_id, .. } => thread_id == target, + AppEvent::StreamChunk { thread_id, .. } + | AppEvent::Thinking { thread_id, .. } + | AppEvent::ToolStarted { thread_id, .. } + | AppEvent::ToolCompleted { thread_id, .. } + | AppEvent::ToolResult { thread_id, .. } + | AppEvent::Error { thread_id, .. } + | AppEvent::TurnCost { thread_id, .. } + | AppEvent::ImageGenerated { thread_id, .. } + | AppEvent::Suggestions { thread_id, .. } + | AppEvent::ReasoningUpdate { thread_id, .. } + | AppEvent::Status { thread_id, .. } + | AppEvent::ApprovalNeeded { thread_id, .. } => thread_id.as_deref() == Some(target), + // Global or job-scoped events are never matched. + _ => false, + } +} + +/// Build an empty in-progress response shell. +fn in_progress_response(resp_id: &str, model: &str) -> ResponseObject { + ResponseObject { + id: resp_id.to_string(), + object: "response", + created_at: unix_timestamp(), + model: model.to_string(), + status: ResponseStatus::InProgress, + output: Vec::new(), + usage: ResponseUsage::default(), + error: None, + } +} + +/// Send an `IncomingMessage` to the agent loop, returning an error response on +/// failure. +async fn send_to_agent(state: &GatewayState, msg: IncomingMessage) -> Result<(), ApiError> { + let tx = { + let guard = state.msg_tx.read().await; + guard.as_ref().cloned().ok_or_else(|| { + api_error( + StatusCode::SERVICE_UNAVAILABLE, + "Agent loop not started", + "server_error", + ) + })? + }; + tx.send(msg).await.map_err(|_| { + api_error( + StatusCode::INTERNAL_SERVER_ERROR, + "Agent loop channel closed", + "server_error", + ) + }) +} + +// --------------------------------------------------------------------------- +// Non-streaming: collect AppEvents into a ResponseObject +// --------------------------------------------------------------------------- + +/// Accumulator for building a `ResponseObject` from a stream of `AppEvent`s. +struct ResponseAccumulator { + resp_id: String, + model: String, + created_at: i64, + output: Vec, + text_chunks: Vec, + usage: ResponseUsage, + failed: bool, + error_message: Option, +} + +impl ResponseAccumulator { + fn new(resp_id: String, model: String) -> Self { + Self { + resp_id, + model, + created_at: unix_timestamp(), + output: Vec::new(), + text_chunks: Vec::new(), + usage: ResponseUsage::default(), + failed: false, + error_message: None, + } + } + + /// Process one `AppEvent` and return `true` if the turn is finished. + fn process(&mut self, event: AppEvent) -> bool { + match event { + AppEvent::StreamChunk { content, .. } => { + self.text_chunks.push(content); + false + } + AppEvent::Response { content, .. } => { + // Final response text supersedes any stream chunks. + let text = if content.is_empty() { + self.text_chunks.join("") + } else { + content + }; + if !text.is_empty() { + self.output.push(ResponseOutputItem::Message { + id: make_item_id(), + role: "assistant".to_string(), + content: vec![MessageContent::OutputText { text }], + }); + } + true // turn complete + } + AppEvent::ToolStarted { name, .. } => { + // Emit function_call placeholder — arguments filled on ToolCompleted. + let call_id = format!("call_{}", Uuid::new_v4().simple()); + self.output.push(ResponseOutputItem::FunctionCall { + id: make_item_id(), + call_id, + name, + arguments: String::new(), + }); + false + } + AppEvent::ToolCompleted { + name, + success, + error, + parameters, + .. + } => { + // Try to attach arguments to the matching FunctionCall. + if let Some(args) = parameters { + for item in self.output.iter_mut().rev() { + if let ResponseOutputItem::FunctionCall { + name: n, + arguments: a, + .. + } = item + && *n == name + && a.is_empty() + { + *a = args; + break; + } + } + } + // On failure, record a FunctionCallOutput with the error. + if !success && let Some(err) = error { + let call_id = self.last_call_id_for(&name); + self.output.push(ResponseOutputItem::FunctionCallOutput { + id: make_item_id(), + call_id, + output: format!("Error: {err}"), + }); + } + false + } + AppEvent::ToolResult { name, preview, .. } => { + let call_id = self.last_call_id_for(&name); + self.output.push(ResponseOutputItem::FunctionCallOutput { + id: make_item_id(), + call_id, + output: preview, + }); + false + } + AppEvent::TurnCost { + input_tokens, + output_tokens, + .. + } => { + self.usage = ResponseUsage { + input_tokens, + output_tokens, + total_tokens: input_tokens + output_tokens, + }; + false + } + AppEvent::Error { message, .. } => { + self.failed = true; + self.error_message = Some(message); + true // turn complete (failed) + } + AppEvent::ApprovalNeeded { tool_name, .. } => { + self.failed = true; + self.error_message = Some(format!( + "Tool '{tool_name}' requires approval which is not supported via the Responses API" + )); + true + } + // Ignore events we don't map (Thinking, Status, etc.). + _ => false, + } + } + + /// Find the `call_id` of the most recent `FunctionCall` for a given tool name. + fn last_call_id_for(&self, name: &str) -> String { + self.output + .iter() + .rev() + .find_map(|item| match item { + ResponseOutputItem::FunctionCall { + call_id, name: n, .. + } if n == name => Some(call_id.clone()), + _ => None, + }) + .unwrap_or_default() + } + + fn finish(self) -> ResponseObject { + ResponseObject { + id: self.resp_id, + object: "response", + created_at: self.created_at, + model: self.model, + status: if self.failed { + ResponseStatus::Failed + } else { + ResponseStatus::Completed + }, + output: self.output, + usage: self.usage, + error: self.error_message.map(|msg| ResponseError { + message: msg, + code: None, + }), + } + } +} + +// --------------------------------------------------------------------------- +// Handlers +// --------------------------------------------------------------------------- + +pub async fn create_response_handler( + State(state): State>, + super::auth::AuthenticatedUser(user): super::auth::AuthenticatedUser, + Json(req): Json, +) -> Result { + if !state.chat_rate_limiter.check(&user.user_id) { + return Err(api_error( + StatusCode::TOO_MANY_REQUESTS, + "Rate limit exceeded. Please try again later.", + "rate_limit_error", + )); + } + + // Reject fields that are accepted but not yet wired into the agent loop. + if req.model != "default" { + return Err(api_error( + StatusCode::BAD_REQUEST, + "Model selection is not yet supported; omit 'model' or use \"default\"", + "invalid_request_error", + )); + } + if req.instructions.is_some() { + return Err(api_error( + StatusCode::BAD_REQUEST, + "The 'instructions' field is not yet supported", + "invalid_request_error", + )); + } + if req.tools.is_some() { + return Err(api_error( + StatusCode::BAD_REQUEST, + "The 'tools' field is not yet supported", + "invalid_request_error", + )); + } + if req.tool_choice.is_some() { + return Err(api_error( + StatusCode::BAD_REQUEST, + "The 'tool_choice' field is not yet supported", + "invalid_request_error", + )); + } + if req.temperature.is_some() { + return Err(api_error( + StatusCode::BAD_REQUEST, + "The 'temperature' field is not yet supported", + "invalid_request_error", + )); + } + if req.max_output_tokens.is_some() { + return Err(api_error( + StatusCode::BAD_REQUEST, + "The 'max_output_tokens' field is not yet supported", + "invalid_request_error", + )); + } + + let content = extract_user_content(&req.input) + .map_err(|e| api_error(StatusCode::BAD_REQUEST, e, "invalid_request_error"))?; + + // Resolve or create thread. + let thread_uuid = match &req.previous_response_id { + Some(prev_id) => { + let (_prev_resp, thread) = decode_response_id(prev_id) + .map_err(|e| api_error(StatusCode::BAD_REQUEST, e, "invalid_request_error"))?; + thread + } + None => Uuid::new_v4(), + }; + let thread_id_str = thread_uuid.to_string(); + + // Each POST gets its own unique response UUID. + let response_uuid = Uuid::new_v4(); + + // Build the message for the agent loop. + let msg = IncomingMessage::new("gateway", &user.user_id, &content) + .with_thread(&thread_id_str) + .with_metadata(serde_json::json!({ + "thread_id": &thread_id_str, + "user_id": &user.user_id, + "source": "responses_api", + })); + + let resp_id = encode_response_id(&response_uuid, &thread_uuid); + let model = req.model.clone(); + let stream = req.stream.unwrap_or(false); + let user_id = user.user_id.clone(); + + if stream { + handle_streaming(state, msg, resp_id, model, thread_id_str, user_id) + .await + .map(IntoResponse::into_response) + } else { + handle_non_streaming(state, msg, resp_id, model, thread_id_str, &user_id) + .await + .map(IntoResponse::into_response) + } +} + +async fn handle_non_streaming( + state: Arc, + msg: IncomingMessage, + resp_id: String, + model: String, + thread_id: String, + user_id: &str, +) -> Result, ApiError> { + // Subscribe BEFORE sending so we don't miss events. + let mut event_stream = state + .sse + .subscribe_raw(Some(user_id.to_string())) + .ok_or_else(|| { + api_error( + StatusCode::SERVICE_UNAVAILABLE, + "Too many concurrent connections", + "server_error", + ) + })?; + + send_to_agent(&state, msg).await?; + + let mut acc = ResponseAccumulator::new(resp_id, model); + + let result = tokio::time::timeout(RESPONSE_TIMEOUT, async { + while let Some(event) = event_stream.next().await { + if !event_matches_thread(&event, &thread_id) { + continue; + } + if acc.process(event) { + break; + } + } + }) + .await; + + if result.is_err() { + acc.failed = true; + acc.error_message = Some("Response timed out".to_string()); + } + + Ok(Json(acc.finish())) +} + +async fn handle_streaming( + state: Arc, + msg: IncomingMessage, + resp_id: String, + model: String, + thread_id: String, + user_id: String, +) -> Result> + Send>, ApiError> { + let event_stream = state.sse.subscribe_raw(Some(user_id)).ok_or_else(|| { + api_error( + StatusCode::SERVICE_UNAVAILABLE, + "Too many concurrent connections", + "server_error", + ) + })?; + + send_to_agent(&state, msg).await?; + + // Use a channel to bridge the spawned task and the SSE stream. + let (tx, rx) = tokio::sync::mpsc::channel::(64); + + tokio::spawn(streaming_worker( + tx, + event_stream, + resp_id, + model, + thread_id, + )); + + let stream = tokio_stream::wrappers::ReceiverStream::new(rx).map(Ok::<_, Infallible>); + + Ok(Sse::new(stream).keep_alive(KeepAlive::new().interval(Duration::from_secs(15)).text(""))) +} + +/// Background task that reads `AppEvent`s and sends SSE `Event`s to the client. +async fn streaming_worker( + tx: tokio::sync::mpsc::Sender, + event_stream: impl Stream + Send + Unpin, + resp_id: String, + model: String, + thread_id: String, +) { + use std::pin::pin; + + fn sse_event(evt_type: &str, data: &str) -> Event { + Event::default().event(evt_type).data(data) + } + + fn emit( + tx: &tokio::sync::mpsc::Sender, + evt_type: &str, + payload: &impl Serialize, + ) -> bool { + if let Ok(data) = serde_json::to_string(payload) { + tx.try_send(sse_event(evt_type, &data)).is_ok() + } else { + true // serialization failure is non-fatal; keep going + } + } + + // Emit response.created + let initial = in_progress_response(&resp_id, &model); + if !emit( + &tx, + "response.created", + &ResponseStreamEvent::ResponseCreated { response: initial }, + ) { + return; + } + + let mut acc = ResponseAccumulator::new(resp_id, model); + let mut message_output_index: Option = None; + let mut current_tool_index: Option = None; + + let mut event_stream = pin!(event_stream); + let timeout = tokio::time::sleep(RESPONSE_TIMEOUT); + tokio::pin!(timeout); + + loop { + let event = tokio::select! { + biased; + ev = event_stream.next() => match ev { + Some(e) => e, + None => break, + }, + () = &mut timeout => { + acc.failed = true; + let resp = acc.finish(); + let _ = emit(&tx, "response.failed", &ResponseStreamEvent::ResponseFailed { response: resp }); + return; + } + }; + + if !event_matches_thread(&event, &thread_id) { + continue; + } + + match &event { + AppEvent::StreamChunk { content, .. } => { + let idx = match message_output_index { + Some(i) => i, + None => { + let i = acc.output.len(); + let item = ResponseOutputItem::Message { + id: make_item_id(), + role: "assistant".to_string(), + content: vec![MessageContent::OutputText { + text: String::new(), + }], + }; + emit( + &tx, + "response.output_item.added", + &ResponseStreamEvent::OutputItemAdded { + output_index: i, + item: item.clone(), + }, + ); + acc.output.push(item); + message_output_index = Some(i); + i + } + }; + emit( + &tx, + "response.output_text.delta", + &ResponseStreamEvent::OutputTextDelta { + output_index: idx, + content_index: 0, + delta: content.clone(), + }, + ); + acc.text_chunks.push(content.clone()); + } + AppEvent::ToolStarted { name, .. } => { + let idx = acc.output.len(); + let call_id = format!("call_{}", Uuid::new_v4().simple()); + let item = ResponseOutputItem::FunctionCall { + id: make_item_id(), + call_id, + name: name.clone(), + arguments: String::new(), + }; + emit( + &tx, + "response.output_item.added", + &ResponseStreamEvent::OutputItemAdded { + output_index: idx, + item: item.clone(), + }, + ); + acc.output.push(item); + current_tool_index = Some(idx); + } + AppEvent::ToolCompleted { + name, + success, + error, + parameters, + .. + } => { + if let Some(args) = parameters { + for item in acc.output.iter_mut().rev() { + if let ResponseOutputItem::FunctionCall { + name: n, + arguments: a, + .. + } = item + && *n == *name + && a.is_empty() + { + *a = args.clone(); + break; + } + } + } + if let Some(idx) = current_tool_index.take() + && let Some(item) = acc.output.get(idx) + { + emit( + &tx, + "response.output_item.done", + &ResponseStreamEvent::OutputItemDone { + output_index: idx, + item: item.clone(), + }, + ); + } + // On failure, emit a FunctionCallOutput with the error. + if !*success && let Some(err) = error { + let call_id = acc.last_call_id_for(name); + let idx = acc.output.len(); + let item = ResponseOutputItem::FunctionCallOutput { + id: make_item_id(), + call_id, + output: format!("Error: {err}"), + }; + emit( + &tx, + "response.output_item.added", + &ResponseStreamEvent::OutputItemAdded { + output_index: idx, + item: item.clone(), + }, + ); + emit( + &tx, + "response.output_item.done", + &ResponseStreamEvent::OutputItemDone { + output_index: idx, + item: item.clone(), + }, + ); + acc.output.push(item); + } + } + AppEvent::ToolResult { name, preview, .. } => { + let call_id = acc.last_call_id_for(name); + let idx = acc.output.len(); + let item = ResponseOutputItem::FunctionCallOutput { + id: make_item_id(), + call_id, + output: preview.clone(), + }; + emit( + &tx, + "response.output_item.added", + &ResponseStreamEvent::OutputItemAdded { + output_index: idx, + item: item.clone(), + }, + ); + emit( + &tx, + "response.output_item.done", + &ResponseStreamEvent::OutputItemDone { + output_index: idx, + item: item.clone(), + }, + ); + acc.output.push(item); + } + AppEvent::TurnCost { + input_tokens, + output_tokens, + .. + } => { + acc.usage = ResponseUsage { + input_tokens: *input_tokens, + output_tokens: *output_tokens, + total_tokens: input_tokens + output_tokens, + }; + } + _ => {} + } + + // Terminal events. + let is_terminal = matches!( + &event, + AppEvent::Response { .. } | AppEvent::Error { .. } | AppEvent::ApprovalNeeded { .. } + ); + + if is_terminal { + if let AppEvent::Response { content, .. } = &event { + let text = if content.is_empty() { + acc.text_chunks.join("") + } else { + content.clone() + }; + if !text.is_empty() { + match message_output_index { + Some(idx) => { + acc.output[idx] = ResponseOutputItem::Message { + id: make_item_id(), + role: "assistant".to_string(), + content: vec![MessageContent::OutputText { text }], + }; + if let Some(item) = acc.output.get(idx) { + emit( + &tx, + "response.output_item.done", + &ResponseStreamEvent::OutputItemDone { + output_index: idx, + item: item.clone(), + }, + ); + } + } + None => { + let idx = acc.output.len(); + let item = ResponseOutputItem::Message { + id: make_item_id(), + role: "assistant".to_string(), + content: vec![MessageContent::OutputText { text }], + }; + emit( + &tx, + "response.output_item.added", + &ResponseStreamEvent::OutputItemAdded { + output_index: idx, + item: item.clone(), + }, + ); + emit( + &tx, + "response.output_item.done", + &ResponseStreamEvent::OutputItemDone { + output_index: idx, + item: item.clone(), + }, + ); + acc.output.push(item); + } + } + } + } + + if matches!( + &event, + AppEvent::Error { .. } | AppEvent::ApprovalNeeded { .. } + ) { + acc.process(event); + } + + let resp = acc.finish(); + let (evt_type, evt) = if resp.status == ResponseStatus::Failed { + ( + "response.failed", + ResponseStreamEvent::ResponseFailed { response: resp }, + ) + } else { + ( + "response.completed", + ResponseStreamEvent::ResponseCompleted { response: resp }, + ) + }; + let _ = emit(&tx, evt_type, &evt); + return; + } + } +} + +// --------------------------------------------------------------------------- +// GET /v1/responses/{id} +// --------------------------------------------------------------------------- + +pub async fn get_response_handler( + State(state): State>, + super::auth::AuthenticatedUser(user): super::auth::AuthenticatedUser, + Path(id): Path, +) -> Result, ApiError> { + let (_response_uuid, thread_uuid) = decode_response_id(&id) + .map_err(|e| api_error(StatusCode::BAD_REQUEST, e, "invalid_request_error"))?; + + let store = state.store.as_ref().ok_or_else(|| { + api_error( + StatusCode::SERVICE_UNAVAILABLE, + "Database not configured", + "server_error", + ) + })?; + + // Verify the authenticated user owns this conversation. + let owns = store + .conversation_belongs_to_user(thread_uuid, &user.user_id) + .await + .map_err(|e| { + api_error( + StatusCode::INTERNAL_SERVER_ERROR, + format!("Failed to verify ownership: {e}"), + "server_error", + ) + })?; + if !owns { + return Err(api_error( + StatusCode::NOT_FOUND, + format!("Response '{id}' not found"), + "invalid_request_error", + )); + } + + // Load messages for this conversation. + let messages = store + .list_conversation_messages(thread_uuid) + .await + .map_err(|e| { + api_error( + StatusCode::INTERNAL_SERVER_ERROR, + format!("Failed to load conversation: {e}"), + "server_error", + ) + })?; + + if messages.is_empty() { + return Err(api_error( + StatusCode::NOT_FOUND, + format!("Response '{id}' not found"), + "invalid_request_error", + )); + } + + // Reconstruct output items from stored messages. + let mut output = Vec::new(); + for msg in &messages { + match msg.role.as_str() { + "assistant" => { + if !msg.content.is_empty() { + output.push(ResponseOutputItem::Message { + id: format!("msg_{}", msg.id.simple()), + role: "assistant".to_string(), + content: vec![MessageContent::OutputText { + text: msg.content.clone(), + }], + }); + } + } + "tool_calls" => { + // Tool calls may be stored as a plain JSON array (legacy) or + // as an object wrapper: `{ "calls": [...], "narrative": "..." }`. + let calls = match serde_json::from_str::(&msg.content) { + Ok(serde_json::Value::Array(arr)) => arr, + Ok(serde_json::Value::Object(ref obj)) => obj + .get("calls") + .and_then(|v| v.as_array()) + .cloned() + .unwrap_or_default(), + _ => Vec::new(), + }; + for call in &calls { + let name = call + .get("name") + .and_then(|v| v.as_str()) + .unwrap_or("unknown") + .to_string(); + // Prefer `call_id`, fall back to `tool_call_id`, then `id`. + let call_id = call + .get("call_id") + .or_else(|| call.get("tool_call_id")) + .or_else(|| call.get("id")) + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + let arguments = call + .get("parameters") + .or_else(|| call.get("arguments")) + .map(|v| { + if v.is_string() { + v.as_str().unwrap_or("{}").to_string() + } else { + serde_json::to_string(v).unwrap_or_default() + } + }) + .unwrap_or_default(); + output.push(ResponseOutputItem::FunctionCall { + id: make_item_id(), + call_id: call_id.clone(), + name, + arguments, + }); + // If there's an inline result, emit a FunctionCallOutput too. + if let Some(result) = call + .get("result_preview") + .or_else(|| call.get("result")) + .and_then(|v| v.as_str()) + { + output.push(ResponseOutputItem::FunctionCallOutput { + id: make_item_id(), + call_id, + output: result.to_string(), + }); + } + } + } + "tool" => { + // Tool results — try to correlate with the preceding FunctionCall. + let call_id = output + .iter() + .rev() + .find_map(|item| match item { + ResponseOutputItem::FunctionCall { call_id, .. } => Some(call_id.clone()), + _ => None, + }) + .unwrap_or_default(); + output.push(ResponseOutputItem::FunctionCallOutput { + id: make_item_id(), + call_id, + output: msg.content.clone(), + }); + } + _ => {} // Skip user/system messages (they are input, not output). + } + } + + Ok(Json(ResponseObject { + id, + object: "response", + created_at: messages + .first() + .map(|m| m.created_at.timestamp()) + .unwrap_or_else(unix_timestamp), + model: "default".to_string(), + status: ResponseStatus::Completed, + output, + usage: ResponseUsage::default(), // Token usage is not persisted per-message. + error: None, + })) +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn response_id_round_trip() { + let resp_uuid = Uuid::new_v4(); + let thread_uuid = Uuid::new_v4(); + let encoded = encode_response_id(&resp_uuid, &thread_uuid); + assert!(encoded.starts_with(RESP_PREFIX)); + let (decoded_resp, decoded_thread) = decode_response_id(&encoded).expect("should decode"); + assert_eq!(resp_uuid, decoded_resp); + assert_eq!(thread_uuid, decoded_thread); + } + + #[test] + fn response_ids_differ_across_turns() { + let thread_uuid = Uuid::new_v4(); + let id1 = encode_response_id(&Uuid::new_v4(), &thread_uuid); + let id2 = encode_response_id(&Uuid::new_v4(), &thread_uuid); + assert_ne!(id1, id2, "each turn must produce a distinct response ID"); + } + + #[test] + fn decode_response_id_rejects_bad_prefix() { + assert!(decode_response_id("bad_prefix").is_err()); + } + + #[test] + fn decode_response_id_rejects_bad_uuid() { + assert!(decode_response_id("resp_not_a_uuid").is_err()); + } + + #[test] + fn extract_user_content_text() { + let input = ResponsesInput::Text("hello".to_string()); + assert_eq!(extract_user_content(&input).unwrap(), "hello"); + } + + #[test] + fn extract_user_content_empty_text_errors() { + let input = ResponsesInput::Text(String::new()); + assert!(extract_user_content(&input).is_err()); + } + + #[test] + fn extract_user_content_messages_uses_last_user() { + let input = ResponsesInput::Messages(vec![ + ResponsesInputMessage { + role: "user".to_string(), + content: "first".to_string(), + }, + ResponsesInputMessage { + role: "assistant".to_string(), + content: "middle".to_string(), + }, + ResponsesInputMessage { + role: "user".to_string(), + content: "last".to_string(), + }, + ]); + assert_eq!(extract_user_content(&input).unwrap(), "last"); + } + + #[test] + fn extract_user_content_no_user_message_errors() { + let input = ResponsesInput::Messages(vec![ResponsesInputMessage { + role: "system".to_string(), + content: "hello".to_string(), + }]); + assert!(extract_user_content(&input).is_err()); + } + + #[test] + fn event_matches_thread_filters_correctly() { + let target = "abc-123"; + let matching = AppEvent::Response { + content: "hi".to_string(), + thread_id: "abc-123".to_string(), + }; + assert!(event_matches_thread(&matching, target)); + + let non_matching = AppEvent::Response { + content: "hi".to_string(), + thread_id: "other".to_string(), + }; + assert!(!event_matches_thread(&non_matching, target)); + + let global = AppEvent::Heartbeat; + assert!(!event_matches_thread(&global, target)); + } + + #[test] + fn accumulator_basic_response() { + let mut acc = ResponseAccumulator::new("resp_test".to_string(), "m".to_string()); + let done = acc.process(AppEvent::Response { + content: "Hello world".to_string(), + thread_id: "t".to_string(), + }); + assert!(done); + let resp = acc.finish(); + assert_eq!(resp.status, ResponseStatus::Completed); + assert_eq!(resp.output.len(), 1); + match &resp.output[0] { + ResponseOutputItem::Message { content, .. } => { + assert!( + matches!(&content[0], MessageContent::OutputText { text } if text == "Hello world") + ); + } + _ => panic!("expected Message output item"), + } + } + + #[test] + fn accumulator_stream_chunks_then_response() { + let mut acc = ResponseAccumulator::new("resp_test".to_string(), "m".to_string()); + assert!(!acc.process(AppEvent::StreamChunk { + content: "Hello ".to_string(), + thread_id: Some("t".to_string()), + })); + assert!(!acc.process(AppEvent::StreamChunk { + content: "world".to_string(), + thread_id: Some("t".to_string()), + })); + // Empty response content → accumulator falls back to chunks. + assert!(acc.process(AppEvent::Response { + content: String::new(), + thread_id: "t".to_string(), + })); + let resp = acc.finish(); + match &resp.output[0] { + ResponseOutputItem::Message { content, .. } => { + assert!( + matches!(&content[0], MessageContent::OutputText { text } if text == "Hello world") + ); + } + _ => panic!("expected Message output item"), + } + } + + #[test] + fn accumulator_tool_flow() { + let mut acc = ResponseAccumulator::new("resp_test".to_string(), "m".to_string()); + assert!(!acc.process(AppEvent::ToolStarted { + name: "memory_search".to_string(), + thread_id: Some("t".to_string()), + })); + assert!(!acc.process(AppEvent::ToolResult { + name: "memory_search".to_string(), + preview: "found 3 results".to_string(), + thread_id: Some("t".to_string()), + })); + assert!(acc.process(AppEvent::Response { + content: "Here are your results.".to_string(), + thread_id: "t".to_string(), + })); + let resp = acc.finish(); + // FunctionCall + FunctionCallOutput + Message = 3 items + assert_eq!(resp.output.len(), 3); + assert!( + matches!(&resp.output[0], ResponseOutputItem::FunctionCall { name, .. } if name == "memory_search") + ); + assert!( + matches!(&resp.output[1], ResponseOutputItem::FunctionCallOutput { output, .. } if output == "found 3 results") + ); + assert!(matches!( + &resp.output[2], + ResponseOutputItem::Message { .. } + )); + } + + #[test] + fn accumulator_error_marks_failed() { + let mut acc = ResponseAccumulator::new("resp_test".to_string(), "m".to_string()); + assert!(acc.process(AppEvent::Error { + message: "something broke".to_string(), + thread_id: Some("t".to_string()), + })); + let resp = acc.finish(); + assert_eq!(resp.status, ResponseStatus::Failed); + } + + #[test] + fn accumulator_approval_needed_marks_failed() { + let mut acc = ResponseAccumulator::new("resp_test".to_string(), "m".to_string()); + assert!(acc.process(AppEvent::ApprovalNeeded { + request_id: "r1".to_string(), + tool_name: "shell".to_string(), + description: "run ls".to_string(), + parameters: "{}".to_string(), + thread_id: Some("t".to_string()), + allow_always: true, + })); + let resp = acc.finish(); + assert_eq!(resp.status, ResponseStatus::Failed); + } + + #[test] + fn response_status_serializes_as_snake_case() { + let json = serde_json::to_string(&ResponseStatus::InProgress).expect("serialize"); + assert_eq!(json, "\"in_progress\""); + let json = serde_json::to_string(&ResponseStatus::Completed).expect("serialize"); + assert_eq!(json, "\"completed\""); + } +} diff --git a/src/channels/web/server.rs b/src/channels/web/server.rs index f46a5367..051b6774 100644 --- a/src/channels/web/server.rs +++ b/src/channels/web/server.rs @@ -534,6 +534,15 @@ pub async fn start_server( post(super::openai_compat::chat_completions_handler), ) .route("/v1/models", get(super::openai_compat::models_handler)) + // OpenAI Responses API (routes through the full agent loop) + .route( + "/v1/responses", + post(super::responses_api::create_response_handler), + ) + .route( + "/v1/responses/{id}", + get(super::responses_api::get_response_handler), + ) .route_layer(middleware::from_fn_with_state( auth_state.clone(), auth_middleware, @@ -1906,7 +1915,7 @@ async fn chat_threads_handler( // Fallback: in-memory only (no assistant thread without DB) let mut sorted_threads: Vec<_> = sess.threads.values().collect(); - sorted_threads.sort_by(|a, b| b.updated_at.cmp(&a.updated_at)); + sorted_threads.sort_by_key(|t| std::cmp::Reverse(t.updated_at)); let threads: Vec = sorted_threads .into_iter() .map(|t| ThreadInfo { @@ -2226,7 +2235,7 @@ async fn extensions_activate_handler( AuthenticatedUser(user): AuthenticatedUser, Path(name): Path, ) -> Result, (StatusCode, String)> { - tracing::debug!( + tracing::trace!( extension = %name, user_id = %user.user_id, "extensions_activate_handler: received activate request" @@ -2260,7 +2269,7 @@ async fn extensions_activate_handler( crate::extensions::ExtensionError::AuthRequired ); - tracing::debug!( + tracing::trace!( extension = %name, error = %activate_err, needs_auth = needs_auth, @@ -2274,7 +2283,7 @@ async fn extensions_activate_handler( // Activation failed due to auth; try authenticating first. match ext_mgr.auth(&name, &user.user_id).await { Ok(auth_result) if auth_result.is_authenticated() => { - tracing::debug!( + tracing::trace!( extension = %name, "extensions_activate_handler: auth reports authenticated, retrying activate" ); diff --git a/src/channels/web/util.rs b/src/channels/web/util.rs index 2e4ffe3b..1ee8e229 100644 --- a/src/channels/web/util.rs +++ b/src/channels/web/util.rs @@ -4,6 +4,11 @@ use crate::channels::web::types::{ToolCallInfo, TurnInfo}; pub use ironclaw_common::truncate_preview; +/// Convert stored tool errors into plain text suitable for UI display. +pub fn tool_error_for_display(error: &str) -> String { + ironclaw_safety::SafetyLayer::unwrap_tool_output(error).unwrap_or_else(|| error.to_string()) +} + /// Parse tool call summary JSON objects into `ToolCallInfo` structs. fn parse_tool_call_infos(calls: &[serde_json::Value]) -> Vec { calls @@ -13,7 +18,7 @@ fn parse_tool_call_infos(calls: &[serde_json::Value]) -> Vec { has_result: c.get("result_preview").is_some_and(|v| !v.is_null()), has_error: c.get("error").is_some_and(|v| !v.is_null()), result_preview: c["result_preview"].as_str().map(String::from), - error: c["error"].as_str().map(String::from), + error: c["error"].as_str().map(tool_error_for_display), rationale: c["rationale"].as_str().map(String::from), }) .collect() @@ -181,6 +186,29 @@ mod tests { assert_eq!(turns[0].response.as_deref(), Some("Done")); } + #[test] + fn test_build_turns_unwrap_wrapped_tool_error_for_display() { + let tc_json = serde_json::json!([ + { + "name": "http", + "error": "\nTool 'http' failed: timeout\n" + } + ]); + let messages = vec![ + make_msg("user", "Run it", 0), + make_msg("tool_calls", &tc_json.to_string(), 500), + ]; + + let turns = build_turns_from_db_messages(&messages); + + assert_eq!(turns.len(), 1); + assert_eq!(turns[0].tool_calls.len(), 1); + assert_eq!( + turns[0].tool_calls[0].error.as_deref(), + Some("Tool 'http' failed: timeout") + ); + } + #[test] fn test_build_turns_malformed_tool_calls() { let messages = vec![ diff --git a/src/extensions/manager.rs b/src/extensions/manager.rs index 55b1e96d..8c0534f2 100644 --- a/src/extensions/manager.rs +++ b/src/extensions/manager.rs @@ -690,7 +690,7 @@ impl ExtensionManager { && parsed.username().is_empty() && parsed.password().is_none() => { - tracing::debug!( + tracing::trace!( extension = %name, relay_url_host = %parsed.host_str().unwrap_or("unknown"), "effective_relay_url: using per-extension override from settings" @@ -968,7 +968,7 @@ impl ExtensionManager { match store.get_setting(&self.user_id, &key).await { Ok(Some(v)) => { let has_id = v.as_str().is_some_and(|s| !s.is_empty()); - tracing::debug!( + tracing::trace!( extension = %name, has_team_id = has_id, "has_stored_team_id: checked store" @@ -976,7 +976,7 @@ impl ExtensionManager { return has_id; } Ok(None) => { - tracing::debug!( + tracing::trace!( extension = %name, "has_stored_team_id: no team_id setting found" ); @@ -4292,7 +4292,7 @@ impl ExtensionManager { name: &str, user_id: &str, ) -> Result { - tracing::debug!( + tracing::trace!( extension = %name, user_id = %user_id, "auth_channel_relay: starting" @@ -4306,14 +4306,14 @@ impl ExtensionManager { // to "authenticated" even when no team_id exists, preventing the OAuth // flow from being offered to the user. if self.has_stored_team_id(name, user_id).await { - tracing::debug!( + tracing::trace!( extension = %name, "auth_channel_relay: already authenticated (team_id in store)" ); return Ok(AuthResult::authenticated(name, ExtensionKind::ChannelRelay)); } - tracing::debug!( + tracing::trace!( extension = %name, "auth_channel_relay: no stored team_id, initiating OAuth" ); @@ -4335,7 +4335,7 @@ impl ExtensionManager { .await .unwrap_or_else(|| relay_config.url.clone()); - tracing::debug!( + tracing::trace!( extension = %name, relay_url = %effective_url, "auth_channel_relay: creating relay client for OAuth" @@ -4377,7 +4377,7 @@ impl ExtensionManager { // Channel-relay derives all URLs from trusted instance_url in chat-api. // We only pass the nonce for CSRF validation on the callback. - tracing::debug!( + tracing::trace!( extension = %name, relay_url = %effective_url, "auth_channel_relay: calling initiate_oauth on channel-relay" @@ -4413,7 +4413,7 @@ impl ExtensionManager { name: &str, user_id: &str, ) -> Result { - tracing::debug!( + tracing::trace!( extension = %name, user_id = %user_id, "activate_channel_relay: starting" @@ -4426,7 +4426,7 @@ impl ExtensionManager { match store.get_setting(user_id, &team_id_key).await { Ok(Some(v)) => { let id = v.as_str().map(|s| s.to_string()).unwrap_or_default(); - tracing::debug!( + tracing::trace!( extension = %name, team_id_empty = id.is_empty(), "activate_channel_relay: loaded team_id from store" @@ -4434,7 +4434,7 @@ impl ExtensionManager { id } Ok(None) => { - tracing::debug!( + tracing::trace!( extension = %name, setting_key = %team_id_key, "activate_channel_relay: no team_id in settings store" @@ -4451,7 +4451,7 @@ impl ExtensionManager { } } } else { - tracing::debug!( + tracing::trace!( extension = %name, "activate_channel_relay: no settings store available" ); @@ -4459,7 +4459,7 @@ impl ExtensionManager { }; if team_id.is_empty() { - tracing::debug!( + tracing::trace!( extension = %name, "activate_channel_relay: team_id is empty, returning AuthRequired" ); @@ -4482,7 +4482,7 @@ impl ExtensionManager { .await .unwrap_or_else(|| relay_config.url.clone()); - tracing::debug!( + tracing::trace!( extension = %name, relay_url = %effective_url, "activate_channel_relay: relay config loaded" @@ -4507,7 +4507,7 @@ impl ExtensionManager { // Fetch the per-instance signing secret from channel-relay. // This must succeed — there is no fallback. - tracing::debug!( + tracing::trace!( extension = %name, relay_url = %effective_url, "activate_channel_relay: fetching signing secret from channel-relay" diff --git a/src/llm/reasoning.rs b/src/llm/reasoning.rs index 6e078ac7..a0852cef 100644 --- a/src/llm/reasoning.rs +++ b/src/llm/reasoning.rs @@ -1376,9 +1376,18 @@ fn overlaps_code_region(start: usize, end: usize, regions: &[CodeRegion]) -> boo } /// Return the byte bounds of the line containing `pos`, excluding the trailing newline. +/// +/// `pos` is clamped to `text.len()` and adjusted to the nearest char boundary, +/// so callers need not guarantee that `pos` falls on a boundary. fn line_bounds(text: &str, pos: usize) -> (usize, usize) { - let start = text[..pos].rfind('\n').map_or(0, |idx| idx + 1); - let end = text[pos..].find('\n').map_or(text.len(), |idx| pos + idx); + let pos = pos.min(text.len()); + // Walk backward to find a valid char boundary (at most 3 bytes for UTF-8). + let mut safe = pos; + while safe > 0 && !text.is_char_boundary(safe) { + safe -= 1; + } + let start = text[..safe].rfind('\n').map_or(0, |idx| idx + 1); + let end = text[safe..].find('\n').map_or(text.len(), |idx| safe + idx); (start, end) } @@ -2302,6 +2311,51 @@ That's my plan."#; assert_eq!(regions[0].end, text.len()); } + // ---- line_bounds UTF-8 safety (issue #1669) ---- + + #[test] + fn test_line_bounds_ascii() { + let text = "hello\nworld\n"; + assert_eq!(line_bounds(text, 0), (0, 5)); + assert_eq!(line_bounds(text, 6), (6, 11)); + } + + #[test] + fn test_line_bounds_at_text_len() { + let text = "abc"; + assert_eq!(line_bounds(text, 3), (0, 3)); + } + + #[test] + fn test_line_bounds_mid_multibyte_char() { + // '🔥' is 4 bytes (F0 9F 94 A5). Passing pos=1 lands inside the char. + // line_bounds must not panic — it should snap to a valid boundary. + let text = "🔥\n"; + // All mid-char positions should snap back to byte 0 (start of '🔥'), + // so line bounds cover the first line: "🔥" = bytes 0..4. + assert_eq!(line_bounds(text, 1), (0, 4)); // would panic before fix + assert_eq!(line_bounds(text, 2), (0, 4)); + assert_eq!(line_bounds(text, 3), (0, 4)); + } + + #[test] + fn test_line_bounds_emoji_before_newline() { + // 'Result: 🔥\n' — end.saturating_sub(1) from the \n position + // should not panic even with multi-byte chars on the same line. + let text = "Result: 🔥\n"; + let newline_pos = text.find('\n').unwrap(); + // saturating_sub(1) lands inside '🔥' (byte 11 → 10, but char ends at 12). + // Snaps back to byte 8 (start of '🔥'), line covers "Result: 🔥" = bytes 0..12. + assert_eq!(line_bounds(text, newline_pos.saturating_sub(1)), (0, 12)); + } + + #[test] + fn test_line_bounds_pos_beyond_len() { + let text = "abc"; + // pos > text.len() should be clamped, not panic + assert_eq!(line_bounds(text, 100), (0, 3)); + } + // ---- recover_tool_calls_from_content tests ---- fn make_tools(names: &[&str]) -> Vec { diff --git a/src/tools/builder/core.rs b/src/tools/builder/core.rs index d4e10e95..8aa89aec 100644 --- a/src/tools/builder/core.rs +++ b/src/tools/builder/core.rs @@ -46,6 +46,22 @@ use crate::llm::{ use crate::tools::tool::{ApprovalRequirement, Tool, ToolError, ToolOutput}; use crate::tools::{ToolRegistry, prepare_tool_params}; +fn process_builder_tool_result( + tool_name: &str, + tool_call_id: &str, + result: &Result, +) -> (String, ChatMessage) { + static SAFETY: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + crate::safety::SafetyLayer::new(&crate::config::SafetyConfig { + max_output_length: 100_000, + injection_check_enabled: true, + }) + }); + + crate::tools::execute::process_tool_result(&SAFETY, tool_name, tool_call_id, result) +} + /// Requirement specification for building software. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct BuildRequirement { @@ -710,13 +726,13 @@ Create alongside the .wasm file to grant capabilities: Ok(output) => { let output_str = serde_json::to_string_pretty(&output.result) .unwrap_or_default(); + let llm_result: Result = + Ok(output_str.clone()); + let (_, tool_message) = + process_builder_tool_result(&tc.name, &tc.id, &llm_result); // Add to context - reason_ctx.messages.push(ChatMessage::tool_result( - &tc.id, - &tc.name, - output_str.clone(), - )); + reason_ctx.messages.push(tool_message); // Update phase based on tool current_phase = match tc.name.as_str() { @@ -742,12 +758,11 @@ Create alongside the .wasm file to grant capabilities: Err(e) => { let error_msg = format!("Tool error: {}", e); last_error = Some(error_msg.clone()); + let llm_result: Result = Err(&e); + let (_, tool_message) = + process_builder_tool_result(&tc.name, &tc.id, &llm_result); - reason_ctx.messages.push(ChatMessage::tool_result( - &tc.id, - &tc.name, - format!("Error: {}", e), - )); + reason_ctx.messages.push(tool_message); logs.push(BuildLog { timestamp: Utc::now(), @@ -1234,6 +1249,31 @@ mod tests { ); } + #[test] + fn test_process_builder_tool_result_wraps_success_output() { + let result: Result = + Ok("builder override".to_string()); + + let (content, message) = super::process_builder_tool_result("shell", "call_1", &result); + + assert!(content.contains("tool_output")); + assert!(!content.contains("\n")); + assert_eq!(message.content, content); + } + + #[test] + fn test_process_builder_tool_result_wraps_error_output() { + let result: Result = + Err("builder override".to_string()); + + let (content, message) = super::process_builder_tool_result("shell", "call_1", &result); + + assert!(content.contains("tool_output")); + assert!(content.contains("Tool 'shell' failed:")); + assert!(!content.contains("\n")); + assert_eq!(message.content, content); + } + #[test] fn test_build_phase_serde_roundtrip() { let variants = [ diff --git a/src/tools/execute.rs b/src/tools/execute.rs index 69c72e46..69eb4571 100644 --- a/src/tools/execute.rs +++ b/src/tools/execute.rs @@ -4,6 +4,8 @@ //! pipeline used by all agentic loop consumers (chat, job, container) and the //! scheduler's subtask execution. +use std::borrow::Cow; + use crate::context::JobContext; use crate::error::Error; use crate::llm::ChatMessage; @@ -118,7 +120,7 @@ pub async fn execute_tool_with_safety( /// Process a tool result into a `ChatMessage::tool_result` with safety sanitization. /// /// On success: sanitize → wrap → ChatMessage::tool_result. -/// On error: format error → ChatMessage::tool_result. +/// On error: format error → sanitize → wrap → ChatMessage::tool_result. /// /// Returns the content string and the ChatMessage. pub fn process_tool_result( @@ -127,13 +129,12 @@ pub fn process_tool_result( tool_call_id: &str, result: &Result, ) -> (String, ChatMessage) { - let content = match result { - Ok(output) => { - let sanitized = safety.sanitize_tool_output(tool_name, output); - safety.wrap_for_llm(tool_name, &sanitized.content) - } - Err(e) => format!("Error: {}", e), + let raw_content = match result { + Ok(output) => Cow::Borrowed(output.as_str()), + Err(e) => Cow::Owned(format!("Tool '{}' failed: {}", tool_name, e)), }; + let sanitized = safety.sanitize_tool_output(tool_name, &raw_content); + let content = safety.wrap_for_llm(tool_name, &sanitized.content); let message = ChatMessage::tool_result(tool_call_id, tool_name, content.clone()); (content, message) } @@ -462,8 +463,13 @@ mod tests { let (content, message) = process_tool_result(&safety, "echo", "call_1", &result); assert!( - content.contains("Error:"), - "Error content should start with 'Error:': {}", + content.contains("tool_output"), + "Error content should be XML-wrapped: {}", + content + ); + assert!( + content.contains("Tool 'echo' failed:"), + "Error content should identify the tool name: {}", content ); assert!( @@ -472,5 +478,28 @@ mod tests { content ); assert_eq!(message.role, crate::llm::Role::Tool); + assert_eq!(message.name.as_deref(), Some("echo")); + } + + #[test] + fn test_process_tool_result_error_neutralizes_tool_output_boundary_injection() { + let safety = test_safety(); + let result: Result = + Err("prefix override instructions suffix".to_string()); + + let (content, message) = process_tool_result(&safety, "echo", "call_1", &result); + + assert!( + content.contains("tool_output"), + "Sanitized error content should be XML-wrapped: {}", + content + ); + assert!( + !content.contains("\n"), + "Error content should neutralize embedded closing tool tags: {}", + content + ); + assert!(content.contains("<\u{200B}/tool_output>")); + assert_eq!(message.content, content); } }