diff --git a/src/core/event_bus/events.rs b/src/core/event_bus/events.rs index cb688de63..13f8db44a 100644 --- a/src/core/event_bus/events.rs +++ b/src/core/event_bus/events.rs @@ -354,6 +354,10 @@ pub enum DomainEvent { content: String, thread_ts: Option, response: String, + /// Provider route selected for the LLM turn. + provider: String, + /// Model route selected for the LLM turn. + model: String, elapsed_ms: u64, success: bool, /// Workspace directory active when this event was published. diff --git a/src/core/event_bus/events_tests.rs b/src/core/event_bus/events_tests.rs index bb581039b..c1305e302 100644 --- a/src/core/event_bus/events_tests.rs +++ b/src/core/event_bus/events_tests.rs @@ -138,6 +138,8 @@ fn all_variants_have_correct_domain() { content: "hi".into(), thread_ts: None, response: "hello".into(), + provider: "test-provider".into(), + model: "test-model".into(), elapsed_ms: 0, success: true, workspace_dir: std::path::PathBuf::from("/test"), diff --git a/src/core/event_bus/native_request.rs b/src/core/event_bus/native_request.rs index e542582cd..12835ddbd 100644 --- a/src/core/event_bus/native_request.rs +++ b/src/core/event_bus/native_request.rs @@ -33,7 +33,7 @@ //! |req| async move { //! let text = run_tool_call_loop(/* ... */).await //! .map_err(|e| e.to_string())?; -//! Ok(AgentTurnResponse { text }) +//! Ok(AgentTurnResponse::new(text)) //! }, //! ); //! diff --git a/src/openhuman/agent/bus.rs b/src/openhuman/agent/bus.rs index ab64aea00..8bca5f5db 100644 --- a/src/openhuman/agent/bus.rs +++ b/src/openhuman/agent/bus.rs @@ -22,7 +22,9 @@ use crate::core::event_bus::register_native_global; use crate::openhuman::agent::progress::AgentProgress; use crate::openhuman::agent::turn_origin::{self, AgentTurnOrigin}; use crate::openhuman::config::MultimodalConfig; -use crate::openhuman::inference::provider::{ChatMessage, Provider}; +use crate::openhuman::inference::provider::{ + current_resolved_provider_route, with_resolved_provider_route_scope, ChatMessage, Provider, +}; use crate::openhuman::prompt_injection::{ enforce_prompt_input, PromptEnforcementAction, PromptEnforcementContext, }; @@ -144,6 +146,37 @@ pub struct AgentTurnRequest { pub struct AgentTurnResponse { /// Final assistant text after all tool calls resolved and the loop terminated. pub text: String, + /// Provider that actually produced the final response, after any routing, + /// retry, or fallback layer. `None` means the provider stack did not expose + /// resolved-route metadata and callers should fall back to the requested + /// provider. + pub resolved_provider: Option, + /// Model that actually produced the final response, after any routing, + /// retry, or fallback layer. `None` means callers should fall back to the + /// requested model. + pub resolved_model: Option, +} + +impl AgentTurnResponse { + pub fn new(text: impl Into) -> Self { + Self { + text: text.into(), + resolved_provider: None, + resolved_model: None, + } + } + + pub fn with_resolved_route( + text: impl Into, + provider: impl Into, + model: impl Into, + ) -> Self { + Self { + text: text.into(), + resolved_provider: Some(provider.into()), + resolved_model: Some(model.into()), + } + } } /// Register the agent domain's native request handlers on the global @@ -262,43 +295,48 @@ pub fn register_agent_handlers() { channel_name, target_agent_id.as_deref().unwrap_or("root") ); - let text = turn_origin::with_origin( - origin, - with_file_state_agent_id( - file_state_id, - with_current_sandbox_mode(sandbox_mode, async { - run_tool_call_loop( - provider.as_ref(), - &mut history, - tools_registry.as_ref(), - &provider_name, - &model, - temperature, - silent, - &channel_name, - &multimodal, - &multimodal_files, - max_tool_iterations, - on_delta, - visible_tool_names.as_ref(), - &extra_tools, - on_progress, - // Bus path runs ad-hoc agent turns without an Agent - // handle, so we pass None — payload summarization is - // wired into the orchestrator session via Agent::turn, - // not the bus dispatcher. - None, - // Use the default (allow-all) tool policy. Custom - // policies can be wired in via AgentTurnRequest when - // per-channel policy configuration is added (#2134). - &crate::openhuman::tools::policy::DefaultToolPolicy, - ) - .await - }), - ), - ) - .await - .map_err(|e| e.to_string())?; + let (text, resolved_route) = with_resolved_provider_route_scope(async { + let text = turn_origin::with_origin( + origin, + with_file_state_agent_id( + file_state_id, + with_current_sandbox_mode(sandbox_mode, async { + run_tool_call_loop( + provider.as_ref(), + &mut history, + tools_registry.as_ref(), + &provider_name, + &model, + temperature, + silent, + &channel_name, + &multimodal, + &multimodal_files, + max_tool_iterations, + on_delta, + visible_tool_names.as_ref(), + &extra_tools, + on_progress, + // Bus path runs ad-hoc agent turns without an Agent + // handle, so we pass None — payload summarization is + // wired into the orchestrator session via Agent::turn, + // not the bus dispatcher. + None, + // Use the default (allow-all) tool policy. Custom + // policies can be wired in via AgentTurnRequest when + // per-channel policy configuration is added (#2134). + &crate::openhuman::tools::policy::DefaultToolPolicy, + ) + .await + }), + ), + ) + .await + .map_err(|e| e.to_string())?; + let resolved_route = current_resolved_provider_route(); + Ok::<_, String>((text, resolved_route)) + }) + .await?; tracing::debug!( channel = %channel_name, @@ -306,7 +344,12 @@ pub fn register_agent_handlers() { "[agent::bus] {AGENT_RUN_TURN_METHOD} completed" ); - Ok(AgentTurnResponse { text }) + Ok(match resolved_route { + Some(route) => { + AgentTurnResponse::with_resolved_route(text, route.provider, route.model) + } + None => AgentTurnResponse::new(text), + }) }, ); tracing::debug!("[agent::bus] registered native handler `{AGENT_RUN_TURN_METHOD}`"); @@ -346,7 +389,7 @@ pub fn register_agent_handlers() { /// async move { /// calls.fetch_add(1, Ordering::SeqCst); /// assert_eq!(req.channel_name, "discord"); -/// Ok(AgentTurnResponse { text: "CANNED".into() }) +/// Ok(AgentTurnResponse::new("CANNED")) /// } /// }) /// .await; @@ -459,9 +502,10 @@ mod tests { assert_eq!(req.provider_name, "fake-provider"); assert_eq!(req.channel_name, "test-channel"); assert_eq!(req.history.len(), 2); - Ok(AgentTurnResponse { - text: format!("handled({})", req.history.len()), - }) + Ok(AgentTurnResponse::new(format!( + "handled({})", + req.history.len() + ))) }, ); @@ -487,9 +531,7 @@ mod tests { .expect("streaming test must supply an on_delta sender"); tx.send("chunk1".into()).await.map_err(|e| e.to_string())?; tx.send("chunk2".into()).await.map_err(|e| e.to_string())?; - Ok(AgentTurnResponse { - text: "streamed".into(), - }) + Ok(AgentTurnResponse::new("streamed")) }, ); diff --git a/src/openhuman/agent/triage/evaluator_tests.rs b/src/openhuman/agent/triage/evaluator_tests.rs index 8d46718a7..a42289424 100644 --- a/src/openhuman/agent/triage/evaluator_tests.rs +++ b/src/openhuman/agent/triage/evaluator_tests.rs @@ -284,12 +284,11 @@ const VALID_JSON_REPLY: &str = "{\"action\":\"acknowledge\",\"reason\":\"all goo async fn happy_path_returns_cloud_resolution() { AgentDefinitionRegistry::init_global_builtins().expect("init_global_builtins"); - let _guard = mock_agent_run_turn(move |_req| async move { - Ok(AgentTurnResponse { - text: VALID_JSON_REPLY.to_string(), - }) - }) - .await; + let _guard = + mock_agent_run_turn( + move |_req| async move { Ok(AgentTurnResponse::new(VALID_JSON_REPLY)) }, + ) + .await; let outcome = run_triage_with_arms_for_test(cloud_arm(), Some(local_arm()), &envelope()) .await @@ -313,9 +312,7 @@ async fn rate_limited_then_ok_marks_cloud_after_retry() { if n == 0 { Err("HTTP 429 Too Many Requests; Retry-After: 0".to_string()) } else { - Ok(AgentTurnResponse { - text: VALID_JSON_REPLY.to_string(), - }) + Ok(AgentTurnResponse::new(VALID_JSON_REPLY)) } } }) @@ -348,9 +345,7 @@ async fn double_429_falls_through_to_local_fallback() { } else { // Third call should be the local arm. assert_eq!(req.provider_name, "stub-local", "fall-through hits local"); - Ok(AgentTurnResponse { - text: VALID_JSON_REPLY.to_string(), - }) + Ok(AgentTurnResponse::new(VALID_JSON_REPLY)) } } }) @@ -381,9 +376,7 @@ async fn cloud_5xx_falls_through_to_local_fallback() { Err("upstream returned 502 Bad Gateway".to_string()) } else { assert_eq!(req.provider_name, "stub-local"); - Ok(AgentTurnResponse { - text: VALID_JSON_REPLY.to_string(), - }) + Ok(AgentTurnResponse::new(VALID_JSON_REPLY)) } } }) @@ -492,9 +485,7 @@ async fn cloud_budget_exhausted_skips_retry_and_falls_to_local() { req.provider_name, "stub-local", "budget-exhausted must skip cloud retry and dispatch to local" ); - Ok(AgentTurnResponse { - text: VALID_JSON_REPLY.to_string(), - }) + Ok(AgentTurnResponse::new(VALID_JSON_REPLY)) } } }) @@ -551,9 +542,7 @@ async fn cloud_budget_exhausted_on_retry_falls_through_to_local() { req.provider_name, "stub-local", "post-budget dispatch must land on local" ); - Ok(AgentTurnResponse { - text: VALID_JSON_REPLY.to_string(), - }) + Ok(AgentTurnResponse::new(VALID_JSON_REPLY)) } } } @@ -677,9 +666,7 @@ async fn cloud_safety_flagged_then_local_recovers_decides_on_local() { req.provider_name, "stub-local", "safety-flagged on cloud must skip cloud retry and dispatch to local" ); - Ok(AgentTurnResponse { - text: VALID_JSON_REPLY.to_string(), - }) + Ok(AgentTurnResponse::new(VALID_JSON_REPLY)) } } }) @@ -788,17 +775,13 @@ async fn double_cloud_parse_failure_falls_through_to_local_fallback() { req.provider_name, "stub-cloud", "first two attempts should stay on the cloud arm" ); - Ok(AgentTurnResponse { - text: "not json".to_string(), - }) + Ok(AgentTurnResponse::new("not json")) } else { assert_eq!( req.provider_name, "stub-local", "malformed cloud retry should fall through to local" ); - Ok(AgentTurnResponse { - text: VALID_JSON_REPLY.to_string(), - }) + Ok(AgentTurnResponse::new(VALID_JSON_REPLY)) } } }) @@ -828,9 +811,7 @@ async fn double_cloud_parse_failure_without_local_returns_deferred_not_err() { let counter = StdArc::clone(&counter_for_stub); async move { counter.fetch_add(1, Ordering::SeqCst); - Ok(AgentTurnResponse { - text: "still not json".to_string(), - }) + Ok(AgentTurnResponse::new("still not json")) } }) .await; diff --git a/src/openhuman/channels/providers/telegram/bus_tests.rs b/src/openhuman/channels/providers/telegram/bus_tests.rs index 3dfd72935..c171ec8f8 100644 --- a/src/openhuman/channels/providers/telegram/bus_tests.rs +++ b/src/openhuman/channels/providers/telegram/bus_tests.rs @@ -34,6 +34,8 @@ async fn subscriber_marks_busy_on_received_and_clears_on_processed() { content: "hi".into(), thread_ts: Some("1".into()), response: "ok".into(), + provider: "test-provider".into(), + model: "test-model".into(), elapsed_ms: 10, success: true, workspace_dir: dir.path().to_path_buf(), @@ -149,6 +151,8 @@ async fn telegram_processed_matching_workspace_clears_busy() { content: "hi".into(), thread_ts: None, response: "done".into(), + provider: "test-provider".into(), + model: "test-model".into(), elapsed_ms: 50, success: true, workspace_dir: dir.path().to_path_buf(), @@ -194,6 +198,8 @@ async fn telegram_processed_stale_workspace_does_not_clear_busy() { content: "hi".into(), thread_ts: None, response: "done".into(), + provider: "test-provider".into(), + model: "test-model".into(), elapsed_ms: 50, success: true, workspace_dir: stale.path().to_path_buf(), diff --git a/src/openhuman/channels/runtime/dispatch/processor.rs b/src/openhuman/channels/runtime/dispatch/processor.rs index a54302cff..9648cd826 100644 --- a/src/openhuman/channels/runtime/dispatch/processor.rs +++ b/src/openhuman/channels/runtime/dispatch/processor.rs @@ -462,7 +462,6 @@ pub(crate) async fn process_channel_message( turn_request, ) .await - .map(|resp| resp.text) .map_err(|err| match err { // Unwrap handler-returned errors so the underlying // message (e.g. "Agent exceeded maximum tool iterations") @@ -514,8 +513,17 @@ pub(crate) async fn process_channel_message( log_worker_join_result(handle.await); } - let (success, response_text) = match llm_result { + let (success, response_text, response_provider, response_model) = match llm_result { Ok(Ok(response)) => { + let resolved_provider = response + .resolved_provider + .clone() + .unwrap_or_else(|| route.provider.clone()); + let resolved_model = response + .resolved_model + .clone() + .unwrap_or_else(|| route.model.clone()); + let response_text = response.text; // Save user + assistant turn to per-sender history { let mut histories = ctx @@ -524,7 +532,7 @@ pub(crate) async fn process_channel_message( .unwrap_or_else(|e| e.into_inner()); let turns = histories.entry(history_key).or_default(); turns.push(ChatMessage::user(&enriched_message)); - turns.push(ChatMessage::assistant(&response)); + turns.push(ChatMessage::assistant(&response_text)); // Trim to MAX_CHANNEL_HISTORY (keep recent turns) while turns.len() > crate::openhuman::channels::context::MAX_CHANNEL_HISTORY { turns.remove(0); @@ -533,7 +541,7 @@ pub(crate) async fn process_channel_message( println!( " 🤖 Reply ({}ms): {}", started_at.elapsed().as_millis(), - truncate_with_ellipsis(&response, REPLY_LOG_TRUNCATE_CHARS) + truncate_with_ellipsis(&response_text, REPLY_LOG_TRUNCATE_CHARS) ); if let Some(channel) = target_channel.as_ref() { if let Some(ref draft_id) = draft_message_id { @@ -541,7 +549,7 @@ pub(crate) async fn process_channel_message( .finalize_draft( &msg.reply_target, draft_id, - &response, + &response_text, msg.thread_ts.as_deref(), ) .await @@ -549,14 +557,14 @@ pub(crate) async fn process_channel_message( tracing::warn!("Failed to finalize draft: {e}; sending as new message"); let _ = channel .send( - &SendMessage::new(&response, &msg.reply_target) + &SendMessage::new(&response_text, &msg.reply_target) .in_thread(msg.thread_ts.clone()), ) .await; } } else if let Err(e) = channel .send( - &SendMessage::new(&response, &msg.reply_target) + &SendMessage::new(&response_text, &msg.reply_target) .in_thread(msg.thread_ts.clone()), ) .await @@ -564,7 +572,7 @@ pub(crate) async fn process_channel_message( eprintln!(" ❌ Failed to reply on {}: {e}", channel.name()); } } - (true, response) + (true, response_text, resolved_provider, resolved_model) } Ok(Err(e)) => { if is_context_window_overflow_error(&e) { @@ -607,6 +615,8 @@ pub(crate) async fn process_channel_message( content: msg.content.clone(), thread_ts: msg.thread_ts.clone(), response: error_text.to_string(), + provider: route.provider.clone(), + model: route.model.clone(), elapsed_ms: started_at.elapsed().as_millis() as u64, success: false, workspace_dir: ctx.workspace_dir.as_ref().clone(), @@ -685,7 +695,12 @@ pub(crate) async fn process_channel_message( .await; } } - (false, error_response) + ( + false, + error_response, + route.provider.clone(), + route.model.clone(), + ) } Err(_) => { let timeout_msg = format!("LLM response timed out after {}s", ctx.message_timeout_secs); @@ -724,7 +739,12 @@ pub(crate) async fn process_channel_message( .await; } } - (false, error_text) + ( + false, + error_text, + route.provider.clone(), + route.model.clone(), + ) } }; @@ -736,6 +756,8 @@ pub(crate) async fn process_channel_message( content: msg.content.clone(), thread_ts: msg.thread_ts.clone(), response: response_text, + provider: response_provider, + model: response_model, elapsed_ms: started_at.elapsed().as_millis() as u64, success, workspace_dir: ctx.workspace_dir.as_ref().clone(), diff --git a/src/openhuman/channels/runtime/test_support.rs b/src/openhuman/channels/runtime/test_support.rs index bd1113e86..a7b32ea3d 100644 --- a/src/openhuman/channels/runtime/test_support.rs +++ b/src/openhuman/channels/runtime/test_support.rs @@ -390,9 +390,7 @@ pub async fn run_dispatch_harness(options: DispatchHarnessOptions) -> DispatchHa match handler_error { Some(message) => Err(message), - None => Ok(AgentTurnResponse { - text: response_text, - }), + None => Ok(AgentTurnResponse::new(response_text)), } } } diff --git a/src/openhuman/channels/tests/discord_integration.rs b/src/openhuman/channels/tests/discord_integration.rs index 7cb518b6e..f8ce7050d 100644 --- a/src/openhuman/channels/tests/discord_integration.rs +++ b/src/openhuman/channels/tests/discord_integration.rs @@ -351,9 +351,7 @@ async fn discord_dispatch_routes_through_agent_run_turn_bus_handler() { req.history.len() >= 2, "history should include at least the system prompt and user message" ); - Ok(AgentTurnResponse { - text: "CANNED_DISCORD_RESPONSE".to_string(), - }) + Ok(AgentTurnResponse::new("CANNED_DISCORD_RESPONSE")) } }) .await; diff --git a/src/openhuman/channels/tests/runtime_dispatch.rs b/src/openhuman/channels/tests/runtime_dispatch.rs index 8caff7170..c02f77616 100644 --- a/src/openhuman/channels/tests/runtime_dispatch.rs +++ b/src/openhuman/channels/tests/runtime_dispatch.rs @@ -2,6 +2,7 @@ use super::super::context::{ChannelRuntimeContext, CHANNEL_MESSAGE_TIMEOUT_SECS} use super::super::runtime::{process_channel_message, run_message_dispatch_loop}; use super::super::{traits, Channel}; use super::common::{use_real_agent_handler, NoopMemory, RecordingChannel, SlowProvider}; +use crate::core::event_bus::{init_global, DomainEvent, DEFAULT_CAPACITY}; use crate::openhuman::agent::bus::{mock_agent_run_turn, AgentTurnRequest, AgentTurnResponse}; use crate::openhuman::inference::provider; use std::collections::HashMap; @@ -27,9 +28,7 @@ async fn message_dispatch_processes_messages_in_parallel() { peak_in_flight.fetch_max(current, Ordering::SeqCst); tokio::time::sleep(Duration::from_millis(250)).await; in_flight.fetch_sub(1, Ordering::SeqCst); - Ok(AgentTurnResponse { - text: "echo: stub".to_string(), - }) + Ok(AgentTurnResponse::new("echo: stub")) } } }) @@ -190,9 +189,7 @@ async fn dispatch_routes_through_agent_run_turn_bus_handler() { req.history.len() >= 2, "history should include at least the system prompt and user message" ); - Ok(AgentTurnResponse { - text: "CANNED_RESPONSE_FROM_BUS_STUB".to_string(), - }) + Ok(AgentTurnResponse::new("CANNED_RESPONSE_FROM_BUS_STUB")) } }) .await; @@ -265,6 +262,103 @@ async fn dispatch_routes_through_agent_run_turn_bus_handler() { // that expects the real path sees a consistent registry. } +#[tokio::test] +async fn channel_processed_event_records_resolved_agent_route() { + init_global(DEFAULT_CAPACITY); + let mut events = crate::core::event_bus::global() + .expect("event bus should be initialized") + .raw_receiver(); + + let _bus_guard = mock_agent_run_turn(move |_req| async move { + Ok(AgentTurnResponse::with_resolved_route( + "CANNED_RESPONSE_FROM_RESOLVED_ROUTE", + "actual-provider", + "actual-model", + )) + }) + .await; + + let channel_impl = Arc::new(RecordingChannel::default()); + let channel: Arc = channel_impl.clone(); + + let mut channels_by_name = HashMap::new(); + channels_by_name.insert(channel.name().to_string(), channel); + + let runtime_ctx = Arc::new(ChannelRuntimeContext { + channels_by_name: Arc::new(channels_by_name), + provider: Arc::new(super::common::DummyProvider), + default_provider: Arc::new("requested-provider".to_string()), + memory: Arc::new(NoopMemory), + tools_registry: Arc::new(vec![]), + system_prompt: Arc::new("test-system-prompt".to_string()), + model: Arc::new("requested-model".to_string()), + temperature: 0.0, + auto_save_memory: false, + max_tool_iterations: 10, + min_relevance_score: 0.0, + conversation_histories: Arc::new(Mutex::new(HashMap::new())), + provider_cache: Arc::new(Mutex::new(HashMap::new())), + route_overrides: Arc::new(Mutex::new(HashMap::new())), + api_url: None, + inference_url: None, + reliability: Arc::new(crate::openhuman::config::ReliabilityConfig::default()), + provider_runtime_options: provider::ProviderRuntimeOptions::default(), + workspace_dir: Arc::new(std::env::temp_dir()), + message_timeout_secs: CHANNEL_MESSAGE_TIMEOUT_SECS, + multimodal: crate::openhuman::config::MultimodalConfig::default(), + multimodal_files: crate::openhuman::config::MultimodalFileConfig::default(), + }); + + process_channel_message( + runtime_ctx, + traits::ChannelMessage { + id: "resolved-route-msg".to_string(), + sender: "alice".to_string(), + reply_target: "alice".to_string(), + content: "hello from resolved route test".to_string(), + channel: "test-channel".to_string(), + timestamp: 1, + thread_ts: None, + }, + ) + .await; + + // Bound the scan so unrelated global event traffic can't hang the test; + // the target event is published by process_channel_message above. + let mut matched = false; + for _ in 0..50 { + let event = tokio::time::timeout(Duration::from_millis(200), events.recv()) + .await + .expect("ChannelMessageProcessed event should be published") + .expect("event receiver should stay open"); + + if let DomainEvent::ChannelMessageProcessed { + message_id, + provider, + model, + response, + success, + .. + } = event + { + if message_id != "resolved-route-msg" { + continue; + } + + assert!(success); + assert_eq!(response, "CANNED_RESPONSE_FROM_RESOLVED_ROUTE"); + assert_eq!(provider, "actual-provider"); + assert_eq!(model, "actual-model"); + matched = true; + break; + } + } + assert!( + matched, + "did not observe ChannelMessageProcessed for resolved-route-msg" + ); +} + /// Security regression for the `[FILE:…]` smuggling vector: a remote /// channel user (Slack/Discord/Telegram/WhatsApp/etc) putting /// `[FILE:/etc/passwd]` (or any other local-path marker) into a normal @@ -282,9 +376,7 @@ async fn process_channel_message_hardens_multimodal_files_against_smuggled_marke let captured = Arc::clone(&captured_for_handler); async move { *captured.lock().unwrap() = Some(req.multimodal_files.clone()); - Ok(AgentTurnResponse { - text: "ok".to_string(), - }) + Ok(AgentTurnResponse::new("ok")) } }) .await; @@ -375,9 +467,7 @@ async fn process_channel_message_hardens_against_relative_path_markers() { let captured = Arc::clone(&captured_for_handler); async move { *captured.lock().unwrap() = Some(req.multimodal_files.clone()); - Ok(AgentTurnResponse { - text: "ok".to_string(), - }) + Ok(AgentTurnResponse::new("ok")) } }) .await; diff --git a/src/openhuman/channels/tests/telegram_integration.rs b/src/openhuman/channels/tests/telegram_integration.rs index 039853d2d..11811cffb 100644 --- a/src/openhuman/channels/tests/telegram_integration.rs +++ b/src/openhuman/channels/tests/telegram_integration.rs @@ -468,9 +468,7 @@ async fn telegram_dispatch_routes_through_agent_run_turn_bus_handler() { req.history.len() >= 2, "history should include at least the system prompt and user message" ); - Ok(AgentTurnResponse { - text: "CANNED_TELEGRAM_RESPONSE".to_string(), - }) + Ok(AgentTurnResponse::new("CANNED_TELEGRAM_RESPONSE")) } }) .await; diff --git a/src/openhuman/inference/provider/mod.rs b/src/openhuman/inference/provider/mod.rs index e532a99fc..3288d3481 100644 --- a/src/openhuman/inference/provider/mod.rs +++ b/src/openhuman/inference/provider/mod.rs @@ -20,6 +20,7 @@ mod openai_codex; pub mod openhuman_backend; pub mod ops; pub mod reliable; +pub mod resolved_route; pub mod router; pub mod schemas; pub mod temperature; @@ -45,3 +46,7 @@ pub use error_code::{ }; pub use factory::{create_chat_provider, provider_for_role, BYOK_INCOMPLETE_SENTINEL}; pub use ops::*; +pub use resolved_route::{ + current_resolved_provider_route, record_resolved_provider_route, + with_resolved_provider_route_scope, ResolvedProviderRoute, +}; diff --git a/src/openhuman/inference/provider/reliable.rs b/src/openhuman/inference/provider/reliable.rs index 1d4953d75..58c6a2461 100644 --- a/src/openhuman/inference/provider/reliable.rs +++ b/src/openhuman/inference/provider/reliable.rs @@ -2,6 +2,7 @@ use super::traits::{ ChatMessage, ChatRequest, ChatResponse, StreamChunk, StreamError, StreamOptions, StreamResult, }; use super::Provider; +use crate::openhuman::inference::provider::record_resolved_provider_route; use async_trait::async_trait; use futures_util::{stream, StreamExt}; use std::collections::HashMap; @@ -511,6 +512,7 @@ impl Provider for ReliableProvider { let mut backoff_ms = self.base_backoff_ms; for attempt in 0..=self.max_retries { + record_resolved_provider_route(provider_name, *current_model); match provider .chat_with_system(system_prompt, message, current_model, temperature) .await @@ -647,6 +649,7 @@ impl Provider for ReliableProvider { let mut backoff_ms = self.base_backoff_ms; for attempt in 0..=self.max_retries { + record_resolved_provider_route(provider_name, *current_model); match provider .chat_with_history(messages, current_model, temperature) .await @@ -814,6 +817,7 @@ impl Provider for ReliableProvider { stream: stream_this_attempt, max_tokens: request.max_tokens, }; + record_resolved_provider_route(provider_name, *current_model); match provider.chat(req, current_model, temperature).await { Ok(resp) => { if attempt > 0 || *current_model != model { @@ -939,6 +943,7 @@ impl Provider for ReliableProvider { let mut backoff_ms = self.base_backoff_ms; for attempt in 0..=self.max_retries { + record_resolved_provider_route(provider_name, *current_model); match provider .chat_with_tools(messages, tools, current_model, temperature) .await diff --git a/src/openhuman/inference/provider/reliable_tests.rs b/src/openhuman/inference/provider/reliable_tests.rs index 51341c98e..4e6b60b4c 100644 --- a/src/openhuman/inference/provider/reliable_tests.rs +++ b/src/openhuman/inference/provider/reliable_tests.rs @@ -146,6 +146,54 @@ async fn falls_back_after_retries_exhausted() { assert_eq!(fallback_calls.load(Ordering::SeqCst), 1); } +#[tokio::test] +async fn records_successful_fallback_provider_route() { + let primary_calls = Arc::new(AtomicUsize::new(0)); + let fallback_calls = Arc::new(AtomicUsize::new(0)); + + let provider = ReliableProvider::new( + vec![ + ( + "primary".into(), + Box::new(MockProvider { + calls: Arc::clone(&primary_calls), + fail_until_attempt: usize::MAX, + response: "never", + error: "primary down", + }), + ), + ( + "fallback".into(), + Box::new(MockProvider { + calls: Arc::clone(&fallback_calls), + fail_until_attempt: 0, + response: "ok", + error: "boom", + }), + ), + ], + 0, + 1, + ); + + let recorded = + crate::openhuman::inference::provider::with_resolved_provider_route_scope(async { + let result = provider + .chat_with_system(Some("system"), "hello", "requested-model", 0.0) + .await + .unwrap(); + assert_eq!(result, "ok"); + crate::openhuman::inference::provider::current_resolved_provider_route() + }) + .await + .expect("reliable provider should record the successful route"); + + assert_eq!(recorded.provider, "fallback"); + assert_eq!(recorded.model, "requested-model"); + assert_eq!(primary_calls.load(Ordering::SeqCst), 1); + assert_eq!(fallback_calls.load(Ordering::SeqCst), 1); +} + #[tokio::test] async fn returns_aggregated_error_when_all_providers_fail() { let provider = ReliableProvider::new( diff --git a/src/openhuman/inference/provider/resolved_route.rs b/src/openhuman/inference/provider/resolved_route.rs new file mode 100644 index 000000000..93cbb527f --- /dev/null +++ b/src/openhuman/inference/provider/resolved_route.rs @@ -0,0 +1,86 @@ +//! Per-turn resolved provider/model metadata. +//! +//! Provider wrappers that translate model aliases or perform fallbacks record +//! the concrete route that actually handled the latest successful provider +//! call. The agent bus reads this after the turn so channel audit events can +//! persist the resolved provider/model instead of the caller's requested route. + +use std::sync::{Arc, Mutex}; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ResolvedProviderRoute { + pub provider: String, + pub model: String, +} + +type RouteSlot = Arc>>; + +tokio::task_local! { + static RESOLVED_PROVIDER_ROUTE: RouteSlot; +} + +pub async fn with_resolved_provider_route_scope(future: F) -> F::Output +where + F: std::future::Future, +{ + tracing::trace!("[provider] resolved-route scope enter"); + let out = RESOLVED_PROVIDER_ROUTE + .scope(Arc::new(Mutex::new(None)), Box::pin(future)) + .await; + tracing::trace!("[provider] resolved-route scope exit"); + out +} + +pub fn record_resolved_provider_route(provider: impl Into, model: impl Into) { + let provider = provider.into(); + let model = model.into(); + let route = ResolvedProviderRoute { + provider: provider.clone(), + model: model.clone(), + }; + let wrote = RESOLVED_PROVIDER_ROUTE + .try_with(|slot| { + *slot.lock().unwrap_or_else(|e| e.into_inner()) = Some(route); + }) + .is_ok(); + tracing::debug!( + provider = %provider, + model = %model, + wrote, + "[provider] resolved-route recorded" + ); +} + +pub fn current_resolved_provider_route() -> Option { + let route = RESOLVED_PROVIDER_ROUTE + .try_with(|slot| slot.lock().unwrap_or_else(|e| e.into_inner()).clone()) + .ok() + .flatten(); + tracing::trace!(present = route.is_some(), "[provider] resolved-route read"); + route +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn resolved_provider_route_scopes_and_clears() { + assert!(current_resolved_provider_route().is_none()); + + let observed = with_resolved_provider_route_scope(async { + record_resolved_provider_route("provider-a", "model-a"); + current_resolved_provider_route() + }) + .await; + + assert_eq!( + observed, + Some(ResolvedProviderRoute { + provider: "provider-a".into(), + model: "model-a".into(), + }) + ); + assert!(current_resolved_provider_route().is_none()); + } +} diff --git a/src/openhuman/inference/provider/router.rs b/src/openhuman/inference/provider/router.rs index 8fb9d1221..38fb7ee41 100644 --- a/src/openhuman/inference/provider/router.rs +++ b/src/openhuman/inference/provider/router.rs @@ -1,5 +1,6 @@ use super::traits::{ChatMessage, ChatRequest, ChatResponse}; use super::Provider; +use crate::openhuman::inference::provider::record_resolved_provider_route; use async_trait::async_trait; use std::collections::HashMap; @@ -179,6 +180,7 @@ impl Provider for RouterProvider { "Router dispatching request" ); + record_resolved_provider_route(provider_name, &resolved_model); provider .chat_with_system(system_prompt, message, &resolved_model, temperature) .await @@ -191,7 +193,8 @@ impl Provider for RouterProvider { temperature: f64, ) -> anyhow::Result { let (provider_idx, resolved_model) = self.resolve(model); - let (_, provider) = &self.providers[provider_idx]; + let (provider_name, provider) = &self.providers[provider_idx]; + record_resolved_provider_route(provider_name, &resolved_model); provider .chat_with_history(messages, &resolved_model, temperature) .await @@ -204,7 +207,8 @@ impl Provider for RouterProvider { temperature: f64, ) -> anyhow::Result { let (provider_idx, resolved_model) = self.resolve(model); - let (_, provider) = &self.providers[provider_idx]; + let (provider_name, provider) = &self.providers[provider_idx]; + record_resolved_provider_route(provider_name, &resolved_model); provider.chat(request, &resolved_model, temperature).await } @@ -216,7 +220,8 @@ impl Provider for RouterProvider { temperature: f64, ) -> anyhow::Result { let (provider_idx, resolved_model) = self.resolve(model); - let (_, provider) = &self.providers[provider_idx]; + let (provider_name, provider) = &self.providers[provider_idx]; + record_resolved_provider_route(provider_name, &resolved_model); provider .chat_with_tools(messages, tools, &resolved_model, temperature) .await diff --git a/src/openhuman/inference/provider/router_tests.rs b/src/openhuman/inference/provider/router_tests.rs index 532654f5c..6b575ddda 100644 --- a/src/openhuman/inference/provider/router_tests.rs +++ b/src/openhuman/inference/provider/router_tests.rs @@ -438,6 +438,30 @@ async fn chat_with_system_passes_system_prompt() { assert_eq!(mock.call_count(), 1); } +#[tokio::test] +async fn records_resolved_route_after_hint_resolution() { + let (router, mocks) = make_router( + vec![("fast", "fast-response"), ("smart", "smart-response")], + vec![("reasoning", "smart", "claude-opus")], + ); + + let recorded = + crate::openhuman::inference::provider::with_resolved_provider_route_scope(async { + let result = router + .chat_with_system(Some("system"), "think", "hint:reasoning", 0.5) + .await + .unwrap(); + assert_eq!(result, "smart-response"); + crate::openhuman::inference::provider::current_resolved_provider_route() + }) + .await + .expect("router should record the concrete provider route"); + + assert_eq!(recorded.provider, "smart"); + assert_eq!(recorded.model, "claude-opus"); + assert_eq!(mocks[1].last_model(), "claude-opus"); +} + #[tokio::test] async fn chat_with_tools_delegates_to_resolved_provider() { let mock = Arc::new(MockProvider::new("tool-response")); diff --git a/src/openhuman/memory_conversations/bus.rs b/src/openhuman/memory_conversations/bus.rs index 9ad114714..44666f829 100644 --- a/src/openhuman/memory_conversations/bus.rs +++ b/src/openhuman/memory_conversations/bus.rs @@ -128,6 +128,8 @@ impl EventHandler for ConversationPersistenceSubscriber { role: "user", success: None, elapsed_ms: None, + model_provider: None, + model: None, source: "channel_received", }, ) { @@ -146,6 +148,8 @@ impl EventHandler for ConversationPersistenceSubscriber { reply_target, thread_ts, response, + provider, + model, elapsed_ms, success, workspace_dir, @@ -179,6 +183,8 @@ impl EventHandler for ConversationPersistenceSubscriber { role: "assistant", success: Some(*success), elapsed_ms: Some(*elapsed_ms), + model_provider: Some(provider), + model: Some(model), source: "channel_processed", }, ) { @@ -205,6 +211,8 @@ struct ChannelTurnDescriptor<'a> { role: &'a str, success: Option, elapsed_ms: Option, + model_provider: Option<&'a str>, + model: Option<&'a str>, source: &'a str, } @@ -267,6 +275,8 @@ fn persist_channel_turn( "sourceEvent": descriptor.source, "success": descriptor.success, "elapsedMs": descriptor.elapsed_ms, + "modelProvider": descriptor.model_provider, + "model": descriptor.model, "sourceMessageId": descriptor.message_id, }), sender: descriptor.role.to_string(), @@ -368,6 +378,8 @@ mod tests { content: "hello".into(), thread_ts: Some("thread-1".into()), response: "hi there".into(), + provider: "test-provider".into(), + model: "test-model".into(), elapsed_ms: 42, success: true, workspace_dir: temp.path().to_path_buf(), @@ -387,6 +399,8 @@ mod tests { assert_eq!(messages[1].sender, "assistant"); assert_eq!(messages[1].extra_metadata["elapsedMs"], 42); assert_eq!(messages[1].extra_metadata["success"], true); + assert_eq!(messages[1].extra_metadata["modelProvider"], "test-provider"); + assert_eq!(messages[1].extra_metadata["model"], "test-model"); } #[tokio::test] @@ -558,6 +572,8 @@ mod tests { content: "hello".into(), thread_ts: None, response: "hi there".into(), + provider: "test-provider".into(), + model: "test-model".into(), elapsed_ms: 10, success: true, workspace_dir: temp.path().to_path_buf(), @@ -603,6 +619,8 @@ mod tests { content: "hello".into(), thread_ts: None, response: "should not persist".into(), + provider: "test-provider".into(), + model: "test-model".into(), elapsed_ms: 10, success: true, workspace_dir: stale.path().to_path_buf(), @@ -652,6 +670,8 @@ mod tests { content: "hello".into(), thread_ts: None, response: "from workspace B — must be dropped".into(), + provider: "test-provider".into(), + model: "test-model".into(), elapsed_ms: 5, success: true, workspace_dir: workspace_b.path().to_path_buf(), @@ -668,6 +688,8 @@ mod tests { content: "hello".into(), thread_ts: None, response: "from workspace A — should persist".into(), + provider: "test-provider".into(), + model: "test-model".into(), elapsed_ms: 10, success: true, workspace_dir: workspace_a.path().to_path_buf(), diff --git a/src/openhuman/routing/provider.rs b/src/openhuman/routing/provider.rs index 60f6ea122..d26f733ed 100644 --- a/src/openhuman/routing/provider.rs +++ b/src/openhuman/routing/provider.rs @@ -20,6 +20,7 @@ use async_trait::async_trait; use crate::openhuman::config::{ MODEL_AGENTIC_V1, MODEL_CHAT_V1, MODEL_CODING_V1, MODEL_REASONING_V1, }; +use crate::openhuman::inference::provider::record_resolved_provider_route; use crate::openhuman::inference::provider::traits::{ ChatMessage, ChatRequest, ChatResponse, Provider, ProviderCapabilities, StreamChunk, StreamError, StreamOptions, StreamResult, ToolsPayload, @@ -189,24 +190,26 @@ impl IntelligentRoutingProvider { match &result { Err(e) => { if !privacy_required { - if let Some(RoutingTarget::Remote { .. }) = fallback { + if let Some(RoutingTarget::Remote { model }) = fallback { tracing::warn!( hint, error = ?e, "[routing] local call failed, retrying with remote" ); + record_resolved_provider_route("remote", model); return (fallback_fn.await, true); } } (result, false) } Ok(text) if !privacy_required && quality::is_low_quality(text) => { - if let Some(RoutingTarget::Remote { .. }) = fallback { + if let Some(RoutingTarget::Remote { model }) = fallback { tracing::warn!( hint, response_preview = truncate_safe(text, 80), "[routing] local response low quality, retrying with remote" ); + record_resolved_provider_route("remote", model); return (fallback_fn.await, true); } (result, false) @@ -228,6 +231,7 @@ impl IntelligentRoutingProvider { let (result, fallback_occurred) = match &primary { RoutingTarget::Local { model: m } => { tracing::debug!(model = m.as_str(), hint = model, "[routing] → local"); + record_resolved_provider_route("local", m); let m = m.clone(); let fb_model = fallback .as_ref() @@ -253,6 +257,7 @@ impl IntelligentRoutingProvider { } RoutingTarget::Remote { model: m } => { tracing::debug!(model = m.as_str(), hint = model, "[routing] → remote"); + record_resolved_provider_route("remote", m); ( self.remote .chat_with_system(system_prompt, message, m, temperature) @@ -312,11 +317,13 @@ impl IntelligentRoutingProvider { let result = match &effective_primary { RoutingTarget::Local { model: m } => { + record_resolved_provider_route("local", m); let r = self.local.chat(request, m, temperature).await; if should_fallback(&r, self.hints.privacy_required, &fallback) { if let Some(RoutingTarget::Remote { model: fb }) = &fallback { tracing::warn!(hint = model, "[routing] local chat fallback → remote"); fallback_occurred = true; + record_resolved_provider_route("remote", fb); self.remote.chat(request, fb, temperature).await } else { r @@ -325,7 +332,10 @@ impl IntelligentRoutingProvider { r } } - RoutingTarget::Remote { model: m } => self.remote.chat(request, m, temperature).await, + RoutingTarget::Remote { model: m } => { + record_resolved_provider_route("remote", m); + self.remote.chat(request, m, temperature).await + } }; let (input_tokens, output_tokens, cost_usd) = match &result { @@ -398,6 +408,7 @@ impl Provider for IntelligentRoutingProvider { let result = match &primary { RoutingTarget::Local { model: m } => { + record_resolved_provider_route("local", m); let r = self.local.chat_with_history(messages, m, temperature).await; let do_fallback = !self.hints.privacy_required && fallback.is_some() @@ -412,6 +423,7 @@ impl Provider for IntelligentRoutingProvider { "[routing] local history failed/low-quality → remote" ); fallback_occurred = true; + record_resolved_provider_route("remote", fb); self.remote .chat_with_history(messages, fb, temperature) .await @@ -423,6 +435,7 @@ impl Provider for IntelligentRoutingProvider { } } RoutingTarget::Remote { model: m } => { + record_resolved_provider_route("remote", m); self.remote .chat_with_history(messages, m, temperature) .await diff --git a/src/openhuman/routing/provider_tests.rs b/src/openhuman/routing/provider_tests.rs index 4369314e7..07e254424 100644 --- a/src/openhuman/routing/provider_tests.rs +++ b/src/openhuman/routing/provider_tests.rs @@ -115,6 +115,37 @@ async fn local_used_when_healthy_and_lightweight() { assert_eq!(local.last_model(), "gemma3:4b-it-qat"); } +#[tokio::test] +async fn records_local_route_for_lightweight_task() { + let local = MockProvider::new("local", "Great reaction!"); + let remote = MockProvider::new("remote", "remote-resp"); + let health = LocalHealthChecker::seeded(true); + + let r = router( + Arc::clone(&local), + Arc::clone(&remote), + health, + RoutingHints::default(), + ); + + let recorded = + crate::openhuman::inference::provider::with_resolved_provider_route_scope(async { + let result = r + .chat_with_system(None, "React to this", "hint:reaction", 0.7) + .await + .unwrap(); + assert_eq!(result, "Great reaction!"); + crate::openhuman::inference::provider::current_resolved_provider_route() + }) + .await + .expect("intelligent routing should record the selected local route"); + + assert_eq!(recorded.provider, "local"); + assert_eq!(recorded.model, "gemma3:4b-it-qat"); + assert_eq!(local.calls(), 1); + assert_eq!(remote.calls(), 0); +} + #[tokio::test] async fn medium_without_hints_uses_remote() { let local = MockProvider::new("local", "Here is a summary."); diff --git a/tests/inference_agent_raw_coverage_e2e.rs b/tests/inference_agent_raw_coverage_e2e.rs index 103fb171c..b4acfb8d7 100644 --- a/tests/inference_agent_raw_coverage_e2e.rs +++ b/tests/inference_agent_raw_coverage_e2e.rs @@ -2927,9 +2927,9 @@ async fn agent_triage_evaluator_covers_native_dispatch_decision_and_deferred_pat && msg.content.contains("SOURCE: webhook") && msg.content.contains("PAYLOAD:") })); - Ok(AgentTurnResponse { - text: r#"{"action":"drop","reason":"already handled"}"#.into(), - }) + Ok(AgentTurnResponse::new( + r#"{"action":"drop","reason":"already handled"}"#, + )) }, ); let cloud = ResolvedProvider { @@ -2989,12 +2989,10 @@ async fn agent_triage_evaluator_covers_native_dispatch_decision_and_deferred_pat async move { let attempt = attempts_for_handler.fetch_add(1, Ordering::SeqCst); match attempt { - 0 | 1 => Ok(AgentTurnResponse { - text: "not json".into(), - }), - _ => Ok(AgentTurnResponse { - text: r#"{"action":"escalate","target_agent":"orchestrator","prompt":"follow up","reason":"needs work"}"#.into(), - }), + 0 | 1 => Ok(AgentTurnResponse::new("not json")), + _ => Ok(AgentTurnResponse::new( + r#"{"action":"escalate","target_agent":"orchestrator","prompt":"follow up","reason":"needs work"}"#, + )), } } },