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) {