From 359559abc3f60f20ab8aed0931e996d4cbac05fc Mon Sep 17 00:00:00 2001 From: Alexander Cristurean Date: Mon, 7 Sep 2026 11:36:48 +0200 Subject: [PATCH 01/10] feat(compact): reconcile openai_responses_compact with main (#30) Signed-off-by: Alexander Cristurean --- apis/src/openai/responses/compact/config.rs | 13 + apis/src/openai/responses/compact/mod.rs | 541 +++++++++++++++--- apis/src/openai/responses/compact/tests.rs | 327 +++++++++-- .../responses/openai_responses_proxy/mod.rs | 6 +- apis/src/openai/responses/store/filter.rs | 17 +- .../src/openai/responses/store/input_items.rs | 8 +- apis/src/openai/responses/store/tests.rs | 35 ++ .../openai/translation/chat_completions.rs | 6 +- docs/filters/openai_responses_compact.md | 10 +- .../tests/suite/examples/compact.rs | 125 +++- 10 files changed, 944 insertions(+), 144 deletions(-) diff --git a/apis/src/openai/responses/compact/config.rs b/apis/src/openai/responses/compact/config.rs index 227962f1f4..9c915d44ac 100644 --- a/apis/src/openai/responses/compact/config.rs +++ b/apis/src/openai/responses/compact/config.rs @@ -52,6 +52,12 @@ pub(super) struct CompactFilterConfig { #[serde(default = "default_tiktoken_encoding")] pub tiktoken_encoding: String, + /// Prefix prepended to the summary when translating compaction + /// items to backend messages. Defaults to + /// `"[Previous conversation summary]\n\n"`. + #[serde(default)] + pub summary_prefix: Option, + /// Callout timeout in milliseconds. #[serde(default)] pub timeout_ms: Option, @@ -94,6 +100,9 @@ pub(super) struct ValidatedConfig { /// Tiktoken encoding name. pub tiktoken_encoding: String, + /// Prefix prepended to the summary in backend messages. + pub summary_prefix: String, + /// Shared callout settings (timeout, failure mode, status). pub callout: CalloutSettings, } @@ -143,6 +152,10 @@ pub(super) fn build_config(raw: &CompactFilterConfig) -> Result Result { + let conversation_text = build_conversation_text(messages); + let model = req.model.as_deref().unwrap_or(&self.config.default_model); + let instructions = req.instructions.as_deref(); + let request = build_summarization_request(&conversation_text, instructions, model); + let timeout = Duration::from_millis(self.config.callout.timeout_ms); + let result = subrequest::execute_url( + &self.client, + &self.config.inference_url, + request, + MAX_SUMMARIZATION_RESPONSE_BYTES, + timeout, + self.config.address_policy, + ) + .await; + match self.handle_subrequest_result(result, false) { + Ok(Some(s)) => Ok(s), + Ok(None) => Err(FilterAction::Reject(responses_error_rejection( + 502, + "server_error", + "compaction callout failed", + false, + ))), + Err(action) => Err(action), + } + } + /// Apply the configured open/closed policy on a callout error. fn on_callout_error(&self, message: &str, streaming: bool) -> Result, FilterAction> { match self.config.callout.on_failure { @@ -235,6 +281,103 @@ impl CompactFilter { ))), } } + + /// Apply compaction results: replace messages and persist the + /// compaction response to the store. + async fn apply_compaction(&self, ctx: &mut HttpFilterContext<'_>, summary: &str, model: &str) { + let compaction_id = format!("compact_{}", ctx.id_generator.generate(ctx.time_source)); + let resp_id = format!("resp_{}", ctx.id_generator.generate(ctx.time_source)); + let created_at = i64::try_from(ctx.time_source.now().as_secs()).unwrap_or(i64::MAX); + let store = ctx + .extensions + .get::() + .and_then(|r| r.get("default")); + let tenant_id = ctx.get_metadata("responses.tenant_id").unwrap_or("default").to_owned(); + + // Mutate state and take an owned snapshot of the compacted messages, + // releasing the `&mut ResponsesState` borrow before the store await. + let persisted_messages = { + let Some(state) = ctx.extensions.get_mut::() else { + warn!("ResponsesState missing in apply_compaction"); + return; + }; + replace_messages( + state, + build_compaction_item(&compaction_id, summary, &self.config.summary_prefix), + ); + state.persisted_messages.clone() + }; + if let Some(store) = store.as_deref() { + persist_compaction_response(store, &resp_id, model, &tenant_id, created_at, persisted_messages).await; + } + } + + /// Check threshold and run summarization if exceeded. + /// + /// Returns `Ok(Some((summary, model)))` when compaction ran, + /// `Ok(None)` when skipped, or `Err(FilterAction)` to + /// short-circuit. + async fn check_and_summarize( + &self, + state: &ResponsesState, + streaming: bool, + ) -> Result, FilterAction> { + let (params, conversation_text) = match should_compact(state, &self.config.tiktoken_encoding) { + Ok(Some(pair)) => pair, + Ok(None) => return Ok(None), + Err(msg) => { + let rej = responses_error_rejection(400, "invalid_request_error", &msg, streaming); + return Err(FilterAction::Reject(rej)); + }, + }; + let summary = match self + .execute_compaction(state, ¶ms, streaming, &conversation_text) + .await + { + Ok(Some(s)) => s, + Ok(None) => return Ok(None), + Err(action) => return Err(action), + }; + let model = params + .compaction_model + .unwrap_or_else(|| self.config.default_model.clone()); + Ok(Some((summary, model))) + } + + /// Handle an explicit `POST /v1/responses/compact` request. + /// + /// Loads a stored conversation by `response_id`, compacts it via + /// a summarization callout, stores the compacted result, and + /// returns the new response. + async fn handle_explicit_compact( + &self, + ctx: &HttpFilterContext<'_>, + body: &Option, + ) -> Result { + match self.do_explicit_compact(ctx, body).await { + Ok(action) | Err(action) => Ok(action), + } + } + + /// Inner logic for explicit compact, using `FilterAction` as the error type. + async fn do_explicit_compact( + &self, + ctx: &HttpFilterContext<'_>, + body: &Option, + ) -> Result { + let req = parse_compact_request_body(body)?; + let (store, tenant_id) = resolve_store_and_tenant(ctx)?; + let record = fetch_response(&*store, &tenant_id, &req).await?; + let messages = extract_stored_messages(record)?; + let summary = self.summarize_messages(&req, &messages).await?; + let response_object = build_and_persist_compaction(self, ctx, &*store, &tenant_id, &req, &summary).await?; + let body_bytes = serde_json::to_vec(&response_object).unwrap_or_default(); + Ok(FilterAction::Reject( + praxis_filter::Rejection::status(200) + .with_header("content-type", "application/json") + .with_body(Bytes::from(body_bytes)), + )) + } } #[async_trait] @@ -257,48 +400,35 @@ impl HttpFilter for CompactFilter { Ok(FilterAction::Continue) } - #[expect( - clippy::too_many_lines, - reason = "filter request body handling is inherently sequential" - )] async fn on_request_body( &self, ctx: &mut HttpFilterContext<'_>, - _body: &mut Option, + body: &mut Option, end_of_stream: bool, ) -> Result { if !end_of_stream { return Ok(FilterAction::Continue); } + if is_explicit_compact_request(ctx) { + return self.handle_explicit_compact(ctx, body).await; + } if !is_responses_request(ctx) { return Ok(FilterAction::Release); } let streaming = is_streaming(ctx); + if !ensure_compactable_state(ctx) { + return Ok(FilterAction::Release); + } let Some(state) = ctx.extensions.get::() else { + warn!("ResponsesState missing after ensure_compactable_state"); return Ok(FilterAction::Release); }; - let (params, conversation_text) = match should_compact(state, &self.config.tiktoken_encoding) { - Ok(Some(pair)) => pair, - Ok(None) => return Ok(FilterAction::Release), - Err(msg) => { - let rej = responses_error_rejection(400, "invalid_request_error", &msg, streaming); - return Ok(FilterAction::Reject(rej)); - }, - }; - if !state.history_rehydrated { - return Ok(FilterAction::Release); - } - let compaction = self.execute_compaction(state, ¶ms, streaming, &conversation_text); - let summary = match compaction.await { - Ok(Some(s)) => s, + let (summary, model) = match self.check_and_summarize(state, streaming).await { + Ok(Some(result)) => result, Ok(None) | Err(FilterAction::Release) => return Ok(FilterAction::Release), Err(action) => return Ok(action), }; - let Some(state) = ctx.extensions.get_mut::() else { - return Ok(FilterAction::Release); - }; - let compaction_id = format!("compact_{}", ctx.id_generator.generate(ctx.time_source)); - replace_messages(state, build_compaction_item(&compaction_id, &summary)); + self.apply_compaction(ctx, &summary, &model).await; ctx.set_metadata("responses.compacted", "true"); Ok(FilterAction::Release) } @@ -308,15 +438,46 @@ impl HttpFilter for CompactFilter { // Compaction Logic // ----------------------------------------------------------------------------- +/// Returns `true` when compaction should proceed. +fn ensure_compactable_state(ctx: &HttpFilterContext<'_>) -> bool { + is_compactable(ctx.extensions.get::()) +} + +/// Check whether the given state qualifies for reactive compaction. +/// +/// Returns `true` only when rehydrated history is present. Direct +/// input requests (no `previous_response_id`) are skipped because +/// `state.input == state.messages` — there is no separable "current +/// turn" to preserve after summarization. Use the explicit +/// `POST /v1/responses/compact` endpoint for non-rehydrated history. +/// +/// This gate is the single decision point for direct-input handling: +/// `should_compact` and `replace_messages` retain direct-input branches +/// (and unit tests for them) documenting the lower-level contract, but +/// those branches are unreachable through the filter while this returns +/// `false` for non-rehydrated state. +fn is_compactable(state: Option<&ResponsesState>) -> bool { + let Some(state) = state else { + return false; + }; + state.history_rehydrated +} + /// 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. /// -/// The token estimate includes instructions and tool definitions in +/// When `previous_usage` is available from the rehydrated response, +/// its `total_tokens` is used directly — avoiding the cost of BPE +/// tokenization. Falls back to tiktoken estimation otherwise. The +/// fallback estimate includes instructions and tool definitions in /// addition to conversation messages, since all three contribute to /// the rendered context sent to the model. -#[expect(clippy::too_many_lines, reason = "token threshold estimation is sequential")] +/// +/// The check is reactive: the token count reflects the *previous* +/// turn's usage, not the current one. If the previous turn exceeded +/// the threshold, we compact before sending this turn. fn should_compact( state: &ResponsesState, tiktoken_encoding: &str, @@ -324,54 +485,65 @@ fn should_compact( let Some(params) = extract_compaction_config(&state.context_management)? else { return Ok(None); }; - - let conversation_text = build_conversation_text(&state.messages); - let Some(message_tokens) = get_token_count(&conversation_text, tiktoken_encoding) else { - return Ok(None); + let history = if state.history_rehydrated { + let end = state.messages.len().saturating_sub(state.input.len()); + state.messages.get(..end).unwrap_or(&state.messages) + } else { + &state.messages }; - let overhead_text = build_context_overhead_text(state); - let overhead_tokens = if overhead_text.is_empty() { - 0 - } else { - get_token_count(&overhead_text, tiktoken_encoding).unwrap_or(0) + if let Some(token_count) = previous_usage_total(state) { + if !exceeds_threshold(token_count, ¶ms) { + return Ok(None); + } + return Ok(Some((params, build_conversation_text(history)))); + } + + debug!("previous_usage unavailable, falling back to tiktoken estimation"); + let conversation_text = build_conversation_text(history); + let overhead = build_context_overhead_text(&state.request_body); + let full_text = format!("{conversation_text}\n\n{overhead}"); + let Some(token_count) = get_token_count(&full_text, tiktoken_encoding) else { + return Ok(None); }; + if !exceeds_threshold(token_count, ¶ms) { + return Ok(None); + } + Ok(Some((params, conversation_text))) +} - let token_count = message_tokens + overhead_tokens; +/// Log and return whether `token_count` exceeds the compaction threshold. +fn exceeds_threshold(token_count: u64, params: &CompactionParams) -> bool { if token_count <= params.compact_threshold { debug!( token_count, - message_tokens, - overhead_tokens, threshold = params.compact_threshold, "under threshold, skipping" ); - return Ok(None); + return false; } debug!( token_count, - message_tokens, - overhead_tokens, threshold = params.compact_threshold, "threshold exceeded, compacting" ); - Ok(Some((params, conversation_text))) + true } -/// Build the text for instructions and tool definitions that live -/// outside the message list but still consume context window tokens. -fn build_context_overhead_text(state: &ResponsesState) -> String { - let mut buf = String::new(); - if let Some(instructions) = state.request_body.get("instructions").and_then(Value::as_str) { - buf.push_str(instructions); - } - for tool in &state.tools { - if !buf.is_empty() { - buf.push('\n'); - } - buf.push_str(&tool.to_string()); - } - buf +/// Extract `total_tokens` from the previous response's usage object. +fn previous_usage_total(state: &ResponsesState) -> Option { + let total = state.previous_usage.as_ref()?.get("total_tokens")?.as_u64()?; + debug!( + count = total, + source = "previous_usage", + "token count from prior response" + ); + Some(total) +} + +/// Check whether this is an explicit `POST /v1/responses/compact` request. +pub(super) fn is_explicit_compact_request(ctx: &HttpFilterContext<'_>) -> bool { + ctx.request.method == http::Method::POST && ctx.request.uri.path().trim_end_matches('/') == "/v1/responses/compact" } /// Check whether this is an OpenAI Responses API request. @@ -385,6 +557,138 @@ fn is_streaming(ctx: &HttpFilterContext<'_>) -> bool { .is_some_and(|v| v == "true") } +// ----------------------------------------------------------------------------- +// Explicit Compact Endpoint Helpers +// ----------------------------------------------------------------------------- + +/// Parsed body for `POST /v1/responses/compact`. +struct ExplicitCompactRequest { + /// The stored response to compact. + response_id: String, + /// Optional model override for the summarization call. + model: Option, + /// Optional instructions to prepend to the summarization prompt. + instructions: Option, +} + +/// Parse and validate the `POST /v1/responses/compact` body. +fn parse_compact_request_body(body: &Option) -> Result { + let bytes = body + .as_ref() + .filter(|b| !b.is_empty()) + .ok_or_else(|| reject_compact(400, "invalid_request_error", "request body is empty"))?; + let parsed: Value = serde_json::from_slice(bytes).map_err(|e| { + debug!(error = %e, "compact request body parse failed"); + reject_compact(400, "invalid_request_error", "invalid JSON body") + })?; + let response_id = parsed + .get("response_id") + .and_then(Value::as_str) + .filter(|s| !s.is_empty()) + .ok_or_else(|| reject_compact(400, "invalid_request_error", "missing required field: response_id"))? + .to_owned(); + Ok(ExplicitCompactRequest { + response_id, + model: parsed.get("model").and_then(Value::as_str).map(ToOwned::to_owned), + instructions: parsed + .get("instructions") + .and_then(Value::as_str) + .map(ToOwned::to_owned), + }) +} + +/// Look up the store and tenant from the request context. +fn resolve_store_and_tenant( + ctx: &HttpFilterContext<'_>, +) -> Result<(std::sync::Arc, String), FilterAction> { + let store = ctx + .extensions + .get::() + .and_then(|r| r.get("default")) + .ok_or_else(|| reject_compact(500, "server_error", "response store not available"))?; + let tenant_id = ctx.get_metadata("responses.tenant_id").unwrap_or("default").to_owned(); + Ok((store, tenant_id)) +} + +/// Fetch a stored response. +async fn fetch_response( + store: &dyn crate::store::ResponseStore, + tenant_id: &str, + req: &ExplicitCompactRequest, +) -> Result { + match store.get_response(tenant_id, &req.response_id).await { + Ok(Some(r)) => Ok(r), + Ok(None) => Err(reject_compact(404, "not_found_error", "response not found")), + Err(e) => { + warn!(error = %e, "failed to fetch response for compact"); + Err(reject_compact(500, "server_error", "failed to fetch response")) + }, + } +} + +/// Extract messages from a stored response. +fn extract_stored_messages(record: ResponseRecord) -> Result, FilterAction> { + match record.messages { + Value::Array(arr) if !arr.is_empty() => Ok(arr), + _ => Err(reject_compact( + 400, + "invalid_request_error", + "response has no messages to compact", + )), + } +} + +/// Build and persist the compaction result for an explicit compact request. +#[expect( + clippy::too_many_arguments, + clippy::too_many_lines, + reason = "all parameters are needed" +)] +async fn build_and_persist_compaction( + filter: &CompactFilter, + ctx: &HttpFilterContext<'_>, + store: &dyn crate::store::ResponseStore, + tenant_id: &str, + req: &ExplicitCompactRequest, + summary: &str, +) -> Result { + let compaction_id = format!("compact_{}", ctx.id_generator.generate(ctx.time_source)); + let resp_id = format!("resp_{}", ctx.id_generator.generate(ctx.time_source)); + let created_at = i64::try_from(ctx.time_source.now().as_secs()).unwrap_or(i64::MAX); + let model = req.model.as_deref().unwrap_or(&filter.config.default_model); + let compaction_item = build_compaction_item(&compaction_id, summary, &filter.config.summary_prefix); + let compacted_messages = Value::Array(vec![compaction_item]); + let response_object = serde_json::json!({ + "id": resp_id, + "object": "response", + "status": "completed", + "model": model, + "created_at": created_at, + "previous_response_id": req.response_id, + "output": compacted_messages, + }); + + let record = ResponseRecord { + id: resp_id, + tenant_id: tenant_id.to_owned(), + created_at, + model: model.to_owned(), + response_object: response_object.clone(), + input: compacted_messages.clone(), + messages: compacted_messages, + }; + store.upsert_response(&record).await.map_err(|e| { + warn!(error = %e, "failed to persist explicit compaction response"); + reject_compact(500, "server_error", "failed to persist compaction response") + })?; + Ok(response_object) +} + +/// Build a `FilterAction::Reject` for an explicit compact error. +fn reject_compact(status: u16, code: &str, message: &str) -> FilterAction { + FilterAction::Reject(responses_error_rejection(status, code, message, false)) +} + /// Parse the `context_management` JSON to find a compaction config. /// /// The `context_management` field is an array like: @@ -533,36 +837,107 @@ fn parse_summarization_response(body: &[u8]) -> Result { /// Build the compaction output item. /// /// Returns: `{"type": "compaction", "id": "", "encrypted_content": ""}` +/// with an optional `"summary_prefix"` when it differs from the default. /// /// The summary is base64-encoded into `encrypted_content` to match the /// OpenAI Responses API compaction item shape and make the content opaque /// to clients. -fn build_compaction_item(id: &str, summary: &str) -> Value { +fn build_compaction_item(id: &str, summary: &str, summary_prefix: &str) -> Value { let encrypted_content = base64::engine::general_purpose::STANDARD.encode(summary); - serde_json::json!({ + let mut item = serde_json::json!({ "type": "compaction", "id": id, "encrypted_content": encrypted_content - }) + }); + if summary_prefix != DEFAULT_SUMMARY_PREFIX + && let Some(obj) = item.as_object_mut() + { + obj.insert("summary_prefix".to_owned(), Value::String(summary_prefix.to_owned())); + } + item } /// Replace conversation history with the compaction item. /// -/// After replacement: -/// - `state.messages` = `[compaction_item, ...state.input]` -/// - `state.persisted_messages` = `[compaction_item, ...state.input]` +/// In the **rehydrated** path `state.input` holds only the current +/// turn's messages (rehydrate prepends history to `state.messages` +/// but leaves `input` untouched). Result: +/// `[compaction_item, ...current_turn_input]`. /// -/// The compaction item is `{"type": "compaction", "encrypted_content": ""}`. -/// `state.input` holds the current request's input items (unchanged -/// by rehydrate), so the current turn's messages are preserved. +/// In the **direct input** path `state.input == state.messages` +/// (no rehydration). Preserving `input` would duplicate the entire +/// conversation after the summary. Result: `[compaction_item]`. fn replace_messages(state: &mut ResponsesState, compaction_item: Value) { - let mut new_messages = Vec::with_capacity(state.input.len() + 1); - new_messages.push(compaction_item); - new_messages.extend(state.input.iter().cloned()); - state.persisted_messages = new_messages.clone(); + let direct_input = !state.history_rehydrated; + let new_messages = if direct_input { + vec![compaction_item] + } else { + let mut msgs = Vec::with_capacity(state.input.len() + 1); + msgs.push(compaction_item); + msgs.extend(state.input.iter().cloned()); + msgs + }; + state.persisted_messages.clone_from(&new_messages); state.messages = new_messages; } +/// Persist a hidden compaction response to the store so that it can +/// be referenced via `previous_response_id` in future requests. +/// +/// Best-effort: a store failure is logged but does not block the +/// request — the main response's store filter will still persist +/// the compacted messages as part of the regular response record. +#[expect(clippy::too_many_arguments, reason = "all fields are needed for the record")] +async fn persist_compaction_response( + store: &dyn crate::store::ResponseStore, + response_id: &str, + model: &str, + tenant_id: &str, + created_at: i64, + persisted_messages: Vec, +) { + let persisted = Value::Array(persisted_messages); + let record = ResponseRecord { + id: response_id.to_owned(), + tenant_id: tenant_id.to_owned(), + created_at, + model: model.to_owned(), + response_object: serde_json::json!({ + "id": response_id, + "object": "response", + "status": "completed", + "model": model, + "created_at": created_at, + }), + input: persisted.clone(), + messages: persisted, + }; + if let Err(e) = store.upsert_response(&record).await { + warn!(error = %e, id = %response_id, "failed to persist compaction response"); + } +} + +/// Build a text representation of instructions and tool definitions for token counting. +/// +/// Returns an empty string when neither field is present. The result is +/// concatenated with conversation text so tiktoken counts the full context +/// window overhead, matching the behavior of `previous_usage.total_tokens`. +fn build_context_overhead_text(request_body: &Value) -> String { + let mut buf = String::new(); + if let Some(instructions) = request_body.get("instructions").and_then(Value::as_str) + && !instructions.is_empty() + { + append_line(&mut buf, "instructions", instructions); + } + if let Some(tools) = request_body.get("tools").and_then(Value::as_array) + && !tools.is_empty() + { + let serialized = serde_json::to_string(tools).unwrap_or_default(); + append_line(&mut buf, "tools", &serialized); + } + buf +} + /// Format a message array as readable text for the summarization prompt. /// /// Each message becomes `