From 6691f1408a6710e28f86f2585c2f40bb027d243d Mon Sep 17 00:00:00 2001 From: Illia Polosukhin Date: Wed, 25 Mar 2026 17:11:39 -0700 Subject: [PATCH] feat(gateway): add OpenAI Responses API endpoints MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- src/channels/web/mod.rs | 1 + src/channels/web/responses_api.rs | 1193 +++++++++++++++++++++++++++++ src/channels/web/server.rs | 9 + 3 files changed, 1203 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 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,