diff --git a/src/openhuman/agent/harness/graph.rs b/src/openhuman/agent/harness/graph.rs index 72617733b..9d9895fe7 100644 --- a/src/openhuman/agent/harness/graph.rs +++ b/src/openhuman/agent/harness/graph.rs @@ -126,6 +126,8 @@ pub(crate) async fn run_channel_turn_via_graph( // Context middlewares: cache-align + default tool-result byte cap (the // channel path has no session `ContextManager` to source config from). crate::openhuman::tinyagents::TurnContextMiddleware::defaults(), + // Channel/CLI path carries its own gating; no session `.tool_policy()`. + None, ) .await?; // Append only this turn's typed suffix (assistant tool-calls + tool results + diff --git a/src/openhuman/agent/harness/session/turn/core.rs b/src/openhuman/agent/harness/session/turn/core.rs index 448f89c47..88bb24866 100644 --- a/src/openhuman/agent/harness/session/turn/core.rs +++ b/src/openhuman/agent/harness/session/turn/core.rs @@ -61,22 +61,30 @@ fn should_run_super_context( // that now owns the first-turn context-collection pass (#4249). /// Flatten the assistant tool calls a turn produced into [`ToolCallRecord`]s for -/// the deterministic cap checkpoint (it lists the tools that ran). Tool success -/// isn't tracked per call here, so each is recorded optimistically; the listing -/// is a human-readable fallback, not authoritative accounting. +/// post-turn hooks + the deterministic cap checkpoint. Per-call success + +/// sanitized output summary are recovered from the turn's captured +/// [`ToolCallOutcome`]s (correlated by provider call id), since the harness folds +/// a tool result into a `Message::tool` that drops its failure flag — matching the +/// engine's honest per-call accounting instead of recording every call as ok. fn tool_records_from_conversation( conversation: &[ConversationMessage], + tool_outcomes: &[crate::openhuman::tinyagents::ToolCallOutcome], ) -> Vec { let mut records = Vec::new(); for msg in conversation { if let ConversationMessage::AssistantToolCalls { tool_calls, .. } = msg { for call in tool_calls { + let outcome = tool_outcomes.iter().find(|o| o.call_id == call.id); + let success = outcome.map(|o| o.success).unwrap_or(true); + let output_summary = outcome + .map(|o| hooks::sanitize_tool_output(&o.content, &call.name, success)) + .unwrap_or_default(); records.push(hooks::ToolCallRecord { name: call.name.clone(), arguments: serde_json::from_str(&call.arguments) .unwrap_or(serde_json::Value::Null), - success: true, - output_summary: String::new(), + success, + output_summary, duration_ms: 0, }); } @@ -903,6 +911,9 @@ impl Agent { user_message: user_message.to_string(), } }), + // Progressive-disclosure handoff is a sub-agent (integrations_agent) + // concern; the top-level chat turn never sets it. + handoff: None, }; // Gather any sub-agent spend delegated during this turn (synchronous @@ -923,6 +934,15 @@ impl Agent { context_window, run_queue: self.run_queue.clone(), context_mw, + // Enforce the builder-configured tool policy at the tool + // boundary (the tinyagents path otherwise bypasses it). + tool_policy: Some(crate::openhuman::tinyagents::ToolPolicyEnforcement { + policy: self.tool_policy.clone(), + session: self.tool_policy_session.clone(), + session_id: self.event_session_id.clone(), + channel: self.event_channel().to_string(), + agent_definition_id: self.agent_definition_id.clone(), + }), }), ) .await; @@ -967,7 +987,7 @@ impl Agent { } let checkpoint = if summary.trim().is_empty() { super::super::turn_checkpoint::build_deterministic_checkpoint( - &tool_records_from_conversation(&outcome.conversation), + &tool_records_from_conversation(&outcome.conversation, &outcome.tool_outcomes), max_iterations, ) } else { @@ -1016,13 +1036,35 @@ impl Agent { ); let persisted = self.tool_dispatcher.to_provider_messages(&self.history); + // Carry the turn's provider (event channel) + effective model and usage + // into the persisted transcript meta. Passing `None` here dropped + // `provider`/`model` from every transcript (they are `TranscriptMeta` + // fields sourced from the turn usage) — parity with the legacy engine, + // which handed `self.last_turn_usage.as_ref()` to this call. + let turn_usage = crate::openhuman::agent::harness::session::transcript::TurnUsage { + provider: self.event_channel().to_string(), + // The model that actually ran this turn (a per-turn override can + // diverge from `self.model_name`); attribute usage to it. + model: effective_model.to_string(), + usage: crate::openhuman::agent::harness::session::transcript::MessageUsage { + input: input_tokens, + output: output_tokens, + cached_input: cached_input_tokens, + context_window: context_window.unwrap_or(0), + cost_usd: charged_amount_usd, + }, + ts: chrono::Utc::now().to_rfc3339(), + reasoning_content: None, + tool_calls: Vec::new(), + iteration: outcome.model_calls as u32, + }; self.persist_session_transcript( &persisted, input_tokens, output_tokens, cached_input_tokens, charged_amount_usd, - None, + Some(&turn_usage), ); // Charge this turn's usage against the thread's active goal (parity with @@ -1055,7 +1097,10 @@ impl Agent { let ctx = TurnContext { user_message: user_message.to_string(), assistant_response: reply.clone(), - tool_calls: tool_records_from_conversation(&outcome.conversation), + tool_calls: tool_records_from_conversation( + &outcome.conversation, + &outcome.tool_outcomes, + ), turn_duration_ms: turn_started.elapsed().as_millis() as u64, session_id: Some(self.event_session_id.clone()) .filter(|session_id| !session_id.trim().is_empty()), diff --git a/src/openhuman/agent/harness/session/turn/graph.rs b/src/openhuman/agent/harness/session/turn/graph.rs index 571f7c765..aab209048 100644 --- a/src/openhuman/agent/harness/session/turn/graph.rs +++ b/src/openhuman/agent/harness/session/turn/graph.rs @@ -69,6 +69,9 @@ pub(crate) struct ChatTurnGraph { /// openhuman context middlewares (cache-align, microcompact, tool-output /// budget + payload summarizer) sourced from the session's `ContextManager`. pub context_mw: TurnContextMiddleware, + /// The agent's builder-configured tool policy + session context, enforced at + /// the tool boundary. `None` when the session has no explicit policy. + pub tool_policy: Option, } /// Drive the chat turn graph: a thin wrapper over the shared tinyagents seam @@ -102,6 +105,8 @@ pub(crate) async fn run_chat_turn_graph(graph: ChatTurnGraph) -> Result, } diff --git a/src/openhuman/agent/harness/subagent_runner/mod.rs b/src/openhuman/agent/harness/subagent_runner/mod.rs index 3c3a7c423..bf0ff0c23 100644 --- a/src/openhuman/agent/harness/subagent_runner/mod.rs +++ b/src/openhuman/agent/harness/subagent_runner/mod.rs @@ -49,6 +49,12 @@ pub use types::{ // renderer. The other `tool_prep` helpers are used only inside this module. pub(crate) use tool_prep::build_text_mode_tool_instructions; +// Progressive-disclosure handoff: the tinyagents `HandoffMiddleware` intercepts +// oversized sub-agent tool results via `apply_handoff`, sharing the per-spawn +// `ResultHandoffCache` with the `extract_from_result` tool. +pub(crate) use handoff::ResultHandoffCache; +pub(crate) use ops::apply_handoff; + // `user_is_signed_in_to_composio` is the mode-aware "can the user call // composio at all?" probe added in Wave 2 (#1710). Re-exported here so // non-composio probe sites (registration gates, heartbeat telemetry) diff --git a/src/openhuman/agent/harness/subagent_runner/ops/graph.rs b/src/openhuman/agent/harness/subagent_runner/ops/graph.rs index 572c76cbc..940148077 100644 --- a/src/openhuman/agent/harness/subagent_runner/ops/graph.rs +++ b/src/openhuman/agent/harness/subagent_runner/ops/graph.rs @@ -61,6 +61,19 @@ pub(super) async fn run_subagent_via_graph( workspace_dir: std::path::PathBuf, max_output_tokens: u32, model_vision: bool, + // Transcript-persistence provenance: the resolved child transcript stem + // (`{parent_chain}__{child_session_key}`) and the provider label (the parent + // turn's event channel) so the child's raw transcript lands in `session_raw` + // with the right stem + provider/model meta — parity with the removed + // `SubagentObserver::persist_transcript`. + transcript_stem: &str, + provider_label: &str, + // Progressive-disclosure handoff cache (integrations_agent with a resolved + // toolkit); `Some` installs the `HandoffMiddleware` that stashes oversized + // tool results and shares the cache with the `extract_from_result` tool. + handoff_cache: Option< + std::sync::Arc, + >, ) -> Result<(String, usize, AggregatedUsage, Option, bool), SubagentRunError> { tracing::info!( model, @@ -88,11 +101,14 @@ pub(super) async fn run_subagent_via_graph( history.clone() }; - // Child-progress attribution: when the parent carries an `on_progress` sink, - // mirror this sub-agent's iterations / tool calls / text + thinking deltas as - // `Subagent*` events scoped to (`agent_id`, `task_id`) so the parent thread - // can nest them under the live subagent row. - let subagent_scope = on_progress.as_ref().map(|_| SubagentScope { + // Child-progress attribution: mirror this sub-agent's iterations / tool calls + // / text + thinking deltas as `Subagent*` events scoped to (`agent_id`, + // `task_id`) so the parent thread can nest them under the live subagent row. + // Always set (not gated on `on_progress`): the scope also tells the shared + // seam this is a sub-agent turn, so the unknown-tool recovery uses the + // sub-agent wording. With no progress sink the scoped events simply have + // nowhere to go, which is harmless. + let subagent_scope = Some(SubagentScope { agent_id: agent_id.to_string(), task_id: task_id.to_string(), extended_policy, @@ -146,8 +162,22 @@ pub(super) async fn run_subagent_via_graph( // Bound the sub-agent's per-call output at its configured budget. Some(max_output_tokens), // Context middlewares: cache-align + default tool-result byte cap so a - // sub-agent's (often large) tool outputs stay bounded in its transcript. - crate::openhuman::tinyagents::TurnContextMiddleware::defaults(), + // sub-agent's (often large) tool outputs stay bounded in its transcript, + // plus the progressive-disclosure handoff when a cache is attached. + { + let mut mw = crate::openhuman::tinyagents::TurnContextMiddleware::defaults(); + if let Some(cache) = handoff_cache { + mw.handoff = Some(crate::openhuman::tinyagents::HandoffConfig { + cache, + agent_id: agent_id.to_string(), + task_id: task_id.to_string(), + }); + } + mw + }, + // Sub-agents gate via their own SubagentToolSource policy path, not the + // session `.tool_policy()`; no enforcement threaded here. + None, )) .await .map_err(SubagentRunError::Provider)?; @@ -208,6 +238,45 @@ pub(super) async fn run_subagent_via_graph( } } + // Persist the sub-agent's raw transcript to `session_raw` (parity with the + // removed `SubagentObserver::persist_transcript`). The graph runner replaced + // the observer but only mirrored to the worker thread, so per-child + // transcripts stopped being written — breaking downstream learning ingestion + // (`learning/transcript_ingest`, which reads `session_raw/*.jsonl`). + // On a cap-hit / early-exit, `outcome.text` is the checkpoint (or clarifying + // question) that stands in for a final assistant turn — append it so the + // persisted transcript reflects the actual final state, not the pre-checkpoint + // history. `history` already carries this turn's typed suffix. + let transcript_history; + let history_for_transcript: &[ChatMessage] = if (outcome.hit_cap + || outcome.early_exit_tool.is_some()) + && !outcome.text.trim().is_empty() + { + transcript_history = { + let mut messages = history.clone(); + messages.push(ChatMessage::assistant(outcome.text.clone())); + messages + }; + &transcript_history + } else { + history.as_slice() + }; + persist_subagent_transcript( + &workspace_dir, + transcript_stem, + agent_id, + task_id, + provider_label, + model, + history_for_transcript, + &usage, + context_window.unwrap_or(0), + // Match the dispatcher the history was actually serialized with (text-mode + // integrations turns write XML), and the real iteration count. + if native_tools { "native" } else { "xml" }, + outcome.model_calls as u32, + ); + // Mirror this turn's conversation to the spawn's worker thread (when one is // attached), matching the legacy `SubagentObserver`: assistant intents + // final answer as `agent` messages, tool results as `user` messages. The @@ -241,6 +310,79 @@ pub(super) async fn run_subagent_via_graph( )) } +/// Persist a sub-agent turn's raw transcript to `session_raw`, mirroring the +/// removed `SubagentObserver::persist_transcript`: `agent_type:"subagent"`, the +/// `task_id`, and the provider/model + usage carried on the last assistant +/// message so per-thread usage reads price the sub-agent at its own model. +#[allow(clippy::too_many_arguments)] +fn persist_subagent_transcript( + workspace_dir: &std::path::Path, + transcript_stem: &str, + agent_id: &str, + task_id: &str, + provider_label: &str, + model: &str, + history: &[ChatMessage], + usage: &AggregatedUsage, + context_window: u64, + dispatcher: &str, + iteration: u32, +) { + use crate::openhuman::agent::harness::session::transcript; + + let path = match transcript::resolve_keyed_transcript_path(workspace_dir, transcript_stem) { + Ok(p) => p, + Err(err) => { + tracing::debug!( + agent_id, + error = %err, + "[subagent_runner:graph] failed to resolve child transcript path" + ); + return; + } + }; + let now = chrono::Utc::now().to_rfc3339(); + let turn_usage = transcript::TurnUsage { + provider: provider_label.to_string(), + model: model.to_string(), + usage: transcript::MessageUsage { + input: usage.input_tokens, + output: usage.output_tokens, + cached_input: usage.cached_input_tokens, + context_window, + cost_usd: usage.charged_amount_usd, + }, + ts: now.clone(), + reasoning_content: None, + tool_calls: Vec::new(), + iteration, + }; + let meta = transcript::TranscriptMeta { + agent_name: agent_id.to_string(), + agent_id: Some(agent_id.to_string()), + agent_type: Some("subagent".to_string()), + dispatcher: dispatcher.into(), + provider: Some(turn_usage.provider.clone()), + model: Some(turn_usage.model.clone()), + created: now.clone(), + updated: now, + turn_count: 1, + input_tokens: usage.input_tokens, + output_tokens: usage.output_tokens, + cached_input_tokens: usage.cached_input_tokens, + charged_amount_usd: usage.charged_amount_usd, + thread_id: crate::openhuman::inference::provider::thread_context::current_thread_id(), + task_id: Some(task_id.to_string()), + }; + if let Err(err) = transcript::write_transcript(&path, history, &meta, Some(&turn_usage)) { + tracing::debug!( + agent_id, + error = %err, + "[subagent_runner:graph] failed to write child transcript" + ); + } +} + /// Mirror a sub-agent turn's structured conversation to its worker thread, /// matching the legacy [`SubagentObserver`]: assistant turns (intents + final) /// become `agent` messages, tool results become `user` messages. `extra_final`, @@ -439,6 +581,9 @@ mod tests { std::env::temp_dir(), 1024, false, + "root-session__real_tools", + "mock-channel", + None, ) .await .expect("graph subagent runs"); @@ -523,6 +668,9 @@ mod tests { std::env::temp_dir(), 1024, false, + "root-session__scoped_deltas", + "mock-channel", + None, ) .await .expect("child-delta subagent runs"); @@ -665,6 +813,9 @@ mod tests { std::env::temp_dir(), 1024, false, + "root-session__clarification", + "mock-channel", + None, ) .await .expect("ask-clarification subagent runs"); @@ -767,6 +918,9 @@ mod tests { std::env::temp_dir(), 1024, false, + "root-session__cap_hit", + "mock-channel", + None, ) .await .expect("cap-hit subagent runs"); diff --git a/src/openhuman/agent/harness/subagent_runner/ops/handoff_helper.rs b/src/openhuman/agent/harness/subagent_runner/ops/handoff_helper.rs index a6fe89624..984b345a0 100644 --- a/src/openhuman/agent/harness/subagent_runner/ops/handoff_helper.rs +++ b/src/openhuman/agent/harness/subagent_runner/ops/handoff_helper.rs @@ -14,7 +14,7 @@ use crate::openhuman::agent::harness::subagent_runner::handoff::{ /// extractor tool, stash the raw payload and substitute a short placeholder the /// sub-agent can drill into with `extract_from_result`. Errors and /// already-extracted output pass through unchanged. -pub(super) fn apply_handoff( +pub(crate) fn apply_handoff( cache: &ResultHandoffCache, tool_name: &str, task_id: &str, diff --git a/src/openhuman/agent/harness/subagent_runner/ops/mod.rs b/src/openhuman/agent/harness/subagent_runner/ops/mod.rs index 5484f9c6c..7c6f154ec 100644 --- a/src/openhuman/agent/harness/subagent_runner/ops/mod.rs +++ b/src/openhuman/agent/harness/subagent_runner/ops/mod.rs @@ -22,6 +22,7 @@ mod checkpoint; mod graph; mod handoff_helper; +pub(crate) use handoff_helper::apply_handoff; mod prompt; mod provider; mod runner; diff --git a/src/openhuman/agent/harness/subagent_runner/ops/runner.rs b/src/openhuman/agent/harness/subagent_runner/ops/runner.rs index 0d1371116..09a28316e 100644 --- a/src/openhuman/agent/harness/subagent_runner/ops/runner.rs +++ b/src/openhuman/agent/harness/subagent_runner/ops/runner.rs @@ -18,7 +18,8 @@ use crate::openhuman::agent::harness::fork_context::{current_parent, ParentExecu use crate::openhuman::agent::harness::subagent_runner::extract_tool::ExtractFromResultTool; use crate::openhuman::agent::harness::subagent_runner::handoff::ResultHandoffCache; use crate::openhuman::agent::harness::subagent_runner::tool_prep::{ - filter_tool_indices, is_subagent_spawn_tool, load_prompt_source, top_k_for_toolkit, + build_text_mode_tool_instructions, filter_tool_indices, is_subagent_spawn_tool, + load_prompt_source, top_k_for_toolkit, }; use crate::openhuman::agent::harness::subagent_runner::types::{ SubagentMode, SubagentRunError, SubagentRunOptions, SubagentRunOutcome, @@ -782,6 +783,32 @@ async fn run_typed_mode( ] }; + // `integrations_agent` with a resolved toolkit runs in **text mode**: its + // large per-action Composio toolkit compiles into a provider grammar that + // blows the native tool-schema ceiling, so omit native tool advertisement and + // describe the tools in the system prompt as prose, parsing `` tags + // from the response (legacy `force_text_mode` parity — the tinyagents rewrite + // dropped it, so integrations turns advertised native schemas the backend then + // rejected). Wrapping the provider clears `native_tool_calling`, which makes + // the model adapter skip native advertisement and fall back to XML parsing. + let subagent_provider: Arc = + if is_integrations_agent_with_toolkit { + if let Some(sys) = history.iter_mut().find(|m| m.role == "system") { + sys.content.push_str("\n\n"); + sys.content + .push_str(&build_text_mode_tool_instructions(&filtered_specs)); + } + tracing::info!( + agent_id = %definition.id, + task_id = %task_id, + tool_count = filtered_specs.len(), + "[subagent_runner:text-mode] omitting native tool schemas; injected XML tool protocol into system prompt" + ); + Arc::new(TextModeProvider::new(subagent_provider)) + } else { + subagent_provider + }; + // ── Run the inner tool-call loop ─────────────────────────────────── // Resolve the sub-agent model's user-configured vision flag; defaults to // `false` when config can't be loaded. Combined with the provider capability @@ -807,7 +834,9 @@ async fn run_typed_mode( // `handoff_cache` — the integrations-agent progressive-disclosure seams — are // not yet re-expressed on the tinyagents path; they need a tool-result // interception middleware and are tracked as a follow-up (issue #4249, 1b). - let _ = (&lazy_resolver, &handoff_cache); + // `handoff_cache` is now threaded into the graph route below (progressive + // disclosure). `lazy_resolver` remains a follow-up (#4249 1b). + let _ = &lazy_resolver; // Per-agent turn graph (issue #4249): `Default` runs the shared sub-agent // graph; `Custom` hands the assembled turn to this agent's own graph runner // (declared in its `graph.rs::graph()`). Every built-in agent selects @@ -816,6 +845,45 @@ async fn run_typed_mode( use crate::openhuman::agent::harness::agent_graph::{ AgentGraph, AgentTurnRequest, AgentTurnUsage, }; + // Resolve the child transcript stem once — `{parent_chain}__{child_session_key}` + // — so the sub-agent's raw transcript lands in `session_raw` under a filename + // that chains the parent session (parity with the removed observer stem). + let child_session_key = { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default(); + let unix_ts = now.as_secs(); + let nanos = now.subsec_nanos(); + let sanitized: String = definition + .id + .chars() + .map(|c| { + if c.is_ascii_alphanumeric() || c == '_' || c == '-' { + c + } else { + '_' + } + }) + .collect(); + let task_suffix: String = task_id + .chars() + .filter(|c| c.is_ascii_alphanumeric() || *c == '_' || *c == '-') + .take(12) + .collect(); + if task_suffix.is_empty() { + format!("{unix_ts}_{nanos:09}_{sanitized}") + } else { + format!("{unix_ts}_{nanos:09}_{sanitized}_{task_suffix}") + } + }; + let transcript_stem = { + let parent_chain = match parent.session_parent_prefix.as_deref() { + Some(prefix) => format!("{}__{}", prefix, parent.session_key), + None => parent.session_key.clone(), + }; + format!("{parent_chain}__{child_session_key}") + }; + let (output, iterations, agg_usage, early_exit_tool, hit_cap) = match &definition.graph { AgentGraph::Default => { super::graph::run_subagent_via_graph( @@ -837,6 +905,15 @@ async fn run_typed_mode( parent.workspace_dir.clone(), max_output_tokens, model_vision, + &transcript_stem, + // Sub-agent turns record their provider label as the literal + // "subagent" (parity with the legacy observer's TurnObserver + // provenance), distinguishing delegated spend from the parent's + // own channel in per-thread usage reads. + "subagent", + // Progressive-disclosure handoff cache (shared with the + // extract_from_result tool registered above). + handoff_cache.clone(), ) .await? } @@ -992,3 +1069,65 @@ async fn run_typed_mode( usage, }) } + +/// A [`Provider`] decorator that reports **no native tool calling**, forcing the +/// tinyagents model adapter to omit native tool schemas and fall back to +/// prompt-guided (`` XML) parsing. Everything else delegates to the +/// inner provider. Used to run `integrations_agent` in text mode (its large +/// toolkit would otherwise blow the provider's native tool-grammar ceiling). +struct TextModeProvider { + inner: Arc, +} + +impl TextModeProvider { + fn new(inner: Arc) -> Self { + Self { inner } + } +} + +#[async_trait::async_trait] +impl crate::openhuman::inference::provider::Provider for TextModeProvider { + fn capabilities(&self) -> crate::openhuman::inference::provider::traits::ProviderCapabilities { + let mut caps = self.inner.capabilities(); + // The whole point: hide native tool calling so the adapter advertises none. + caps.native_tool_calling = false; + caps + } + + async fn chat_with_system( + &self, + system_prompt: Option<&str>, + message: &str, + model: &str, + temperature: f64, + ) -> anyhow::Result { + self.inner + .chat_with_system(system_prompt, message, model, temperature) + .await + } + + async fn chat( + &self, + request: crate::openhuman::inference::provider::ChatRequest<'_>, + model: &str, + temperature: f64, + ) -> anyhow::Result { + self.inner.chat(request, model, temperature).await + } + + fn supports_vision(&self) -> bool { + self.inner.supports_vision() + } + + fn supports_streaming(&self) -> bool { + self.inner.supports_streaming() + } + + async fn effective_context_window(&self, model: &str) -> Option { + self.inner.effective_context_window(model).await + } + + async fn warmup(&self) -> anyhow::Result<()> { + self.inner.warmup().await + } +} diff --git a/src/openhuman/tinyagents/convert.rs b/src/openhuman/tinyagents/convert.rs index a84e322a4..61ed9a371 100644 --- a/src/openhuman/tinyagents/convert.rs +++ b/src/openhuman/tinyagents/convert.rs @@ -23,6 +23,45 @@ use tinyagents::harness::tool::{ToolCall as TaToolCall, ToolSchema}; use crate::openhuman::inference::provider::{ChatMessage, ConversationMessage, ToolResultMessage}; use crate::openhuman::tools::ToolSpec; +/// Key under which a thinking model's `reasoning_content` is stashed on an +/// assistant [`Message`]'s content blocks (as a [`ContentBlock::ProviderExtension`]) +/// and echoed back through openhuman [`ChatMessage::extra_metadata`]. tinyagents' +/// `AssistantMessage` has no reasoning channel of its own, so we carry it here so +/// it survives the round-trip and can be replayed verbatim on the next turn +/// (thinking-mode providers reject a multi-turn request that drops it). +pub(super) const REASONING_EXT_KEY: &str = "reasoning_content"; + +/// Build the [`ContentBlock`] that stashes a response's `reasoning_content` on an +/// assistant message, if any. +pub(super) fn reasoning_content_block(reasoning: Option<&str>) -> Option { + let reasoning = reasoning?; + // Store verbatim (only gate on non-empty after a trim): thinking-mode + // providers validate the prior reasoning block byte-for-byte on a resumed + // multi-turn request, so trimming boundary whitespace could break replay. + (!reasoning.trim().is_empty()).then(|| { + ContentBlock::ProviderExtension(serde_json::json!({ REASONING_EXT_KEY: reasoning })) + }) +} + +/// Recover the stashed `reasoning_content` from an assistant message's content +/// blocks (see [`reasoning_content_block`]). +fn reasoning_from_content(content: &[ContentBlock]) -> Option { + content.iter().find_map(|block| match block { + ContentBlock::ProviderExtension(value) => value + .get(REASONING_EXT_KEY) + .and_then(serde_json::Value::as_str) + .map(str::to_string), + _ => None, + }) +} + +/// The `extra_metadata` an assistant [`ChatMessage`] should carry so a stashed +/// `reasoning_content` replays on the next provider request. +fn reasoning_extra_metadata(content: &[ContentBlock]) -> Option { + reasoning_from_content(content) + .map(|reasoning| serde_json::json!({ REASONING_EXT_KEY: reasoning })) +} + /// Convert one openhuman [`ChatMessage`] into a harness [`Message`]. /// /// Role strings map onto the typed arms. A seeded **native** tool round is @@ -42,17 +81,29 @@ pub(super) fn chat_message_to_message(msg: &ChatMessage) -> Message { content: vec![ContentBlock::Text(text)], }), "assistant" => { + // Restore any `reasoning_content` stashed on the persisted message so a + // multi-turn thinking-mode conversation replays it verbatim (see + // [`reasoning_content_block`]). + let reasoning = msg + .extra_metadata + .as_ref() + .and_then(|meta| meta.get(REASONING_EXT_KEY)) + .and_then(serde_json::Value::as_str); if let Some((inner, tool_calls)) = parse_native_assistant_envelope(&text) { + let mut content = vec![ContentBlock::Text(inner)]; + content.extend(reasoning_content_block(reasoning)); Message::Assistant(AssistantMessage { id: msg.id.clone(), - content: vec![ContentBlock::Text(inner)], + content, tool_calls, usage: None, }) } else { + let mut content = vec![ContentBlock::Text(text)]; + content.extend(reasoning_content_block(reasoning)); Message::Assistant(AssistantMessage { id: msg.id.clone(), - content: vec![ContentBlock::Text(text)], + content, tool_calls: Vec::new(), usage: None, }) @@ -140,7 +191,11 @@ pub(super) fn message_to_chat_message(msg: &Message) -> ChatMessage { match msg { Message::System(_) => ChatMessage::system(msg.text()), Message::User(_) => ChatMessage::user(msg.text()), - Message::Assistant(_) => ChatMessage::assistant(msg.text()), + Message::Assistant(a) => { + let mut cm = ChatMessage::assistant(msg.text()); + cm.extra_metadata = reasoning_extra_metadata(&a.content); + cm + } Message::Tool(t) => { let mut cm = ChatMessage::tool(msg.text()); cm.id = Some(t.tool_call_id.clone()); @@ -173,9 +228,15 @@ pub(super) fn message_to_native_chat_message(msg: &Message) -> ChatMessage { "content": msg.text(), "tool_calls": tool_calls, }); - ChatMessage::assistant(payload.to_string()) + let mut cm = ChatMessage::assistant(payload.to_string()); + cm.extra_metadata = reasoning_extra_metadata(&a.content); + cm + } + Message::Assistant(a) => { + let mut cm = ChatMessage::assistant(msg.text()); + cm.extra_metadata = reasoning_extra_metadata(&a.content); + cm } - Message::Assistant(_) => ChatMessage::assistant(msg.text()), Message::Tool(t) => { let payload = serde_json::json!({ "tool_call_id": t.tool_call_id, @@ -225,16 +286,16 @@ pub(super) fn messages_to_conversation(messages: &[Message]) -> Vec { flush(&mut out, &mut pending); if a.tool_calls.is_empty() { - out.push(ConversationMessage::Chat(ChatMessage::assistant( - msg.text(), - ))); + let mut chat = ChatMessage::assistant(msg.text()); + chat.extra_metadata = reasoning_extra_metadata(&a.content); + out.push(ConversationMessage::Chat(chat)); } else { let text = msg.text(); out.push(ConversationMessage::AssistantToolCalls { text: (!text.is_empty()).then_some(text), tool_calls: a.tool_calls.iter().map(ta_call_to_oh_call).collect(), - reasoning_content: None, - extra_metadata: None, + reasoning_content: reasoning_from_content(&a.content), + extra_metadata: reasoning_extra_metadata(&a.content), }); } } diff --git a/src/openhuman/tinyagents/middleware.rs b/src/openhuman/tinyagents/middleware.rs index 227b50e5a..a6ea26b9d 100644 --- a/src/openhuman/tinyagents/middleware.rs +++ b/src/openhuman/tinyagents/middleware.rs @@ -75,6 +75,20 @@ pub struct TurnContextMiddleware { /// non-chat path) skips it. Only the chat turn sets this — and only when its /// gate (`should_run_super_context`) passes. pub super_context: Option, + /// Progressive-disclosure handoff: when set (integrations_agent with a + /// resolved toolkit), oversized tool results are stashed in the shared + /// [`ResultHandoffCache`] and replaced with an `extract_from_result` drill-in + /// placeholder. `None` everywhere else. + pub handoff: Option, +} + +/// Config for the [`HandoffMiddleware`]: the per-spawn cache (shared with the +/// `extract_from_result` tool) plus the ids used in handoff log lines. +#[derive(Clone)] +pub struct HandoffConfig { + pub cache: Arc, + pub agent_id: String, + pub task_id: String, } /// Inputs the [`SuperContextMiddleware`] node needs to run its first-turn @@ -97,6 +111,7 @@ impl TurnContextMiddleware { microcompact_keep_recent: 0, autocompact_enabled: true, super_context: None, + handoff: None, } } @@ -133,6 +148,16 @@ impl TurnContextMiddleware { keep_recent: self.microcompact_keep_recent, })); } + // Handoff runs BEFORE the tool-output budget so an oversized payload is + // stashed + replaced with a short placeholder first; the byte cap would + // otherwise shrink it below the handoff threshold and defeat the drill-in. + if let Some(handoff) = self.handoff { + harness.push_middleware(Arc::new(HandoffMiddleware { + cache: handoff.cache, + agent_id: handoff.agent_id, + task_id: handoff.task_id, + })); + } if self.tool_result_budget_bytes > 0 || self.payload_summarizer.is_some() { harness.push_middleware(Arc::new(ToolOutputMiddleware { budget_bytes: self.tool_result_budget_bytes, @@ -143,6 +168,42 @@ impl TurnContextMiddleware { } } +/// `after_tool`: progressive-disclosure handoff (issue #4249 1b). An oversized +/// sub-agent tool result is stashed in the shared [`ResultHandoffCache`] and its +/// content replaced with a short placeholder naming a `result_id` the model can +/// drill into via `extract_from_result`. Restores the seam the legacy +/// `SubagentToolSource` ran on every tool result (via `apply_handoff`), which the +/// agent_graph rewrite dropped. Errors and `extract_from_result`'s own output +/// pass through unchanged (handled inside `apply_handoff`). +pub struct HandoffMiddleware { + cache: Arc, + agent_id: String, + task_id: String, +} + +#[async_trait] +impl Middleware<()> for HandoffMiddleware { + fn name(&self) -> &str { + "result_handoff" + } + + async fn after_tool( + &self, + _ctx: &mut RunContext<()>, + _state: &(), + result: &mut TaToolResult, + ) -> TaResult<()> { + result.content = crate::openhuman::agent::harness::subagent_runner::apply_handoff( + &self.cache, + &result.name, + &self.task_id, + &self.agent_id, + std::mem::take(&mut result.content), + ); + Ok(()) + } +} + /// `before_model` (first call only): "super context" — the graph node analogue /// of the harness-driven first-turn context collection that used to run /// imperatively in `session/turn/core.rs`. On the first model call it runs the @@ -354,16 +415,13 @@ struct ToolOutputMiddleware { } impl ToolOutputMiddleware { - /// Effective byte cap for `name`: the tool's own `max_result_size_chars` - /// when it declares one (treated as bytes — a conservative approximation), - /// else the shared fallback `budget_bytes`. - fn effective_budget(&self, name: &str) -> usize { + /// The tool's own declared character cap, if any. + fn tool_char_cap(&self, name: &str) -> Option { self.tool_sets .iter() .flat_map(|set| set.iter()) .find(|t| t.name() == name) .and_then(|t| t.max_result_size_chars()) - .unwrap_or(self.budget_bytes) } } @@ -397,12 +455,35 @@ impl Middleware<()> for ToolOutputMiddleware { } } - // 2. Hard byte cap backstop — truncate at a UTF-8 boundary with a marker. - // Honor the tool's own declared cap first, else the shared fallback. - let budget = self.effective_budget(&result.name); - if budget > 0 { + // 2. Per-tool **char** cap — a tool that declares `max_result_size_chars` + // caps its own output in characters, with the tool-cap marker the model + // was taught to read (legacy engine parity). Distinct from the generic + // byte budget below: the tool cap is the tool's own contract. + let tool_cap = self.tool_char_cap(&result.name); + if let Some(cap) = tool_cap { + let char_count = result.content.chars().count(); + if char_count > cap { + let truncated: String = result.content.chars().take(cap).collect(); + let dropped = char_count - cap; + tracing::debug!( + tool = %result.name, + cap, + char_count, + dropped, + "[tinyagents::mw] per-tool char cap applied" + ); + result.content = format!( + "{truncated}\n\n[truncated by tool cap: {dropped} more chars not shown]" + ); + } + } + + // 3. Shared byte-cap backstop — truncate at a UTF-8 boundary with a marker. + // Only for tools with no cap of their own (a capped tool already bounded + // itself above; stacking the two markers would double-truncate). + if tool_cap.is_none() && self.budget_bytes > 0 { let (capped, outcome) = - apply_tool_result_budget(std::mem::take(&mut result.content), budget); + apply_tool_result_budget(std::mem::take(&mut result.content), self.budget_bytes); if outcome.truncated { tracing::debug!( tool = %result.name, @@ -515,6 +596,278 @@ impl ToolMiddleware<()> for ApprovalSecurityMiddleware { } } +/// `wrap_tool`: refuse a tool whose scope is +/// [`ToolScope::CliRpcOnly`](crate::openhuman::tools::ToolScope) inside the +/// autonomous agent loop (issue #4249). The in-house engine ran this gate in +/// `engine::tools`; the tinyagents path dropped it, so a CLI/RPC-only tool +/// (e.g. phone calls) would execute from the model loop. Applies on every path +/// (channel, session, sub-agent) since the restriction is intrinsic to the tool, +/// not the session — installed unconditionally. +pub struct CliRpcOnlyMiddleware { + tool_sets: Vec>>>, +} + +impl CliRpcOnlyMiddleware { + pub fn new(tool_sets: Vec>>>) -> Self { + Self { tool_sets } + } + + fn is_cli_rpc_only(&self, name: &str) -> bool { + self.tool_sets + .iter() + .flat_map(|set| set.iter()) + .find(|t| t.name() == name) + .map(|t| t.scope() == crate::openhuman::tools::ToolScope::CliRpcOnly) + .unwrap_or(false) + } +} + +#[async_trait] +impl ToolMiddleware<()> for CliRpcOnlyMiddleware { + fn name(&self) -> &str { + "cli_rpc_only" + } + + async fn wrap_tool( + &self, + ctx: &mut RunContext<()>, + state: &(), + call: TaToolCall, + next: ToolHandler<'_, (), ()>, + ) -> TaResult { + if self.is_cli_rpc_only(&call.name) { + tracing::warn!( + tool = call.name.as_str(), + "[tinyagents::mw] tool scope is CliRpcOnly — denied in agent loop" + ); + let content = format!( + "Tool '{}' is only available via explicit CLI/RPC invocation, not in the autonomous agent loop.", + call.name + ); + return Ok(MiddlewareToolOutcome::Result(TaToolResult { + call_id: call.id, + name: call.name, + content: content.clone(), + raw: None, + error: Some(content), + elapsed_ms: 0, + })); + } + next.run(ctx, state, call).await + } +} + +/// `wrap_tool`: enforce the agent's builder-configured [`ToolPolicy`] at the tool +/// boundary (issue #4249). The in-house engine ran this check in +/// `agent_tool_exec` (`ctx.tool_policy.check(...)`); the tinyagents path bypassed +/// it, so a `.tool_policy()` deny/require-approval silently no-opped and the tool +/// executed anyway — a security regression. This middleware restores it: a +/// blocking decision short-circuits with a model-consumable result carrying the +/// same `"Tool '' by policy '': "` +/// wording the engine produced. +pub struct ToolPolicyMiddleware { + policy: Arc, + /// The session's channel-permission snapshot — enforces the per-channel deny + /// + per-call permission-level ceiling the engine ran in `agent_tool_exec`. + session: crate::openhuman::agent_tool_policy::ToolPolicySession, + /// Shared tool sets (same `Arc`s the runner registers) so a call's OpenHuman + /// `Tool` can be resolved for its generated-tool runtime context and its + /// per-call permission level. + tool_sets: Vec>>>, + /// The advertised (visible) tool-name whitelist. Non-empty = restricted; a + /// call outside it is "not available to this agent" (the engine's first gate). + /// A non-visible tool is never registered, so it reaches here rewritten onto + /// the recovery sentinel — its original name rides `requested_tool`. + visible_tool_names: HashSet, + session_id: String, + channel: String, + agent_definition_id: String, +} + +impl ToolPolicyMiddleware { + pub fn new( + policy: Arc, + session: crate::openhuman::agent_tool_policy::ToolPolicySession, + tool_sets: Vec>>>, + visible_tool_names: HashSet, + session_id: String, + channel: String, + agent_definition_id: String, + ) -> Self { + Self { + policy, + session, + tool_sets, + visible_tool_names, + session_id, + channel, + agent_definition_id, + } + } + + fn resolve_tool(&self, name: &str) -> Option<&Box> { + self.tool_sets + .iter() + .flat_map(|set| set.iter()) + .find(|t| t.name() == name) + } + + /// The channel-permission gate the engine ran before the builder policy: a + /// session-level deny, then a per-call permission-level ceiling check. Returns + /// the blocking message when the call must not execute. + fn channel_permission_block(&self, call: &TaToolCall) -> Option { + let decision = self.session.decision_for(&call.name); + if decision.is_denied() { + let required = decision + .required_permission + .map(|permission| permission.to_string()) + .unwrap_or_else(|| "unknown".to_string()); + return Some(format!( + "Tool '{}' blocked by tool policy: requires {}, channel '{}' allows {}", + call.name, required, self.channel, decision.allowed_permission + )); + } + let tool = self.resolve_tool(&call.name)?; + let call_required = tool.permission_level_with_args(&call.arguments); + if call_required > decision.allowed_permission { + return Some(format!( + "Tool '{}' action requires {} permission, channel '{}' allows {}", + call.name, call_required, self.channel, decision.allowed_permission + )); + } + None + } + + fn generated_context( + &self, + name: &str, + args: &serde_json::Value, + ) -> Option { + self.tool_sets + .iter() + .flat_map(|set| set.iter()) + .find(|t| t.name() == name) + .and_then(|t| t.generated_runtime_context(args)) + } +} + +#[async_trait] +impl ToolMiddleware<()> for ToolPolicyMiddleware { + fn name(&self) -> &str { + "tool_policy" + } + + async fn wrap_tool( + &self, + ctx: &mut RunContext<()>, + state: &(), + call: TaToolCall, + next: ToolHandler<'_, (), ()>, + ) -> TaResult { + use crate::openhuman::agent::tool_policy::{ + ToolCallContext, ToolPolicyDecision, ToolPolicyRequest, + }; + + // A call the model made to a tool outside the visible set was registered + // nowhere, so it arrives rewritten onto the recovery sentinel with the + // original name on `requested_tool`. When the session restricts visibility + // (non-empty set) and that original tool isn't visible, the engine's first + // gate produced "not available to this agent" — reproduce it here (takes + // precedence over the generic "Unknown tool" sentinel wording). A genuinely + // unknown call with no visibility restriction still falls through to the + // sentinel's "Unknown tool" result. + if call.name == UNKNOWN_TOOL_SENTINEL { + if !self.visible_tool_names.is_empty() { + let requested = call + .arguments + .get("requested_tool") + .and_then(|v| v.as_str()) + .unwrap_or_default(); + if !requested.is_empty() && !self.visible_tool_names.contains(requested) { + let content = format!("Tool '{requested}' is not available to this agent"); + return Ok(MiddlewareToolOutcome::Result(TaToolResult { + call_id: call.id, + name: call.name, + content: content.clone(), + raw: None, + error: Some(content), + elapsed_ms: 0, + })); + } + } + return next.run(ctx, state, call).await; + } + + // Channel-permission ceiling first (session deny + per-call permission + // level), mirroring the engine order in `agent_tool_exec`. + if let Some(message) = self.channel_permission_block(&call) { + tracing::debug!( + tool = call.name.as_str(), + channel = self.channel.as_str(), + "[tinyagents::mw] tool blocked by channel permission ceiling" + ); + return Ok(MiddlewareToolOutcome::Result(TaToolResult { + call_id: call.id, + name: call.name, + content: message.clone(), + raw: None, + error: Some(message), + elapsed_ms: 0, + })); + } + + let context = ToolCallContext::session( + self.session_id.clone(), + self.channel.clone(), + self.agent_definition_id.clone(), + call.id.clone(), + 1, + ); + let mut request = + ToolPolicyRequest::new(call.name.clone(), call.arguments.clone(), context); + if let Some(generated) = self.generated_context(&call.name, &call.arguments) { + request = request.with_generated_tool_context(generated); + } + + let decision = self.policy.check(&request).await; + if let Some(reason) = decision.blocking_reason() { + let blocked_action = match &decision { + ToolPolicyDecision::RequireApproval { .. } => "requires approval", + ToolPolicyDecision::Deny { .. } => "denied", + ToolPolicyDecision::Allow => "allowed", + }; + crate::openhuman::tool_registry::denials::record( + call.name.as_str(), + self.policy.name(), + blocked_action, + reason, + ); + tracing::debug!( + tool = call.name.as_str(), + policy = self.policy.name(), + action = blocked_action, + reason = %reason, + "[tinyagents::mw] tool blocked by policy" + ); + let content = format!( + "Tool '{}' {blocked_action} by policy '{}': {reason}", + call.name, + self.policy.name() + ); + return Ok(MiddlewareToolOutcome::Result(TaToolResult { + call_id: call.id, + name: call.name, + content: content.clone(), + raw: None, + error: Some(content), + elapsed_ms: 0, + })); + } + + next.run(ctx, state, call).await + } +} + /// `before_tool`: rewrite a call to an **unadvertised** tool onto the recovery /// sentinel (issue #4249, Phase 1 Task B) so a hallucinated tool name is a /// recoverable [`UnknownToolAdapter`](super::tools::UnknownToolAdapter) result @@ -564,6 +917,79 @@ impl Middleware<()> for UnknownToolRewriteMiddleware { } } +/// `after_tool`: capture each tool call's execution outcome (success + content) +/// into a shared sink before the harness folds the result into a `Message::tool` +/// that drops the `error` flag (issue #4249). Without this, a post-turn +/// `ToolCallRecord` could only report every call as an optimistic success — the +/// in-house engine tracked real per-call success. Runs last in the `after_tool` +/// chain so it records the final (summarized/capped) content the transcript keeps. +pub struct ToolOutcomeCaptureMiddleware { + sink: super::ToolOutcomeSink, +} + +impl ToolOutcomeCaptureMiddleware { + pub fn new(sink: super::ToolOutcomeSink) -> Self { + Self { sink } + } +} + +#[async_trait] +impl Middleware<()> for ToolOutcomeCaptureMiddleware { + fn name(&self) -> &str { + "tool_outcome_capture" + } + + async fn after_tool( + &self, + _ctx: &mut RunContext<()>, + _state: &(), + result: &mut TaToolResult, + ) -> TaResult<()> { + if let Ok(mut sink) = self.sink.lock() { + sink.push(super::ToolCallOutcome { + call_id: result.call_id.clone(), + name: result.name.clone(), + success: result.error.is_none(), + content: result.content.clone(), + }); + } + Ok(()) + } +} + +/// `before_tool`: coerce a tool call's arguments to an empty object when they +/// are not a JSON object (issue #4249). A model can emit malformed native +/// arguments (invalid JSON, or a bare scalar/array); the model adapter parses +/// those to `Value::Null`, which the harness then rejects against an object +/// schema and aborts the whole turn. The in-house engine recovered such a call by +/// running the tool with `{}`; restore that so a single bad tool call is +/// recoverable rather than fatal. The recovery sentinel's own +/// `{ "requested_tool": … }` payload is already an object, so it is untouched. +pub struct ArgRecoveryMiddleware; + +#[async_trait] +impl Middleware<()> for ArgRecoveryMiddleware { + fn name(&self) -> &str { + "arg_recovery" + } + + async fn before_tool( + &self, + _ctx: &mut RunContext<()>, + _state: &(), + call: &mut TaToolCall, + ) -> TaResult<()> { + if !call.arguments.is_object() { + tracing::debug!( + tool = call.name.as_str(), + "[tinyagents::mw] recovering non-object tool arguments to {{}}" + ); + call.arguments = serde_json::json!({}); + } + Ok(()) + } +} + /// `before_model`: enforce OpenHuman's daily/monthly cost budgets **before** a /// model call spends (issue #4249, Phase 5). Reads the global /// [`CostTracker`](crate::openhuman::cost) and, when cost budgets are configured @@ -626,77 +1052,206 @@ impl Middleware<()> for CostBudgetMiddleware { } } -/// `after_tool`: stop the run when a tool returns the **same** error result -/// repeatedly (issue #4249). The legacy tool loop's progress guard halted on a -/// repeated deterministic failure — a security/approval denial, or a terminal -/// tool error the model keeps reissuing — so the run surfaced the root cause -/// instead of burning the whole iteration budget and then a generic cap failure. -/// The tinyagents path kept only the generic model/tool call caps, so this -/// reinstates the breaker as a graph middleware: after `threshold` consecutive -/// identical error signatures (`tool name` + error text), it pauses the run via -/// the shared steering handle (the same mechanism as the stop-hook / cap pausers). -/// Any successful tool result resets the counter — progress was made. +/// Consecutive **any**-failure no-progress backstop: different commands all +/// failing means the goal is unreachable here. Matches the legacy +/// `NO_PROGRESS_FAILURE_THRESHOLD`. +const NO_PROGRESS_FAILURE_THRESHOLD: usize = 6; +/// Consecutive **identical** hard-policy-rejection repeats before halting — a +/// blocked call re-issued unchanged can never succeed. Legacy +/// `HARD_REJECT_REPEAT_THRESHOLD`. +const HARD_REJECT_REPEAT_THRESHOLD: usize = 2; + +/// `after_tool`: stop the run when tool calls keep failing with no progress +/// (issue #4249). The legacy tool loop's progress guard surfaced a root-cause +/// halt summary — a security/approval denial re-issued unchanged, an identical +/// error retried, or *different* commands all failing — instead of burning the +/// whole iteration budget and ending on a generic cap error. The tinyagents path +/// kept only the model/tool call caps, so this reinstates the guard as a graph +/// middleware. Three halt conditions, checked per failure (any success resets +/// every counter — progress was made): +/// +/// 1. **Hard policy rejection** (`[policy-blocked]`) repeated `HARD_REJECT_REPEAT_THRESHOLD` +/// times with an identical signature — "blocked by the security policy … re-issued". +/// 2. **Identical** error signature repeated `identical_threshold` times — +/// "retried N times with identical arguments". +/// 3. **Any** failure `NO_PROGRESS_FAILURE_THRESHOLD` times in a row (even with +/// varied errors) — "N tool calls in a row failed". +/// +/// On trip it records a root-cause summary into the shared [`HaltSummarySlot`] +/// (the turn overrides its final text with it) and pauses the run via the shared +/// steering handle (same mechanism as the stop-hook / cap pausers). pub struct RepeatedToolFailureMiddleware { handle: SteeringHandle, - threshold: usize, - last: std::sync::Mutex>, + identical_threshold: usize, + halt_summary: super::HaltSummarySlot, + state: std::sync::Mutex, + /// call_id → argument fingerprint, captured in `before_tool` (the tool result + /// carries no arguments). Folded into the identical-repeat signature so the + /// "identical arguments" halt only trips on the *same* args — two different + /// argument sets that happen to share a first error line don't count as a + /// repeat and can't pre-empt the generic no-progress backstop. + arg_sigs: std::sync::Mutex>, +} + +#[derive(Default)] +struct FailureState { + last_sig: Option, + same_count: usize, + consecutive: usize, } impl RepeatedToolFailureMiddleware { - /// Build the breaker. `threshold` is clamped to at least 2 (a single failure - /// is never a loop). - pub fn new(handle: SteeringHandle, threshold: usize) -> Self { + /// Build the breaker. `identical_threshold` (the identical-signature retry + /// ceiling) is clamped to at least 2 — a single failure is never a loop. + pub fn new( + handle: SteeringHandle, + identical_threshold: usize, + halt_summary: super::HaltSummarySlot, + ) -> Self { Self { handle, - threshold: threshold.max(2), - last: std::sync::Mutex::new(None), + identical_threshold: identical_threshold.max(2), + halt_summary, + state: std::sync::Mutex::new(FailureState::default()), + arg_sigs: std::sync::Mutex::new(std::collections::HashMap::new()), } } } +/// A stable, bounded fingerprint of a tool call's arguments for the identical- +/// repeat signature (hashed so a huge payload doesn't bloat the map/comparison). +fn args_fingerprint(arguments: &serde_json::Value) -> String { + use std::hash::{Hash, Hasher}; + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + arguments.to_string().hash(&mut hasher); + format!("{:x}", hasher.finish()) +} + +/// Trim a tool error for inclusion in a halt summary (keep it bounded but retain +/// the deterministic leading detail the model/user needs). +fn truncate_for_halt(text: &str) -> String { + const MAX: usize = 600; + crate::openhuman::util::truncate_with_ellipsis(text, MAX) +} + #[async_trait] impl Middleware<()> for RepeatedToolFailureMiddleware { fn name(&self) -> &str { "repeated_tool_failure" } + async fn before_tool( + &self, + _ctx: &mut RunContext<()>, + _state: &(), + call: &mut TaToolCall, + ) -> TaResult<()> { + // The tool result carries no arguments, so capture a fingerprint here and + // correlate it by call_id in `after_tool`. + if let Ok(mut sigs) = self.arg_sigs.lock() { + sigs.insert(call.id.clone(), args_fingerprint(&call.arguments)); + } + Ok(()) + } + async fn after_tool( &self, _ctx: &mut RunContext<()>, _state: &(), result: &mut TaToolResult, ) -> TaResult<()> { - let mut guard = self.last.lock().unwrap(); + let mut state = self.state.lock().unwrap(); + let arg_fp = self + .arg_sigs + .lock() + .ok() + .and_then(|mut sigs| sigs.remove(&result.call_id)) + .unwrap_or_default(); let Some(err) = result.error.as_deref() else { - // Success → progress was made; reset the breaker. - *guard = None; + // Success → progress was made; reset every counter. + *state = FailureState::default(); return Ok(()); }; - // Signature: tool name + error text. Truncate the error so a huge payload - // doesn't dominate the comparison (the first line is the deterministic part). - let sig = format!("{}\u{1f}{}", result.name, err.lines().next().unwrap_or(err)); - let count = match guard.as_mut() { - Some((prev, c)) if *prev == sig => { - *c += 1; - *c + + // Signature: tool name + argument fingerprint + first error line (the + // deterministic parts; a huge payload tail must not dominate the + // identical-repeat comparison). Including the args means the "identical + // arguments" halt only fires when the args truly repeat. + let err_line = err.lines().next().unwrap_or(err); + let sig = format!("{}\u{1f}{arg_fp}\u{1f}{err_line}", result.name); + // The unknown-tool recovery is a failure the model can correct (it got the + // "unknown tool" feedback), so it must NOT feed the generic *any*-failure + // no-progress counter — else a turn that recovers from one bad tool name + // and then legitimately exhausts its iteration budget would trip the + // backstop instead of hitting the cap. It still feeds the *identical*-repeat + // counter, so re-issuing the SAME unavailable tool halts with a root cause. + if result.name != UNKNOWN_TOOL_SENTINEL { + state.consecutive += 1; + } + let same_count = match &state.last_sig { + Some(prev) if *prev == sig => { + state.same_count += 1; + state.same_count } _ => { - *guard = Some((sig, 1)); + state.last_sig = Some(sig); + state.same_count = 1; 1 } }; - if count >= self.threshold { + + // A hard policy rejection is marked in the tool output; it can never + // succeed when re-issued unchanged, so it trips faster. + let is_hard_reject = result + .content + .contains(crate::openhuman::security::POLICY_BLOCKED_MARKER) + || err.contains(crate::openhuman::security::POLICY_BLOCKED_MARKER); + + let summary = if is_hard_reject && same_count >= HARD_REJECT_REPEAT_THRESHOLD { + Some(format!( + "Stopping: the `{}` call is blocked by the security policy and was re-issued with \ + identical arguments — it can never succeed this way. Reason:\n{}\n\nDo not repeat \ + this call; use an allowed alternative or report that it can't be done here.", + result.name, + truncate_for_halt(err), + )) + } else if same_count >= self.identical_threshold { + Some(format!( + "Stopping: the `{}` call was retried {same_count} times with identical arguments \ + and kept failing — repeating it will not help. Last error:\n{}\n\nThis looks \ + unrecoverable in the current environment. Report this back instead of retrying.", + result.name, + truncate_for_halt(err), + )) + } else if state.consecutive >= NO_PROGRESS_FAILURE_THRESHOLD { + Some(format!( + "Stopping: {} tool calls in a row failed with no progress. Last error (from \ + `{}`):\n{}\n\nDifferent commands are all failing — the goal looks unreachable in \ + this environment. Report this back instead of retrying.", + state.consecutive, + result.name, + truncate_for_halt(err), + )) + } else { + None + }; + + if let Some(summary) = summary { tracing::warn!( tool = %result.name, - count, - threshold = self.threshold, - "[tinyagents::mw] repeated identical tool failure — pausing run so the root cause surfaces" + consecutive = state.consecutive, + same_count, + is_hard_reject, + "[tinyagents::mw] repeated tool failure — halting run so the root cause surfaces" ); + if let Ok(mut slot) = self.halt_summary.lock() { + *slot = Some(summary); + } // Pause at the top of the next iteration (before the next model call), // matching the stop-hook / cap pause path. Reset so a resumed run does - // not immediately re-pause on the same latched signature. + // not immediately re-pause on the same latched state. self.handle.send(SteeringCommand::Pause); - *guard = None; + *state = FailureState::default(); } Ok(()) } @@ -939,7 +1494,7 @@ mod tests { } #[test] - fn effective_budget_prefers_the_tools_own_cap() { + fn tool_char_cap_reads_the_tools_own_declared_cap() { let tools: Arc>> = Arc::new(vec![Box::new(FakeTool { name: "big", cap: Some(10), @@ -950,10 +1505,10 @@ mod tests { payload_summarizer: None, tool_sets: vec![tools], }; - // Tool declares its own cap → used instead of the flat fallback. - assert_eq!(mw.effective_budget("big"), 10); - // Unknown tool → the flat fallback. - assert_eq!(mw.effective_budget("other"), 1_000); + // Tool declares its own char cap → surfaced for the per-tool truncation. + assert_eq!(mw.tool_char_cap("big"), Some(10)); + // Unknown tool → no per-tool cap (the flat byte budget applies instead). + assert_eq!(mw.tool_char_cap("other"), None); } #[tokio::test] @@ -971,8 +1526,10 @@ mod tests { let mut result = tool_result("capped", &"y".repeat(500)); mw.after_tool(&mut ctx(), &(), &mut result).await.unwrap(); assert!( - result.content.contains("truncated by tool_result_budget"), - "the tool's own 20-byte cap should truncate: {}", + result + .content + .contains("truncated by tool cap: 480 more chars not shown"), + "the tool's own 20-char cap should truncate with the tool-cap marker: {}", result.content ); } @@ -1042,7 +1599,11 @@ mod tests { #[tokio::test] async fn repeated_tool_failure_pauses_only_after_the_threshold() { let handle = SteeringHandle::allow_all(); - let mw = RepeatedToolFailureMiddleware::new(handle.clone(), 3); + let mw = RepeatedToolFailureMiddleware::new( + handle.clone(), + 3, + std::sync::Arc::new(std::sync::Mutex::new(None)), + ); // Two identical failures: below the threshold, no pause. for _ in 0..2 { let mut r = failing_result("flaky", "boom"); @@ -1061,7 +1622,11 @@ mod tests { #[tokio::test] async fn repeated_tool_failure_resets_on_a_success() { let handle = SteeringHandle::allow_all(); - let mw = RepeatedToolFailureMiddleware::new(handle.clone(), 3); + let mw = RepeatedToolFailureMiddleware::new( + handle.clone(), + 3, + std::sync::Arc::new(std::sync::Mutex::new(None)), + ); // Two failures, then a success clears the counter. for _ in 0..2 { let mut r = failing_result("t", "boom"); @@ -1080,7 +1645,11 @@ mod tests { #[tokio::test] async fn repeated_tool_failure_ignores_distinct_errors() { let handle = SteeringHandle::allow_all(); - let mw = RepeatedToolFailureMiddleware::new(handle.clone(), 3); + let mw = RepeatedToolFailureMiddleware::new( + handle.clone(), + 3, + std::sync::Arc::new(std::sync::Mutex::new(None)), + ); // Three *different* errors never trip the breaker — only an identical, // deterministic failure loop does. for err in ["e1", "e2", "e3"] { diff --git a/src/openhuman/tinyagents/mod.rs b/src/openhuman/tinyagents/mod.rs index 188870f53..a41937ff8 100644 --- a/src/openhuman/tinyagents/mod.rs +++ b/src/openhuman/tinyagents/mod.rs @@ -46,7 +46,7 @@ use crate::openhuman::agent::progress::AgentProgress; use crate::openhuman::inference::provider::{ChatMessage, ConversationMessage, Provider}; pub use checkpoint::SqlRunLedgerCheckpointer; -pub use middleware::{SuperContextConfig, TurnContextMiddleware}; +pub use middleware::{HandoffConfig, SuperContextConfig, TurnContextMiddleware}; pub use model::{ProviderModel, ThinkingForwarder}; pub use observability::{CapPauser, IterationCursor, OpenhumanEventBridge, SubagentScope}; pub use tools::{ @@ -55,8 +55,25 @@ pub use tools::{ }; use std::collections::HashSet; +use std::sync::Arc as StdArc; use tokio::sync::mpsc::Sender; +/// The builder-configured [`ToolPolicy`](crate::openhuman::agent::tool_policy::ToolPolicy) +/// plus the session context a policy check needs, handed to the shared turn seam +/// so it can install the [`ToolPolicyMiddleware`](middleware::ToolPolicyMiddleware). +/// `None` means "no policy enforcement on this turn" (the channel/CLI + sub-agent +/// paths, which carry their own gating). +pub struct ToolPolicyEnforcement { + pub policy: StdArc, + /// The session's channel-permission snapshot — enforces the per-channel + /// permission ceiling (deny + per-call permission-level gate) the in-house + /// engine ran in `agent_tool_exec`. + pub session: crate::openhuman::agent_tool_policy::ToolPolicySession, + pub session_id: String, + pub channel: String, + pub agent_definition_id: String, +} + /// Drain the run queue's pending steer messages and forward them to the /// tinyagents [`SteeringHandle`] as injected user turns (the harness applies /// them to the working transcript at the next iteration checkpoint). This is the @@ -70,6 +87,21 @@ async fn forward_steers(queue: &RunQueue, handle: &SteeringHandle) { } } +/// Forward any queued **collect** messages (orchestrator/monitor lines enqueued +/// via `QueueMode::Collect`) into the run as injected user turns so they reach the +/// next LLM call as additional context. The in-house loop drained these each +/// iteration (`drain_collects`); the tinyagents rewrite wired only `forward_steers` +/// (issue #4249), so monitor lines never reached the model. Mirrors the legacy +/// `[Additional context from user]:` framing the model was taught to read. +async fn forward_collects(queue: &RunQueue, handle: &SteeringHandle) { + for msg in queue.drain_collects().await { + handle.send(SteeringCommand::InjectMessage(TaMessage::user(format!( + "[Additional context from user]: {}", + msg.text + )))); + } +} + /// Build the harness [`RunPolicy`] for an openhuman turn. /// /// The loop enforces limits from `self.policy.limits` (not the per-run @@ -144,8 +176,36 @@ pub struct TinyagentsTurnOutcome { /// should summarize a resumable checkpoint rather than treat `text` as a /// final answer — the tinyagents analogue of `CheckpointStrategy::on_max_iter`. pub hit_cap: bool, + /// Per-tool-call execution outcomes (success + raw result content), keyed by + /// provider call id, captured at the tool boundary. The harness folds a tool + /// result into a `Message::tool` that drops its `error` flag, so this is the + /// only place the caller can recover whether each call actually failed — used + /// to build honest `ToolCallRecord`s for post-turn hooks + the cap checkpoint. + pub tool_outcomes: Vec, } +/// One tool call's execution outcome, captured at the tool boundary before the +/// harness discards the failure flag. `success` mirrors the absence of a +/// `TaToolResult::error`; `content` is the (possibly summarized/capped) result +/// text used to derive a sanitized post-turn summary. +#[derive(Debug, Clone)] +pub struct ToolCallOutcome { + pub call_id: String, + pub name: String, + pub success: bool, + pub content: String, +} + +/// Shared sink the [`ToolOutcomeCaptureMiddleware`](middleware::ToolOutcomeCaptureMiddleware) +/// appends each tool call's outcome to, drained into the turn outcome. +pub type ToolOutcomeSink = std::sync::Arc>>; + +/// Shared slot the repeated-failure breaker writes a root-cause halt summary into +/// when it trips. The turn overrides its final text with this summary so the +/// no-progress halt surfaces the cause instead of an empty/last-model reply +/// (legacy `RepeatFailureGuard` parity). +pub type HaltSummarySlot = std::sync::Arc>>; + /// Drive an agent turn through the `tinyagents` agent-loop harness. /// /// Registers `provider` as the default model and every entry in `resolved_tools` @@ -222,6 +282,8 @@ pub async fn run_turn_via_tinyagents( ), early_exit_tool: None, hit_cap: false, + // This thin variant carries no per-call outcome capture middleware. + tool_outcomes: Vec::new(), }) } @@ -268,6 +330,7 @@ pub async fn run_turn_via_tinyagents_shared( pause_at_cap: bool, max_output_tokens: Option, context_mw: TurnContextMiddleware, + tool_policy: Option, ) -> Result { // `0` means "unset" → the legacy default (a native-bus / test convention); // otherwise the harness model-call cap would be zero and abort the run before @@ -391,10 +454,12 @@ pub async fn run_turn_via_tinyagents_shared( // error `REPEATED_TOOL_FAILURE_THRESHOLD` times in a row, so a deterministic // security/approval denial or terminal tool error surfaces its root cause // instead of burning the whole iteration budget (legacy ProgressGuard parity). + let halt_summary: HaltSummarySlot = std::sync::Arc::new(std::sync::Mutex::new(None)); if let Some(handle) = &handle { harness.push_middleware(Arc::new(middleware::RepeatedToolFailureMiddleware::new( handle.clone(), REPEATED_TOOL_FAILURE_THRESHOLD, + halt_summary.clone(), ))); } @@ -454,6 +519,38 @@ pub async fn run_turn_via_tinyagents_shared( tool_sets.clone(), ))); + // CLI/RPC-only scope gate — a tool restricted to explicit CLI/RPC invocation + // must not run from the model loop. Intrinsic to the tool, so installed on + // every path (channel/session/sub-agent). + harness.push_tool_middleware(Arc::new(middleware::CliRpcOnlyMiddleware::new( + tool_sets.clone(), + ))); + + // Capture each tool call's real success + content before the harness folds the + // result into a `Message::tool` that drops the failure flag, so the turn can + // build honest per-call `ToolCallRecord`s (post-turn hooks + cap checkpoint). + let tool_outcome_sink: ToolOutcomeSink = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + harness.push_middleware(Arc::new(middleware::ToolOutcomeCaptureMiddleware::new( + tool_outcome_sink.clone(), + ))); + + // Builder-configured tool policy (`.tool_policy()`), enforced at the tool + // boundary. The in-house engine ran this in `agent_tool_exec`; the tinyagents + // path bypassed it, so a deny/require-approval silently no-opped (security + // regression). Installed only when the caller threads an enforcement context + // (the session chat path); channel/CLI + sub-agent paths pass `None`. + if let Some(enforcement) = tool_policy { + harness.push_tool_middleware(Arc::new(middleware::ToolPolicyMiddleware::new( + enforcement.policy, + enforcement.session, + tool_sets.clone(), + allowed.clone(), + enforcement.session_id, + enforcement.channel, + enforcement.agent_definition_id, + ))); + } + // Unknown-tool recovery as a `before_tool` middleware (issue #4249, Phase 1 // Task B): a call to an unadvertised tool is rewritten onto the recovery // sentinel before the harness resolves it, so a hallucinated tool name is a @@ -463,6 +560,12 @@ pub async fn run_turn_via_tinyagents_shared( valid_tools, ))); + // Malformed-argument recovery (`before_tool`): coerce a call's non-object + // arguments (invalid JSON parses to Null) to `{}` so a single bad tool call is + // recoverable — the harness would otherwise reject it against an object schema + // and abort the whole turn. Engine parity. + harness.push_middleware(Arc::new(middleware::ArgRecoveryMiddleware)); + let config = RunConfig::new("agent_turn") .with_max_model_calls(max_iterations) .with_max_tool_calls(max_iterations.saturating_mul(8).max(8)); @@ -483,6 +586,14 @@ pub async fn run_turn_via_tinyagents_shared( let mut ctx = RunContext::new(config, ()); let streaming = on_progress.is_some(); + // Retain a clone of the progress sink so the turn can emit a terminal + // `TurnCompleted` after the run (the harness event stream the bridge mirrors + // has no run-completed event). Parent turns only — a sub-agent turn reports + // via its `Subagent*` events, not a top-level `TurnCompleted`. + let turn_completed_sink = subagent_scope + .is_none() + .then(|| on_progress.clone()) + .flatten(); // A sink is needed to mirror progress (bridge) or to observe model-call // completions for the cap pauser. let events = (on_progress.is_some() || pause_at_cap).then(EventSink::new); @@ -521,6 +632,7 @@ pub async fn run_turn_via_tinyagents_shared( let steering_forwarder = if let Some(handle) = handle { if let Some(queue) = run_queue.clone() { forward_steers(&queue, &handle).await; + forward_collects(&queue, &handle).await; } ctx = ctx.with_steering(handle.clone()); run_queue.map(|queue| { @@ -528,6 +640,7 @@ pub async fn run_turn_via_tinyagents_shared( loop { tokio::time::sleep(std::time::Duration::from_millis(50)).await; forward_steers(&queue, &handle).await; + forward_collects(&queue, &handle).await; } }) }) @@ -573,6 +686,15 @@ pub async fn run_turn_via_tinyagents_shared( return Err(anyhow::anyhow!("tinyagents harness run failed: {e}")); } }; + // Terminal turn event (parity with the legacy engine's `progress::emit`): the + // harness stream has no run-completed event, so emit `TurnCompleted` here with + // the model-call count as the iteration total. Parent turns only; best-effort. + if let Some(sink) = &turn_completed_sink { + let _ = sink.try_send(AgentProgress::TurnCompleted { + iterations: run.model_calls as u32, + }); + } + let bridge_totals = bridge.map(|bridge| bridge.totals_with_cost()); // Prefer the bridge's accumulated usage (per-call, authoritative — including @@ -601,16 +723,40 @@ pub async fn run_turn_via_tinyagents_shared( // cap hit. An early-exit is a clean pause and takes precedence; under // `pause_at_cap` the only other `Pause` source is the cap pauser, so this is // unambiguous. (`run_queue` steering injects messages, never pauses.) + // The repeated-failure breaker halts the run with a root-cause summary instead + // of a final model turn; surface it as the turn's text so the no-progress cause + // reaches the caller/user rather than an empty reply. + let breaker_halt = halt_summary.lock().ok().and_then(|mut s| s.take()); + + // Cap detection: the harness sets `final_response` only when the loop + // finishes naturally (the model stopped requesting tools). When the cap + // pauser stops the loop mid-work, `final_response` stays `None` — that's the + // cap hit. An early-exit is a clean pause and takes precedence; under + // `pause_at_cap` the only other `Pause` source is the cap pauser, so this is + // unambiguous. (`run_queue` steering injects messages, never pauses.) A + // breaker halt is *not* a cap hit: it already carries a root-cause summary, so + // treating it as a cap would let the caller (sub-agent runner) overwrite that + // summary with a generic checkpoint digest. let hit_cap = pause_at_cap && early_exit.is_none() + && breaker_halt.is_none() && run.model_calls >= max_iterations && run.final_response.is_none(); - let (early_exit_tool, text) = match early_exit { + let (early_exit_tool, mut text) = match early_exit { Some(exit) => (Some(exit.tool), exit.question), None => (None, run.text().unwrap_or_default()), }; + if let Some(summary) = breaker_halt { + text = summary; + } + + let tool_outcomes = tool_outcome_sink + .lock() + .map(|guard| guard.clone()) + .unwrap_or_default(); + Ok(TinyagentsTurnOutcome { text, history: convert::messages_to_history(&run.messages), @@ -625,6 +771,7 @@ pub async fn run_turn_via_tinyagents_shared( charged_amount_usd, early_exit_tool, hit_cap, + tool_outcomes, }) } diff --git a/src/openhuman/tinyagents/model.rs b/src/openhuman/tinyagents/model.rs index 6ee025433..67601f9b9 100644 --- a/src/openhuman/tinyagents/model.rs +++ b/src/openhuman/tinyagents/model.rs @@ -141,9 +141,14 @@ fn build_chat_inputs( } else { super::convert::messages_to_text_mode_chat(&request.messages) }; + // The unknown-tool sentinel is a recovery adapter, never a real capability — + // its contract (see `tools::UNKNOWN_TOOL_SENTINEL`) is that it is never + // advertised to the model. Filter it out of the advertised specs so it never + // leaks into the provider's tool list. let specs = request .tools .iter() + .filter(|s| s.name != super::tools::UNKNOWN_TOOL_SENTINEL) .map(|s| ToolSpec { name: s.name.clone(), description: s.description.clone(), @@ -202,6 +207,16 @@ fn response_to_model_response(response: &ChatResponse) -> ModelResponse { if !visible_text.is_empty() { content.push(ContentBlock::Text(visible_text)); } + // Thinking models return `reasoning_content` separately from the visible + // reply; tinyagents' `AssistantMessage` has no reasoning channel, so stash it + // on a provider-extension content block. It stays out of `Message::text()` + // (which only concatenates `Text` blocks) but survives into persistence and + // the next turn's request — where thinking-mode providers require it back. + if let Some(block) = + super::convert::reasoning_content_block(response.reasoning_content.as_deref()) + { + content.push(block); + } let usage = response.usage.as_ref().map(|u| { // Carry the provider's cached-prefix input count through the crate // `Usage` (it has a `cache_read_tokens` field) so downstream cost diff --git a/src/openhuman/tinyagents/tests.rs b/src/openhuman/tinyagents/tests.rs index 0a59a4a69..3c51204b6 100644 --- a/src/openhuman/tinyagents/tests.rs +++ b/src/openhuman/tinyagents/tests.rs @@ -174,6 +174,7 @@ async fn streaming_path_forwards_text_deltas_and_cost() { false, None, TurnContextMiddleware::defaults(), + None, ) .await .expect("streaming turn runs"); @@ -273,6 +274,7 @@ async fn pre_queued_steer_message_is_injected_into_the_request() { false, None, TurnContextMiddleware::defaults(), + None, ) .await .expect("steered turn runs"); @@ -367,6 +369,7 @@ async fn concurrent_shared_turns_each_get_a_distinct_result() { true, None, TurnContextMiddleware::defaults(), + None, ); let two = run_turn_via_tinyagents_shared( provider.clone(), @@ -384,6 +387,7 @@ async fn concurrent_shared_turns_each_get_a_distinct_result() { true, None, TurnContextMiddleware::defaults(), + None, ); let (a, b) = tokio::join!(one, two); diff --git a/src/openhuman/tinyagents/tools.rs b/src/openhuman/tinyagents/tools.rs index 131a3282b..6f1227702 100644 --- a/src/openhuman/tinyagents/tools.rs +++ b/src/openhuman/tinyagents/tools.rs @@ -72,16 +72,21 @@ impl Tool<()> for UnknownToolAdapter { ) } else { format!( - "Unknown tool: '{requested}'. It is not available; do not call it again. \ + "Unknown tool: {requested}. It is not available; do not call it again. \ Use one of the advertised tools, or answer directly." ) }; Ok(TaToolResult { call_id: call.id, name: call.name, + // Surface the recovery result as an error too: an unknown-tool call IS + // a failure, so the repeated-failure breaker counts it and halts the + // run with a root cause when the model keeps re-issuing the same + // unavailable tool (instead of looping to the iteration cap). The + // model-visible content is unchanged. + error: Some(content.clone()), content, raw: None, - error: None, elapsed_ms: 0, }) } @@ -246,7 +251,7 @@ async fn execute_openhuman_tool( TaToolResult { call_id: call.id, name: call.name.clone(), - content: format!("Error executing '{}': {e}", call.name), + content: format!("Error executing {}: {e}", call.name), raw: None, error: Some(e.to_string()), elapsed_ms: 0, diff --git a/tests/agent_archivist_debug_round21_raw_coverage_e2e.rs b/tests/agent_archivist_debug_round21_raw_coverage_e2e.rs index d70b8a46b..109959427 100644 --- a/tests/agent_archivist_debug_round21_raw_coverage_e2e.rs +++ b/tests/agent_archivist_debug_round21_raw_coverage_e2e.rs @@ -262,6 +262,7 @@ fn definition(max_iterations: usize) -> AgentDefinition { delegate_name: None, agent_tier: Default::default(), source: DefinitionSource::Builtin, + graph: Default::default(), } } diff --git a/tests/agent_harness_e2e.rs b/tests/agent_harness_e2e.rs index f75be57da..4605f037d 100644 --- a/tests/agent_harness_e2e.rs +++ b/tests/agent_harness_e2e.rs @@ -2485,7 +2485,6 @@ mod streaming_support { }) .auto_save(true) .explicit_preferences_enabled(false) - .unified_compaction_enabled(false) .build() .unwrap() } diff --git a/tests/agent_harness_leftovers_raw_coverage_e2e.rs b/tests/agent_harness_leftovers_raw_coverage_e2e.rs index ba639d6bd..3ee54d2ad 100644 --- a/tests/agent_harness_leftovers_raw_coverage_e2e.rs +++ b/tests/agent_harness_leftovers_raw_coverage_e2e.rs @@ -370,6 +370,7 @@ fn definition(max_result_chars: Option) -> AgentDefinition { delegate_name: None, agent_tier: AgentTier::Worker, source: DefinitionSource::Builtin, + graph: Default::default(), } } diff --git a/tests/agent_harness_raw_coverage_e2e.rs b/tests/agent_harness_raw_coverage_e2e.rs index 552223903..66f19cf8c 100644 --- a/tests/agent_harness_raw_coverage_e2e.rs +++ b/tests/agent_harness_raw_coverage_e2e.rs @@ -328,6 +328,7 @@ fn coverage_definition() -> AgentDefinition { delegate_name: None, agent_tier: Default::default(), source: Default::default(), + graph: Default::default(), } } diff --git a/tests/agent_large_round25_raw_coverage_e2e.rs b/tests/agent_large_round25_raw_coverage_e2e.rs index 58936274c..8a8f67608 100644 --- a/tests/agent_large_round25_raw_coverage_e2e.rs +++ b/tests/agent_large_round25_raw_coverage_e2e.rs @@ -302,6 +302,7 @@ fn integrations_definition() -> AgentDefinition { delegate_name: None, agent_tier: Default::default(), source: DefinitionSource::Builtin, + graph: Default::default(), } } diff --git a/tests/agent_prompts_subagent_raw_coverage_e2e.rs b/tests/agent_prompts_subagent_raw_coverage_e2e.rs index 6f985911f..d58e321a2 100644 --- a/tests/agent_prompts_subagent_raw_coverage_e2e.rs +++ b/tests/agent_prompts_subagent_raw_coverage_e2e.rs @@ -268,6 +268,7 @@ fn definition(prompt: PromptSource) -> AgentDefinition { delegate_name: None, agent_tier: Default::default(), source: DefinitionSource::Builtin, + graph: Default::default(), } } diff --git a/tests/agent_round26_raw_coverage_e2e.rs b/tests/agent_round26_raw_coverage_e2e.rs index 52c550c33..40ba10606 100644 --- a/tests/agent_round26_raw_coverage_e2e.rs +++ b/tests/agent_round26_raw_coverage_e2e.rs @@ -433,7 +433,6 @@ async fn builder_dedupes_visible_native_tools_and_seed_resume_bounds_history() - ..AgentConfig::default() }) .explicit_preferences_enabled(false) - .unified_compaction_enabled(false) .build()?; let original_key = agent.session_key().to_string(); diff --git a/tests/agent_session_round24_raw_coverage_e2e.rs b/tests/agent_session_round24_raw_coverage_e2e.rs index 1473c2a16..6e887e03a 100644 --- a/tests/agent_session_round24_raw_coverage_e2e.rs +++ b/tests/agent_session_round24_raw_coverage_e2e.rs @@ -444,7 +444,6 @@ async fn max_iteration_checkpoint_uses_deterministic_fallback_and_hooks() { }) .context_config(ContextConfig::default()) .explicit_preferences_enabled(false) - .unified_compaction_enabled(false) .build() .unwrap(); let (progress_tx, mut progress_rx) = tokio::sync::mpsc::channel(16); @@ -518,7 +517,6 @@ async fn builder_validation_and_system_prompt_cover_defaults_and_learning() { .explicit_preferences_enabled(true) .omit_profile(false) .omit_memory_md(false) - .unified_compaction_enabled(true) .build() .unwrap(); diff --git a/tests/agent_session_turn_raw_coverage_e2e.rs b/tests/agent_session_turn_raw_coverage_e2e.rs index 7eca68a1b..be87c85f9 100644 --- a/tests/agent_session_turn_raw_coverage_e2e.rs +++ b/tests/agent_session_turn_raw_coverage_e2e.rs @@ -548,7 +548,6 @@ fn agent_with( .context_config(context_config) .auto_save(true) .explicit_preferences_enabled(false) - .unified_compaction_enabled(false) .build() .unwrap() } @@ -786,7 +785,6 @@ async fn turn_xml_failures_checkpoint_policy_visibility_and_hooks_are_publicly_e })]) .tool_policy(Arc::new(DenyNamedPolicy("round17_ok"))) .explicit_preferences_enabled(false) - .unified_compaction_enabled(false) .build() .unwrap(); let mut visible = HashSet::new(); @@ -1027,5 +1025,6 @@ fn definition( delegate_name: None, agent_tier: AgentTier::Worker, source: DefinitionSource::Builtin, + graph: Default::default(), } } diff --git a/tests/agent_tool_loop_raw_coverage_e2e.rs b/tests/agent_tool_loop_raw_coverage_e2e.rs index d92b2b2e0..7aea7103a 100644 --- a/tests/agent_tool_loop_raw_coverage_e2e.rs +++ b/tests/agent_tool_loop_raw_coverage_e2e.rs @@ -580,7 +580,6 @@ async fn agent_builder_prompt_and_debug_dump_cover_public_session_paths() { .event_context("round15-session", "round15-channel") .omit_profile(false) .omit_memory_md(false) - .unified_compaction_enabled(false) .build() .unwrap(); diff --git a/tests/agent_turn_builder_leftovers_raw_coverage_e2e.rs b/tests/agent_turn_builder_leftovers_raw_coverage_e2e.rs index 92ecb16b3..3cc4971e5 100644 --- a/tests/agent_turn_builder_leftovers_raw_coverage_e2e.rs +++ b/tests/agent_turn_builder_leftovers_raw_coverage_e2e.rs @@ -397,7 +397,6 @@ async fn native_turn_dedups_duplicate_tool_specs_and_recovers_invalid_arguments( ..ContextConfig::default() }) .explicit_preferences_enabled(false) - .unified_compaction_enabled(false) .build() .unwrap(); @@ -457,7 +456,6 @@ async fn xml_turn_persists_tool_cycle_and_fires_failure_hook_context() { ..ContextConfig::default() }) .explicit_preferences_enabled(false) - .unified_compaction_enabled(true) .build() .unwrap(); @@ -541,7 +539,6 @@ async fn session_memory_threshold_path_runs_only_after_successful_turn() { ..ContextConfig::default() }) .explicit_preferences_enabled(false) - .unified_compaction_enabled(true) .build() .unwrap(); @@ -566,7 +563,6 @@ async fn session_memory_threshold_path_runs_only_after_successful_turn() { .event_context("round20-empty-session", "round20-empty-channel") .agent_definition_name("round20/empty") .explicit_preferences_enabled(false) - .unified_compaction_enabled(false) .build() .unwrap(); let err = failed_agent.run_single("return blank").await.unwrap_err(); diff --git a/tests/agent_turn_toolloop_round22_raw_coverage_e2e.rs b/tests/agent_turn_toolloop_round22_raw_coverage_e2e.rs index 0da3290ae..73e17ee71 100644 --- a/tests/agent_turn_toolloop_round22_raw_coverage_e2e.rs +++ b/tests/agent_turn_toolloop_round22_raw_coverage_e2e.rs @@ -321,19 +321,33 @@ async fn glm_style_tool_call_executes_then_final_streams_in_chunks_and_progress( assert!(response .text .starts_with("This is a deliberately long final response")); - let mut deltas = Vec::new(); + // The raw `on_delta` Sender path is retired (superseded by + // `on_progress` text deltas — see `agent/bus.rs`), so its channel stays empty; + // streaming is observed on `on_progress` below. + let mut on_delta_chunks = Vec::new(); while let Ok(delta) = delta_rx.try_recv() { - deltas.push(delta); + on_delta_chunks.push(delta); } assert!( - deltas.len() >= 2, - "long final response should stream in at least two chunks, got {deltas:?}" + on_delta_chunks.is_empty(), + "retired on_delta channel must stay empty, got {on_delta_chunks:?}" ); let mut progress = Vec::new(); while let Ok(event) = progress_rx.try_recv() { progress.push(event); } + // Streaming surfaces as `AgentProgress::TextDelta` on `on_progress`: each + // streamed model call forwards the provider's delta, so a two-iteration turn + // (tool round + final) emits at least two text deltas. + let text_deltas = progress + .iter() + .filter(|event| matches!(event, AgentProgress::TextDelta { .. })) + .count(); + assert!( + text_deltas >= 2, + "streaming should emit at least two on_progress text deltas, got {text_deltas}" + ); assert!(progress .iter() .any(|event| matches!(event, AgentProgress::TextDelta { delta, iteration: 1 } if delta == "draft from provider"))); diff --git a/tests/inference_agent_raw_coverage_e2e.rs b/tests/inference_agent_raw_coverage_e2e.rs index b0e659bae..490501208 100644 --- a/tests/inference_agent_raw_coverage_e2e.rs +++ b/tests/inference_agent_raw_coverage_e2e.rs @@ -1608,6 +1608,7 @@ named = ["todo", "plan_exit"] delegate_name: None, agent_tier: AgentTier::Worker, source: DefinitionSource::Builtin, + graph: Default::default(), }; assert_eq!(fallback_name.display_name(), "fallback_id"); diff --git a/tests/monitor_agent_e2e.rs b/tests/monitor_agent_e2e.rs index a5430536f..4203ea433 100644 --- a/tests/monitor_agent_e2e.rs +++ b/tests/monitor_agent_e2e.rs @@ -294,7 +294,6 @@ fn build_agent( .omit_profile(true) .omit_memory_md(true) .explicit_preferences_enabled(false) - .unified_compaction_enabled(false) .build() .unwrap(); agent.set_connected_integrations(Vec::new()); diff --git a/tests/tools_agent_credentials_state_raw_coverage_e2e.rs b/tests/tools_agent_credentials_state_raw_coverage_e2e.rs index 21227d83c..fe083baf5 100644 --- a/tests/tools_agent_credentials_state_raw_coverage_e2e.rs +++ b/tests/tools_agent_credentials_state_raw_coverage_e2e.rs @@ -415,6 +415,7 @@ fn agent_definition(id: &str, max_result_chars: Option) -> AgentDefinitio delegate_name: None, agent_tier: Default::default(), source: Default::default(), + graph: Default::default(), } } diff --git a/tests/tools_approval_channels_raw_coverage_e2e.rs b/tests/tools_approval_channels_raw_coverage_e2e.rs index 23daa4738..192962dfd 100644 --- a/tests/tools_approval_channels_raw_coverage_e2e.rs +++ b/tests/tools_approval_channels_raw_coverage_e2e.rs @@ -308,6 +308,7 @@ fn coverage_agent_definition( delegate_name: delegate_name.map(str::to_string), agent_tier: AgentTier::Worker, source: DefinitionSource::Builtin, + graph: Default::default(), } }