diff --git a/src/openhuman/agent/schemas.rs b/src/openhuman/agent/schemas.rs index 525155c63..f56f9cd13 100644 --- a/src/openhuman/agent/schemas.rs +++ b/src/openhuman/agent/schemas.rs @@ -316,26 +316,47 @@ fn handle_triage_evaluate(params: Map) -> ControllerFuture { "[rpc][agent] running triage pipeline" ); - let run = crate::openhuman::agent::triage::run_triage(&envelope) + let outcome = crate::openhuman::agent::triage::run_triage(&envelope) .await .map_err(|e| format!("triage evaluation failed: {e}"))?; let dry_run = p.dry_run.unwrap_or(false); - if !dry_run { - crate::openhuman::agent::triage::apply_decision(run.clone(), &envelope) - .await - .map_err(|e| format!("apply_decision failed: {e}"))?; - } + match outcome { + crate::openhuman::agent::triage::TriageOutcome::Decision(run) => { + if !dry_run { + crate::openhuman::agent::triage::apply_decision(run.clone(), &envelope) + .await + .map_err(|e| format!("apply_decision failed: {e}"))?; + } - Ok(serde_json::json!({ - "decision": run.decision.action.as_str(), - "target_agent": run.decision.target_agent, - "prompt": run.decision.prompt, - "reason": run.decision.reason, - "used_local": run.used_local, - "latency_ms": run.latency_ms, - "dry_run": dry_run, - })) + Ok(serde_json::json!({ + "decision": run.decision.action.as_str(), + "target_agent": run.decision.target_agent, + "prompt": run.decision.prompt, + "reason": run.decision.reason, + "used_local": run.used_local, + "latency_ms": run.latency_ms, + "resolution_path": run.resolution_path.as_str(), + "dry_run": dry_run, + })) + } + crate::openhuman::agent::triage::TriageOutcome::Deferred { + defer_until_ms, + reason, + } => { + // Deferred outcome: the chain (cloud → cloud-retry → + // local) all failed; the caller is expected to + // re-issue this trigger after `defer_until_ms`. No + // side effects fire on this path. + Ok(serde_json::json!({ + "decision": "deferred", + "resolution_path": "deferred", + "defer_until_ms": defer_until_ms, + "reason": reason, + "dry_run": dry_run, + })) + } + } }) } diff --git a/src/openhuman/agent/triage/escalation.rs b/src/openhuman/agent/triage/escalation.rs index c25d20a25..bbf16689f 100644 --- a/src/openhuman/agent/triage/escalation.rs +++ b/src/openhuman/agent/triage/escalation.rs @@ -232,6 +232,7 @@ mod tests { }, used_local: false, latency_ms: 9, + resolution_path: super::super::evaluator::TriageResolutionPath::Cloud, } } @@ -245,6 +246,7 @@ mod tests { }, used_local: false, latency_ms: 9, + resolution_path: super::super::evaluator::TriageResolutionPath::Cloud, } } diff --git a/src/openhuman/agent/triage/evaluator.rs b/src/openhuman/agent/triage/evaluator.rs index f08b15b9d..99e623d54 100644 --- a/src/openhuman/agent/triage/evaluator.rs +++ b/src/openhuman/agent/triage/evaluator.rs @@ -1,21 +1,22 @@ //! Build the turn, dispatch `agent.run_turn`, parse the reply. //! -//! This is the core of the triage pipeline. It: +//! This is the core of the triage pipeline. It implements a tiered +//! fallback chain (issue #1257): //! -//! 1. Resolves a provider via [`super::routing::resolve_provider`] -//! (commit 1 = always remote; commit 2 = local-or-remote with probe -//! + cache). -//! 2. Looks up the `trigger_triage` built-in agent definition from -//! the global [`AgentDefinitionRegistry`]. -//! 3. Builds a [`ChatMessage`] history: the definition's system -//! prompt body + a user message summarising the envelope. -//! 4. Dispatches the turn through the existing -//! [`crate::openhuman::agent::bus::AGENT_RUN_TURN_METHOD`] native -//! request so tests can override the handler via -//! [`crate::openhuman::agent::bus::mock_agent_run_turn`]. -//! 5. Parses the reply with -//! [`super::decision::parse_triage_decision`] — tolerant enough to -//! accept whatever 1B-parameter output looks like today. +//! ```text +//! cloud (initial) +//! ├── 429 / transient (5xx / timeout / connection) ──► retry once +//! │ └── still failing ──► local fallback +//! └── ok ──► resolution_path = Cloud | CloudAfterRetry +//! +//! local fallback +//! ├── ok ──► resolution_path = LocalFallback +//! └── failed ──► TriageOutcome::Deferred { until_ms, reason } +//! ``` +//! +//! Non-transient cloud failures (auth, malformed prompt, model not +//! found, parse failure) bubble up immediately — there's no point +//! retrying them and the local arm wouldn't help either. //! //! ## Why `run_tool_call_loop` doesn't care about `tools_registry = []` //! @@ -23,11 +24,10 @@ //! `run_tool_call_loop` implementation in //! `src/openhuman/agent/harness/tool_loop.rs` handles an empty registry //! by just doing a plain `chat_with_history` under the hood — no tool -//! schemas are sent to the backend. That's exactly what we want: one -//! LLM round-trip, no chained tool calls. +//! schemas are sent to the backend. use std::sync::Arc; -use std::time::Instant; +use std::time::{Duration, Instant}; use anyhow::{anyhow, Context}; @@ -36,6 +36,9 @@ use crate::openhuman::agent::bus::{AgentTurnRequest, AgentTurnResponse, AGENT_RU use crate::openhuman::agent::harness::definition::{AgentDefinition, PromptSource}; use crate::openhuman::agent::harness::AgentDefinitionRegistry; use crate::openhuman::config::MultimodalConfig; +use crate::openhuman::providers::reliable::{ + is_rate_limited, is_upstream_unhealthy, parse_retry_after_ms, +}; use crate::openhuman::providers::ChatMessage; use crate::openhuman::config::Config; @@ -43,103 +46,214 @@ use crate::openhuman::config::Config; use super::decision::{parse_triage_decision, ParseError, TriageDecision}; use super::envelope::TriggerEnvelope; use super::events; -use super::routing::{resolve_provider_with_config, ResolvedProvider}; +use super::routing::{ + build_local_provider_with_config, resolve_provider_with_config, ResolvedProvider, +}; -/// Agent definition id for the built-in triage classifier. Hard-coded -/// so a rogue workspace TOML can't override it to point at a different -/// agent — the triage pipeline is tightly coupled to the prompt -/// contract in `trigger_triage/prompt.md`. +/// Agent definition id for the built-in triage classifier. pub const TRIGGER_TRIAGE_AGENT_ID: &str = "trigger_triage"; -/// How much of the raw payload we inline into the user message. Picked -/// so a huge Gmail body cannot blow the local model's context window. -/// The classifier only needs the gist — downstream agents (orchestrator, -/// trigger_reactor) can re-read the full payload if they need it. +/// How much of the raw payload we inline into the user message. const PAYLOAD_INLINE_LIMIT_BYTES: usize = 8 * 1024; -/// Final output of a single triage run — the parsed decision plus -/// bookkeeping fields published on the domain event. -#[derive(Debug, Clone)] -pub struct TriageRun { - pub decision: TriageDecision, - pub used_local: bool, - pub latency_ms: u64, +/// Cap on how long to wait for a server-supplied `Retry-After` before +/// giving up on the cloud arm and falling through to local. Mirrors +/// the cap in `ReliableProvider::compute_backoff`. +const RETRY_AFTER_CAP: Duration = Duration::from_millis(30_000); + +/// Default backoff for transient (non-rate-limit) cloud failures +/// before the single retry. Short enough to keep tail latency +/// bounded; long enough for a wedged TCP connection to give up. +const TRANSIENT_BACKOFF: Duration = Duration::from_millis(500); + +/// How far in the future a Deferred outcome asks the caller to retry. +/// A short tick mirrors the issue's "next tick retries the whole +/// chain" language — long enough to shed a thundering herd, short +/// enough that user-visible latency on transient outages stays in the +/// tens of seconds. +const DEFER_WAKEUP_MS: i64 = 30_000; + +/// Which arm produced this triage decision. Surfaced on `TriageRun` +/// so the orchestrator can colour-code degraded turns and show the +/// state in `/debug` views. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum TriageResolutionPath { + /// Cloud succeeded on the initial attempt. + Cloud, + /// Cloud succeeded on the retry after a 429 / transient failure. + CloudAfterRetry, + /// Cloud failed twice; the local arm produced the decision. + LocalFallback, } -/// Run the triage classifier against a trigger envelope. -/// -/// This is the main entry point for trigger classification. It performs the following: -/// 1. Resolves an appropriate provider (preferring local LLMs for speed). -/// 2. Dispatches a single LLM turn using the `trigger_triage` archetype. -/// 3. Parses the resulting JSON decision. -/// 4. If the local attempt fails or produces garbage, automatically retries on a -/// remote provider for maximum reliability. -/// -/// On success returns a [`TriageRun`] containing the decision and performance metrics. -pub async fn run_triage(envelope: &TriggerEnvelope) -> anyhow::Result { - // Load the config once and reuse it for both the first attempt and - // any retry that falls back to remote. `Config::load_or_init` is - // relatively heavy (disk + env merge) so paying it twice would - // double the tail latency of a degraded-local trigger. - let config = Config::load_or_init() - .await - .context("loading config for triage turn")?; - let resolved = resolve_provider_with_config(&config) - .await - .context("resolving provider for triage turn")?; - - // Single attempt — triage always uses the remote provider. - // The local-AI retry path has been removed: used_local is always false - // so that branch could never fire, and mark_degraded no longer exists. - match run_triage_with_resolved(resolved, envelope).await { - Ok(run) => Ok(run), - Err(err) => { - let reason = format!("{err}"); - events::publish_failed(envelope, &reason); - Err(err) +impl TriageResolutionPath { + pub fn as_str(self) -> &'static str { + match self { + Self::Cloud => "cloud", + Self::CloudAfterRetry => "cloud-after-retry", + Self::LocalFallback => "local-fallback", } } } -/// Sentinel error wrapper the retry path looks for. We attach it to -/// recoverable failures (handler error on local, parse failure on -/// local) so `run_triage` can decide whether a second attempt is worth -/// running. Unrecoverable failures (missing registry, missing built-in -/// definition, etc.) surface as plain `anyhow::Error` and skip the -/// retry loop. -#[derive(Debug)] -struct TurnOutcomeFailure { - used_local: bool, - kind: &'static str, - message: String, +/// Final output of a single triage run when a decision was produced. +#[derive(Debug, Clone)] +pub struct TriageRun { + pub decision: TriageDecision, + /// `true` when the producing arm was local — kept for telemetry + /// compatibility with subscribers that read this field. Equivalent + /// to `resolution_path == LocalFallback`. + pub used_local: bool, + pub latency_ms: u64, + pub resolution_path: TriageResolutionPath, } -impl std::fmt::Display for TurnOutcomeFailure { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!( - f, - "triage {} ({}): {}", - self.kind, - if self.used_local { "local" } else { "remote" }, - self.message - ) +/// Outcome of [`run_triage`]. Either a parsed decision or a +/// deferral asking the caller to retry the whole chain after +/// `defer_until_ms` (Unix epoch millis). +#[derive(Debug, Clone)] +pub enum TriageOutcome { + Decision(TriageRun), + Deferred { + /// Unix epoch millis at which the caller should re-run the + /// triage chain. + defer_until_ms: i64, + /// Short human-readable reason — already scrubbed; safe to log. + reason: String, + }, +} + +impl TriageOutcome { + pub fn into_decision(self) -> Option { + match self { + TriageOutcome::Decision(run) => Some(run), + TriageOutcome::Deferred { .. } => None, + } } } -impl std::error::Error for TurnOutcomeFailure {} - -/// Inner half of [`run_triage`] that takes an already-resolved -/// [`ResolvedProvider`] instead of calling `routing::resolve_provider`. +/// Run the triage classifier with the full tiered fallback chain. /// -/// Split out so unit tests can inject a stub provider without loading -/// real config from disk or constructing a real backend client. The -/// public [`run_triage`] path delegates here after resolving. Commit 2 -/// will also call this from the "parse-failed → retry on remote" path -/// with a second resolved provider. -pub async fn run_triage_with_resolved( - resolved: ResolvedProvider, +/// 1. Resolve the cloud provider. +/// 2. Try cloud; on 429 / transient, sleep and retry once. +/// 3. On a second 429 / transient, build the local provider and +/// fall back to it (acquiring the global LLM permit). +/// 4. On local failure, return `TriageOutcome::Deferred` so the +/// caller (typically a trigger-handler RPC) can reschedule. +pub async fn run_triage(envelope: &TriggerEnvelope) -> anyhow::Result { + let config = Config::load_or_init() + .await + .context("loading config for triage turn")?; + let cloud = resolve_provider_with_config(&config) + .await + .context("resolving provider for triage turn")?; + let local = build_local_provider_with_config(&config); + + let outcome = run_triage_with_arms(cloud, local, envelope).await; + if let Err(err) = &outcome { + events::publish_failed(envelope, &format!("{err}")); + } + outcome +} + +/// Inner driver for [`run_triage`] that takes already-resolved arms. +/// Tests inject stub providers via this entry point. +pub async fn run_triage_with_arms( + cloud: ResolvedProvider, + local: Option, envelope: &TriggerEnvelope, -) -> anyhow::Result { +) -> anyhow::Result { + // ── Cloud arm ────────────────────────────────────────────────── + match try_arm(&cloud, envelope, TriageResolutionPath::Cloud).await { + Ok(run) => return Ok(TriageOutcome::Decision(run)), + Err(ArmError::Fatal(err)) => return Err(err), + Err(ArmError::Retryable { retry_after_ms, .. }) => { + // Sleep before the cloud retry. Honour Retry-After when + // present; otherwise use a short backoff so the second + // attempt has a real chance of finding the upstream + // recovered. + let sleep_ms = retry_after_ms + .map(|ms| Duration::from_millis(ms).min(RETRY_AFTER_CAP)) + .unwrap_or(TRANSIENT_BACKOFF); + tracing::info!( + sleep_ms = sleep_ms.as_millis() as u64, + had_retry_after = retry_after_ms.is_some(), + "[triage::evaluator] cloud retry pending after retryable failure" + ); + tokio::time::sleep(sleep_ms).await; + + match try_arm(&cloud, envelope, TriageResolutionPath::CloudAfterRetry).await { + Ok(run) => return Ok(TriageOutcome::Decision(run)), + Err(ArmError::Fatal(err)) => return Err(err), + Err(ArmError::Retryable { .. }) => { + // Exhausted cloud budget — fall through to local. + tracing::warn!( + "[triage::evaluator] cloud retry budget exhausted; \ + falling back to local arm" + ); + } + } + } + } + + // ── Local fallback ───────────────────────────────────────────── + let Some(local) = local else { + // No local arm available at all (runtime disabled, no model + // configured) — the only honest outcome is a deferral so the + // next tick retries the whole chain. + return Ok(TriageOutcome::Deferred { + defer_until_ms: now_ms().saturating_add(DEFER_WAKEUP_MS), + reason: "cloud retry exhausted; local arm unavailable".to_string(), + }); + }; + + // Hold the global LLM permit for the lifetime of the local turn — + // protects laptop RAM from concurrent local model calls (#1073). + let _gate_permit = crate::openhuman::scheduler_gate::wait_for_capacity().await; + + match try_arm(&local, envelope, TriageResolutionPath::LocalFallback).await { + Ok(run) => Ok(TriageOutcome::Decision(run)), + Err(ArmError::Fatal(err)) | Err(ArmError::Retryable { source: err, .. }) => { + // Local also failed — defer rather than surface a hard + // error. Today's "hard fail" is the wrong default for a + // transient blocker per #1257. + let reason = format!("cloud + local both failed: {err}"); + tracing::warn!( + error = %reason, + defer_ms = DEFER_WAKEUP_MS, + "[triage::evaluator] both arms failed; deferring" + ); + Ok(TriageOutcome::Deferred { + defer_until_ms: now_ms().saturating_add(DEFER_WAKEUP_MS), + reason, + }) + } + } +} + +/// Single-arm execution result. `Retryable` lets the orchestrator +/// decide whether to sleep + retry on the same arm (cloud) or to fall +/// through (local). `Fatal` short-circuits the whole chain. +enum ArmError { + /// 429 / 5xx / timeout / connection — the kind of failure where + /// trying again later might help. + Retryable { + retry_after_ms: Option, + source: anyhow::Error, + }, + /// Auth failure, missing model, prompt parse error, registry + /// missing, etc. — retry / fallback would not change the result. + Fatal(anyhow::Error), +} + +/// Run a single arm: dispatch the agent turn through the native bus +/// and parse the reply. Classifies any error so the caller can decide +/// what to do next. +async fn try_arm( + resolved: &ResolvedProvider, + envelope: &TriggerEnvelope, + intended_path: TriageResolutionPath, +) -> Result { let started = Instant::now(); tracing::debug!( @@ -148,81 +262,50 @@ pub async fn run_triage_with_resolved( external_id = %envelope.external_id, provider = %resolved.provider_name, used_local = resolved.used_local, + path = intended_path.as_str(), "[triage::evaluator] starting triage turn" ); - let ResolvedProvider { - provider, - provider_name, - model, - used_local, - } = resolved; - - // ── Look up the built-in agent definition ────────────────────── let registry = AgentDefinitionRegistry::global().ok_or_else(|| { - anyhow!( + ArmError::Fatal(anyhow!( "AgentDefinitionRegistry not initialised — did startup wiring \ skip `init_global`?" - ) + )) })?; let definition = registry.get(TRIGGER_TRIAGE_AGENT_ID).ok_or_else(|| { - anyhow!("built-in `{TRIGGER_TRIAGE_AGENT_ID}` definition missing from registry") + ArmError::Fatal(anyhow!( + "built-in `{TRIGGER_TRIAGE_AGENT_ID}` definition missing from registry" + )) })?; - // ── Build the turn history ────────────────────────────────────── - let system_prompt = extract_inline_prompt(definition).context( - "trigger_triage agent definition must ship an inline prompt body \ - (bug: the built-in loader injects one at startup)", - )?; + let system_prompt = extract_inline_prompt(&definition).ok_or_else(|| { + ArmError::Fatal(anyhow!( + "trigger_triage agent definition must ship an inline prompt body" + )) + })?; let user_message = render_user_message(envelope); let history = vec![ ChatMessage::system(&system_prompt), ChatMessage::user(&user_message), ]; - // ── Dispatch via the native bus so tests can stub it ──────────── let request = AgentTurnRequest { - provider: Arc::clone(&provider), + provider: Arc::clone(&resolved.provider), history, tools_registry: Arc::new(Vec::new()), - provider_name: provider_name.clone(), - model: model.clone(), + provider_name: resolved.provider_name.clone(), + model: resolved.model.clone(), temperature: definition.temperature, silent: true, channel_name: "triage".to_string(), multimodal: MultimodalConfig::default(), - // Single round-trip: the triage agent has zero tools so this - // cap is only a safety net. max_tool_iterations: 1, on_delta: None, - // The triage classifier runs against an empty tools registry - // by design and emits a structured JSON decision rather than - // calling tools — record the agent identity for tracing but - // leave the visible-tool filter unset so the legacy unfiltered - // behaviour is preserved. target_agent_id: Some("trigger_triage".to_string()), visible_tool_names: None, extra_tools: Vec::new(), on_progress: None, }; - tracing::debug!( - provider = %provider_name, - model = %model, - used_local = used_local, - "[triage::evaluator] dispatching {AGENT_RUN_TURN_METHOD}" - ); - - // Gate the LLM round-trip on the scheduler's global single slot - // ONLY when the resolved provider is local. Cloud routes already - // have their own (forthcoming) rate-limiter and are not the - // resource that the gate exists to protect — gating cloud here - // would just delay triage for no reason. - let _gate_permit = if used_local { - log::debug!("[triage::evaluator] local route — acquiring scheduler_gate llm permit"); - crate::openhuman::scheduler_gate::wait_for_capacity().await - } else { - None - }; let response = match request_native_global::( AGENT_RUN_TURN_METHOD, @@ -238,44 +321,52 @@ pub async fn run_triage_with_resolved( }; tracing::warn!( error = %message, - used_local = used_local, + path = intended_path.as_str(), "[triage::evaluator] agent turn dispatch failed" ); - // Wrap in TurnOutcomeFailure so the outer `run_triage` can - // decide whether to retry on remote. Only local failures - // are retry-eligible. - return Err(anyhow!(TurnOutcomeFailure { - used_local, - kind: "handler", - message, - })); + return Err(classify_error(message)); } }; - // ── Parse the classifier's reply ──────────────────────────────── let decision = match parse_triage_decision(&response.text) { Ok(d) => d, Err(parse_err) => { tracing::warn!( error = %parse_err, reply_chars = response.text.chars().count(), - used_local = used_local, + path = intended_path.as_str(), "[triage::evaluator] classifier reply did not parse" ); - return Err(anyhow!(TurnOutcomeFailure { - used_local, - kind: "parser", - message: format_parse_error(&parse_err), - })); + // A parse failure means the model produced unusable + // output. Retrying the same arm with the same prompt + // won't help, but on the *cloud* arm a parse failure is + // worth retrying once because the cloud model can be + // non-deterministic across calls. On the local arm we've + // already exhausted cloud and would just spin — treat it + // as fatal so the chain progresses to Deferred. + return Err(match intended_path { + TriageResolutionPath::Cloud => ArmError::Retryable { + retry_after_ms: None, + source: anyhow!( + "classifier reply did not parse: {}", + format_parse_error(&parse_err) + ), + }, + _ => ArmError::Fatal(anyhow!( + "classifier reply did not parse on {} arm: {}", + intended_path.as_str(), + format_parse_error(&parse_err) + )), + }); } }; let latency_ms = started.elapsed().as_millis() as u64; + let used_local = matches!(intended_path, TriageResolutionPath::LocalFallback); tracing::info!( source = %envelope.source.slug(), - label = %envelope.display_label, action = %decision.action.as_str(), - used_local = used_local, + path = intended_path.as_str(), latency_ms = latency_ms, "[triage::evaluator] classifier decision produced" ); @@ -284,20 +375,63 @@ pub async fn run_triage_with_resolved( decision, used_local, latency_ms, + resolution_path: intended_path, }) } -/// Pull the prompt body out of the definition. -/// -/// Built-ins use [`PromptSource::Dynamic`] (function-driven) and -/// custom TOML definitions may use `Inline`. Only `Inline` and -/// `Dynamic` are handled here — `File`-backed sources fall into the -/// wildcard arm and return `None`. For `Dynamic`, the builder is -/// invoked with a minimal -/// [`crate::openhuman::agent::harness::definition::PromptContext`] -/// since the triage classifier does not need tool lists or memory -/// context. Returning an option here (rather than panicking) lets the -/// caller surface a clean error to downstream logging. +/// Classify a handler-failure message string from the agent bus into +/// either a retryable (sleep + try again) or fatal (give up) error. +fn classify_error(message: String) -> ArmError { + let err = anyhow!("{message}"); + if is_rate_limited(&err) { + return ArmError::Retryable { + retry_after_ms: parse_retry_after_ms(&err), + source: err, + }; + } + if is_upstream_unhealthy(&err) || is_transient_string(&message) { + return ArmError::Retryable { + retry_after_ms: None, + source: err, + }; + } + ArmError::Fatal(err) +} + +/// Heuristic for transient cloud failures the provider stack didn't +/// already classify — connection resets, timeouts, generic 5xx text. +/// Mirrors the conservative match shape used by `is_upstream_unhealthy`. +fn is_transient_string(msg: &str) -> bool { + let lower = msg.to_lowercase(); + let hints = [ + "timed out", + "timeout", + "connection", + "connect error", + "broken pipe", + "reset by peer", + "deadline exceeded", + "temporarily unavailable", + ]; + if hints.iter().any(|h| lower.contains(h)) { + return true; + } + // Bare 5xx in the message body. Be careful not to match arbitrary + // numerals — only treat 5xx as transient. + for token in lower.split(|c: char| !c.is_ascii_digit()) { + if let Ok(code) = token.parse::() { + if (500..600).contains(&code) { + return true; + } + } + } + false +} + +fn now_ms() -> i64 { + chrono::Utc::now().timestamp_millis() +} + fn extract_inline_prompt(def: &AgentDefinition) -> Option { match &def.system_prompt { PromptSource::Inline(body) if !body.is_empty() => Some(body.clone()), @@ -342,9 +476,6 @@ fn extract_inline_prompt(def: &AgentDefinition) -> Option { } } -/// Render the user-side message that the classifier reads. The prompt -/// contract in `trigger_triage/prompt.md` describes the field layout; -/// we keep it trivially serialisable so a 1B model can reason over it. fn render_user_message(envelope: &TriggerEnvelope) -> String { let payload_string = truncate_payload(&envelope.payload, PAYLOAD_INLINE_LIMIT_BYTES); format!( @@ -359,9 +490,6 @@ fn render_user_message(envelope: &TriggerEnvelope) -> String { ) } -/// Format a [`ParseError`] for inclusion in a [`TurnOutcomeFailure`] -/// message. Keeps the source-error wrapper out of the string so the -/// retry log stays readable. fn format_parse_error(err: &ParseError) -> String { match err { ParseError::NoJsonObject => "classifier reply had no JSON object".to_string(), @@ -372,18 +500,12 @@ fn format_parse_error(err: &ParseError) -> String { } } -/// Pretty-print `payload` as JSON, truncate if it exceeds `max_bytes`, -/// and leave a `[...truncated N bytes]` marker so the classifier -/// understands the input was abridged. fn truncate_payload(payload: &serde_json::Value, max_bytes: usize) -> String { let pretty = serde_json::to_string_pretty(payload).unwrap_or_else(|_| payload.to_string()); if pretty.len() <= max_bytes { return pretty; } let dropped = pretty.len() - max_bytes; - // Split at a char boundary at or below `max_bytes` so we never - // produce an invalid UTF-8 slice when the payload contains - // multi-byte characters. let mut end = max_bytes; while end > 0 && !pretty.is_char_boundary(end) { end -= 1; diff --git a/src/openhuman/agent/triage/evaluator_tests.rs b/src/openhuman/agent/triage/evaluator_tests.rs index 0b4c1d2a1..e876e6360 100644 --- a/src/openhuman/agent/triage/evaluator_tests.rs +++ b/src/openhuman/agent/triage/evaluator_tests.rs @@ -29,17 +29,12 @@ fn truncate_payload_marks_truncation_and_stays_valid_utf8() { let big = serde_json::Value::String("😀".repeat(10_000)); let out = truncate_payload(&big, 128); assert!(out.contains("[...truncated")); - assert!(out.len() <= 128 + 64); // generous upper bound for the marker - // Round-trip to prove it's valid UTF-8 (otherwise format! would - // have panicked — this assertion is belt-and-braces). + assert!(out.len() <= 128 + 64); let _ = out.as_str(); } #[test] fn extract_inline_prompt_returns_body_for_trigger_triage_builtin() { - // Load the baked-in TOML+prompt directly so this test doesn't - // depend on `AgentDefinitionRegistry::init_global` having been - // called by the test runner. let builtin = BUILTINS .iter() .find(|b| b.id == TRIGGER_TRIAGE_AGENT_ID) @@ -53,23 +48,56 @@ fn extract_inline_prompt_returns_body_for_trigger_triage_builtin() { ); } -// ── Bus dispatch integration test ─────────────────────────────── +#[test] +fn classify_string_recognises_429_with_retry_after() { + let err = classify_error("HTTP 429 Too Many Requests; Retry-After: 2".to_string()); + match err { + ArmError::Retryable { + retry_after_ms: Some(ms), + .. + } => { + assert_eq!(ms, 2_000, "Retry-After: 2 → 2000 ms"); + } + _ => panic!("expected Retryable with retry_after_ms"), + } +} + +#[test] +fn classify_string_recognises_5xx_as_transient() { + let err = classify_error("upstream returned 503 Service Unavailable".to_string()); + assert!( + matches!(err, ArmError::Retryable { .. }), + "5xx should be Retryable" + ); +} + +#[test] +fn classify_string_recognises_timeout_as_transient() { + let err = classify_error("request timed out after 30s".to_string()); + assert!( + matches!(err, ArmError::Retryable { .. }), + "timeout should be Retryable" + ); +} + +#[test] +fn classify_string_treats_auth_failure_as_fatal() { + let err = classify_error("HTTP 401 unauthorized: invalid api key".to_string()); + assert!( + matches!(err, ArmError::Fatal(_)), + "auth failure should be Fatal" + ); +} + +// ── Tiered fallback integration tests ─────────────────────────── // -// Stubs `agent.run_turn` via `mock_agent_run_turn` and drives -// `run_triage_with_resolved` with an injected `ResolvedProvider`. -// Proves: -// 1. the evaluator routes its turn through the native bus -// 2. the dispatched `AgentTurnRequest` has the triage system -// prompt, a user message carrying the envelope label, empty -// tools_registry, and the `provider_name` / `model` the -// resolver returned -// 3. a canned JSON reply is parsed into the correct -// `TriageDecision` +// These drive `run_triage_with_arms` end-to-end through the agent +// bus, with a stateful stub that decides per-call whether to return +// success, a 429, a 5xx, or a fatal auth error. Each `cloud-then- +// local` test relies on call-ordering: cloud arm is exercised +// first; falling through to local arm uses a different +// `provider_name` we inspect to disambiguate. -/// Minimal `Provider` impl that satisfies the `Arc` -/// type in `ResolvedProvider`. The stubbed bus handler never -/// actually invokes any provider methods — if it did, these -/// methods would bail out loudly so the test fails fast. struct NoopProvider; #[async_trait] @@ -81,247 +109,258 @@ impl Provider for NoopProvider { _model: &str, _temperature: f64, ) -> anyhow::Result { - anyhow::bail!( - "NoopProvider::chat_with_system should never be called — \ - the mock_agent_run_turn stub short-circuits before the \ - real handler hits any provider method" - ) + anyhow::bail!("NoopProvider should never be called — bus mock short-circuits") } } -fn fake_resolved(used_local: bool) -> ResolvedProvider { +fn cloud_arm() -> ResolvedProvider { ResolvedProvider { provider: StdArc::new(NoopProvider) as StdArc, - provider_name: "stub-provider".to_string(), - model: "stub-model".to_string(), - used_local, + provider_name: "stub-cloud".to_string(), + model: "stub-cloud-model".to_string(), + used_local: false, } } -#[tokio::test] -async fn run_triage_dispatches_through_agent_run_turn_bus() { - // Registry must be available before `run_triage_with_resolved` - // looks up the `trigger_triage` definition. `init_global_builtins` - // is a no-op on subsequent calls so parallel tests are safe. - AgentDefinitionRegistry::init_global_builtins().expect("init_global_builtins"); +fn local_arm() -> ResolvedProvider { + ResolvedProvider { + provider: StdArc::new(NoopProvider) as StdArc, + provider_name: "stub-local".to_string(), + model: "stub-local-model".to_string(), + used_local: true, + } +} - let envelope = TriggerEnvelope::from_composio( +fn envelope() -> TriggerEnvelope { + TriggerEnvelope::from_composio( "gmail", "GMAIL_NEW_GMAIL_MESSAGE", - "trig-42", - "uuid-42", + "trig-x", + "uuid-x", json!({ "from": "ada@example.com", "subject": "ship it" }), - ); + ) +} - let stub_calls = StdArc::new(AtomicUsize::new(0)); - let stub_calls_handler = StdArc::clone(&stub_calls); +const VALID_JSON_REPLY: &str = "{\"action\":\"acknowledge\",\"reason\":\"all good\"}"; - // Capture of the dispatched request for deeper assertions - // after the bus round-trip completes. - let captured = StdArc::new(tokio::sync::Mutex::new( - None::<( - String, // provider_name - String, // model - usize, // history length - usize, // tools_registry length - String, // channel_name - String, // system prompt body (first 200 chars) - String, // user message body - )>, - )); - let captured_handler = StdArc::clone(&captured); +#[tokio::test] +async fn happy_path_returns_cloud_resolution() { + AgentDefinitionRegistry::init_global_builtins().expect("init_global_builtins"); - let _guard = mock_agent_run_turn(move |req| { - let calls = StdArc::clone(&stub_calls_handler); - let cap = StdArc::clone(&captured_handler); - async move { - calls.fetch_add(1, Ordering::SeqCst); - let system_preview = req - .history - .first() - .map(|m| m.content.chars().take(200).collect::()) - .unwrap_or_default(); - let user_msg = req - .history - .get(1) - .map(|m| m.content.clone()) - .unwrap_or_default(); - *cap.lock().await = Some(( - req.provider_name.clone(), - req.model.clone(), - req.history.len(), - req.tools_registry.len(), - req.channel_name.clone(), - system_preview, - user_msg, - )); - Ok(AgentTurnResponse { - text: - "Here's my call:\n```json\n{\"action\":\"drop\",\"reason\":\"test noise\"}\n```" - .to_string(), - }) - } + let _guard = mock_agent_run_turn(move |_req| async move { + Ok(AgentTurnResponse { + text: VALID_JSON_REPLY.to_string(), + }) }) .await; - let run = run_triage_with_resolved(fake_resolved(false), &envelope) + let outcome = run_triage_with_arms(cloud_arm(), Some(local_arm()), &envelope()) .await - .expect("run_triage should succeed with stub"); + .expect("happy path must succeed"); - // ── Stub was hit exactly once. - assert_eq!( - stub_calls.load(Ordering::SeqCst), - 1, - "stub handler must be invoked exactly once per triage run" - ); - - // ── Dispatched request shape. - let cap = captured.lock().await; - let (provider_name, model, hist_len, tools_len, channel, sys_preview, user_msg) = - cap.clone().expect("captured request"); - assert_eq!(provider_name, "stub-provider"); - assert_eq!(model, "stub-model"); - assert_eq!(hist_len, 2, "expected system + user message"); - assert_eq!(tools_len, 0, "trigger_triage has zero tools"); - assert_eq!(channel, "triage"); - assert!( - sys_preview.to_lowercase().contains("trigger"), - "system prompt should come from trigger_triage/prompt.md" - ); - assert!( - user_msg.contains("composio/gmail/GMAIL_NEW_GMAIL_MESSAGE"), - "user message must carry the envelope display label, got: {user_msg}" - ); - assert!( - user_msg.contains("ada@example.com"), - "user message must carry the payload, got: {user_msg}" - ); - - // ── Parsed decision matches the canned reply. - assert_eq!( - run.decision.action, - crate::openhuman::agent::triage::TriageAction::Drop - ); - assert_eq!(run.decision.reason, "test noise"); + let run = outcome.into_decision().expect("decision"); + assert_eq!(run.resolution_path, TriageResolutionPath::Cloud); assert!(!run.used_local); } #[tokio::test] -async fn remote_parse_failure_surfaces_as_error() { - // When a remote turn (used_local=false) produces an unparseable - // reply, the error surfaces directly — no retry is attempted - // because there's no "better" provider to fall back to. +async fn rate_limited_then_ok_marks_cloud_after_retry() { AgentDefinitionRegistry::init_global_builtins().expect("init_global_builtins"); - - let envelope = - TriggerEnvelope::from_composio("notion", "NOTION_PAGE_UPDATED", "t", "u", json!({})); - - let _guard = mock_agent_run_turn(move |_req| async move { - Ok(AgentTurnResponse { - text: "totally unparseable, no json here at all".to_string(), - }) - }) - .await; - - let err = run_triage_with_resolved(fake_resolved(false), &envelope) - .await - .expect_err("remote parse failure must surface as error"); - let msg = err.to_string(); - assert!( - msg.contains("parser") || msg.contains("JSON"), - "expected parser error message, got: {msg}" - ); -} - -#[tokio::test] -async fn local_parse_failure_is_retry_eligible() { - // When a local turn (used_local=true) produces an unparseable - // reply, the error is wrapped in TurnOutcomeFailure with - // used_local=true, so the outer `run_triage` can detect it - // and retry on remote. - AgentDefinitionRegistry::init_global_builtins().expect("init_global_builtins"); - - let envelope = TriggerEnvelope::from_composio("slack", "SLACK_MESSAGE", "t", "u", json!({})); - - let _guard = mock_agent_run_turn(move |_req| async move { - Ok(AgentTurnResponse { - text: "no json here".to_string(), - }) - }) - .await; - - let err = run_triage_with_resolved(fake_resolved(true), &envelope) - .await - .expect_err("local parse failure must surface as error"); - - // The error must be a TurnOutcomeFailure with used_local=true - // so the outer run_triage can detect it for retry. - let failure = err - .downcast_ref::() - .expect("error must be a TurnOutcomeFailure"); - assert!( - failure.used_local, - "TurnOutcomeFailure must report used_local=true for retry eligibility" - ); - assert_eq!(failure.kind, "parser"); -} - -#[tokio::test] -async fn stateful_stub_simulates_local_garbage_then_remote_success() { - // Proves the bus round-trip works with a stateful stub that - // returns garbage on call 1 (simulating local) and valid JSON - // on call 2 (simulating remote). This validates the exact - // sequence the retry path in `run_triage` exercises. - AgentDefinitionRegistry::init_global_builtins().expect("init_global_builtins"); - - let envelope = TriggerEnvelope::from_composio( - "github", - "GITHUB_PUSH", - "t", - "u", - json!({ "ref": "refs/heads/main" }), - ); - - let call_counter = StdArc::new(AtomicUsize::new(0)); - let counter_for_stub = StdArc::clone(&call_counter); + let counter = StdArc::new(AtomicUsize::new(0)); + let counter_for_stub = StdArc::clone(&counter); let _guard = mock_agent_run_turn(move |_req| { let counter = StdArc::clone(&counter_for_stub); async move { let n = counter.fetch_add(1, Ordering::SeqCst); if n == 0 { - // First call: simulate garbage local response - Ok(AgentTurnResponse { - text: "I have no idea what to do with this".to_string(), - }) + Err("HTTP 429 Too Many Requests; Retry-After: 0".to_string()) } else { - // Second call: valid JSON Ok(AgentTurnResponse { - text: "{\"action\":\"acknowledge\",\"reason\":\"valid on retry\"}".to_string(), + text: VALID_JSON_REPLY.to_string(), }) } } }) .await; - // Call 1: local (used_local=true) → expect parse failure - let err = run_triage_with_resolved(fake_resolved(true), &envelope) + let outcome = run_triage_with_arms(cloud_arm(), Some(local_arm()), &envelope()) .await - .expect_err("first call should fail (garbage)"); - let failure = err.downcast_ref::().unwrap(); - assert!(failure.used_local); + .expect("retry path must succeed"); - // Call 2: remote (used_local=false) → expect success - let run = run_triage_with_resolved(fake_resolved(false), &envelope) - .await - .expect("second call should succeed (valid JSON)"); - assert_eq!( - run.decision.action, - crate::openhuman::agent::triage::TriageAction::Acknowledge - ); - assert_eq!(run.decision.reason, "valid on retry"); + let run = outcome.into_decision().expect("decision"); + assert_eq!(run.resolution_path, TriageResolutionPath::CloudAfterRetry); assert!(!run.used_local); + assert_eq!(counter.load(Ordering::SeqCst), 2); +} - // Total: exactly 2 bus calls - assert_eq!(call_counter.load(Ordering::SeqCst), 2); +#[tokio::test] +async fn double_429_falls_through_to_local_fallback() { + AgentDefinitionRegistry::init_global_builtins().expect("init_global_builtins"); + let counter = StdArc::new(AtomicUsize::new(0)); + let counter_for_stub = StdArc::clone(&counter); + + let _guard = mock_agent_run_turn(move |req| { + let counter = StdArc::clone(&counter_for_stub); + async move { + let n = counter.fetch_add(1, Ordering::SeqCst); + if n < 2 { + // Cloud calls #1 and #2 both 429. + assert_eq!(req.provider_name, "stub-cloud", "first two calls hit cloud"); + Err("HTTP 429 Too Many Requests; Retry-After: 0".to_string()) + } 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(), + }) + } + } + }) + .await; + + let outcome = run_triage_with_arms(cloud_arm(), Some(local_arm()), &envelope()) + .await + .expect("local fallback must succeed"); + + let run = outcome.into_decision().expect("decision"); + assert_eq!(run.resolution_path, TriageResolutionPath::LocalFallback); + assert!(run.used_local); + assert_eq!(counter.load(Ordering::SeqCst), 3); +} + +#[tokio::test] +async fn cloud_5xx_falls_through_to_local_fallback() { + AgentDefinitionRegistry::init_global_builtins().expect("init_global_builtins"); + let counter = StdArc::new(AtomicUsize::new(0)); + let counter_for_stub = StdArc::clone(&counter); + + let _guard = mock_agent_run_turn(move |req| { + let counter = StdArc::clone(&counter_for_stub); + async move { + let n = counter.fetch_add(1, Ordering::SeqCst); + if n < 2 { + assert_eq!(req.provider_name, "stub-cloud"); + Err("upstream returned 502 Bad Gateway".to_string()) + } else { + assert_eq!(req.provider_name, "stub-local"); + Ok(AgentTurnResponse { + text: VALID_JSON_REPLY.to_string(), + }) + } + } + }) + .await; + + let outcome = run_triage_with_arms(cloud_arm(), Some(local_arm()), &envelope()) + .await + .expect("local fallback must succeed after 5xx"); + + let run = outcome.into_decision().expect("decision"); + assert_eq!(run.resolution_path, TriageResolutionPath::LocalFallback); +} + +#[tokio::test] +async fn cloud_then_local_failure_returns_deferred() { + AgentDefinitionRegistry::init_global_builtins().expect("init_global_builtins"); + let counter = StdArc::new(AtomicUsize::new(0)); + let counter_for_stub = StdArc::clone(&counter); + + let _guard = mock_agent_run_turn(move |_req| { + let counter = StdArc::clone(&counter_for_stub); + async move { + counter.fetch_add(1, Ordering::SeqCst); + // Every call fails transiently — cloud retry #1, retry #2, local. + Err("HTTP 503 Service Unavailable".to_string()) + } + }) + .await; + + let outcome = run_triage_with_arms(cloud_arm(), Some(local_arm()), &envelope()) + .await + .expect("Deferred is Ok, not Err"); + + match outcome { + TriageOutcome::Deferred { + defer_until_ms, + reason, + } => { + assert!( + defer_until_ms > chrono::Utc::now().timestamp_millis(), + "defer_until_ms must be in the future" + ); + assert!( + reason.to_lowercase().contains("503") || reason.contains("cloud"), + "reason should reference the upstream failure: {reason}" + ); + } + TriageOutcome::Decision(_) => panic!("expected Deferred, got Decision"), + } + assert_eq!(counter.load(Ordering::SeqCst), 3, "1 + retry + local = 3"); +} + +#[tokio::test] +async fn fatal_cloud_error_short_circuits_without_local_attempt() { + AgentDefinitionRegistry::init_global_builtins().expect("init_global_builtins"); + let counter = StdArc::new(AtomicUsize::new(0)); + let counter_for_stub = StdArc::clone(&counter); + + let _guard = mock_agent_run_turn(move |_req| { + let counter = StdArc::clone(&counter_for_stub); + async move { + counter.fetch_add(1, Ordering::SeqCst); + Err("HTTP 401 unauthorized: invalid api key".to_string()) + } + }) + .await; + + let err = run_triage_with_arms(cloud_arm(), Some(local_arm()), &envelope()) + .await + .expect_err("auth failure must surface as Err"); + + assert!( + err.to_string().to_lowercase().contains("401") + || err.to_string().to_lowercase().contains("unauthorized"), + "expected auth-related error message, got: {err}" + ); + assert_eq!( + counter.load(Ordering::SeqCst), + 1, + "fatal cloud error should not retry or fall back" + ); +} + +#[tokio::test] +async fn no_local_arm_returns_deferred_after_cloud_exhaustion() { + AgentDefinitionRegistry::init_global_builtins().expect("init_global_builtins"); + let counter = StdArc::new(AtomicUsize::new(0)); + let counter_for_stub = StdArc::clone(&counter); + + let _guard = mock_agent_run_turn(move |_req| { + let counter = StdArc::clone(&counter_for_stub); + async move { + counter.fetch_add(1, Ordering::SeqCst); + Err("HTTP 503 Service Unavailable".to_string()) + } + }) + .await; + + let outcome = run_triage_with_arms(cloud_arm(), None, &envelope()) + .await + .expect("Deferred is Ok"); + + match outcome { + TriageOutcome::Deferred { reason, .. } => { + assert!( + reason.contains("local arm unavailable"), + "reason should explain the missing local arm: {reason}" + ); + } + TriageOutcome::Decision(_) => panic!("expected Deferred"), + } + assert_eq!( + counter.load(Ordering::SeqCst), + 2, + "1 cloud + 1 retry, no local" + ); } diff --git a/src/openhuman/agent/triage/mod.rs b/src/openhuman/agent/triage/mod.rs index c58a19705..28a2f3e5a 100644 --- a/src/openhuman/agent/triage/mod.rs +++ b/src/openhuman/agent/triage/mod.rs @@ -41,5 +41,5 @@ pub mod routing; pub use decision::{parse_triage_decision, ParseError, TriageAction, TriageDecision}; pub use envelope::{TriggerEnvelope, TriggerSource}; pub use escalation::apply_decision; -pub use evaluator::{run_triage, TriageRun}; -pub use routing::{resolve_provider, ResolvedProvider}; +pub use evaluator::{run_triage, TriageOutcome, TriageResolutionPath, TriageRun}; +pub use routing::{build_local_provider_with_config, resolve_provider, ResolvedProvider}; diff --git a/src/openhuman/agent/triage/routing.rs b/src/openhuman/agent/triage/routing.rs index a4d4b545c..1be0f92c1 100644 --- a/src/openhuman/agent/triage/routing.rs +++ b/src/openhuman/agent/triage/routing.rs @@ -54,6 +54,75 @@ pub async fn resolve_provider_with_config(config: &Config) -> anyhow::Result Option { + use crate::openhuman::providers::compatible::{AuthStyle, OpenAiCompatibleProvider}; + + let local_cfg = &config.local_ai; + if !local_cfg.runtime_enabled { + tracing::debug!("[triage::routing] local arm disabled (runtime_enabled=false)"); + return None; + } + if local_cfg.chat_model_id.trim().is_empty() { + tracing::debug!("[triage::routing] local arm skipped (no chat_model_id configured)"); + return None; + } + + let override_base = std::env::var("OPENHUMAN_LOCAL_INFERENCE_URL") + .ok() + .map(|s| s.trim().trim_end_matches('/').to_string()) + .filter(|s| !s.is_empty()); + let provider_kind = local_cfg.provider.trim().to_ascii_lowercase(); + let use_openai_compat = override_base.is_some() + || matches!( + provider_kind.as_str(), + "llamacpp" | "llama-server" | "custom_openai" + ); + + let (label, base) = if use_openai_compat { + let base = override_base + .or_else(|| local_cfg.base_url.clone()) + .unwrap_or_else(|| "http://127.0.0.1:8080/v1".to_string()); + let label = if provider_kind == "custom_openai" { + "custom_openai" + } else { + "llamacpp" + }; + (label, base) + } else { + let ollama_base = crate::openhuman::local_ai::ollama_base_url(); + ("ollama", format!("{ollama_base}/v1")) + }; + + let provider: Arc = Arc::new(OpenAiCompatibleProvider::new( + label, + &base, + local_cfg.api_key.as_deref(), + AuthStyle::Bearer, + )); + tracing::debug!( + provider = %label, + model = %local_cfg.chat_model_id, + "[triage::routing] resolved local fallback provider" + ); + Some(ResolvedProvider { + provider, + provider_name: label.to_string(), + model: local_cfg.chat_model_id.clone(), + used_local: true, + }) +} + // ── Provider builder ──────────────────────────────────────────────────── /// Build the default remote routed backend provider. Same wiring as diff --git a/src/openhuman/composio/bus.rs b/src/openhuman/composio/bus.rs index cff919995..95743643f 100644 --- a/src/openhuman/composio/bus.rs +++ b/src/openhuman/composio/bus.rs @@ -52,7 +52,7 @@ use std::time::Duration; use async_trait::async_trait; use crate::core::event_bus::{subscribe_global, DomainEvent, EventHandler, SubscriptionHandle}; -use crate::openhuman::agent::triage::{apply_decision, run_triage, TriggerEnvelope}; +use crate::openhuman::agent::triage::{apply_decision, run_triage, TriageOutcome, TriggerEnvelope}; use crate::openhuman::composio::trigger_history; use crate::openhuman::config::rpc as config_rpc; @@ -279,7 +279,7 @@ impl EventHandler for ComposioTriggerSubscriber { // triage turn is an LLM round-trip that may take seconds. tokio::spawn(async move { match run_triage(&envelope).await { - Ok(run) => { + Ok(TriageOutcome::Decision(run)) => { if let Err(e) = apply_decision(run, &envelope).await { tracing::error!( label = %envelope.display_label, @@ -288,6 +288,21 @@ impl EventHandler for ComposioTriggerSubscriber { ); } } + Ok(TriageOutcome::Deferred { + defer_until_ms, + reason, + }) => { + // Tiered fallback exhausted both arms; the caller + // surface (composio bus) has no scheduler of its + // own — log and drop. The next composio fire will + // re-enter the chain. + tracing::warn!( + label = %envelope.display_label, + defer_until_ms = defer_until_ms, + reason = %reason, + "[composio][triage] run_triage deferred" + ); + } Err(e) => { tracing::error!( label = %envelope.display_label, diff --git a/src/openhuman/notifications/rpc.rs b/src/openhuman/notifications/rpc.rs index a504574bb..6d124d90c 100644 --- a/src/openhuman/notifications/rpc.rs +++ b/src/openhuman/notifications/rpc.rs @@ -13,7 +13,9 @@ use serde_json::{json, Map, Value}; use uuid::Uuid; use crate::core::event_bus::{publish_global, DomainEvent}; -use crate::openhuman::agent::triage::{apply_decision, run_triage, TriggerEnvelope, TriggerSource}; +use crate::openhuman::agent::triage::{ + apply_decision, run_triage, TriageOutcome, TriggerEnvelope, TriggerSource, +}; use crate::openhuman::config::rpc as config_rpc; use crate::rpc::RpcOutcome; @@ -105,7 +107,7 @@ pub async fn handle_ingest(params: Map) -> Result }; match run_triage(&envelope).await { - Ok(triage_run) => { + Ok(TriageOutcome::Decision(triage_run)) => { let action = triage_run.decision.action.as_str().to_string(); let reason = triage_run.decision.reason.clone(); // Map TriageAction → importance score heuristic. @@ -201,6 +203,21 @@ pub async fn handle_ingest(params: Map) -> Result "[notification_intel] published NotificationTriaged event" ); } + Ok(TriageOutcome::Deferred { + defer_until_ms, + reason, + }) => { + // Tiered fallback exhausted both arms; the next + // notification ingest re-enters the chain. Log only — + // notifications are inherently retryable on the next + // user fetch. + tracing::warn!( + id = %id_for_triage, + defer_until_ms = defer_until_ms, + reason = %reason, + "[notification_intel] triage deferred" + ); + } Err(e) => { tracing::warn!( id = %id_for_triage, diff --git a/src/openhuman/providers/reliable.rs b/src/openhuman/providers/reliable.rs index 064ecfb0e..a50015522 100644 --- a/src/openhuman/providers/reliable.rs +++ b/src/openhuman/providers/reliable.rs @@ -77,7 +77,7 @@ fn is_context_window_exceeded(err: &anyhow::Error) -> bool { /// Detect provider-side temporary capacity/outage errors that are often surfaced /// as HTTP 5xx with text like "no healthy upstream". -fn is_upstream_unhealthy(err: &anyhow::Error) -> bool { +pub(crate) fn is_upstream_unhealthy(err: &anyhow::Error) -> bool { let lower = err.to_string().to_lowercase(); lower.contains("no healthy upstream") || lower.contains("upstream unavailable") @@ -85,7 +85,7 @@ fn is_upstream_unhealthy(err: &anyhow::Error) -> bool { } /// Check if an error is a rate-limit (429) error. -fn is_rate_limited(err: &anyhow::Error) -> bool { +pub(crate) fn is_rate_limited(err: &anyhow::Error) -> bool { if let Some(reqwest_err) = err.downcast_ref::() { if let Some(status) = reqwest_err.status() { return status.as_u16() == 429; @@ -144,7 +144,7 @@ fn is_non_retryable_rate_limit(err: &anyhow::Error) -> bool { /// Try to extract a Retry-After value (in milliseconds) from an error message. /// Looks for patterns like `Retry-After: 5` or `retry_after: 2.5` in the error string. -fn parse_retry_after_ms(err: &anyhow::Error) -> Option { +pub(crate) fn parse_retry_after_ms(err: &anyhow::Error) -> Option { let msg = err.to_string(); let lower = msg.to_lowercase(); diff --git a/src/openhuman/webhooks/bus.rs b/src/openhuman/webhooks/bus.rs index 2a86e2a5f..bb68752c3 100644 --- a/src/openhuman/webhooks/bus.rs +++ b/src/openhuman/webhooks/bus.rs @@ -302,19 +302,27 @@ fn decode_webhook_body(base64_body: &str) -> Result { async fn run_agent_trigger( envelope: &crate::openhuman::agent::triage::TriggerEnvelope, ) -> Result { - let run = crate::openhuman::agent::triage::run_triage(envelope) + let outcome = crate::openhuman::agent::triage::run_triage(envelope) .await .map_err(|e| format!("triage evaluation failed: {e}"))?; - crate::openhuman::agent::triage::apply_decision(run.clone(), envelope) - .await - .map_err(|e| format!("apply_decision failed: {e}"))?; + match outcome { + crate::openhuman::agent::triage::TriageOutcome::Decision(run) => { + crate::openhuman::agent::triage::apply_decision(run.clone(), envelope) + .await + .map_err(|e| format!("apply_decision failed: {e}"))?; - Ok(format!( - "Triage decision: {} (agent: {:?})", - run.decision.action.as_str(), - run.decision.target_agent - )) + Ok(format!( + "Triage decision: {} (agent: {:?})", + run.decision.action.as_str(), + run.decision.target_agent + )) + } + crate::openhuman::agent::triage::TriageOutcome::Deferred { + defer_until_ms, + reason, + } => Ok(format!("Triage deferred until {defer_until_ms}: {reason}")), + } } /// Build a base64-encoded JSON response body for an agent trigger result. diff --git a/src/openhuman/webhooks/ops.rs b/src/openhuman/webhooks/ops.rs index 782d69d95..f01a2ffc9 100644 --- a/src/openhuman/webhooks/ops.rs +++ b/src/openhuman/webhooks/ops.rs @@ -179,7 +179,7 @@ pub async fn trigger_agent( } }; - let run = tokio::time::timeout( + let outcome = tokio::time::timeout( std::time::Duration::from_secs(60), crate::openhuman::agent::triage::run_triage(&envelope), ) @@ -187,23 +187,40 @@ pub async fn trigger_agent( .map_err(|_| "triage timed out after 60s".to_string())? .map_err(|e| format!("triage failed: {e}"))?; - tokio::time::timeout( - std::time::Duration::from_secs(60), - crate::openhuman::agent::triage::apply_decision(run.clone(), &envelope), - ) - .await - .map_err(|_| "apply_decision timed out after 60s".to_string())? - .map_err(|e| format!("apply_decision failed: {e}"))?; + match outcome { + crate::openhuman::agent::triage::TriageOutcome::Decision(run) => { + tokio::time::timeout( + std::time::Duration::from_secs(60), + crate::openhuman::agent::triage::apply_decision(run.clone(), &envelope), + ) + .await + .map_err(|_| "apply_decision timed out after 60s".to_string())? + .map_err(|e| format!("apply_decision failed: {e}"))?; - Ok(RpcOutcome::single_log( - serde_json::json!({ - "decision": run.decision.action.as_str(), - "target_agent": run.decision.target_agent, - "prompt": run.decision.prompt, - "reason": run.decision.reason, - }), - format!("webhooks.trigger_agent completed for {source}/{caller_id}"), - )) + Ok(RpcOutcome::single_log( + serde_json::json!({ + "decision": run.decision.action.as_str(), + "target_agent": run.decision.target_agent, + "prompt": run.decision.prompt, + "reason": run.decision.reason, + "resolution_path": run.resolution_path.as_str(), + }), + format!("webhooks.trigger_agent completed for {source}/{caller_id}"), + )) + } + crate::openhuman::agent::triage::TriageOutcome::Deferred { + defer_until_ms, + reason, + } => Ok(RpcOutcome::single_log( + serde_json::json!({ + "decision": "deferred", + "resolution_path": "deferred", + "defer_until_ms": defer_until_ms, + "reason": reason, + }), + format!("webhooks.trigger_agent deferred for {source}/{caller_id}"), + )), + } } pub fn build_echo_response(request: &WebhookRequest) -> WebhookResponseData {