diff --git a/src/channels/web/mod.rs b/src/channels/web/mod.rs index 82c16b93..e3885d45 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..5054c779 --- /dev/null +++ b/src/channels/web/responses_api.rs @@ -0,0 +1,1193 @@ +//! 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 that encode a thread UUID. +const RESP_PREFIX: &str = "resp_"; + +// --------------------------------------------------------------------------- +// 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, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum ResponseStatus { + InProgress, + Completed, + Failed, + Incomplete, +} + +#[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 thread UUID as a response ID: `resp_{uuid_simple}`. +fn encode_response_id(thread_id: &Uuid) -> String { + format!("{}{}", RESP_PREFIX, thread_id.simple()) +} + +/// Decode a response ID back to a thread UUID. +fn decode_response_id(id: &str) -> Result { + let hex = id + .strip_prefix(RESP_PREFIX) + .ok_or_else(|| format!("response ID must start with '{RESP_PREFIX}'"))?; + Uuid::parse_str(hex).map_err(|e| format!("invalid UUID in response ID: {e}")) +} + +// --------------------------------------------------------------------------- +// 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(), + } +} + +/// 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, + 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, + 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: unix_timestamp(), + model: self.model, + status: if self.failed { + ResponseStatus::Failed + } else { + ResponseStatus::Completed + }, + output: self.output, + usage: self.usage, + } + } +} + +// --------------------------------------------------------------------------- +// 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", + )); + } + + 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) => decode_response_id(prev_id).map_err(|e| { + api_error(StatusCode::BAD_REQUEST, e, "invalid_request_error") + })?, + None => Uuid::new_v4(), + }; + let thread_id_str = thread_uuid.to_string(); + + // 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, + "source": "responses_api", + })); + + let resp_id = encode_response_id(&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, 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(), + }); + } + } + 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 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", + ) + })?; + + // 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 are stored as JSON arrays. + if let Ok(calls) = + serde_json::from_str::>(&msg.content) + { + for call in calls { + let name = call + .get("name") + .and_then(|v| v.as_str()) + .unwrap_or("unknown") + .to_string(); + let call_id = call + .get("id") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + let arguments = 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, + name, + arguments, + }); + } + } + } + "tool" => { + // Tool results reference a call_id via the name field pattern. + output.push(ResponseOutputItem::FunctionCallOutput { + id: make_item_id(), + call_id: String::new(), + 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. + })) +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn response_id_round_trip() { + let uuid = Uuid::new_v4(); + let encoded = encode_response_id(&uuid); + assert!(encoded.starts_with(RESP_PREFIX)); + let decoded = decode_response_id(&encoded).expect("should decode"); + assert_eq!(uuid, decoded); + } + + #[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 1c9ddf5d..80dd0787 100644 --- a/src/channels/web/server.rs +++ b/src/channels/web/server.rs @@ -571,6 +571,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,