From 45cd6682d3435b4652c062f768e1d5314237dafd Mon Sep 17 00:00:00 2001 From: Achieve Date: Fri, 27 Mar 2026 14:54:38 +0800 Subject: [PATCH 1/5] fix: downgrade excessive debug logging in hot path (closes #1686) (#1694) PR #1681 introduced 23 debug-level log statements across relay client, web server handlers, and extension manager functions. Many of these fire on every HTTP request or in loops (e.g. has_stored_team_id called per extension in list_installed). Downgrade them to trace level to reduce noise at the default debug log level while preserving warn/info logs for actionable diagnostics. Co-authored-by: Claude Opus 4.6 (1M context) --- src/channels/relay/client.rs | 10 +++++----- src/channels/web/server.rs | 6 +++--- src/extensions/manager.rs | 30 +++++++++++++++--------------- 3 files changed, 23 insertions(+), 23 deletions(-) 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/web/server.rs b/src/channels/web/server.rs index 06870ace..a6727f79 100644 --- a/src/channels/web/server.rs +++ b/src/channels/web/server.rs @@ -2201,7 +2201,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" @@ -2235,7 +2235,7 @@ async fn extensions_activate_handler( crate::extensions::ExtensionError::AuthRequired ); - tracing::debug!( + tracing::trace!( extension = %name, error = %activate_err, needs_auth = needs_auth, @@ -2249,7 +2249,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/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" From 9c5ba43ccd493c7fad40575ac5210b92648d9545 Mon Sep 17 00:00:00 2001 From: Illia Polosukhin Date: Fri, 27 Mar 2026 00:00:25 -0700 Subject: [PATCH 2/5] feat(gateway): add OpenAI Responses API endpoints (#1656) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(gateway): add OpenAI Responses API endpoints Add POST /v1/responses and GET /v1/responses/{id} to the web gateway, implementing the OpenAI Responses API. Unlike the existing Chat Completions proxy which passes through to the raw LLM, the Responses API routes requests through the full agent loop — giving external clients access to tools, memory, safety, and server-side conversation state via a standard OpenAI-compatible interface. Key design decisions: - Response IDs encode thread UUIDs statelessly (resp_{uuid_simple}) - previous_response_id enables multi-turn conversations - Streaming maps AppEvent variants to Responses API SSE events - Tool approval returns response.failed (no interactive approval flow) - GET endpoint reconstructs ResponseObject from conversation_messages Co-Authored-By: Claude Opus 4.6 (1M context) * fix(responses-api): address all review feedback on PR #1656 - Decouple response ID from thread ID: encode both a per-call response_uuid and the thread_uuid so each POST produces a unique ID - Reject unsupported fields (instructions, tools, tool_choice, temperature, max_output_tokens, non-default model) with 400 - Add user_id to IncomingMessage metadata for user-scoped SSE events - Add conversation_belongs_to_user() ownership check on GET endpoint - Fix tool call parsing: handle both legacy array and object wrapper format; use call_id/tool_call_id/id key fallback chain - Correlate tool role messages to preceding FunctionCall call_id - Stabilize created_at (capture once in accumulator, reuse everywhere) - Surface error_message via new ResponseObject.error field - Handle streaming tool failures (emit FunctionCallOutput on error) - Remove dead Incomplete status variant - Fix formatting (cargo fmt) Co-Authored-By: Claude Opus 4.6 (1M context) --------- Co-authored-by: Claude Opus 4.6 (1M context) --- src/channels/web/mod.rs | 1 + src/channels/web/responses_api.rs | 1411 +++++++++++++++++++++++++++++ src/channels/web/server.rs | 9 + 3 files changed, 1421 insertions(+) create mode 100644 src/channels/web/responses_api.rs diff --git a/src/channels/web/mod.rs b/src/channels/web/mod.rs index 63aedaa0..8ae9051f 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 a6727f79..ab50c94a 100644 --- a/src/channels/web/server.rs +++ b/src/channels/web/server.rs @@ -520,6 +520,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, From 7234700c78d985ddc872721bc2a7130eeaa0b8c3 Mon Sep 17 00:00:00 2001 From: "Will.hou" <1205157517@qq.com> Date: Fri, 27 Mar 2026 15:01:22 +0800 Subject: [PATCH 3/5] fix(llm): prevent UTF-8 panic in line_bounds() (fixes #1669) (#1679) * fix(llm): prevent UTF-8 panic in line_bounds() (fixes #1669) `line_bounds()` used `text[..pos]` slicing which panics when `pos` lands inside a multi-byte UTF-8 character. This happens when `end.saturating_sub(1)` in `is_recoverable_tool_call_segment()` steps back into a multi-byte char like emoji. Fix: clamp `pos` to `text.len()` and walk backward to the nearest char boundary before slicing. Add 5 regression tests covering mid-char positions, emoji boundaries, and out-of-bounds pos. Also fix pre-existing clippy `unnecessary_sort_by` warnings in web gateway handlers. Generated with [Claude Code](https://claude.ai/code) via [Happy](https://happy.engineering) Co-Authored-By: Claude Co-Authored-By: Happy * test: assert expected values in line_bounds UTF-8 tests Address Gemini review: strengthen regression tests to verify correct return values (not just absence of panic) when pos lands mid-char. Generated with [Claude Code](https://claude.ai/code) via [Happy](https://happy.engineering) Co-Authored-By: Claude Co-Authored-By: Happy --------- Co-authored-by: willamhou Co-authored-by: Claude Co-authored-by: Happy --- src/channels/web/handlers/chat.rs | 2 +- src/channels/web/server.rs | 2 +- src/llm/reasoning.rs | 58 +++++++++++++++++++++++++++++-- 3 files changed, 58 insertions(+), 4 deletions(-) diff --git a/src/channels/web/handlers/chat.rs b/src/channels/web/handlers/chat.rs index bc4e3dbc..e2458571 100644 --- a/src/channels/web/handlers/chat.rs +++ b/src/channels/web/handlers/chat.rs @@ -533,7 +533,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/server.rs b/src/channels/web/server.rs index ab50c94a..73e183d2 100644 --- a/src/channels/web/server.rs +++ b/src/channels/web/server.rs @@ -1890,7 +1890,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 { 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 { From 30db07c58e02fa9bb3a6757823e9aa64322aee6f Mon Sep 17 00:00:00 2001 From: "firat.sertgoz" Date: Fri, 27 Mar 2026 10:49:02 +0300 Subject: [PATCH 4/5] fix: require Feishu webhook authentication (#1638) * fix: require Feishu webhook authentication * fix: handle Feishu v2 webhook token auth * fix: skip empty verification token write, consistent with app_id/app_secret Address zmanian review nit #4: only write verification_token to workspace when present, matching the if-let pattern used for app_id and app_secret. Functionally identical (the auth check filters empty strings), but consistent. Co-Authored-By: Claude Opus 4.6 (1M context) --------- Co-authored-by: Claude Opus 4.6 (1M context) --- channels-src/feishu/Cargo.lock | 7 ++ channels-src/feishu/Cargo.toml | 1 + channels-src/feishu/feishu.capabilities.json | 6 +- channels-src/feishu/src/lib.rs | 122 ++++++++++++++++++- src/channels/wasm/loader.rs | 8 ++ src/channels/wasm/schema.rs | 40 ++++++ src/channels/wasm/setup.rs | 17 ++- 7 files changed, 192 insertions(+), 9 deletions(-) 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/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, }; From 2f4eb08613cefff1af8b7b1a475fda00c84dd855 Mon Sep 17 00:00:00 2001 From: "firat.sertgoz" Date: Fri, 27 Mar 2026 10:49:28 +0300 Subject: [PATCH 5/5] fix: sanitize tool error results before llm injection (#1639) * fix: sanitize tool error results before llm injection Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus * fix: wrap preflight tool rejection errors for llm safety Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus * style: apply rustfmt to error-path regressions Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus * fix: preserve wrapped tool errors in history replay * fix: address review findings on PR #1639 - Simplify legacy error handling in rebuild_chat_messages_from_db: remove redundant "Error: " prefix since legacy errors already contain descriptive text (e.g. "Tool 'http' failed: timeout"). Both wrapped (new) and plain (legacy) errors now pass through as-is. - Update existing test assertion to match simplified format. - Restore error-path doc line on process_tool_result. Co-Authored-By: Claude Opus 4.6 (1M context) * fix: satisfy clippy on builder tool safety helper --------- Co-authored-by: Sisyphus Co-authored-by: Claude Opus 4.6 (1M context) --- src/agent/dispatcher.rs | 88 +++++++++++++++++++++---------- src/agent/thread_ops.rs | 32 ++++++++++- src/channels/web/handlers/chat.rs | 6 ++- src/channels/web/util.rs | 30 ++++++++++- src/tools/builder/core.rs | 60 +++++++++++++++++---- src/tools/execute.rs | 47 +++++++++++++---- 6 files changed, 211 insertions(+), 52 deletions(-) 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/web/handlers/chat.rs b/src/channels/web/handlers/chat.rs index e2458571..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(), 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/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); } }