From b85590182673b102a7e8b9fa96195e7969228a1d Mon Sep 17 00:00:00 2001 From: Alexander Cristurean Date: Fri, 17 Jul 2026 14:06:54 +0200 Subject: [PATCH 01/11] feat: initial scaffold for compact filter. Signed-off-by: Alexander Cristurean # Conflicts: # apis/src/openai/mod.rs # apis/src/openai/responses/mod.rs # examples/configs/openai/responses/full-flow.yaml --- Cargo.lock | 33 ++++++++++++++++++++++++++++++++ Cargo.toml | 1 + apis/Cargo.toml | 1 + apis/src/openai/responses/mod.rs | 2 ++ server/src/lib.rs | 4 ++++ 5 files changed, 41 insertions(+) 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/responses/mod.rs b/apis/src/openai/responses/mod.rs index 0bf00516..d27cde23 100644 --- a/apis/src/openai/responses/mod.rs +++ b/apis/src/openai/responses/mod.rs @@ -462,10 +462,12 @@ fn defaulted_openresponses_item_type(object: &serde_json::Map 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); } From 901e0d55a9aa77bbdf7d5bc012bc7146f4822b72 Mon Sep 17 00:00:00 2001 From: Alexander Cristurean Date: Fri, 17 Jul 2026 16:30:09 +0200 Subject: [PATCH 02/11] fix: CompactFilter scaffold. Signed-off-by: Alexander Cristurean --- apis/src/openai/responses/compact/config.rs | 128 +++++++ apis/src/openai/responses/compact/mod.rs | 403 ++++++++++++++++++++ apis/src/openai/responses/compact/tests.rs | 206 ++++++++++ 3 files changed, 737 insertions(+) create mode 100644 apis/src/openai/responses/compact/config.rs create mode 100644 apis/src/openai/responses/compact/mod.rs create mode 100644 apis/src/openai/responses/compact/tests.rs diff --git a/apis/src/openai/responses/compact/config.rs b/apis/src/openai/responses/compact/config.rs new file mode 100644 index 00000000..2e5b65a4 --- /dev/null +++ b/apis/src/openai/responses/compact/config.rs @@ -0,0 +1,128 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Praxis Contributors + +//! Configuration for the `openai_responses_compact` filter. + +use praxis_filter::FilterError; +use serde::Deserialize; + +/// 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; + +// ----------------------------------------------------------------------------- +// FailureMode +// ----------------------------------------------------------------------------- + +/// What happens when the summarization callout fails. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "snake_case")] +pub(super) enum FailureMode { + /// Reject the request on summarization failure (default). + Closed, + /// Continue with full (uncompacted) history on failure. + Open, +} + +// ----------------------------------------------------------------------------- +// CompactFilterConfig (YAML deserialization) +// ----------------------------------------------------------------------------- + +#[expect( + dead_code, + reason = "scaffolding — fields used once build_config is implemented" +)] +/// 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, +} + +fn default_model() -> String { + "gpt-4o-mini".to_owned() +} + +fn default_tiktoken_encoding() -> String { + "cl100k_base".to_owned() +} + +// ----------------------------------------------------------------------------- +// ValidatedConfig (post-validation) +// ----------------------------------------------------------------------------- + +#[expect( + dead_code, + reason = "scaffolding — fields used once from_config and on_request_body are implemented" +)] +/// 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, + + /// Callout timeout in milliseconds. + pub timeout_ms: u64, + + /// Failure mode for the inference callout. + pub failure_mode: FailureMode, + + /// HTTP status on error. + pub status_on_error: u16, +} + +/// 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. +/// +/// Follow the pattern in `web_search/config.rs:build_config()`: +/// +/// - Validate `inference_url` is not empty +/// - Validate `timeout_ms` (default [`DEFAULT_TIMEOUT_MS`], reject 0) +/// - Validate `status_on_error` (default [`DEFAULT_STATUS_ON_ERROR`], +/// must be 100..=599) +/// - Validate `tiktoken_encoding` is a known encoding name +/// (use `tiktoken_rs` to check) +/// - Return `ValidatedConfig` with defaults applied +#[expect( + clippy::todo, + reason = "scaffolding — implement config validation" +)] +pub(super) fn build_config(raw: &CompactFilterConfig) -> Result { + let _ = (raw, DEFAULT_TIMEOUT_MS, DEFAULT_STATUS_ON_ERROR); + todo!("implement config validation") +} diff --git a/apis/src/openai/responses/compact/mod.rs b/apis/src/openai/responses/compact/mod.rs new file mode 100644 index 00000000..bf8555e7 --- /dev/null +++ b/apis/src/openai/responses/compact/mod.rs @@ -0,0 +1,403 @@ +// 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; + +#[expect( + unused_imports, + reason = "scaffolding — all imports needed once todo!()s are implemented" +)] +use { + async_trait::async_trait, + bytes::Bytes, + praxis_core::callout::{ + CalloutClient, CalloutConfig, CalloutRequest, CalloutResult, CircuitBreakerConfig, + FailureMode as CoreFailureMode, + }, + praxis_filter::{ + BodyAccess, BodyMode, FilterAction, FilterError, HttpFilter, HttpFilterContext, + body::MAX_JSON_BODY_BYTES, parse_filter_config, + }, + serde_json::Value, + tracing::{debug, warn}, + self::config::{CompactFilterConfig, FailureMode, ValidatedConfig, build_config}, + super::{error::responses_error_rejection, state::ResponsesState}, +}; + +// ----------------------------------------------------------------------------- +// Constants +// ----------------------------------------------------------------------------- + +/// Metadata key for previous response total token count. +/// Written by the rehydrate filter. +const PREV_USAGE_TOTAL_KEY: &str = "responses.previous_usage_total_tokens"; + +#[expect(dead_code, reason = "scaffolding — used once build_summarization_request is implemented")] +/// 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 +// ----------------------------------------------------------------------------- + +#[expect(dead_code, reason = "scaffolding — used once extract_compaction_config is implemented")] +/// 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 +/// ``` +#[expect( + dead_code, + reason = "scaffolding — fields used once from_config and on_request_body are implemented" +)] +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. + #[expect( + clippy::todo, + unused_variables, + reason = "scaffolding — implement CalloutClient construction" + )] + 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)?; + + // TODO: build CalloutConfig from validated config + // (follow the pattern in web_search/provider.rs:build_callout_config) + // + // Map FailureMode -> CoreFailureMode: + // FailureMode::Closed -> CoreFailureMode::Closed + // FailureMode::Open -> CoreFailureMode::Open + // + // Include CircuitBreakerConfig with: + // consecutive_failures: 5 + // recovery_window_ms: 30_000 + // + // Then construct Self { callout_client, config: validated } + todo!("construct CalloutClient and return CompactFilter") + } +} + +#[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) + } + + #[expect( + clippy::todo, + unused_variables, + reason = "scaffolding — implement the compaction flow" + )] + 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); + } + + // Skip non-Responses requests + if ctx.get_metadata("openai_responses_format.format") != Some("openai_responses") { + return Ok(FilterAction::Release); + } + + let streaming = ctx + .get_metadata("openai_responses_format.stream") + .is_some_and(|v| v == "true"); + + // TODO: implement the compaction flow: + // + // 1. Get ResponsesState from ctx.extensions + // - If absent, return Release (no state to compact) + // + // 2. Call extract_compaction_config() on state.context_management + // - If None, return Release (no compaction requested) + // + // 3. Call get_token_count() to get the current token count + // - Try previous_usage metadata first, fall back to tiktoken + // - If None (no count available), return Release + // + // 4. Compare token_count against params.compact_threshold + // - If token_count <= threshold, return Release + // + // 5. Build summarization request via build_summarization_request() + // - Use params.compaction_model or self.config.default_model + // - Include state.request_body["instructions"] if present + // + // 6. Execute via self.callout_client.execute(request).await + // - On Success: parse response, build compaction item, + // replace messages + // - On Failed (open mode): log warning, return Release + // - On Rejected: return Reject with responses_error_rejection() + // + // 7. Set metadata: "responses.compacted" = "true" + // + // 8. Return Release + todo!("implement compaction flow") + } +} + +// ----------------------------------------------------------------------------- +// Compaction Logic +// ----------------------------------------------------------------------------- + +/// 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. +#[expect( + clippy::todo, + dead_code, + unused_variables, + reason = "scaffolding — implement context_management parsing" +)] +fn extract_compaction_config(context_management: &Option) -> Option { + // TODO: iterate the array, find the first entry where + // type == "compaction", extract compact_threshold (u64) + // and optional compaction_model (String) + todo!("parse context_management array") +} + +/// Get the current token count. +/// +/// Tries `previous_usage` metadata first (exact, written by +/// rehydrate). Falls back to tiktoken estimation on +/// `state.messages` using the configured encoding. +#[expect( + clippy::todo, + dead_code, + unused_variables, + reason = "scaffolding — implement tiktoken fallback" +)] +fn get_token_count( + ctx: &HttpFilterContext<'_>, + messages: &[Value], + tiktoken_encoding: &str, +) -> Option { + // Try metadata from rehydrate first (exact count from the + // previous response's usage stats) + if let Some(total) = ctx.get_metadata(PREV_USAGE_TOTAL_KEY) { + if let Ok(count) = total.parse::() { + debug!(count, source = "previous_usage", "token count from metadata"); + return Some(count); + } + } + + // TODO: fall back to tiktoken estimation + // + // 1. Call build_conversation_text(messages) to serialize + // 2. Use tiktoken_rs to get a BPE encoder for tiktoken_encoding + // (e.g., tiktoken_rs::cl100k_base() for "cl100k_base") + // 3. Encode the text and return the token count as u64 + // 4. Log with source = "tiktoken_estimate" + // 5. Return None if encoding fails + todo!("tiktoken fallback estimation") +} + +/// Build a Chat Completions request for summarization. +/// +/// The request body has this shape: +/// ```json +/// { +/// "model": "", +/// "messages": [ +/// {"role": "system", "content": ""}, +/// {"role": "user", "content": ""} +/// ] +/// } +/// ``` +#[expect( + clippy::todo, + dead_code, + unused_variables, + reason = "scaffolding — implement request construction" +)] +fn build_summarization_request( + messages: &[Value], + instructions: Option<&str>, + model: &str, + inference_url: &str, +) -> CalloutRequest { + // TODO: + // 1. Build the system prompt: + // - If instructions is Some, prepend them before + // SUMMARIZATION_SYSTEM_PROMPT + // - Otherwise just use SUMMARIZATION_SYSTEM_PROMPT + // + // 2. Build user content via build_conversation_text(messages) + // + // 3. Construct the Chat Completions JSON body + // + // 4. Return a CalloutRequest with: + // - method: POST + // - url: inference_url.to_owned() + // - headers: Content-Type + Accept application/json + // - body: Some(serialized JSON bytes) + // - depth: 0 + todo!("build summarization CalloutRequest") +} + +/// Parse the Chat Completions response and extract the summary text. +/// +/// Expected shape: `{"choices": [{"message": {"content": "..."}}]}` +#[expect( + clippy::todo, + dead_code, + unused_variables, + reason = "scaffolding — implement response parsing" +)] +fn parse_summarization_response(body: &[u8]) -> Result { + // TODO: parse JSON, navigate to choices[0].message.content, + // return the string. Return Err with a descriptive + // message if parsing fails or the path doesn't exist. + todo!("parse Chat Completions response") +} + +/// Build the compaction output item. +/// +/// Returns: `{"type": "compaction", "encrypted_content": ""}` +/// +/// Note: `encrypted_content` is a misnomer from the OpenAI spec — +/// in the proxy context this is plain text. +#[expect( + clippy::todo, + dead_code, + unused_variables, + reason = "scaffolding — implement compaction item construction" +)] +fn build_compaction_item(summary: &str) -> Value { + // TODO: construct the JSON value + todo!("build compaction item JSON") +} + +/// 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. +#[expect( + clippy::todo, + dead_code, + unused_variables, + reason = "scaffolding — implement message replacement" +)] +fn replace_messages(state: &mut ResponsesState, compaction_item: Value) { + // TODO: replace state.messages and state.persisted_messages + // with [compaction_item] followed by state.input items + todo!("replace messages with compacted history") +} + +/// Format a message array as readable text for the summarization prompt. +/// +/// Each message becomes: `: ` +/// Messages are separated by blank lines. +#[expect( + clippy::todo, + dead_code, + unused_variables, + reason = "scaffolding — implement message serialization" +)] +fn build_conversation_text(messages: &[Value]) -> String { + // TODO: iterate messages, extract role and content from each, + // format as "role: content" separated by "\n\n" + // + // Content can be: + // - A string: use directly + // - An array of content parts: extract "text" from each + // input_text/output_text part and join them + // - Missing/null: skip the message + todo!("serialize messages to readable text") +} diff --git a/apis/src/openai/responses/compact/tests.rs b/apis/src/openai/responses/compact/tests.rs new file mode 100644 index 00000000..4388a9c6 --- /dev/null +++ b/apis/src/openai/responses/compact/tests.rs @@ -0,0 +1,206 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Praxis Contributors + +// Uncomment as you implement and enable tests: +// use serde_json::json; +// use super::*; + +// ============================================================================= +// Config tests +// ============================================================================= + +// TODO: implement these once build_config() is done + +// #[test] +// fn build_config_applies_defaults() { ... } + +// #[test] +// fn build_config_rejects_empty_inference_url() { ... } + +// #[test] +// fn build_config_rejects_zero_timeout() { ... } + +// #[test] +// fn build_config_rejects_invalid_status() { ... } + +// #[test] +// fn build_config_custom_values() { ... } + +// ============================================================================= +// extract_compaction_config tests +// ============================================================================= + +// TODO: uncomment once extract_compaction_config() is implemented + +// #[test] +// fn extract_compaction_config_with_compaction_entry() { +// let cm = Some(json!([{"type": "compaction", "compact_threshold": 50000}])); +// let params = extract_compaction_config(&cm); +// assert!(params.is_some()); +// let params = params.unwrap(); +// assert_eq!(params.compact_threshold, 50000); +// assert!(params.compaction_model.is_none()); +// } +// +// #[test] +// fn extract_compaction_config_with_model_override() { +// let cm = Some(json!([{ +// "type": "compaction", +// "compact_threshold": 100000, +// "compaction_model": "gpt-4o" +// }])); +// let params = extract_compaction_config(&cm).unwrap(); +// assert_eq!(params.compact_threshold, 100000); +// 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()); +// } + +// ============================================================================= +// build_compaction_item tests +// ============================================================================= + +// TODO: uncomment once build_compaction_item() is implemented + +// #[test] +// fn compaction_item_has_correct_shape() { +// let item = build_compaction_item("This is a summary."); +// assert_eq!(item["type"], "compaction"); +// assert_eq!(item["encrypted_content"], "This is a summary."); +// } + +// ============================================================================= +// parse_summarization_response tests +// ============================================================================= + +// TODO: uncomment once parse_summarization_response() is implemented + +// #[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()); +// } + +// ============================================================================= +// build_conversation_text tests +// ============================================================================= + +// TODO: uncomment once build_conversation_text() is implemented + +// #[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()); +// } + +// ============================================================================= +// replace_messages tests +// ============================================================================= + +// TODO: uncomment once replace_messages() and build_compaction_item() +// are implemented + +// #[test] +// fn replace_messages_preserves_current_input() { +// // Simulate state after rehydrate: +// // input = [current_user_msg] +// // messages = [old_msg1, old_msg2, current_user_msg] +// let mut state = ResponsesState::from_request_body(json!({ +// "model": "gpt-4o", +// "input": "What's next?" +// })); +// // Prepend some "rehydrated" history +// 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); +// +// // After replacement: [compaction_item, current_user_msg] +// assert_eq!( +// state.messages.len(), +// 2, +// "should have compaction + current input" +// ); +// assert_eq!(state.messages[0]["type"], "compaction"); +// assert_eq!(state.messages[1]["role"], "user"); +// assert_eq!(state.messages[1]["content"], "What's next?"); +// +// // persisted_messages should match +// assert_eq!(state.persisted_messages.len(), 2); +// assert_eq!(state.persisted_messages[0]["type"], "compaction"); +// } + +// ============================================================================= +// Filter-level tests +// ============================================================================= + +// TODO: implement once from_config() works + +// #[test] +// fn from_config_with_valid_yaml() { ... } + +// #[test] +// fn from_config_rejects_missing_inference_url() { ... } From 70b61877c643855076b4cf9cd4e1bc67b76f5ea1 Mon Sep 17 00:00:00 2001 From: Alexander Cristurean Date: Fri, 17 Jul 2026 17:43:33 +0200 Subject: [PATCH 03/11] fix: extracted common config with web search and config for compaction. Signed-off-by: Alexander Cristurean  Conflicts:  apis/src/openai/responses/mod.rs --- apis/src/openai/responses/compact/config.rs | 69 +++---- apis/src/openai/responses/compact/mod.rs | 33 +--- .../src/openai/responses/config_validation.rs | 186 ++++++++++++++++++ apis/src/openai/responses/mod.rs | 1 + .../src/openai/responses/web_search/config.rs | 78 ++------ .../openai/responses/web_search/provider.rs | 56 +++--- 6 files changed, 266 insertions(+), 157 deletions(-) create mode 100644 apis/src/openai/responses/config_validation.rs diff --git a/apis/src/openai/responses/compact/config.rs b/apis/src/openai/responses/compact/config.rs index 2e5b65a4..588efdbb 100644 --- a/apis/src/openai/responses/compact/config.rs +++ b/apis/src/openai/responses/compact/config.rs @@ -6,34 +6,18 @@ 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; -// ----------------------------------------------------------------------------- -// FailureMode -// ----------------------------------------------------------------------------- - -/// What happens when the summarization callout fails. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)] -#[serde(rename_all = "snake_case")] -pub(super) enum FailureMode { - /// Reject the request on summarization failure (default). - Closed, - /// Continue with full (uncompacted) history on failure. - Open, -} - // ----------------------------------------------------------------------------- // CompactFilterConfig (YAML deserialization) // ----------------------------------------------------------------------------- -#[expect( - dead_code, - reason = "scaffolding — fields used once build_config is implemented" -)] /// Raw YAML config, deserialized then validated. #[derive(Deserialize)] #[serde(deny_unknown_fields)] @@ -92,14 +76,8 @@ pub(super) struct ValidatedConfig { /// Tiktoken encoding name. pub tiktoken_encoding: String, - /// Callout timeout in milliseconds. - pub timeout_ms: u64, - - /// Failure mode for the inference callout. - pub failure_mode: FailureMode, - - /// HTTP status on error. - pub status_on_error: u16, + /// Shared callout settings (timeout, failure mode, status). + pub callout: CalloutSettings, } /// Validate raw config and apply defaults. @@ -108,21 +86,28 @@ pub(super) struct ValidatedConfig { /// /// Returns [`FilterError`] if `inference_url` is empty, /// `timeout_ms` is zero, or `status_on_error` is out of range. -/// -/// Follow the pattern in `web_search/config.rs:build_config()`: -/// -/// - Validate `inference_url` is not empty -/// - Validate `timeout_ms` (default [`DEFAULT_TIMEOUT_MS`], reject 0) -/// - Validate `status_on_error` (default [`DEFAULT_STATUS_ON_ERROR`], -/// must be 100..=599) -/// - Validate `tiktoken_encoding` is a known encoding name -/// (use `tiktoken_rs` to check) -/// - Return `ValidatedConfig` with defaults applied -#[expect( - clippy::todo, - reason = "scaffolding — implement config validation" -)] pub(super) fn build_config(raw: &CompactFilterConfig) -> Result { - let _ = (raw, DEFAULT_TIMEOUT_MS, DEFAULT_STATUS_ON_ERROR); - todo!("implement config validation") + 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.to_owned(), + default_model: raw.default_model.to_owned(), + tiktoken_encoding: raw.tiktoken_encoding.to_owned(), + 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 index bf8555e7..384767f4 100644 --- a/apis/src/openai/responses/compact/mod.rs +++ b/apis/src/openai/responses/compact/mod.rs @@ -30,17 +30,14 @@ mod tests; use { async_trait::async_trait, bytes::Bytes, - praxis_core::callout::{ - CalloutClient, CalloutConfig, CalloutRequest, CalloutResult, CircuitBreakerConfig, - FailureMode as CoreFailureMode, - }, + praxis_core::callout::{CalloutClient, CalloutRequest, CalloutResult}, praxis_filter::{ BodyAccess, BodyMode, FilterAction, FilterError, HttpFilter, HttpFilterContext, body::MAX_JSON_BODY_BYTES, parse_filter_config, }, serde_json::Value, tracing::{debug, warn}, - self::config::{CompactFilterConfig, FailureMode, ValidatedConfig, build_config}, + self::config::{CompactFilterConfig, ValidatedConfig, build_config}, super::{error::responses_error_rejection, state::ResponsesState}, }; @@ -121,28 +118,16 @@ impl CompactFilter { /// /// Returns [`FilterError`] if config validation fails or the /// callout client cannot be constructed. - #[expect( - clippy::todo, - unused_variables, - reason = "scaffolding — implement CalloutClient construction" - )] 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)?; - - // TODO: build CalloutConfig from validated config - // (follow the pattern in web_search/provider.rs:build_callout_config) - // - // Map FailureMode -> CoreFailureMode: - // FailureMode::Closed -> CoreFailureMode::Closed - // FailureMode::Open -> CoreFailureMode::Open - // - // Include CircuitBreakerConfig with: - // consecutive_failures: 5 - // recovery_window_ms: 30_000 - // - // Then construct Self { callout_client, config: validated } - todo!("construct CalloutClient and return CompactFilter") + 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, + })) } } diff --git a/apis/src/openai/responses/config_validation.rs b/apis/src/openai/responses/config_validation.rs new file mode 100644 index 00000000..51addb06 --- /dev/null +++ b/apis/src/openai/responses/config_validation.rs @@ -0,0 +1,186 @@ +// 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 d27cde23..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; diff --git a/apis/src/openai/responses/web_search/config.rs b/apis/src/openai/responses/web_search/config.rs index 9f65ad58..53c7a6c1 100644 --- a/apis/src/openai/responses/web_search/config.rs +++ b/apis/src/openai/responses/web_search/config.rs @@ -10,6 +10,9 @@ use praxis_filter::{ use secrecy::{ExposeSecret as _, SecretString}; use serde::Deserialize; +use crate::openai::responses::config_validation::{self, CalloutSettings, FailureMode}; + + /// Default callout timeout (10 seconds — search APIs can be slow). const DEFAULT_TIMEOUT_MS: u64 = 10_000; @@ -88,20 +91,6 @@ impl SearchContextSize { } } -// ----------------------------------------------------------------------------- -// FailureMode -// ----------------------------------------------------------------------------- - -/// What happens when a search callout fails. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)] -#[serde(rename_all = "snake_case")] -pub(crate) enum FailureMode { - /// Reject the request on search failure (default). - Closed, - /// Continue without search results on failure. - Open, -} - // ----------------------------------------------------------------------------- // WebSearchFilterConfig (YAML deserialization) // ----------------------------------------------------------------------------- @@ -155,17 +144,11 @@ pub(crate) struct ValidatedConfig { /// Default search context size. pub default_context_size: SearchContextSize, - /// Callout timeout in milliseconds. - pub timeout_ms: u64, - /// Maximum request body bytes to buffer. pub max_body_bytes: usize, - /// Failure mode for search callouts. - pub failure_mode: FailureMode, - - /// HTTP status on error. - pub status_on_error: u16, + /// Shared callout settings (timeout, failure mode, status). + pub callout: CalloutSettings, } impl std::fmt::Debug for ValidatedConfig { @@ -174,10 +157,8 @@ impl std::fmt::Debug for ValidatedConfig { .field("provider", &self.provider) .field("api_key", &"[REDACTED]") .field("default_context_size", &self.default_context_size) - .field("timeout_ms", &self.timeout_ms) .field("max_body_bytes", &self.max_body_bytes) - .field("failure_mode", &self.failure_mode) - .field("status_on_error", &self.status_on_error) + .field("callout", &self.callout) .finish() } } @@ -200,40 +181,23 @@ pub(super) fn build_config(raw: &WebSearchFilterConfig) -> 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 +270,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 +323,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..2ffd3301 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 crate::openai::responses::config_validation::FailureMode; +use super::config::{SearchContextSize, SearchProvider, ValidatedConfig}; // ----------------------------------------------------------------------------- // 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, }) } @@ -360,6 +341,7 @@ mod tests { use secrecy::SecretString; use serde_json::json; + use crate::openai::responses::config_validation::CalloutSettings; use super::*; #[test] @@ -508,10 +490,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 +507,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 +528,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"); From 68bc7389ef055bffca4ba659be39ff6d248ccdb9 Mon Sep 17 00:00:00 2001 From: Alexander Cristurean Date: Mon, 20 Jul 2026 17:50:46 +0200 Subject: [PATCH 04/11] feat: added token counting with tiktoken. Signed-off-by: Alexander Cristurean --- apis/src/openai/responses/compact/mod.rs | 174 ++++++++++-------- .../configs/openai/responses/compact.yaml | 71 +++++++ 2 files changed, 168 insertions(+), 77 deletions(-) create mode 100644 examples/configs/openai/responses/compact.yaml diff --git a/apis/src/openai/responses/compact/mod.rs b/apis/src/openai/responses/compact/mod.rs index 384767f4..28f0e135 100644 --- a/apis/src/openai/responses/compact/mod.rs +++ b/apis/src/openai/responses/compact/mod.rs @@ -45,10 +45,6 @@ use { // Constants // ----------------------------------------------------------------------------- -/// Metadata key for previous response total token count. -/// Written by the rehydrate filter. -const PREV_USAGE_TOTAL_KEY: &str = "responses.previous_usage_total_tokens"; - #[expect(dead_code, reason = "scaffolding — used once build_summarization_request is implemented")] /// System prompt for the summarization call. const SUMMARIZATION_SYSTEM_PROMPT: &str = "\ @@ -178,23 +174,32 @@ impl HttpFilter for CompactFilter { .get_metadata("openai_responses_format.stream") .is_some_and(|v| v == "true"); - // TODO: implement the compaction flow: - // - // 1. Get ResponsesState from ctx.extensions - // - If absent, return Release (no state to compact) - // - // 2. Call extract_compaction_config() on state.context_management - // - If None, return Release (no compaction requested) - // - // 3. Call get_token_count() to get the current token count - // - Try previous_usage metadata first, fall back to tiktoken - // - If None (no count available), return Release - // - // 4. Compare token_count against params.compact_threshold - // - If token_count <= threshold, return Release + let Some(state) = ctx.extensions.get::() else { + debug!("no ResponsesState in extensions, passthrough"); + return Ok(FilterAction::Release); + }; + + + let Some(compaction_config) = extract_compaction_config(&state.context_management) else { + debug!("no compaction config in context management"); + return Ok(FilterAction::Release); + }; + + let Some(token_count) = get_token_count(&state.messages, &self.config.tiktoken_encoding) else { + return Ok(FilterAction::Release); + }; + + if token_count <= compaction_config.compact_threshold { + debug!(token_count, threshold = compaction_config.compact_threshold, "under threshold, skipping compaction"); + return Ok(FilterAction::Release); + } + + debug!(token_count, threshold = compaction_config.compact_threshold, "threshold exceeded, compaction needed"); + + // TODO: implement remaining compaction flow: // // 5. Build summarization request via build_summarization_request() - // - Use params.compaction_model or self.config.default_model + // - Use compaction_config.compaction_model or self.config.default_model // - Include state.request_body["instructions"] if present // // 6. Execute via self.callout_client.execute(request).await @@ -206,7 +211,8 @@ impl HttpFilter for CompactFilter { // 7. Set metadata: "responses.compacted" = "true" // // 8. Return Release - todo!("implement compaction flow") + warn!("compaction not yet implemented, passing through"); + Ok(FilterAction::Release) } } @@ -220,53 +226,51 @@ impl HttpFilter for CompactFilter { /// `[{"type": "compaction", "compact_threshold": 50000}]` /// /// Returns `None` if no compaction entry is found. -#[expect( - clippy::todo, - dead_code, - unused_variables, - reason = "scaffolding — implement context_management parsing" -)] fn extract_compaction_config(context_management: &Option) -> Option { - // TODO: iterate the array, find the first entry where - // type == "compaction", extract compact_threshold (u64) - // and optional compaction_model (String) - todo!("parse context_management array") -} + let array = context_management.as_ref()?.as_array()?; -/// Get the current token count. -/// -/// Tries `previous_usage` metadata first (exact, written by -/// rehydrate). Falls back to tiktoken estimation on -/// `state.messages` using the configured encoding. -#[expect( - clippy::todo, - dead_code, - unused_variables, - reason = "scaffolding — implement tiktoken fallback" -)] -fn get_token_count( - ctx: &HttpFilterContext<'_>, - messages: &[Value], - tiktoken_encoding: &str, -) -> Option { - // Try metadata from rehydrate first (exact count from the - // previous response's usage stats) - if let Some(total) = ctx.get_metadata(PREV_USAGE_TOTAL_KEY) { - if let Ok(count) = total.parse::() { - debug!(count, source = "previous_usage", "token count from metadata"); - return Some(count); + 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(|v| v.as_u64()) + .unwrap_or(0); + let compaction_model = entry + .get("compaction_model") + .and_then(|v| v.as_str()) + .map(|s| s.to_owned()); + return Some(CompactionParams { + compact_threshold, + compaction_model, + }); } + None +} - // TODO: fall back to tiktoken estimation - // - // 1. Call build_conversation_text(messages) to serialize - // 2. Use tiktoken_rs to get a BPE encoder for tiktoken_encoding - // (e.g., tiktoken_rs::cl100k_base() for "cl100k_base") - // 3. Encode the text and return the token count as u64 - // 4. Log with source = "tiktoken_estimate" - // 5. Return None if encoding fails - todo!("tiktoken fallback estimation") +/// 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(messages: &[Value], 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 text = build_conversation_text(messages); + let count = bpe.encode_ordinary(&text).len() as u64; + debug!(count, source = "tiktoken", encoding = tiktoken_encoding, "token count estimated"); + Some(count) } /// Build a Chat Completions request for summarization. @@ -369,20 +373,36 @@ fn replace_messages(state: &mut ResponsesState, compaction_item: Value) { /// /// Each message becomes: `: ` /// Messages are separated by blank lines. -#[expect( - clippy::todo, - dead_code, - unused_variables, - reason = "scaffolding — implement message serialization" -)] fn build_conversation_text(messages: &[Value]) -> String { - // TODO: iterate messages, extract role and content from each, - // format as "role: content" separated by "\n\n" - // - // Content can be: - // - A string: use directly - // - An array of content parts: extract "text" from each - // input_text/output_text part and join them - // - Missing/null: skip the message - todo!("serialize messages to readable text") + let mut parts = Vec::new(); + 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; + } + parts.push(format!("{role}: {content}")); + } + parts.join("\n\n") +} + +/// 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. +fn extract_content(msg: &Value) -> String { + let Some(content) = msg.get("content") else { + return String::new(); + }; + if let Some(s) = content.as_str() { + return s.to_owned(); + } + 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 texts.join(" "); + } + String::new() } diff --git a/examples/configs/openai/responses/compact.yaml b/examples/configs/openai/responses/compact.yaml new file mode 100644 index 00000000..035e7cce --- /dev/null +++ b/examples/configs/openai/responses/compact.yaml @@ -0,0 +1,71 @@ +# 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 + + - 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 From 94aa6d252d3b0d40c73760e70880ad3b3241de78 Mon Sep 17 00:00:00 2001 From: Alexander Cristurean Date: Tue, 21 Jul 2026 13:54:05 +0200 Subject: [PATCH 05/11] feat: wrapped up compaction filter. Signed-off-by: Alexander Cristurean --- apis/src/openai/responses/compact/config.rs | 4 - apis/src/openai/responses/compact/mod.rs | 171 ++++++++++---------- 2 files changed, 90 insertions(+), 85 deletions(-) diff --git a/apis/src/openai/responses/compact/config.rs b/apis/src/openai/responses/compact/config.rs index 588efdbb..1d4600ec 100644 --- a/apis/src/openai/responses/compact/config.rs +++ b/apis/src/openai/responses/compact/config.rs @@ -61,10 +61,6 @@ fn default_tiktoken_encoding() -> String { // ValidatedConfig (post-validation) // ----------------------------------------------------------------------------- -#[expect( - dead_code, - reason = "scaffolding — fields used once from_config and on_request_body are implemented" -)] /// Validated configuration with defaults applied. pub(super) struct ValidatedConfig { /// URL of the inference backend for summarization calls. diff --git a/apis/src/openai/responses/compact/mod.rs b/apis/src/openai/responses/compact/mod.rs index 28f0e135..2906abca 100644 --- a/apis/src/openai/responses/compact/mod.rs +++ b/apis/src/openai/responses/compact/mod.rs @@ -23,10 +23,6 @@ pub(super) mod config; )] mod tests; -#[expect( - unused_imports, - reason = "scaffolding — all imports needed once todo!()s are implemented" -)] use { async_trait::async_trait, bytes::Bytes, @@ -45,7 +41,6 @@ use { // Constants // ----------------------------------------------------------------------------- -#[expect(dead_code, reason = "scaffolding — used once build_summarization_request is implemented")] /// System prompt for the summarization call. const SUMMARIZATION_SYSTEM_PROMPT: &str = "\ Summarize the following conversation concisely. \ @@ -58,7 +53,6 @@ capture everything needed to continue coherently."; // CompactionParams // ----------------------------------------------------------------------------- -#[expect(dead_code, reason = "scaffolding — used once extract_compaction_config is implemented")] /// Parsed compaction parameters from the request's `context_management`. struct CompactionParams { /// Token threshold above which compaction triggers. @@ -93,10 +87,6 @@ struct CompactionParams { /// failure_mode: closed /// status_on_error: 502 /// ``` -#[expect( - dead_code, - reason = "scaffolding — fields used once from_config and on_request_body are implemented" -)] pub struct CompactFilter { /// HTTP client for the summarization inference call. callout_client: CalloutClient, @@ -150,11 +140,7 @@ impl HttpFilter for CompactFilter { Ok(FilterAction::Continue) } - #[expect( - clippy::todo, - unused_variables, - reason = "scaffolding — implement the compaction flow" - )] + async fn on_request_body( &self, ctx: &mut HttpFilterContext<'_>, @@ -196,22 +182,46 @@ impl HttpFilter for CompactFilter { debug!(token_count, threshold = compaction_config.compact_threshold, "threshold exceeded, compaction needed"); - // TODO: implement remaining compaction flow: - // - // 5. Build summarization request via build_summarization_request() - // - Use compaction_config.compaction_model or self.config.default_model - // - Include state.request_body["instructions"] if present - // - // 6. Execute via self.callout_client.execute(request).await - // - On Success: parse response, build compaction item, - // replace messages - // - On Failed (open mode): log warning, return Release - // - On Rejected: return Reject with responses_error_rejection() - // - // 7. Set metadata: "responses.compacted" = "true" - // - // 8. Return Release - warn!("compaction not yet implemented, passing through"); + let model = compaction_config.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(&state.messages, instructions, model, &self.config.inference_url); + let response = match self.callout_client.execute(request).await { + CalloutResult::Success(response) => parse_summarization_response(&response.body), + CalloutResult::Failed => { + warn!("summarization callout failed, skipping compaction"); + return Ok(FilterAction::Release); + } + CalloutResult::Rejected(rejection) => { + warn!(status = rejection.status, "summarization callout rejected (closed failure mode)"); + return Ok(FilterAction::Reject(responses_error_rejection( + rejection.status, + "server_error", + "summarization callout rejected", + streaming, + ))); + } + }; + + let summary = match response { + Ok(s) => s, + Err(e) => { + warn!(error = %e, "failed to parse summarization response, skipping compaction"); + return Ok(FilterAction::Release); + } + }; + + let compaction_item = build_compaction_item(&summary); + + let state = ctx.extensions.get_mut::().expect("ResponsesState was present above"); + replace_messages(state, compaction_item); + + ctx.set_metadata("responses.compacted", "true"); Ok(FilterAction::Release) } } @@ -285,51 +295,59 @@ fn get_token_count(messages: &[Value], tiktoken_encoding: &str) -> Option { /// ] /// } /// ``` -#[expect( - clippy::todo, - dead_code, - unused_variables, - reason = "scaffolding — implement request construction" -)] fn build_summarization_request( messages: &[Value], instructions: Option<&str>, model: &str, inference_url: &str, ) -> CalloutRequest { - // TODO: - // 1. Build the system prompt: - // - If instructions is Some, prepend them before - // SUMMARIZATION_SYSTEM_PROMPT - // - Otherwise just use SUMMARIZATION_SYSTEM_PROMPT - // - // 2. Build user content via build_conversation_text(messages) - // - // 3. Construct the Chat Completions JSON body - // - // 4. Return a CalloutRequest with: - // - method: POST - // - url: inference_url.to_owned() - // - headers: Content-Type + Accept application/json - // - body: Some(serialized JSON bytes) - // - depth: 0 - todo!("build summarization CalloutRequest") + let system_content = match instructions { + Some(inst) => format!("{inst}\n\n{SUMMARIZATION_SYSTEM_PROMPT}"), + None => SUMMARIZATION_SYSTEM_PROMPT.to_owned(), + }; + + let conversation_text = build_conversation_text(messages); + + 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).expect("json! values always serialize"); + + 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": "..."}}]}` -#[expect( - clippy::todo, - dead_code, - unused_variables, - reason = "scaffolding — implement response parsing" -)] fn parse_summarization_response(body: &[u8]) -> Result { - // TODO: parse JSON, navigate to choices[0].message.content, - // return the string. Return Err with a descriptive - // message if parsing fails or the path doesn't exist. - todo!("parse Chat Completions response") + + match serde_json::from_slice::(body) { + Ok(body) => { + body.get("choices") + .and_then(Value::as_array) + .and_then(|choices| choices.get(0)) + .and_then(|choice| choice.get("message")) + .and_then(|msg| msg.get("content")) + .and_then(Value::as_str) + .map(|s| s.to_owned()) + .ok_or_else(|| "Chat Completions response missing choices[0].message.content".to_string()) + } + Err(err) => { Err(format!("failed to parse Chat Completions response JSON: {err}"))} + } } /// Build the compaction output item. @@ -338,15 +356,11 @@ fn parse_summarization_response(body: &[u8]) -> Result { /// /// Note: `encrypted_content` is a misnomer from the OpenAI spec — /// in the proxy context this is plain text. -#[expect( - clippy::todo, - dead_code, - unused_variables, - reason = "scaffolding — implement compaction item construction" -)] fn build_compaction_item(summary: &str) -> Value { - // TODO: construct the JSON value - todo!("build compaction item JSON") + serde_json::json!({ + "type": "compaction", + "encrypted_content": summary + }) } /// Replace conversation history with the compaction item. @@ -357,16 +371,11 @@ fn build_compaction_item(summary: &str) -> Value { /// /// `state.input` holds the current request's input items (unchanged /// by rehydrate), so the current turn's messages are preserved. -#[expect( - clippy::todo, - dead_code, - unused_variables, - reason = "scaffolding — implement message replacement" -)] fn replace_messages(state: &mut ResponsesState, compaction_item: Value) { - // TODO: replace state.messages and state.persisted_messages - // with [compaction_item] followed by state.input items - todo!("replace messages with compacted history") + let mut new_messages = vec![compaction_item.clone()]; + new_messages.extend(state.input.iter().cloned()); + state.messages = new_messages.clone(); + state.persisted_messages = new_messages; } /// Format a message array as readable text for the summarization prompt. From b0be5891a34d8359afbc49fe2d160908143a838b Mon Sep 17 00:00:00 2001 From: Alexander Cristurean Date: Tue, 21 Jul 2026 15:45:44 +0200 Subject: [PATCH 06/11] fix: cosmetic changes. Signed-off-by: Alexander Cristurean --- apis/src/openai/responses/compact/config.rs | 8 +- apis/src/openai/responses/compact/mod.rs | 131 ++++++++---------- .../configs/openai/responses/compact.yaml | 1 + 3 files changed, 66 insertions(+), 74 deletions(-) diff --git a/apis/src/openai/responses/compact/config.rs b/apis/src/openai/responses/compact/config.rs index 1d4600ec..81eb689e 100644 --- a/apis/src/openai/responses/compact/config.rs +++ b/apis/src/openai/responses/compact/config.rs @@ -49,10 +49,12 @@ pub(super) struct CompactFilterConfig { 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() } @@ -97,9 +99,9 @@ pub(super) fn build_config(raw: &CompactFilterConfig) -> Result 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(&state.messages, 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] @@ -147,80 +183,33 @@ impl HttpFilter for CompactFilter { _body: &mut Option, end_of_stream: bool, ) -> Result { - if !end_of_stream { - return Ok(FilterAction::Continue); - } - - // Skip non-Responses requests + if !end_of_stream { return Ok(FilterAction::Continue); } if ctx.get_metadata("openai_responses_format.format") != Some("openai_responses") { return Ok(FilterAction::Release); } - - let streaming = ctx - .get_metadata("openai_responses_format.stream") - .is_some_and(|v| v == "true"); - + let streaming = ctx.get_metadata("openai_responses_format.stream").is_some_and(|v| v == "true"); let Some(state) = ctx.extensions.get::() else { - debug!("no ResponsesState in extensions, passthrough"); return Ok(FilterAction::Release); }; - - - let Some(compaction_config) = extract_compaction_config(&state.context_management) else { - debug!("no compaction config in context management"); + let Some(params) = extract_compaction_config(&state.context_management) else { return Ok(FilterAction::Release); }; - let Some(token_count) = get_token_count(&state.messages, &self.config.tiktoken_encoding) else { return Ok(FilterAction::Release); }; - - if token_count <= compaction_config.compact_threshold { - debug!(token_count, threshold = compaction_config.compact_threshold, "under threshold, skipping compaction"); + if token_count <= params.compact_threshold { + debug!(token_count, threshold = params.compact_threshold, "under threshold, skipping"); return Ok(FilterAction::Release); } - - debug!(token_count, threshold = compaction_config.compact_threshold, "threshold exceeded, compaction needed"); - - let model = compaction_config.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(&state.messages, instructions, model, &self.config.inference_url); - let response = match self.callout_client.execute(request).await { - CalloutResult::Success(response) => parse_summarization_response(&response.body), - CalloutResult::Failed => { - warn!("summarization callout failed, skipping compaction"); - return Ok(FilterAction::Release); - } - CalloutResult::Rejected(rejection) => { - warn!(status = rejection.status, "summarization callout rejected (closed failure mode)"); - return Ok(FilterAction::Reject(responses_error_rejection( - rejection.status, - "server_error", - "summarization callout rejected", - streaming, - ))); - } + let summary = match self.execute_compaction(state, ¶ms, streaming).await { + Ok(Some(s)) => s, + Ok(None) | Err(FilterAction::Release) => return Ok(FilterAction::Release), + Err(action) => return Ok(action), }; - - let summary = match response { - Ok(s) => s, - Err(e) => { - warn!(error = %e, "failed to parse summarization response, skipping compaction"); - return Ok(FilterAction::Release); - } + let Some(state) = ctx.extensions.get_mut::() else { + return Ok(FilterAction::Release); }; - - let compaction_item = build_compaction_item(&summary); - - let state = ctx.extensions.get_mut::().expect("ResponsesState was present above"); - replace_messages(state, compaction_item); - + replace_messages(state, build_compaction_item(&summary)); ctx.set_metadata("responses.compacted", "true"); Ok(FilterAction::Release) } @@ -248,12 +237,12 @@ fn extract_compaction_config(context_management: &Option) -> Option Result { Ok(body) => { body.get("choices") .and_then(Value::as_array) - .and_then(|choices| choices.get(0)) + .and_then(|choices| choices.first()) .and_then(|choice| choice.get("message")) .and_then(|msg| msg.get("content")) .and_then(Value::as_str) - .map(|s| s.to_owned()) - .ok_or_else(|| "Chat Completions response missing choices[0].message.content".to_string()) + .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}"))} } @@ -358,8 +347,8 @@ fn parse_summarization_response(body: &[u8]) -> Result { /// in the proxy context this is plain text. fn build_compaction_item(summary: &str) -> Value { serde_json::json!({ - "type": "compaction", - "encrypted_content": summary + "role": "system", + "content": summary }) } @@ -372,10 +361,10 @@ fn build_compaction_item(summary: &str) -> Value { /// `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![compaction_item.clone()]; + let mut new_messages = vec![compaction_item]; new_messages.extend(state.input.iter().cloned()); - state.messages = new_messages.clone(); - state.persisted_messages = new_messages; + state.persisted_messages = new_messages.clone(); + state.messages = new_messages; } /// Format a message array as readable text for the summarization prompt. diff --git a/examples/configs/openai/responses/compact.yaml b/examples/configs/openai/responses/compact.yaml index 035e7cce..98c1eb73 100644 --- a/examples/configs/openai/responses/compact.yaml +++ b/examples/configs/openai/responses/compact.yaml @@ -53,6 +53,7 @@ filter_chains: - 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 From 18c419e8b1ed3353fa9a1e7df9c67c05b04c3ce6 Mon Sep 17 00:00:00 2001 From: Alexander Cristurean Date: Tue, 21 Jul 2026 17:14:52 +0200 Subject: [PATCH 07/11] fix: reduce allocations in compact filter hot path. Signed-off-by: Alexander Cristurean --- apis/src/openai/responses/compact/mod.rs | 50 +++++++++++++----------- 1 file changed, 28 insertions(+), 22 deletions(-) diff --git a/apis/src/openai/responses/compact/mod.rs b/apis/src/openai/responses/compact/mod.rs index 8cdefdbc..ef231a4a 100644 --- a/apis/src/openai/responses/compact/mod.rs +++ b/apis/src/openai/responses/compact/mod.rs @@ -24,6 +24,7 @@ pub(super) mod config; mod tests; use { + std::borrow::Cow, async_trait::async_trait, bytes::Bytes, praxis_core::callout::{CalloutClient, CalloutRequest, CalloutResult}, @@ -126,11 +127,12 @@ impl CompactFilter { 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(&state.messages, instructions, model, &self.config.inference_url); + 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) @@ -194,14 +196,15 @@ impl HttpFilter for CompactFilter { let Some(params) = extract_compaction_config(&state.context_management) else { return Ok(FilterAction::Release); }; - let Some(token_count) = get_token_count(&state.messages, &self.config.tiktoken_encoding) else { + let conversation_text = build_conversation_text(&state.messages); + let Some(token_count) = get_token_count(&conversation_text, &self.config.tiktoken_encoding) else { return Ok(FilterAction::Release); }; if token_count <= params.compact_threshold { debug!(token_count, threshold = params.compact_threshold, "under threshold, skipping"); return Ok(FilterAction::Release); } - let summary = match self.execute_compaction(state, ¶ms, streaming).await { + let summary = match self.execute_compaction(state, ¶ms, streaming, &conversation_text).await { Ok(Some(s)) => s, Ok(None) | Err(FilterAction::Release) => return Ok(FilterAction::Release), Err(action) => return Ok(action), @@ -257,7 +260,7 @@ fn extract_compaction_config(context_management: &Option) -> Option Option { +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(), @@ -266,8 +269,7 @@ fn get_token_count(messages: &[Value], tiktoken_encoding: &str) -> Option { return None; } }; - let text = build_conversation_text(messages); - let count = bpe.encode_ordinary(&text).len() as u64; + let count = bpe.count_ordinary(conversation_text) as u64; debug!(count, source = "tiktoken", encoding = tiktoken_encoding, "token count estimated"); Some(count) } @@ -285,7 +287,7 @@ fn get_token_count(messages: &[Value], tiktoken_encoding: &str) -> Option { /// } /// ``` fn build_summarization_request( - messages: &[Value], + conversation_text: &str, instructions: Option<&str>, model: &str, inference_url: &str, @@ -295,8 +297,6 @@ fn build_summarization_request( None => SUMMARIZATION_SYSTEM_PROMPT.to_owned(), }; - let conversation_text = build_conversation_text(messages); - let body = serde_json::json!({ "model": model, "messages": [ @@ -341,10 +341,7 @@ fn parse_summarization_response(body: &[u8]) -> Result { /// Build the compaction output item. /// -/// Returns: `{"type": "compaction", "encrypted_content": ""}` -/// -/// Note: `encrypted_content` is a misnomer from the OpenAI spec — -/// in the proxy context this is plain text. +/// Returns: `{"role": "system", "content": ""}` fn build_compaction_item(summary: &str) -> Value { serde_json::json!({ "role": "system", @@ -361,7 +358,8 @@ fn build_compaction_item(summary: &str) -> Value { /// `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![compaction_item]; + 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; @@ -372,35 +370,43 @@ fn replace_messages(state: &mut ResponsesState, compaction_item: Value) { /// Each message becomes: `: ` /// Messages are separated by blank lines. fn build_conversation_text(messages: &[Value]) -> String { - let mut parts = Vec::new(); + 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; } - parts.push(format!("{role}: {content}")); + if !buf.is_empty() { + buf.push_str("\n\n"); + } + buf.push_str(role); + buf.push_str(": "); + buf.push_str(&content); } - parts.join("\n\n") + 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. -fn extract_content(msg: &Value) -> String { +/// +/// 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 String::new(); + return Cow::Borrowed(""); }; if let Some(s) = content.as_str() { - return s.to_owned(); + 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 texts.join(" "); + return Cow::Owned(texts.join(" ")); } - String::new() + Cow::Borrowed("") } From 672dfe7006b41278b1678be755fc6c006603f878 Mon Sep 17 00:00:00 2001 From: Alexander Cristurean Date: Tue, 21 Jul 2026 17:24:21 +0200 Subject: [PATCH 08/11] test: added unit tests for functions used in compact filter. Signed-off-by: Alexander Cristurean --- apis/src/openai/responses/compact/tests.rs | 453 +++++++++++++-------- 1 file changed, 286 insertions(+), 167 deletions(-) diff --git a/apis/src/openai/responses/compact/tests.rs b/apis/src/openai/responses/compact/tests.rs index 4388a9c6..aebab23d 100644 --- a/apis/src/openai/responses/compact/tests.rs +++ b/apis/src/openai/responses/compact/tests.rs @@ -1,206 +1,325 @@ // SPDX-License-Identifier: MIT // Copyright (c) 2026 Praxis Contributors -// Uncomment as you implement and enable tests: -// use serde_json::json; -// use super::*; +use serde_json::json; + +use super::*; +use crate::openai::responses::config_validation::FailureMode; // ============================================================================= // Config tests // ============================================================================= -// TODO: implement these once build_config() is done +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() { ... } +#[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() { ... } +#[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() { ... } +#[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() { ... } +#[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() { ... } +#[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 // ============================================================================= -// TODO: uncomment once extract_compaction_config() is implemented - -// #[test] -// fn extract_compaction_config_with_compaction_entry() { -// let cm = Some(json!([{"type": "compaction", "compact_threshold": 50000}])); -// let params = extract_compaction_config(&cm); -// assert!(params.is_some()); -// let params = params.unwrap(); -// assert_eq!(params.compact_threshold, 50000); -// assert!(params.compaction_model.is_none()); -// } -// -// #[test] -// fn extract_compaction_config_with_model_override() { -// let cm = Some(json!([{ -// "type": "compaction", -// "compact_threshold": 100000, -// "compaction_model": "gpt-4o" -// }])); -// let params = extract_compaction_config(&cm).unwrap(); -// assert_eq!(params.compact_threshold, 100000); -// 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_with_compaction_entry() { + let cm = Some(json!([{"type": "compaction", "compact_threshold": 50000}])); + let params = extract_compaction_config(&cm); + assert!(params.is_some()); + let params = params.unwrap(); + assert_eq!(params.compact_threshold, 50000); + assert!(params.compaction_model.is_none()); +} + +#[test] +fn extract_compaction_config_with_model_override() { + let cm = Some(json!([{ + "type": "compaction", + "compact_threshold": 100000, + "compaction_model": "gpt-4o" + }])); + let params = extract_compaction_config(&cm).unwrap(); + assert_eq!(params.compact_threshold, 100000); + 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 // ============================================================================= -// TODO: uncomment once build_compaction_item() is implemented - -// #[test] -// fn compaction_item_has_correct_shape() { -// let item = build_compaction_item("This is a summary."); -// assert_eq!(item["type"], "compaction"); -// assert_eq!(item["encrypted_content"], "This is a summary."); -// } +#[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 // ============================================================================= -// TODO: uncomment once parse_summarization_response() is implemented - -// #[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_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 // ============================================================================= -// TODO: uncomment once build_conversation_text() is implemented +#[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 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 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 // ============================================================================= -// TODO: uncomment once replace_messages() and build_compaction_item() -// are implemented - -// #[test] -// fn replace_messages_preserves_current_input() { -// // Simulate state after rehydrate: -// // input = [current_user_msg] -// // messages = [old_msg1, old_msg2, current_user_msg] -// let mut state = ResponsesState::from_request_body(json!({ -// "model": "gpt-4o", -// "input": "What's next?" -// })); -// // Prepend some "rehydrated" history -// 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); -// -// // After replacement: [compaction_item, current_user_msg] -// assert_eq!( -// state.messages.len(), -// 2, -// "should have compaction + current input" -// ); -// assert_eq!(state.messages[0]["type"], "compaction"); -// assert_eq!(state.messages[1]["role"], "user"); -// assert_eq!(state.messages[1]["content"], "What's next?"); -// -// // persisted_messages should match -// assert_eq!(state.persisted_messages.len(), 2); -// assert_eq!(state.persisted_messages[0]["type"], "compaction"); -// } - -// ============================================================================= -// Filter-level tests -// ============================================================================= - -// TODO: implement once from_config() works - -// #[test] -// fn from_config_with_valid_yaml() { ... } - -// #[test] -// fn from_config_rejects_missing_inference_url() { ... } +#[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); +} \ No newline at end of file From a68d19933223f97855451440393acb74a864ffd1 Mon Sep 17 00:00:00 2001 From: Alexander Cristurean Date: Tue, 21 Jul 2026 17:43:57 +0200 Subject: [PATCH 09/11] test: added integration tests. Signed-off-by: Alexander Cristurean --- .../tests/suite/examples/compact.rs | 126 ++++++++++++++++++ tests/integration/tests/suite/examples/mod.rs | 1 + 2 files changed, 127 insertions(+) create mode 100644 tests/integration/tests/suite/examples/compact.rs diff --git a/tests/integration/tests/suite/examples/compact.rs b/tests/integration/tests/suite/examples/compact.rs new file mode 100644 index 00000000..0ec229fa --- /dev/null +++ b/tests/integration/tests/suite/examples/compact.rs @@ -0,0 +1,126 @@ +// 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..e2996482 100644 --- a/tests/integration/tests/suite/examples/mod.rs +++ b/tests/integration/tests/suite/examples/mod.rs @@ -7,6 +7,7 @@ mod test_utils; #[expect(unreachable_pub)] pub use test_utils::load_example_config; +mod compact; mod agentic_routing; mod anthropic_messages; mod credential_injection; From 9aa173301569b5f11d38ce128f28085336dca0a2 Mon Sep 17 00:00:00 2001 From: Alexander Cristurean Date: Tue, 21 Jul 2026 18:01:06 +0200 Subject: [PATCH 10/11] fix: cosmetic changes. Signed-off-by: Alexander Cristurean --- apis/src/openai/responses/compact/mod.rs | 156 +++++++++++------- apis/src/openai/responses/compact/tests.rs | 40 +++-- .../src/openai/responses/config_validation.rs | 31 +--- .../src/openai/responses/web_search/config.rs | 3 +- .../openai/responses/web_search/provider.rs | 4 +- docs/filters/openai_responses_compact.md | 39 +++++ docs/filters/reference.md | 1 + .../tests/suite/examples/compact.rs | 26 +-- tests/integration/tests/suite/examples/mod.rs | 2 +- 9 files changed, 180 insertions(+), 122 deletions(-) create mode 100644 docs/filters/openai_responses_compact.md diff --git a/apis/src/openai/responses/compact/mod.rs b/apis/src/openai/responses/compact/mod.rs index ef231a4a..a3f0983d 100644 --- a/apis/src/openai/responses/compact/mod.rs +++ b/apis/src/openai/responses/compact/mod.rs @@ -23,20 +23,20 @@ pub(super) mod config; )] mod tests; -use { - std::borrow::Cow, - async_trait::async_trait, - bytes::Bytes, - praxis_core::callout::{CalloutClient, CalloutRequest, CalloutResult}, - praxis_filter::{ - BodyAccess, BodyMode, FilterAction, FilterError, HttpFilter, HttpFilterContext, - body::MAX_JSON_BODY_BYTES, parse_filter_config, - }, - serde_json::Value, - tracing::{debug, warn}, - self::config::{CompactFilterConfig, ValidatedConfig, build_config}, - super::{error::responses_error_rejection, state::ResponsesState}, +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 @@ -109,8 +109,7 @@ impl CompactFilter { 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()))?; + let callout_client = CalloutClient::new(callout_config).map_err(|e| FilterError::from(e.to_string()))?; Ok(Box::new(Self { callout_client, config: validated, @@ -131,26 +130,26 @@ impl CompactFilter { ) -> 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); + 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::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, + rejection.status, + "server_error", + "summarization callout rejected", + streaming, ))) - } + }, } } } @@ -171,40 +170,31 @@ impl HttpFilter for CompactFilter { } } - async fn on_request( - &self, - _ctx: &mut HttpFilterContext<'_>, - ) -> Result { + 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 ctx.get_metadata("openai_responses_format.format") != Some("openai_responses") { + if !end_of_stream { + return Ok(FilterAction::Continue); + } + if !is_responses_request(ctx) { return Ok(FilterAction::Release); } - let streaming = ctx.get_metadata("openai_responses_format.stream").is_some_and(|v| v == "true"); + let streaming = is_streaming(ctx); let Some(state) = ctx.extensions.get::() else { return Ok(FilterAction::Release); }; - let Some(params) = extract_compaction_config(&state.context_management) else { + let Some((params, conversation_text)) = should_compact(state, &self.config.tiktoken_encoding) else { return Ok(FilterAction::Release); }; - let conversation_text = build_conversation_text(&state.messages); - let Some(token_count) = get_token_count(&conversation_text, &self.config.tiktoken_encoding) else { - return Ok(FilterAction::Release); - }; - if token_count <= params.compact_threshold { - debug!(token_count, threshold = params.compact_threshold, "under threshold, skipping"); - return Ok(FilterAction::Release); - } - let summary = match self.execute_compaction(state, ¶ms, streaming, &conversation_text).await { + 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), @@ -222,6 +212,41 @@ impl HttpFilter for CompactFilter { // 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: @@ -229,7 +254,7 @@ impl HttpFilter for CompactFilter { /// /// Returns `None` if no compaction entry is found. fn extract_compaction_config(context_management: &Option) -> Option { - let array = context_management.as_ref()?.as_array()?; + 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 { @@ -238,10 +263,7 @@ fn extract_compaction_config(context_management: &Option) -> Option Option { 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"); + debug!( + count, + source = "tiktoken", + encoding = tiktoken_encoding, + "token count estimated" + ); Some(count) } @@ -311,7 +338,10 @@ fn build_summarization_request( method: http::Method::POST, url: inference_url.to_owned(), headers: vec![ - (http::header::CONTENT_TYPE, http::HeaderValue::from_static("application/json")), + ( + http::header::CONTENT_TYPE, + http::HeaderValue::from_static("application/json"), + ), (http::header::ACCEPT, http::HeaderValue::from_static("application/json")), ], body: Some(body_bytes), @@ -323,19 +353,17 @@ fn build_summarization_request( /// /// 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}"))} + 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}")), } } diff --git a/apis/src/openai/responses/compact/tests.rs b/apis/src/openai/responses/compact/tests.rs index aebab23d..ee578460 100644 --- a/apis/src/openai/responses/compact/tests.rs +++ b/apis/src/openai/responses/compact/tests.rs @@ -71,11 +71,11 @@ fn build_config_custom_values() { #[test] fn extract_compaction_config_with_compaction_entry() { - let cm = Some(json!([{"type": "compaction", "compact_threshold": 50000}])); + 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, 50000); + assert_eq!(params.compact_threshold, 50_000); assert!(params.compaction_model.is_none()); } @@ -83,11 +83,11 @@ fn extract_compaction_config_with_compaction_entry() { fn extract_compaction_config_with_model_override() { let cm = Some(json!([{ "type": "compaction", - "compact_threshold": 100000, + "compact_threshold": 100_000, "compaction_model": "gpt-4o" }])); let params = extract_compaction_config(&cm).unwrap(); - assert_eq!(params.compact_threshold, 100000); + assert_eq!(params.compact_threshold, 100_000); assert_eq!(params.compaction_model.as_deref(), Some("gpt-4o")); } @@ -248,7 +248,12 @@ fn extract_content_null() { 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"); + 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()); @@ -266,7 +271,12 @@ fn summarization_request_without_instructions() { 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 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"); @@ -283,10 +293,18 @@ fn replace_messages_preserves_current_input() { "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"})); + 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); @@ -322,4 +340,4 @@ fn token_count_supports_o200k() { let count = get_token_count(&text, "o200k_base"); assert!(count.is_some()); assert!(count.unwrap() > 0); -} \ No newline at end of file +} diff --git a/apis/src/openai/responses/config_validation.rs b/apis/src/openai/responses/config_validation.rs index 51addb06..82f43352 100644 --- a/apis/src/openai/responses/config_validation.rs +++ b/apis/src/openai/responses/config_validation.rs @@ -65,11 +65,7 @@ impl CalloutSettings { /// # Errors /// /// Returns [`FilterError`] when the resolved value is zero. -pub(crate) fn validate_timeout_ms( - filter: &str, - raw: Option, - default: u64, -) -> Result { +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()); @@ -84,16 +80,10 @@ pub(crate) fn validate_timeout_ms( /// /// 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 { +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(), - ); + return Err(format!("{filter}: status_on_error must be between 100 and 599, got {value}").into()); } Ok(value) } @@ -111,10 +101,7 @@ mod tests { #[test] fn timeout_uses_provided_value() { - assert_eq!( - validate_timeout_ms("test", Some(10_000), 5000).unwrap(), - 10_000 - ); + assert_eq!(validate_timeout_ms("test", Some(10_000), 5000).unwrap(), 10_000); } #[test] @@ -137,18 +124,12 @@ mod tests { #[test] fn status_applies_default() { - assert_eq!( - validate_status_on_error("test", None, 502).unwrap(), - 502 - ); + 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 - ); + assert_eq!(validate_status_on_error("test", Some(503), 502).unwrap(), 503); } #[test] diff --git a/apis/src/openai/responses/web_search/config.rs b/apis/src/openai/responses/web_search/config.rs index 53c7a6c1..8312d52d 100644 --- a/apis/src/openai/responses/web_search/config.rs +++ b/apis/src/openai/responses/web_search/config.rs @@ -181,8 +181,7 @@ pub(super) fn build_config(raw: &WebSearchFilterConfig) -> Result + + +# `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/tests/integration/tests/suite/examples/compact.rs b/tests/integration/tests/suite/examples/compact.rs index 0ec229fa..518ab43a 100644 --- a/tests/integration/tests/suite/examples/compact.rs +++ b/tests/integration/tests/suite/examples/compact.rs @@ -9,8 +9,7 @@ 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, + Backend, TempSqlite, example_config_path, free_port, http_send, json_post, parse_status, patch_yaml, start_proxy, }; // ----------------------------------------------------------------------------- @@ -32,12 +31,7 @@ const CHAT_COMPLETIONS_RESPONSE: &str = r#"{"id":"chatcmpl-1","object":"chat.com /// 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 { +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}")); @@ -67,13 +61,14 @@ fn compact_passthrough() { let raw = http_send( proxy.addr(), - &json_post( - "/v1/responses", - r#"{"model":"gpt-4.1","input":"Hello"}"#, - ), + &json_post("/v1/responses", r#"{"model":"gpt-4.1","input":"Hello"}"#), ); - assert_eq!(parse_status(&raw), 200, "request without context_management should pass through"); + assert_eq!( + parse_status(&raw), + 200, + "request without context_management should pass through" + ); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -92,10 +87,7 @@ async fn compact_multi_turn_compaction() { let raw1 = http_send( proxy1.addr(), - &json_post( - "/v1/responses", - r#"{"model":"gpt-4.1","input":"Explain TCP vs UDP"}"#, - ), + &json_post("/v1/responses", r#"{"model":"gpt-4.1","input":"Explain TCP vs UDP"}"#), ); assert_eq!(parse_status(&raw1), 200, "first request should succeed"); diff --git a/tests/integration/tests/suite/examples/mod.rs b/tests/integration/tests/suite/examples/mod.rs index e2996482..67cc8e57 100644 --- a/tests/integration/tests/suite/examples/mod.rs +++ b/tests/integration/tests/suite/examples/mod.rs @@ -7,9 +7,9 @@ mod test_utils; #[expect(unreachable_pub)] pub use test_utils::load_example_config; -mod compact; mod agentic_routing; mod anthropic_messages; +mod compact; mod credential_injection; mod full_flow; mod mcp_broker; From 2f05a1cbc506bc1345f221e3cca262dc3a4bf9f5 Mon Sep 17 00:00:00 2001 From: Alexander Cristurean Date: Tue, 21 Jul 2026 19:15:15 +0200 Subject: [PATCH 11/11] fix: add CompactFilter re-export and fix web_search test after config refactor Signed-off-by: Alexander Cristurean --- apis/src/openai/mod.rs | 2 +- apis/src/openai/responses/web_search/provider.rs | 8 +++++--- 2 files changed, 6 insertions(+), 4 deletions(-) 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/web_search/provider.rs b/apis/src/openai/responses/web_search/provider.rs index 8ae769df..235f3507 100644 --- a/apis/src/openai/responses/web_search/provider.rs +++ b/apis/src/openai/responses/web_search/provider.rs @@ -433,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();