diff --git a/README.md b/README.md index 2f804363d..e3ef2857b 100644 --- a/README.md +++ b/README.md @@ -152,7 +152,7 @@ Aether Tunnel 是配套的正向代理节点，部署在海外 VPS 上，为墙 - `AETHER_GATEWAY_SECURITY_CACHE_TTL_MS`：IP 黑白名单本地缓存时间，默认 `1000ms`，写操作会主动失效相关缓存 - `AETHER_MAX_REDACTED_SYNC_RESPONSE_BODY_MB`：可选的 PII 恢复同步响应缓冲上限；未配置或设为 `0` 时不限制 - `REDIS_URL`：Redis 连接串；仅 Postgres + Redis 的 Docker Compose 部署需要配置 -- `AETHER_RUNTIME_BACKEND=memory|redis`：运行时缓存/协调后端。SQLite 默认用 `memory`，不会连接 Redis +- `AETHER_RUNTIME_BACKEND=memory|redis`：运行时缓存/协调后端。SQLite 默认用 `memory`，不会连接 Redis；多节点部署和需要跨 gateway 重启恢复 OpenAI Responses continuation history 的部署必须使用共享 Redis - `AETHER_GATEWAY_AUTO_PREPARE_DATABASE`：常规启动前自动执行挂起的 schema migration 和 backfill；仓库自带的 `docker-compose.yml` 默认开启 - `JWT_SECRET_KEY` / `ENCRYPTION_KEY`：认证和敏感数据加密所需密钥 - `API_KEY_PREFIX`：用户和管理员新建 API Key 时使用的前缀，默认 `sk` diff --git a/apps/aether-gateway/src/ai_serving/finalize/internal/stream_rewrite.rs b/apps/aether-gateway/src/ai_serving/finalize/internal/stream_rewrite.rs index a4a06bd63..bad26766f 100644 --- a/apps/aether-gateway/src/ai_serving/finalize/internal/stream_rewrite.rs +++ b/apps/aether-gateway/src/ai_serving/finalize/internal/stream_rewrite.rs @@ -1,5 +1,7 @@ use serde_json::Value; +use aether_ai_formats::api::ResponseHistoryRecord; + use crate::ai_serving::{ maybe_build_ai_surface_stream_rewriter, AiSurfaceFinalizeError, AiSurfaceStreamRewriter, }; @@ -24,6 +26,10 @@ impl LocalStreamRewriter<'_> { pub(crate) fn finish(&mut self) -> Result, GatewayError> { self.inner.finish().map_err(map_surface_error) } + + pub(crate) fn take_response_history_record(&mut self) -> Option { + self.inner.take_response_history_record() + } } fn map_surface_error(error: AiSurfaceFinalizeError) -> GatewayError { diff --git a/apps/aether-gateway/src/ai_serving/mod.rs b/apps/aether-gateway/src/ai_serving/mod.rs index 9038c71ae..8dfe53dc1 100644 --- a/apps/aether-gateway/src/ai_serving/mod.rs +++ b/apps/aether-gateway/src/ai_serving/mod.rs @@ -3,6 +3,7 @@ pub(crate) mod api; mod finalize; mod planner; mod pure; +mod response_history; pub(crate) mod transport; use axum::body::Body; @@ -64,6 +65,10 @@ pub(crate) use self::planner::{ SkippedLocalExecutionCandidate, }; pub(crate) use self::pure::*; +pub(crate) use self::response_history::{ + hydrate_openai_response_history, persist_converted_response_history, + persist_response_history_record, +}; pub(crate) use self::transport::{ append_transport_diagnostics_to_value, build_request_trace_proxy_value, candidate_common_transport_skip_reason, candidate_transport_pair_skip_reason, diff --git a/apps/aether-gateway/src/ai_serving/planner/standard/family/request.rs b/apps/aether-gateway/src/ai_serving/planner/standard/family/request.rs index e2c29c99e..e6480e8c2 100644 --- a/apps/aether-gateway/src/ai_serving/planner/standard/family/request.rs +++ b/apps/aether-gateway/src/ai_serving/planner/standard/family/request.rs @@ -584,6 +584,14 @@ pub(crate) async fn resolve_local_standard_candidate_payload_parts( return Ok(None); } }; + crate::ai_serving::hydrate_openai_response_history( + state.runtime_state(), + body_json, + spec_metadata.api_format, + provider_api_format, + input.auth_context.api_key_id.as_str(), + ) + .await?; let redaction = resolve_provider_chat_pii_redaction( state, parts, diff --git a/apps/aether-gateway/src/ai_serving/planner/standard/normalize/responses.rs b/apps/aether-gateway/src/ai_serving/planner/standard/normalize/responses.rs index dabdc5895..6ac9b14cc 100644 --- a/apps/aether-gateway/src/ai_serving/planner/standard/normalize/responses.rs +++ b/apps/aether-gateway/src/ai_serving/planner/standard/normalize/responses.rs @@ -3,7 +3,7 @@ use serde_json::Value; use crate::ai_serving::transport::apply_standard_provider_request_body_rules_with_request_headers; use crate::ai_serving::{ apply_openai_responses_compact_special_body_edits, - build_cross_format_openai_responses_request_body_with_model_directives as surface_build_cross_format_openai_responses_request_body, + build_cross_format_openai_responses_request_body_with_model_directives_and_history_scope as surface_build_cross_format_openai_responses_request_body, build_local_openai_responses_request_body_with_model_directives as surface_build_local_openai_responses_request_body, GatewayProviderTransportSnapshot, }; @@ -105,7 +105,7 @@ pub(crate) fn build_cross_format_openai_responses_request_body( force_body_stream_field: bool, provider_type: &str, body_rules: Option<&Value>, - _user_api_key_id: Option<&str>, + user_api_key_id: Option<&str>, request_headers: &http::HeaderMap, enable_model_directives: bool, ) -> Option { @@ -119,6 +119,7 @@ pub(crate) fn build_cross_format_openai_responses_request_body( provider_type, body_rules, request_headers, + user_api_key_id, None, enable_model_directives, ) @@ -134,6 +135,7 @@ pub(crate) fn build_cross_format_openai_responses_request_body_with_codex_model_ provider_type: &str, body_rules: Option<&Value>, request_headers: &http::HeaderMap, + history_scope: Option<&str>, model_capabilities: Option<&crate::ai_serving::CodexResponsesModelCapabilities>, enable_model_directives: bool, ) -> Option { @@ -144,6 +146,7 @@ pub(crate) fn build_cross_format_openai_responses_request_body_with_codex_model_ provider_api_format, upstream_is_stream, enable_model_directives, + history_scope, )?; let mut provider_request_body = apply_standard_provider_request_body_rules_with_request_headers( diff --git a/apps/aether-gateway/src/ai_serving/planner/standard/openai/responses/decision/request.rs b/apps/aether-gateway/src/ai_serving/planner/standard/openai/responses/decision/request.rs index 449a7cd14..90775a906 100644 --- a/apps/aether-gateway/src/ai_serving/planner/standard/openai/responses/decision/request.rs +++ b/apps/aether-gateway/src/ai_serving/planner/standard/openai/responses/decision/request.rs @@ -82,6 +82,96 @@ fn is_grok_text_provider_api_format(provider_api_format: &str) -> bool { ) } +fn response_function_tool_names(body: &Value) -> Vec { + body.get("tools") + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter(|tool| tool.get("type").and_then(Value::as_str) == Some("function")) + .filter_map(|tool| tool.get("name").and_then(Value::as_str)) + .map(ToOwned::to_owned) + .collect() +} + +fn chat_function_tool_names(body: &Value) -> Vec { + body.get("tools") + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter_map(|tool| { + tool.get("function") + .and_then(|function| function.get("name")) + .and_then(Value::as_str) + }) + .map(ToOwned::to_owned) + .collect() +} + +fn response_input_call_ids(body: &Value) -> Vec { + body.get("input") + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter(|item| { + matches!( + item.get("type").and_then(Value::as_str), + Some("function_call" | "function_call_output") + ) + }) + .filter_map(|item| item.get("call_id").and_then(Value::as_str)) + .map(ToOwned::to_owned) + .collect() +} + +fn chat_message_call_ids(body: &Value) -> Vec { + let mut call_ids = Vec::new(); + for message in body + .get("messages") + .and_then(Value::as_array) + .into_iter() + .flatten() + { + if let Some(tool_calls) = message.get("tool_calls").and_then(Value::as_array) { + call_ids.extend( + tool_calls + .iter() + .filter_map(|tool_call| tool_call.get("id").and_then(Value::as_str)) + .map(ToOwned::to_owned), + ); + } + if let Some(tool_call_id) = message.get("tool_call_id").and_then(Value::as_str) { + call_ids.push(tool_call_id.to_string()); + } + } + call_ids +} + +fn log_responses_to_chat_tool_conversion(trace_id: &str, inbound: &Value, outbound: &Value) { + let inbound_tool_names = response_function_tool_names(inbound); + let outbound_tool_names = chat_function_tool_names(outbound); + let inbound_call_ids = response_input_call_ids(inbound); + let outbound_call_ids = chat_message_call_ids(outbound); + let previous_response_id = inbound + .get("previous_response_id") + .and_then(Value::as_str) + .unwrap_or_default(); + debug!( + event_name = "openai_responses_to_chat_tool_conversion", + log_type = "debug", + trace_id = %trace_id, + inbound_tool_count = inbound_tool_names.len(), + outbound_tool_count = outbound_tool_names.len(), + inbound_tool_names = ?inbound_tool_names, + outbound_tool_names = ?outbound_tool_names, + previous_response_id = %previous_response_id, + inbound_call_ids = ?inbound_call_ids, + outbound_call_ids = ?outbound_call_ids, + history_recovered = !previous_response_id.is_empty() + && outbound_call_ids.iter().any(|call_id| inbound_call_ids.contains(call_id)), + "converted OpenAI Responses tools and continuation context to OpenAI Chat" + ); +} + pub(crate) struct LocalOpenAiResponsesCandidatePayloadParts { pub(super) auth_header: String, pub(super) auth_value: String, @@ -306,6 +396,14 @@ pub(crate) async fn resolve_local_openai_responses_candidate_payload_parts( return Ok(None); } }; + crate::ai_serving::hydrate_openai_response_history( + state.runtime_state(), + body_json, + spec_metadata.api_format, + provider_api_format, + input.auth_context.api_key_id.as_str(), + ) + .await?; let redaction = resolve_provider_chat_pii_redaction( state, parts, @@ -367,6 +465,7 @@ pub(crate) async fn resolve_local_openai_responses_candidate_payload_parts( transport.endpoint.body_rules.as_ref() }, effective_headers, + Some(input.auth_context.api_key_id.as_str()), codex_model_capabilities.as_ref(), false, ) @@ -471,6 +570,11 @@ pub(crate) async fn resolve_local_openai_responses_candidate_payload_parts( .await; return Ok(None); } + if needs_bidirectional_conversion + && crate::ai_serving::api_format_alias_matches(provider_api_format, "openai:chat") + { + log_responses_to_chat_tool_conversion(trace_id, body_json, &base_provider_request_body); + } let provider_request_body = base_provider_request_body; if let Some(kiro_auth) = kiro_auth.as_ref() { diff --git a/apps/aether-gateway/src/ai_serving/pure/mod.rs b/apps/aether-gateway/src/ai_serving/pure/mod.rs index d710ff9d7..f291b0596 100644 --- a/apps/aether-gateway/src/ai_serving/pure/mod.rs +++ b/apps/aether-gateway/src/ai_serving/pure/mod.rs @@ -17,6 +17,7 @@ pub(crate) use aether_ai_formats::api::{ build_cross_format_openai_chat_request_body_with_model_directives, build_cross_format_openai_responses_request_body, build_cross_format_openai_responses_request_body_with_model_directives, + build_cross_format_openai_responses_request_body_with_model_directives_and_history_scope, build_gemini_image_request_body_from_openai_image_request, build_gemini_image_response_from_openai_image_response, build_gemini_image_response_from_openai_responses_image_response, build_generated_tool_call_id, diff --git a/apps/aether-gateway/src/ai_serving/response_history.rs b/apps/aether-gateway/src/ai_serving/response_history.rs new file mode 100644 index 000000000..69f57f9f3 --- /dev/null +++ b/apps/aether-gateway/src/ai_serving/response_history.rs @@ -0,0 +1,96 @@ +use aether_ai_formats::api::{ + hydrate_response_history, normalize_api_format_alias, record_converted_response_history, + response_history_is_loaded, response_history_storage_key, ResponseHistoryRecord, +}; +use aether_runtime_state::RuntimeState; +use serde_json::Value; +use tracing::warn; + +use crate::GatewayError; + +pub(crate) async fn hydrate_openai_response_history( + runtime_state: &RuntimeState, + request: &Value, + client_api_format: &str, + provider_api_format: &str, + history_scope: &str, +) -> Result<(), GatewayError> { + if normalize_api_format_alias(client_api_format) != "openai:responses" + || normalize_api_format_alias(provider_api_format) != "openai:chat" + { + return Ok(()); + } + let Some(previous_response_id) = request + .get("previous_response_id") + .and_then(Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + else { + return Ok(()); + }; + if response_history_is_loaded(previous_response_id, Some(history_scope)) { + return Ok(()); + } + + let storage_key = response_history_storage_key(previous_response_id, Some(history_scope)); + let payload = runtime_state.kv_get(&storage_key).await.map_err(|error| { + warn!( + event_name = "openai_response_history_read_failed", + log_type = "ops", + backend = runtime_state.backend_kind().as_str(), + error = ?error, + "gateway failed to read shared OpenAI response history" + ); + GatewayError::Internal("OpenAI response history lookup failed".to_string()) + })?; + let Some(payload) = payload else { + return Ok(()); + }; + if let Err(error) = + hydrate_response_history(previous_response_id, Some(history_scope), &payload) + { + let _ = runtime_state.kv_delete(&storage_key).await; + warn!( + event_name = "openai_response_history_invalid", + log_type = "ops", + backend = runtime_state.backend_kind().as_str(), + error = %error, + "gateway rejected invalid shared OpenAI response history" + ); + return Err(GatewayError::Internal( + "OpenAI response history validation failed".to_string(), + )); + } + Ok(()) +} + +pub(crate) async fn persist_response_history_record( + runtime_state: &RuntimeState, + record: ResponseHistoryRecord, +) { + if let Err(error) = runtime_state + .kv_set(&record.storage_key, record.payload, Some(record.ttl)) + .await + { + warn!( + event_name = "openai_response_history_write_failed", + log_type = "ops", + backend = runtime_state.backend_kind().as_str(), + error = ?error, + "gateway failed to persist shared OpenAI response history" + ); + } +} + +pub(crate) async fn persist_converted_response_history( + runtime_state: &RuntimeState, + report_context: &Value, + response: Option<&Value>, +) { + let Some(response) = response else { + return; + }; + if let Some(record) = record_converted_response_history(report_context, response) { + persist_response_history_record(runtime_state, record).await; + } +} diff --git a/apps/aether-gateway/src/execution_runtime/stream/execution.rs b/apps/aether-gateway/src/execution_runtime/stream/execution.rs index d7fd67e77..13b0c70ca 100644 --- a/apps/aether-gateway/src/execution_runtime/stream/execution.rs +++ b/apps/aether-gateway/src/execution_runtime/stream/execution.rs @@ -6366,6 +6366,13 @@ async fn execute_stream_from_frame_stream_with_retry_scope( report_context.as_ref(), ) { Ok(Some(outcome)) => { + if let Some(record) = outcome.response_history_record { + crate::ai_serving::persist_response_history_record( + state.runtime_state(), + record, + ) + .await; + } headers.remove("content-encoding"); headers.remove("content-length"); headers.insert( @@ -6570,6 +6577,15 @@ async fn execute_stream_from_frame_stream_with_retry_scope( if stream_commit_gate.is_uncommitted() { stream_commit_gate.commit(); } + let prefetched_response_history_persisted = if let Some(record) = local_stream_rewriter + .as_mut() + .and_then(|rewriter| rewriter.take_response_history_record()) + { + crate::ai_serving::persist_response_history_record(state.runtime_state(), record).await; + true + } else { + false + }; drop(private_stream_normalizer); drop(local_stream_rewriter); @@ -6906,6 +6922,11 @@ async fn execute_stream_from_frame_stream_with_retry_scope( } } } + if prefetched_response_history_persisted { + if let Some(rewriter) = local_stream_rewriter.as_mut() { + let _ = rewriter.take_response_history_record(); + } + } } if terminal_failure.is_none() && !reached_eof { @@ -7159,6 +7180,19 @@ async fn execute_stream_from_frame_stream_with_retry_scope( normalized_chunk }; + if provider_private_error_body_json.is_none() { + if let Some(record) = local_stream_rewriter + .as_mut() + .and_then(|rewriter| rewriter.take_response_history_record()) + { + crate::ai_serving::persist_response_history_record( + state_for_report.runtime_state(), + record, + ) + .await; + } + } + if rewritten_chunk.is_empty() { if let Some(error_body_json) = provider_private_error_body_json { let error_status_code = resolve_provider_stream_error_status_code( @@ -7350,6 +7384,18 @@ async fn execute_stream_from_frame_stream_with_retry_scope( } else { normalized_chunk }; + if provider_private_error_body_json.is_none() { + if let Some(record) = local_stream_rewriter + .as_mut() + .and_then(|rewriter| rewriter.take_response_history_record()) + { + crate::ai_serving::persist_response_history_record( + state_for_report.runtime_state(), + record, + ) + .await; + } + } if !rewritten_chunk.is_empty() { append_stream_capture_bytes( &mut buffered_body, @@ -7423,7 +7469,15 @@ async fn execute_stream_from_frame_stream_with_retry_scope( } if !downstream_dropped && terminal_failure.is_none() { if let Some(rewriter) = local_stream_rewriter.as_mut() { - match rewriter.finish() { + let finish_result = rewriter.finish(); + if let Some(record) = rewriter.take_response_history_record() { + crate::ai_serving::persist_response_history_record( + state_for_report.runtime_state(), + record, + ) + .await; + } + match finish_result { Ok(flushed_chunk) if !flushed_chunk.is_empty() => { append_stream_capture_bytes( &mut buffered_body, @@ -7481,6 +7535,19 @@ async fn execute_stream_from_frame_stream_with_retry_scope( } } + if terminal_failure.is_none() { + if let Some(record) = local_stream_rewriter + .as_mut() + .and_then(|rewriter| rewriter.take_response_history_record()) + { + crate::ai_serving::persist_response_history_record( + state_for_report.runtime_state(), + record, + ) + .await; + } + } + if !downstream_dropped { if let Some(failure) = terminal_failure.as_ref() { let terminal_event = if is_openai_image_stream_for_report { diff --git a/apps/aether-gateway/src/execution_runtime/sync/execution.rs b/apps/aether-gateway/src/execution_runtime/sync/execution.rs index a5ada874c..5cdc6bc0d 100644 --- a/apps/aether-gateway/src/execution_runtime/sync/execution.rs +++ b/apps/aether-gateway/src/execution_runtime/sync/execution.rs @@ -1789,6 +1789,17 @@ async fn apply_sync_success_effects( report_context: Option<&serde_json::Value>, payload: &GatewaySyncReportRequest, ) { + if let Some(report_context) = report_context { + crate::ai_serving::persist_converted_response_history( + state.runtime_state(), + report_context, + payload + .client_body_json + .as_ref() + .or(payload.body_json.as_ref()), + ) + .await; + } apply_local_execution_effect( state, LocalExecutionEffectContext { diff --git a/crates/aether-ai/formats/src/api.rs b/crates/aether-ai/formats/src/api.rs index d38bc0cdf..1ae51737c 100644 --- a/crates/aether-ai/formats/src/api.rs +++ b/crates/aether-ai/formats/src/api.rs @@ -190,6 +190,10 @@ pub use crate::formats::{ CODEX_OPENAI_IMAGE_DEFAULT_VARIATION_MODEL, CODEX_OPENAI_IMAGE_DEFAULT_VARIATION_PROMPT, CODEX_OPENAI_IMAGE_INTERNAL_MODEL, }, + history::{ + hydrate_response_history, record_converted_response_history, + response_history_is_loaded, response_history_storage_key, ResponseHistoryRecord, + }, spec::{ resolve_stream_spec as resolve_openai_responses_stream_spec, resolve_sync_spec as resolve_openai_responses_sync_spec, LocalOpenAiResponsesSpec, @@ -207,6 +211,7 @@ pub use crate::formats::{ build_cross_format_openai_chat_request_body_with_model_directives, build_cross_format_openai_responses_request_body, build_cross_format_openai_responses_request_body_with_model_directives, + build_cross_format_openai_responses_request_body_with_model_directives_and_history_scope, build_local_openai_chat_request_body, build_local_openai_chat_request_body_with_model_directives, build_local_openai_responses_request_body, @@ -269,7 +274,9 @@ pub use aether_ai_formats::formats::conversion::request::{ convert_openai_chat_request_to_openai_responses_request, extract_openai_text_content, normalize_claude_request_to_openai_chat_request, normalize_gemini_request_to_openai_chat_request, - normalize_openai_responses_request_to_openai_chat_request, parse_openai_tool_result_content, + normalize_openai_responses_request_to_openai_chat_request, + normalize_openai_responses_request_to_openai_chat_request_with_history_scope, + parse_openai_tool_result_content, }; pub use aether_ai_formats::formats::conversion::response::{ build_openai_responses_response, build_openai_responses_response_with_content, diff --git a/crates/aether-ai/formats/src/formats/context.rs b/crates/aether-ai/formats/src/formats/context.rs index bbef7dcca..6c09e9114 100644 --- a/crates/aether-ai/formats/src/formats/context.rs +++ b/crates/aether-ai/formats/src/formats/context.rs @@ -9,6 +9,7 @@ pub struct FormatContext { pub request_path: Option, pub upstream_is_stream: bool, pub report_context: Option, + pub history_scope: Option, } impl FormatContext { @@ -32,12 +33,18 @@ impl FormatContext { self } + pub fn with_history_scope(mut self, history_scope: impl Into) -> Self { + self.history_scope = Some(history_scope.into()); + self + } + pub fn without_runtime_request_edits(&self) -> Self { Self { mapped_model: None, request_path: self.request_path.clone(), upstream_is_stream: false, report_context: self.report_context.clone(), + history_scope: self.history_scope.clone(), } } diff --git a/crates/aether-ai/formats/src/formats/conversion/request.rs b/crates/aether-ai/formats/src/formats/conversion/request.rs index 993cd67db..2c884dc9f 100644 --- a/crates/aether-ai/formats/src/formats/conversion/request.rs +++ b/crates/aether-ai/formats/src/formats/conversion/request.rs @@ -59,13 +59,18 @@ pub fn convert_openai_chat_request_to_openai_responses_request( pub fn normalize_openai_responses_request_to_openai_chat_request( body_json: &Value, ) -> Option { - registry::convert_request( - "openai:responses", - "openai:chat", - body_json, - &FormatContext::default(), - ) - .ok() + normalize_openai_responses_request_to_openai_chat_request_with_history_scope(body_json, None) +} + +pub fn normalize_openai_responses_request_to_openai_chat_request_with_history_scope( + body_json: &Value, + history_scope: Option<&str>, +) -> Option { + let mut context = FormatContext::default(); + if let Some(history_scope) = history_scope { + context = context.with_history_scope(history_scope); + } + registry::convert_request("openai:responses", "openai:chat", body_json, &context).ok() } pub fn normalize_claude_request_to_openai_chat_request(body_json: &Value) -> Option { diff --git a/crates/aether-ai/formats/src/formats/openai/chat/stream.rs b/crates/aether-ai/formats/src/formats/openai/chat/stream.rs index 32992dceb..2fc8fc99a 100644 --- a/crates/aether-ai/formats/src/formats/openai/chat/stream.rs +++ b/crates/aether-ai/formats/src/formats/openai/chat/stream.rs @@ -21,6 +21,7 @@ fn normalize_openai_service_tier(value: Option<&str>) -> Option { struct OpenAIChatProviderToolState { id: Option, name: Option, + pending_arguments: String, started_emitted: bool, } @@ -279,59 +280,54 @@ impl OpenAIChatProviderState { if let Some(call_id) = tool_call_object.get("id").and_then(Value::as_str) { state.id = Some(call_id.to_string()); } + let mut arguments = None; if let Some(function) = tool_call_object.get("function").and_then(Value::as_object) { if let Some(name) = function.get("name").and_then(Value::as_str) { state.name = Some(name.to_string()); } - if !state.started_emitted && (state.id.is_some() || state.name.is_some()) { + arguments = function + .get("arguments") + .and_then(Value::as_str) + .filter(|arguments| !arguments.is_empty()); + } + if !state.started_emitted { + if let Some(arguments) = arguments { + state.pending_arguments.push_str(arguments); + } + if let (Some(call_id), Some(name)) = (state.id.clone(), state.name.clone()) + { out.push(CanonicalStreamFrame { id: id.clone(), model: model.clone(), event: CanonicalStreamEvent::ToolCallStart { index, - call_id: state - .id - .clone() - .unwrap_or_else(|| build_generated_tool_call_id(index)), - name: state - .name - .clone() - .unwrap_or_else(|| "unknown".to_string()), + call_id, + name, }, }); state.started_emitted = true; - } - if let Some(arguments) = function.get("arguments").and_then(Value::as_str) { - if !arguments.is_empty() { - if !state.started_emitted { - out.push(CanonicalStreamFrame { - id: id.clone(), - model: model.clone(), - event: CanonicalStreamEvent::ToolCallStart { - index, - call_id: state.id.clone().unwrap_or_else(|| { - build_generated_tool_call_id(index) - }), - name: state - .name - .clone() - .unwrap_or_else(|| "unknown".to_string()), - }, - }); - state.started_emitted = true; - } + if !state.pending_arguments.is_empty() { out.push(CanonicalStreamFrame { id: id.clone(), model: model.clone(), event: CanonicalStreamEvent::ToolCallArgumentsDelta { index, - arguments: arguments.to_string(), + arguments: std::mem::take(&mut state.pending_arguments), }, }); } } + } else if let Some(arguments) = arguments { + out.push(CanonicalStreamFrame { + id: id.clone(), + model: model.clone(), + event: CanonicalStreamEvent::ToolCallArgumentsDelta { + index, + arguments: arguments.to_string(), + }, + }); } } } else if delta.contains_key("tool_calls") { @@ -651,6 +647,22 @@ impl OpenAIResponsesProviderState { }); } + fn emit_ready_function_call( + &mut self, + report_context: &Value, + out: &mut Vec, + index: usize, + ) { + if self + .tool_calls + .get(&index) + .is_none_or(|state| state.call_id.trim().is_empty()) + { + return; + } + self.emit_ready_tool_call(report_context, out, index); + } + fn merge_tool_call_arguments(state: &mut OpenAIResponsesProviderToolState, arguments: &str) { if arguments.is_empty() { return; @@ -715,7 +727,6 @@ impl OpenAIResponsesProviderState { let state = self.tool_calls.entry(index).or_default(); state.call_id = item .get("call_id") - .or_else(|| item.get("id")) .and_then(Value::as_str) .unwrap_or(state.call_id.as_str()) .to_string(); @@ -730,7 +741,7 @@ impl OpenAIResponsesProviderState { .unwrap_or_default() .to_string(); Self::merge_tool_call_arguments(state, &completed_arguments); - self.emit_ready_tool_call(report_context, out, index); + self.emit_ready_function_call(report_context, out, index); } fn emit_custom_tool_call_item( @@ -1553,21 +1564,11 @@ impl OpenAIResponsesProviderState { .map(|value| value as usize); let index = self.tool_index_for_key(key, output_index); let state = self.tool_calls.entry(index).or_default(); - if let Some(call_id) = value - .get("call_id") - .or_else(|| value.get("id")) - .and_then(Value::as_str) - { + if let Some(call_id) = value.get("call_id").and_then(Value::as_str) { state.call_id = call_id.to_string(); - } else if state.call_id.is_empty() { - state.call_id = value - .get("item_id") - .and_then(Value::as_str) - .unwrap_or_default() - .to_string(); } state.arguments.push_str(delta); - self.emit_ready_tool_call(report_context, &mut out, index); + self.emit_ready_function_call(report_context, &mut out, index); } "response.function_call_arguments.done" => { let arguments = value @@ -1604,16 +1605,14 @@ impl OpenAIResponsesProviderState { let state = self.tool_calls.entry(index).or_default(); state.call_id = value .get("call_id") - .or_else(|| value.get("id")) .and_then(Value::as_str) .or_else(|| { value .get("item") .and_then(Value::as_object) - .and_then(|item| item.get("call_id").or_else(|| item.get("id"))) + .and_then(|item| item.get("call_id")) .and_then(Value::as_str) }) - .or_else(|| value.get("item_id").and_then(Value::as_str)) .unwrap_or(state.call_id.as_str()) .to_string(); state.name = value @@ -1629,7 +1628,7 @@ impl OpenAIResponsesProviderState { .unwrap_or(state.name.as_str()) .to_string(); Self::merge_tool_call_arguments(state, arguments); - self.emit_ready_tool_call(report_context, &mut out, index); + self.emit_ready_function_call(report_context, &mut out, index); } "response.function_call_output.delta" | "response.function_call_output.done" => { let tool_use_id = value @@ -1896,6 +1895,7 @@ pub struct OpenAIResponsesClientEmitter { image_generation_items: BTreeMap, opaque_output_items: BTreeMap, opaque_output_indexes: BTreeMap, + completed_history_response: Option, } impl OpenAIChatClientEmitter { @@ -2210,6 +2210,10 @@ impl OpenAIResponsesClientEmitter { self.response_id.as_deref().unwrap_or("resp-local-stream") } + pub(crate) fn completed_response_for_history(&self) -> Option<&Value> { + self.completed_history_response.as_ref() + } + fn model(&self) -> &str { self.model.as_deref().unwrap_or("unknown") } @@ -2226,6 +2230,14 @@ impl OpenAIResponsesClientEmitter { .unwrap_or_else(|| format!("{}_rs_0", self.response_id())) } + fn tool_call_item_id(&self, index: usize) -> String { + format!( + "fc_{}_{}", + self.response_id().trim_start_matches("resp_"), + index + ) + } + fn ensure_message_item_id(&mut self) -> String { if self.message_item_id.is_none() { self.message_item_id = Some(format!("{}_msg", self.response_id())); @@ -2601,11 +2613,12 @@ impl OpenAIResponsesClientEmitter { for index in indices { let output_index = self.ensure_tool_output_index(index); let state = self.tool_calls.get(&index).cloned().unwrap_or_default(); - let item_id = if state.call_id.is_empty() { + let call_id = if state.call_id.is_empty() { build_generated_tool_call_id(index) } else { state.call_id.clone() }; + let item_id = self.tool_call_item_id(index); let name = if state.name.is_empty() { "unknown".to_string() } else { @@ -2638,7 +2651,7 @@ impl OpenAIResponsesClientEmitter { "response_id": self.response_id(), "output_index": output_index, "item_id": item_id.clone(), - "call_id": item_id.clone(), + "call_id": call_id.clone(), "name": name, "arguments": state.arguments.as_str(), }), @@ -2652,7 +2665,7 @@ impl OpenAIResponsesClientEmitter { "item": { "type": "function_call", "id": item_id.clone(), - "call_id": item_id, + "call_id": call_id, "name": name, "arguments": state.arguments.as_str(), "status": "completed", @@ -2795,11 +2808,12 @@ impl OpenAIResponsesClientEmitter { } for (index, state) in &self.tool_calls { if let Some(output_index) = state.output_index { - let item_id = if state.call_id.is_empty() { + let call_id = if state.call_id.is_empty() { build_generated_tool_call_id(*index) } else { state.call_id.clone() }; + let item_id = self.tool_call_item_id(*index); if state.web_search { ordered_output.push(( output_index, @@ -2820,7 +2834,7 @@ impl OpenAIResponsesClientEmitter { json!({ "type": "function_call", "id": item_id.clone(), - "call_id": item_id, + "call_id": call_id, "name": if state.name.is_empty() { "unknown".to_string() } else { @@ -2948,6 +2962,9 @@ impl OpenAIResponsesClientEmitter { out.extend(self.finish_text_item()?); out.extend(self.finish_tool_items()?); out.extend(self.finish_tool_result_items()?); + if event_type == "response.completed" { + self.completed_history_response = Some(response.clone()); + } out.extend(self.encode_response_event( event_type, json!({ @@ -3110,6 +3127,7 @@ impl OpenAIResponsesClientEmitter { let mut out = self.ensure_started()?; let output_index = self.ensure_tool_output_index(index); let response_id = self.response_id().to_string(); + let item_id = self.tool_call_item_id(index); let state = self.tool_calls.entry(index).or_default(); state.call_id = call_id.clone(); state.name = name.clone(); @@ -3119,7 +3137,7 @@ impl OpenAIResponsesClientEmitter { let item = if state.web_search { json!({ "type": "web_search_call", - "id": emitted_call_id, + "id": item_id, "status": "in_progress", "action": { "type": "search", @@ -3129,7 +3147,7 @@ impl OpenAIResponsesClientEmitter { } else { json!({ "type": "function_call", - "id": call_id, + "id": item_id, "call_id": emitted_call_id, "name": emitted_name, "arguments": "", @@ -3156,19 +3174,20 @@ impl OpenAIResponsesClientEmitter { if state.web_search { return Ok(out); } - let item_id = if state.call_id.is_empty() { + let call_id = if state.call_id.is_empty() { build_generated_tool_call_id(index) } else { state.call_id.clone() }; + let item_id = self.tool_call_item_id(index); out.extend(self.encode_response_event( "response.function_call_arguments.delta", json!({ "type": "response.function_call_arguments.delta", "response_id": response_id, "output_index": output_index, - "item_id": item_id.clone(), - "call_id": item_id, + "item_id": item_id, + "call_id": call_id, "delta": arguments, }), )?); @@ -3297,6 +3316,9 @@ impl OpenAIResponsesClientEmitter { ), _ => ("response.completed", self.completed_response(usage)), }; + if event_type == "response.completed" { + self.completed_history_response = Some(response.clone()); + } out.extend(self.encode_response_event( event_type, json!({ @@ -3623,6 +3645,81 @@ mod tests { ))); } + #[test] + fn openai_chat_provider_state_waits_for_real_tool_call_identity() { + let mut state = OpenAIChatProviderState::default(); + let report_context = json!({}); + let mut frames = state + .push_line( + &report_context, + data_line(json!({ + "id": "chatcmpl_tool_123", + "model": "deepseek-v4-flash", + "choices": [{ + "index": 0, + "delta": { + "tool_calls": [{ + "index": 0, + "function": {"arguments": "{\"path\":"} + }] + } + }] + })), + ) + .expect("arguments-first delta should parse"); + frames.extend( + state + .push_line( + &report_context, + data_line(json!({ + "id": "chatcmpl_tool_123", + "model": "deepseek-v4-flash", + "choices": [{ + "index": 0, + "delta": { + "tool_calls": [{ + "index": 0, + "function": {"name": "security_scan", "arguments": "\"src\"}"} + }] + } + }] + })), + ) + .expect("name delta should parse"), + ); + + assert!(!frames.iter().any(|frame| matches!( + frame.event, + CanonicalStreamEvent::ToolCallStart { .. } + | CanonicalStreamEvent::ToolCallArgumentsDelta { .. } + ))); + + let identified = state + .push_line( + &report_context, + data_line(json!({ + "id": "chatcmpl_tool_123", + "model": "deepseek-v4-flash", + "choices": [{ + "index": 0, + "delta": {"tool_calls": [{"index": 0, "id": "call_security_123"}]} + }] + })), + ) + .expect("id delta should parse"); + + assert!(matches!( + identified.first().map(|frame| &frame.event), + Some(CanonicalStreamEvent::ToolCallStart { call_id, name, .. }) + if call_id == "call_security_123" && name == "security_scan" + )); + assert!(matches!( + identified.get(1).map(|frame| &frame.event), + Some(CanonicalStreamEvent::ToolCallArgumentsDelta { arguments, .. }) + if arguments == "{\"path\":\"src\"}" + )); + } + #[test] fn openai_responses_provider_state_emits_unknown_events_for_unknown_response_types() { let mut state = OpenAIResponsesProviderState::default(); @@ -4543,6 +4640,91 @@ mod tests { assert!(!sse.contains("\\\"pages\\\":\\\"\\\"")); } + #[test] + fn openai_responses_provider_state_waits_for_call_id_distinct_from_item_id() { + let mut state = OpenAIResponsesProviderState::default(); + let report_context = json!({}); + + let item_frames = state + .push_line( + &report_context, + data_line(json!({ + "type": "response.output_item.added", + "response_id": "resp_delayed_call_id", + "output_index": 0, + "item": { + "type": "function_call", + "id": "fc_delayed_call_id", + "name": "Read", + "arguments": "" + } + })), + ) + .expect("function-call item should parse"); + assert!(!item_frames.iter().any(|frame| matches!( + frame.event, + CanonicalStreamEvent::ToolCallStart { .. } + | CanonicalStreamEvent::ToolCallArgumentsDelta { .. } + ))); + + let delta_frames = state + .push_line( + &report_context, + data_line(json!({ + "type": "response.function_call_arguments.delta", + "response_id": "resp_delayed_call_id", + "output_index": 0, + "item_id": "fc_delayed_call_id", + "delta": "{\"path\":" + })), + ) + .expect("arguments delta should be buffered"); + assert!(!delta_frames.iter().any(|frame| matches!( + frame.event, + CanonicalStreamEvent::ToolCallStart { .. } + | CanonicalStreamEvent::ToolCallArgumentsDelta { .. } + ))); + + let identity_frames = state + .push_line( + &report_context, + data_line(json!({ + "type": "response.function_call_arguments.done", + "response_id": "resp_delayed_call_id", + "output_index": 0, + "item_id": "fc_delayed_call_id", + "item": { + "type": "function_call", + "id": "fc_delayed_call_id", + "call_id": "call_delayed_call_id", + "name": "Read", + "arguments": "{\"path\":\"src/lib.rs\"}" + }, + "arguments": "{\"path\":\"src/lib.rs\"}" + })), + ) + .expect("real call identity should flush buffered arguments"); + + assert!(identity_frames.iter().any(|frame| matches!( + frame.event, + CanonicalStreamEvent::ToolCallStart { + ref call_id, + ref name, + .. + } if call_id == "call_delayed_call_id" && name == "Read" + ))); + assert!(!identity_frames.iter().any(|frame| matches!( + frame.event, + CanonicalStreamEvent::ToolCallStart { ref call_id, .. } + if call_id == "fc_delayed_call_id" + ))); + assert!(identity_frames.iter().any(|frame| matches!( + frame.event, + CanonicalStreamEvent::ToolCallArgumentsDelta { ref arguments, .. } + if arguments == "{\"path\":\"src/lib.rs\"}" + ))); + } + #[test] fn openai_responses_provider_state_parses_function_call_output_as_tool_result() { let mut state = OpenAIResponsesProviderState::default(); @@ -4825,6 +5007,68 @@ mod tests { assert!(sse.contains("\"output\":\"{\\\"ok\\\":true}\"")); } + #[test] + fn openai_responses_client_emitter_keeps_call_id_distinct_and_stable() { + let mut emitter = OpenAIResponsesClientEmitter::default(); + let mut bytes = emitter + .emit(CanonicalStreamFrame { + id: "resp_tool_identity_123".to_string(), + model: "deepseek-v4-flash".to_string(), + event: CanonicalStreamEvent::ToolCallStart { + index: 0, + call_id: "call_security_123".to_string(), + name: "security_scan".to_string(), + }, + }) + .expect("tool start should encode"); + bytes.extend( + emitter + .emit(CanonicalStreamFrame { + id: "resp_tool_identity_123".to_string(), + model: "deepseek-v4-flash".to_string(), + event: CanonicalStreamEvent::ToolCallArgumentsDelta { + index: 0, + arguments: "{\"depth\":\"deep\"}".to_string(), + }, + }) + .expect("tool arguments should encode"), + ); + bytes.extend( + emitter + .emit(CanonicalStreamFrame { + id: "resp_tool_identity_123".to_string(), + model: "deepseek-v4-flash".to_string(), + event: CanonicalStreamEvent::Finish { + finish_reason: Some("tool_calls".to_string()), + usage: None, + }, + }) + .expect("tool finish should encode"), + ); + + let sse = String::from_utf8(bytes).expect("sse should be utf8"); + let item_id = "fc_tool_identity_123_0"; + assert!(sse.contains(&format!("\"id\":\"{item_id}\""))); + assert!(sse.contains(&format!("\"item_id\":\"{item_id}\""))); + assert!(sse.contains("\"call_id\":\"call_security_123\"")); + assert!(!sse.contains("\"id\":\"call_security_123\"")); + assert!( + sse.find("event: response.output_item.added") + < sse.find("event: response.function_call_arguments.delta") + ); + assert!( + sse.find("event: response.function_call_arguments.delta") + < sse.find("event: response.function_call_arguments.done") + ); + assert!( + sse.find("event: response.function_call_arguments.done") + < sse.find("event: response.output_item.done") + ); + assert!( + sse.find("event: response.output_item.done") < sse.find("event: response.completed") + ); + } + #[test] fn openai_responses_client_emitter_emits_web_search_call_item() { let mut emitter = OpenAIResponsesClientEmitter::default(); diff --git a/crates/aether-ai/formats/src/formats/openai/responses/history.rs b/crates/aether-ai/formats/src/formats/openai/responses/history.rs new file mode 100644 index 000000000..43dbf008c --- /dev/null +++ b/crates/aether-ai/formats/src/formats/openai/responses/history.rs @@ -0,0 +1,815 @@ +use std::{ + collections::{HashMap, VecDeque}, + sync::{Mutex, OnceLock}, + time::{Duration, Instant, SystemTime, UNIX_EPOCH}, +}; + +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; + +use crate::formats::{context::FormatContext, registry}; + +const HISTORY_TTL: Duration = Duration::from_secs(6 * 60 * 60); +const MAX_HISTORY_ENTRIES: usize = 2_048; +const MAX_HISTORY_BYTES: usize = 64 * 1024 * 1024; +const MAX_HISTORY_ENTRY_BYTES: usize = 8 * 1024 * 1024; +const HISTORY_STORAGE_VERSION: u8 = 1; +const HISTORY_STORAGE_KEY_PREFIX: &str = "ai:responses:history:v1"; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ResponseHistoryRecord { + pub storage_key: String, + pub payload: String, + pub ttl: Duration, +} + +#[derive(Serialize, Deserialize)] +struct PersistedResponseHistory { + version: u8, + response_id: String, + scope_fingerprint: String, + expires_at_unix_secs: u64, + transcript: Vec, +} + +#[derive(Clone)] +struct ResponseHistoryEntry { + transcript: Vec, + inserted_at: Instant, + expires_at: Instant, + size_bytes: usize, +} + +#[derive(Clone, Hash, PartialEq, Eq)] +struct ResponseHistoryKey { + scope: Option, + response_id: String, +} + +#[derive(Default)] +struct ResponseHistoryStore { + entries: HashMap, + insertion_order: VecDeque<(ResponseHistoryKey, Instant)>, + total_bytes: usize, +} + +impl ResponseHistoryStore { + fn remove(&mut self, key: &ResponseHistoryKey) { + if let Some(entry) = self.entries.remove(key) { + self.total_bytes = self.total_bytes.saturating_sub(entry.size_bytes); + } + } + + fn prune(&mut self, now: Instant) { + let expired_keys = self + .entries + .iter() + .filter(|(_, entry)| entry.expires_at <= now) + .map(|(key, _)| key.clone()) + .collect::>(); + for key in expired_keys { + self.remove(&key); + } + + while self.entries.len() > MAX_HISTORY_ENTRIES || self.total_bytes > MAX_HISTORY_BYTES { + let Some((key, inserted_at)) = self.insertion_order.pop_front() else { + break; + }; + if self + .entries + .get(&key) + .is_some_and(|entry| entry.inserted_at == inserted_at) + { + self.remove(&key); + } + } + + while let Some((key, inserted_at)) = self.insertion_order.front() { + if self + .entries + .get(key) + .is_some_and(|entry| entry.inserted_at == *inserted_at) + { + break; + } + self.insertion_order.pop_front(); + } + } + + fn get( + &mut self, + response_id: &str, + history_scope: Option<&str>, + now: Instant, + ) -> Option> { + self.prune(now); + self.entries + .get(&response_history_key(response_id, history_scope)) + .map(|entry| entry.transcript.clone()) + } + + fn insert( + &mut self, + response_id: String, + history_scope: Option<&str>, + transcript: Vec, + now: Instant, + ttl: Duration, + ) { + if ttl.is_zero() { + return; + } + let size_bytes = serde_json::to_vec(&transcript) + .map(|bytes| bytes.len()) + .unwrap_or(MAX_HISTORY_ENTRY_BYTES.saturating_add(1)); + if size_bytes > MAX_HISTORY_ENTRY_BYTES { + return; + } + let key = response_history_key(&response_id, history_scope); + self.remove(&key); + self.total_bytes = self.total_bytes.saturating_add(size_bytes); + self.entries.insert( + key.clone(), + ResponseHistoryEntry { + transcript, + inserted_at: now, + expires_at: now.checked_add(ttl).unwrap_or(now), + size_bytes, + }, + ); + self.insertion_order.push_back((key, now)); + self.prune(now); + } +} + +fn response_history_key(response_id: &str, history_scope: Option<&str>) -> ResponseHistoryKey { + ResponseHistoryKey { + scope: history_scope + .map(str::trim) + .filter(|scope| !scope.is_empty()) + .map(ToOwned::to_owned), + response_id: response_id.to_string(), + } +} + +fn normalized_history_scope(history_scope: Option<&str>) -> Option<&str> { + history_scope + .map(str::trim) + .filter(|scope| !scope.is_empty()) +} + +fn sha256_hex(value: &[u8]) -> String { + Sha256::digest(value) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect() +} + +fn history_scope_fingerprint(history_scope: Option<&str>) -> String { + sha256_hex( + normalized_history_scope(history_scope) + .unwrap_or("") + .as_bytes(), + ) +} + +pub fn response_history_storage_key(response_id: &str, history_scope: Option<&str>) -> String { + let mut hasher = Sha256::new(); + hasher.update(history_scope_fingerprint(history_scope)); + hasher.update([0]); + hasher.update(response_id.trim().as_bytes()); + let digest = hasher + .finalize() + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::(); + format!("{HISTORY_STORAGE_KEY_PREFIX}:{digest}") +} + +fn current_unix_secs() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs() +} + +fn response_history_store() -> &'static Mutex { + static STORE: OnceLock> = OnceLock::new(); + STORE.get_or_init(|| Mutex::new(ResponseHistoryStore::default())) +} + +pub fn response_history_is_loaded(response_id: &str, history_scope: Option<&str>) -> bool { + response_history_store() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .get(response_id, history_scope, Instant::now()) + .is_some() +} + +pub fn hydrate_response_history( + response_id: &str, + history_scope: Option<&str>, + payload: &str, +) -> Result<(), String> { + if payload.len() > MAX_HISTORY_ENTRY_BYTES { + return Err("persisted response history exceeds the maximum entry size".to_string()); + } + let persisted: PersistedResponseHistory = serde_json::from_str(payload) + .map_err(|error| format!("invalid persisted response history: {error}"))?; + if persisted.version != HISTORY_STORAGE_VERSION { + return Err(format!( + "unsupported response history version {}", + persisted.version + )); + } + if persisted.response_id != response_id.trim() { + return Err( + "persisted response history id does not match the requested response".to_string(), + ); + } + if persisted.scope_fingerprint != history_scope_fingerprint(history_scope) { + return Err("persisted response history scope does not match the requester".to_string()); + } + let now_unix_secs = current_unix_secs(); + let remaining_ttl = persisted + .expires_at_unix_secs + .checked_sub(now_unix_secs) + .filter(|seconds| *seconds > 0) + .map(Duration::from_secs) + .ok_or_else(|| "persisted response history has expired".to_string())?; + response_history_store() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .insert( + response_id.trim().to_string(), + history_scope, + persisted.transcript, + Instant::now(), + remaining_ttl.min(HISTORY_TTL), + ); + Ok(()) +} + +fn request_input_items(request: &Value) -> Vec { + match request.get("input") { + Some(Value::Array(items)) => items.clone(), + Some(Value::String(text)) if !text.is_empty() => vec![json!({ + "type": "message", + "role": "user", + "content": text, + })], + _ if request.get("messages").and_then(Value::as_array).is_some() => { + registry::convert_request( + "openai:chat", + "openai:responses", + request, + &FormatContext::default(), + ) + .ok() + .and_then(|converted| converted.get("input").and_then(Value::as_array).cloned()) + .unwrap_or_default() + } + _ => Vec::new(), + } +} + +fn history_conversion_enabled(report_context: &Value) -> bool { + report_context + .get("needs_conversion") + .and_then(Value::as_bool) + .unwrap_or(false) + && report_context + .get("client_api_format") + .and_then(Value::as_str) + .is_some_and(|format| format.eq_ignore_ascii_case("openai:responses")) + && report_context + .get("provider_api_format") + .and_then(Value::as_str) + .is_some_and(|format| format.eq_ignore_ascii_case("openai:chat")) +} + +pub(crate) fn expand_previous_response_for_chat( + request: &Value, + history_scope: Option<&str>, +) -> Result { + let Some(previous_response_id) = request + .get("previous_response_id") + .and_then(Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + else { + return Ok(request.clone()); + }; + let mut store = response_history_store() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let Some(mut transcript) = store.get(previous_response_id, history_scope, Instant::now()) + else { + return Err(format!( + "response history not found for previous_response_id {previous_response_id}" + )); + }; + transcript.extend(request_input_items(request)); + let mut expanded = request.clone(); + let Some(object) = expanded.as_object_mut() else { + return Err("OpenAI Responses request must be a JSON object".to_string()); + }; + object.remove("previous_response_id"); + object.insert("input".to_string(), Value::Array(transcript)); + Ok(expanded) +} + +pub fn record_converted_response_history( + report_context: &Value, + response: &Value, +) -> Option { + if !history_conversion_enabled(report_context) + || response.get("status").and_then(Value::as_str) != Some("completed") + { + return None; + } + let Some(response_id) = response + .get("id") + .and_then(Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + else { + return None; + }; + let Some(request) = report_context.get("original_request_body") else { + return None; + }; + let history_scope = report_context + .get("api_key_id") + .and_then(Value::as_str) + .map(str::trim) + .filter(|scope| !scope.is_empty()); + let mut transcript = if request.get("messages").and_then(Value::as_array).is_some() { + Vec::new() + } else { + request + .get("previous_response_id") + .and_then(Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + .and_then(|previous_response_id| { + response_history_store() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .get(previous_response_id, history_scope, Instant::now()) + }) + .unwrap_or_default() + }; + transcript.extend(request_input_items(request)); + if let Some(output) = response.get("output").and_then(Value::as_array) { + transcript.extend(output.iter().cloned()); + } + let expires_at_unix_secs = current_unix_secs().saturating_add(HISTORY_TTL.as_secs()); + let payload = serde_json::to_string(&PersistedResponseHistory { + version: HISTORY_STORAGE_VERSION, + response_id: response_id.to_string(), + scope_fingerprint: history_scope_fingerprint(history_scope), + expires_at_unix_secs, + transcript: transcript.clone(), + }) + .ok()?; + if payload.len() > MAX_HISTORY_ENTRY_BYTES { + return None; + } + response_history_store() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .insert( + response_id.to_string(), + history_scope, + transcript, + Instant::now(), + HISTORY_TTL, + ); + Some(ResponseHistoryRecord { + storage_key: response_history_storage_key(response_id, history_scope), + payload, + ttl: HISTORY_TTL, + }) +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use crate::formats::{context::FormatContext, registry::convert_request}; + + use super::{ + expand_previous_response_for_chat, hydrate_response_history, + record_converted_response_history, response_history_storage_key, response_history_store, + ResponseHistoryStore, + }; + + fn conversion_report_context(original_request_body: serde_json::Value) -> serde_json::Value { + json!({ + "needs_conversion": true, + "client_api_format": "openai:responses", + "provider_api_format": "openai:chat", + "original_request_body": original_request_body, + }) + } + + #[test] + fn expands_previous_response_with_assistant_call_before_tool_output() { + let report_context = conversion_report_context(json!({ + "model": "deepseek-v4-flash", + "input": [{"role": "user", "content": "scan the repository"}] + })); + record_converted_response_history( + &report_context, + &json!({ + "id": "resp_history_expand_test_1", + "status": "completed", + "output": [{ + "type": "function_call", + "id": "fc_history_expand_test_1", + "call_id": "call_history_expand_test_1", + "name": "security_scan", + "arguments": "{\"depth\":\"deep\"}" + }] + }), + ); + + let expanded = expand_previous_response_for_chat( + &json!({ + "model": "deepseek-v4-flash", + "previous_response_id": "resp_history_expand_test_1", + "input": [{ + "type": "function_call_output", + "call_id": "call_history_expand_test_1", + "output": "manifest-created" + }] + }), + None, + ) + .expect("stored previous response should expand"); + + assert!(expanded.get("previous_response_id").is_none()); + assert_eq!(expanded["input"][1]["type"], "function_call"); + assert_eq!(expanded["input"][1]["id"], "fc_history_expand_test_1"); + assert_eq!( + expanded["input"][1]["call_id"], + "call_history_expand_test_1" + ); + assert_eq!(expanded["input"][2]["type"], "function_call_output"); + assert_eq!( + expanded["input"][2]["call_id"], + "call_history_expand_test_1" + ); + } + + #[test] + fn restores_persisted_history_after_local_cache_reset() { + let report_context = json!({ + "needs_conversion": true, + "client_api_format": "openai:responses", + "provider_api_format": "openai:chat", + "api_key_id": "distributed-history-key-a", + "original_request_body": { + "model": "deepseek-v4-flash", + "input": [{"role": "user", "content": "inspect the repository"}] + } + }); + let record = record_converted_response_history( + &report_context, + &json!({ + "id": "resp_distributed_history_1", + "status": "completed", + "output": [{ + "type": "function_call", + "id": "fc_distributed_history_1", + "call_id": "call_distributed_history_1", + "name": "inspect_repository", + "arguments": "{}" + }] + }), + ) + .expect("completed conversion should produce a persistence record"); + assert_eq!( + record.storage_key, + response_history_storage_key( + "resp_distributed_history_1", + Some("distributed-history-key-a") + ) + ); + assert!(!record.storage_key.contains("distributed-history-key-a")); + assert!(!record.storage_key.contains("resp_distributed_history_1")); + + *response_history_store() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = ResponseHistoryStore::default(); + let continuation = json!({ + "model": "deepseek-v4-flash", + "previous_response_id": "resp_distributed_history_1", + "input": [{ + "type": "function_call_output", + "call_id": "call_distributed_history_1", + "output": "inspection-complete" + }] + }); + assert!(expand_previous_response_for_chat( + &continuation, + Some("distributed-history-key-a") + ) + .is_err()); + + hydrate_response_history( + "resp_distributed_history_1", + Some("distributed-history-key-a"), + &record.payload, + ) + .expect("another instance should hydrate the persisted transcript"); + let expanded = + expand_previous_response_for_chat(&continuation, Some("distributed-history-key-a")) + .expect("hydrated history should support the continuation"); + assert_eq!( + expanded["input"][1]["call_id"], + "call_distributed_history_1" + ); + assert_eq!( + expanded["input"][2]["call_id"], + "call_distributed_history_1" + ); + } + + #[test] + fn restores_history_recorded_from_redacted_chat_messages() { + record_converted_response_history( + &conversion_report_context(json!({ + "model": "deepseek-v4-flash", + "messages": [ + {"role": "user", "content": "scan the redacted repository"}, + { + "role": "assistant", + "content": null, + "tool_calls": [{ + "id": "call_redacted_previous_1", + "type": "function", + "function": { + "name": "security_scan", + "arguments": "{\"depth\":\"quick\"}" + } + }] + }, + { + "role": "tool", + "tool_call_id": "call_redacted_previous_1", + "content": "quick-scan-complete" + } + ] + })), + &json!({ + "id": "resp_history_redacted_test_1", + "status": "completed", + "output": [{ + "type": "function_call", + "id": "fc_history_redacted_test_1", + "call_id": "call_redacted_next_1", + "name": "deep_scan", + "arguments": "{\"scope\":\"changed-files\"}" + }] + }), + ); + + let expanded = expand_previous_response_for_chat( + &json!({ + "model": "deepseek-v4-flash", + "previous_response_id": "resp_history_redacted_test_1", + "input": [{ + "type": "function_call_output", + "call_id": "call_redacted_next_1", + "output": "deep-scan-complete" + }] + }), + None, + ) + .expect("redacted Chat history should expand"); + let input = expanded["input"] + .as_array() + .expect("expanded input should be an array"); + + assert_eq!(input.len(), 5); + assert_eq!(input[1]["type"], "function_call"); + assert_eq!(input[1]["call_id"], "call_redacted_previous_1"); + assert_eq!(input[2]["type"], "function_call_output"); + assert_eq!(input[2]["call_id"], "call_redacted_previous_1"); + assert_eq!(input[3]["id"], "fc_history_redacted_test_1"); + assert_eq!(input[3]["call_id"], "call_redacted_next_1"); + assert_eq!(input[4]["type"], "function_call_output"); + assert_eq!(input[4]["call_id"], "call_redacted_next_1"); + } + + #[test] + fn isolates_response_history_by_api_key_scope() { + let report_context = json!({ + "needs_conversion": true, + "client_api_format": "openai:responses", + "provider_api_format": "openai:chat", + "api_key_id": "key-history-scope-a", + "original_request_body": { + "model": "deepseek-v4-flash", + "input": [{"role": "user", "content": "private request"}] + } + }); + record_converted_response_history( + &report_context, + &json!({ + "id": "resp_history_scope_test_1", + "status": "completed", + "output": [{"type": "message", "role": "assistant", "content": []}] + }), + ); + let continuation = json!({ + "model": "deepseek-v4-flash", + "previous_response_id": "resp_history_scope_test_1", + "input": "continue" + }); + + let owner_result = convert_request( + "openai:responses", + "openai:chat", + &continuation, + &FormatContext::default().with_history_scope("key-history-scope-a"), + ); + assert!(owner_result.is_ok()); + + let other_user_error = convert_request( + "openai:responses", + "openai:chat", + &continuation, + &FormatContext::default().with_history_scope("key-history-scope-b"), + ) + .expect_err("another API key must not recover scoped history"); + assert!(matches!( + other_user_error, + crate::formats::context::FormatError::UnsupportedField { ref field, .. } + if field == "previous_response_id" + )); + } + + #[test] + fn converts_all_forty_two_tools_and_forces_late_tools() { + let tools = (0..42) + .map(|index| { + json!({ + "type": "function", + "name": format!("security_tool_{index:02}"), + "description": format!("Security tool {index}"), + "parameters": { + "type": "object", + "properties": {"path": {"type": "string"}}, + "required": ["path"] + } + }) + }) + .collect::>(); + + for forced_index in [18usize, 41usize] { + let body = json!({ + "model": "deepseek-v4-flash", + "input": [{"role": "user", "content": "run the selected scan"}], + "tools": tools, + "tool_choice": { + "type": "function", + "name": format!("security_tool_{forced_index:02}") + } + }); + let converted = convert_request( + "openai:responses", + "openai:chat", + &body, + &FormatContext::default(), + ) + .expect("all Responses tools should convert to Chat"); + let converted_tools = converted["tools"] + .as_array() + .expect("chat tools should be an array"); + + assert_eq!(converted_tools.len(), 42); + for (index, tool) in converted_tools.iter().enumerate() { + assert_eq!( + tool["function"]["name"], + json!(format!("security_tool_{index:02}")) + ); + } + assert_eq!( + converted["tool_choice"]["function"]["name"], + json!(format!("security_tool_{forced_index:02}")) + ); + } + } + + #[test] + fn restores_two_consecutive_tool_call_rounds() { + let first_request = json!({ + "model": "deepseek-v4-flash", + "input": [{"role": "user", "content": "perform a deep scan"}], + "parallel_tool_calls": true + }); + record_converted_response_history( + &conversion_report_context(first_request.clone()), + &json!({ + "id": "resp_history_multiturn_test_1", + "status": "completed", + "output": [{ + "type": "function_call", + "id": "fc_history_multiturn_test_1", + "call_id": "call_discovery_manifest_1", + "name": "create_discovery_manifest", + "arguments": "{\"root\":\"src\"}" + }] + }), + ); + + let second_request = json!({ + "model": "deepseek-v4-flash", + "previous_response_id": "resp_history_multiturn_test_1", + "input": [{ + "type": "function_call_output", + "call_id": "call_discovery_manifest_1", + "output": "manifest-1" + }], + "parallel_tool_calls": true + }); + let second_chat = convert_request( + "openai:responses", + "openai:chat", + &second_request, + &FormatContext::default(), + ) + .expect("first continuation should convert"); + assert_eq!(second_chat["messages"][1]["role"], "assistant"); + assert_eq!( + second_chat["messages"][1]["tool_calls"][0]["id"], + "call_discovery_manifest_1" + ); + assert_eq!(second_chat["messages"][2]["role"], "tool"); + assert_eq!( + second_chat["messages"][2]["tool_call_id"], + "call_discovery_manifest_1" + ); + + record_converted_response_history( + &conversion_report_context(second_request.clone()), + &json!({ + "id": "resp_history_multiturn_test_2", + "status": "completed", + "output": [ + { + "type": "function_call", + "id": "fc_history_multiturn_test_2a", + "call_id": "call_deep_scan_2a", + "name": "deep_scan", + "arguments": "{\"manifest\":\"manifest-1\"}" + }, + { + "type": "function_call", + "id": "fc_history_multiturn_test_2b", + "call_id": "call_audit_2b", + "name": "audit_results", + "arguments": "{\"manifest\":\"manifest-1\"}" + } + ] + }), + ); + + let third_chat = convert_request( + "openai:responses", + "openai:chat", + &json!({ + "model": "deepseek-v4-flash", + "previous_response_id": "resp_history_multiturn_test_2", + "input": [ + { + "type": "function_call_output", + "call_id": "call_deep_scan_2a", + "output": "scan-complete" + }, + { + "type": "function_call_output", + "call_id": "call_audit_2b", + "output": "audit-complete" + } + ] + }), + &FormatContext::default(), + ) + .expect("second continuation should convert"); + let messages = third_chat["messages"] + .as_array() + .expect("chat messages should be an array"); + assert_eq!(messages.len(), 6); + assert_eq!(messages[3]["role"], "assistant"); + assert_eq!(messages[3]["tool_calls"].as_array().map(Vec::len), Some(2)); + assert_eq!(messages[3]["tool_calls"][0]["id"], "call_deep_scan_2a"); + assert_eq!(messages[3]["tool_calls"][1]["id"], "call_audit_2b"); + assert_eq!(messages[4]["tool_call_id"], "call_deep_scan_2a"); + assert_eq!(messages[5]["tool_call_id"], "call_audit_2b"); + } +} diff --git a/crates/aether-ai/formats/src/formats/openai/responses/mod.rs b/crates/aether-ai/formats/src/formats/openai/responses/mod.rs index 7fca6658e..679584d2d 100644 --- a/crates/aether-ai/formats/src/formats/openai/responses/mod.rs +++ b/crates/aether-ai/formats/src/formats/openai/responses/mod.rs @@ -1,6 +1,7 @@ use serde_json::Value; pub mod codex; +pub(crate) mod history; pub mod request; pub mod response; pub mod spec; diff --git a/crates/aether-ai/formats/src/formats/openai/responses/response.rs b/crates/aether-ai/formats/src/formats/openai/responses/response.rs index 088ccc60d..07890af20 100644 --- a/crates/aether-ai/formats/src/formats/openai/responses/response.rs +++ b/crates/aether-ai/formats/src/formats/openai/responses/response.rs @@ -5,7 +5,7 @@ use std::{ use serde_json::{json, Map, Value}; -use super::encode_tool_result_error; +use super::{encode_tool_result_error, history::record_converted_response_history}; use crate::{ formats::context::FormatContext, @@ -28,7 +28,10 @@ pub fn from(body: &Value, _ctx: &FormatContext) -> Option { } pub fn to(response: &CanonicalResponse, ctx: &FormatContext) -> Option { - Some(to_raw(response, &ctx.report_context_value(), false)) + let report_context = ctx.report_context_value(); + let response = to_raw(response, &report_context, false); + record_converted_response_history(&report_context, &response); + Some(response) } pub fn to_compact(response: &CanonicalResponse, ctx: &FormatContext) -> Option { diff --git a/crates/aether-ai/formats/src/formats/registry.rs b/crates/aether-ai/formats/src/formats/registry.rs index 71a57747b..64d632615 100644 --- a/crates/aether-ai/formats/src/formats/registry.rs +++ b/crates/aether-ai/formats/src/formats/registry.rs @@ -135,6 +135,22 @@ pub fn convert_request( ) -> Result { let source = parse_format(source_format)?; let target = parse_format(target_format)?; + let expanded_body = if source == FormatId::OpenAiResponses && target == FormatId::OpenAiChat { + Some( + openai_responses::history::expand_previous_response_for_chat( + body, + ctx.history_scope.as_deref(), + ) + .map_err(|reason| FormatError::UnsupportedField { + format: source.as_str().to_string(), + field: "previous_response_id".to_string(), + reason, + })?, + ) + } else { + None + }; + let body = expanded_body.as_ref().unwrap_or(body); validate_openai_responses_target_contract(target_format, body)?; let mut request = parse_request(source_format, body, ctx)?; validate_runtime_request_conversion( @@ -3181,7 +3197,7 @@ mod tests { use super::{ convert_request, convert_request_pure, convert_request_pure_with_context, - convert_response_pure, FormatContext, + convert_response_pure, FormatContext, FormatError, }; use crate::formats::id::FormatId; @@ -5487,7 +5503,7 @@ mod tests { } #[test] - fn legacy_openai_responses_to_chat_does_not_leak_responses_only_extensions() { + fn runtime_openai_responses_to_chat_rejects_missing_previous_response_history() { let body = json!({ "model": "gpt-source", "input": [{"role": "user", "content": "hello"}], @@ -5496,17 +5512,19 @@ mod tests { "stream": true }); - let converted = convert_request( + let error = convert_request( "openai:responses", "openai:chat", &body, &FormatContext::default(), ) - .expect("legacy conversion should still emit a chat body"); + .expect_err("missing previous response history must fail closed"); - assert!(converted.get("stream").is_none()); - assert!(converted.get("include").is_none()); - assert!(converted.get("previous_response_id").is_none()); + assert!(matches!( + error, + FormatError::UnsupportedField { ref field, .. } + if field == "previous_response_id" + )); } #[test] diff --git a/crates/aether-ai/formats/src/formats/shared/standard_matrix.rs b/crates/aether-ai/formats/src/formats/shared/standard_matrix.rs index 5e79591cf..db550abcd 100644 --- a/crates/aether-ai/formats/src/formats/shared/standard_matrix.rs +++ b/crates/aether-ai/formats/src/formats/shared/standard_matrix.rs @@ -90,10 +90,13 @@ pub fn build_standard_request_body_with_model_directives_and_request_headers( request_headers: Option<&http::HeaderMap>, enable_model_directives: bool, ) -> Option { - let format_context = FormatContext::default() + let mut format_context = FormatContext::default() .with_mapped_model(mapped_model) .with_request_path(request_path) .with_upstream_stream(upstream_is_stream); + if let Some(history_scope) = user_api_key_id { + format_context = format_context.with_history_scope(history_scope); + } let source_api_format = compatible_source_format_for_standard_request( body_json, client_api_format, @@ -340,6 +343,8 @@ fn normalize_standard_request_to_openai_chat_request_cow<'a>( #[cfg(test)] mod tests { + use crate::formats::openai::responses::history::record_converted_response_history; + use super::{ build_standard_request_body, build_standard_request_body_from_canonical, build_standard_request_body_with_model_directives, @@ -491,6 +496,76 @@ mod tests { } } + #[test] + fn standard_request_body_scopes_previous_response_history_by_api_key() { + record_converted_response_history( + &json!({ + "needs_conversion": true, + "client_api_format": "openai:responses", + "provider_api_format": "openai:chat", + "api_key_id": "standard-history-key-a", + "original_request_body": { + "model": "source-model", + "input": [{"role": "user", "content": "inspect the repository"}] + } + }), + &json!({ + "id": "resp_standard_history_scope_1", + "status": "completed", + "output": [{ + "type": "function_call", + "id": "fc_standard_history_scope_1", + "call_id": "call_standard_history_scope_1", + "name": "inspect_repository", + "arguments": "{}" + }] + }), + ); + let continuation = json!({ + "model": "source-model", + "previous_response_id": "resp_standard_history_scope_1", + "input": [{ + "type": "function_call_output", + "call_id": "call_standard_history_scope_1", + "output": "inspection-complete" + }] + }); + + let owner = build_standard_request_body( + &continuation, + "openai:responses", + "mapped-model", + "custom", + "openai:chat", + "/v1/responses", + false, + None, + Some("standard-history-key-a"), + ) + .expect("the owning API key should restore response history"); + assert_eq!( + owner["messages"][1]["tool_calls"][0]["id"], + "call_standard_history_scope_1" + ); + assert_eq!( + owner["messages"][2]["tool_call_id"], + "call_standard_history_scope_1" + ); + + assert!(build_standard_request_body( + &continuation, + "openai:responses", + "mapped-model", + "custom", + "openai:chat", + "/v1/responses", + false, + None, + Some("standard-history-key-b"), + ) + .is_none()); + } + #[test] fn standard_request_body_stream_policy_wins_after_body_rules() { let request = json!({ diff --git a/crates/aether-ai/formats/src/formats/shared/standard_normalize.rs b/crates/aether-ai/formats/src/formats/shared/standard_normalize.rs index 470336daa..de6d1f070 100644 --- a/crates/aether-ai/formats/src/formats/shared/standard_normalize.rs +++ b/crates/aether-ai/formats/src/formats/shared/standard_normalize.rs @@ -6,6 +6,7 @@ use aether_ai_formats::formats::conversion::request::{ normalize_claude_request_to_openai_chat_request, normalize_gemini_request_to_openai_chat_request, normalize_openai_responses_request_to_openai_chat_request, + normalize_openai_responses_request_to_openai_chat_request_with_history_scope, }; use aether_ai_formats::{request_conversion_kind, FormatContext, RequestConversionKind}; use serde_json::{json, Value}; @@ -67,11 +68,16 @@ fn chat_compatible_body_for_openai_chat_endpoint(body_json: &Value) -> Option( body_json: &'a Value, client_api_format: &str, + history_scope: Option<&str>, ) -> Option> { match aether_ai_formats::normalize_api_format_alias(client_api_format).as_str() { "openai:chat" => chat_compatible_body_for_openai_chat_endpoint(body_json), "openai:responses" | "openai:responses:compact" => { - normalize_openai_responses_request_to_openai_chat_request(body_json).map(Cow::Owned) + normalize_openai_responses_request_to_openai_chat_request_with_history_scope( + body_json, + history_scope, + ) + .map(Cow::Owned) } "claude:messages" => { normalize_claude_request_to_openai_chat_request(body_json).map(Cow::Owned) @@ -325,7 +331,28 @@ pub fn build_cross_format_openai_responses_request_body_with_model_directives( upstream_is_stream: bool, enable_model_directives: bool, ) -> Option { - let chat_like_request = chat_compatible_body_for_standard_source(body_json, client_api_format)?; + build_cross_format_openai_responses_request_body_with_model_directives_and_history_scope( + body_json, + mapped_model, + client_api_format, + provider_api_format, + upstream_is_stream, + enable_model_directives, + None, + ) +} + +pub fn build_cross_format_openai_responses_request_body_with_model_directives_and_history_scope( + body_json: &Value, + mapped_model: &str, + client_api_format: &str, + provider_api_format: &str, + upstream_is_stream: bool, + enable_model_directives: bool, + history_scope: Option<&str>, +) -> Option { + let chat_like_request = + chat_compatible_body_for_standard_source(body_json, client_api_format, history_scope)?; let conversion_kind = request_conversion_kind(client_api_format, provider_api_format)?; let provider_request_body = match conversion_kind { RequestConversionKind::ToOpenAIChat => { diff --git a/crates/aether-ai/formats/src/formats/shared/stream_core/format_matrix.rs b/crates/aether-ai/formats/src/formats/shared/stream_core/format_matrix.rs index 67f946fc2..e6ad2e912 100644 --- a/crates/aether-ai/formats/src/formats/shared/stream_core/format_matrix.rs +++ b/crates/aether-ai/formats/src/formats/shared/stream_core/format_matrix.rs @@ -9,6 +9,9 @@ use crate::formats::openai::chat::stream::{ OpenAIResponsesProviderState, }; use crate::formats::openai::image::stream::OpenAiImageStreamTerminalState; +use crate::formats::openai::responses::history::{ + record_converted_response_history, ResponseHistoryRecord, +}; use crate::formats::shared::error_body::{ build_core_error_body_for_client_format, LocalCoreSyncErrorKind, }; @@ -25,6 +28,8 @@ pub struct StreamingStandardFormatMatrix { client: Option, propagated_actual_service_tier: Option, terminated: bool, + history_recorded: bool, + pending_history_record: Option, } impl StreamingStandardFormatMatrix { @@ -56,7 +61,7 @@ impl StreamingStandardFormatMatrix { client.set_actual_service_tier(propagated_actual_service_tier.as_deref()); } } - self.emit_frames(frames) + self.emit_frames(report_context, frames) } pub fn finish(&mut self, report_context: &Value) -> Result, AiSurfaceFinalizeError> { @@ -79,10 +84,11 @@ impl StreamingStandardFormatMatrix { client.set_actual_service_tier(propagated_actual_service_tier.as_deref()); } } - let mut out = self.emit_frames(frames)?; + let mut out = self.emit_frames(report_context, frames)?; if let Some(client) = self.client.as_mut() { out.extend(client.finish()?); } + self.record_response_history(report_context); Ok(out) } @@ -100,6 +106,7 @@ impl StreamingStandardFormatMatrix { fn emit_frames( &mut self, + report_context: &Value, frames: Vec, ) -> Result, AiSurfaceFinalizeError> { let Some(client) = self.client.as_mut() else { @@ -133,9 +140,30 @@ impl StreamingStandardFormatMatrix { } out.extend(client.emit(frame)?); } + self.record_response_history(report_context); Ok(out) } + fn record_response_history(&mut self, report_context: &Value) { + if self.history_recorded { + return; + } + let Some(response) = self.client.as_ref().and_then(|client| match client { + ClientStreamEmitter::OpenAIResponses(emitter) => { + emitter.completed_response_for_history() + } + _ => None, + }) else { + return; + }; + self.pending_history_record = record_converted_response_history(report_context, response); + self.history_recorded = true; + } + + pub fn take_response_history_record(&mut self) -> Option { + self.pending_history_record.take() + } + fn emit_error(&mut self, error_body: Value) -> Result, AiSurfaceFinalizeError> { let Some(client) = self.client.as_mut() else { return Ok(Vec::new()); @@ -619,6 +647,7 @@ fn parse_gemini_error(payload: &Value) -> Option<(String, Option, LocalC #[cfg(test)] mod tests { use super::{StreamingStandardFormatMatrix, StreamingStandardTerminalObserver}; + use crate::formats::{context::FormatContext, registry::convert_request}; use serde_json::{json, Value}; fn report_context(provider_api_format: &str, client_api_format: &str) -> Value { @@ -642,6 +671,121 @@ mod tests { .collect() } + #[test] + fn streamed_chat_tool_call_records_responses_continuation_history() { + let report_context = json!({ + "provider_api_format": "openai:chat", + "client_api_format": "openai:responses", + "mapped_model": "deepseek-v4-flash", + "needs_conversion": true, + "original_request_body": { + "model": "deepseek-v4-flash", + "input": [{"role": "user", "content": "perform a deep scan"}] + } + }); + let mut matrix = StreamingStandardFormatMatrix::default(); + matrix + .transform_line( + &report_context, + data_line(json!({ + "id": "chatcmpl_history_stream_test_1", + "model": "deepseek-v4-flash", + "choices": [{ + "index": 0, + "delta": { + "tool_calls": [{ + "index": 0, + "id": "call_history_stream_test_1", + "type": "function", + "function": { + "name": "create_discovery_manifest", + "arguments": "{\"root\":" + } + }] + }, + "finish_reason": Value::Null + }] + })), + ) + .expect("tool start should convert"); + matrix + .transform_line( + &report_context, + data_line(json!({ + "id": "chatcmpl_history_stream_test_1", + "model": "deepseek-v4-flash", + "choices": [{ + "index": 0, + "delta": { + "tool_calls": [{ + "index": 0, + "function": {"arguments": "\"src\"}"} + }] + }, + "finish_reason": Value::Null + }] + })), + ) + .expect("tool arguments should convert"); + let terminal = matrix + .transform_line( + &report_context, + data_line(json!({ + "id": "chatcmpl_history_stream_test_1", + "model": "deepseek-v4-flash", + "choices": [{ + "index": 0, + "delta": {}, + "finish_reason": "tool_calls" + }], + "usage": { + "prompt_tokens": 10, + "completion_tokens": 4, + "total_tokens": 14 + } + })), + ) + .expect("tool finish should convert"); + let terminal_sse = String::from_utf8(terminal).expect("SSE should be utf8"); + assert!(terminal_sse.contains("event: response.function_call_arguments.done")); + assert!(terminal_sse.contains("event: response.output_item.done")); + assert!(terminal_sse.contains("event: response.completed")); + let persisted = matrix + .take_response_history_record() + .expect("completed stream should expose one persistence record"); + assert!(persisted + .storage_key + .starts_with("ai:responses:history:v1:")); + assert!(persisted.payload.contains("resp_history_stream_test_1")); + assert!(matrix.take_response_history_record().is_none()); + + let continuation = convert_request( + "openai:responses", + "openai:chat", + &json!({ + "model": "deepseek-v4-flash", + "previous_response_id": "resp_history_stream_test_1", + "input": [{ + "type": "function_call_output", + "call_id": "call_history_stream_test_1", + "output": "manifest-created" + }] + }), + &FormatContext::default(), + ) + .expect("streamed response history should restore the next Chat request"); + assert_eq!(continuation["messages"][1]["role"], "assistant"); + assert_eq!( + continuation["messages"][1]["tool_calls"][0]["id"], + "call_history_stream_test_1" + ); + assert_eq!(continuation["messages"][2]["role"], "tool"); + assert_eq!( + continuation["messages"][2]["tool_call_id"], + "call_history_stream_test_1" + ); + } + #[test] fn transforms_provider_errors_to_openai_chat_error_bodies() { let cases = [ diff --git a/crates/aether-ai/formats/src/formats/shared/stream_rewrite.rs b/crates/aether-ai/formats/src/formats/shared/stream_rewrite.rs index 537fe9639..5bf216051 100644 --- a/crates/aether-ai/formats/src/formats/shared/stream_rewrite.rs +++ b/crates/aether-ai/formats/src/formats/shared/stream_rewrite.rs @@ -3,6 +3,7 @@ use std::collections::BTreeMap; use serde_json::{json, Map, Value}; use crate::formats::openai::image::stream::{OpenAiImageChatStreamState, OpenAiImageStreamState}; +use crate::formats::openai::responses::history::ResponseHistoryRecord; use crate::formats::openai::responses::response::ensure_modern_openai_responses_response_fields; use crate::formats::shared::model_directives::model_directive_display_model_from_report_context; use crate::formats::shared::response::{ @@ -345,6 +346,16 @@ impl AiSurfaceStreamRewriter<'_> { } } + pub fn take_response_history_record(&mut self) -> Option { + match &mut self.state { + AiSurfaceStreamRewriteState::Standard(state) => state.take_response_history_record(), + AiSurfaceStreamRewriteState::KiroToClaudeCliThenStandard { standard, .. } => { + standard.take_response_history_record() + } + _ => None, + } + } + fn transform_line(&mut self, line: Vec) -> Result, AiSurfaceFinalizeError> { match &mut self.state { AiSurfaceStreamRewriteState::EnvelopeUnwrap => { diff --git a/crates/aether-ai/formats/src/formats/shared/sync_to_stream.rs b/crates/aether-ai/formats/src/formats/shared/sync_to_stream.rs index bd5291d71..111cf6e50 100644 --- a/crates/aether-ai/formats/src/formats/shared/sync_to_stream.rs +++ b/crates/aether-ai/formats/src/formats/shared/sync_to_stream.rs @@ -12,6 +12,9 @@ use crate::formats::gemini::generate_content::stream::GeminiClientEmitter; use crate::formats::openai::chat::stream::{ OpenAIChatClientEmitter, OpenAIResponsesClientEmitter, OpenAIResponsesProviderState, }; +use crate::formats::openai::responses::history::{ + record_converted_response_history, ResponseHistoryRecord, +}; use crate::formats::shared::sse::{encode_done_sse, encode_json_sse}; use crate::formats::shared::stream_core::common::{ build_openai_chat_chunk, build_openai_chat_finish_chunk, @@ -27,6 +30,7 @@ use crate::formats::shared::AiSurfaceFinalizeError; pub struct SyncToStreamBridgeOutcome { pub sse_body: Vec, pub terminal_summary: Option, + pub response_history_record: Option, } pub fn maybe_bridge_standard_sync_json_to_stream( @@ -101,6 +105,8 @@ pub fn maybe_bridge_standard_sync_json_to_stream( &openai_responses_response, provider_actual_service_tier.clone(), ); + let response_history_record = + record_converted_response_history(&bridge_context, &openai_responses_response); let canonical_frames = build_canonical_frames_from_openai_responses_response( &openai_responses_response, &bridge_context, @@ -124,6 +130,7 @@ pub fn maybe_bridge_standard_sync_json_to_stream( Ok(Some(SyncToStreamBridgeOutcome { sse_body, terminal_summary, + response_history_record, })) } @@ -168,6 +175,7 @@ fn bridge_openai_responses_same_family_sync_json_to_stream( response, provider_actual_service_tier_from_sync_response(response, provider_api_format), ), + response_history_record: None, })) } @@ -206,6 +214,7 @@ fn maybe_bridge_openai_image_sync_json_to_stream( report_context, image_count, )), + response_history_record: None, })) } @@ -277,6 +286,7 @@ fn maybe_bridge_openai_image_sync_json_to_chat_stream( Ok(Some(SyncToStreamBridgeOutcome { sse_body, terminal_summary: Some(summary), + response_history_record: None, })) } @@ -344,6 +354,7 @@ fn maybe_bridge_openai_image_sync_json_to_responses_stream( report_context, image_count, )), + response_history_record: None, })) } @@ -769,6 +780,7 @@ fn maybe_bridge_aether_sse_response_capture_to_stream( Ok(Some(SyncToStreamBridgeOutcome { sse_body, terminal_summary, + response_history_record: None, })) } @@ -1329,6 +1341,45 @@ mod tests { ); } + #[test] + fn chat_sync_bridge_exposes_responses_history_record() { + let report_context = json!({ + "provider_api_format": "openai:chat", + "client_api_format": "openai:responses", + "needs_conversion": true, + "api_key_id": "sync-bridge-history-key", + "original_request_body": { + "model": "deepseek-v4-flash", + "input": [{"role": "user", "content": "inspect the repository"}] + } + }); + let outcome = maybe_bridge_standard_sync_json_to_stream( + &json!({ + "id": "chatcmpl_sync_history_1", + "object": "chat.completion", + "model": "deepseek-v4-flash", + "choices": [{ + "index": 0, + "message": {"role": "assistant", "content": "done"}, + "finish_reason": "stop" + }] + }), + "openai:chat", + "openai:responses", + Some(&report_context), + ) + .expect("bridge should succeed") + .expect("bridge should produce sse"); + + let history_record = outcome + .response_history_record + .expect("completed bridge should expose shared history"); + assert!(history_record + .storage_key + .starts_with("ai:responses:history:v1:")); + assert!(history_record.payload.contains("resp_sync_history_1")); + } + #[test] fn openai_sync_usage_derives_missing_input_tokens_from_total() { let usage = standardized_usage_from_openai_usage(&json!({ diff --git a/crates/aether-runtime/state/src/lib.rs b/crates/aether-runtime/state/src/lib.rs index 1e0c5e99a..5c5915b5a 100644 --- a/crates/aether-runtime/state/src/lib.rs +++ b/crates/aether-runtime/state/src/lib.rs @@ -1807,6 +1807,73 @@ mod tests { ); } + #[tokio::test] + async fn redis_runtime_instances_share_ttl_kv_across_reinitialization() { + let external_redis_url = std::env::var("AETHER_TEST_REDIS_URL") + .ok() + .filter(|value| !value.trim().is_empty()); + let redis = if external_redis_url.is_none() { + TestRedisServer::start().await + } else { + None + }; + let Some(redis_url) = + external_redis_url.or_else(|| redis.as_ref().map(|redis| redis.redis_url.clone())) + else { + return; + }; + let key_prefix = format!("aether-history-test-{}", std::process::id()); + let runtime_config = || RedisClientConfig { + url: redis_url.clone(), + key_prefix: Some(key_prefix.clone()), + }; + let writer = RuntimeState::redis(runtime_config(), Some(1_000)) + .await + .expect("writer runtime should connect"); + let reader = RuntimeState::redis(runtime_config(), Some(1_000)) + .await + .expect("reader runtime should connect"); + let history_key = "ai:responses:history:v1:shared-record"; + + writer + .kv_set( + history_key, + "persisted-history", + Some(Duration::from_secs(30)), + ) + .await + .expect("writer should persist history"); + assert_eq!( + reader + .kv_get(history_key) + .await + .expect("reader get") + .as_deref(), + Some("persisted-history") + ); + + drop(writer); + drop(reader); + let restarted = RuntimeState::redis(runtime_config(), Some(1_000)) + .await + .expect("restarted runtime should connect"); + assert_eq!( + restarted + .kv_get(history_key) + .await + .expect("restarted get") + .as_deref(), + Some("persisted-history") + ); + assert!(matches!( + restarted + .kv_ttl_seconds(history_key) + .await + .expect("history ttl"), + Some(1..=30) + )); + } + #[tokio::test] async fn redis_lock_fencing_tokens_increase_and_expired_lease_cannot_renew() { let Some((_redis, runtime)) = redis_runtime_for_test("lock-fencing").await else { diff --git a/docs/operations/redis-runtime-runbook.md b/docs/operations/redis-runtime-runbook.md index b3c3264aa..4d3184be6 100644 --- a/docs/operations/redis-runtime-runbook.md +++ b/docs/operations/redis-runtime-runbook.md @@ -31,6 +31,27 @@ crash should restore persistence in the Redis command: Expect higher tail latency when Redis persistence shares disks with Postgres or application logs. +### OpenAI Responses continuation history + +When an OpenAI Responses request is converted to an OpenAI Chat provider, +Aether stores the completed continuation transcript in `RuntimeState` under the +`ai:responses:history:v1` namespace. Records are immutable, scoped by a hashed +API key identity, limited to 8 MiB, and expire after six hours. Redis `SET` with +TTL makes completion writes atomic and idempotent. + +All gateway instances must use the same Redis URL and key prefix. This allows a +continuation request to land on another instance and allows gateway processes +to restart without losing history. `AETHER_RUNTIME_BACKEND=memory` remains a +single-process development mode and cannot provide either guarantee; multi-node +startup rejects it. + +The bundled non-persistent Redis policy survives gateway restarts but not a +Redis container or host restart. Deployments that require continuation history +to survive Redis restarts must enable the AOF/RDB policy above and mount `/data`, +or use an externally managed persistent Redis service. Monitor +`openai_response_history_read_failed`, `openai_response_history_write_failed`, +and `openai_response_history_invalid` events for backend or payload failures. + ## Latency Triage Redis `INFO commandstats` reports `latency_percentiles_usec_*` values in