admin/openhuman
1
0
Fork 0
mirror of https://github.com/tinyhumansai/openhuman.git synced 2026-07-27 21:08:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

[codex] Record channel response model metadata (#3682)

Browse Source 
Co-authored-by: Ron <ron@SpinMaster.local>
Co-authored-by: M3gA-Mind <megamind@mahadao.com>
This commit is contained in:
Ron Liu
2026-06-30 19:07:50 +05:30
committed by GitHub
co-authored by Ron M3gA-Mind
parent b237c33a0c
commit f84eec5335
21 changed files with 504 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/event_bus/events.rs
+4
View File
@@ -354,6 +354,10 @@ pub enum DomainEvent {
content: String,
thread_ts: Option<String>,
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.
src/core/event_bus/events_tests.rs
+2
View File
@@ -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"),
src/core/event_bus/native_request.rs
+1 -1
View File
@@ -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))
//! },
//! );
//!
src/openhuman/agent/bus.rs
+88 -46
View File
@@ -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<String>,
/// 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<String>,
}
impl AgentTurnResponse {
pub fn new(text: impl Into<String>) -> Self {
Self {
text: text.into(),
resolved_provider: None,
resolved_model: None,
}
}
pub fn with_resolved_route(
text: impl Into<String>,
provider: impl Into<String>,
model: impl Into<String>,
) -> 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"))
},
);
src/openhuman/agent/triage/evaluator_tests.rs
+14 -33
View File
@@ -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;
src/openhuman/channels/providers/telegram/bus_tests.rs
+6
View File
@@ -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(),
src/openhuman/channels/runtime/dispatch/processor.rs
+32 -10
View File
@@ -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(),
src/openhuman/channels/runtime/test_support.rs
+1 -3
View File
@@ -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)),
}
}
}
src/openhuman/channels/tests/discord_integration.rs
+1 -3
View File
@@ -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;
src/openhuman/channels/tests/runtime_dispatch.rs
+102 -12
View File
@@ -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<dyn Channel> = 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;
src/openhuman/channels/tests/telegram_integration.rs
+1 -3
View File
@@ -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;
src/openhuman/inference/provider/mod.rs
+5
View File
@@ -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,
};
src/openhuman/inference/provider/reliable.rs
+5
View File
@@ -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
src/openhuman/inference/provider/reliable_tests.rs
+48
View File
@@ -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(
src/openhuman/inference/provider/resolved_route.rs
+86
View File
@@ -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<Mutex<Option<ResolvedProviderRoute>>>;
tokio::task_local! {
static RESOLVED_PROVIDER_ROUTE: RouteSlot;
}
pub async fn with_resolved_provider_route_scope<F>(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<String>, model: impl Into<String>) {
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<ResolvedProviderRoute> {
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());
}
}
src/openhuman/inference/provider/router.rs
+8 -3
View File
@@ -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<String> {
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<ChatResponse> {
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<ChatResponse> {
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
src/openhuman/inference/provider/router_tests.rs
+24
View File
@@ -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"));
src/openhuman/memory_conversations/bus.rs
+22
View File
@@ -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<bool>,
elapsed_ms: Option<u64>,
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(),
src/openhuman/routing/provider.rs
+16 -3
View File
@@ -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
src/openhuman/routing/provider_tests.rs
+31
View File
@@ -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.");
tests/inference_agent_raw_coverage_e2e.rs
+7 -9
View File
@@ -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"}"#,
)),
}
}
},