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,