From bb7fee81abe69d8c953b883317c27f85cf720e36 Mon Sep 17 00:00:00 2001 From: wsp Date: Sun, 25 Oct 2026 22:33:10 +0800 Subject: [PATCH] fix(ai): Preserve conversations across model switches Project signed and opaque history for the selected model and account instead of blocking manual switches or pinning fixed accounts to old history. Preserve visible messages, completed tool exchanges, and the original records so compatible state can be replayed when switching back. Keep replay validation after projection and log stripped message counts. Cover provider changes, fixed accounts, legacy provenance, and returning to the original model with focused regression tests. Fixes: #3537 --- .../core/src/agentic/execution/AGENTS.md | 11 +- .../src/agentic/execution/round_executor.rs | 196 +++++++++++++++++- .../ai/channel_client_source.rs | 135 ++++++++++++ 3 files changed, 332 insertions(+), 10 deletions(-) diff --git a/src/crates/assembly/core/src/agentic/execution/AGENTS.md b/src/crates/assembly/core/src/agentic/execution/AGENTS.md index 26ea267362..15d518c054 100644 --- a/src/crates/assembly/core/src/agentic/execution/AGENTS.md +++ b/src/crates/assembly/core/src/agentic/execution/AGENTS.md @@ -70,7 +70,7 @@ For cross-provider channels, run the real factory/round mock and replay-binding checks with the same minimal features: ```bash -cargo test --locked -p openbitfun-core --no-default-features --features agent-runtime,git,subscription-auth --lib multi_channel_pool +cargo test --locked -p openbitfun-core --no-default-features --features agent-runtime,git,subscription-auth --lib -- multi_channel_pool manual_model_switch_ cargo test --locked -p openbitfun-core --no-default-features --features agent-runtime,git,subscription-auth --lib infrastructure::ai::channel_client_source::tests ``` @@ -99,9 +99,12 @@ signed reasoning text or opaque replay; text, tool calls and results stay). Every account or channel switch other than same-account authentication recovery uses portable history. Session records keep the originals. Anthropic requests whose last assistant tool turn lacks a signed thinking block run without -thinking (adapter-owned). Fixed accounts and unrouted models keep the original -replay binding and its `recovery_constraint` diagnostics; configuration/storage -errors remain strict. +thinking (adapter-owned). Fixed accounts and unrouted models select the configured +destination without pinning it to historical account provenance. Before every +send, project incompatible or unproven signed/opaque state per message against +the actual channel fingerprint and account, then validate remaining replay state. +Compatible state and durable history remain intact; legacy account-only provenance +can still replay to its proven seat. Configuration/storage errors remain strict. Native turn initialization and manual compression scaffold preparation resolve diff --git a/src/crates/assembly/core/src/agentic/execution/round_executor.rs b/src/crates/assembly/core/src/agentic/execution/round_executor.rs index aa55244fab..eef80572e5 100644 --- a/src/crates/assembly/core/src/agentic/execution/round_executor.rs +++ b/src/crates/assembly/core/src/agentic/execution/round_executor.rs @@ -37,7 +37,7 @@ use crate::util::elapsed_ms_u64; use crate::util::errors::{OpenBitFunError, OpenBitFunResult}; use crate::util::types::Message as AIMessage; use crate::util::types::ToolDefinition; -use log::{debug, error, warn}; +use log::{debug, error, info, warn}; use openbitfun_agent_runtime::turn_cancellation::DialogTurnCancellationTokenStore; use openbitfun_agent_tools::{parse_call_deferred_tool_input, CALL_DEFERRED_TOOL_NAME}; use openbitfun_ai_adapters::{ @@ -762,7 +762,8 @@ impl RoundExecutor { // Routed rounds may switch account or channel, so signed reasoning is // a cache preference, not a binding: keep the latest producer's state // and degrade it to portable history whenever that producer is not - // available. Fixed rounds keep the original replay contract. + // available. Fixed rounds select the configured destination first; + // all attempts project private replay state before sending. let routed = channels.is_some() || account_source.as_ref().is_some_and(|source| { matches!( @@ -779,8 +780,13 @@ impl RoundExecutor { ai_messages = crate::infrastructure::ai::channel_client_source::ChannelClientSource::portable_history(&ai_messages); } } - let mut replay_origin = - crate::infrastructure::ai::account_client_source::opaque_replay_origin(&ai_messages); + let mut replay_origin = if routed { + crate::infrastructure::ai::account_client_source::opaque_replay_origin(&ai_messages) + } else { + // A manual model/account selection must not be pinned back to the + // producer of old history. Its private state is projected below. + Ok(None) + }; let mut has_opaque_replay = crate::infrastructure::ai::channel_client_source::ChannelClientSource::has_nonportable_history(&ai_messages); if let Some(channels) = &channels { tokio::select! { @@ -990,13 +996,26 @@ impl RoundExecutor { } } - crate::infrastructure::ai::channel_client_source::ChannelClientSource::validate_replay_binding(&ai_messages, &context.model_config_id, &ai_client, channels.is_some()) - .map_err(|error| OpenBitFunError::AIClient(error.to_string()).with_channel_bound_replay())?; let account = selected_origin.clone().or_else(|| { ai_client .subscription_provider_key() .and_then(|provider| ai_client.subscription_account_origin(provider)) }); + let stripped = crate::infrastructure::ai::channel_client_source::ChannelClientSource::project_replay_for_target( + &mut ai_messages, &context.model_config_id, &ai_client, account.as_ref(), + ); + if stripped > 0 { + has_opaque_replay = crate::infrastructure::ai::channel_client_source::ChannelClientSource::has_nonportable_history(&ai_messages); + replay_origin = + crate::infrastructure::ai::account_client_source::opaque_replay_origin( + &ai_messages, + ) + .ok() + .flatten(); + info!("Projected incompatible model history: session_id={}, model_config_id={}, stripped_messages={}, reason=unproven_or_changed_replay_identity", context.session_id, context.model_config_id, stripped); + } + crate::infrastructure::ai::channel_client_source::ChannelClientSource::validate_replay_binding(&ai_messages, &context.model_config_id, &ai_client, channels.is_some()) + .map_err(|error| OpenBitFunError::AIClient(error.to_string()).with_channel_bound_replay())?; let mut usage_context = openbitfun_services_core::token_usage::types::ModelRequestUsageContext { version: 1, @@ -6817,6 +6836,171 @@ pub(super) mod tests { } } + #[tokio::test] + async fn manual_model_switch_projects_history_before_sending() { + for format in ["responses", "anthropic", "gemini", "fixed-account"] { + let events = match format { + "anthropic" => vec![ + json!({"type":"message_start","message":{"id":"continued","type":"message","role":"assistant","model":"target","usage":{"input_tokens":20,"output_tokens":0}}}), + json!({"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}), + json!({"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Continued"}}), + json!({"type":"content_block_stop","index":0}), + json!({"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":5}}), + json!({"type":"message_stop"}), + ], + "gemini" => vec![ + json!({"candidates":[{"content":{"role":"model","parts":[{"text":"Continued"}]},"finishReason":"STOP"}]}), + ], + _ => vec![ + json!({"type":"response.output_text.delta","delta":"Continued"}), + json!({"type":"response.completed","response":{"id":"continued","output":[]}}), + ], + }; + let reply = ( + 200, + events + .iter() + .map(|event| { + format!( + "event: {}\ndata: {event}\n\n", + event["type"].as_str().unwrap_or("message") + ) + }) + .collect::(), + ); + let server = + RetryTestServer::new(vec![reply; if format == "responses" { 2 } else { 1 }]); + let mut client = (*server.client()).clone(); + client.config.format = if format == "fixed-account" { + "responses" + } else { + format + } + .into(); + client.config.model = if format == "gemini" { + "gemini-3-pro" + } else { + "target-model" + } + .into(); + let client = Arc::new(client); + let mut context = test_round_context(); + let mut original_client = (*client).clone(); + original_client.config.format = "responses".into(); + original_client.config.model = "original-model".into(); + let mut previous = + super::AIMessage::assistant_with_tools(vec![openbitfun_core_types::ToolCall { + id: "already-executed".into(), + name: "write_file".into(), + arguments: json!({"path":"result.txt", "content":"saved once"}), + raw_arguments: None, + }]); + previous.content = Some("Saved the requested file".into()); + previous.reasoning_content = Some("Original private reasoning".into()); + previous.thinking_signature = Some("original-signature".into()); + previous.model_response_replay = Some(openbitfun_core_types::ModelResponseReplay { + protocol: "openai_responses".into(), + channel_origin: Some(context.model_config_id.clone()), + channel_fingerprint: Some(original_client.channel_replay_fingerprint()), + account_origin: None, + items: vec![ + openbitfun_core_types::ModelResponseReplayItem::OpaqueReasoning { + item_id: Some("original-reasoning".into()), + summary: vec![], + opaque_state: "original-encrypted-state".into(), + }, + openbitfun_core_types::ModelResponseReplayItem::AssistantMessage, + openbitfun_core_types::ModelResponseReplayItem::FunctionCall { + call_id: "already-executed".into(), + }, + ], + }); + if format == "fixed-account" { + let source = test_account_source(vec![client.clone()], true); + let mut old_account = source.clients[0].0.clone(); + old_account.account_id = "previous-account".into(); + old_account.upstream_account_id = Some("previous-seat".into()); + previous + .model_response_replay + .as_mut() + .unwrap() + .account_origin = Some(old_account); + context.account_client_source = Some(source); + } + let mut result = + super::AIMessage::user("Saved result.txt; checksum=manual-switch".into()); + result.role = "tool".into(); + result.name = Some("write_file".into()); + result.tool_call_id = Some("already-executed".into()); + let history = vec![ + super::AIMessage::user("Save the file".into()), + previous, + result, + super::AIMessage::user("Continue".into()), + ]; + // Exercise the persisted shape without changing the original record. + let before = serde_json::to_value(&history).unwrap(); + let restored = serde_json::from_value(before.clone()).unwrap(); + let result = tokio::time::timeout( + Duration::from_secs(5), + test_round_executor().execute_round(client, context, restored, None, None), + ) + .await + .unwrap() + .unwrap(); + assert!(result.had_assistant_text, "{format}"); + assert!( + result.tool_result_messages.is_empty(), + "completed tools must not execute again" + ); + assert_eq!(serde_json::to_value(&history).unwrap(), before); + let requests = server.requests.lock().unwrap(); + assert_eq!(requests.len(), 1, "{format}"); + let sent = requests[0].to_string(); + for secret in [ + "original-signature", + "original-encrypted-state", + "Original private reasoning", + ] { + assert!(!sent.contains(secret), "{format}: {secret}"); + } + for retained in [ + "Saved the requested file", + "write_file", + "saved once", + "checksum=manual-switch", + ] { + assert!(sent.contains(retained), "{format}: {retained}"); + } + if format == "gemini" { + assert!(sent.contains("skip_thought_signature_validator")); + } + drop(requests); + if format == "responses" { + // Switching back projects the original durable history again; + // compatible encrypted reasoning must still reach its producer. + let result = tokio::time::timeout( + Duration::from_secs(5), + test_round_executor().execute_round( + Arc::new(original_client), + test_round_context(), + history.clone(), + None, + None, + ), + ) + .await + .unwrap() + .unwrap(); + assert!(result.had_assistant_text); + let requests = server.requests.lock().unwrap(); + assert_eq!(requests.len(), 2); + assert!(requests[1].to_string().contains("original-encrypted-state")); + assert_eq!(serde_json::to_value(&history).unwrap(), before); + } + } + } + #[tokio::test] async fn routed_opaque_history_prefers_its_producer_and_degrades_on_switch() { for has_origin in [false, true] { diff --git a/src/crates/assembly/core/src/infrastructure/ai/channel_client_source.rs b/src/crates/assembly/core/src/infrastructure/ai/channel_client_source.rs index af17c5fc8c..7a9542b377 100644 --- a/src/crates/assembly/core/src/infrastructure/ai/channel_client_source.rs +++ b/src/crates/assembly/core/src/infrastructure/ai/channel_client_source.rs @@ -234,7 +234,43 @@ mod tests { &history, channel, &changed, true ) .is_err()); + let mut projected = history.clone(); + assert_eq!( + ChannelClientSource::project_replay_for_target( + &mut projected, + channel, + &changed, + None + ), + 1 + ); + assert!(projected[0].thinking_signature.is_none()); + assert!(projected[0].model_response_replay.is_none()); + assert_eq!(projected[0].content, history[0].content); + assert!(ChannelClientSource::validate_replay_binding( + &projected, channel, &changed, true + ) + .is_ok()); } + let mut changed_model = client.clone(); + changed_model.config.model = "another-model".into(); + let mut mixed = history.clone(); + let mut other = history[0].clone(); + other + .model_response_replay + .as_mut() + .unwrap() + .channel_fingerprint = Some(changed_model.channel_replay_fingerprint()); + mixed.push(other); + assert_eq!( + ChannelClientSource::project_replay_for_target(&mut mixed, "primary", &client, None), + 1 + ); + assert_eq!( + serde_json::to_value(&mixed[0]).unwrap(), + serde_json::to_value(&history[0]).unwrap() + ); + assert!(mixed[1].thinking_signature.is_none()); let mut legacy = history[0].clone(); legacy.model_response_replay = None; assert!(ChannelClientSource::validate_replay_binding( @@ -249,6 +285,60 @@ mod tests { .is_ok() ); } + + #[test] + fn replay_projection_preserves_legacy_seat_provenance_but_drops_unknown_state() { + let client = AIClient::new( + openbitfun_core_types::AIConfig::try_from(AIModelConfig::default()).unwrap(), + ); + let account = openbitfun_core_types::ModelAccountOrigin { + provider: "codex".into(), + account_id: "account-a".into(), + upstream_account_id: Some("seat-a".into()), + }; + let mut message = openbitfun_core_types::Message::assistant("Saved answer".into()); + message.thinking_signature = Some("original-signature".into()); + message.reasoning_content = Some("Signed reasoning".into()); + let mut unknown = vec![message.clone()]; + assert_eq!( + ChannelClientSource::project_replay_for_target(&mut unknown, "primary", &client, None), + 1 + ); + assert!(unknown[0].reasoning_content.is_none()); + message.model_response_replay = Some(openbitfun_core_types::ModelResponseReplay { + protocol: "openai_responses".into(), + channel_origin: None, + channel_fingerprint: None, + account_origin: Some(account.clone()), + items: vec![], + }); + let before = serde_json::to_value(&message).unwrap(); + let mut projected = vec![message]; + assert_eq!( + ChannelClientSource::project_replay_for_target( + &mut projected, + "primary", + &client, + Some(&account) + ), + 0 + ); + assert_eq!(serde_json::to_value(&projected[0]).unwrap(), before); + let changed = openbitfun_core_types::ModelAccountOrigin { + upstream_account_id: Some("replacement-seat".into()), + ..account + }; + assert_eq!( + ChannelClientSource::project_replay_for_target( + &mut projected, + "primary", + &client, + Some(&changed) + ), + 1 + ); + assert!(projected[0].model_response_replay.is_none()); + } } pub(crate) struct ChannelClientSource { @@ -379,6 +469,51 @@ impl ChannelClientSource { .as_ref() .is_none_or(|route| route.manual_channel.is_none()) } + + /// Project the request copy for the actual destination, including manually + /// selected models and fixed accounts. Keep compatible state per message so + /// switching back can still reuse the originals in durable history. + pub(crate) fn project_replay_for_target( + messages: &mut [openbitfun_core_types::Message], + channel: &str, + client: &AIClient, + account: Option<&openbitfun_core_types::ModelAccountOrigin>, + ) -> usize { + let fingerprint = client.channel_replay_fingerprint(); + let mut stripped = 0; + for message in messages { + if !Self::has_nonportable_history(std::slice::from_ref(message)) { + continue; + } + let compatible = message + .model_response_replay + .as_ref() + .is_some_and(|replay| { + let same_account = replay.account_origin.as_ref() == account + && account.is_none_or(|origin| { + origin + .upstream_account_id + .as_deref() + .is_some_and(|id| !id.trim().is_empty()) + }); + let same_channel = match (&replay.channel_origin, &replay.channel_fingerprint) { + (Some(origin), Some(original)) => { + origin == channel && original == &fingerprint + } + // Older subscription history can prove its producing seat + // without channel fields. Preserve that existing replay path. + (None, None) => account.is_some(), + _ => false, + }; + same_channel && same_account + }); + if !compatible { + Self::make_portable(message); + stripped += 1; + } + } + stripped + } /// Removes provider-signed reasoning from one message. Text, tool calls /// and tool results stay; any account or channel can then replay it. fn make_portable(message: &mut openbitfun_core_types::Message) {