From 2ce5e6027fbb8710e086a345d045879de1e7d1af Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Mon, 6 Jul 2026 23:28:40 -0700 Subject: [PATCH] =?UTF-8?q?refactor(inference):=20Phase=205=20=E2=80=94=20?= =?UTF-8?q?decouple=20per-turn=20state=20from=20ProviderModel=20(usage=20m?= =?UTF-8?q?iddleware,=20G2,=20build=5Fturn=5Fmodels)=20(#4625)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/openhuman/agent/harness/graph.rs | 11 +- .../agent/harness/session/turn/graph.rs | 11 +- .../harness/subagent_runner/ops/graph.rs | 11 +- src/openhuman/agent_registry/agents/loader.rs | 1 + src/openhuman/tinyagents/mod.rs | 157 ++++++++----- src/openhuman/tinyagents/model.rs | 216 ++++-------------- src/openhuman/tinyagents/observability.rs | 69 ++++-- src/openhuman/tinyagents/routes.rs | 63 ++++- src/openhuman/tinyagents/summarize.rs | 59 +++-- src/openhuman/tinyagents/tests.rs | 44 ++-- vendor/tinyagents | 2 +- 11 files changed, 332 insertions(+), 312 deletions(-) diff --git a/src/openhuman/agent/harness/graph.rs b/src/openhuman/agent/harness/graph.rs index ace8c0d50..eae27c482 100644 --- a/src/openhuman/agent/harness/graph.rs +++ b/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, diff --git a/src/openhuman/agent/harness/session/turn/graph.rs b/src/openhuman/agent/harness/session/turn/graph.rs index 11a044d90..171c25e4f 100644 --- a/src/openhuman/agent/harness/session/turn/graph.rs +++ b/src/openhuman/agent/harness/session/turn/graph.rs @@ -89,10 +89,19 @@ pub(crate) async fn run_chat_turn_graph(graph: ChatTurnGraph) -> Result bool { #[allow(clippy::too_many_arguments)] pub(crate) async fn run_turn_via_tinyagents_shared( - provider: Arc, + turn_models: TurnModels, + provider_id: String, model: &str, - temperature: f64, history: Vec, tool_sets: Vec>>>, allowed: Option>, @@ -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>, + /// Additive workload-tier routes (registry name → model), excluding the + /// primary; the crate registry resolves fallback/selection across them. + routes: Vec<(String, Arc>)>, + /// 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>, + /// 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, + model: &str, + temperature: f64, + context_window: Option, +) -> 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> = Arc::new(primary); + + let routes = routes::build_route_models(&summary_provider, temperature, model) + .into_iter() + .map(|route| { + let model: Arc> = 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> = + 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, + turn_models: TurnModels, model: &str, - temperature: f64, tool_sets: Vec>>>, allowed: Option>, 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, ); diff --git a/src/openhuman/tinyagents/model.rs b/src/openhuman/tinyagents/model.rs index f7fea1972..c19396f45 100644 --- a/src/openhuman/tinyagents/model.rs +++ b/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, - scope: Option, - 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, - scope: Option, - 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, - 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, 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, - /// When set, the adapter forwards tool-argument progress and post-hoc - /// non-streaming reasoning onto the progress sink. - thinking: Option, /// 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::(); @@ -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) => { diff --git a/src/openhuman/tinyagents/observability.rs b/src/openhuman/tinyagents/observability.rs index 39766b9d4..c38685a98 100644 --- a/src/openhuman/tinyagents/observability.rs +++ b/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 { diff --git a/src/openhuman/tinyagents/routes.rs b/src/openhuman/tinyagents/routes.rs index b077d4f76..fb057a2f7 100644 --- a/src/openhuman/tinyagents/routes.rs +++ b/src/openhuman/tinyagents/routes.rs @@ -88,11 +88,6 @@ pub(super) fn build_route_models( provider: &Arc, 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 { 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 { + 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)) + } +} diff --git a/src/openhuman/tinyagents/summarize.rs b/src/openhuman/tinyagents/summarize.rs index 1e95131a6..206c304ed 100644 --- a/src/openhuman/tinyagents/summarize.rs +++ b/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, - model: String, - temperature: f64, + model: Arc>, + /// 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, - model: impl Into, - temperature: f64, - ) -> Self { + /// Build a summarizer over `model` (its id/temperature pinned). + pub(super) fn new(model: Arc>, model_id: impl Into) -> 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 ), }, diff --git a/src/openhuman/tinyagents/tests.rs b/src/openhuman/tinyagents/tests.rs index facd39d7e..27f8b32c8 100644 --- a/src/openhuman/tinyagents/tests.rs +++ b/src/openhuman/tinyagents/tests.rs @@ -159,10 +159,13 @@ async fn streaming_path_forwards_text_deltas_and_cost() { let registry: Arc>> = Arc::new(vec![]); let history = vec![ChatMessage::user("hi")]; + let provider: Arc = 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>> = 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>> = 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])]; 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])]; 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 = 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, diff --git a/vendor/tinyagents b/vendor/tinyagents index 6bf67ace4..b1f0d82d1 160000 --- a/vendor/tinyagents +++ b/vendor/tinyagents @@ -1 +1 @@ -Subproject commit 6bf67ace49914ed53684a8b1f3ff52dbbdd7fbd4 +Subproject commit b1f0d82d1845d5e3de020882b8817565f529cad9