diff --git a/model_gateway/src/memory/context.rs b/model_gateway/src/memory/context.rs index 7395ca9f61..cce9ed58dd 100644 --- a/model_gateway/src/memory/context.rs +++ b/model_gateway/src/memory/context.rs @@ -48,6 +48,10 @@ pub struct MemoryExecutionContext { pub subject_id: Option, pub embedding_model: Option, pub extraction_model: Option, + /// Whether STM condensation was requested by the caller. + pub stm_enabled: bool, + /// Model to use for STM condensation, if specified by the caller. + pub stm_condenser_model_id: Option, } impl MemoryExecutionContext { @@ -82,6 +86,8 @@ impl MemoryExecutionContext { subject_id: headers.subject_id.clone(), embedding_model: headers.embedding_model.clone(), extraction_model: headers.extraction_model.clone(), + stm_enabled: headers.stm_enabled && runtime.enabled, + stm_condenser_model_id: headers.stm_condenser_model_id.clone(), } } } @@ -183,4 +189,16 @@ mod tests { ); assert_eq!(ctx.extraction_model.as_deref(), Some("gpt-4.1-mini")); } + + #[test] + fn stm_enabled_gated_off_when_runtime_disabled() { + let headers = MemoryHeaderView { + stm_enabled: true, + ..MemoryHeaderView::default() + }; + + let ctx = MemoryExecutionContext::from_headers(&headers, &runtime(false)); + + assert!(!ctx.stm_enabled); + } } diff --git a/model_gateway/src/routers/common/header_utils.rs b/model_gateway/src/routers/common/header_utils.rs index 1e9fb63431..19ab6662ae 100644 --- a/model_gateway/src/routers/common/header_utils.rs +++ b/model_gateway/src/routers/common/header_utils.rs @@ -19,6 +19,10 @@ pub struct MemoryHeaderView { pub subject_id: Option, pub embedding_model: Option, pub extraction_model: Option, + /// Whether STM condensation was requested by the caller. + pub stm_enabled: bool, + /// Model to use for STM condensation, if specified by the caller. + pub stm_condenser_model_id: Option, } impl MemoryHeaderView { @@ -52,6 +56,12 @@ impl MemoryHeaderView { extraction_model: ltm_enabled .then_some(config.long_term_memory.extraction_model_id) .flatten(), + stm_enabled: config.short_term_memory.enabled, + stm_condenser_model_id: config + .short_term_memory + .enabled + .then_some(config.short_term_memory.condenser_model_id) + .flatten(), } } } diff --git a/model_gateway/src/routers/common/persistence_utils.rs b/model_gateway/src/routers/common/persistence_utils.rs index 8eb9a3bbd0..ae7f09a5a3 100644 --- a/model_gateway/src/routers/common/persistence_utils.rs +++ b/model_gateway/src/routers/common/persistence_utils.rs @@ -7,14 +7,17 @@ use openai_protocol::responses::{ generate_id, MessagePhase, ResponseInput, ResponseInputOutputItem, ResponsesRequest, StringOrContentParts, }; -use serde_json::{json, Value}; +use serde_json::{json, Map, Value}; use smg_data_connector::{ with_request_context, ConversationId, ConversationItem, ConversationItemId, - ConversationItemStorage, ConversationStorage, NewConversationItem, + ConversationItemStorage, ConversationMemoryStatus, ConversationMemoryType, + ConversationMemoryWriter, ConversationStorage, NewConversationItem, NewConversationMemory, RequestContext as StorageRequestContext, ResponseId, ResponseStorage, StoredResponse, }; use tracing::{debug, info, warn}; +use crate::memory::MemoryExecutionContext; + // ============================================================================ // Constants // ============================================================================ @@ -37,6 +40,13 @@ pub const ITEM_TYPE_FIELDS: &[(&str, &[&str])] = &[ ("function_call_output", &["call_id", "output"]), ]; +const STMO_FIRST_TURN: usize = 4; +const STMO_TURN_INTERVAL: usize = 3; + +const STMO_CFG_KEY_CONDENSER_MODEL: &str = "condenser_model"; +const STMO_CFG_KEY_LAST_INDEX: &str = "last_index"; +const STMO_CFG_KEY_TARGET_ITEM_END: &str = "target_item_end"; + // ============================================================================ // JSON Serialization // ============================================================================ @@ -44,7 +54,7 @@ pub const ITEM_TYPE_FIELDS: &[(&str, &[&str])] = &[ /// Convert a ConversationItem to JSON, extracting specified fields based on item type /// or including content as-is for standard message types. pub fn item_to_json(item: &ConversationItem) -> Value { - let mut obj = serde_json::Map::new(); + let mut obj = Map::new(); obj.insert("id".to_string(), json!(item.id.0)); obj.insert("type".to_string(), json!(item.item_type)); @@ -378,10 +388,17 @@ async fn link_items_to_conversation( /// 2. Extracts output items from the response /// 3. Stores ALL items in response storage (always) /// 4. If conversation provided, also links items to conversation +#[expect( + clippy::too_many_arguments, + reason = "threads storage handles plus request payload/context through a shared persistence entrypoint" +)] pub async fn persist_conversation_items( conversation_storage: Arc, item_storage: Arc, response_storage: Arc, + conversation_memory_writer: Arc, + memory_execution_context: MemoryExecutionContext, + conversation_user_turn_count: Option, response_json: &Value, original_body: &ResponsesRequest, request_context: Option, @@ -390,6 +407,9 @@ pub async fn persist_conversation_items( conversation_storage, item_storage, response_storage, + conversation_memory_writer, + memory_execution_context, + conversation_user_turn_count, response_json, original_body, ); @@ -399,10 +419,17 @@ pub async fn persist_conversation_items( } } +#[expect( + clippy::too_many_arguments, + reason = "keeps storage + memory execution inputs explicit in the shared persistence path" +)] async fn persist_conversation_items_inner( conversation_storage: Arc, item_storage: Arc, response_storage: Arc, + conversation_memory_writer: Arc, + memory_execution_context: MemoryExecutionContext, + conversation_user_turn_count: Option, response_json: &Value, original_body: &ResponsesRequest, ) -> Result<(), String> { @@ -463,6 +490,16 @@ async fn persist_conversation_items_inner( response_id_str, ) .await?; + enqueue_stmo_if_needed( + &conversation_memory_writer, + &memory_execution_context, + conversation_user_turn_count, + &conv_id, + &response_id, + &output_items, + input_items.len(), + ) + .await; info!( conversation_id = %conv_id.0, response_id = %response_id.0, @@ -481,3 +518,277 @@ async fn persist_conversation_items_inner( Ok(()) } + +/// Returns true when STMO should be enqueued for the current user-turn count. +/// +/// Trigger pattern: first at turn 4, then every 3 turns after that (4, 7, 10, 13, ...). +fn should_enqueue_stmo_for_current_turn(user_turns: usize) -> bool { + user_turns >= STMO_FIRST_TURN && (user_turns - 1).is_multiple_of(STMO_TURN_INTERVAL) +} + +/// Enqueues an STMO job when the current request is eligible. +/// +/// Eligibility gates: +/// - STM is enabled in `memory_execution_context` +/// - current user turn count (from assembled conversation input) is present and on trigger boundary +/// - optional `stm_condenser_model_id` is forwarded when present +/// +/// On success, this creates a `NewConversationMemory` row with type `Stmo`, status `Ready`, +/// and a minimal `memory_config` payload (`condenser_model`, `last_index`, `target_item_end`). +/// +/// This function is best-effort: enqueue failures are logged and swallowed. +async fn enqueue_stmo_if_needed( + conversation_memory_writer: &Arc, + memory_execution_context: &MemoryExecutionContext, + conversation_user_turn_count: Option, + conversation_id: &ConversationId, + response_id: &ResponseId, + output_items: &[Value], + input_item_count: usize, +) { + if !memory_execution_context.stm_enabled { + return; + } + + let Some(user_turns) = conversation_user_turn_count else { + return; + }; + + if !should_enqueue_stmo_for_current_turn(user_turns) { + return; + } + + let target_item_end = input_item_count + output_items.len(); + // STMO worker config semantics: + // - `last_index`: latest observed user-turn count at enqueue time. + // - `target_item_end`: exclusive end index for items included in this run. + let mut job_config = Map::new(); + if let Some(condenser_model) = memory_execution_context.stm_condenser_model_id.as_deref() { + job_config.insert( + STMO_CFG_KEY_CONDENSER_MODEL.to_string(), + Value::String(condenser_model.to_string()), + ); + } + job_config.insert(STMO_CFG_KEY_LAST_INDEX.to_string(), json!(user_turns)); + job_config.insert( + STMO_CFG_KEY_TARGET_ITEM_END.to_string(), + json!(target_item_end), + ); + let job_config = Value::Object(job_config).to_string(); + + let row = NewConversationMemory { + conversation_id: conversation_id.clone(), + conversation_version: None, + response_id: Some(response_id.clone()), + memory_type: ConversationMemoryType::Stmo, + status: ConversationMemoryStatus::Ready, + attempt: 0, + owner_id: None, + next_run_at: Utc::now(), + lease_until: None, + content: None, + memory_config: Some(job_config), + scope_id: None, + error_msg: None, + }; + + if let Err(err) = conversation_memory_writer.create_memory(row).await { + warn!( + conversation_id = %conversation_id.0, + response_id = %response_id.0, + error = %err, + "Failed to enqueue STMO job (best-effort; request flow continues)" + ); + } +} + +#[cfg(test)] +mod tests { + use std::sync::{Arc, Mutex}; + + use serde_json::Value; + use smg_data_connector::{ConversationMemoryId, ConversationMemoryResult}; + + use super::*; + + struct RecordingConversationMemoryWriter { + rows: Mutex>, + } + + impl RecordingConversationMemoryWriter { + fn new() -> Self { + Self { + rows: Mutex::new(Vec::new()), + } + } + } + + #[async_trait::async_trait] + impl ConversationMemoryWriter for RecordingConversationMemoryWriter { + async fn create_memory( + &self, + input: NewConversationMemory, + ) -> ConversationMemoryResult { + self.rows.lock().expect("rows mutex poisoned").push(input); + Ok(ConversationMemoryId::from("mem_test")) + } + } + + #[test] + fn stmo_turn_boundary_matches_expected_sequence() { + let cases = [ + (1, false), + (2, false), + (3, false), + (4, true), + (5, false), + (6, false), + (7, true), + (8, false), + (9, false), + (10, true), + (11, false), + (12, false), + (13, true), + ]; + + for (turn, expected) in cases { + assert_eq!( + should_enqueue_stmo_for_current_turn(turn), + expected, + "turn={turn}" + ); + } + } + + #[tokio::test] + async fn enqueue_stmo_if_needed_enqueues_expected_row_on_boundary() { + let writer = Arc::new(RecordingConversationMemoryWriter::new()); + let writer_dyn: Arc = writer.clone(); + + let memory_execution_context = MemoryExecutionContext { + stm_enabled: true, + stm_condenser_model_id: Some("condense-1".to_string()), + ..MemoryExecutionContext::default() + }; + + let output_items = vec![json!({ "type": "message", "role": "assistant" })]; + let conversation_id = ConversationId::from("conv_test"); + let response_id = ResponseId::from("resp_test"); + + enqueue_stmo_if_needed( + &writer_dyn, + &memory_execution_context, + Some(4), + &conversation_id, + &response_id, + &output_items, + 4, + ) + .await; + + let rows = writer.rows.lock().expect("rows mutex poisoned"); + assert_eq!(rows.len(), 1, "should enqueue exactly one STMO row"); + + let row = &rows[0]; + assert_eq!(row.conversation_id, conversation_id); + assert_eq!(row.response_id, Some(response_id)); + assert_eq!(row.memory_type, ConversationMemoryType::Stmo); + assert_eq!(row.status, ConversationMemoryStatus::Ready); + + let config = row + .memory_config + .as_deref() + .expect("memory_config should be set"); + let config_json: Value = + serde_json::from_str(config).expect("memory_config must be valid JSON"); + + assert_eq!( + config_json + .get(STMO_CFG_KEY_CONDENSER_MODEL) + .and_then(Value::as_str), + Some("condense-1") + ); + assert_eq!( + config_json + .get(STMO_CFG_KEY_LAST_INDEX) + .and_then(Value::as_u64), + Some(4) + ); + assert_eq!( + config_json + .get(STMO_CFG_KEY_TARGET_ITEM_END) + .and_then(Value::as_u64), + Some(5) + ); + } + + #[tokio::test] + async fn enqueue_stmo_if_needed_skips_when_turn_count_missing() { + let writer = Arc::new(RecordingConversationMemoryWriter::new()); + let writer_dyn: Arc = writer.clone(); + + let memory_execution_context = MemoryExecutionContext { + stm_enabled: true, + stm_condenser_model_id: Some("condense-1".to_string()), + ..MemoryExecutionContext::default() + }; + + enqueue_stmo_if_needed( + &writer_dyn, + &memory_execution_context, + None, + &ConversationId::from("conv_test"), + &ResponseId::from("resp_test"), + &[], + 0, + ) + .await; + + let rows = writer.rows.lock().expect("rows mutex poisoned"); + assert!( + rows.is_empty(), + "should not enqueue when turn count is absent" + ); + } + + #[tokio::test] + async fn enqueue_stmo_if_needed_enqueues_without_condenser_model() { + let writer = Arc::new(RecordingConversationMemoryWriter::new()); + let writer_dyn: Arc = writer.clone(); + + let memory_execution_context = MemoryExecutionContext { + stm_enabled: true, + stm_condenser_model_id: None, + ..MemoryExecutionContext::default() + }; + + enqueue_stmo_if_needed( + &writer_dyn, + &memory_execution_context, + Some(4), + &ConversationId::from("conv_test"), + &ResponseId::from("resp_test"), + &[json!({ "type": "message", "role": "assistant" })], + 4, + ) + .await; + + let rows = writer.rows.lock().expect("rows mutex poisoned"); + assert_eq!(rows.len(), 1, "should enqueue without condenser model"); + + let config = rows[0] + .memory_config + .as_deref() + .expect("memory_config should be set"); + let config_json: Value = + serde_json::from_str(config).expect("memory_config must be valid JSON"); + assert!(config_json.get(STMO_CFG_KEY_CONDENSER_MODEL).is_none()); + assert_eq!( + config_json + .get(STMO_CFG_KEY_LAST_INDEX) + .and_then(Value::as_u64), + Some(4) + ); + } +} diff --git a/model_gateway/src/routers/grpc/common/responses/utils.rs b/model_gateway/src/routers/grpc/common/responses/utils.rs index 5221fb283d..c585273bb0 100644 --- a/model_gateway/src/routers/grpc/common/responses/utils.rs +++ b/model_gateway/src/routers/grpc/common/responses/utils.rs @@ -9,13 +9,14 @@ use openai_protocol::{ }; use serde_json::to_value; use smg_data_connector::{ - ConversationItemStorage, ConversationStorage, RequestContext as StorageRequestContext, - ResponseStorage, + ConversationItemStorage, ConversationStorage, NoOpConversationMemoryWriter, + RequestContext as StorageRequestContext, ResponseStorage, }; use smg_mcp::{McpOrchestrator, McpServerBinding}; use tracing::{debug, error, warn}; use crate::{ + memory::MemoryExecutionContext, routers::{ common::{ mcp_utils::ensure_request_mcp_client, persistence_utils::persist_conversation_items, @@ -154,6 +155,9 @@ pub(crate) async fn persist_response_if_needed( conversation_storage, conversation_item_storage, response_storage, + Arc::new(NoOpConversationMemoryWriter::new()), + MemoryExecutionContext::default(), + None, &response_json, original_request, request_context, diff --git a/model_gateway/src/routers/openai/context.rs b/model_gateway/src/routers/openai/context.rs index 9e94ccd24c..1efee4392d 100644 --- a/model_gateway/src/routers/openai/context.rs +++ b/model_gateway/src/routers/openai/context.rs @@ -132,6 +132,7 @@ pub struct PayloadState { pub struct ResponsesPayloadState { pub previous_response_id: Option, pub existing_mcp_list_tools_labels: Vec, + pub conversation_user_turn_count: Option, } impl RequestContext { @@ -260,6 +261,7 @@ pub struct StorageHandles { pub conversation_memory_writer: Arc, pub request_context: Option, pub memory_execution_context: MemoryExecutionContext, + pub conversation_user_turn_count: Option, } pub struct OwnedStreamingContext { @@ -313,6 +315,7 @@ impl RequestContext { conversation_memory_writer, request_context: self.storage_request_context, memory_execution_context: self.memory_execution_context, + conversation_user_turn_count: responses_payload_state.conversation_user_turn_count, }, }) } diff --git a/model_gateway/src/routers/openai/responses/history.rs b/model_gateway/src/routers/openai/responses/history.rs index b457e1c2cd..ed55950a4f 100644 --- a/model_gateway/src/routers/openai/responses/history.rs +++ b/model_gateway/src/routers/openai/responses/history.rs @@ -12,20 +12,16 @@ use openai_protocol::{ }; use serde_json::Value; use smg_data_connector::{ConversationId, ListParams, ResponseId, ResponseStorageError, SortOrder}; -use tracing::{debug, warn}; +use tracing::warn; use super::super::context::ResponsesComponents; use crate::{ observability::metrics::{metrics_labels, Metrics}, - routers::{ - common::{ - header_utils::ConversationMemoryConfig, persistence_utils::split_stored_message_content, - }, - error, - }, + routers::{common::persistence_utils::split_stored_message_content, error}, }; const MAX_CONVERSATION_HISTORY_ITEMS: usize = 100; +const USER_ROLE: &str = "user"; pub(crate) struct LoadedInputHistory { pub previous_response_id: Option, @@ -294,67 +290,69 @@ fn append_current_input( } } -/// Memory hook entrypoint for Responses API. +/// Counts conversation user turns from the assembled Responses input payload. /// -/// This is intentionally a no-op in this PR: it confirms header parsing is -/// connected to request flow and logs activation state for follow-up retrieval work. -pub(crate) fn inject_memory_context( - config: &ConversationMemoryConfig, - _request_body: &mut ResponsesRequest, -) { - if config.long_term_memory.enabled { - debug!( - has_subject_id = config.long_term_memory.subject_id.is_some(), - has_embedding_model = config.long_term_memory.embedding_model_id.is_some(), - has_extraction_model = config.long_term_memory.extraction_model_id.is_some(), - "LTM recall requested - retrieval not yet implemented" - ); - } - - if config.short_term_memory.enabled { - debug!( - has_condenser_model = config.short_term_memory.condenser_model_id.is_some(), - "STM recall requested - retrieval not yet implemented" - ); +/// Rules: +/// - `ResponseInput::Text` counts as one user turn. +/// - `ResponseInput::Items` counts message-like items with role `"user"`. +/// - Non-message items are ignored. +/// +/// This function is resilient by design: unknown or non-user items are treated +/// as non-turns and skipped. +pub(crate) fn count_conversation_user_turns(input: &ResponseInput) -> usize { + match input { + ResponseInput::Text(_) => 1, + ResponseInput::Items(items) => items + .iter() + .filter(|item| match item { + ResponseInputOutputItem::Message { role, .. } + | ResponseInputOutputItem::SimpleInputMessage { role, .. } => { + role.eq_ignore_ascii_case(USER_ROLE) + } + _ => false, + }) + .count(), } } #[cfg(test)] mod tests { - use openai_protocol::responses::{ResponseInput, ResponsesRequest}; - - use super::inject_memory_context; - use crate::routers::common::header_utils::{ - ConversationMemoryConfig, LongTermMemoryConfig, ShortTermMemoryConfig, + use openai_protocol::responses::{ + ResponseInput, ResponseInputOutputItem, SimpleInputMessageTypeTag, StringOrContentParts, }; + use super::count_conversation_user_turns; + + #[test] + fn count_conversation_user_turns_counts_text_as_single_turn() { + let input = ResponseInput::Text("hello".to_string()); + assert_eq!(count_conversation_user_turns(&input), 1); + } + #[test] - fn inject_memory_context_is_no_op_for_now() { - let config = ConversationMemoryConfig { - long_term_memory: LongTermMemoryConfig { - enabled: true, - policy: None, - subject_id: Some("subj-1".to_string()), - embedding_model_id: Some("embed-1".to_string()), - extraction_model_id: Some("extract-1".to_string()), + fn count_conversation_user_turns_counts_user_messages_only() { + let input = ResponseInput::Items(vec![ + ResponseInputOutputItem::Message { + id: "m1".to_string(), + role: "user".to_string(), + content: vec![], + status: None, + phase: None, }, - short_term_memory: ShortTermMemoryConfig { - enabled: true, - condenser_model_id: Some("condense-1".to_string()), + ResponseInputOutputItem::SimpleInputMessage { + content: StringOrContentParts::String("hi".to_string()), + role: "assistant".to_string(), + r#type: Some(SimpleInputMessageTypeTag::Message), + phase: None, }, - }; - let mut request = ResponsesRequest { - input: ResponseInput::Text("hello".to_string()), - ..Default::default() - }; - - inject_memory_context(&config, &mut request); + ResponseInputOutputItem::SimpleInputMessage { + content: StringOrContentParts::String("again".to_string()), + role: "USER".to_string(), + r#type: Some(SimpleInputMessageTypeTag::Message), + phase: None, + }, + ]); - match request.input { - ResponseInput::Text(text) => assert_eq!(text, "hello"), - ResponseInput::Items(_) => { - panic!("request input should remain unchanged for no-op hook") - } - } + assert_eq!(count_conversation_user_turns(&input), 2); } } diff --git a/model_gateway/src/routers/openai/responses/non_streaming.rs b/model_gateway/src/routers/openai/responses/non_streaming.rs index b3b3efd2da..10f2833f2e 100644 --- a/model_gateway/src/routers/openai/responses/non_streaming.rs +++ b/model_gateway/src/routers/openai/responses/non_streaming.rs @@ -40,6 +40,7 @@ pub async fn handle_non_streaming_response(mut ctx: RequestContext) -> Response let ResponsesPayloadState { previous_response_id, existing_mcp_list_tools_labels, + conversation_user_turn_count, } = ctx.take_responses_payload().unwrap_or_default(); let original_body = match ctx.responses_request() { @@ -157,15 +158,19 @@ pub async fn handle_non_streaming_response(mut ctx: RequestContext) -> Response previous_response_id.as_deref(), ); - if let (Some(conv_storage), Some(item_storage), Some(resp_storage)) = ( + if let (Some(conv_storage), Some(item_storage), Some(resp_storage), Some(mem_writer)) = ( ctx.components.conversation_storage(), ctx.components.conversation_item_storage(), ctx.components.response_storage(), + ctx.components.conversation_memory_writer(), ) { if let Err(err) = persist_conversation_items( conv_storage.clone(), item_storage.clone(), resp_storage.clone(), + mem_writer.clone(), + ctx.memory_execution_context.clone(), + conversation_user_turn_count, &response_json, original_body, ctx.storage_request_context.clone(), diff --git a/model_gateway/src/routers/openai/responses/route.rs b/model_gateway/src/routers/openai/responses/route.rs index cf131cae08..f424ffae88 100644 --- a/model_gateway/src/routers/openai/responses/route.rs +++ b/model_gateway/src/routers/openai/responses/route.rs @@ -25,10 +25,7 @@ use crate::{ middleware::TenantRequestMeta, observability::metrics::{bool_to_static_str, metrics_labels, Metrics}, routers::{ - common::{ - header_utils::extract_conversation_memory_config, - worker_selection::{SelectWorkerRequest, WorkerSelector}, - }, + common::worker_selection::{SelectWorkerRequest, WorkerSelector}, error, }, worker::{Endpoint, ProviderType, WorkerRegistry}, @@ -128,10 +125,6 @@ pub(in crate::routers::openai) async fn route_responses( Err(response) => return response, }; - if let Some(memory_config) = extract_conversation_memory_config(headers) { - super::history::inject_memory_context(&memory_config, &mut request_body); - } - request_body.store = Some(false); if let ResponseInput::Items(ref mut items) = request_body.input { items.retain(|item| !matches!(item, ResponseInputOutputItem::Reasoning { .. })); @@ -186,9 +179,18 @@ pub(in crate::routers::openai) async fn route_responses( json: payload, url: format!("{}/v1/responses", worker.url()), }); + let conversation_user_turn_count = if ctx.memory_execution_context.stm_enabled { + Some(super::history::count_conversation_user_turns( + &request_body.input, + )) + } else { + None + }; + ctx.state.responses_payload = Some(ResponsesPayloadState { previous_response_id: loaded_history.previous_response_id, existing_mcp_list_tools_labels: loaded_history.existing_mcp_list_tools_labels, + conversation_user_turn_count, }); let response = if ctx.is_streaming() { diff --git a/model_gateway/src/routers/openai/responses/streaming.rs b/model_gateway/src/routers/openai/responses/streaming.rs index b9599587af..ec997d24d0 100644 --- a/model_gateway/src/routers/openai/responses/streaming.rs +++ b/model_gateway/src/routers/openai/responses/streaming.rs @@ -627,6 +627,9 @@ pub(super) async fn handle_simple_streaming_passthrough( storage.conversation.clone(), storage.conversation_item.clone(), storage.response.clone(), + storage.conversation_memory_writer.clone(), + storage.memory_execution_context.clone(), + storage.conversation_user_turn_count, &response_json, &original_request, storage.request_context.clone(), @@ -924,6 +927,9 @@ pub(super) fn handle_streaming_with_tool_interception( storage.conversation.clone(), storage.conversation_item.clone(), storage.response.clone(), + storage.conversation_memory_writer.clone(), + storage.memory_execution_context.clone(), + storage.conversation_user_turn_count, &response_json, &original_request, storage.request_context.clone(),