fix(responses): preserve replay state and prompt cache lineage

src/apps/desktop/src/api/btw_api.rs

@@ -12,9 +12,15 @@ use tauri::State;
 
 use crate::api::app_state::AppState;
 
-use bitfun_core::agentic::coordination::ConversationCoordinator;
+use bitfun_core::agentic::coordination::{
+    ConversationCoordinator, DialogSubmissionPolicy, DialogTriggerSource,
+};
 use bitfun_core::agentic::image_analysis::ImageContextData;
 
+fn desktop_btw_submission_policy() -> DialogSubmissionPolicy {
+    DialogSubmissionPolicy::for_source(DialogTriggerSource::DesktopUi)
+}
+
 #[derive(Debug, Clone, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct BtwAskStreamRequest {
@@ -105,6 +111,7 @@ pub async fn btw_ask_stream(
             &child_session_id,
             child_session_name.as_deref(),
             &request.question,
+            desktop_btw_submission_policy(),
             model_id.as_deref(),
             image_contexts,
             request.parent_dialog_turn_id.as_deref(),
@@ -150,3 +157,16 @@ pub async fn btw_ask_stream(
 
     Ok(BtwAskStreamResponse { ok: true })
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn btw_turns_use_the_desktop_chat_output_surface() {
+        assert_eq!(
+            desktop_btw_submission_policy(),
+            DialogSubmissionPolicy::for_source(DialogTriggerSource::DesktopUi)
+        );
+    }
+}

src/crates/adapters/ai-adapters/Cargo.toml

@@ -36,7 +36,8 @@ log = { workspace = true }
 reqwest = { workspace = true, features = ["http2", "json", "rustls", "socks", "stream"] }
 serde = { workspace = true }
 serde_json = { workspace = true }
-sha2 = { workspace = true, optional = true }
+sha2 = { workspace = true }
+hex = { workspace = true }
 tokio = { workspace = true, features = ["macros", "rt", "sync", "time"] }
 tokio-stream = { workspace = true }
 tokio-util = { workspace = true }
@@ -62,7 +63,6 @@ subscription-auth = [
     "dep:keyring-core",
     "dep:libc",
     "reqwest/form",
-    "dep:sha2",
     "tokio/fs",
     "tokio/io-util",
     "tokio/net",

src/crates/adapters/ai-adapters/src/client.rs

@@ -230,6 +230,7 @@ impl AIClient {
             extra_body,
             SEND_MESSAGE_STREAM_ATTEMPTS,
             trace,
+            None,
         )
         .await
     }
@@ -244,14 +245,44 @@ impl AIClient {
         messages: Vec<Message>,
         tools: Option<Vec<ToolDefinition>>,
         trace: Option<ModelExchangeTraceConfig>,
+    ) -> Result<StreamResponse> {
+        self.send_message_stream_once_with_request_context(messages, tools, None, trace)
+            .await
+    }
+
+    /// Open one model stream without an adapter-owned retry loop, carrying
+    /// provider-neutral request-scoped facts to adapters that support them.
+    pub async fn send_message_stream_once_with_request_context(
+        &self,
+        messages: Vec<Message>,
+        tools: Option<Vec<ToolDefinition>>,
+        request_context: Option<ModelRequestContext>,
+        trace: Option<ModelExchangeTraceConfig>,
     ) -> Result<StreamResponse> {
         let custom_body = self.config.custom_request_body.clone();
+        if matches!(
+            ApiFormat::parse(&self.config.format)?,
+            ApiFormat::OpenAIResponses
+        ) {
+            return openai::responses::send_stream(
+                self,
+                messages,
+                tools,
+                custom_body,
+                1,
+                trace,
+                request_context,
+            )
+            .await;
+        }
+
         self.send_message_stream_with_extra_body_and_max_attempts(
             messages,
             tools,
             custom_body,
             1,
             trace,
+            None,
         )
         .await
     }
@@ -282,11 +313,31 @@ impl AIClient {
         tools: Option<Vec<ToolDefinition>>,
         trace: Option<ModelExchangeTraceConfig>,
     ) -> Result<GeminiResponse> {
-        let custom_body = self.config.custom_request_body.clone();
-        self.send_message_with_extra_body_and_trace(messages, tools, custom_body, trace)
+        self.send_message_with_trace_and_request_context(messages, tools, None, trace)
             .await
     }
 
+    /// Aggregate one model response while carrying provider-neutral
+    /// request-scoped facts to adapters that support them.
+    pub async fn send_message_with_trace_and_request_context(
+        &self,
+        messages: Vec<Message>,
+        tools: Option<Vec<ToolDefinition>>,
+        request_context: Option<ModelRequestContext>,
+        trace: Option<ModelExchangeTraceConfig>,
+    ) -> Result<GeminiResponse> {
+        let custom_body = self.config.custom_request_body.clone();
+        self.send_message_with_extra_body_trace_and_max_attempts(
+            messages,
+            tools,
+            custom_body,
+            request_context,
+            trace,
+            SEND_MESSAGE_STREAM_ATTEMPTS,
+        )
+        .await
+    }
+
     pub async fn send_message_with_extra_body_and_trace(
         &self,
         messages: Vec<Message>,
@@ -298,6 +349,7 @@ impl AIClient {
             messages,
             tools,
             extra_body,
+            None,
             trace,
             SEND_MESSAGE_STREAM_ATTEMPTS,
         )
@@ -309,6 +361,7 @@ impl AIClient {
         messages: Vec<Message>,
         tools: Option<Vec<ToolDefinition>>,
         extra_body: Option<serde_json::Value>,
+        request_context: Option<ModelRequestContext>,
         trace: Option<ModelExchangeTraceConfig>,
         max_attempts: usize,
     ) -> Result<GeminiResponse> {
@@ -320,6 +373,7 @@ impl AIClient {
                     extra_body.clone(),
                     1,
                     trace.clone(),
+                    request_context.clone(),
                 )
                 .await
             {
@@ -381,14 +435,23 @@ impl AIClient {
         extra_body: Option<serde_json::Value>,
         max_tries: usize,
         trace: Option<ModelExchangeTraceConfig>,
+        request_context: Option<ModelRequestContext>,
     ) -> Result<StreamResponse> {
         match ApiFormat::parse(&self.config.format)? {
             ApiFormat::OpenAIChat => {
                 openai::chat::send_stream(self, messages, tools, extra_body, max_tries, trace).await
             }
             ApiFormat::OpenAIResponses => {
-                openai::responses::send_stream(self, messages, tools, extra_body, max_tries, trace)
-                    .await
+                openai::responses::send_stream(
+                    self,
+                    messages,
+                    tools,
+                    extra_body,
+                    max_tries,
+                    trace,
+                    request_context,
+                )
+                .await
             }
             ApiFormat::Anthropic => {
                 anthropic::request::send_stream(self, messages, tools, extra_body, max_tries, trace)
@@ -427,6 +490,7 @@ impl AIClient {
             tools,
             custom_body,
             None,
+            None,
             max_attempts,
         )
         .await

src/crates/adapters/ai-adapters/src/client/healthcheck.rs

@@ -219,6 +219,7 @@ pub(crate) async fn test_image_input_connection(
         name: None,
         is_error: None,
         tool_image_attachments: None,
+        model_response_replay: None,
     }];
 
     match client

src/crates/adapters/ai-adapters/src/client/response_aggregator.rs

@@ -33,6 +33,7 @@ pub(crate) async fn aggregate_stream_response(
                     tool_call_completion: _,
                     finish_reason: chunk_finish_reason,
                     provider_metadata: chunk_provider_metadata,
+                    model_response_replay: _,
                 } = chunk;
 
                 if let Some(text) = text {

src/crates/adapters/ai-adapters/src/lib.rs

@@ -24,6 +24,6 @@ pub use trace::{
 };
 pub use types::{
     resolve_request_url, AIConfig, ConnectionTestMessageCode, ConnectionTestResult, GeminiResponse,
-    GeminiUsage, Message, ProxyConfig, ReasoningPresetAction, ReasoningPresetDescriptor,
-    RemoteModelInfo, ToolCall, ToolDefinition, ToolImageAttachment,
+    GeminiUsage, Message, ModelRequestContext, ProxyConfig, ReasoningPresetAction,
+    ReasoningPresetDescriptor, RemoteModelInfo, ToolCall, ToolDefinition, ToolImageAttachment,
 };

src/crates/adapters/ai-adapters/src/providers/anthropic/message_converter.rs

@@ -246,6 +246,7 @@ mod tests {
             name: None,
             is_error: None,
             tool_image_attachments: None,
+            model_response_replay: None,
         };
 
         let (_, messages) = AnthropicMessageConverter::convert_messages(vec![msg]);
@@ -288,6 +289,7 @@ mod tests {
             name: None,
             is_error: None,
             tool_image_attachments: None,
+            model_response_replay: None,
         };
 
         let (_, messages) = AnthropicMessageConverter::convert_messages(vec![

src/crates/adapters/ai-adapters/src/providers/gemini/message_converter.rs

@@ -673,6 +673,7 @@ mod tests {
                 name: None,
                 is_error: None,
                 tool_image_attachments: None,
+                model_response_replay: None,
             },
             Message {
                 role: "tool".to_string(),
@@ -684,6 +685,7 @@ mod tests {
                 name: Some("get_weather".to_string()),
                 is_error: None,
                 tool_image_attachments: None,
+                model_response_replay: None,
             },
         ];
 
@@ -726,6 +728,7 @@ mod tests {
             name: None,
             is_error: None,
             tool_image_attachments: None,
+            model_response_replay: None,
         }];
 
         let (_, contents) =
@@ -764,6 +767,7 @@ mod tests {
             name: None,
             is_error: None,
             tool_image_attachments: None,
+            model_response_replay: None,
         }];
 
         let (_, contents) = GeminiMessageConverter::convert_messages(messages, "gemini-2.5-pro");

src/crates/adapters/ai-adapters/src/providers/openai/codex_chatgpt.rs

@@ -208,7 +208,14 @@ pub(crate) async fn send_stream(
         trace,
         || common::apply_headers(client, client.client.post(&url)),
         move |response, tx, tx_raw, remaining_ttft_timeout| {
-            handle_responses_stream(response, tx, tx_raw, remaining_ttft_timeout, idle_timeout)
+            handle_responses_stream(
+                response,
+                tx,
+                tx_raw,
+                remaining_ttft_timeout,
+                idle_timeout,
+                None,
+            )
         },
     )
     .await