refactor(inference): Phase 5 — decouple per-turn state from ProviderModel (usage middleware, G2, build_turn_models) (#4625)

src/openhuman/agent/harness/graph.rs

@@ -106,10 +106,19 @@ pub(crate) async fn run_channel_turn_via_graph(
         context_window,
         "[channel:graph] routing channel turn through tinyagents harness"
     );
-    let outcome = run_turn_via_tinyagents_shared(
+    // Build the turn's crate `ChatModel` set from the resolved provider; the seam
+    // entry is crate-native (issue #4249, Phase 5).
+    let provider_id = provider.telemetry_provider_id();
+    let turn_models = crate::openhuman::tinyagents::build_turn_models(
         provider,
         model,
         temperature,
+        context_window,
+    );
+    let outcome = run_turn_via_tinyagents_shared(
+        turn_models,
+        provider_id,
+        model,
         prepared,
         vec![extra_arc, tools_registry],
         allowed,

src/openhuman/agent/harness/session/turn/graph.rs

@@ -89,10 +89,19 @@ pub(crate) async fn run_chat_turn_graph(graph: ChatTurnGraph) -> Result<Tinyagen
     } else {
         Some(graph.visible_tool_names)
     };
-    run_turn_via_tinyagents_shared(
+    // Build the turn's crate `ChatModel` set from the session provider; the seam
+    // entry is crate-native (issue #4249, Phase 5).
+    let provider_id = graph.provider.telemetry_provider_id();
+    let turn_models = crate::openhuman::tinyagents::build_turn_models(
         graph.provider,
         &graph.model,
         graph.temperature,
+        graph.context_window,
+    );
+    run_turn_via_tinyagents_shared(
+        turn_models,
+        provider_id,
+        &graph.model,
         graph.messages,
         vec![graph.tools],
         visible_tool_names,

src/openhuman/agent/harness/subagent_runner/ops/graph.rs

@@ -262,10 +262,19 @@ pub(super) async fn run_subagent_via_graph(
     // Capture native-tool support before `provider` is moved: the durable-history
     // append below serializes this turn's typed suffix with the matching dispatcher.
     let native_tools = provider.supports_native_tools();
-    let run_result = Box::pin(run_turn_via_tinyagents_shared(
+    // Build the child turn's crate `ChatModel` set from the resolved provider; the
+    // seam entry is crate-native (issue #4249, Phase 5).
+    let provider_id = provider.telemetry_provider_id();
+    let turn_models = crate::openhuman::tinyagents::build_turn_models(
         provider,
         model,
         temperature,
+        context_window,
+    );
+    let run_result = Box::pin(run_turn_via_tinyagents_shared(
+        turn_models,
+        provider_id,
+        model,
         dispatch_history,
         // Dynamic (per-spawn) tools first so a dynamic tool that intentionally
         // shadows a parent-registry tool of the same name is the one that

src/openhuman/agent_registry/agents/loader.rs

@@ -1063,6 +1063,7 @@ mod tests {
                     "get_flow_run",
                     "list_flow_connections",
                     "search_tool_catalog",
+                    "get_tool_contract",
                     "list_agent_profiles",
                     "dry_run_workflow",
                     "run_flow",

src/openhuman/tinyagents/mod.rs

@@ -69,7 +69,6 @@ use crate::openhuman::agent::harness::tool_result_artifacts::{
 use crate::openhuman::agent::harness::{run_queue::RunQueue, MAX_SPAWN_DEPTH};
 use crate::openhuman::agent::progress::AgentProgress;
 use crate::openhuman::inference::provider::{ChatMessage, ConversationMessage, Provider};
-use model::ThinkingForwarder;
 
 #[allow(unused_imports)] // Wired into the recall/retrieval facade in workstream 09.2.
 pub(crate) use embeddings::ProviderEmbeddingModel;
@@ -451,9 +450,9 @@ fn is_subagent_spawn_or_delegate_tool(name: &str) -> bool {
 
 #[allow(clippy::too_many_arguments)]
 pub(crate) async fn run_turn_via_tinyagents_shared(
-    provider: Arc<dyn Provider>,
+    turn_models: TurnModels,
+    provider_id: String,
     model: &str,
-    temperature: f64,
     history: Vec<ChatMessage>,
     tool_sets: Vec<Arc<Vec<Box<dyn crate::openhuman::tools::Tool>>>>,
     allowed: Option<HashSet<String>>,
@@ -483,10 +482,10 @@ pub(crate) async fn run_turn_via_tinyagents_shared(
     // otherwise the harness model-call cap would be zero and abort the run before
     // the first provider call.
     let max_iterations = effective_max_iterations(max_iterations);
-    // Snapshot the provider's telemetry id before `provider` moves into the
-    // harness assembly — the event bridge stamps it on every per-call
-    // generation event (`{provider_id}.{model}` in Langfuse).
-    let provider_id = provider.telemetry_provider_id();
+    // The turn's crate `ChatModel` set (`turn_models`) and the provider telemetry
+    // id are built by the caller via `build_turn_models` — the seam entry is
+    // crate-native and no longer names `Provider` (issue #4249, Phase 5). The
+    // telemetry id (`{provider_id}.{model}` in Langfuse) rides in as a param.
     let AssembledTurnHarness {
         harness,
         cursor,
@@ -505,9 +504,8 @@ pub(crate) async fn run_turn_via_tinyagents_shared(
         compression_mw,
         prompt_cache_guard,
     } = assemble_turn_harness(
-        provider,
+        turn_models,
         model,
-        temperature,
         tool_sets,
         allowed,
         max_iterations,
@@ -1113,6 +1111,69 @@ fn tinyagents_depth_error(
     }
 }
 
+/// The per-turn crate [`ChatModel`](tinyagents::harness::model::ChatModel) set,
+/// built once from an openhuman [`Provider`] by [`build_turn_models`] — the
+/// single place a turn's `ProviderModel`s are constructed (issue #4249, Phase 5).
+///
+/// [`assemble_turn_harness`] takes this bundle instead of the raw provider, so
+/// the harness assembly is expressed purely in crate model types; the
+/// `Provider` → `ChatModel` adaptation is confined to `build_turn_models`.
+pub(crate) struct TurnModels {
+    /// The turn's effective/primary model (registry default + dispatch target).
+    primary: Arc<dyn tinyagents::harness::model::ChatModel<()>>,
+    /// Additive workload-tier routes (registry name → model), excluding the
+    /// primary; the crate registry resolves fallback/selection across them.
+    routes: Vec<(String, Arc<dyn tinyagents::harness::model::ChatModel<()>>)>,
+    /// A model for the context-window summarizer (a distinct adapter instance so
+    /// its provider errors don't touch the turn's `error_slot`).
+    summarizer: Arc<dyn tinyagents::harness::model::ChatModel<()>>,
+    /// Recovers the primary's original (downcastable) provider error on failure.
+    error_slot: crate::openhuman::tinyagents::model::ProviderErrorSlot,
+}
+
+/// Build the per-turn [`TurnModels`] from an openhuman [`Provider`] — the sole
+/// `ProviderModel` construction site for a turn (issue #4249, Phase 5). The
+/// primary carries the model's context window on its capability profile; the
+/// workload-tier routes are projected via [`routes::build_route_models`]; the
+/// summarizer is a separate adapter over the same provider/model.
+pub(crate) fn build_turn_models(
+    provider: Arc<dyn Provider>,
+    model: &str,
+    temperature: f64,
+    context_window: Option<u64>,
+) -> TurnModels {
+    let summary_provider = provider.clone();
+    let mut primary = ProviderModel::new(provider, model, temperature);
+    // Record the model's context window on its capability profile (issue #4249,
+    // Phase 2) so the crate can validate input capacity before dispatch. The
+    // per-call output cap rides `RunConfig.max_turn_output_tokens` instead.
+    if let Some(window) = context_window.filter(|w| *w > 0) {
+        primary = primary.with_context_window(window);
+    }
+    let error_slot = primary.error_slot();
+    let primary: Arc<dyn tinyagents::harness::model::ChatModel<()>> = Arc::new(primary);
+
+    let routes = routes::build_route_models(&summary_provider, temperature, model)
+        .into_iter()
+        .map(|route| {
+            let model: Arc<dyn tinyagents::harness::model::ChatModel<()>> = route.model;
+            (route.name, model)
+        })
+        .collect();
+
+    // A distinct adapter instance for the summarizer (own error_slot), matching
+    // the pre-Phase-5 separate `summary_provider` clone.
+    let summarizer: Arc<dyn tinyagents::harness::model::ChatModel<()>> =
+        Arc::new(ProviderModel::new(summary_provider, model, temperature));
+
+    TurnModels {
+        primary,
+        routes,
+        summarizer,
+        error_slot,
+    }
+}
+
 /// Everything [`assemble_turn_harness`] wires up for one turn: the configured
 /// harness plus the shared slots/handles the run loop reads after the drive
 /// future returns.
@@ -1177,9 +1238,8 @@ struct AssembledTurnHarness {
 /// exposes the harness registries without driving a run.
 #[allow(clippy::too_many_arguments)]
 fn assemble_turn_harness(
-    provider: Arc<dyn Provider>,
+    turn_models: TurnModels,
     model: &str,
-    temperature: f64,
     tool_sets: Vec<Arc<Vec<Box<dyn crate::openhuman::tools::Tool>>>>,
     allowed: Option<HashSet<String>>,
     max_iterations: usize,
@@ -1228,61 +1288,47 @@ fn assemble_turn_harness(
     // tool-call start (the crate `ToolDelta` carries none), the bridge reads it
     // to label the argument fragments now streamed via `MessageDelta.tool_call`.
     let tool_names: ToolNameMap = Arc::default();
-    // Shared FIFO carry of per-call provider `UsageInfo`: the model adapter
-    // pushes each successful response's usage (charged USD + context window +
-    // cache-creation/reasoning tokens the crate `Usage` drops), the event bridge
-    // pops it when recording that call's usage (#4467, item 1).
+    // Shared FIFO carry of per-call provider `UsageInfo`: `UsageCarryMiddleware`
+    // pushes each response's usage (charged USD + context window +
+    // cache-creation/reasoning tokens, read off the response via G1), the event
+    // bridge pops it when recording that call's usage (#4467, item 1). The carry
+    // is produced by a wrap-model middleware now, not the adapter, so route models
+    // carry no usage side-channel (Phase 5).
     let provider_usage_carry: ProviderUsageCarry = Arc::default();
-    // Keep a provider handle for the context-window summarizer (the run consumes
-    // the other clone into the `ProviderModel`).
-    let summary_provider = provider.clone();
-    let mut provider_model = ProviderModel::new(provider, model, temperature)
-        .with_usage_carry(provider_usage_carry.clone());
-    // The per-call output cap now rides `RunConfig.max_turn_output_tokens`
-    // (Phase 5 groundwork), set by the caller: the loop stamps it onto every
-    // `ModelRequest` and the adapter honors `request.max_tokens`, so the cap no
-    // longer needs to be baked into the primary model or each route model.
-    // Record the model's context window on its capability profile (issue #4249,
-    // Phase 2) so the crate can validate input capacity before dispatch.
-    if let Some(window) = context_window.filter(|w| *w > 0) {
-        provider_model = provider_model.with_context_window(window);
-    }
-    if let Some(tx) = &on_progress {
-        provider_model = provider_model.with_thinking(ThinkingForwarder::new(
-            tx.clone(),
-            subagent_scope.clone(),
-            cursor.clone(),
-            tool_names.clone(),
-        ));
-    }
-    // Recover the original (downcastable) provider error if the run fails — the
-    // harness only carries a stringified copy.
-    let error_slot = provider_model.error_slot();
-    let provider_model = Arc::new(provider_model);
-    capability_registry.replace_model(model, provider_model.clone());
+    // The turn's models are pre-built by `build_turn_models` (the single
+    // `ProviderModel` construction site) and handed in as crate `ChatModel`s —
+    // the assembly no longer touches the raw provider (issue #4249, Phase 5).
+    let TurnModels {
+        primary,
+        routes,
+        summarizer: summarizer_model,
+        error_slot,
+    } = turn_models;
+    capability_registry.replace_model(model, primary.clone());
     harness
-        .register_model(model, provider_model)
+        .register_model(model, primary)
         .set_default_model(model);
 
     // Project the full workload-route set into the registry (issue #4249,
     // Workstream 02.1). Each route is an additive registry entry carrying its
     // per-route capability profile; `set_default_model` above keeps the turn's
     // effective model as the dispatch target, so behavior is preserved until
-    // fallback/selection (02.2) chooses among the routes. `summary_provider` is
-    // the retained provider handle (the other clone was consumed into the
-    // primary `ProviderModel`); `build_route_models` clones it per route and
-    // skips the turn's own model so we don't shadow the default.
-    for route in
-        routes::build_route_models(&summary_provider, temperature, model, &provider_usage_carry)
-    {
-        let routes::RouteModel {
-            name,
-            model: route_model,
-        } = route;
+    // fallback/selection (02.2) chooses among the routes. `build_turn_models`
+    // already skipped the turn's own model, so we don't shadow the default.
+    for (name, route_model) in routes {
         capability_registry.replace_model(name.as_str(), route_model.clone());
         harness.register_model(name, route_model);
     }
 
+    // Cost usage capture (issue #4249, Phase 5): feed the event bridge's usage
+    // carry from a wrap-model middleware that reads the full `UsageInfo` off each
+    // response, instead of every `ProviderModel` pushing it. Installed
+    // unconditionally — usage flows on every turn — and shares the same carry the
+    // bridge drains on `UsageRecorded`.
+    harness.push_model_middleware(Arc::new(routes::UsageCarryMiddleware::new(
+        provider_usage_carry.clone(),
+    )));
+
     // Per-call capability gate (issue #4249, Workstream 02.1): when the turn has
     // derivable capability needs (today: vision for a `vision-v1` turn), stamp
     // them onto every `ModelRequest` via `with_required_capabilities` so an unfit
@@ -1705,9 +1751,8 @@ fn assemble_turn_harness(
             // identical re-issued input slice must not re-run the summarizer LLM.
             let summarizer = summarize::FaultTolerantCachingSummarizer::new(
                 Box::new(summarize::ProviderModelSummarizer::new(
-                    summary_provider,
+                    summarizer_model,
                     model,
-                    temperature,
                 )),
                 &policy,
             );

src/openhuman/tinyagents/model.rs

@@ -7,7 +7,6 @@
 //! translate the [`ChatResponse`] back into a harness [`ModelResponse`] —
 //! carrying through text, native tool calls, and token usage.
 
-use std::sync::atomic::Ordering;
 use std::sync::{Arc, Mutex};
 
 use async_trait::async_trait;
@@ -17,11 +16,9 @@ use tinyagents::harness::model::{
 };
 use tinyagents::harness::tool::{ToolCall as TaToolCall, ToolDelta};
 use tinyagents::harness::usage::Usage;
-use tokio::sync::mpsc::{Sender, UnboundedSender};
+use tokio::sync::mpsc::UnboundedSender;
 
 use super::abort_guard::AbortOnDrop;
-use super::observability::{IterationCursor, ProviderUsageCarry, SubagentScope, ToolNameMap};
-use crate::openhuman::agent::progress::AgentProgress;
 use crate::openhuman::inference::provider::thread_context::{current_thread_id, with_thread_id};
 use crate::openhuman::inference::provider::{
     current_route_slot, with_route_slot, ChatMessage, ChatRequest, ChatResponse, Provider,
@@ -29,99 +26,6 @@ use crate::openhuman::inference::provider::{
 };
 use crate::openhuman::tools::ToolSpec;
 
-/// Out-of-band forwarder for progress events that do not yet round-trip through
-/// tinyagents with OpenHuman parity: non-streaming post-hoc reasoning and the
-/// tool-call **start** marker (tool name).
-///
-/// Streaming reasoning now rides tinyagents' native `MessageDelta.reasoning`
-/// channel, and the incremental tool-call **argument** fragments ride the native
-/// `MessageDelta.tool_call` channel (crate `ToolDelta`); both are projected by
-/// [`OpenhumanEventBridge`](super::OpenhumanEventBridge). What remains here is
-/// the split the crate can't express: the crate `ToolDelta` has only
-/// `call_id`/`content` (no `tool_name`), so the tool-call **start** event — the
-/// empty-delta `ToolCallArgsDelta` that carries the tool name and opens the UI
-/// timeline row — is still emitted straight onto the progress sink, and the
-/// learned `call_id → tool_name` map is *shared* with the bridge (via
-/// [`ToolNameMap`]) so it can label the argument fragments it now projects off
-/// the crate stream. This forwarder also still emits non-streaming post-hoc
-/// reasoning (see [`ProviderModel::invoke`]). It shares the bridge's
-/// [`IterationCursor`] so each event is attributed to the right model call.
-/// Parent runs emit the top-level variants; child runs emit the `Subagent`
-/// counterpart for thinking. Tool-arg/start events have no child variant, so
-/// they ride the top-level event.
-#[derive(Clone)]
-pub(super) struct ThinkingForwarder {
-    sink: Sender<AgentProgress>,
-    scope: Option<SubagentScope>,
-    cursor: IterationCursor,
-    /// call_id → tool_name, learned from `ToolCallStart`. Shared with the
-    /// [`OpenhumanEventBridge`](super::OpenhumanEventBridge) so the streamed
-    /// argument fragments (which ride the crate `ToolDelta`, sans name) can be
-    /// labelled with the tool the UI shows.
-    tool_names: ToolNameMap,
-}
-
-impl ThinkingForwarder {
-    pub(super) fn new(
-        sink: Sender<AgentProgress>,
-        scope: Option<SubagentScope>,
-        cursor: IterationCursor,
-        tool_names: ToolNameMap,
-    ) -> Self {
-        Self {
-            sink,
-            scope,
-            cursor,
-            tool_names,
-        }
-    }
-
-    /// Best-effort, non-blocking emit of one reasoning chunk (drops on a full
-    /// channel, matching the streaming text path).
-    fn emit(&self, delta: String) {
-        if delta.is_empty() {
-            return;
-        }
-        let iteration = self.cursor.load(Ordering::SeqCst);
-        let progress = match &self.scope {
-            None => AgentProgress::ThinkingDelta { delta, iteration },
-            Some(s) => AgentProgress::SubagentThinkingDelta {
-                agent_id: s.agent_id.clone(),
-                task_id: s.task_id.clone(),
-                delta,
-                iteration,
-            },
-        };
-        let _ = self.sink.try_send(progress);
-    }
-
-    /// Record the tool name a streaming tool call starts with (into the map
-    /// shared with the bridge, so it can label the argument fragments it
-    /// projects off the crate stream), and emit the start marker — an
-    /// empty-delta `ToolCallArgsDelta` — so consumers see the call begin before
-    /// its arguments arrive (matching the legacy `ProviderDelta::ToolCallStart`
-    /// mapping). The crate `ToolDelta` has no `tool_name` field, so this half of
-    /// the tool-arg contract can't ride the crate stream and stays here.
-    fn note_tool_call(&self, call_id: String, tool_name: String) {
-        self.tool_names
-            .lock()
-            .unwrap()
-            .insert(call_id.clone(), tool_name.clone());
-        tracing::trace!(
-            call_id = call_id.as_str(),
-            tool_name = tool_name.as_str(),
-            child = self.scope.is_some(),
-            "[stream] tool-call start marker (name recorded for crate-stream arg fragments)"
-        );
-        let _ = self.sink.try_send(AgentProgress::ToolCallArgsDelta {
-            call_id,
-            tool_name,
-            delta: String::new(),
-            iteration: self.cursor.load(Ordering::SeqCst),
-        });
-    }
-}
-
 /// Translate a harness [`ModelRequest`] into openhuman's message list + tool
 /// specs (shared by the buffered and streaming paths).
 fn build_chat_inputs(
@@ -346,19 +250,17 @@ pub(crate) fn usage_info_from_response(response: &ModelResponse) -> Option<Usage
 /// the [`OpenhumanEventBridge`](super::OpenhumanEventBridge) mirrors them as
 /// progress deltas from the crate stream alone): text/reasoning as
 /// [`MessageDelta`], and each argument fragment as
-/// [`ModelStreamItem::ToolCallDelta`] correlated by `call_id`. The crate
-/// `ToolDelta` has no `tool_name`, so the tool-call **start** marker (which
-/// carries the name and opens the UI timeline row) still rides the out-of-band
-/// [`ThinkingForwarder`]; it also records the name into the map shared with the
-/// bridge so the streamed fragments stay labelled. The model adapter still
-/// assembles the final native tool calls from the `Completed` response (the
-/// `StreamAccumulator` treats it as authoritative), so these fragments are
-/// progress-only — the UI can show the call being composed.
-fn forward_delta(
-    tx: &UnboundedSender<ModelStreamItem>,
-    thinking: Option<&ThinkingForwarder>,
-    delta: ProviderDelta,
-) {
+/// [`ModelStreamItem::ToolCallDelta`] correlated by `call_id`. The tool-call
+/// **start** marker now also rides the native stream: with the crate `ToolDelta`
+/// carrying an optional `tool_name` (G2), the call-opening delta is a
+/// `ToolCallDelta` with the name set and empty content, so the
+/// [`OpenhumanEventBridge`](super::OpenhumanEventBridge) records the name and
+/// opens the UI timeline row off the crate stream alone — no out-of-band
+/// forwarder. The model adapter still assembles the final native tool calls from
+/// the `Completed` response (the `StreamAccumulator` treats it as
+/// authoritative), so these fragments are progress-only — the UI can show the
+/// call being composed.
+fn forward_delta(tx: &UnboundedSender<ModelStreamItem>, delta: ProviderDelta) {
     match delta {
         ProviderDelta::TextDelta { delta } => {
             if !delta.is_empty() {
@@ -373,9 +275,19 @@ fn forward_delta(
             }
         }
         ProviderDelta::ToolCallStart { call_id, tool_name } => {
-            if let Some(forwarder) = thinking {
-                forwarder.note_tool_call(call_id, tool_name);
-            }
+            // Call-opening marker: name set, empty content. Rides the native
+            // crate stream (G2) so the bridge can label the call before its
+            // arguments arrive.
+            tracing::trace!(
+                call_id = call_id.as_str(),
+                tool_name = tool_name.as_str(),
+                "[stream] forwarding tool-call start onto crate ToolCallDelta"
+            );
+            let _ = tx.send(ModelStreamItem::ToolCallDelta(ToolDelta {
+                call_id,
+                content: String::new(),
+                tool_name: Some(tool_name),
+            }));
         }
         ProviderDelta::ToolCallArgsDelta { call_id, delta } => {
             if !delta.is_empty() {
@@ -387,6 +299,7 @@ fn forward_delta(
                 let _ = tx.send(ModelStreamItem::ToolCallDelta(ToolDelta {
                     call_id,
                     content: delta,
+                    tool_name: None,
                 }));
             }
         }
@@ -413,17 +326,8 @@ pub(super) struct ProviderModel {
     model: String,
     temperature: f64,
     max_tokens: Option<u32>,
-    /// When set, the adapter forwards tool-argument progress and post-hoc
-    /// non-streaming reasoning onto the progress sink.
-    thinking: Option<ThinkingForwarder>,
     /// Preserves the last original provider error for the runner to re-surface.
     error_slot: ProviderErrorSlot,
-    /// FIFO side-channel shared with the event bridge: on each successful chat
-    /// response the adapter pushes the provider `UsageInfo` (which carries the
-    /// backend-charged USD + context window + cache-creation/reasoning tokens
-    /// the crate `Usage` mapping drops), and the bridge pops it when recording
-    /// that call's usage — restoring charged-USD precedence (#4467, item 1).
-    usage_carry: ProviderUsageCarry,
     /// Capability profile derived from the wrapped provider (issue #4249,
     /// Phase 2): lets the crate validate a request against the model's actual
     /// capabilities (vision, tool calling, streaming, token limits) *before*
@@ -490,9 +394,7 @@ impl ProviderModel {
             model,
             temperature,
             max_tokens: None,
-            thinking: None,
             error_slot: Arc::new(Mutex::new(None)),
-            usage_carry: Arc::default(),
             profile,
         }
     }
@@ -503,14 +405,6 @@ impl ProviderModel {
         self.error_slot.clone()
     }
 
-    /// Attach the shared provider-usage carry the event bridge drains, so the
-    /// backend-charged USD + context window this adapter observes reach the cost
-    /// accounting (#4467, item 1). Clone the same handle into the bridge.
-    pub(super) fn with_usage_carry(mut self, carry: ProviderUsageCarry) -> Self {
-        self.usage_carry = carry;
-        self
-    }
-
     /// Cap the output tokens requested from the provider for every call.
     pub(super) fn with_max_tokens(mut self, max_tokens: u32) -> Self {
         self.max_tokens = Some(max_tokens);
@@ -547,13 +441,6 @@ impl ProviderModel {
         self.profile.reasoning = reasoning;
         self
     }
-
-    /// Forward provider thinking/tool-argument progress onto a progress sink via
-    /// `forwarder` (parent or sub-agent scoped). See [`ThinkingForwarder`].
-    pub(super) fn with_thinking(mut self, forwarder: ThinkingForwarder) -> Self {
-        self.thinking = Some(forwarder);
-        self
-    }
 }
 
 #[async_trait]
@@ -649,27 +536,15 @@ impl ChatModel<()> for ProviderModel {
                 });
             }
         };
-        // Non-streaming path: surface any reasoning the provider returned as a
-        // single post-hoc thinking delta (it had no per-token channel to ride).
-        if let Some(forwarder) = &self.thinking {
-            if let Some(reasoning) = response
-                .reasoning_content
-                .as_ref()
-                .filter(|r| !r.is_empty())
-            {
-                forwarder.emit(reasoning.clone());
-            }
-        }
-        // Push this call's provider usage onto the shared carry so the event
-        // bridge records charged USD / context window with provider precedence
-        // (#4467, item 1). One push per successful response, matching the single
-        // `UsageRecorded` the crate emits for this call.
-        if let Some(u) = &response.usage {
-            self.usage_carry
-                .lock()
-                .unwrap_or_else(|p| p.into_inner())
-                .push_back(u.clone());
-        }
+        // The buffered path is used only for unobserved turns (no progress sink):
+        // the seam sets `streaming = on_progress.is_some()`, so any post-hoc
+        // reasoning here would have nowhere to go. Observed turns take `stream()`,
+        // which forwards reasoning natively. Reasoning still rides the response as
+        // a typed thinking block (see `response_to_model_response`) for
+        // persistence/replay.
+        // Provider usage (charged USD / context window / cache-creation-reasoning)
+        // now reaches the event bridge via `UsageCarryMiddleware`, which reads it
+        // off the returned `ModelResponse` (G1) — the adapter no longer carries it.
         Ok(response_to_model_response(&response, &pformat_registry))
     }
 
@@ -695,12 +570,7 @@ impl ChatModel<()> for ProviderModel {
         let temperature = request.temperature.unwrap_or(self.temperature);
         // Same precedence for the output cap (see `invoke`).
         let max_tokens = request.max_tokens.or(self.max_tokens);
-        let thinking = self.thinking.clone();
         let error_slot = self.error_slot.clone();
-        // Captured for the spawned producer (task-locals/`self` do not cross the
-        // spawn): the streaming path pushes provider usage onto the same carry
-        // the buffered path uses, so charged USD reaches the bridge (#4467, item 1).
-        let usage_carry = self.usage_carry.clone();
 
         let (item_tx, item_rx) = tokio::sync::mpsc::unbounded_channel::<ModelStreamItem>();
 
@@ -752,7 +622,7 @@ impl ChatModel<()> for ProviderModel {
                     maybe = delta_rx.recv() => {
                         if let Some(delta) = maybe {
                             streamed_thinking |= matches!(delta, ProviderDelta::ThinkingDelta { .. });
-                            forward_delta(&item_tx, thinking.as_ref(), delta);
+                            forward_delta(&item_tx, delta);
                         }
                     }
                     res = &mut chat_fut => break res,
@@ -761,7 +631,7 @@ impl ChatModel<()> for ProviderModel {
             // Drain any deltas that landed before the call returned.
             while let Ok(delta) = delta_rx.try_recv() {
                 streamed_thinking |= matches!(delta, ProviderDelta::ThinkingDelta { .. });
-                forward_delta(&item_tx, thinking.as_ref(), delta);
+                forward_delta(&item_tx, delta);
             }
 
             let terminal = match response {
@@ -789,15 +659,9 @@ impl ChatModel<()> for ProviderModel {
                             ));
                         }
                     }
-                    // Push provider usage onto the shared carry (#4467, item 1),
-                    // mirroring the buffered path — before building the terminal
-                    // item, so it is queued ahead of the crate `UsageRecorded`.
-                    if let Some(u) = &resp.usage {
-                        usage_carry
-                            .lock()
-                            .unwrap_or_else(|p| p.into_inner())
-                            .push_back(u.clone());
-                    }
+                    // Provider usage rides the `Completed` response's crate `Usage`
+                    // + raw (G1); `UsageCarryMiddleware` reads it off the folded
+                    // response for the bridge, so the adapter no longer pushes here.
                     ModelStreamItem::Completed(response_to_model_response(&resp, &pformat_registry))
                 }
                 Err(e) => {

src/openhuman/tinyagents/observability.rs

@@ -600,36 +600,59 @@ impl EventListener for OpenhumanEventBridge {
                         }),
                     }
                 }
-                // Tool-call **argument** fragments now ride the crate stream
-                // (`MessageDelta.tool_call`) instead of the out-of-band
-                // `ThinkingForwarder`. Project them onto the same
-                // `ToolCallArgsDelta` the UI timeline consumes so the model can
-                // be shown composing the call before it executes. The crate
-                // `ToolDelta` carries no `tool_name`, so we recover it from the
-                // shared map the forwarder populated on the tool-call start
-                // event (empty until the start marker lands — matching the
-                // legacy forwarder's own default). There is no `Subagent*`
-                // tool-arg variant, and an UNSCOPED top-level `ToolCallArgsDelta`
-                // emitted from a child run would render the child's argument
-                // composition as the *parent's* own timeline activity (#4467,
-                // item 6). v0.58.7 dropped child arg fragments, so restore that:
-                // only a parent/top-level turn projects these fragments; a child
-                // run drops them (its Started/Completed rows already carry the
-                // final arguments under the `Subagent*` scope).
+                // Tool-call **start** + **argument** fragments both ride the crate
+                // stream (`MessageDelta.tool_call`) now — the out-of-band
+                // `ThinkingForwarder` is gone. The call-opening delta carries the
+                // tool name (crate `ToolDelta::tool_name`, G2) with empty content;
+                // argument fragments carry content with no name. We record the
+                // name on the opening delta so subsequent fragments can be
+                // labelled, and project both onto the `ToolCallArgsDelta` the UI
+                // timeline consumes so the model can be shown composing the call
+                // before it executes.
                 if let Some(tool_call) = &delta.tool_call {
-                    if self.scope.is_none() && !tool_call.content.is_empty() {
-                        let tool_name = self
-                            .tool_names
+                    // Record the tool name as soon as the call opens (matching the
+                    // legacy forwarder's `note_tool_call`), and emit the start
+                    // marker — an empty-delta `ToolCallArgsDelta` — top-level
+                    // regardless of scope, exactly as the forwarder did.
+                    if let Some(name) = tool_call.tool_name.as_deref().filter(|n| !n.is_empty()) {
+                        self.tool_names
                             .lock()
                             .unwrap()
-                            .get(&tool_call.call_id)
-                            .cloned()
-                            .unwrap_or_default();
+                            .insert(tool_call.call_id.clone(), name.to_string());
+                        if tool_call.content.is_empty() {
+                            self.send(AgentProgress::ToolCallArgsDelta {
+                                call_id: tool_call.call_id.clone(),
+                                tool_name: name.to_string(),
+                                delta: String::new(),
+                                iteration,
+                            });
+                        }
+                    }
+                    // Argument fragments are parent-only: there is no `Subagent*`
+                    // tool-arg variant, and an UNSCOPED top-level `ToolCallArgsDelta`
+                    // emitted from a child run would render the child's argument
+                    // composition as the *parent's* own timeline activity (#4467,
+                    // item 6; v0.58.7 dropped child arg fragments). A child run's
+                    // Started/Completed rows already carry the final arguments
+                    // under the `Subagent*` scope.
+                    if self.scope.is_none() && !tool_call.content.is_empty() {
+                        let tool_name = tool_call
+                            .tool_name
+                            .as_deref()
+                            .filter(|n| !n.is_empty())
+                            .map(str::to_string)
+                            .unwrap_or_else(|| {
+                                self.tool_names
+                                    .lock()
+                                    .unwrap()
+                                    .get(&tool_call.call_id)
+                                    .cloned()
+                                    .unwrap_or_default()
+                            });
                         tracing::trace!(
                             call_id = tool_call.call_id.as_str(),
                             tool_name = tool_name.as_str(),
                             len = tool_call.content.len(),
-                            child = self.scope.is_some(),
                             "[stream] projecting crate tool-arg fragment onto ToolCallArgsDelta"
                         );
                         self.send(AgentProgress::ToolCallArgsDelta {

src/openhuman/tinyagents/routes.rs

@@ -88,11 +88,6 @@ pub(super) fn build_route_models(
     provider: &Arc<dyn Provider>,
     temperature: f64,
     skip_model: &str,
-    // Shared provider-usage carry (#4467): fallback route models must drain the
-    // SAME carry the bridge reads, or a successful fallback call never feeds its
-    // backend-charged USD / context-window / cache-creation-reasoning breakdown
-    // to the cost surfaces (it would fall back to a catalogue estimate).
-    usage_carry: &super::observability::ProviderUsageCarry,
 ) -> Vec<RouteModel> {
     let mut out = Vec::new();
     for &tier in WORKLOAD_ROUTE_TIERS {
@@ -112,8 +107,10 @@ pub(super) fn build_route_models(
         }
         let mut model = ProviderModel::new(provider.clone(), tier, temperature)
             .with_vision(vision)
-            .with_reasoning(reasoning)
-            .with_usage_carry(usage_carry.clone());
+            .with_reasoning(reasoning);
+        // Provider usage (incl. fallback-route calls) reaches the cost bridge via
+        // `UsageCarryMiddleware`, which reads it off each response — so route
+        // models no longer carry the usage side-channel.
         // The per-turn output cap now rides `RunConfig.max_turn_output_tokens`
         // (Phase 5 groundwork): the loop stamps it onto every `ModelRequest`, so
         // route models no longer bake it in — they carry only model identity +
@@ -318,3 +315,55 @@ impl ModelMiddleware<()> for FallbackObserverMiddleware {
         Ok(MiddlewareModelOutcome::from(response))
     }
 }
+
+/// Around-model middleware that feeds the cost event bridge (issue #4249,
+/// Phase 5): after the real model call, it reads the full host [`UsageInfo`] off
+/// the returned [`ModelResponse`] — token breakdowns from the crate `Usage`,
+/// backend-charged USD + context window from the G1 `raw` passthrough
+/// ([`usage_info_from_response`](super::model::usage_info_from_response)) — and
+/// pushes it onto the shared [`ProviderUsageCarry`](super::observability::ProviderUsageCarry)
+/// the [`OpenhumanEventBridge`](super::OpenhumanEventBridge) drains on
+/// `UsageRecorded`.
+///
+/// This replaces the per-[`ProviderModel`] usage push (buffered + streamed), so
+/// the adapter — and every projected route model — carries only model identity +
+/// capability profile. It wraps the whole retry/fallback core, so it fires
+/// exactly once per logical model call (matching the single `UsageRecorded` the
+/// crate emits), for both the buffered and streamed paths (the streamed response
+/// is folded back to a `ModelResponse` with usage + raw intact). Push happens
+/// after the call returns, before the loop emits `UsageRecorded`, preserving the
+/// FIFO ordering the bridge relies on.
+pub(super) struct UsageCarryMiddleware {
+    carry: super::observability::ProviderUsageCarry,
+}
+
+impl UsageCarryMiddleware {
+    pub(super) fn new(carry: super::observability::ProviderUsageCarry) -> Self {
+        Self { carry }
+    }
+}
+
+#[async_trait]
+impl ModelMiddleware<()> for UsageCarryMiddleware {
+    fn name(&self) -> &str {
+        "openhuman.usage_carry"
+    }
+
+    async fn wrap_model(
+        &self,
+        ctx: &mut RunContext<()>,
+        state: &(),
+        request: ModelRequest,
+        next: ModelHandler<'_, (), ()>,
+    ) -> tinyagents::Result<MiddlewareModelOutcome> {
+        let outcome = next.run(ctx, state, request).await?;
+        let response = outcome.into_response();
+        if let Some(usage) = super::model::usage_info_from_response(&response) {
+            self.carry
+                .lock()
+                .unwrap_or_else(|p| p.into_inner())
+                .push_back(usage);
+        }
+        Ok(MiddlewareModelOutcome::from(response))
+    }
+}

src/openhuman/tinyagents/summarize.rs

@@ -33,7 +33,8 @@ use tinyagents::harness::summarization::{
     estimate_tokens, CompressionProvenance, SummarizationPolicy, Summarizer, SummaryRecord,
 };
 
-use crate::openhuman::inference::provider::Provider;
+use tinyagents::harness::message::Message as HarnessMessage;
+use tinyagents::harness::model::{ChatModel, ModelRequest};
 
 /// Fraction of the model's context window at which summarization fires.
 ///
@@ -47,29 +48,25 @@ const SUMMARIZE_THRESHOLD_FRACTION: f64 = 0.90;
 const SUMMARIZE_KEEP_LAST: usize = 8;
 
 /// An LLM-backed [`Summarizer`] that condenses a slice of harness [`TaMessage`]s
-/// into a single system summary via an openhuman [`Provider`] chat call.
+/// into a single system summary via a crate [`ChatModel`] call.
 ///
-/// Wraps the **same** provider + model the turn is already running on, so the
-/// summary is produced by the active model (a cheaper summarizer model can be
-/// threaded later if compaction on the main model proves expensive — the legacy
-/// `ContextConfig::summarizer_model` hook).
+/// Wraps the **same** model the turn is already running on (its id/temperature
+/// baked in), so the summary is produced by the active model (a cheaper
+/// summarizer model can be threaded later if compaction on the main model proves
+/// expensive — the legacy `ContextConfig::summarizer_model` hook).
 pub(super) struct ProviderModelSummarizer {
-    provider: Arc<dyn Provider>,
-    model: String,
-    temperature: f64,
+    model: Arc<dyn ChatModel<()>>,
+    /// Model id, kept for logging/provenance only (the id rides the wrapped
+    /// [`ChatModel`]).
+    model_id: String,
 }
 
 impl ProviderModelSummarizer {
-    /// Build a summarizer over `provider`/`model` at `temperature`.
-    pub(super) fn new(
-        provider: Arc<dyn Provider>,
-        model: impl Into<String>,
-        temperature: f64,
-    ) -> Self {
+    /// Build a summarizer over `model` (its id/temperature pinned).
+    pub(super) fn new(model: Arc<dyn ChatModel<()>>, model_id: impl Into<String>) -> Self {
         Self {
-            provider,
-            model: model.into(),
-            temperature,
+            model,
+            model_id: model_id.into(),
         }
     }
 }
@@ -107,25 +104,25 @@ impl Summarizer for ProviderModelSummarizer {
             .join("\n");
 
         tracing::info!(
-            model = %self.model,
+            model = %self.model_id,
             head_messages = messages.len(),
             approx_input_tokens = original_token_estimate,
             "[tinyagents::summarize] dispatching context-window summary"
         );
 
+        let request = ModelRequest::new(vec![
+            HarnessMessage::system(SUMMARIZER_SYSTEM_PROMPT),
+            HarnessMessage::user(transcript),
+        ]);
         let summary = self
-            .provider
-            .chat_with_system(
-                Some(SUMMARIZER_SYSTEM_PROMPT),
-                &transcript,
-                &self.model,
-                self.temperature,
-            )
+            .model
+            .invoke(&(), request)
             .await
             .map_err(|e| {
-                tracing::warn!(error = %e, "[tinyagents::summarize] summarizer provider call failed");
-                TinyAgentsError::Model(format!("summarizer provider call failed: {e}"))
-            })?;
+                tracing::warn!(error = %e, "[tinyagents::summarize] summarizer model call failed");
+                TinyAgentsError::Model(format!("summarizer model call failed: {e}"))
+            })?
+            .text();
 
         let summary = summary.trim();
         if summary.is_empty() {
@@ -138,7 +135,7 @@ impl Summarizer for ProviderModelSummarizer {
         let summary_token_estimate = estimate_tokens(&body);
 
         tracing::info!(
-            model = %self.model,
+            model = %self.model_id,
             summary_tokens = summary_token_estimate,
             freed_tokens = original_token_estimate.saturating_sub(summary_token_estimate),
             "[tinyagents::summarize] context-window summary complete"
@@ -152,7 +149,7 @@ impl Summarizer for ProviderModelSummarizer {
                 summary_token_estimate,
                 reason: format!(
                     "ProviderModelSummarizer via {} (LLM compaction at {:.0}% of context window)",
-                    self.model,
+                    self.model_id,
                     SUMMARIZE_THRESHOLD_FRACTION * 100.0
                 ),
             },

src/openhuman/tinyagents/tests.rs

@@ -159,10 +159,13 @@ async fn streaming_path_forwards_text_deltas_and_cost() {
     let registry: Arc<Vec<Box<dyn Tool>>> = Arc::new(vec![]);
     let history = vec![ChatMessage::user("hi")];
 
+    let provider: Arc<dyn Provider> = Arc::new(StreamingProvider);
+    let provider_id = provider.telemetry_provider_id();
+    let turn_models = build_turn_models(provider, "mock-model", 0.0, None);
     let outcome = run_turn_via_tinyagents_shared(
-        Arc::new(StreamingProvider),
+        turn_models,
+        provider_id,
         "mock-model",
-        0.0,
         history,
         vec![registry],
         None,
@@ -262,10 +265,12 @@ async fn pre_queued_steer_message_is_injected_into_the_request() {
         .await;
 
     let registry: Arc<Vec<Box<dyn Tool>>> = Arc::new(vec![]);
+    let provider_id = provider.telemetry_provider_id();
+    let turn_models = build_turn_models(provider.clone(), "mock-model", 0.0, None);
     let outcome = run_turn_via_tinyagents_shared(
-        provider.clone(),
+        turn_models,
+        provider_id,
         "mock-model",
-        0.0,
         vec![ChatMessage::user("investigate the bug")],
         vec![registry],
         None,
@@ -360,10 +365,11 @@ async fn concurrent_shared_turns_each_get_a_distinct_result() {
     });
     let registry: Arc<Vec<Box<dyn Tool>>> = Arc::new(vec![]);
 
+    let provider_id = provider.telemetry_provider_id();
     let one = run_turn_via_tinyagents_shared(
-        provider.clone(),
+        build_turn_models(provider.clone(), "mock-model", 0.0, None),
+        provider_id.clone(),
         "mock-model",
-        0.0,
         vec![ChatMessage::user("task one")],
         vec![registry.clone()],
         None,
@@ -382,9 +388,9 @@ async fn concurrent_shared_turns_each_get_a_distinct_result() {
         false, // defer_turn_completed_to_caller (#4457)
     );
     let two = run_turn_via_tinyagents_shared(
-        provider.clone(),
+        build_turn_models(provider.clone(), "mock-model", 0.0, None),
+        provider_id,
         "mock-model",
-        0.0,
         vec![ChatMessage::user("task two")],
         vec![registry],
         None,
@@ -436,9 +442,8 @@ fn adapter_inventory_registers_model_tools_and_middleware() {
         vec![Arc::new(vec![Box::new(EchoTool) as Box<dyn Tool>])];
 
     let assembled = assemble_turn_harness(
-        provider,
+        build_turn_models(provider, "mock-model", 0.0, Some(200_000)),
         "mock-model",
-        0.0,
         tool_sets,
         None,
         4,
@@ -527,7 +532,14 @@ fn adapter_inventory_registers_model_tools_and_middleware() {
     // CLI/RPC-only scope gate + credential scrub (#4453, innermost). No builder
     // tool policy on this call.
     assert_eq!(mw.tool_middleware_len(), 4, "tool middleware inventory");
-    assert_eq!(mw.model_middleware_len(), 0, "no around-model wraps");
+    // One around-model wrap: the cost `UsageCarryMiddleware` (always installed).
+    // RequiredCapabilities/FallbackObserver are not installed on this call
+    // (no required caps; `mock-model` is not a tier, so no fallback chain).
+    assert_eq!(
+        mw.model_middleware_len(),
+        1,
+        "usage-carry around-model wrap"
+    );
     assert_eq!(
         assembled.harness.policy().limits.max_depth,
         crate::openhuman::agent::harness::MAX_SPAWN_DEPTH,
@@ -573,9 +585,8 @@ fn adapter_inventory_gates_context_middleware_on_window() {
         vec![Arc::new(vec![Box::new(EchoTool) as Box<dyn Tool>])];
 
     let assembled = assemble_turn_harness(
-        provider,
+        build_turn_models(provider, "mock-model", 0.0, None),
         "mock-model",
-        0.0,
         tool_sets,
         None,
         4,
@@ -646,10 +657,13 @@ async fn unobserved_turn_reports_aggregate_usage_for_the_cost_fallback() {
         }
     }
 
+    let provider: Arc<dyn Provider> = Arc::new(DoneWithUsage);
+    let provider_id = provider.telemetry_provider_id();
+    let turn_models = build_turn_models(provider, "mock-model", 0.0, None);
     let outcome = run_turn_via_tinyagents_shared(
-        Arc::new(DoneWithUsage),
+        turn_models,
+        provider_id,
         "mock-model",
-        0.0,
         vec![ChatMessage::user("hello")],
         Vec::new(),
         None,