fix(agent-harness): repair #4249 agent_graph regressions caught by raw-coverage fixtures (#4372)

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 +

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<hooks::ToolCallRecord> {
     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()),

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<crate::openhuman::tinyagents::ToolPolicyEnforcement>,
 }
 
 /// 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<Tinyagen
         Some(AGENT_TURN_MAX_OUTPUT_TOKENS),
         // Context middlewares sourced from the session's ContextManager.
         graph.context_mw,
+        // Builder-configured tool policy enforcement (session chat path).
+        graph.tool_policy,
     )
     .await
 }

src/openhuman/agent/harness/subagent_runner/handoff.rs

@@ -56,7 +56,7 @@ pub(super) const HANDOFF_MAX_ENTRIES: usize = 8;
 /// the top of `run_typed_mode` and shared (via `Arc`) with both the inner
 /// tool-call loop (writes) and the `extract_from_result` tool (reads).
 #[derive(Default)]
-pub(super) struct ResultHandoffCache {
+pub(crate) struct ResultHandoffCache {
     inner: StdMutex<HandoffInner>,
 }
 

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)

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<crate::openhuman::agent::harness::subagent_runner::ResultHandoffCache>,
+    >,
 ) -> Result<(String, usize, AggregatedUsage, Option<String>, 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");

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,

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;

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 `<tool_call>` 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<dyn crate::openhuman::inference::provider::Provider> =
+        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 (`<tool_call>` 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<dyn crate::openhuman::inference::provider::Provider>,
+}
+
+impl TextModeProvider {
+    fn new(inner: Arc<dyn crate::openhuman::inference::provider::Provider>) -> 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<String> {
+        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<crate::openhuman::inference::provider::ChatResponse> {
+        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<u64> {
+        self.inner.effective_context_window(model).await
+    }
+
+    async fn warmup(&self) -> anyhow::Result<()> {
+        self.inner.warmup().await
+    }
+}

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<ContentBlock> {
+    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<String> {
+    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<serde_json::Value> {
+    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<Conversation
             Message::Assistant(a) => {
                 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),
                     });
                 }
             }

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<SuperContextConfig>,
+    /// 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<HandoffConfig>,
+}
+
+/// 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<crate::openhuman::agent::harness::subagent_runner::ResultHandoffCache>,
+    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<crate::openhuman::agent::harness::subagent_runner::ResultHandoffCache>,
+    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<usize> {
         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<Arc<Vec<Box<dyn Tool>>>>,
+}
+
+impl CliRpcOnlyMiddleware {
+    pub fn new(tool_sets: Vec<Arc<Vec<Box<dyn Tool>>>>) -> 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<MiddlewareToolOutcome> {
+        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 '<name>' <denied|requires approval> by policy '<policy>': <reason>"`
+/// wording the engine produced.
+pub struct ToolPolicyMiddleware {
+    policy: Arc<dyn crate::openhuman::agent::tool_policy::ToolPolicy>,
+    /// 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<Arc<Vec<Box<dyn Tool>>>>,
+    /// 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<String>,
+    session_id: String,
+    channel: String,
+    agent_definition_id: String,
+}
+
+impl ToolPolicyMiddleware {
+    pub fn new(
+        policy: Arc<dyn crate::openhuman::agent::tool_policy::ToolPolicy>,
+        session: crate::openhuman::agent_tool_policy::ToolPolicySession,
+        tool_sets: Vec<Arc<Vec<Box<dyn Tool>>>>,
+        visible_tool_names: HashSet<String>,
+        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<dyn Tool>> {
+        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<String> {
+        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<crate::openhuman::agent::tool_policy::GeneratedToolRuntimeContext> {
+        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<MiddlewareToolOutcome> {
+        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<Option<(String, usize)>>,
+    identical_threshold: usize,
+    halt_summary: super::HaltSummarySlot,
+    state: std::sync::Mutex<FailureState>,
+    /// 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<std::collections::HashMap<String, String>>,
+}
+
+#[derive(Default)]
+struct FailureState {
+    last_sig: Option<String>,
+    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<Vec<Box<dyn Tool>>> = 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"] {

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<dyn crate::openhuman::agent::tool_policy::ToolPolicy>,
+    /// 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<ToolCallOutcome>,
 }
 
+/// 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<std::sync::Mutex<Vec<ToolCallOutcome>>>;
+
+/// 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<std::sync::Mutex<Option<String>>>;
+
 /// 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<u32>,
     context_mw: TurnContextMiddleware,
+    tool_policy: Option<ToolPolicyEnforcement>,
 ) -> Result<TinyagentsTurnOutcome> {
     // `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,
     })
 }
 

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

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

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,

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(),
     }
 }
 

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()
     }

tests/agent_harness_leftovers_raw_coverage_e2e.rs

@@ -370,6 +370,7 @@ fn definition(max_result_chars: Option<usize>) -> AgentDefinition {
         delegate_name: None,
         agent_tier: AgentTier::Worker,
         source: DefinitionSource::Builtin,
+        graph: Default::default(),
     }
 }
 

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(),
     }
 }
 

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(),
     }
 }
 

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(),
     }
 }
 

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();

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();
 

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(),
     }
 }

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();
 

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();

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<String> 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")));

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");
 

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());

tests/tools_agent_credentials_state_raw_coverage_e2e.rs

@@ -415,6 +415,7 @@ fn agent_definition(id: &str, max_result_chars: Option<usize>) -> AgentDefinitio
         delegate_name: None,
         agent_tier: Default::default(),
         source: Default::default(),
+        graph: Default::default(),
     }
 }
 

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(),
     }
 }