diff --git a/Cargo.lock b/Cargo.lock index 801299b3..1592e765 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -145,6 +145,12 @@ dependencies = [ "uuid", ] +[[package]] +name = "anyhow" +version = "1.0.103" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" + [[package]] name = "apl-cmf" version = "0.2.1" @@ -1457,6 +1463,17 @@ dependencies = [ "smallvec", ] +[[package]] +name = "fancy-regex" +version = "0.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72cf461f865c862bb7dc573f643dd6a2b6842f7c30b07882b56bd148cc2761b8" +dependencies = [ + "bit-set", + "regex-automata", + "regex-syntax", +] + [[package]] name = "fastrand" version = "2.4.1" @@ -2931,6 +2948,7 @@ dependencies = [ "sqlx", "tempfile", "thiserror 2.0.18", + "tiktoken-rs", "tokio", "tokio-util", "tracing", @@ -4824,6 +4842,21 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "tiktoken-rs" +version = "0.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "027853bbf8c7763b77c5c595f1c271c7d536ced7d6f83452911b944621e57fc2" +dependencies = [ + "anyhow", + "base64", + "bstr", + "fancy-regex", + "lazy_static", + "regex", + "rustc-hash", +] + [[package]] name = "tikv-jemalloc-sys" version = "0.7.1+5.3.1-0-g81034ce1f1373e37dc865038e1bc8eeecf559ce8" diff --git a/Cargo.toml b/Cargo.toml index 3c367b5a..c2a2f74c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -49,6 +49,7 @@ serde_yaml = { package = "yaml_serde", version = "0.10.4" } sqlx = { version = "0.9.0", default-features = false, features = ["runtime-tokio", "tls-rustls-ring", "sqlite", "postgres"] } tempfile = "3.27.0" thiserror = "2.0.18" +tiktoken-rs = "0.12.0" tikv-jemallocator = "0.7.0" tokio = { version = "1.52.3", features = ["macros", "rt"] } tokio-rustls = "0.26.4" diff --git a/apis/Cargo.toml b/apis/Cargo.toml index 44661925..f3ab0117 100644 --- a/apis/Cargo.toml +++ b/apis/Cargo.toml @@ -37,6 +37,7 @@ serde_json = { workspace = true } serde_yaml = { workspace = true } sqlx = { workspace = true, optional = true } thiserror = { workspace = true } +tiktoken-rs = { workspace = true } tokio = { workspace = true, features = ["rt", "time", "net"] } tracing = { workspace = true } zeroize = { workspace = true } diff --git a/apis/src/openai/mod.rs b/apis/src/openai/mod.rs index 1879ea43..d1f46b66 100644 --- a/apis/src/openai/mod.rs +++ b/apis/src/openai/mod.rs @@ -15,7 +15,7 @@ pub(crate) mod translation; pub use conversations::OpenaiConversationsFilter; pub use responses::{ - DocExtractFilter, FileResolveFilter, McpDispatchFilter, McpToolResolveFilter, ModelRewriteFilter, + CompactFilter, DocExtractFilter, FileResolveFilter, McpDispatchFilter, McpToolResolveFilter, ModelRewriteFilter, OpenaiResponsesValidateFilter, RehydrateFilter, ResponseStoreFilter, ResponsesFormatFilter, ToolParseFilter, WebSearchFilter, openai_responses_proxy::ResponsesProxyFilter, stream_events::OpenaiStreamEventsFilter, }; diff --git a/apis/src/openai/responses/compact/config.rs b/apis/src/openai/responses/compact/config.rs new file mode 100644 index 00000000..81eb689e --- /dev/null +++ b/apis/src/openai/responses/compact/config.rs @@ -0,0 +1,111 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Praxis Contributors + +//! Configuration for the `openai_responses_compact` filter. + +use praxis_filter::FilterError; +use serde::Deserialize; + +use crate::openai::responses::config_validation::{self, CalloutSettings, FailureMode}; + +/// Default callout timeout (30 seconds — summarization can be slow). +const DEFAULT_TIMEOUT_MS: u64 = 30_000; + +/// Default HTTP status when the summarization callout fails in closed mode. +const DEFAULT_STATUS_ON_ERROR: u16 = 502; + +// ----------------------------------------------------------------------------- +// CompactFilterConfig (YAML deserialization) +// ----------------------------------------------------------------------------- + +/// Raw YAML config, deserialized then validated. +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +pub(super) struct CompactFilterConfig { + /// URL of the inference backend for summarization calls. + /// E.g., `"http://localhost:11434/v1/chat/completions"` + pub inference_url: String, + + /// Default model for summarization when not overridden + /// in the request's `context_management`. + #[serde(default = "default_model")] + pub default_model: String, + + /// Tiktoken encoding name for local token estimation. + /// Used as fallback when `previous_usage` is unavailable. + #[serde(default = "default_tiktoken_encoding")] + pub tiktoken_encoding: String, + + /// Callout timeout in milliseconds. + #[serde(default)] + pub timeout_ms: Option, + + /// Failure mode for the inference callout. + #[serde(default)] + pub failure_mode: Option, + + /// HTTP status code to return when rejecting on error. + #[serde(default)] + pub status_on_error: Option, +} + +/// Default summarization model when not overridden per-request. +fn default_model() -> String { + "gpt-4o-mini".to_owned() +} + +/// Default tiktoken encoding for local token estimation. +fn default_tiktoken_encoding() -> String { + "cl100k_base".to_owned() +} + +// ----------------------------------------------------------------------------- +// ValidatedConfig (post-validation) +// ----------------------------------------------------------------------------- + +/// Validated configuration with defaults applied. +pub(super) struct ValidatedConfig { + /// URL of the inference backend for summarization calls. + pub inference_url: String, + + /// Default model for summarization. + pub default_model: String, + + /// Tiktoken encoding name. + pub tiktoken_encoding: String, + + /// Shared callout settings (timeout, failure mode, status). + pub callout: CalloutSettings, +} + +/// Validate raw config and apply defaults. +/// +/// # Errors +/// +/// Returns [`FilterError`] if `inference_url` is empty, +/// `timeout_ms` is zero, or `status_on_error` is out of range. +pub(super) fn build_config(raw: &CompactFilterConfig) -> Result { + if raw.inference_url.is_empty() { + return Err(FilterError::from("openai_responses_compact: inference_url is empty")); + } + + let timeout_ms = + config_validation::validate_timeout_ms("openai_responses_compact", raw.timeout_ms, DEFAULT_TIMEOUT_MS)?; + + let status_on_error = config_validation::validate_status_on_error( + "openai_responses_compact", + raw.status_on_error, + DEFAULT_STATUS_ON_ERROR, + )?; + + Ok(ValidatedConfig { + inference_url: raw.inference_url.clone(), + default_model: raw.default_model.clone(), + tiktoken_encoding: raw.tiktoken_encoding.clone(), + callout: CalloutSettings { + timeout_ms, + failure_mode: raw.failure_mode.unwrap_or(FailureMode::Closed), + status_on_error, + }, + }) +} diff --git a/apis/src/openai/responses/compact/mod.rs b/apis/src/openai/responses/compact/mod.rs new file mode 100644 index 00000000..a3f0983d --- /dev/null +++ b/apis/src/openai/responses/compact/mod.rs @@ -0,0 +1,440 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Praxis Contributors + +//! Compact filter: token counting and context window management. +//! +//! When a request's `context_management` contains a compaction +//! configuration and the token count exceeds the specified threshold, +//! this filter summarizes the conversation history via a sub-request +//! to an inference backend, replacing it with a single compaction +//! item. Runs after `rehydrate` (which populates messages and +//! previous usage) and before `openai_tool_parse`. + +pub(super) mod config; + +#[cfg(test)] +#[expect(clippy::allow_attributes, reason = "blanket test suppressions")] +#[allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::indexing_slicing, + clippy::panic, + reason = "tests" +)] +mod tests; + +use std::borrow::Cow; + +use async_trait::async_trait; +use bytes::Bytes; +use praxis_core::callout::{CalloutClient, CalloutRequest, CalloutResult}; +use praxis_filter::{ + BodyAccess, BodyMode, FilterAction, FilterError, HttpFilter, HttpFilterContext, body::MAX_JSON_BODY_BYTES, + parse_filter_config, +}; +use serde_json::Value; +use tracing::{debug, warn}; + +use self::config::{CompactFilterConfig, ValidatedConfig, build_config}; +use super::{error::responses_error_rejection, state::ResponsesState}; + +// ----------------------------------------------------------------------------- +// Constants +// ----------------------------------------------------------------------------- + +/// System prompt for the summarization call. +const SUMMARIZATION_SYSTEM_PROMPT: &str = "\ +Summarize the following conversation concisely. \ +Preserve all key facts, decisions, code snippets, \ +user preferences, and important context. The summary \ +will replace the full conversation history, so it must \ +capture everything needed to continue coherently."; + +// ----------------------------------------------------------------------------- +// CompactionParams +// ----------------------------------------------------------------------------- + +/// Parsed compaction parameters from the request's `context_management`. +struct CompactionParams { + /// Token threshold above which compaction triggers. + compact_threshold: u64, + /// Optional model override for the summarization call. + compaction_model: Option, +} + +// ----------------------------------------------------------------------------- +// CompactFilter +// ----------------------------------------------------------------------------- + +/// Summarizes conversation history when the token count exceeds a +/// configured threshold. +/// +/// # YAML +/// +/// ```yaml +/// filter: openai_responses_compact +/// inference_url: "http://localhost:11434/v1/chat/completions" +/// default_model: llama3.2:1b +/// ``` +/// +/// # Full YAML +/// +/// ```yaml +/// filter: openai_responses_compact +/// inference_url: "http://localhost:11434/v1/chat/completions" +/// default_model: gpt-4o-mini +/// tiktoken_encoding: cl100k_base +/// timeout_ms: 30000 +/// failure_mode: closed +/// status_on_error: 502 +/// ``` +pub struct CompactFilter { + /// HTTP client for the summarization inference call. + callout_client: CalloutClient, + /// Validated filter configuration. + config: ValidatedConfig, +} + +impl CompactFilter { + /// Create a filter from parsed YAML config. + /// + /// Constructs the [`CalloutClient`] at startup so it can be + /// reused across requests. + /// + /// # Errors + /// + /// Returns [`FilterError`] if config validation fails or the + /// callout client cannot be constructed. + pub fn from_config(config: &serde_yaml::Value) -> Result, FilterError> { + let cfg: CompactFilterConfig = parse_filter_config("openai_responses_compact", config)?; + let validated = build_config(&cfg)?; + let callout_config = validated.callout.build_callout_config(); + let callout_client = CalloutClient::new(callout_config).map_err(|e| FilterError::from(e.to_string()))?; + Ok(Box::new(Self { + callout_client, + config: validated, + })) + } + + /// Run the summarization callout and return the summary text. + /// + /// Returns `Ok(Some(summary))` on success, `Ok(None)` when + /// compaction should be skipped, or `Err(FilterAction)` to + /// short-circuit the request. + async fn execute_compaction( + &self, + state: &ResponsesState, + params: &CompactionParams, + streaming: bool, + conversation_text: &str, + ) -> Result, FilterAction> { + let model = params.compaction_model.as_deref().unwrap_or(&self.config.default_model); + let instructions = state.request_body.get("instructions").and_then(Value::as_str); + let request = build_summarization_request(conversation_text, instructions, model, &self.config.inference_url); + + match self.callout_client.execute(request).await { + CalloutResult::Success(resp) => parse_summarization_response(&resp.body).map(Some).or_else(|e| { + warn!(error = %e, "failed to parse summarization response, skipping compaction"); + Ok(None) + }), + CalloutResult::Failed => { + warn!("summarization callout failed, skipping compaction"); + Ok(None) + }, + CalloutResult::Rejected(rejection) => { + warn!(status = rejection.status, "summarization callout rejected"); + Err(FilterAction::Reject(responses_error_rejection( + rejection.status, + "server_error", + "summarization callout rejected", + streaming, + ))) + }, + } + } +} + +#[async_trait] +impl HttpFilter for CompactFilter { + fn name(&self) -> &'static str { + "openai_responses_compact" + } + + fn request_body_access(&self) -> BodyAccess { + BodyAccess::ReadOnly + } + + fn request_body_mode(&self) -> BodyMode { + BodyMode::StreamBuffer { + max_bytes: Some(MAX_JSON_BODY_BYTES), + } + } + + async fn on_request(&self, _ctx: &mut HttpFilterContext<'_>) -> Result { + Ok(FilterAction::Continue) + } + + async fn on_request_body( + &self, + ctx: &mut HttpFilterContext<'_>, + _body: &mut Option, + end_of_stream: bool, + ) -> Result { + if !end_of_stream { + return Ok(FilterAction::Continue); + } + if !is_responses_request(ctx) { + return Ok(FilterAction::Release); + } + let streaming = is_streaming(ctx); + let Some(state) = ctx.extensions.get::() else { + return Ok(FilterAction::Release); + }; + let Some((params, conversation_text)) = should_compact(state, &self.config.tiktoken_encoding) else { + return Ok(FilterAction::Release); + }; + let compaction = self.execute_compaction(state, ¶ms, streaming, &conversation_text); + let summary = match compaction.await { + Ok(Some(s)) => s, + Ok(None) | Err(FilterAction::Release) => return Ok(FilterAction::Release), + Err(action) => return Ok(action), + }; + let Some(state) = ctx.extensions.get_mut::() else { + return Ok(FilterAction::Release); + }; + replace_messages(state, build_compaction_item(&summary)); + ctx.set_metadata("responses.compacted", "true"); + Ok(FilterAction::Release) + } +} + +// ----------------------------------------------------------------------------- +// Compaction Logic +// ----------------------------------------------------------------------------- + +/// Check whether compaction should run and return the params + text. +/// +/// Returns `None` if there is no compaction config, the encoding is +/// unknown, or the token count is below the threshold. +fn should_compact(state: &ResponsesState, tiktoken_encoding: &str) -> Option<(CompactionParams, String)> { + let params = extract_compaction_config(&state.context_management)?; + let conversation_text = build_conversation_text(&state.messages); + let token_count = get_token_count(&conversation_text, tiktoken_encoding)?; + if token_count <= params.compact_threshold { + debug!( + token_count, + threshold = params.compact_threshold, + "under threshold, skipping" + ); + return None; + } + debug!( + token_count, + threshold = params.compact_threshold, + "threshold exceeded, compacting" + ); + Some((params, conversation_text)) +} + +/// Check whether this is an OpenAI Responses API request. +fn is_responses_request(ctx: &HttpFilterContext<'_>) -> bool { + ctx.get_metadata("openai_responses_format.format") == Some("openai_responses") +} + +/// Check whether the client requested streaming. +fn is_streaming(ctx: &HttpFilterContext<'_>) -> bool { + ctx.get_metadata("openai_responses_format.stream") + .is_some_and(|v| v == "true") +} + +/// Parse the `context_management` JSON to find a compaction config. +/// +/// The `context_management` field is an array like: +/// `[{"type": "compaction", "compact_threshold": 50000}]` +/// +/// Returns `None` if no compaction entry is found. +fn extract_compaction_config(context_management: &Option) -> Option { + let array = context_management.as_ref()?.as_array()?; + + for entry in array { + let Some(entry_type) = entry.get("type").and_then(|v| v.as_str()) else { + continue; + }; + if entry_type != "compaction" { + continue; + } + let compact_threshold = entry.get("compact_threshold").and_then(Value::as_u64).unwrap_or(0); + let compaction_model = entry + .get("compaction_model") + .and_then(Value::as_str) + .map(ToOwned::to_owned); + return Some(CompactionParams { + compact_threshold, + compaction_model, + }); + } + None +} + +/// Estimate the token count for the given messages using tiktoken. +/// +/// Uses the configured encoding (e.g. `cl100k_base`, `o200k_base`) +/// to tokenize the serialized conversation text. +/// +/// Returns `None` if the encoding name is not recognized. +fn get_token_count(conversation_text: &str, tiktoken_encoding: &str) -> Option { + let bpe = match tiktoken_encoding { + "cl100k_base" => tiktoken_rs::cl100k_base_singleton(), + "o200k_base" => tiktoken_rs::o200k_base_singleton(), + other => { + warn!(encoding = other, "unknown tiktoken encoding, cannot estimate tokens"); + return None; + }, + }; + let count = bpe.count_ordinary(conversation_text) as u64; + debug!( + count, + source = "tiktoken", + encoding = tiktoken_encoding, + "token count estimated" + ); + Some(count) +} + +/// Build a Chat Completions request for summarization. +/// +/// The request body has this shape: +/// ```json +/// { +/// "model": "", +/// "messages": [ +/// {"role": "system", "content": ""}, +/// {"role": "user", "content": ""} +/// ] +/// } +/// ``` +fn build_summarization_request( + conversation_text: &str, + instructions: Option<&str>, + model: &str, + inference_url: &str, +) -> CalloutRequest { + let system_content = match instructions { + Some(inst) => format!("{inst}\n\n{SUMMARIZATION_SYSTEM_PROMPT}"), + None => SUMMARIZATION_SYSTEM_PROMPT.to_owned(), + }; + + let body = serde_json::json!({ + "model": model, + "messages": [ + {"role": "system", "content": system_content}, + {"role": "user", "content": conversation_text} + ] + }); + + let body_bytes = serde_json::to_vec(&body).unwrap_or_default(); + + CalloutRequest { + method: http::Method::POST, + url: inference_url.to_owned(), + headers: vec![ + ( + http::header::CONTENT_TYPE, + http::HeaderValue::from_static("application/json"), + ), + (http::header::ACCEPT, http::HeaderValue::from_static("application/json")), + ], + body: Some(body_bytes), + depth: 0, + } +} + +/// Parse the Chat Completions response and extract the summary text. +/// +/// Expected shape: `{"choices": [{"message": {"content": "..."}}]}` +fn parse_summarization_response(body: &[u8]) -> Result { + match serde_json::from_slice::(body) { + Ok(body) => body + .get("choices") + .and_then(Value::as_array) + .and_then(|choices| choices.first()) + .and_then(|choice| choice.get("message")) + .and_then(|msg| msg.get("content")) + .and_then(Value::as_str) + .map(ToOwned::to_owned) + .ok_or_else(|| "Chat Completions response missing choices[0].message.content".to_owned()), + Err(err) => Err(format!("failed to parse Chat Completions response JSON: {err}")), + } +} + +/// Build the compaction output item. +/// +/// Returns: `{"role": "system", "content": ""}` +fn build_compaction_item(summary: &str) -> Value { + serde_json::json!({ + "role": "system", + "content": summary + }) +} + +/// Replace conversation history with the compaction item. +/// +/// After replacement: +/// - `state.messages` = `[compaction_item, ...state.input]` +/// - `state.persisted_messages` = `[compaction_item, ...state.input]` +/// +/// `state.input` holds the current request's input items (unchanged +/// by rehydrate), so the current turn's messages are preserved. +fn replace_messages(state: &mut ResponsesState, compaction_item: Value) { + let mut new_messages = Vec::with_capacity(state.input.len() + 1); + new_messages.push(compaction_item); + new_messages.extend(state.input.iter().cloned()); + state.persisted_messages = new_messages.clone(); + state.messages = new_messages; +} + +/// Format a message array as readable text for the summarization prompt. +/// +/// Each message becomes: `: ` +/// Messages are separated by blank lines. +fn build_conversation_text(messages: &[Value]) -> String { + let mut buf = String::with_capacity(messages.len() * 100); + for msg in messages { + let role = msg.get("role").and_then(Value::as_str).unwrap_or("unknown"); + let content = extract_content(msg); + if content.is_empty() { + continue; + } + if !buf.is_empty() { + buf.push_str("\n\n"); + } + buf.push_str(role); + buf.push_str(": "); + buf.push_str(&content); + } + buf +} + +/// Extract text content from a message's `content` field. +/// +/// Content can be a plain string, an array of content parts +/// (each with a `"text"` field), or absent/null. +/// +/// Returns `Cow::Borrowed` for plain strings (zero-copy) and +/// `Cow::Owned` for array content that must be joined. +fn extract_content(msg: &Value) -> Cow<'_, str> { + let Some(content) = msg.get("content") else { + return Cow::Borrowed(""); + }; + if let Some(s) = content.as_str() { + return Cow::Borrowed(s); + } + if let Some(arr) = content.as_array() { + let texts: Vec<&str> = arr + .iter() + .filter_map(|part| part.get("text").and_then(Value::as_str)) + .collect(); + return Cow::Owned(texts.join(" ")); + } + Cow::Borrowed("") +} diff --git a/apis/src/openai/responses/compact/tests.rs b/apis/src/openai/responses/compact/tests.rs new file mode 100644 index 00000000..ee578460 --- /dev/null +++ b/apis/src/openai/responses/compact/tests.rs @@ -0,0 +1,343 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Praxis Contributors + +use serde_json::json; + +use super::*; +use crate::openai::responses::config_validation::FailureMode; + +// ============================================================================= +// Config tests +// ============================================================================= + +fn base_config() -> CompactFilterConfig { + CompactFilterConfig { + inference_url: "http://localhost:11434/v1/chat/completions".to_owned(), + default_model: "gpt-4o-mini".to_owned(), + tiktoken_encoding: "cl100k_base".to_owned(), + timeout_ms: None, + failure_mode: None, + status_on_error: None, + } +} + +#[test] +fn build_config_applies_defaults() { + let cfg = build_config(&base_config()).unwrap(); + assert_eq!(cfg.inference_url, "http://localhost:11434/v1/chat/completions"); + assert_eq!(cfg.default_model, "gpt-4o-mini"); + assert_eq!(cfg.tiktoken_encoding, "cl100k_base"); + assert_eq!(cfg.callout.timeout_ms, 30_000); + assert_eq!(cfg.callout.failure_mode, FailureMode::Closed); + assert_eq!(cfg.callout.status_on_error, 502); +} + +#[test] +fn build_config_rejects_empty_inference_url() { + let mut cfg = base_config(); + cfg.inference_url = String::new(); + assert!(build_config(&cfg).is_err()); +} + +#[test] +fn build_config_rejects_zero_timeout() { + let mut cfg = base_config(); + cfg.timeout_ms = Some(0); + assert!(build_config(&cfg).is_err()); +} + +#[test] +fn build_config_rejects_invalid_status() { + let mut cfg = base_config(); + cfg.status_on_error = Some(999); + assert!(build_config(&cfg).is_err()); +} + +#[test] +fn build_config_custom_values() { + let mut cfg = base_config(); + cfg.timeout_ms = Some(60_000); + cfg.failure_mode = Some(FailureMode::Open); + cfg.status_on_error = Some(503); + let validated = build_config(&cfg).unwrap(); + assert_eq!(validated.callout.timeout_ms, 60_000); + assert_eq!(validated.callout.failure_mode, FailureMode::Open); + assert_eq!(validated.callout.status_on_error, 503); +} + +// ============================================================================= +// extract_compaction_config tests +// ============================================================================= + +#[test] +fn extract_compaction_config_with_compaction_entry() { + let cm = Some(json!([{"type": "compaction", "compact_threshold": 50_000}])); + let params = extract_compaction_config(&cm); + assert!(params.is_some()); + let params = params.unwrap(); + assert_eq!(params.compact_threshold, 50_000); + assert!(params.compaction_model.is_none()); +} + +#[test] +fn extract_compaction_config_with_model_override() { + let cm = Some(json!([{ + "type": "compaction", + "compact_threshold": 100_000, + "compaction_model": "gpt-4o" + }])); + let params = extract_compaction_config(&cm).unwrap(); + assert_eq!(params.compact_threshold, 100_000); + assert_eq!(params.compaction_model.as_deref(), Some("gpt-4o")); +} + +#[test] +fn extract_compaction_config_no_compaction_entry() { + let cm = Some(json!([{"type": "truncation", "max_tokens": 4096}])); + assert!(extract_compaction_config(&cm).is_none()); +} + +#[test] +fn extract_compaction_config_none() { + assert!(extract_compaction_config(&None).is_none()); +} + +#[test] +fn extract_compaction_config_empty_array() { + let cm = Some(json!([])); + assert!(extract_compaction_config(&cm).is_none()); +} + +#[test] +fn extract_compaction_config_defaults_threshold_to_zero() { + let cm = Some(json!([{"type": "compaction"}])); + let params = extract_compaction_config(&cm).unwrap(); + assert_eq!(params.compact_threshold, 0); +} + +// ============================================================================= +// build_compaction_item tests +// ============================================================================= + +#[test] +fn compaction_item_has_correct_shape() { + let item = build_compaction_item("This is a summary."); + assert_eq!(item["role"], "system"); + assert_eq!(item["content"], "This is a summary."); +} + +// ============================================================================= +// parse_summarization_response tests +// ============================================================================= + +#[test] +fn parse_valid_chat_completion_response() { + let response = json!({ + "choices": [{ + "message": { + "role": "assistant", + "content": "Here is the summary." + } + }] + }); + let body = serde_json::to_vec(&response).unwrap(); + let result = parse_summarization_response(&body); + assert_eq!(result.unwrap(), "Here is the summary."); +} + +#[test] +fn parse_malformed_response_returns_error() { + let result = parse_summarization_response(b"not json"); + assert!(result.is_err()); +} + +#[test] +fn parse_response_missing_choices_returns_error() { + let response = json!({"id": "chatcmpl-123"}); + let body = serde_json::to_vec(&response).unwrap(); + assert!(parse_summarization_response(&body).is_err()); +} + +#[test] +fn parse_response_empty_choices_returns_error() { + let response = json!({"choices": []}); + let body = serde_json::to_vec(&response).unwrap(); + assert!(parse_summarization_response(&body).is_err()); +} + +// ============================================================================= +// build_conversation_text tests +// ============================================================================= + +#[test] +fn conversation_text_simple_messages() { + let messages = vec![ + json!({"role": "user", "content": "Hello"}), + json!({"role": "assistant", "content": "Hi there!"}), + ]; + let text = build_conversation_text(&messages); + assert!(text.contains("user: Hello")); + assert!(text.contains("assistant: Hi there!")); +} + +#[test] +fn conversation_text_empty_messages() { + let text = build_conversation_text(&[]); + assert!(text.is_empty()); +} + +#[test] +fn conversation_text_skips_empty_content() { + let messages = vec![ + json!({"role": "user", "content": "Hello"}), + json!({"role": "assistant"}), + json!({"role": "user", "content": "Still here"}), + ]; + let text = build_conversation_text(&messages); + assert!(!text.contains("assistant")); + assert!(text.contains("user: Hello")); + assert!(text.contains("user: Still here")); +} + +#[test] +fn conversation_text_array_content() { + let messages = vec![json!({ + "role": "user", + "content": [ + {"type": "text", "text": "Part one"}, + {"type": "text", "text": "Part two"} + ] + })]; + let text = build_conversation_text(&messages); + assert!(text.contains("user: Part one Part two")); +} + +// ============================================================================= +// extract_content tests +// ============================================================================= + +#[test] +fn extract_content_string() { + let msg = json!({"content": "hello"}); + assert_eq!(extract_content(&msg), "hello"); +} + +#[test] +fn extract_content_array() { + let msg = json!({"content": [{"type": "text", "text": "a"}, {"type": "text", "text": "b"}]}); + assert_eq!(extract_content(&msg), "a b"); +} + +#[test] +fn extract_content_missing() { + let msg = json!({"role": "user"}); + assert_eq!(extract_content(&msg), ""); +} + +#[test] +fn extract_content_null() { + let msg = json!({"content": null}); + assert_eq!(extract_content(&msg), ""); +} + +// ============================================================================= +// build_summarization_request tests +// ============================================================================= + +#[test] +fn summarization_request_without_instructions() { + let messages = vec![json!({"role": "user", "content": "Hello"})]; + let conversation_text = build_conversation_text(&messages); + let req = build_summarization_request( + &conversation_text, + None, + "gpt-4o-mini", + "http://localhost/v1/chat/completions", + ); + assert_eq!(req.method, http::Method::POST); + assert_eq!(req.url, "http://localhost/v1/chat/completions"); + assert!(req.body.is_some()); + let body: Value = serde_json::from_slice(&req.body.unwrap()).unwrap(); + assert_eq!(body["model"], "gpt-4o-mini"); + let msgs = body["messages"].as_array().unwrap(); + assert_eq!(msgs.len(), 2); + assert_eq!(msgs[0]["role"], "system"); + assert!(msgs[0]["content"].as_str().unwrap().contains("Summarize")); + assert_eq!(msgs[1]["role"], "user"); + assert!(msgs[1]["content"].as_str().unwrap().contains("user: Hello")); +} + +#[test] +fn summarization_request_with_instructions() { + let messages = vec![json!({"role": "user", "content": "Hello"})]; + let conversation_text = build_conversation_text(&messages); + let req = build_summarization_request( + &conversation_text, + Some("Be concise"), + "gpt-4o-mini", + "http://localhost/v1/chat/completions", + ); + let body: Value = serde_json::from_slice(&req.body.unwrap()).unwrap(); + let system = body["messages"][0]["content"].as_str().unwrap(); + assert!(system.starts_with("Be concise"), "instructions should be prepended"); + assert!(system.contains("Summarize"), "system prompt should follow"); +} + +// ============================================================================= +// replace_messages tests +// ============================================================================= + +#[test] +fn replace_messages_preserves_current_input() { + let mut state = ResponsesState::from_request_body(json!({ + "model": "gpt-4o", + "input": "What's next?" + })); + state + .messages + .insert(0, json!({"role": "user", "content": "old question"})); + state + .messages + .insert(1, json!({"role": "assistant", "content": "old answer"})); + state + .persisted_messages + .insert(0, json!({"role": "user", "content": "old question"})); + state + .persisted_messages + .insert(1, json!({"role": "assistant", "content": "old answer"})); + + let compaction_item = build_compaction_item("Summary of old conversation."); + replace_messages(&mut state, compaction_item); + + assert_eq!(state.messages.len(), 2, "should have compaction + current input"); + assert_eq!(state.messages[0]["role"], "system"); + assert_eq!(state.messages[0]["content"], "Summary of old conversation."); + assert_eq!(state.persisted_messages.len(), 2); + assert_eq!(state.persisted_messages[0]["role"], "system"); +} + +// ============================================================================= +// get_token_count tests +// ============================================================================= + +#[test] +fn token_count_returns_some_for_known_encoding() { + let text = build_conversation_text(&[json!({"role": "user", "content": "Hello world"})]); + let count = get_token_count(&text, "cl100k_base"); + assert!(count.is_some()); + assert!(count.unwrap() > 0); +} + +#[test] +fn token_count_returns_none_for_unknown_encoding() { + let text = build_conversation_text(&[json!({"role": "user", "content": "Hello"})]); + assert!(get_token_count(&text, "unknown_encoding").is_none()); +} + +#[test] +fn token_count_supports_o200k() { + let text = build_conversation_text(&[json!({"role": "user", "content": "Hello world"})]); + let count = get_token_count(&text, "o200k_base"); + assert!(count.is_some()); + assert!(count.unwrap() > 0); +} diff --git a/apis/src/openai/responses/config_validation.rs b/apis/src/openai/responses/config_validation.rs new file mode 100644 index 00000000..82f43352 --- /dev/null +++ b/apis/src/openai/responses/config_validation.rs @@ -0,0 +1,167 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Praxis Contributors + +//! Shared config validation helpers for Responses API filters. + +use praxis_core::callout::{CalloutConfig, CircuitBreakerConfig, FailureMode as CoreFailureMode}; +use praxis_filter::FilterError; +use serde::Deserialize; + +// ----------------------------------------------------------------------------- +// FailureMode +// ----------------------------------------------------------------------------- + +/// What happens when a callout to an external service fails. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum FailureMode { + /// Reject the request on failure (default). + Closed, + /// Continue without the callout result on failure. + Open, +} + +// ----------------------------------------------------------------------------- +// CalloutSettings +// ----------------------------------------------------------------------------- + +/// Common callout fields shared by filters that make HTTP callouts. +#[derive(Debug, Clone, Copy)] +pub(crate) struct CalloutSettings { + /// Callout timeout in milliseconds. + pub timeout_ms: u64, + /// Failure mode for the callout. + pub failure_mode: FailureMode, + /// HTTP status code to return when rejecting on error. + pub status_on_error: u16, +} + +impl CalloutSettings { + /// Build a [`CalloutConfig`] from these settings. + pub(crate) fn build_callout_config(self) -> CalloutConfig { + let failure_mode = match self.failure_mode { + FailureMode::Closed => CoreFailureMode::Closed, + FailureMode::Open => CoreFailureMode::Open, + }; + CalloutConfig { + circuit_breaker: Some(CircuitBreakerConfig { + consecutive_failures: 5, + recovery_window_ms: 30_000, + }), + failure_mode, + status_on_error: self.status_on_error, + timeout_ms: self.timeout_ms, + ..CalloutConfig::default() + } + } +} + +// ----------------------------------------------------------------------------- +// Validation helpers +// ----------------------------------------------------------------------------- + +/// Validate `timeout_ms`, applying a default and rejecting zero. +/// +/// # Errors +/// +/// Returns [`FilterError`] when the resolved value is zero. +pub(crate) fn validate_timeout_ms(filter: &str, raw: Option, default: u64) -> Result { + let value = raw.unwrap_or(default); + if value == 0 { + return Err(format!("{filter}: timeout_ms must be greater than 0").into()); + } + Ok(value) +} + +/// Validate `status_on_error`, applying a default and rejecting +/// values outside the HTTP status range. +/// +/// # Errors +/// +/// Returns [`FilterError`] when the resolved value is not in +/// `100..=599`. +pub(crate) fn validate_status_on_error(filter: &str, raw: Option, default: u16) -> Result { + let value = raw.unwrap_or(default); + if !(100..=599).contains(&value) { + return Err(format!("{filter}: status_on_error must be between 100 and 599, got {value}").into()); + } + Ok(value) +} + +#[cfg(test)] +#[expect(clippy::allow_attributes, reason = "blanket test suppressions")] +#[allow(clippy::unwrap_used, clippy::expect_used, reason = "tests")] +mod tests { + use super::*; + + #[test] + fn timeout_applies_default() { + assert_eq!(validate_timeout_ms("test", None, 5000).unwrap(), 5000); + } + + #[test] + fn timeout_uses_provided_value() { + assert_eq!(validate_timeout_ms("test", Some(10_000), 5000).unwrap(), 10_000); + } + + #[test] + fn timeout_zero_rejected() { + let err = validate_timeout_ms("test", Some(0), 5000).unwrap_err(); + assert!( + err.to_string().contains("greater than 0"), + "zero should be rejected, got: {err}" + ); + } + + #[test] + fn timeout_error_includes_filter_name() { + let err = validate_timeout_ms("my_filter", Some(0), 5000).unwrap_err(); + assert!( + err.to_string().contains("my_filter"), + "error should include filter name, got: {err}" + ); + } + + #[test] + fn status_applies_default() { + assert_eq!(validate_status_on_error("test", None, 502).unwrap(), 502); + } + + #[test] + fn status_uses_provided_value() { + assert_eq!(validate_status_on_error("test", Some(503), 502).unwrap(), 503); + } + + #[test] + fn status_below_range_rejected() { + let err = validate_status_on_error("test", Some(99), 502).unwrap_err(); + assert!( + err.to_string().contains("between 100 and 599"), + "below range should be rejected, got: {err}" + ); + } + + #[test] + fn status_above_range_rejected() { + let err = validate_status_on_error("test", Some(600), 502).unwrap_err(); + assert!( + err.to_string().contains("between 100 and 599"), + "above range should be rejected, got: {err}" + ); + } + + #[test] + fn status_boundaries_accepted() { + validate_status_on_error("test", Some(100), 502).expect("100 should be accepted"); + validate_status_on_error("test", Some(599), 502).expect("599 should be accepted"); + } + + #[test] + fn status_error_includes_filter_name() { + let err = validate_status_on_error("my_filter", Some(0), 502).unwrap_err(); + assert!( + err.to_string().contains("my_filter"), + "error should include filter name, got: {err}" + ); + } +} diff --git a/apis/src/openai/responses/mod.rs b/apis/src/openai/responses/mod.rs index 0bf00516..9e032ff2 100644 --- a/apis/src/openai/responses/mod.rs +++ b/apis/src/openai/responses/mod.rs @@ -18,6 +18,7 @@ //! to validate parameter combinations and extract additional fields. mod config; +pub(crate) mod config_validation; pub(crate) mod doc_extract; pub(crate) mod error; pub(crate) mod file_resolve; @@ -462,10 +463,12 @@ fn defaulted_openresponses_item_type(object: &serde_json::Map Result) -> Result { - let value = raw.unwrap_or(DEFAULT_TIMEOUT_MS); - if value == 0 { - return Err(FilterError::from( - "openai_web_search: timeout_ms must be greater than 0".to_owned(), - )); - } - Ok(value) -} - -/// Validate HTTP status code, applying the default and rejecting out-of-range. -fn validate_status_on_error(raw: Option) -> Result { - let value = raw.unwrap_or(DEFAULT_STATUS_ON_ERROR); - if !(100..=599).contains(&value) { - return Err(FilterError::from(format!( - "openai_web_search: status_on_error must be between 100 and 599, got {value}" - ))); - } - Ok(value) -} /// Validate `default_context_size`, defaulting to `Medium` when /// absent and rejecting unknown values. @@ -306,10 +269,10 @@ mod tests { assert_eq!(cfg.provider, SearchProvider::Brave); assert_eq!(cfg.api_key.expose_secret(), "test-key-123"); assert_eq!(cfg.default_context_size, SearchContextSize::Medium); - assert_eq!(cfg.timeout_ms, DEFAULT_TIMEOUT_MS); + assert_eq!(cfg.callout.timeout_ms, DEFAULT_TIMEOUT_MS); assert_eq!(cfg.max_body_bytes, MAX_JSON_BODY_BYTES); - assert_eq!(cfg.failure_mode, FailureMode::Closed); - assert_eq!(cfg.status_on_error, DEFAULT_STATUS_ON_ERROR); + assert_eq!(cfg.callout.failure_mode, FailureMode::Closed); + assert_eq!(cfg.callout.status_on_error, DEFAULT_STATUS_ON_ERROR); } #[test] @@ -359,9 +322,9 @@ mod tests { cfg.status_on_error = Some(503); let validated = build_config(&cfg).unwrap(); assert_eq!(validated.default_context_size, SearchContextSize::High); - assert_eq!(validated.timeout_ms, 15_000); - assert_eq!(validated.failure_mode, FailureMode::Open); - assert_eq!(validated.status_on_error, 503); + assert_eq!(validated.callout.timeout_ms, 15_000); + assert_eq!(validated.callout.failure_mode, FailureMode::Open); + assert_eq!(validated.callout.status_on_error, 503); } #[test] diff --git a/apis/src/openai/responses/web_search/provider.rs b/apis/src/openai/responses/web_search/provider.rs index 55a22838..235f3507 100644 --- a/apis/src/openai/responses/web_search/provider.rs +++ b/apis/src/openai/responses/web_search/provider.rs @@ -6,15 +6,14 @@ //! Uses [`CalloutClient`] from praxis-core for HTTP callouts with //! circuit breaking, timeout, and loop prevention. -use praxis_core::callout::{ - CalloutClient, CalloutConfig, CalloutRequest, CalloutResult, CircuitBreakerConfig, FailureMode as CoreFailureMode, -}; +use praxis_core::callout::{CalloutClient, CalloutRequest, CalloutResult}; use praxis_filter::FilterError; use secrecy::{ExposeSecret as _, SecretString}; use serde_json::Value; use tracing::{debug, warn}; -use super::config::{FailureMode, SearchContextSize, SearchProvider, ValidatedConfig}; +use super::config::{SearchContextSize, SearchProvider, ValidatedConfig}; +use crate::openai::responses::config_validation::FailureMode; // ----------------------------------------------------------------------------- // SearchResult @@ -82,24 +81,6 @@ impl std::fmt::Debug for SearchClient { } } -/// Build a [`CalloutConfig`] from validated filter config. -fn build_callout_config(config: &ValidatedConfig) -> CalloutConfig { - let failure_mode = match config.failure_mode { - FailureMode::Closed => CoreFailureMode::Closed, - FailureMode::Open => CoreFailureMode::Open, - }; - CalloutConfig { - circuit_breaker: Some(CircuitBreakerConfig { - consecutive_failures: 5, - recovery_window_ms: 30_000, - }), - failure_mode, - status_on_error: config.status_on_error, - timeout_ms: config.timeout_ms, - ..CalloutConfig::default() - } -} - impl SearchClient { /// Build a search client from validated filter config. /// @@ -108,15 +89,15 @@ impl SearchClient { /// Returns [`FilterError`] if the underlying [`CalloutClient`] /// cannot be constructed. pub(crate) fn from_config(config: &ValidatedConfig) -> Result { - let callout_config = build_callout_config(config); + let callout_config = config.callout.build_callout_config(); let client = CalloutClient::new(callout_config).map_err(|e| FilterError::from(e.to_string()))?; Ok(Self { client, provider: config.provider, api_key: config.api_key.clone(), default_context_size: config.default_context_size, - failure_mode: config.failure_mode, - status_on_error: config.status_on_error, + failure_mode: config.callout.failure_mode, + status_on_error: config.callout.status_on_error, }) } @@ -361,6 +342,7 @@ mod tests { use serde_json::json; use super::*; + use crate::openai::responses::config_validation::CalloutSettings; #[test] fn parse_brave_results_normal() { @@ -451,10 +433,12 @@ mod tests { provider: SearchProvider::You, api_key: SecretString::from("test-key".to_owned()), default_context_size: SearchContextSize::Medium, - timeout_ms: 5000, max_body_bytes: 64 * 1024 * 1024, - failure_mode: FailureMode::Closed, - status_on_error: 502, + callout: CalloutSettings { + timeout_ms: 5000, + failure_mode: FailureMode::Closed, + status_on_error: 502, + }, }; let client = SearchClient::from_config(&config).unwrap(); @@ -508,10 +492,12 @@ mod tests { provider: SearchProvider::Brave, api_key: SecretString::from("test-key".to_owned()), default_context_size: SearchContextSize::Medium, - timeout_ms: 5000, max_body_bytes: 64 * 1024 * 1024, - failure_mode: FailureMode::Closed, - status_on_error: 502, + callout: CalloutSettings { + timeout_ms: 5000, + failure_mode: FailureMode::Closed, + status_on_error: 502, + }, }; let client = SearchClient::from_config(&config); assert!(client.is_ok()); @@ -523,10 +509,12 @@ mod tests { provider: SearchProvider::Brave, api_key: SecretString::from("test-key".to_owned()), default_context_size: SearchContextSize::Medium, - timeout_ms: 5000, max_body_bytes: 64 * 1024 * 1024, - failure_mode: FailureMode::Closed, - status_on_error: 502, + callout: CalloutSettings { + timeout_ms: 5000, + failure_mode: FailureMode::Closed, + status_on_error: 502, + }, }; let client = SearchClient::from_config(&config).unwrap(); let outcome = client.parse_response(b"not json"); @@ -542,10 +530,12 @@ mod tests { provider: SearchProvider::Brave, api_key: SecretString::from("test-key".to_owned()), default_context_size: SearchContextSize::Medium, - timeout_ms: 5000, max_body_bytes: 64 * 1024 * 1024, - failure_mode: FailureMode::Open, - status_on_error: 502, + callout: CalloutSettings { + timeout_ms: 5000, + failure_mode: FailureMode::Open, + status_on_error: 502, + }, }; let client = SearchClient::from_config(&config).unwrap(); let outcome = client.parse_response(b"not json"); diff --git a/docs/filters/openai_responses_compact.md b/docs/filters/openai_responses_compact.md new file mode 100644 index 00000000..f5e14d63 --- /dev/null +++ b/docs/filters/openai_responses_compact.md @@ -0,0 +1,39 @@ + + + +# `openai_responses_compact` + +Summarizes conversation history when the token count exceeds a configured threshold. + +## Configuration + +| Field | Type | Required | Description | +|-------|------|---------|-------------| +| `inference_url` | string | yes | URL of the inference backend for summarization calls. E.g., `"http://localhost:11434/v1/chat/completions"` | +| `default_model` | string | no | Default model for summarization when not overridden in the request's `context_management`. | +| `tiktoken_encoding` | string | no | Tiktoken encoding name for local token estimation. Used as fallback when `previous_usage` is unavailable. | +| `timeout_ms` | integer | no | Callout timeout in milliseconds. | +| `failure_mode` | `closed` \| `open` | no | Failure mode for the inference callout. | +| `status_on_error` | integer | no | HTTP status code to return when rejecting on error. | + +## Examples + +### Example 1 + +```yaml +filter: openai_responses_compact +inference_url: "http://localhost:11434/v1/chat/completions" +default_model: llama3.2:1b +``` + +### Example 2 + +```yaml +filter: openai_responses_compact +inference_url: "http://localhost:11434/v1/chat/completions" +default_model: gpt-4o-mini +tiktoken_encoding: cl100k_base +timeout_ms: 30000 +failure_mode: closed +status_on_error: 502 +``` diff --git a/docs/filters/reference.md b/docs/filters/reference.md index a26173b6..389b6877 100644 --- a/docs/filters/reference.md +++ b/docs/filters/reference.md @@ -31,6 +31,7 @@ see the [Praxis core filter reference][core-ref]. | [`openai_mcp_dispatch`](openai_mcp_dispatch.md) | Executes MCP tool calls against upstream MCP servers within the Responses API agentic loop. | | [`openai_mcp_tool_resolve`](openai_mcp_tool_resolve.md) | Resolves MCP tool entries from the Responses API `tools` array into concrete tool definitions by calling `tools/list` on each upstream MCP server. | | [`openai_response_store`](openai_response_store.md) | Persists Responses API responses to the configured response store backend. | +| [`openai_responses_compact`](openai_responses_compact.md) | Summarizes conversation history when the token count exceeds a configured threshold. | | [`openai_responses_format`](openai_responses_format.md) | Classifies AI API request bodies and promotes routing facts to headers, metadata, and filter results without mutating the body. | | [`openai_responses_model_rewrite`](openai_responses_model_rewrite.md) | Rewrites the `model` field in Responses API request bodies. | | [`openai_responses_proxy`](openai_responses_proxy.md) | Rebuilds the request body from `ResponsesState` when present. | diff --git a/examples/configs/openai/responses/compact.yaml b/examples/configs/openai/responses/compact.yaml new file mode 100644 index 00000000..98c1eb73 --- /dev/null +++ b/examples/configs/openai/responses/compact.yaml @@ -0,0 +1,72 @@ +# Compact Filter Example +# +# Demonstrates the compaction flow: store a response, rehydrate it +# on the next turn, and count tokens to check if compaction is needed. +# +# Usage: +# cargo run -p praxis-ai-proxy -- -c examples/configs/openai/responses/compact.yaml +# +# Requires an OpenAI-compatible backend on port 11434 (e.g. Ollama). +# +# Test: +# # Step 1 — store a response +# RESP_ID=$(curl -s -X POST http://localhost:8080/v1/responses \ +# -H "Content-Type: application/json" \ +# -d '{"model":"llama3.2:1b","input":"Explain TCP vs UDP in detail"}' \ +# | python3 -c "import sys,json; print(json.load(sys.stdin)['id'])") +# +# # Step 2 — follow up with compaction (threshold=200 tokens) +# curl -s -X POST http://localhost:8080/v1/responses \ +# -H "Content-Type: application/json" \ +# -d "{\"model\":\"llama3.2:1b\",\"input\":\"Compare with QUIC\",\"previous_response_id\":\"$RESP_ID\",\"context_management\":[{\"type\":\"compaction\",\"compact_threshold\":200}],\"store\":false}" + +listeners: + - name: ai-gateway + address: "127.0.0.1:8080" + filter_chains: [compact-pipeline] + +filter_chains: + - name: compact-pipeline + filters: + - filter: openai_responses_format + on_invalid: continue + headers: + format: x-praxis-ai-format + model: x-praxis-ai-model + stream: x-praxis-ai-stream + mode: x-praxis-responses-mode + + - filter: openai_responses_validate + + - filter: openai_tool_parse + + - filter: openai_response_store + backend: sqlite + database_url: "sqlite://responses.db?mode=rwc" + responses_table: openai_responses + conversations_table: openai_conversations + + - filter: openai_stream_events + + - filter: openai_responses_rehydrate + + - filter: openai_responses_compact + inference_url: "http://localhost:11434/v1/chat/completions" + default_model: llama3.2:1b + timeout_ms: 60000 + + - filter: openai_responses_proxy + name: inference + + - filter: router + routes: + - path: "/v1/responses" + headers: + x-praxis-ai-format: "openai_responses" + cluster: "inference-backend" + + - filter: load_balancer + clusters: + - name: "inference-backend" + endpoints: + - "127.0.0.1:11434" \ No newline at end of file diff --git a/server/src/lib.rs b/server/src/lib.rs index c510580e..9f1f6a06 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -133,6 +133,10 @@ fn register_openai_responses_filters(registry: &mut praxis_filter::FilterRegistr @register registry, http "openai_responses_rehydrate" => praxis_ai_apis::openai::RehydrateFilter::from_config ); + praxis_filter::register_filters!( + @register registry, + http "openai_responses_compact" => praxis_ai_apis::openai::CompactFilter::from_config + ); register_openai_response_filters(registry); } diff --git a/tests/integration/tests/suite/examples/compact.rs b/tests/integration/tests/suite/examples/compact.rs new file mode 100644 index 00000000..518ab43a --- /dev/null +++ b/tests/integration/tests/suite/examples/compact.rs @@ -0,0 +1,118 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Praxis Contributors + +//! Functional tests for the `compact` example config. +//! +//! Verifies that the example pipeline builds, simple requests pass +//! through, and the multi-turn compaction flow works end-to-end. + +use std::collections::HashMap; + +use praxis_test_utils::{ + Backend, TempSqlite, example_config_path, free_port, http_send, json_post, parse_status, patch_yaml, start_proxy, +}; + +// ----------------------------------------------------------------------------- +// Constants +// ----------------------------------------------------------------------------- + +/// Backend response for the first turn — stored by response_store. +/// The output text is long enough to exceed a low compact_threshold. +const FIRST_RESPONSE_JSON: &str = r#"{"id":"resp_compact","created_at":1000,"model":"gpt-4.1","object":"response","status":"completed","input":"Explain TCP vs UDP","output":[{"type":"message","content":[{"type":"output_text","text":"TCP is a connection-oriented protocol that provides reliable, ordered delivery of data. It establishes a connection through a three-way handshake before transmitting data. UDP is a connectionless protocol that sends data without establishing a connection first. TCP guarantees delivery through acknowledgments and retransmissions while UDP does not. TCP is used for applications requiring reliability like web browsing and email while UDP is used for real-time applications like video streaming and gaming where speed matters more than reliability."}]}]}"#; + +/// Chat Completions response returned by the mock backend on the second +/// turn — used by both the compact filter's summarization callout and +/// the main inference forwarded by `openai_responses_proxy`. +const CHAT_COMPLETIONS_RESPONSE: &str = r#"{"id":"chatcmpl-1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"Summary of the conversation."},"finish_reason":"stop"}],"usage":{"prompt_tokens":50,"completion_tokens":10,"total_tokens":60}}"#; + +// ----------------------------------------------------------------------------- +// Helpers +// ----------------------------------------------------------------------------- + +/// Load the compact example config, replacing the SQLite URL and +/// patching listener/backend addresses. +fn load_compact_config(yaml: &str, db_url: &str, proxy_port: u16, backend_port: u16) -> praxis_core::config::Config { + let replaced = yaml + .replace("sqlite://responses.db?mode=rwc", db_url) + .replace("localhost:11434", &format!("127.0.0.1:{backend_port}")); + let patched = patch_yaml( + &replaced, + proxy_port, + &HashMap::from([("127.0.0.1:11434", backend_port)]), + ); + praxis_core::config::Config::from_yaml(&patched).expect("patched config should parse") +} + +// ----------------------------------------------------------------------------- +// Tests +// ----------------------------------------------------------------------------- + +#[test] +fn compact_passthrough() { + let backend_guard = Backend::fixed(FIRST_RESPONSE_JSON) + .header("content-type", "application/json") + .start_with_shutdown(); + let proxy_port = free_port(); + + let yaml = std::fs::read_to_string(example_config_path("openai/responses/compact.yaml")) + .expect("example config should exist"); + let config = load_compact_config(&yaml, "sqlite::memory:", proxy_port, backend_guard.port()); + let proxy = start_proxy(&config); + + let raw = http_send( + proxy.addr(), + &json_post("/v1/responses", r#"{"model":"gpt-4.1","input":"Hello"}"#), + ); + + assert_eq!( + parse_status(&raw), + 200, + "request without context_management should pass through" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn compact_multi_turn_compaction() { + let backend1 = Backend::fixed(FIRST_RESPONSE_JSON) + .header("content-type", "application/json") + .start_with_shutdown(); + let proxy_port = free_port(); + + let db = TempSqlite::new("compact"); + let yaml = std::fs::read_to_string(example_config_path("openai/responses/compact.yaml")) + .expect("example config should exist"); + + let config1 = load_compact_config(&yaml, db.url(), proxy_port, backend1.port()); + let proxy1 = start_proxy(&config1); + + let raw1 = http_send( + proxy1.addr(), + &json_post("/v1/responses", r#"{"model":"gpt-4.1","input":"Explain TCP vs UDP"}"#), + ); + assert_eq!(parse_status(&raw1), 200, "first request should succeed"); + + drop(backend1); + drop(proxy1); + + let backend2 = Backend::fixed(CHAT_COMPLETIONS_RESPONSE) + .header("content-type", "application/json") + .start_with_shutdown(); + + let config2 = load_compact_config(&yaml, db.url(), proxy_port, backend2.port()); + let proxy2 = start_proxy(&config2); + + let raw2 = http_send( + proxy2.addr(), + &json_post( + "/v1/responses", + r#"{"model":"gpt-4.1","input":"Compare with QUIC","previous_response_id":"resp_compact","context_management":[{"type":"compaction","compact_threshold":50}]}"#, + ), + ); + let status2 = parse_status(&raw2); + assert_eq!( + status2, 200, + "second request with compaction should succeed (callout + pipeline completed)" + ); + + drop(proxy2); +} diff --git a/tests/integration/tests/suite/examples/mod.rs b/tests/integration/tests/suite/examples/mod.rs index c28c270b..67cc8e57 100644 --- a/tests/integration/tests/suite/examples/mod.rs +++ b/tests/integration/tests/suite/examples/mod.rs @@ -9,6 +9,7 @@ pub use test_utils::load_example_config; mod agentic_routing; mod anthropic_messages; +mod compact; mod credential_injection; mod full_flow; mod mcp_broker;