diff --git a/TERMINOLOGY.md b/TERMINOLOGY.md index 7e13b8ce..e2df7035 100644 --- a/TERMINOLOGY.md +++ b/TERMINOLOGY.md @@ -129,6 +129,29 @@ Describes a flow in which the service retains or resolves prior state, such as R Describes a flow in which the request supplies all required context and the service does not rely on retained response or conversation state. `store: false` disables stored-response state, although callers may still replay prior items. +### compaction + +Replacing a long item history with a smaller canonical window that preserves user messages and a model-generated +summary. Compaction is distinct from truncation: it uses an inference call to summarize context rather than silently +dropping items. + +### compaction item + +An item with `type: "compaction"` whose `encrypted_content` carries the context checkpoint consumed by a later model +request. Preserve the exact field name in wire-format discussion. In this project's local implementation the field is +plaintext despite its protocol name. + +### compacted window + +The canonical output of a compaction request: retained user-message items followed by one compaction item. Replay the +complete window as later Responses input; do not extract only the summary. + +### context management + +The Responses request policy expressed by `context_management`. A `type: "compaction"` entry with +`compact_threshold` asks the gateway to compact resolved input before inference when its estimated token count exceeds +the threshold. + ## Tools and tool calling ### tool @@ -340,3 +363,4 @@ These definitions follow current OpenAI documentation: - [MCP and Connectors](https://developers.openai.com/api/docs/guides/tools-connectors-mcp) - [Streaming API responses](https://developers.openai.com/api/docs/guides/streaming-responses) - [Reasoning models](https://developers.openai.com/api/docs/guides/reasoning) +- [Compaction](https://developers.openai.com/api/docs/guides/compaction) diff --git a/crates/agentic-server-core/benches/executor_throughput.rs b/crates/agentic-server-core/benches/executor_throughput.rs index 7c3803f2..d62f807f 100644 --- a/crates/agentic-server-core/benches/executor_throughput.rs +++ b/crates/agentic-server-core/benches/executor_throughput.rs @@ -147,6 +147,7 @@ fn make_request(input: &str, stream: bool, prev_id: Option) -> RequestPa metadata: None, parallel_tool_calls: None, cache_salt: None, + context_management: None, } } diff --git a/crates/agentic-server-core/benches/storage_concurrent.rs b/crates/agentic-server-core/benches/storage_concurrent.rs index ee2736d1..55d0c4e6 100644 --- a/crates/agentic-server-core/benches/storage_concurrent.rs +++ b/crates/agentic-server-core/benches/storage_concurrent.rs @@ -61,7 +61,9 @@ fn next_id() -> String { fn make_items() -> Vec { let input = InputItem::Message(InputMessage { + id: None, role: "user".to_string(), + status: None, content: InputMessageContent::Text("Test message".to_string()), }); vec![ diff --git a/crates/agentic-server-core/benches/storage_crud.rs b/crates/agentic-server-core/benches/storage_crud.rs index 72442a42..deac34cf 100644 --- a/crates/agentic-server-core/benches/storage_crud.rs +++ b/crates/agentic-server-core/benches/storage_crud.rs @@ -13,7 +13,9 @@ fn next_id() -> String { fn create_test_items() -> Vec { let input_item = InputItem::Message(InputMessage { + id: None, role: "user".to_string(), + status: None, content: InputMessageContent::Text("Test message".to_string()), }); diff --git a/crates/agentic-server-core/src/executor/compaction.rs b/crates/agentic-server-core/src/executor/compaction.rs new file mode 100644 index 00000000..cc811d1f --- /dev/null +++ b/crates/agentic-server-core/src/executor/compaction.rs @@ -0,0 +1,580 @@ +use crate::executor::error::{ExecutorError, ExecutorResult}; +use crate::executor::rehydrate::rehydrate_conversation; +use crate::executor::request::{ExecutionContext, RequestContext}; +use crate::executor::upstream::fetch_blocking_payload; +use crate::types::event::MessageStatus; +use crate::types::io::input::latest_compaction_window; +use crate::types::io::{ + CompactionItem, InputContent, InputItem, InputMessage, InputMessageContent, OutputItem, ResponseUsage, + ResponsesInput, +}; +use crate::types::request_response::{CompactRequest, CompactedResponse, RequestPayload, ResponsePayload}; +use crate::utils::common::{serialize_to_string, utcnow_str, uuid7_str}; + +const COMPACTION_PROMPT: &str = "You are performing a CONTEXT CHECKPOINT COMPACTION. Create a concise handoff summary that preserves current progress, decisions, constraints, unresolved work, and critical references for the next model. Return only the summary."; + +fn retained_user_window(items: &[InputItem]) -> Vec { + let window = latest_compaction_window(items); + items + .iter() + .enumerate() + .filter_map(|(index, item)| { + let InputItem::Message(message) = item else { + return None; + }; + if message.role != "user" + || window.is_some_and(|window| index < window.latest_index() && !window.retains_user_item(index, item)) + { + return None; + } + let mut retained = message.clone(); + retained.id = Some(uuid7_str("msg_")); + retained.status = Some(MessageStatus::Completed); + Some(InputItem::Message(retained)) + }) + .collect() +} + +fn finish_compacted_window(mut output: Vec, summary: String) -> Vec { + output.push(InputItem::Compaction(CompactionItem { + id: Some(uuid7_str("cmp_")), + encrypted_content: summary, + })); + output +} + +#[cfg(test)] +fn build_compacted_window(items: &[InputItem], summary: String) -> Vec { + finish_compacted_window(retained_user_window(items), summary) +} + +fn response_output_text(output: &[OutputItem]) -> Option { + let text = output + .iter() + .filter_map(|item| match item { + OutputItem::Message(message) => Some(message), + _ => None, + }) + .flat_map(|message| message.content.iter()) + .map(|content| content.text.trim()) + .filter(|text| !text.is_empty()) + .collect::>() + .join("\n"); + (!text.is_empty()).then_some(text) +} + +fn value_has_content(value: &serde_json::Value) -> bool { + match value { + serde_json::Value::Null => false, + serde_json::Value::String(text) => !text.trim().is_empty(), + serde_json::Value::Array(values) => values.iter().any(value_has_content), + serde_json::Value::Object(values) => values.values().any(value_has_content), + serde_json::Value::Bool(_) | serde_json::Value::Number(_) => true, + } +} + +fn item_has_meaningful_context(item: &InputItem) -> bool { + match item { + InputItem::Message(message) => match &message.content { + InputMessageContent::Text(text) => !text.trim().is_empty(), + InputMessageContent::Parts(parts) => parts.iter().any(|part| match part { + InputContent::InputText(text) | InputContent::OutputText(text) | InputContent::ReasoningText(text) => { + !text.text.trim().is_empty() + } + InputContent::InputImage(image) => image.image_url.as_deref().is_some_and(|url| !url.trim().is_empty()), + InputContent::Unknown => false, + }), + }, + InputItem::FunctionCall(call) => !call.name.trim().is_empty() || !call.arguments.trim().is_empty(), + InputItem::FunctionCallOutput(output) => !output.output.trim().is_empty(), + InputItem::CustomToolCall(call) => !call.name.trim().is_empty() || !call.input.trim().is_empty(), + InputItem::CustomToolCallOutput(output) => value_has_content(&output.output), + InputItem::Reasoning(reasoning) => { + reasoning.content.iter().any(|content| !content.text.trim().is_empty()) + || reasoning.summary.iter().any(value_has_content) + || reasoning.encrypted_content.as_ref().is_some_and(value_has_content) + } + InputItem::Compaction(compaction) => !compaction.encrypted_content.trim().is_empty(), + InputItem::Unknown => false, + } +} + +fn completed_summary_text(response: &ResponsePayload) -> ExecutorResult { + if response.status != "completed" || response.error.is_some() { + let details = response + .error + .as_ref() + .and_then(|error| serialize_to_string(error).ok()) + .or_else(|| { + response + .incomplete_details + .as_ref() + .and_then(|details| details.reason.clone()) + }) + .unwrap_or_else(|| "upstream returned no failure details".to_owned()); + return Err(ExecutorError::CompactionFailed { + status: response.status.clone(), + details, + }); + } + response_output_text(&response.output) + .ok_or_else(|| ExecutorError::InvalidRequest("compaction model returned no summary text".to_owned())) +} + +/// Estimate the current model-facing context size without requiring a model-specific tokenizer. +/// +/// The approximation deliberately includes JSON structure and rounds up at four UTF-8 bytes per +/// token. It is deterministic, inexpensive, and errs slightly toward compacting early. +#[must_use] +pub(crate) fn estimate_input_tokens(input: &ResponsesInput) -> u64 { + let serialized = serialize_to_string(&input.model_input()).unwrap_or_default(); + let bytes = u64::try_from(serialized.len()).unwrap_or(u64::MAX); + bytes.saturating_add(3) / 4 +} + +fn request_payload(model: String, input: ResponsesInput, instructions: Option) -> RequestPayload { + RequestPayload { + model, + input, + instructions, + previous_response_id: None, + conversation_id: None, + tools: None, + tool_choice: None, + stream: false, + store: false, + include: None, + temperature: None, + top_p: None, + max_output_tokens: None, + truncation: None, + metadata: None, + parallel_tool_calls: None, + cache_salt: None, + context_management: None, + } +} + +/// Summarize an already-resolved item history and return its canonical compacted window. +/// +/// # Errors +/// +/// Returns an invalid-request error for empty input or an empty model summary, and propagates +/// inference and serialization failures. +pub(crate) async fn compact_items( + model: &str, + input: ResponsesInput, + instructions: Option<&str>, + exec_ctx: &ExecutionContext, + auth: Option<&str>, +) -> ExecutorResult<(Vec, ResponseUsage)> { + let original_items = Vec::from(input); + if !original_items.iter().any(item_has_meaningful_context) { + return Err(ExecutorError::InvalidRequest( + "compaction requires non-empty input or previous_response_id context".to_owned(), + )); + } + + let compacted = retained_user_window(&original_items); + let mut summary_items = original_items; + summary_items.push(InputItem::Message(InputMessage { + id: None, + role: "user".to_owned(), + status: None, + content: InputMessageContent::Text(COMPACTION_PROMPT.to_owned()), + })); + let instructions = instructions.map(str::to_owned); + let original_request = request_payload( + model.to_owned(), + ResponsesInput::Items(Vec::new()), + instructions.clone(), + ); + let enriched_request = request_payload(model.to_owned(), ResponsesInput::Items(summary_items), instructions); + let ctx = RequestContext { + original_request, + enriched_request, + new_input_items: Vec::new(), + response_id: uuid7_str("resp_"), + conversation_id: None, + }; + let response = fetch_blocking_payload(&ctx, exec_ctx, auth).await?; + let summary = completed_summary_text(&response)?; + + Ok(( + finish_compacted_window(compacted, summary), + response.usage.unwrap_or_default(), + )) +} + +/// Apply the first configured compaction threshold to the resolved request history. +/// +/// Returns the summarization usage when compaction ran. The configuration remains on the +/// client-facing request context but is never part of [`crate::types::request_response::UpstreamRequest`]. +/// +/// # Errors +/// +/// Propagates inference errors from the summarization request. +pub(crate) async fn maybe_compact_context( + ctx: &mut RequestContext, + exec_ctx: &ExecutionContext, + auth: Option<&str>, +) -> ExecutorResult> { + let threshold = ctx + .enriched_request + .context_management + .as_deref() + .unwrap_or_default() + .iter() + .find(|entry| entry.type_ == "compaction") + .and_then(|entry| entry.compact_threshold); + let Some(threshold) = threshold else { + return Ok(None); + }; + let estimated_tokens = estimate_input_tokens(&ctx.enriched_request.input); + if estimated_tokens <= threshold { + return Ok(None); + } + + tracing::debug!( + estimated_tokens, + threshold, + "automatically compacting resolved response input" + ); + let model = ctx.enriched_request.model.clone(); + let instructions = ctx.enriched_request.instructions.clone(); + let input = std::mem::replace(&mut ctx.enriched_request.input, ResponsesInput::Items(Vec::new())); + let (compacted, usage) = compact_items(&model, input, instructions.as_deref(), exec_ctx, auth).await?; + ctx.enriched_request.input = ResponsesInput::Items(compacted.clone()); + ctx.new_input_items = compacted; + Ok(Some(usage)) +} + +/// Compact direct input or a stored previous-response chain into a reusable item window. +/// +/// # Errors +/// +/// Returns an invalid-request error when neither input nor a previous response ID is supplied, +/// and propagates history, inference, and persistence failures. +pub async fn compact_response( + request: CompactRequest, + exec_ctx: &ExecutionContext, + auth: Option<&str>, +) -> ExecutorResult { + if request.input.is_none() && request.previous_response_id.is_none() { + return Err(ExecutorError::InvalidRequest( + "compaction requires input or previous_response_id".to_owned(), + )); + } + + let mut payload = request_payload( + request.model, + request.input.unwrap_or_else(|| ResponsesInput::Items(Vec::new())), + request.instructions, + ); + payload.previous_response_id = request.previous_response_id; + let mut ctx = rehydrate_conversation(payload, exec_ctx).await?; + let model = ctx.enriched_request.model.clone(); + let instructions = ctx.enriched_request.instructions.clone(); + let input = std::mem::replace(&mut ctx.enriched_request.input, ResponsesInput::Items(Vec::new())); + let (output, usage) = compact_items(&model, input, instructions.as_deref(), exec_ctx, auth).await?; + + let response_id = ctx.response_id.clone(); + ctx.new_input_items.clone_from(&output); + match exec_ctx.resp_handler.execute_turn(ctx, Vec::new()).await { + Ok(()) | Err(ExecutorError::Storage(crate::StorageError::NotConfigured)) => {} + Err(error) => return Err(error), + } + + Ok(CompactedResponse { + id: response_id, + object: "response.compaction".to_owned(), + created_at: utcnow_str(), + output, + usage, + }) +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use axum::Router; + use axum::routing::post; + + use crate::executor::modes::{ConversationHandler, ResponseHandler}; + use crate::executor::request::ExecutionContext; + use crate::storage::{ConversationStore, InOutItem, ResponseMetadata, ResponseStore, create_pool_with_schema}; + use crate::types::event::MessageStatus; + use crate::types::io::{ + CompactionItem, FunctionToolResultMessage, InputItem, InputMessage, InputMessageContent, ResponsesInput, + }; + + use super::{build_compacted_window, compact_response, completed_summary_text, estimate_input_tokens}; + + fn user_message(text: &str) -> InputItem { + InputItem::Message(InputMessage { + id: None, + role: "user".to_owned(), + status: None, + content: InputMessageContent::Text(text.to_owned()), + }) + } + + async fn mock_execution_context(response_store: ResponseStore) -> (ExecutionContext, tokio::task::JoinHandle<()>) { + let app = Router::new().route( + "/v1/responses", + post(|| async { + axum::Json(serde_json::json!({ + "id": "resp_upstream", + "object": "response", + "created_at": 0, + "model": "test-model", + "status": "completed", + "output": [{ + "id": "msg_upstream", + "type": "message", + "role": "assistant", + "status": "completed", + "content": [{ + "type": "output_text", + "text": "durable summary", + "annotations": [] + }] + }], + "usage": { + "input_tokens": 12, + "output_tokens": 3, + "total_tokens": 15 + }, + "incomplete_details": null, + "error": null, + "previous_response_id": null, + "conversation_id": null, + "instructions": null + })) + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind mock inference server"); + let address = listener.local_addr().expect("mock server address"); + let server = tokio::spawn(async move { + axum::serve(listener, app).await.ok(); + }); + let exec_ctx = ExecutionContext::new( + ConversationHandler::new(ConversationStore::disabled()), + ResponseHandler::new(response_store), + Arc::new(reqwest::Client::new()), + format!("http://{address}"), + ); + (exec_ctx, server) + } + + fn compact_request() -> crate::CompactRequest { + serde_json::from_value(serde_json::json!({ + "model": "test-model", + "input": [{"role": "user", "content": "important context"}] + })) + .expect("valid compact request") + } + + #[test] + fn compacted_window_retains_user_messages_and_replaces_old_compaction() { + let items = vec![ + InputItem::Compaction(CompactionItem { + id: Some("cmp_old".to_owned()), + encrypted_content: "old summary".to_owned(), + }), + user_message("first"), + InputItem::FunctionCallOutput(FunctionToolResultMessage { + call_id: "call_1".to_owned(), + output: "tool output".to_owned(), + }), + user_message("second"), + ]; + + let output = build_compacted_window(&items, "new summary".to_owned()); + + assert_eq!( + output + .iter() + .filter(|item| matches!(item, InputItem::Message(_))) + .count(), + 2 + ); + assert_eq!( + output + .iter() + .filter(|item| matches!(item, InputItem::Compaction(_))) + .count(), + 1 + ); + assert!(matches!(output.last(), Some(InputItem::Compaction(_)))); + for item in output.iter().filter_map(|item| match item { + InputItem::Message(message) => Some(message), + _ => None, + }) { + assert_eq!(item.status, Some(MessageStatus::Completed)); + assert!(item.id.as_deref().is_some_and(|id| id.starts_with("msg_"))); + } + } + + #[test] + fn token_estimate_counts_replayed_content() { + let input = ResponsesInput::Items(vec![ + user_message("hello context"), + InputItem::FunctionCallOutput(FunctionToolResultMessage { + call_id: "call_1".to_owned(), + output: "substantial tool output".to_owned(), + }), + ]); + + assert!(estimate_input_tokens(&input) > 0); + } + + #[test] + fn partial_text_from_incomplete_or_failed_summary_is_rejected() { + for (status, error, incomplete_details) in [ + ( + "incomplete", + serde_json::Value::Null, + serde_json::json!({"reason": "max_output_tokens"}), + ), + ( + "failed", + serde_json::json!({"message": "model failed"}), + serde_json::Value::Null, + ), + ] { + let response: crate::ResponsePayload = serde_json::from_value(serde_json::json!({ + "id": "resp_upstream", + "object": "response", + "created_at": 0, + "model": "test-model", + "status": status, + "output": [{ + "id": "msg_partial", + "type": "message", + "role": "assistant", + "status": "completed", + "content": [{"type": "output_text", "text": "partial summary", "annotations": []}] + }], + "usage": null, + "incomplete_details": incomplete_details, + "error": error, + "previous_response_id": null, + "conversation_id": null, + "instructions": null + })) + .expect("valid upstream response"); + + let error = completed_summary_text(&response).expect_err("partial summary must be rejected"); + assert!(matches!(error, crate::executor::ExecutorError::CompactionFailed { .. })); + } + } + + #[tokio::test] + async fn disabled_storage_does_not_fail_compaction() { + let (exec_ctx, server) = mock_execution_context(ResponseStore::disabled()).await; + + let response = compact_response(compact_request(), &exec_ctx, None) + .await + .expect("disabled persistence should be ignored"); + + assert_eq!(response.object, "response.compaction"); + assert_eq!(response.usage.total_tokens, 15); + assert!(response.id.starts_with("resp_")); + assert!( + matches!(response.output.last(), Some(InputItem::Compaction(item)) if item.encrypted_content == "durable summary") + ); + server.abort(); + } + + #[tokio::test] + async fn empty_context_is_rejected_before_summarization() { + let (exec_ctx, server) = mock_execution_context(ResponseStore::disabled()).await; + for input in [ + serde_json::json!(""), + serde_json::json!([{"role": "user", "content": " "}]), + serde_json::json!([{"type": "future_item"}]), + ] { + let request = serde_json::from_value(serde_json::json!({ + "model": "test-model", + "input": input + })) + .expect("structurally valid compact request"); + let error = compact_response(request, &exec_ctx, None) + .await + .expect_err("empty context must fail"); + assert!(matches!(error, crate::executor::ExecutorError::InvalidRequest(_))); + } + server.abort(); + } + + #[tokio::test] + async fn compaction_persists_a_reusable_response_checkpoint() { + let pool = create_pool_with_schema(Some("sqlite::memory:")) + .await + .expect("create response store"); + let response_store = ResponseStore::new(pool); + let (exec_ctx, server) = mock_execution_context(response_store.clone()).await; + + let response = compact_response(compact_request(), &exec_ctx, None) + .await + .expect("compaction succeeds"); + let history = response_store + .rehydrate(&response.id) + .await + .expect("compaction checkpoint can be rehydrated"); + + assert_eq!(history.len(), 2); + assert!(matches!(history[0], InOutItem::Input(InputItem::Message(_)))); + assert!(matches!(history[1], InOutItem::Input(InputItem::Compaction(_)))); + server.abort(); + } + + #[tokio::test] + async fn compaction_resolves_and_replaces_previous_response_history() { + let pool = create_pool_with_schema(Some("sqlite::memory:")) + .await + .expect("create response store"); + let response_store = ResponseStore::new(pool); + response_store + .persist( + "resp_previous", + None, + vec![InOutItem::Input(user_message("remember banana"))], + &ResponseMetadata { + model: "test-model".to_owned(), + previous_response_id: None, + effective_tools: None, + effective_tool_choice: crate::ToolChoice::Auto, + effective_instructions: None, + }, + ) + .await + .expect("seed previous response"); + let (exec_ctx, server) = mock_execution_context(response_store.clone()).await; + let request = serde_json::from_value(serde_json::json!({ + "model": "test-model", + "previous_response_id": "resp_previous" + })) + .expect("valid previous-response compact request"); + + let response = compact_response(request, &exec_ctx, None) + .await + .expect("previous response compacts"); + let history = response_store + .rehydrate(&response.id) + .await + .expect("compacted continuation rehydrates"); + let model_input = ResponsesInput::Items(InOutItem::into_input_items(history)); + let serialized = serde_json::to_value(model_input.model_input()).expect("model input serializes"); + + assert_eq!(serialized.as_array().map(Vec::len), Some(2)); + assert_eq!(serialized[0]["content"], "remember banana"); + assert_eq!(serialized[1]["role"], "assistant"); + assert_eq!(serialized[1]["content"][0]["text"], "durable summary"); + server.abort(); + } +} diff --git a/crates/agentic-server-core/src/executor/engine.rs b/crates/agentic-server-core/src/executor/engine.rs index fe4fa127..efc3430e 100644 --- a/crates/agentic-server-core/src/executor/engine.rs +++ b/crates/agentic-server-core/src/executor/engine.rs @@ -12,6 +12,7 @@ use either::Either; use tokio::sync::mpsc; use tracing::{debug, warn}; +use super::compaction::maybe_compact_context; use super::gateway::{ GatewayCallResult, LoopDecision, append_gateway_calls_to_new_input, append_output_items_to_input, append_tool_outputs, classify_round, emit_gateway_completed_events, emit_gateway_start_events, @@ -98,13 +99,14 @@ async fn run_until_gateway_tools_complete( auth: Option<&str>, stream_upstream: bool, mut stream: Option<(&mut GatewayStreamAccumulator, &mpsc::UnboundedSender)>, + initial_usage: Option, ) -> ExecutorResult<(ResponsePayload, RequestContext)> { let registry: ToolRegistry = match ctx.enriched_request.tools.as_ref() { Some(tools) => ToolRegistry::build_with_handlers(tools, &exec_ctx.gateway_executors).await?, None => ToolRegistry::default(), }; let mut combined_output: Vec = Vec::new(); - let mut combined_usage: Option = None; + let mut combined_usage = initial_usage; for round in 0..MAX_GATEWAY_TOOL_ROUNDS { let output_offset = combined_output.len(); @@ -327,8 +329,9 @@ async fn run_blocking( ctx: RequestContext, exec_ctx: &ExecutionContext, auth: Option<&str>, + initial_usage: Option, ) -> ExecutorResult { - let (payload, ctx) = run_until_gateway_tools_complete(ctx, exec_ctx, auth, false, None).await?; + let (payload, ctx) = run_until_gateway_tools_complete(ctx, exec_ctx, auth, false, None, initial_usage).await?; let ch = exec_ctx.conv_handler.clone(); let rh = exec_ctx.resp_handler.clone(); @@ -339,7 +342,12 @@ async fn run_blocking( Ok(payload) } -fn run_stream(ctx: RequestContext, exec_ctx: Arc, auth: Option) -> BoxStream { +fn run_stream( + ctx: RequestContext, + exec_ctx: Arc, + auth: Option, + initial_usage: Option, +) -> BoxStream { Box::pin(stream! { let (event_tx, mut event_rx) = mpsc::unbounded_channel(); let exec_ctx_for_run = Arc::clone(&exec_ctx); @@ -353,6 +361,7 @@ fn run_stream(ctx: RequestContext, exec_ctx: Arc, auth: Option auth.as_deref(), true, Some((&mut stream_accumulator, &event_tx_for_run)), + initial_usage, ) .await; (result, stream_accumulator) @@ -488,12 +497,18 @@ impl ExecuteRequest { tools = self.payload.tools.as_ref().map_or(0, Vec::len), "executor received responses request" ); - let ctx = rehydrate_conversation(self.payload, &self.exec_ctx).await?; + let mut ctx = rehydrate_conversation(self.payload, &self.exec_ctx).await?; + let compaction_usage = maybe_compact_context(&mut ctx, &self.exec_ctx, self.client_auth.as_deref()).await?; if ctx.original_request.stream { - Ok(Either::Right(run_stream(ctx, self.exec_ctx, self.client_auth))) + Ok(Either::Right(run_stream( + ctx, + self.exec_ctx, + self.client_auth, + compaction_usage, + ))) } else { Ok(Either::Left( - run_blocking(ctx, &self.exec_ctx, self.client_auth.as_deref()).await?, + run_blocking(ctx, &self.exec_ctx, self.client_auth.as_deref(), compaction_usage).await?, )) } } diff --git a/crates/agentic-server-core/src/executor/error.rs b/crates/agentic-server-core/src/executor/error.rs index 845044e2..f2eb5307 100644 --- a/crates/agentic-server-core/src/executor/error.rs +++ b/crates/agentic-server-core/src/executor/error.rs @@ -56,6 +56,9 @@ pub enum ExecutorError { #[error("invalid request: {0}")] InvalidRequest(String), + #[error("compaction summarization failed with status '{status}': {details}")] + CompactionFailed { status: String, details: String }, + #[error("tool error: {0}")] Tool(#[from] ToolError), } @@ -68,7 +71,7 @@ impl ExecutorError { Self::Storage(e) if e.is_not_found() => StatusCode::NOT_FOUND, Self::LLMRequest { status, .. } => *status, Self::Tool(ToolError::Config(_)) | Self::InvalidRequest(_) | Self::JsonError(_) => StatusCode::BAD_REQUEST, - Self::Tool(ToolError::Execution(_)) => StatusCode::BAD_GATEWAY, + Self::Tool(ToolError::Execution(_)) | Self::CompactionFailed { .. } => StatusCode::BAD_GATEWAY, Self::ParseError(_) => StatusCode::UNPROCESSABLE_ENTITY, _ => StatusCode::INTERNAL_SERVER_ERROR, } @@ -79,7 +82,7 @@ impl ExecutorError { pub fn error_code(&self) -> &'static str { match self { Self::Storage(e) if e.is_not_found() => "not_found", - Self::LLMRequest { .. } => "upstream_error", + Self::LLMRequest { .. } | Self::CompactionFailed { .. } => "upstream_error", Self::Tool(ToolError::Config(_)) | Self::InvalidRequest(_) | Self::ParseError(_) | Self::JsonError(_) => { "invalid_request_error" } diff --git a/crates/agentic-server-core/src/executor/gateway.rs b/crates/agentic-server-core/src/executor/gateway.rs index 5880445c..f670b30b 100644 --- a/crates/agentic-server-core/src/executor/gateway.rs +++ b/crates/agentic-server-core/src/executor/gateway.rs @@ -438,7 +438,7 @@ pub(super) fn append_gateway_calls_to_new_input( let OutputItem::FunctionCall(call) = item else { return None; }; - is_gateway_owned_call(call, registry).then(|| InputItem::FunctionCall(call.clone())) + is_gateway_owned_call(call, registry).then(|| InputItem::FunctionCall(call.clone().into())) })); } diff --git a/crates/agentic-server-core/src/executor/mod.rs b/crates/agentic-server-core/src/executor/mod.rs index 134fd12e..e45f8633 100644 --- a/crates/agentic-server-core/src/executor/mod.rs +++ b/crates/agentic-server-core/src/executor/mod.rs @@ -1,6 +1,7 @@ //! Agentic executor. pub mod accumulator; +pub mod compaction; pub mod engine; pub mod error; pub mod inference; @@ -15,6 +16,7 @@ mod gateway; pub mod gateway_accumulator; mod upstream; +pub use compaction::compact_response; pub use engine::{BoxStream, ExecuteRequest, create_conversation, execute}; pub use error::{ExecutorError, ExecutorResult}; pub use inference::call_inference; diff --git a/crates/agentic-server-core/src/executor/modes/conversation.rs b/crates/agentic-server-core/src/executor/modes/conversation.rs index 3229bd29..1e185708 100644 --- a/crates/agentic-server-core/src/executor/modes/conversation.rs +++ b/crates/agentic-server-core/src/executor/modes/conversation.rs @@ -143,6 +143,7 @@ mod tests { metadata: None, parallel_tool_calls: None, cache_salt: None, + context_management: None, }; RequestContext { enriched_request: req.clone(), diff --git a/crates/agentic-server-core/src/executor/modes/response.rs b/crates/agentic-server-core/src/executor/modes/response.rs index ae771a47..eb933e22 100644 --- a/crates/agentic-server-core/src/executor/modes/response.rs +++ b/crates/agentic-server-core/src/executor/modes/response.rs @@ -122,6 +122,7 @@ mod tests { metadata: None, parallel_tool_calls: None, cache_salt: None, + context_management: None, }; RequestContext { enriched_request: req.clone(), diff --git a/crates/agentic-server-core/src/lib.rs b/crates/agentic-server-core/src/lib.rs index 1eefba43..8ce039f6 100644 --- a/crates/agentic-server-core/src/lib.rs +++ b/crates/agentic-server-core/src/lib.rs @@ -19,9 +19,10 @@ pub use tool::{ ToolOutput, ToolRegistry, ToolType, WebSearchHandler, }; pub use types::{ - CodeInterpreterToolParam, CodexNamespaceMember, CodexNamespaceToolParam, CustomToolCall, - CustomToolCallOutputMessage, CustomToolParam, EmptyToolNameError, FileSearchToolParam, FunctionTool, - FunctionToolCall, FunctionToolParam, FunctionToolResultMessage, GatewayCallStatus, IncompleteDetails, InputContent, + CodeInterpreterToolParam, CodexNamespaceMember, CodexNamespaceToolParam, CompactRequest, CompactedResponse, + CompactionItem, ContextManagement, CustomToolCall, CustomToolCallOutputMessage, CustomToolParam, + EmptyToolNameError, FileSearchToolParam, FunctionTool, FunctionToolCall, FunctionToolParam, + FunctionToolResultMessage, GatewayCallStatus, IncompleteDetails, InputContent, InputFunctionToolCall, InputImageContent, InputItem, InputMessage, InputMessageContent, InputTextContent, InputTokenDetails, McpToolCall, McpToolParam, NonEmptyToolName, OutputItem, OutputMessage, OutputTextContent, OutputTokenDetails, ReasoningOutput, ReasoningTextContent, RequestPayload, ResponsePayload, ResponseUsage, ResponsesInput, ResponsesTool, ToolChoice, diff --git a/crates/agentic-server-core/src/storage/types/item.rs b/crates/agentic-server-core/src/storage/types/item.rs index 925af600..df654c49 100644 --- a/crates/agentic-server-core/src/storage/types/item.rs +++ b/crates/agentic-server-core/src/storage/types/item.rs @@ -107,7 +107,9 @@ mod tests { #[test] fn test_inout_item_from_input() { let input = InputItem::Message(InputMessage { + id: None, role: "user".to_string(), + status: None, content: InputMessageContent::Text("hello".to_string()), }); let item: InOutItem = input.into(); @@ -124,7 +126,9 @@ mod tests { #[test] fn test_inout_item_to_string() { let input = InputItem::Message(InputMessage { + id: None, role: "user".to_string(), + status: None, content: InputMessageContent::Text("test".to_string()), }); let item = InOutItem::Input(input); @@ -141,12 +145,16 @@ mod tests { output.content.push(OutputTextContent::new("answer")); let items = vec![ InOutItem::Input(InputItem::Message(InputMessage { + id: None, role: "user".to_string(), + status: None, content: InputMessageContent::Text("msg1".to_string()), })), InOutItem::Output(OutputItem::Message(output)), InOutItem::Input(InputItem::Message(InputMessage { + id: None, role: "user".to_string(), + status: None, content: InputMessageContent::Text("msg2".to_string()), })), ]; diff --git a/crates/agentic-server-core/src/types/io/input.rs b/crates/agentic-server-core/src/types/io/input.rs index 4c040c9d..37460b92 100644 --- a/crates/agentic-server-core/src/types/io/input.rs +++ b/crates/agentic-server-core/src/types/io/input.rs @@ -1,6 +1,10 @@ +use std::borrow::Cow; + use serde::{Deserialize, Serialize}; use serde_json::Value; +use crate::types::event::MessageStatus; + use super::output::{CustomToolCall, FunctionToolCall, ReasoningOutput}; #[derive(Debug, Clone, Serialize, Deserialize)] @@ -36,7 +40,11 @@ pub enum InputContent { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct InputMessage { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub id: Option, pub role: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub status: Option, pub content: InputMessageContent, } @@ -53,6 +61,45 @@ pub struct FunctionToolResultMessage { pub output: String, } +/// A model-generated function call replayed as Responses input. +/// +/// Input replay is intentionally more permissive than [`FunctionToolCall`] +/// output: clients may omit `id` and `status` when passing prior items to a +/// later request or to the compact endpoint. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct InputFunctionToolCall { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub id: Option, + pub call_id: String, + pub name: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub namespace: Option, + pub arguments: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub status: Option, +} + +impl From for InputFunctionToolCall { + fn from(call: FunctionToolCall) -> Self { + Self { + id: Some(call.id), + call_id: call.call_id, + name: call.name, + namespace: call.namespace, + arguments: call.arguments, + status: Some(call.status), + } + } +} + +/// An opaque compacted context checkpoint accepted as Responses input. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct CompactionItem { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub id: Option, + pub encrypted_content: String, +} + /// Client result for a freeform custom tool call. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct CustomToolCallOutputMessage { @@ -62,7 +109,7 @@ pub struct CustomToolCallOutputMessage { pub output: Value, } -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize)] #[serde(tag = "type")] pub enum InputItem { #[serde(rename = "message")] @@ -70,7 +117,7 @@ pub enum InputItem { /// The model's tool invocation — appears in rehydrated history so vLLM sees /// the full call/output pair across turns. #[serde(rename = "function_call")] - FunctionCall(FunctionToolCall), + FunctionCall(InputFunctionToolCall), #[serde(rename = "function_call_output")] FunctionCallOutput(FunctionToolResultMessage), /// The model's freeform invocation, retained when rehydrating the matching @@ -81,10 +128,67 @@ pub enum InputItem { CustomToolCallOutput(CustomToolCallOutputMessage), #[serde(rename = "reasoning")] Reasoning(ReasoningOutput), + #[serde(rename = "compaction")] + Compaction(CompactionItem), #[serde(other)] Unknown, } +#[derive(Deserialize)] +#[serde(tag = "type")] +enum TaggedInputItem { + #[serde(rename = "message")] + Message(InputMessage), + #[serde(rename = "function_call")] + FunctionCall(InputFunctionToolCall), + #[serde(rename = "function_call_output")] + FunctionCallOutput(FunctionToolResultMessage), + #[serde(rename = "custom_tool_call")] + CustomToolCall(CustomToolCall), + #[serde(rename = "custom_tool_call_output")] + CustomToolCallOutput(CustomToolCallOutputMessage), + #[serde(rename = "reasoning")] + Reasoning(ReasoningOutput), + #[serde(rename = "compaction")] + Compaction(CompactionItem), + #[serde(other)] + Unknown, +} + +impl From for InputItem { + fn from(item: TaggedInputItem) -> Self { + match item { + TaggedInputItem::Message(message) => Self::Message(message), + TaggedInputItem::FunctionCall(call) => Self::FunctionCall(call), + TaggedInputItem::FunctionCallOutput(output) => Self::FunctionCallOutput(output), + TaggedInputItem::CustomToolCall(call) => Self::CustomToolCall(call), + TaggedInputItem::CustomToolCallOutput(output) => Self::CustomToolCallOutput(output), + TaggedInputItem::Reasoning(reasoning) => Self::Reasoning(reasoning), + TaggedInputItem::Compaction(compaction) => Self::Compaction(compaction), + TaggedInputItem::Unknown => Self::Unknown, + } + } +} + +impl<'de> Deserialize<'de> for InputItem { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + #[derive(Deserialize)] + #[serde(untagged)] + enum InputItemWire { + Tagged(TaggedInputItem), + Message(InputMessage), + } + + Ok(match InputItemWire::deserialize(deserializer)? { + InputItemWire::Tagged(item) => item.into(), + InputItemWire::Message(message) => Self::Message(message), + }) + } +} + impl InputItem { #[must_use] pub(crate) fn is_unknown(&self) -> bool { @@ -98,3 +202,146 @@ pub enum ResponsesInput { Text(String), Items(Vec), } + +#[derive(Debug, Clone, Copy)] +pub(crate) struct CompactionWindow { + latest_index: usize, + retained_start: usize, +} + +impl CompactionWindow { + #[must_use] + pub(crate) const fn latest_index(self) -> usize { + self.latest_index + } + + #[must_use] + pub(crate) fn retains_user_item(self, index: usize, item: &InputItem) -> bool { + index >= self.retained_start + && index < self.latest_index + && matches!(item, InputItem::Message(message) + if message.role == "user" + && message.id.is_some() + && message.status == Some(MessageStatus::Completed)) + } + + pub(crate) fn retained_user_items(self, items: &[InputItem]) -> impl Iterator { + items + .iter() + .enumerate() + .filter(move |(index, item)| self.retains_user_item(*index, item)) + .map(|(_, item)| item) + } +} + +#[must_use] +pub(crate) fn latest_compaction_window(items: &[InputItem]) -> Option { + let latest_index = items + .iter() + .rposition(|item| matches!(item, InputItem::Compaction(_)))?; + let retained_start = items[..latest_index] + .iter() + .rposition(|item| matches!(item, InputItem::Compaction(_))) + .map_or(0, |index| index + 1); + Some(CompactionWindow { + latest_index, + retained_start, + }) +} + +impl ResponsesInput { + #[must_use] + pub fn contains_compaction(&self) -> bool { + matches!(self, Self::Items(items) if items.iter().any(|item| matches!(item, InputItem::Compaction(_)))) + } + + /// Return the canonical context sent to vLLM. + /// + /// vLLM does not understand public `compaction` items, so the latest item + /// becomes an assistant message containing the locally generated summary. + /// Items before that checkpoint are superseded and are omitted. + #[must_use] + pub fn model_input(&self) -> Cow<'_, Self> { + let Self::Items(items) = self else { + return Cow::Borrowed(self); + }; + let Some(window) = latest_compaction_window(items) else { + return Cow::Borrowed(self); + }; + + let model_items = window + .retained_user_items(items) + .chain(items[window.latest_index()..].iter()) + .map(|item| match item { + InputItem::Compaction(compaction) => InputItem::Message(InputMessage { + id: None, + role: "assistant".to_owned(), + status: None, + content: InputMessageContent::Parts(vec![InputContent::OutputText(InputTextContent { + text: compaction.encrypted_content.clone(), + })]), + }), + other => other.clone(), + }) + .collect(); + Cow::Owned(Self::Items(model_items)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn function_call_input_accepts_missing_status() { + let item: InputItem = serde_json::from_value(serde_json::json!({ + "type": "function_call", + "id": "fc_1", + "call_id": "call_1", + "name": "lookup", + "arguments": "{}" + })) + .expect("valid replay input"); + + let InputItem::FunctionCall(call) = item else { + panic!("expected function call"); + }; + assert_eq!(call.status, None); + } + + #[test] + fn compaction_item_becomes_assistant_model_context() { + let input: ResponsesInput = serde_json::from_value(serde_json::json!([{ + "type": "compaction", + "id": "cmp_1", + "encrypted_content": "summary" + }])) + .expect("valid compaction input"); + + let model_input = input.model_input(); + let serialized = serde_json::to_value(model_input).expect("model input serializes"); + assert_eq!(serialized[0]["role"], "assistant"); + assert_eq!(serialized[0]["content"][0]["type"], "output_text"); + assert_eq!(serialized[0]["content"][0]["text"], "summary"); + } + + #[test] + fn latest_compaction_preserves_canonical_user_messages_and_supersedes_prior_context() { + let input: ResponsesInput = serde_json::from_value(serde_json::json!([ + {"role": "user", "content": "discard me"}, + {"type": "compaction", "encrypted_content": "old summary"}, + {"role": "assistant", "content": "also discard me"}, + {"type": "message", "id": "msg_keep", "role": "user", "status": "completed", "content": "retained user"}, + {"type": "compaction", "encrypted_content": "latest summary"}, + {"role": "user", "content": "keep me"} + ])) + .expect("valid compacted history"); + + let serialized = serde_json::to_value(input.model_input()).expect("model input serializes"); + assert_eq!(serialized.as_array().map(Vec::len), Some(3)); + assert_eq!(serialized[0]["content"], "retained user"); + assert_eq!(serialized[1]["role"], "assistant"); + assert_eq!(serialized[1]["content"][0]["text"], "latest summary"); + assert_eq!(serialized[2]["content"], "keep me"); + } +} diff --git a/crates/agentic-server-core/src/types/io/mod.rs b/crates/agentic-server-core/src/types/io/mod.rs index 514774c5..c0d7e4f1 100644 --- a/crates/agentic-server-core/src/types/io/mod.rs +++ b/crates/agentic-server-core/src/types/io/mod.rs @@ -4,8 +4,8 @@ pub mod tools; pub mod usage; pub use input::{ - CustomToolCallOutputMessage, FunctionToolResultMessage, InputContent, InputImageContent, InputItem, InputMessage, - InputMessageContent, InputTextContent, ResponsesInput, + CompactionItem, CustomToolCallOutputMessage, FunctionToolResultMessage, InputContent, InputFunctionToolCall, + InputImageContent, InputItem, InputMessage, InputMessageContent, InputTextContent, ResponsesInput, }; pub use output::{ ApplyDone, CustomToolCall, FunctionToolCall, GatewayCallStatus, McpToolCall, OutputItem, OutputMessage, diff --git a/crates/agentic-server-core/src/types/io/output.rs b/crates/agentic-server-core/src/types/io/output.rs index 387b0b1f..b567b134 100644 --- a/crates/agentic-server-core/src/types/io/output.rs +++ b/crates/agentic-server-core/src/types/io/output.rs @@ -7,7 +7,9 @@ use crate::tool::ToolRegistry; use crate::types::event::MessageStatus; use crate::utils::uuid7_str; -use super::input::{InputContent, InputItem, InputMessage, InputMessageContent, InputTextContent}; +use super::input::{ + InputContent, InputFunctionToolCall, InputItem, InputMessage, InputMessageContent, InputTextContent, +}; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct OutputTextContent { @@ -72,7 +74,9 @@ impl From for InputMessage { .map(|c| InputContent::OutputText(InputTextContent { text: c.text })) .collect(); Self { + id: Some(msg.id), role: msg.role, + status: Some(msg.status), content: InputMessageContent::Parts(parts), } } @@ -447,7 +451,7 @@ impl OutputItem { match self { Self::Message(message) => Some(InputItem::Message(message.clone().into())), Self::Reasoning(reasoning) => Some(InputItem::Reasoning(reasoning.clone())), - Self::FunctionCall(call) => Some(InputItem::FunctionCall(call.clone())), + Self::FunctionCall(call) => Some(InputItem::FunctionCall(InputFunctionToolCall::from(call.clone()))), Self::CustomToolCall(call) => Some(InputItem::CustomToolCall(call.clone())), Self::WebSearchCall(_) | Self::McpToolCall(_) | Self::Unknown => None, } diff --git a/crates/agentic-server-core/src/types/mod.rs b/crates/agentic-server-core/src/types/mod.rs index 0a0b0e27..e61ac141 100644 --- a/crates/agentic-server-core/src/types/mod.rs +++ b/crates/agentic-server-core/src/types/mod.rs @@ -5,13 +5,16 @@ pub mod request_response; pub mod tools; pub use io::{ - CustomToolCall, CustomToolCallOutputMessage, FunctionTool, FunctionToolCall, FunctionToolResultMessage, - GatewayCallStatus, InputContent, InputImageContent, InputItem, InputMessage, InputMessageContent, InputTextContent, - InputTokenDetails, McpToolCall, OutputItem, OutputMessage, OutputTextContent, OutputTokenDetails, ReasoningOutput, - ReasoningTextContent, ResponseUsage, ResponsesInput, ToolChoice, WebSearchActionSearch, WebSearchCall, - WebSearchCallStatus, WebSearchSource, + CompactionItem, CustomToolCall, CustomToolCallOutputMessage, FunctionTool, FunctionToolCall, + FunctionToolResultMessage, GatewayCallStatus, InputContent, InputFunctionToolCall, InputImageContent, InputItem, + InputMessage, InputMessageContent, InputTextContent, InputTokenDetails, McpToolCall, OutputItem, OutputMessage, + OutputTextContent, OutputTokenDetails, ReasoningOutput, ReasoningTextContent, ResponseUsage, ResponsesInput, + ToolChoice, WebSearchActionSearch, WebSearchCall, WebSearchCallStatus, WebSearchSource, +}; +pub use request_response::{ + CompactRequest, CompactedResponse, ContextManagement, IncompleteDetails, RequestPayload, ResponsePayload, + UpstreamRequest, UpstreamTool, }; -pub use request_response::{IncompleteDetails, RequestPayload, ResponsePayload, UpstreamRequest, UpstreamTool}; pub use tools::{ CodeInterpreterToolParam, CodexNamespaceMember, CodexNamespaceToolParam, CustomToolParam, EmptyToolNameError, FileSearchToolParam, FunctionToolParam, McpToolParam, NonEmptyToolName, ResponsesTool, WebSearchContextSize, diff --git a/crates/agentic-server-core/src/types/request_response.rs b/crates/agentic-server-core/src/types/request_response.rs index 56e43fe0..9dcec60b 100644 --- a/crates/agentic-server-core/src/types/request_response.rs +++ b/crates/agentic-server-core/src/types/request_response.rs @@ -1,3 +1,6 @@ +use std::borrow::Cow; +use std::collections::HashMap; + use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; @@ -31,6 +34,8 @@ pub struct RequestPayload { pub parallel_tool_calls: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub cache_salt: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub context_management: Option>, } fn default_true() -> bool { @@ -40,7 +45,7 @@ fn default_true() -> bool { #[derive(Debug, Serialize)] pub struct UpstreamRequest<'a> { pub model: &'a str, - pub input: &'a ResponsesInput, + pub input: Cow<'a, ResponsesInput>, pub stream: bool, #[serde(skip_serializing_if = "Option::is_none")] pub instructions: Option<&'a str>, @@ -154,7 +159,7 @@ impl RequestPayload { let tool_choice = CodexNamespaceHandler.resolve_tool_choice(namespace_map.as_ref(), self.tool_choice.as_ref()); Ok(UpstreamRequest { model: &self.model, - input: &self.input, + input: self.input.model_input(), stream, instructions: self.instructions.as_deref(), tools, @@ -177,6 +182,40 @@ impl RequestPayload { } } +/// Server-side context management configuration for a Responses request. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ContextManagement { + #[serde(rename = "type")] + pub type_: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub compact_threshold: Option, +} + +/// Request body accepted by `POST /v1/responses/compact`. +#[derive(Debug, Clone, Deserialize)] +pub struct CompactRequest { + pub model: String, + #[serde(default)] + pub input: Option, + #[serde(default)] + pub instructions: Option, + #[serde(default)] + pub previous_response_id: Option, + /// Compatibility fields sent by current SDK and Codex clients. + #[serde(flatten)] + pub compatibility: HashMap, +} + +/// Result returned by `POST /v1/responses/compact`. +#[derive(Debug, Clone, Serialize)] +pub struct CompactedResponse { + pub id: String, + pub object: String, + pub created_at: i64, + pub output: Vec, + pub usage: ResponseUsage, +} + fn upstream_tools(tool: ResponsesTool) -> Vec { match tool { ResponsesTool::Custom(declaration) => { @@ -260,7 +299,9 @@ impl From<&ResponsesInput> for Vec { fn from(input: &ResponsesInput) -> Self { match input { ResponsesInput::Text(text) => vec![InputItem::Message(InputMessage { + id: None, role: "user".into(), + status: None, content: InputMessageContent::Text(text.clone()), })], ResponsesInput::Items(items) => items.iter().filter(|item| !item.is_unknown()).cloned().collect(), @@ -268,10 +309,41 @@ impl From<&ResponsesInput> for Vec { } } +impl From for Vec { + fn from(input: ResponsesInput) -> Self { + match input { + ResponsesInput::Text(text) => vec![InputItem::Message(InputMessage { + id: None, + role: "user".into(), + status: None, + content: InputMessageContent::Text(text), + })], + ResponsesInput::Items(items) => items.into_iter().filter(|item| !item.is_unknown()).collect(), + } + } +} + #[cfg(test)] mod tests { use super::*; + #[test] + fn compact_request_accepts_codex_compatibility_fields() { + let request: CompactRequest = serde_json::from_value(serde_json::json!({ + "model": "test-model", + "input": [{"role": "user", "content": "hello"}], + "tools": [], + "parallel_tool_calls": true, + "reasoning": {"effort": "medium"}, + "text": {"verbosity": "low"} + })) + .expect("compact request should parse"); + + assert_eq!(request.model, "test-model"); + assert!(request.input.is_some()); + assert_eq!(request.compatibility.len(), 4); + } + #[test] fn request_payload_forwards_cache_salt_upstream() { let payload: RequestPayload = serde_json::from_value(serde_json::json!({ diff --git a/crates/agentic-server-core/tests/cassette_support_test.rs b/crates/agentic-server-core/tests/cassette_support_test.rs new file mode 100644 index 00000000..ed97577f --- /dev/null +++ b/crates/agentic-server-core/tests/cassette_support_test.rs @@ -0,0 +1,49 @@ +mod support; + +#[test] +fn cassette_accepts_structured_responses_input() { + let cassette = serde_yaml::from_str::( + r" +turns: +- request: + path: /v1/responses + body: + input: + - role: user + content: hello + response: + body: {} +", + ); + + assert!( + cassette.is_ok(), + "structured Responses input should deserialize: {cassette:?}" + ); +} + +#[test] +fn request_builder_preserves_structured_responses_input() { + let input = serde_json::json!([ + { + "role": "user", + "content": [ + {"type": "input_text", "text": "Describe this image."}, + { + "type": "input_image", + "image_url": "data:image/png;base64,abc", + "detail": "low" + } + ] + } + ]); + + let request = support::make_request(&input, false, false, None, None); + + let serialized = serde_json::to_value(request.input).expect("serialize request input"); + assert_eq!(serialized[0]["role"], "user"); + assert_eq!(serialized[0]["content"][0]["text"], "Describe this image."); + assert_eq!(serialized[0]["content"][1]["type"], "input_image"); + assert_eq!(serialized[0]["content"][1]["image_url"], "data:image/png;base64,abc"); + assert_eq!(serialized[0]["content"][1]["detail"], "low"); +} diff --git a/crates/agentic-server-core/tests/cassettes/README.md b/crates/agentic-server-core/tests/cassettes/README.md index aabcfe8d..f5aea881 100644 --- a/crates/agentic-server-core/tests/cassettes/README.md +++ b/crates/agentic-server-core/tests/cassettes/README.md @@ -22,6 +22,9 @@ printf 'First prompt\nSecond prompt\n' | python tests/cassettes/record_cassette. # gateway-backed cassette -- records the gateway-facing request/response printf 'Use web search to look up potato, then summarize in one sentence.\n' | python tests/cassettes/record_cassette.py --mode responses --turns 1 --no-stream --gateway http://localhost:9000 --model openai/gpt-oss-20b --output out.yaml + +# structured single-turn input -- sends the JSON string or item array from input.json +python tests/cassettes/record_cassette.py --mode responses --turns 1 --no-stream --no-store --max-output-tokens 0 --input-file input.json --model gpt-4o --output out.yaml ``` The recorder scripts (`record_reasoning_cassettes.sh`, `record_tool_call_cassettes.sh`, etc.) use `printf` to feed fixed prompts per test so no manual input is needed. @@ -51,6 +54,7 @@ The recorder scripts (`record_reasoning_cassettes.sh`, `record_tool_call_cassett --openai URL OpenAI upstream (default https://api.openai.com) --tools FILE JSON file containing a tools array (responses mode only) --tool-choice VALUE "auto", "none", "required", or JSON e.g. '{"type":"function","name":"foo"}' +--input-file FILE JSON string or item array for one HTTP Responses turn --max-output-tokens N max_output_tokens for Responses requests (default 1024; use 0 to omit) --proxy-port PORT Local proxy port (default 7070) --branch-from TURN Branch from this turn's response id (repeatable) @@ -204,3 +208,37 @@ OPENAI_API_KEY=sk-... \ OPENAI_CUSTOM_MODEL=gpt-5.6 \ bash tests/cassettes/record_codex_cli_tool_call_cassettes.sh openai-custom ``` + +### Compaction replay (OpenAI) + +These recordings capture the non-streaming `/v1/responses` inference calls replayed by the compaction integration +tests. The JSON inputs contain the exact model-facing item arrays, including the context-checkpoint prompt. Use the +existing recorder directly from `crates/agentic-server-core`: + +```bash +record_compaction() { + input_name="$1" + output_name="$2" + uv run \ + --with click \ + --with fastapi \ + --with httpx \ + --with uvicorn \ + --with pyyaml \ + python tests/cassettes/record_cassette.py \ + --mode responses \ + --turns 1 \ + --no-stream \ + --no-store \ + --max-output-tokens 0 \ + --openai https://api.openai.com \ + --model gpt-4o \ + --input-file "tests/cassettes/compaction/inputs/${input_name}.json" \ + --output "tests/cassettes/compaction/compact-${output_name}-gpt-4o-nonstreaming.yaml" +} + +export OPENAI_API_KEY=sk-... +record_compaction basic basic +record_compaction tool-prior-compaction tool-prior +record_compaction followup followup +``` diff --git a/crates/agentic-server-core/tests/cassettes/compaction/compact-basic-gpt-4o-nonstreaming.yaml b/crates/agentic-server-core/tests/cassettes/compaction/compact-basic-gpt-4o-nonstreaming.yaml new file mode 100644 index 00000000..ea511397 --- /dev/null +++ b/crates/agentic-server-core/tests/cassettes/compaction/compact-basic-gpt-4o-nonstreaming.yaml @@ -0,0 +1,153 @@ +turns: +- filename: t1 + request: + body: + input: + - content: Remember the stable marker BANANA-CASSETTE for the next model. + role: user + type: message + - content: I will preserve BANANA-CASSETTE in the checkpoint. + role: assistant + type: message + - content: The implementation uses Rust and must keep offline replay tests. + role: user + type: message + - content: The constraints are OpenAI-only recordings and no network access + in CI. + role: assistant + type: message + - content: You are performing a CONTEXT CHECKPOINT COMPACTION. Create a concise + handoff summary that preserves current progress, decisions, constraints, + unresolved work, and critical references for the next model. Return only + the summary. + role: user + type: message + model: gpt-4o + store: false + stream: false + headers: + accept: '*/*' + authorization: Bearer *** + content-type: application/json + user-agent: python-httpx/0.28.1 + method: POST + path: /v1/responses + query_params: {} + response: + body: + background: false + billing: + payer: developer + completed_at: 1784780443 + created_at: 1784780441 + error: null + frequency_penalty: 0.0 + id: resp_0bb2e72c44a04a1c016a6196995f0c81a288ee80d885c0a0b1 + incomplete_details: null + instructions: null + max_output_tokens: null + max_tool_calls: null + metadata: {} + model: gpt-4o-2024-08-06 + moderation: null + object: response + output: + - content: + - annotations: [] + logprobs: [] + text: '### Handoff Summary + + + **Current Progress:** + + - Stable marker "BANANA-CASSETTE" set for consistency. + + - Implementation is in Rust, ensuring offline replay tests. + + + **Decisions:** + + - OpenAI-only recordings are to be maintained. + + - CI environment must not have network access. + + + **Constraints:** + + - Adhere strictly to offline testing procedures within CI. + + - Ensure stability and reproducibility of test results using "BANANA-CASSETTE." + + + **Unresolved Work:** + + - Further optimization may be needed for test case efficiency. + + - Validation of complete test coverage in offline mode. + + + **Critical References:** + + - Marker: "BANANA-CASSETTE" + + - Language: Rust implementation + + - Testing Mode: Offline replay tests + + + This summary should guide the next model in maintaining current parameters + and addressing pending tasks.' + type: output_text + id: msg_0bb2e72c44a04a1c016a619699b65881a28944eeac947d31ee + role: assistant + status: completed + type: message + parallel_tool_calls: true + presence_penalty: 0.0 + previous_response_id: null + prompt_cache_key: null + prompt_cache_retention: in_memory + reasoning: + context: null + effort: null + summary: null + safety_identifier: null + service_tier: default + status: completed + store: false + temperature: 1.0 + text: + format: + type: text + verbosity: medium + tool_choice: auto + tool_usage: + image_gen: + input_tokens: 0 + input_tokens_details: + image_tokens: 0 + text_tokens: 0 + output_tokens: 0 + output_tokens_details: + image_tokens: 0 + text_tokens: 0 + total_tokens: 0 + web_search: + num_requests: 0 + tools: [] + top_logprobs: 0 + top_p: 1.0 + truncation: disabled + usage: + input_tokens: 119 + input_tokens_details: + cache_write_tokens: 0 + cached_tokens: 0 + output_tokens: 167 + output_tokens_details: + reasoning_tokens: 0 + total_tokens: 286 + user: null + headers: + content-type: application/json + status_code: 200 diff --git a/crates/agentic-server-core/tests/cassettes/compaction/compact-followup-gpt-4o-nonstreaming.yaml b/crates/agentic-server-core/tests/cassettes/compaction/compact-followup-gpt-4o-nonstreaming.yaml new file mode 100644 index 00000000..4090e10d --- /dev/null +++ b/crates/agentic-server-core/tests/cassettes/compaction/compact-followup-gpt-4o-nonstreaming.yaml @@ -0,0 +1,105 @@ +turns: +- filename: t1 + request: + body: + input: + - content: Remember the stable marker BANANA-CASSETTE for the next model. + role: user + type: message + - content: + - text: 'Checkpoint: preserve the stable marker BANANA-CASSETTE. The implementation + uses Rust, OpenAI-only recordings, and offline replay tests.' + type: output_text + role: assistant + type: message + - content: Reply with only the stable marker from the checkpoint. + role: user + type: message + model: gpt-4o + store: false + stream: false + headers: + accept: '*/*' + authorization: Bearer *** + content-type: application/json + user-agent: python-httpx/0.28.1 + method: POST + path: /v1/responses + query_params: {} + response: + body: + background: false + billing: + payer: developer + completed_at: 1784780450 + created_at: 1784780449 + error: null + frequency_penalty: 0.0 + id: resp_0368501b2bc20ade016a6196a1cd5081928c8de9f57bb3c53d + incomplete_details: null + instructions: null + max_output_tokens: null + max_tool_calls: null + metadata: {} + model: gpt-4o-2024-08-06 + moderation: null + object: response + output: + - content: + - annotations: [] + logprobs: [] + text: BANANA-CASSETTE + type: output_text + id: msg_0368501b2bc20ade016a6196a238448192a7d716bdb9ae8558 + role: assistant + status: completed + type: message + parallel_tool_calls: true + presence_penalty: 0.0 + previous_response_id: null + prompt_cache_key: null + prompt_cache_retention: in_memory + reasoning: + context: null + effort: null + summary: null + safety_identifier: null + service_tier: default + status: completed + store: false + temperature: 1.0 + text: + format: + type: text + verbosity: medium + tool_choice: auto + tool_usage: + image_gen: + input_tokens: 0 + input_tokens_details: + image_tokens: 0 + text_tokens: 0 + output_tokens: 0 + output_tokens_details: + image_tokens: 0 + text_tokens: 0 + total_tokens: 0 + web_search: + num_requests: 0 + tools: [] + top_logprobs: 0 + top_p: 1.0 + truncation: disabled + usage: + input_tokens: 68 + input_tokens_details: + cache_write_tokens: 0 + cached_tokens: 0 + output_tokens: 7 + output_tokens_details: + reasoning_tokens: 0 + total_tokens: 75 + user: null + headers: + content-type: application/json + status_code: 200 diff --git a/crates/agentic-server-core/tests/cassettes/compaction/compact-tool-prior-gpt-4o-nonstreaming.yaml b/crates/agentic-server-core/tests/cassettes/compaction/compact-tool-prior-gpt-4o-nonstreaming.yaml new file mode 100644 index 00000000..d244c856 --- /dev/null +++ b/crates/agentic-server-core/tests/cassettes/compaction/compact-tool-prior-gpt-4o-nonstreaming.yaml @@ -0,0 +1,160 @@ +turns: +- filename: t1 + request: + body: + input: + - content: Keep the stable marker BANANA-CASSETTE. + id: msg_retained + role: user + status: completed + type: message + - content: + - text: 'Prior checkpoint: preserve BANANA-CASSETTE and continue the Rust + replay work.' + type: output_text + role: assistant + type: message + - content: Use the documentation lookup result when updating the tests. + role: user + type: message + - arguments: '{"topic":"compaction"}' + call_id: call_lookup + id: fc_lookup + name: lookup_docs + status: completed + type: function_call + - call_id: call_lookup + output: '{"requirement":"record OpenAI replay cassettes"}' + type: function_call_output + - content: You are performing a CONTEXT CHECKPOINT COMPACTION. Create a concise + handoff summary that preserves current progress, decisions, constraints, + unresolved work, and critical references for the next model. Return only + the summary. + role: user + type: message + model: gpt-4o + store: false + stream: false + headers: + accept: '*/*' + authorization: Bearer *** + content-type: application/json + user-agent: python-httpx/0.28.1 + method: POST + path: /v1/responses + query_params: {} + response: + body: + background: false + billing: + payer: developer + completed_at: 1784780448 + created_at: 1784780444 + error: null + frequency_penalty: 0.0 + id: resp_085f046ab3b0de52016a61969ccbec81a0957d7e1633c06a15 + incomplete_details: null + instructions: null + max_output_tokens: null + max_tool_calls: null + metadata: {} + model: gpt-4o-2024-08-06 + moderation: null + object: response + output: + - content: + - annotations: [] + logprobs: [] + text: '**Handoff Summary:** + + + **Current Progress:** + + - Stable marker BANANA-CASSETTE set. + + - Rust replay work ongoing. + + - Documentation lookup for compaction completed. + + + **Decisions:** + + - Use OpenAI replay cassettes for compaction references. + + - Update tests using the documentation insights. + + + **Constraints:** + + - Maintain BANANA-CASSETTE as a stable marker. + + - Integrate documentation requirements for recording OpenAI replay cassettes. + + + **Unresolved Work:** + + - Further integration of documentation findings into test updates. + + - Continuation of Rust replay task. + + + **Critical References:** + + - Documentation on recording OpenAI replay cassettes for compaction. + + + End of summary.' + type: output_text + id: msg_085f046ab3b0de52016a61969d538081a0b4486be5f1edb083 + role: assistant + status: completed + type: message + parallel_tool_calls: true + presence_penalty: 0.0 + previous_response_id: null + prompt_cache_key: null + prompt_cache_retention: in_memory + reasoning: + context: null + effort: null + summary: null + safety_identifier: null + service_tier: default + status: completed + store: false + temperature: 1.0 + text: + format: + type: text + verbosity: medium + tool_choice: auto + tool_usage: + image_gen: + input_tokens: 0 + input_tokens_details: + image_tokens: 0 + text_tokens: 0 + output_tokens: 0 + output_tokens_details: + image_tokens: 0 + text_tokens: 0 + total_tokens: 0 + web_search: + num_requests: 0 + tools: [] + top_logprobs: 0 + top_p: 1.0 + truncation: disabled + usage: + input_tokens: 133 + input_tokens_details: + cache_write_tokens: 0 + cached_tokens: 0 + output_tokens: 141 + output_tokens_details: + reasoning_tokens: 0 + total_tokens: 274 + user: null + headers: + content-type: application/json + status_code: 200 diff --git a/crates/agentic-server-core/tests/cassettes/compaction/inputs/basic.json b/crates/agentic-server-core/tests/cassettes/compaction/inputs/basic.json new file mode 100644 index 00000000..d4e279b5 --- /dev/null +++ b/crates/agentic-server-core/tests/cassettes/compaction/inputs/basic.json @@ -0,0 +1,27 @@ +[ + { + "type": "message", + "role": "user", + "content": "Remember the stable marker BANANA-CASSETTE for the next model." + }, + { + "type": "message", + "role": "assistant", + "content": "I will preserve BANANA-CASSETTE in the checkpoint." + }, + { + "type": "message", + "role": "user", + "content": "The implementation uses Rust and must keep offline replay tests." + }, + { + "type": "message", + "role": "assistant", + "content": "The constraints are OpenAI-only recordings and no network access in CI." + }, + { + "type": "message", + "role": "user", + "content": "You are performing a CONTEXT CHECKPOINT COMPACTION. Create a concise handoff summary that preserves current progress, decisions, constraints, unresolved work, and critical references for the next model. Return only the summary." + } +] diff --git a/crates/agentic-server-core/tests/cassettes/compaction/inputs/followup.json b/crates/agentic-server-core/tests/cassettes/compaction/inputs/followup.json new file mode 100644 index 00000000..936d0c70 --- /dev/null +++ b/crates/agentic-server-core/tests/cassettes/compaction/inputs/followup.json @@ -0,0 +1,22 @@ +[ + { + "type": "message", + "role": "user", + "content": "Remember the stable marker BANANA-CASSETTE for the next model." + }, + { + "type": "message", + "role": "assistant", + "content": [ + { + "type": "output_text", + "text": "Checkpoint: preserve the stable marker BANANA-CASSETTE. The implementation uses Rust, OpenAI-only recordings, and offline replay tests." + } + ] + }, + { + "type": "message", + "role": "user", + "content": "Reply with only the stable marker from the checkpoint." + } +] diff --git a/crates/agentic-server-core/tests/cassettes/compaction/inputs/tool-prior-compaction.json b/crates/agentic-server-core/tests/cassettes/compaction/inputs/tool-prior-compaction.json new file mode 100644 index 00000000..1f94d745 --- /dev/null +++ b/crates/agentic-server-core/tests/cassettes/compaction/inputs/tool-prior-compaction.json @@ -0,0 +1,42 @@ +[ + { + "type": "message", + "id": "msg_retained", + "role": "user", + "status": "completed", + "content": "Keep the stable marker BANANA-CASSETTE." + }, + { + "type": "message", + "role": "assistant", + "content": [ + { + "type": "output_text", + "text": "Prior checkpoint: preserve BANANA-CASSETTE and continue the Rust replay work." + } + ] + }, + { + "type": "message", + "role": "user", + "content": "Use the documentation lookup result when updating the tests." + }, + { + "type": "function_call", + "id": "fc_lookup", + "call_id": "call_lookup", + "name": "lookup_docs", + "arguments": "{\"topic\":\"compaction\"}", + "status": "completed" + }, + { + "type": "function_call_output", + "call_id": "call_lookup", + "output": "{\"requirement\":\"record OpenAI replay cassettes\"}" + }, + { + "type": "message", + "role": "user", + "content": "You are performing a CONTEXT CHECKPOINT COMPACTION. Create a concise handoff summary that preserves current progress, decisions, constraints, unresolved work, and critical references for the next model. Return only the summary." + } +] diff --git a/crates/agentic-server-core/tests/cassettes/record_cassette.py b/crates/agentic-server-core/tests/cassettes/record_cassette.py index 340027db..ff00d80c 100644 --- a/crates/agentic-server-core/tests/cassettes/record_cassette.py +++ b/crates/agentic-server-core/tests/cassettes/record_cassette.py @@ -634,6 +634,18 @@ def _prompt(label: str) -> str: sys.exit(0) +def _load_response_input(path: str | None) -> str | list | None: + if path is None: + return None + try: + value = json.loads(Path(path).read_text(encoding="utf-8")) + except json.JSONDecodeError as error: + raise click.UsageError(f"--input-file is not valid JSON: {error}") from error + if not isinstance(value, (str, list)): + raise click.UsageError("--input-file must contain a JSON string or array.") + return value + + def _inject_tools(body: dict, tools: list | None, tool_choice: Any) -> None: if tools is not None: body["tools"] = tools @@ -916,6 +928,7 @@ def run_responses( tool_choice: Any = None, tool_outputs: dict[str, str] | None = None, max_output_tokens: int | None = None, + preset_input: str | list | None = None, ) -> None: response_ids: dict[int, str] = {} responses: dict[int, dict] = {} @@ -942,19 +955,22 @@ def run_responses( click.echo( f"\n[Branch] turn {turn} chains from turn {branch_from} (response_id={previous_response_id})" ) - prompt = _prompt(f"Turn {turn}/{turns} — enter prompt: ") - - # Inject matching function/custom output items before the user message. - pending_calls = _extract_tool_calls(last_response) if tool_outputs else [] - if pending_calls and tool_outputs: - input_value: Any = _build_tool_output_input( - pending_calls, tool_outputs, prompt if prompt else None - ) - click.echo( - f" [injecting {len(pending_calls)} tool output(s) before user message]" - ) + if preset_input is not None: + input_value: Any = preset_input else: - input_value = prompt + prompt = _prompt(f"Turn {turn}/{turns} — enter prompt: ") + + # Inject matching function/custom output items before the user message. + pending_calls = _extract_tool_calls(last_response) if tool_outputs else [] + if pending_calls and tool_outputs: + input_value = _build_tool_output_input( + pending_calls, tool_outputs, prompt if prompt else None + ) + click.echo( + f" [injecting {len(pending_calls)} tool output(s) before user message]" + ) + else: + input_value = prompt body: dict = {"model": model, "input": input_value, "stream": stream, "store": store} if max_output_tokens is not None: @@ -1130,6 +1146,11 @@ def run_responses( "When provided, matching function_call_output or custom_tool_call_output items are injected " "between turns (required for OpenAI Responses API).", ) +@click.option( + "--input-file", + type=click.Path(exists=True, dir_okay=False), + help="JSON file containing one Responses input value; requires HTTP --mode responses --turns 1.", +) @click.option( "--max-output-tokens", type=int, @@ -1154,6 +1175,7 @@ def main( tools_file: str | None, tool_choice_raw: str | None, tool_outputs_file: str | None, + input_file: str | None, max_output_tokens: int, ) -> None: """Interactive multi-turn cassette recorder (proxy embedded).""" @@ -1181,6 +1203,11 @@ def main( ) if max_output_tokens < 0: raise click.UsageError("--max-output-tokens must be >= 0.") + if input_file and (mode != "responses" or turns != 1 or branches or transport != "http"): + raise click.UsageError( + "--input-file requires HTTP --mode responses --turns 1 without branches." + ) + preset_input = _load_response_input(input_file) tools: list | None = None if tools_file: @@ -1258,6 +1285,7 @@ def main( tool_choice, tool_outputs, response_max_output_tokens, + preset_input, ) else: click.echo(f"Proxy: {proxy_url} (requests go through here for recording)") @@ -1289,6 +1317,7 @@ def main( tool_choice, tool_outputs, response_max_output_tokens, + preset_input, ) elif mode == "messages": run_messages( diff --git a/crates/agentic-server-core/tests/compaction_cassette_test.rs b/crates/agentic-server-core/tests/compaction_cassette_test.rs new file mode 100644 index 00000000..774eec75 --- /dev/null +++ b/crates/agentic-server-core/tests/compaction_cassette_test.rs @@ -0,0 +1,247 @@ +mod support; + +use std::sync::Arc; + +use agentic_core::executor::{compact_response, execute}; +use agentic_core::{CompactRequest, InputItem, RequestPayload}; +use serde_json::{Value, json}; +use support::{MockResponse, TestFixture, load_cassette, output_text, unwrap_blocking}; + +const DIR: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/tests/cassettes/compaction"); +const COMPACTION_PROMPT: &str = "You are performing a CONTEXT CHECKPOINT COMPACTION. Create a concise handoff summary that preserves current progress, decisions, constraints, unresolved work, and critical references for the next model. Return only the summary."; + +fn basic_source_input() -> Value { + json!([ + { + "type": "message", + "role": "user", + "content": "Remember the stable marker BANANA-CASSETTE for the next model." + }, + { + "type": "message", + "role": "assistant", + "content": "I will preserve BANANA-CASSETTE in the checkpoint." + }, + { + "type": "message", + "role": "user", + "content": "The implementation uses Rust and must keep offline replay tests." + }, + { + "type": "message", + "role": "assistant", + "content": "The constraints are OpenAI-only recordings and no network access in CI." + } + ]) +} + +fn tool_prior_source_input() -> Value { + json!([ + { + "type": "message", + "id": "msg_retained", + "role": "user", + "status": "completed", + "content": "Keep the stable marker BANANA-CASSETTE." + }, + { + "type": "compaction", + "id": "cmp_prior", + "encrypted_content": "Prior checkpoint: preserve BANANA-CASSETTE and continue the Rust replay work." + }, + { + "type": "message", + "role": "user", + "content": "Use the documentation lookup result when updating the tests." + }, + { + "type": "function_call", + "id": "fc_lookup", + "call_id": "call_lookup", + "name": "lookup_docs", + "arguments": "{\"topic\":\"compaction\"}", + "status": "completed" + }, + { + "type": "function_call_output", + "call_id": "call_lookup", + "output": "{\"requirement\":\"record OpenAI replay cassettes\"}" + } + ]) +} + +fn compact_request(input: &Value) -> CompactRequest { + serde_json::from_value(json!({ + "model": "gpt-4o", + "input": input + })) + .expect("valid compact request") +} + +fn response_request(input: &Value, context_management: Option<&Value>) -> RequestPayload { + serde_json::from_value(json!({ + "model": "gpt-4o", + "input": input, + "stream": false, + "store": false, + "context_management": context_management + })) + .expect("valid Responses request") +} + +fn cassette(name: &str) -> support::Cassette { + load_cassette(&format!("{DIR}/{name}")) +} + +fn cassette_response(turn: &support::Turn) -> MockResponse { + MockResponse::from_turn(turn) +} + +fn recorded_total_tokens(turn: &support::Turn) -> i64 { + turn.response + .body + .as_ref() + .and_then(|body| body["usage"]["total_tokens"].as_i64()) + .expect("recorded response usage") +} + +#[tokio::test] +async fn standalone_compaction_replays_openai_summary() { + let recording = cassette("compact-basic-gpt-4o-nonstreaming.yaml"); + let turn = &recording.turns[0]; + let fixture = TestFixture::new(&[turn]).await; + + let compacted = compact_response(compact_request(&basic_source_input()), &fixture.exec_ctx, None) + .await + .expect("compact recorded history"); + + assert_eq!(compacted.object, "response.compaction"); + assert_eq!( + compacted + .output + .iter() + .filter(|item| matches!(item, InputItem::Message(_))) + .count(), + 2 + ); + assert!(matches!(compacted.output.last(), Some(InputItem::Compaction(_)))); + + let requests = fixture.request_bodies().await; + assert_eq!(requests.len(), 1); + assert_eq!(requests[0]["input"], turn.request.body.input); + assert_eq!( + requests[0]["input"] + .as_array() + .and_then(|items| items.last()) + .expect("summary prompt item")["content"], + COMPACTION_PROMPT + ); +} + +#[tokio::test] +async fn repeated_compaction_drops_tool_pair_and_replaces_prior_checkpoint() { + let recording = cassette("compact-tool-prior-gpt-4o-nonstreaming.yaml"); + let turn = &recording.turns[0]; + let fixture = TestFixture::new(&[turn]).await; + + let compacted = compact_response(compact_request(&tool_prior_source_input()), &fixture.exec_ctx, None) + .await + .expect("compact recorded tool history"); + + assert_eq!( + compacted + .output + .iter() + .filter(|item| matches!(item, InputItem::Message(_))) + .count(), + 2 + ); + assert_eq!( + compacted + .output + .iter() + .filter(|item| matches!(item, InputItem::Compaction(_))) + .count(), + 1 + ); + assert!( + compacted + .output + .iter() + .all(|item| !matches!(item, InputItem::FunctionCall(_) | InputItem::FunctionCallOutput(_))) + ); + + let requests = fixture.request_bodies().await; + assert_eq!(requests[0]["input"], turn.request.body.input); +} + +#[tokio::test] +async fn compacted_window_round_trips_with_recorded_followup() { + let summary = cassette("compact-basic-gpt-4o-nonstreaming.yaml"); + let followup = cassette("compact-followup-gpt-4o-nonstreaming.yaml"); + let fixture = TestFixture::new_with_responses(vec![ + cassette_response(&summary.turns[0]), + cassette_response(&followup.turns[0]), + ]) + .await; + + let compacted = compact_response(compact_request(&basic_source_input()), &fixture.exec_ctx, None) + .await + .expect("compact recorded history"); + let mut input = compacted.output; + input.push( + serde_json::from_value(json!({ + "type": "message", + "role": "user", + "content": "Reply with only the stable marker from the checkpoint." + })) + .expect("valid follow-up item"), + ); + let response = unwrap_blocking( + execute( + response_request(&serde_json::to_value(input).expect("serialize compacted input"), None), + Arc::clone(&fixture.exec_ctx), + ) + .await + .expect("execute recorded follow-up"), + ); + + assert_eq!(output_text(&response).trim(), "BANANA-CASSETTE"); +} + +#[tokio::test] +async fn automatic_compaction_replays_both_rounds_and_accumulates_usage() { + let summary = cassette("compact-basic-gpt-4o-nonstreaming.yaml"); + let followup = cassette("compact-followup-gpt-4o-nonstreaming.yaml"); + let expected_usage = recorded_total_tokens(&summary.turns[0]) + recorded_total_tokens(&followup.turns[0]); + let fixture = TestFixture::new_with_responses(vec![ + cassette_response(&summary.turns[0]), + cassette_response(&followup.turns[0]), + ]) + .await; + + let response = unwrap_blocking( + execute( + response_request( + &basic_source_input(), + Some(&json!([{"type": "compaction", "compact_threshold": 1}])), + ), + Arc::clone(&fixture.exec_ctx), + ) + .await + .expect("execute automatic compaction"), + ); + + assert_eq!( + response.usage.as_ref().map(|usage| usage.total_tokens), + Some(expected_usage) + ); + let requests = fixture.request_bodies().await; + assert_eq!(requests.len(), 2); + assert_eq!(requests[0]["input"], summary.turns[0].request.body.input); + assert!( + requests + .iter() + .all(|request| request.get("context_management").is_none()) + ); +} diff --git a/crates/agentic-server-core/tests/dispatch_loop_cassette_test.rs b/crates/agentic-server-core/tests/dispatch_loop_cassette_test.rs index 772a4649..dd854a70 100644 --- a/crates/agentic-server-core/tests/dispatch_loop_cassette_test.rs +++ b/crates/agentic-server-core/tests/dispatch_loop_cassette_test.rs @@ -129,6 +129,7 @@ fn request(text: &str, tools: Option>) -> RequestPayload { metadata: None, parallel_tool_calls: None, cache_salt: None, + context_management: None, } } diff --git a/crates/agentic-server-core/tests/stateful_conversation_integration.rs b/crates/agentic-server-core/tests/stateful_conversation_integration.rs index 6ab4817d..5c715633 100644 --- a/crates/agentic-server-core/tests/stateful_conversation_integration.rs +++ b/crates/agentic-server-core/tests/stateful_conversation_integration.rs @@ -357,9 +357,9 @@ async fn test_store_false_with_conversation_id_hydrates_and_persists() { assert_eq!( request_input_texts(&requests[1]), vec![ - t1.request.body.input.as_str(), + t1.request.body.input.as_str().expect("text cassette input"), expected_text(t1).as_str(), - t2.request.body.input.as_str() + t2.request.body.input.as_str().expect("text cassette input") ] ); } diff --git a/crates/agentic-server-core/tests/storage_integration.rs b/crates/agentic-server-core/tests/storage_integration.rs index 72b0a041..49e1f19c 100644 --- a/crates/agentic-server-core/tests/storage_integration.rs +++ b/crates/agentic-server-core/tests/storage_integration.rs @@ -14,7 +14,9 @@ use support::setup_pool; fn create_input_item(text: &str) -> InOutItem { InOutItem::Input(InputItem::Message(InputMessage { + id: None, role: "user".to_string(), + status: None, content: InputMessageContent::Text(text.to_string()), })) } diff --git a/crates/agentic-server-core/tests/support/mod.rs b/crates/agentic-server-core/tests/support/mod.rs index 6167ab46..b979f3fa 100644 --- a/crates/agentic-server-core/tests/support/mod.rs +++ b/crates/agentic-server-core/tests/support/mod.rs @@ -15,14 +15,14 @@ use axum::response::{IntoResponse, Response}; use axum::routing::post; use either::Either; use futures::StreamExt; -use serde::Deserialize; +use serde::{Deserialize, Serialize}; use serde_json::Value; use tokio::sync::Mutex; use tokio::task::JoinHandle; use agentic_core::executor::{BoxStream, ConversationHandler, ExecutionContext, ResponseHandler}; use agentic_core::storage::{ConversationStore, DbPool, ResponseStore, create_pool_with_schema}; -use agentic_core::types::io::{OutputItem, ResponsesInput}; +use agentic_core::types::io::OutputItem; use agentic_core::types::request_response::{RequestPayload, ResponsePayload}; #[derive(Debug, Deserialize)] @@ -45,7 +45,7 @@ pub struct TurnRequest { #[derive(Debug, Deserialize, Default)] pub struct TurnBody { #[serde(default)] - pub input: String, + pub input: Value, #[serde(default = "default_true")] pub store: bool, #[serde(default)] @@ -347,7 +347,7 @@ pub fn request_input_texts(body: &Value) -> Vec { } pub fn make_request( - input: &str, + input: impl Serialize, store: bool, stream: bool, previous_response_id: Option, @@ -355,7 +355,8 @@ pub fn make_request( ) -> RequestPayload { RequestPayload { model: "test-model".to_string(), - input: ResponsesInput::Text(input.to_string()), + input: serde_json::from_value(serde_json::to_value(input).expect("serialize Responses input")) + .expect("request should contain valid Responses input"), instructions: None, previous_response_id, conversation_id, @@ -371,6 +372,7 @@ pub fn make_request( metadata: None, parallel_tool_calls: None, cache_salt: None, + context_management: None, } } diff --git a/crates/agentic-server-core/tests/web_search_tool_test.rs b/crates/agentic-server-core/tests/web_search_tool_test.rs index bd5bcb73..5ba3565c 100644 --- a/crates/agentic-server-core/tests/web_search_tool_test.rs +++ b/crates/agentic-server-core/tests/web_search_tool_test.rs @@ -804,6 +804,7 @@ async fn execute_runs_web_search_and_sends_tool_output_back_to_model() { metadata: None, parallel_tool_calls: None, cache_salt: None, + context_management: None, }; let result = ExecuteRequest::new(payload, Arc::clone(&exec_ctx)).run().await.unwrap(); @@ -889,6 +890,7 @@ async fn execute_relaxes_forced_tool_choice_after_web_search_result() { metadata: None, parallel_tool_calls: None, cache_salt: None, + context_management: None, }; let result = ExecuteRequest::new(payload, Arc::clone(&exec_ctx)).run().await.unwrap(); @@ -938,6 +940,7 @@ async fn execute_returns_mixed_client_tool_calls_without_followup_model_request( metadata: None, parallel_tool_calls: None, cache_salt: None, + context_management: None, }; let result = ExecuteRequest::new(payload, Arc::clone(&exec_ctx)).run().await.unwrap(); @@ -986,6 +989,7 @@ async fn execute_returns_mixed_client_tool_calls_without_followup_model_request( metadata: None, parallel_tool_calls: None, cache_salt: None, + context_management: None, }; let continuation = ExecuteRequest::new(continuation_payload, exec_ctx).run().await.unwrap(); assert!(matches!(continuation, Either::Left(_))); @@ -1057,6 +1061,7 @@ async fn execute_accumulates_usage_across_web_search_model_rounds() { metadata: None, parallel_tool_calls: None, cache_salt: None, + context_management: None, }; let result = ExecuteRequest::new(payload, exec_ctx).run().await.unwrap(); @@ -1101,6 +1106,7 @@ async fn stream_emits_web_search_lifecycle_events_before_final_payload() { metadata: None, parallel_tool_calls: None, cache_salt: None, + context_management: None, }; let result = ExecuteRequest::new(payload, Arc::clone(&exec_ctx)).run().await.unwrap(); @@ -1229,6 +1235,7 @@ async fn multi_round_stream_has_single_lifecycle_and_monotonic_public_sequence() cache_salt: None, metadata: None, parallel_tool_calls: None, + context_management: None, }; let result = ExecuteRequest::new(payload, Arc::clone(&exec_ctx)).run().await.unwrap(); @@ -1309,6 +1316,7 @@ async fn stream_hides_web_search_function_events_when_name_arrives_on_done() { metadata: None, parallel_tool_calls: None, cache_salt: None, + context_management: None, }; let result = ExecuteRequest::new(payload, Arc::clone(&exec_ctx)).run().await.unwrap(); @@ -1381,6 +1389,7 @@ async fn stream_orders_gateway_lifecycle_before_later_client_function_events() { metadata: None, parallel_tool_calls: None, cache_salt: None, + context_management: None, }; let result = ExecuteRequest::new(payload, exec_ctx).run().await.unwrap(); @@ -1469,6 +1478,7 @@ async fn execute_runs_multiple_web_search_calls_concurrently() { metadata: None, parallel_tool_calls: None, cache_salt: None, + context_management: None, }; let result = tokio::time::timeout(Duration::from_secs(2), ExecuteRequest::new(payload, exec_ctx).run()) @@ -1518,6 +1528,7 @@ async fn execute_feeds_web_search_execution_errors_back_to_model() { metadata: None, parallel_tool_calls: None, cache_salt: None, + context_management: None, }; let result = ExecuteRequest::new(payload, exec_ctx).run().await.unwrap(); @@ -1569,6 +1580,7 @@ async fn execute_returns_incomplete_after_max_gateway_tool_rounds() { metadata: None, parallel_tool_calls: None, cache_salt: None, + context_management: None, }; // Budget exhausted while the model keeps requesting tools → the response is @@ -1620,6 +1632,7 @@ async fn execute_feeds_invalid_web_search_arguments_back_to_model() { metadata: None, parallel_tool_calls: None, cache_salt: None, + context_management: None, }; let result = ExecuteRequest::new(payload, exec_ctx).run().await.unwrap(); @@ -1678,6 +1691,7 @@ async fn execute_runs_large_gateway_fanout_without_hard_cap() { metadata: None, parallel_tool_calls: None, cache_salt: None, + context_management: None, }; let result = ExecuteRequest::new(payload, exec_ctx) @@ -1742,6 +1756,7 @@ async fn stream_error_events_escape_error_messages() { metadata: None, parallel_tool_calls: None, cache_salt: None, + context_management: None, }; let result = ExecuteRequest::new(payload, exec_ctx).run().await.unwrap(); @@ -1818,6 +1833,7 @@ async fn incomplete_turn_persists_a_consistent_conversation_for_continuation() { metadata: None, parallel_tool_calls: None, cache_salt: None, + context_management: None, }; let result = ExecuteRequest::new(payload, Arc::clone(&exec_ctx)).run().await.unwrap(); @@ -1845,6 +1861,7 @@ async fn incomplete_turn_persists_a_consistent_conversation_for_continuation() { metadata: None, parallel_tool_calls: None, cache_salt: None, + context_management: None, }; let _ = ExecuteRequest::new(continuation_payload, exec_ctx).run().await.unwrap(); @@ -1916,6 +1933,7 @@ async fn stream_returns_incomplete_after_max_gateway_tool_rounds() { metadata: None, parallel_tool_calls: None, cache_salt: None, + context_management: None, }; let result = ExecuteRequest::new(payload, exec_ctx).run().await.unwrap(); diff --git a/crates/agentic-server/src/app.rs b/crates/agentic-server/src/app.rs index c2771c57..1c62a10f 100644 --- a/crates/agentic-server/src/app.rs +++ b/crates/agentic-server/src/app.rs @@ -11,7 +11,9 @@ use tower_http::cors::{AllowOrigin, Any, CorsLayer}; use agentic_core::executor::ExecutionContext; use agentic_core::proxy::ProxyState; -use crate::handler::{conversations, count_tokens, health, messages, models, ready, responses, responses_ws}; +use crate::handler::{ + compact_response, conversations, count_tokens, health, messages, models, ready, responses, responses_ws, +}; #[derive(Clone, Default)] pub struct WebSocketTracker { @@ -127,6 +129,7 @@ pub fn build_router(state: AppState, server_config: &ServerConfig) -> Router { .route("/v1/messages", post(messages)) .route("/v1/messages/count_tokens", post(count_tokens)) .route("/v1/responses", post(responses).get(responses_ws)) + .route("/v1/responses/compact", post(compact_response)) .layer(server_config.cors_layer()) .with_state(state) } diff --git a/crates/agentic-server/src/handler/common.rs b/crates/agentic-server/src/handler/common.rs index 4558fa46..7d83efd8 100644 --- a/crates/agentic-server/src/handler/common.rs +++ b/crates/agentic-server/src/handler/common.rs @@ -4,6 +4,7 @@ use axum::response::Response; use bytes::Bytes; use futures::StreamExt; use http::StatusCode; +use serde::de::DeserializeOwned; use tracing::warn; use agentic_core::executor::{BoxStream, ExecutorError}; @@ -61,6 +62,11 @@ pub(super) async fn read_and_parse(body: Body) -> Result<(Bytes, RequestPayload) Ok((bytes, payload)) } +pub(super) async fn read_json(body: Body) -> Result { + let bytes = read_bytes(body).await?; + serde_json::from_slice::(&bytes).map_err(|error| executor_error_response(ExecutorError::from(error))) +} + pub(super) fn extract_store(bytes: &[u8]) -> bool { serde_json::from_slice::(bytes) .ok() diff --git a/crates/agentic-server/src/handler/http/mod.rs b/crates/agentic-server/src/handler/http/mod.rs index 2ba66ea4..e92526d7 100644 --- a/crates/agentic-server/src/handler/http/mod.rs +++ b/crates/agentic-server/src/handler/http/mod.rs @@ -6,4 +6,4 @@ mod responses; pub use conversations::conversations; pub use messages::{count_tokens, messages}; pub use models::{health, models, ready}; -pub use responses::responses; +pub use responses::{compact_response, responses}; diff --git a/crates/agentic-server/src/handler/http/responses.rs b/crates/agentic-server/src/handler/http/responses.rs index 4e129cf1..02d3d648 100644 --- a/crates/agentic-server/src/handler/http/responses.rs +++ b/crates/agentic-server/src/handler/http/responses.rs @@ -7,12 +7,14 @@ use tracing::debug; use std::sync::Arc; -use agentic_core::executor::ExecuteRequest; +use agentic_core::executor::{ExecuteRequest, compact_response as execute_compaction}; use agentic_core::proxy::{ProxyRequest, proxy_request}; -use agentic_core::types::request_response::RequestPayload; +use agentic_core::types::request_response::{CompactRequest, RequestPayload}; use agentic_core::types::tools::ResponsesTool; -use super::super::common::{convert_response, executor_error_response, extract_bearer, read_and_parse, sse_response}; +use super::super::common::{ + convert_response, executor_error_response, extract_bearer, read_and_parse, read_json, sse_response, +}; use crate::app::AppState; async fn proxy_responses(state: &AppState, parts: Parts, body: Bytes) -> Response { @@ -54,6 +56,11 @@ pub async fn responses(State(state): State, req: Request) -> Response let should_execute = payload.store || payload.previous_response_id.is_some() || payload.conversation_id.is_some() + || payload.input.contains_compaction() + || payload + .context_management + .as_ref() + .is_some_and(|entries| !entries.is_empty()) || has_gateway_tools(&payload); debug!( route = if should_execute { "executor" } else { "proxy" }, @@ -61,6 +68,8 @@ pub async fn responses(State(state): State, req: Request) -> Response stream = payload.stream, has_previous_response_id = payload.previous_response_id.is_some(), has_conversation_id = payload.conversation_id.is_some(), + has_compaction = payload.input.contains_compaction(), + context_management = payload.context_management.as_ref().map_or(0, Vec::len), tools = payload.tools.as_ref().map_or(0, Vec::len), "routing HTTP responses request" ); @@ -71,3 +80,16 @@ pub async fn responses(State(state): State, req: Request) -> Response proxy_responses(&state, parts, bytes).await } } + +pub async fn compact_response(State(state): State, req: Request) -> Response { + let (parts, body) = req.into_parts(); + let request: CompactRequest = match read_json(body).await { + Ok(request) => request, + Err(response) => return response, + }; + let auth = extract_bearer(&parts.headers, state.openai_api_key.as_deref()); + match execute_compaction(request, state.exec_ctx.as_ref(), auth.as_deref()).await { + Ok(response) => axum::Json(response).into_response(), + Err(error) => executor_error_response(error), + } +} diff --git a/crates/agentic-server/src/handler/mod.rs b/crates/agentic-server/src/handler/mod.rs index 157a6366..2cf0dc43 100644 --- a/crates/agentic-server/src/handler/mod.rs +++ b/crates/agentic-server/src/handler/mod.rs @@ -3,5 +3,5 @@ pub mod http; pub mod websocket; pub use common::{convert_response, executor_error_response}; -pub use http::{conversations, count_tokens, health, messages, models, ready, responses}; +pub use http::{compact_response, conversations, count_tokens, health, messages, models, ready, responses}; pub use websocket::responses_ws; diff --git a/crates/agentic-server/tests/compaction_test.rs b/crates/agentic-server/tests/compaction_test.rs new file mode 100644 index 00000000..de858df3 --- /dev/null +++ b/crates/agentic-server/tests/compaction_test.rs @@ -0,0 +1,432 @@ +#[allow(dead_code)] +mod common; + +use std::collections::VecDeque; +use std::future::Future; +use std::pin::Pin; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; + +use axum::Router; +use axum::body::Bytes; +use axum::routing::post; +use tokio::net::TcpListener; +use tokio::sync::Mutex; + +use agentic_core::tool::{GatewayExecutor, ToolError, ToolHandler, ToolOutput, ToolType}; +use agentic_core::types::io::FunctionTool; + +use common::{spawn_gateway, test_config, test_state}; + +#[derive(Clone)] +struct TestWebSearchExecutor { + calls: Arc, +} + +impl ToolHandler for TestWebSearchExecutor { + fn tool_type(&self) -> ToolType { + ToolType::WebSearch + } + + fn validate(&self, _param: &serde_json::Value) -> Result<(), ToolError> { + Ok(()) + } + + fn normalize(&self, _param: &serde_json::Value) -> Vec { + vec![FunctionTool { + type_: "function".to_owned(), + name: "web_search".to_owned(), + description: Some("Search the web.".to_owned()), + parameters: Some(serde_json::json!({ + "type": "object", + "properties": {"query": {"type": "string"}}, + "required": ["query"] + })), + strict: Some(false), + }] + } +} + +impl GatewayExecutor for TestWebSearchExecutor { + fn execute( + &self, + call_id: &str, + _tool_name: &str, + _arguments: &str, + _config: &serde_json::Value, + ) -> Pin> + Send + '_>> { + self.calls.fetch_add(1, Ordering::Relaxed); + let call_id = call_id.to_owned(); + Box::pin(async move { + Ok(ToolOutput { + call_id, + output: serde_json::json!({ + "query": "rust compaction", + "results": {"web": [], "news": []}, + "metadata": {} + }) + .to_string(), + }) + }) + } +} + +async fn spawn_compaction_model() -> (String, Arc>>, tokio::task::JoinHandle<()>) { + let requests = Arc::new(Mutex::new(Vec::new())); + let captured = Arc::clone(&requests); + let app = Router::new().route( + "/v1/responses", + post(move |body: Bytes| { + let captured = Arc::clone(&captured); + async move { + captured + .lock() + .await + .push(serde_json::from_slice(&body).expect("model request is JSON")); + axum::Json(serde_json::json!({ + "id": "resp_upstream", + "object": "response", + "created_at": 0, + "model": "test-model", + "status": "completed", + "output": [{ + "id": "msg_upstream", + "type": "message", + "role": "assistant", + "status": "completed", + "content": [{ + "type": "output_text", + "text": "preserved checkpoint summary", + "annotations": [] + }] + }], + "usage": { + "input_tokens": 20, + "output_tokens": 5, + "total_tokens": 25 + } + })) + } + }), + ); + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind model"); + let address = listener.local_addr().expect("model address"); + let handle = tokio::spawn(async move { + axum::serve(listener, app).await.expect("serve model"); + }); + (format!("http://{address}"), requests, handle) +} + +async fn spawn_sequential_model( + responses: Vec<(&str, i64)>, +) -> (String, Arc>>, tokio::task::JoinHandle<()>) { + let requests = Arc::new(Mutex::new(Vec::new())); + let captured = Arc::clone(&requests); + let responses = Arc::new(Mutex::new(VecDeque::from( + responses + .into_iter() + .map(|(text, tokens)| (text.to_owned(), tokens)) + .collect::>(), + ))); + let app = Router::new().route( + "/v1/responses", + post(move |body: Bytes| { + let captured = Arc::clone(&captured); + let responses = Arc::clone(&responses); + async move { + captured + .lock() + .await + .push(serde_json::from_slice(&body).expect("model request is JSON")); + let (text, tokens) = responses.lock().await.pop_front().expect("model response available"); + axum::Json(serde_json::json!({ + "id": "resp_upstream", + "object": "response", + "created_at": 0, + "model": "test-model", + "status": "completed", + "output": [{ + "id": "msg_upstream", + "type": "message", + "role": "assistant", + "status": "completed", + "content": [{"type": "output_text", "text": text, "annotations": []}] + }], + "usage": { + "input_tokens": tokens - 1, + "output_tokens": 1, + "total_tokens": tokens + } + })) + } + }), + ); + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind model"); + let address = listener.local_addr().expect("model address"); + let handle = tokio::spawn(async move { + axum::serve(listener, app).await.expect("serve model"); + }); + (format!("http://{address}"), requests, handle) +} + +async fn spawn_compaction_tool_loop_model() -> (String, Arc>>, tokio::task::JoinHandle<()>) +{ + let requests = Arc::new(Mutex::new(Vec::new())); + let captured = Arc::clone(&requests); + let responses = Arc::new(Mutex::new(VecDeque::from([ + serde_json::json!({ + "id": "resp_summary", + "object": "response", + "created_at": 0, + "model": "test-model", + "status": "completed", + "output": [{ + "id": "msg_summary", + "type": "message", + "role": "assistant", + "status": "completed", + "content": [{"type": "output_text", "text": "automatic summary", "annotations": []}] + }], + "usage": {"input_tokens": 20, "output_tokens": 5, "total_tokens": 25} + }), + serde_json::json!({ + "id": "resp_tool", + "object": "response", + "created_at": 0, + "model": "test-model", + "status": "completed", + "output": [{ + "id": "fc_search", + "type": "function_call", + "call_id": "call_search", + "name": "web_search", + "arguments": "{\"query\":\"rust compaction\"}", + "status": "completed" + }], + "usage": {"input_tokens": 8, "output_tokens": 2, "total_tokens": 10} + }), + serde_json::json!({ + "id": "resp_answer", + "object": "response", + "created_at": 0, + "model": "test-model", + "status": "completed", + "output": [{ + "id": "msg_answer", + "type": "message", + "role": "assistant", + "status": "completed", + "content": [{"type": "output_text", "text": "final answer", "annotations": []}] + }], + "usage": {"input_tokens": 4, "output_tokens": 1, "total_tokens": 5} + }), + ]))); + let app = Router::new().route( + "/v1/responses", + post(move |body: Bytes| { + let captured = Arc::clone(&captured); + let responses = Arc::clone(&responses); + async move { + captured + .lock() + .await + .push(serde_json::from_slice(&body).expect("model request is JSON")); + axum::Json(responses.lock().await.pop_front().expect("model response available")) + } + }), + ); + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind model"); + let address = listener.local_addr().expect("model address"); + let handle = tokio::spawn(async move { + axum::serve(listener, app).await.expect("serve model"); + }); + (format!("http://{address}"), requests, handle) +} + +#[tokio::test] +async fn compact_endpoint_returns_reusable_canonical_window() { + let (model_url, model_requests, _model) = spawn_compaction_model().await; + let (gateway_url, _gateway) = spawn_gateway(test_state(&test_config(&model_url))).await; + + let response = reqwest::Client::new() + .post(format!("{gateway_url}/v1/responses/compact")) + .json(&serde_json::json!({ + "model": "test-model", + "input": [ + {"role": "user", "content": "retain this request"}, + {"type": "function_call_output", "call_id": "call_1", "output": "large result"} + ], + "tools": [], + "parallel_tool_calls": true, + "reasoning": {"effort": "medium"}, + "text": {"verbosity": "low"} + })) + .send() + .await + .expect("compact request"); + + assert_eq!(response.status(), reqwest::StatusCode::OK); + let body: serde_json::Value = response.json().await.expect("compact response JSON"); + assert_eq!(body["object"], "response.compaction"); + assert_eq!(body["usage"]["total_tokens"], 25); + assert_eq!(body["output"][0]["type"], "message"); + assert_eq!(body["output"][0]["status"], "completed"); + assert_eq!(body["output"][1]["type"], "compaction"); + assert_eq!(body["output"][1]["encrypted_content"], "preserved checkpoint summary"); + + let reused = reqwest::Client::new() + .post(format!("{gateway_url}/v1/responses")) + .json(&serde_json::json!({ + "model": "test-model", + "input": body["output"].clone(), + "store": false, + "stream": false + })) + .send() + .await + .expect("reuse compacted output"); + assert_eq!(reused.status(), reqwest::StatusCode::OK); + let reused_body: serde_json::Value = reused.json().await.expect("reuse response JSON"); + assert!(reused_body["id"].as_str().is_some_and(|id| id.starts_with("resp_"))); + + let requests = model_requests.lock().await; + assert_eq!(requests.len(), 2); + assert_eq!(requests[0]["stream"], false); + let final_input = requests[0]["input"] + .as_array() + .expect("model input array") + .last() + .unwrap(); + assert!( + final_input["content"] + .as_str() + .expect("summary prompt") + .contains("CONTEXT CHECKPOINT COMPACTION") + ); + assert_eq!(requests[1]["input"].as_array().map(Vec::len), Some(2)); + assert_eq!(requests[1]["input"][0]["role"], "user"); + assert_eq!(requests[1]["input"][1]["role"], "assistant"); + assert_eq!(requests[1]["input"][1]["content"][0]["type"], "output_text"); + assert_eq!( + requests[1]["input"][1]["content"][0]["text"], + "preserved checkpoint summary" + ); +} + +#[tokio::test] +async fn automatic_compaction_runs_above_threshold_and_accumulates_usage() { + let (model_url, requests, _model) = + spawn_sequential_model(vec![("automatic summary", 25), ("final answer", 10)]).await; + let (gateway_url, _gateway) = spawn_gateway(test_state(&test_config(&model_url))).await; + + let response = reqwest::Client::new() + .post(format!("{gateway_url}/v1/responses")) + .json(&serde_json::json!({ + "model": "test-model", + "input": [{"role": "user", "content": "x".repeat(200)}], + "store": false, + "stream": false, + "context_management": [{"type": "compaction", "compact_threshold": 10}] + })) + .send() + .await + .expect("automatic compaction request"); + + assert_eq!(response.status(), reqwest::StatusCode::OK); + let body: serde_json::Value = response.json().await.expect("automatic response JSON"); + assert_eq!(body["usage"]["total_tokens"], 35); + let requests = requests.lock().await; + assert_eq!(requests.len(), 2); + assert!( + requests + .iter() + .all(|request| request.get("context_management").is_none()) + ); + assert_eq!(requests[1]["input"].as_array().map(Vec::len), Some(2)); + assert_eq!(requests[1]["input"][0]["role"], "user"); + assert_eq!(requests[1]["input"][1]["role"], "assistant"); + assert_eq!(requests[1]["input"][1]["content"][0]["text"], "automatic summary"); +} + +#[tokio::test] +async fn automatic_compaction_accumulates_usage_across_gateway_tool_loop() { + let (model_url, requests, _model) = spawn_compaction_tool_loop_model().await; + let tool_calls = Arc::new(AtomicUsize::new(0)); + let mut state = test_state(&test_config(&model_url)); + state.exec_ctx = Arc::new( + state + .exec_ctx + .as_ref() + .clone() + .with_gateway_executor(Arc::new(TestWebSearchExecutor { + calls: Arc::clone(&tool_calls), + })), + ); + let (gateway_url, _gateway) = spawn_gateway(state).await; + + let response = reqwest::Client::new() + .post(format!("{gateway_url}/v1/responses")) + .json(&serde_json::json!({ + "model": "test-model", + "input": [{"role": "user", "content": "x".repeat(200)}], + "tools": [{"type": "web_search_preview"}], + "store": false, + "stream": false, + "context_management": [{"type": "compaction", "compact_threshold": 10}] + })) + .send() + .await + .expect("automatic compaction tool-loop request"); + + assert_eq!(response.status(), reqwest::StatusCode::OK); + let body: serde_json::Value = response.json().await.expect("automatic response JSON"); + assert_eq!(body["usage"]["total_tokens"], 40); + assert_eq!(tool_calls.load(Ordering::Relaxed), 1); + + let requests = requests.lock().await; + assert_eq!(requests.len(), 3); + assert_eq!(requests[2]["input"].as_array().map(Vec::len), Some(4)); + assert_eq!(requests[2]["input"][1]["role"], "assistant"); + assert_eq!(requests[2]["input"][2]["type"], "function_call"); + assert_eq!(requests[2]["input"][3]["type"], "function_call_output"); +} + +#[tokio::test] +async fn automatic_compaction_skips_below_threshold() { + let (model_url, requests, _model) = spawn_sequential_model(vec![("final answer", 10)]).await; + let (gateway_url, _gateway) = spawn_gateway(test_state(&test_config(&model_url))).await; + + let response = reqwest::Client::new() + .post(format!("{gateway_url}/v1/responses")) + .json(&serde_json::json!({ + "model": "test-model", + "input": [{"role": "user", "content": "short"}], + "store": false, + "stream": false, + "context_management": [{"type": "compaction", "compact_threshold": 100_000}] + })) + .send() + .await + .expect("below-threshold request"); + + assert_eq!(response.status(), reqwest::StatusCode::OK); + let requests = requests.lock().await; + assert_eq!(requests.len(), 1); + assert_eq!(requests[0]["input"][0]["content"], "short"); +} + +#[tokio::test] +async fn compact_endpoint_rejects_missing_context() { + let (model_url, requests, _model) = spawn_sequential_model(Vec::new()).await; + let (gateway_url, _gateway) = spawn_gateway(test_state(&test_config(&model_url))).await; + + let response = reqwest::Client::new() + .post(format!("{gateway_url}/v1/responses/compact")) + .json(&serde_json::json!({"model": "test-model"})) + .send() + .await + .expect("invalid compact request"); + + assert_eq!(response.status(), reqwest::StatusCode::BAD_REQUEST); + assert!(requests.lock().await.is_empty()); +} diff --git a/docs/api/index.md b/docs/api/index.md index f1b8d370..7220372c 100644 --- a/docs/api/index.md +++ b/docs/api/index.md @@ -5,9 +5,16 @@ ### `POST /v1/responses` HTTP Responses requests use the OpenAI-compatible Responses shape. Requests -with `store=true`, `previous_response_id`, or `conversation_id` run through the -stateful executor. Stateless `store=false` requests without continuation state -are proxied directly to the configured vLLM backend. +with `store=true`, `previous_response_id`, `conversation_id`, compaction input, +or `context_management` run through the executor. Other stateless `store=false` +requests are passed directly to the configured vLLM backend. + +### `POST /v1/responses/compact` + +Compacts direct input or a stored previous-response chain into a canonical +window of retained user messages plus one `compaction` item. See +[Responses compaction](../guides/responses-compaction.md) for request examples, +automatic threshold management, and the local plaintext limitation. ### `WS /v1/responses` diff --git a/docs/design/core-public-api.md b/docs/design/core-public-api.md index c5a11f4d..5eebc522 100644 --- a/docs/design/core-public-api.md +++ b/docs/design/core-public-api.md @@ -33,6 +33,19 @@ The base loop handles text messages. This design extends it with: 3. **Streaming tee** — forward SSE to client in real-time while accumulating for tool detection 4. **Extended SSE events** — function_call, reasoning, file_search, web_search event types 5. **Tool executor traits** — MCP, web_search, vector_store as pluggable implementations +6. **Context compaction** — standalone and threshold-driven canonical item-history replacement + +### Responses compaction extension + +The executor exposes `compact_response()` alongside the normal `ExecuteRequest` path. Both standalone +`POST /v1/responses/compact` requests and automatic `context_management` use the same inference-backed summary +service. Public compaction items remain protocol types in stored or replayed item history; the upstream adapter renders +the latest canonical window as retained user messages plus an assistant summary because vLLM does not accept the +public `compaction` item type. + +Automatic compaction runs after rehydration so its threshold applies to resolved history, not only the newest input. +Its summary-call usage is added to the final answer usage. The `context_management` configuration is executor-only and +is not forwarded to the upstream inference server. --- @@ -246,7 +259,7 @@ How the complete pipeline maps to @leseb's proposed filter chain: | 8 | `mcp_tool` | `McpToolExecutor::execute()` | Phase 4 | @ashwing | | 9 | `web_search` | `WebSearchProvider::search()` | Phase 4 | @franciscojavierarceo | | 10 | `file_search` | `VectorStoreClient::search()` | Phase 4 | @franciscojavierarceo | -| 11 | `compact` | `compact_context()` | Future | — | +| 11 | `compact` | `compact_response()` / automatic context management | Implemented | — | | 12 | `reasoning` | `summarize_reasoning()` | Future | — | | 13 | `response_store` (resp) | `persist_response()` | PR #46 | @maralbahari | diff --git a/docs/guides/responses-compaction.md b/docs/guides/responses-compaction.md new file mode 100644 index 00000000..2342e896 --- /dev/null +++ b/docs/guides/responses-compaction.md @@ -0,0 +1,104 @@ +# Responses compaction + +Compaction turns a long Responses item history into a smaller, reusable context checkpoint. vLLM Agentic API supports +both the standalone `POST /v1/responses/compact` endpoint and automatic compaction on `POST /v1/responses`. + +## Standalone compaction + +Send direct input, a stored previous response ID, or both: + +```bash +curl http://localhost:8000/v1/responses/compact \ + -H 'Content-Type: application/json' \ + -d '{ + "model": "served-model", + "input": [ + {"role": "user", "content": "Remember the deployment constraints."}, + {"role": "assistant", "content": "I will preserve them."} + ] + }' +``` + +The endpoint makes a non-streaming inference call to summarize the resolved item history. Its response has +`object: "response.compaction"`; `usage` describes that summary call. The `output` array is the canonical compacted +window: normalized user messages followed by exactly one `compaction` item. + +```json +{ + "id": "resp_...", + "object": "response.compaction", + "created_at": 1784740000, + "output": [ + { + "type": "message", + "id": "msg_...", + "role": "user", + "status": "completed", + "content": "Remember the deployment constraints." + }, + { + "type": "compaction", + "id": "cmp_...", + "encrypted_content": "The user asked that the deployment constraints be preserved..." + } + ], + "usage": { + "input_tokens": 1200, + "output_tokens": 180, + "total_tokens": 1380, + "input_tokens_details": {"cached_tokens": 0}, + "output_tokens_details": {"reasoning_tokens": 0} + } +} +``` + +Pass the complete `output` array back as later Responses input. You may append new input items after it: + +```json +{ + "model": "served-model", + "input": [ + {"type": "message", "id": "msg_...", "role": "user", "status": "completed", "content": "..."}, + {"type": "compaction", "id": "cmp_...", "encrypted_content": "..."}, + {"role": "user", "content": "Continue the task."} + ], + "store": false +} +``` + +When response storage is configured, the returned response ID is also a continuation checkpoint and can be supplied +as `previous_response_id`. Compatibility fields such as `tools`, `parallel_tool_calls`, `reasoning`, and `text` are +accepted by the compact endpoint but are not used by the summary call. + +## Automatic context management + +Add a compaction policy to an ordinary Responses request: + +```json +{ + "model": "served-model", + "input": [{"role": "user", "content": "Long-running task context"}], + "context_management": [ + {"type": "compaction", "compact_threshold": 120000} + ] +} +``` + +The gateway first rehydrates any conversation or previous-response history, estimates the model-facing input size, +and compacts when that estimate is greater than `compact_threshold`. It then answers using the compacted window. +`context_management` is local executor configuration and is not sent to vLLM. If compaction runs, the response usage +is the saturating sum of the summary call and all answer/tool-loop inference rounds. + +The current estimate is deterministic JSON size divided by four, rounded up. It is not a model-specific tokenizer, so +choose a threshold with headroom. An entry without `compact_threshold` is ignored because this server has no global +default threshold. + +## Local plaintext limitation + +The wire field is named `encrypted_content` for Responses compatibility, but summaries generated by this local +implementation are **plaintext and are not encrypted**. Treat the field as conversation content when logging, +storing, transmitting, or applying data-retention controls. The gateway converts it to assistant context only for the +model-facing vLLM request; the public and stored item remains a `compaction` item. + +OpenAI-generated encrypted compaction items are not decryptable or interoperable with this implementation. If one is +submitted, its `encrypted_content` bytes are treated literally as plaintext assistant context. diff --git a/mkdocs.yaml b/mkdocs.yaml index f50ee255..1861ae86 100644 --- a/mkdocs.yaml +++ b/mkdocs.yaml @@ -92,6 +92,8 @@ markdown_extensions: nav: - Home: index.md - API Reference: api/index.md + - Guides: + - Responses Compaction: guides/responses-compaction.md - Developing: - Getting Started: developing/getting-started.md - Deploying: