From b03412111d0df3796d045d7a1530a0fc13090d64 Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Tue, 5 May 2026 22:36:01 -0700 Subject: [PATCH] Agent harness: USD cost tracking, stop hooks, per-tool result caps (#1268) --- src/openhuman/agent/cost.rs | 264 ++++++++++++++++ src/openhuman/agent/debug/mod.rs | 4 +- .../agent/harness/builtin_definitions.rs | 85 +---- src/openhuman/agent/harness/definition.rs | 8 - .../agent/harness/definition_loader.rs | 1 - .../agent/harness/definition_tests.rs | 1 - src/openhuman/agent/harness/fork_context.rs | 79 +---- src/openhuman/agent/harness/mod.rs | 11 +- .../agent/harness/payload_summarizer.rs | 1 - .../agent/harness/sandbox_context.rs | 12 +- src/openhuman/agent/harness/session/tests.rs | 108 ------- src/openhuman/agent/harness/session/turn.rs | 69 +--- .../agent/harness/session/turn_tests.rs | 18 -- .../agent/harness/subagent_runner/ops.rs | 134 +------- .../harness/subagent_runner/ops_tests.rs | 74 +---- .../agent/harness/subagent_runner/types.rs | 14 +- src/openhuman/agent/harness/tool_loop.rs | 150 +++++++-- .../agent/harness/tool_loop_tests.rs | 219 +++++++++++++ src/openhuman/agent/mod.rs | 2 + src/openhuman/agent/progress.rs | 30 +- src/openhuman/agent/stop_hooks.rs | 294 ++++++++++++++++++ src/openhuman/channels/providers/web.rs | 17 + src/openhuman/channels/runtime/dispatch.rs | 1 - src/openhuman/threads/turn_state/mirror.rs | 8 + .../tools/impl/agent/spawn_subagent.rs | 34 +- .../tools/impl/filesystem/file_read.rs | 5 + .../tools/impl/filesystem/glob_search.rs | 5 + src/openhuman/tools/impl/filesystem/grep.rs | 5 + src/openhuman/tools/impl/network/web_fetch.rs | 16 + src/openhuman/tools/impl/system/shell.rs | 9 + src/openhuman/tools/orchestrator_tools.rs | 1 - src/openhuman/tools/traits.rs | 54 ++++ tests/agent_harness_public.rs | 22 +- 33 files changed, 1096 insertions(+), 659 deletions(-) create mode 100644 src/openhuman/agent/cost.rs create mode 100644 src/openhuman/agent/stop_hooks.rs diff --git a/src/openhuman/agent/cost.rs b/src/openhuman/agent/cost.rs new file mode 100644 index 000000000..325a1cde9 --- /dev/null +++ b/src/openhuman/agent/cost.rs @@ -0,0 +1,264 @@ +//! Per-turn cost accounting for an agent's tool-call loop. +//! +//! Each provider response carries an optional [`UsageInfo`] block with +//! `input_tokens`, `output_tokens`, `cached_input_tokens`, and an +//! authoritative `charged_amount_usd` populated by the OpenHuman +//! backend. [`TurnCost`] sums those across every provider call inside a +//! single turn so the harness can: +//! +//! - emit per-iteration cost telemetry via +//! [`crate::openhuman::agent::progress::AgentProgress::TurnCostUpdated`]; +//! - feed an upcoming budget stop-hook (mid-turn USD cap); +//! - log accurate end-of-turn cost lines. +//! +//! When `charged_amount_usd` is zero (older backend builds, providers +//! that don't surface billing), we fall back to a simple token-rate +//! estimate via [`estimate_call_cost_usd`] keyed on the model tier +//! name. The estimate is a floor — directly-billed cost from the +//! backend always wins when available. +//! +//! The pricing table is intentionally tiny and only keyed on the +//! abstract tier names the core uses (`agentic-v1`, `reasoning-v1`, +//! `coding-v1`). The backend resolves them to concrete vendor models; +//! cents-per-Mtok at the tier level is good enough for client-side +//! telemetry and budget gating. PRs adding new tiers should add a row. + +use crate::openhuman::providers::UsageInfo; + +/// Per-million-token rates for a single model tier. +/// +/// All prices are USD per million tokens. `cached_input_per_mtok_usd` +/// applies to the `cached_input_tokens` portion of the usage block (KV +/// prefix cache hits on supporting backends); the remaining +/// `input_tokens - cached_input_tokens` are charged at +/// `input_per_mtok_usd`. +#[derive(Debug, Clone, Copy)] +pub struct ModelPricing { + /// Tier identifier, e.g. `"agentic-v1"`. + pub model: &'static str, + /// Standard prompt rate, USD per million input tokens. + pub input_per_mtok_usd: f64, + /// Cached-prefix prompt rate, USD per million cached input tokens. + pub cached_input_per_mtok_usd: f64, + /// Completion rate, USD per million output tokens. + pub output_per_mtok_usd: f64, +} + +/// Conservative fallback when nothing in the table matches. Picked so +/// budget caps still bite on unknown models rather than reading as $0. +const FALLBACK_PRICING: ModelPricing = ModelPricing { + model: "", + input_per_mtok_usd: 3.00, + cached_input_per_mtok_usd: 0.30, + output_per_mtok_usd: 15.00, +}; + +/// Static price table keyed by tier name. +/// +/// These are the OpenHuman tier handles, not concrete vendor model +/// strings — the backend chooses which underlying Claude / GPT / etc. +/// model serves each tier. Numbers track the public Anthropic price +/// list at the time of writing for the tiers' default mappings; treat +/// them as best-effort estimates for cases where the backend doesn't +/// echo `charged_amount_usd`. +pub const PRICING_TABLE: &[ModelPricing] = &[ + // Reasoning tier — currently maps to Claude Opus 4.x family. + ModelPricing { + model: "reasoning-v1", + input_per_mtok_usd: 15.00, + cached_input_per_mtok_usd: 1.50, + output_per_mtok_usd: 75.00, + }, + // Agentic tier — maps to Sonnet-class models. + ModelPricing { + model: "agentic-v1", + input_per_mtok_usd: 3.00, + cached_input_per_mtok_usd: 0.30, + output_per_mtok_usd: 15.00, + }, + // Coding tier — Sonnet-class. + ModelPricing { + model: "coding-v1", + input_per_mtok_usd: 3.00, + cached_input_per_mtok_usd: 0.30, + output_per_mtok_usd: 15.00, + }, +]; + +/// Look up pricing for a model name, falling back to [`FALLBACK_PRICING`]. +/// +/// Matching is exact on the canonical tier name and case-insensitive on +/// concrete vendor names (so `"claude-opus"` still hits the +/// reasoning-tier row when callers pass an underlying model string). +pub fn lookup_pricing(model: &str) -> ModelPricing { + if let Some(row) = PRICING_TABLE.iter().find(|row| row.model == model) { + return *row; + } + let lower = model.to_ascii_lowercase(); + if lower.contains("opus") { + return PRICING_TABLE[0]; + } + if lower.contains("coding") { + return PRICING_TABLE[2]; + } + if lower.contains("sonnet") || lower.contains("agentic") { + return PRICING_TABLE[1]; + } + FALLBACK_PRICING +} + +/// Estimate the USD cost of a single provider call from its token +/// usage. Used as a fallback when `charged_amount_usd` is missing. +pub fn estimate_call_cost_usd(model: &str, usage: &UsageInfo) -> f64 { + let pricing = lookup_pricing(model); + let cached = usage.cached_input_tokens; + let standard_input = usage.input_tokens.saturating_sub(cached); + let m = 1_000_000.0_f64; + (standard_input as f64) / m * pricing.input_per_mtok_usd + + (cached as f64) / m * pricing.cached_input_per_mtok_usd + + (usage.output_tokens as f64) / m * pricing.output_per_mtok_usd +} + +/// Pick the most authoritative USD figure for a single provider call. +/// +/// Backend-reported `charged_amount_usd` wins whenever it's > 0; +/// otherwise we fall back to [`estimate_call_cost_usd`]. +pub fn call_cost_usd(model: &str, usage: &UsageInfo) -> f64 { + if usage.charged_amount_usd > 0.0 { + usage.charged_amount_usd + } else { + estimate_call_cost_usd(model, usage) + } +} + +/// Running cost / token tally across every provider call inside a +/// single turn of the tool-call loop. +/// +/// `charged_usd` is the sum of authoritative `charged_amount_usd` +/// values; `estimated_usd` adds the fallback estimate for any call that +/// lacked one. `total_usd()` returns whichever has more signal. +#[derive(Debug, Clone, Default)] +pub struct TurnCost { + pub input_tokens: u64, + pub output_tokens: u64, + pub cached_input_tokens: u64, + pub charged_usd: f64, + pub estimated_usd: f64, + pub call_count: u32, +} + +impl TurnCost { + /// New empty accumulator. + pub fn new() -> Self { + Self::default() + } + + /// Fold a single provider call's usage into the running totals. + pub fn add_call(&mut self, model: &str, usage: &UsageInfo) { + self.input_tokens = self.input_tokens.saturating_add(usage.input_tokens); + self.output_tokens = self.output_tokens.saturating_add(usage.output_tokens); + self.cached_input_tokens = self + .cached_input_tokens + .saturating_add(usage.cached_input_tokens); + if usage.charged_amount_usd > 0.0 { + self.charged_usd += usage.charged_amount_usd; + } else { + self.estimated_usd += estimate_call_cost_usd(model, usage); + } + self.call_count = self.call_count.saturating_add(1); + } + + /// Best-available USD figure: authoritative charged amount plus + /// estimated cost for any calls that didn't carry one. + pub fn total_usd(&self) -> f64 { + self.charged_usd + self.estimated_usd + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn usage(input: u64, output: u64, cached: u64, charged: f64) -> UsageInfo { + UsageInfo { + input_tokens: input, + output_tokens: output, + cached_input_tokens: cached, + charged_amount_usd: charged, + ..Default::default() + } + } + + #[test] + fn lookup_pricing_matches_canonical_tiers() { + assert_eq!(lookup_pricing("reasoning-v1").input_per_mtok_usd, 15.0); + assert_eq!(lookup_pricing("agentic-v1").output_per_mtok_usd, 15.0); + } + + #[test] + fn lookup_pricing_falls_back_for_unknown_model() { + let p = lookup_pricing("totally-unknown-model"); + assert_eq!(p.model, ""); + } + + #[test] + fn lookup_pricing_handles_concrete_vendor_names() { + assert_eq!(lookup_pricing("claude-opus-4.7").input_per_mtok_usd, 15.0); + assert_eq!( + lookup_pricing("claude-sonnet-4-6").output_per_mtok_usd, + 15.0 + ); + } + + #[test] + fn lookup_pricing_routes_coding_to_coding_row_not_agentic() { + // Pinned per CodeRabbit feedback: when the coding-tier row + // diverges from agentic, "coding" model strings must hit + // PRICING_TABLE[2], not [1]. + assert_eq!(lookup_pricing("coding-v1").model, "coding-v1"); + assert_eq!(lookup_pricing("agentic-v1").model, "agentic-v1"); + } + + #[test] + fn estimate_call_cost_subtracts_cached_input() { + // 1M standard input + 1M cached input + 1M output on agentic-v1. + let u = usage(2_000_000, 1_000_000, 1_000_000, 0.0); + let est = estimate_call_cost_usd("agentic-v1", &u); + // 1M * 3 + 1M * 0.3 + 1M * 15 = 18.3 + assert!((est - 18.3).abs() < 1e-6, "got {est}"); + } + + #[test] + fn call_cost_prefers_charged_when_present() { + let u = usage(100_000, 200_000, 0, 0.42); + assert_eq!(call_cost_usd("reasoning-v1", &u), 0.42); + } + + #[test] + fn call_cost_falls_back_to_estimate_when_charged_zero() { + let u = usage(1_000_000, 0, 0, 0.0); + // 1M input * 3 = 3 + assert!((call_cost_usd("agentic-v1", &u) - 3.0).abs() < 1e-6); + } + + #[test] + fn turn_cost_accumulates_charged_and_estimated_separately() { + let mut tc = TurnCost::new(); + tc.add_call("reasoning-v1", &usage(0, 0, 0, 0.10)); + tc.add_call("agentic-v1", &usage(1_000_000, 0, 0, 0.0)); // est: 3.00 + assert_eq!(tc.call_count, 2); + assert!((tc.charged_usd - 0.10).abs() < 1e-6); + assert!((tc.estimated_usd - 3.0).abs() < 1e-6); + assert!((tc.total_usd() - 3.10).abs() < 1e-6); + } + + #[test] + fn turn_cost_aggregates_token_counts() { + let mut tc = TurnCost::new(); + tc.add_call("agentic-v1", &usage(100, 50, 20, 0.0)); + tc.add_call("agentic-v1", &usage(200, 75, 0, 0.0)); + assert_eq!(tc.input_tokens, 300); + assert_eq!(tc.output_tokens, 125); + assert_eq!(tc.cached_input_tokens, 20); + } +} diff --git a/src/openhuman/agent/debug/mod.rs b/src/openhuman/agent/debug/mod.rs index 5fc9229ef..e1a9bcaac 100644 --- a/src/openhuman/agent/debug/mod.rs +++ b/src/openhuman/agent/debug/mod.rs @@ -401,8 +401,8 @@ async fn render_integrations_agent(config: &Config, toolkit: &str) -> Result = rendered_tools diff --git a/src/openhuman/agent/harness/builtin_definitions.rs b/src/openhuman/agent/harness/builtin_definitions.rs index 33d0c262e..1619193a3 100644 --- a/src/openhuman/agent/harness/builtin_definitions.rs +++ b/src/openhuman/agent/harness/builtin_definitions.rs @@ -3,16 +3,12 @@ //! The authoritative list of built-in agents lives in //! [`crate::openhuman::agent::agents`] — each agent is a subfolder //! containing `agent.toml` + `prompt.md`. This module is a thin -//! wrapper that loads that set and appends the synthetic `fork` -//! definition (used for byte-exact prompt-cache reuse by the sub-agent -//! runner). +//! wrapper that loads that set. //! //! Custom TOML definitions loaded later by //! [`super::definition_loader`] override any built-in with the same id. -use super::definition::{ - AgentDefinition, DefinitionSource, ModelSpec, PromptSource, SandboxMode, ToolScope, -}; +use super::definition::{AgentDefinition, DefinitionSource}; /// All built-in definitions, in stable order. /// @@ -22,58 +18,8 @@ use super::definition::{ /// [`crate::openhuman::agent::agents`] verify in CI that every entry in /// [`crate::openhuman::agent::agents::BUILTINS`] still parses cleanly. pub fn all() -> Vec { - let mut out = crate::openhuman::agent::agents::load_builtins() - .expect("built-in agent TOML must always parse (see agents/*/agent.toml)"); - out.push(fork_definition()); - out -} - -/// The synthetic `fork` definition. Tells the runner to bypass normal -/// prompt construction and replay the parent's exact rendered system -/// prompt + tool schemas + message prefix from -/// [`super::fork_context::ForkContext`]. The OpenAI-compatible backend's -/// automatic prefix caching turns this into a real token win. -pub fn fork_definition() -> AgentDefinition { - AgentDefinition { - id: "fork".into(), - when_to_use: "Spawn a parallel sub-task that shares the parent's full system \ - prompt, tool set, and message history byte-for-byte. Use when \ - decomposing a task into independent parallel work streams that \ - benefit from prefix-cache reuse on the inference backend." - .into(), - display_name: Some("Fork".into()), - // Prompt source is irrelevant — the runner reads from ForkContext. - system_prompt: PromptSource::Inline(String::new()), - // Fork preserves bytes — DO NOT strip anything from the parent's prompt. - omit_identity: false, - omit_memory_context: false, - omit_safety_preamble: false, - omit_skills_catalog: false, - // Fork preserves the parent's exact prompt bytes — mirror whatever - // PROFILE.md injection state the parent already produced rather - // than re-gating here. - omit_profile: false, - omit_memory_md: false, - model: ModelSpec::Inherit, - // Inherit the parent's temperature too — set to a sentinel that the - // runner replaces with the parent's actual temp at spawn time. - // (We use 0.7 here as a safe default for documentation; the runner - // overrides it from `ParentExecutionContext::temperature`.) - temperature: 0.7, - tools: ToolScope::Wildcard, - disallowed_tools: vec![], - skill_filter: None, - extra_tools: vec![], - // Fork inherits the parent's max iterations from the runtime. - max_iterations: 15, - timeout_secs: None, - sandbox_mode: SandboxMode::None, - background: false, - uses_fork_context: true, - subagents: vec![], - delegate_name: None, - source: DefinitionSource::Builtin, - } + crate::openhuman::agent::agents::load_builtins() + .expect("built-in agent TOML must always parse (see agents/*/agent.toml)") } #[cfg(test)] @@ -83,11 +29,7 @@ mod tests { #[test] fn all_definitions_present() { let defs = all(); - // Every entry in `agents::BUILTINS` plus 1 synthetic `fork`. - assert_eq!( - defs.len(), - crate::openhuman::agent::agents::BUILTINS.len() + 1 - ); + assert_eq!(defs.len(), crate::openhuman::agent::agents::BUILTINS.len()); } #[test] @@ -102,22 +44,6 @@ mod tests { } } - #[test] - fn fork_definition_has_uses_fork_context_true() { - let def = fork_definition(); - assert_eq!(def.id, "fork"); - assert!(def.uses_fork_context); - assert!(matches!(def.model, ModelSpec::Inherit)); - assert!(matches!(def.tools, ToolScope::Wildcard)); - // Fork preserves bytes — must NOT strip anything. - assert!(!def.omit_identity); - assert!(!def.omit_memory_context); - assert!(!def.omit_safety_preamble); - assert!(!def.omit_skills_catalog); - assert!(!def.omit_profile); - assert!(!def.omit_memory_md); - } - #[test] fn expected_builtin_ids_are_present() { let ids: Vec = all().into_iter().map(|d| d.id).collect(); @@ -131,7 +57,6 @@ mod tests { "critic", "archivist", "summarizer", - "fork", ] { assert!(ids.contains(&expected.to_string()), "missing {expected}"); } diff --git a/src/openhuman/agent/harness/definition.rs b/src/openhuman/agent/harness/definition.rs index 0ca894702..864ef757b 100644 --- a/src/openhuman/agent/harness/definition.rs +++ b/src/openhuman/agent/harness/definition.rs @@ -139,10 +139,6 @@ pub struct AgentDefinition { #[serde(default)] pub background: bool, - /// Internal flag for `fork` mode sub-agents. - #[serde(default, skip_serializing_if = "is_false")] - pub uses_fork_context: bool, - // ── delegation surface ───────────────────────────────────────────── /// Subagents this agent is allowed to spawn via synthesised /// `delegate_*` tools. Each entry expands at agent-build time into @@ -227,10 +223,6 @@ impl SkillsWildcard { } } -fn is_false(b: &bool) -> bool { - !b -} - impl AgentDefinition { /// Display name with fallback to id. pub fn display_name(&self) -> &str { diff --git a/src/openhuman/agent/harness/definition_loader.rs b/src/openhuman/agent/harness/definition_loader.rs index 603c602ae..eda6dd059 100644 --- a/src/openhuman/agent/harness/definition_loader.rs +++ b/src/openhuman/agent/harness/definition_loader.rs @@ -233,7 +233,6 @@ hint = "agentic" ); assert!(reg.get("notion_specialist").is_some()); assert!(reg.get("code_executor").is_some()); - assert!(reg.get("fork").is_some()); } #[test] diff --git a/src/openhuman/agent/harness/definition_tests.rs b/src/openhuman/agent/harness/definition_tests.rs index 3c78a9964..e48018654 100644 --- a/src/openhuman/agent/harness/definition_tests.rs +++ b/src/openhuman/agent/harness/definition_tests.rs @@ -22,7 +22,6 @@ fn make_def(id: &str) -> AgentDefinition { timeout_secs: None, sandbox_mode: SandboxMode::None, background: false, - uses_fork_context: false, subagents: vec![], delegate_name: None, source: DefinitionSource::Builtin, diff --git a/src/openhuman/agent/harness/fork_context.rs b/src/openhuman/agent/harness/fork_context.rs index 8a8864d38..9df42566d 100644 --- a/src/openhuman/agent/harness/fork_context.rs +++ b/src/openhuman/agent/harness/fork_context.rs @@ -2,25 +2,18 @@ //! agent's runtime context (provider, tools, model, …) without widening //! the [`crate::openhuman::tools::Tool`] trait. //! -//! Two distinct task-locals live here: +//! [`PARENT_CONTEXT`] is set by the parent +//! [`crate::openhuman::agent::Agent`] around its `turn` so that any tool +//! executing inside that turn (in particular `spawn_subagent`) can read +//! the parent's provider, tool list, and model information. //! -//! 1. [`PARENT_CONTEXT`] — set by the parent [`crate::openhuman::agent::Agent`] -//! around its `turn` so that any tool executing inside that turn (in -//! particular `spawn_subagent`) can read the parent's provider, tool -//! list, and model information. -//! -//! 2. [`FORK_CONTEXT`] — set only when the parent dispatches a `fork`-mode -//! sub-agent. Carries the parent's *exact* rendered system prompt, tool -//! schemas, and message prefix so the forked child can replay the same -//! bytes and the inference backend's automatic prefix caching kicks in. -//! -//! Both contexts are stashed in `Arc`s so that cloning into the child -//! costs a refcount bump rather than a full copy. +//! Stashed in `Arc`s so cloning into a child costs a refcount bump +//! rather than a full copy. use crate::openhuman::agent::progress::AgentProgress; use crate::openhuman::config::AgentConfig; use crate::openhuman::memory::Memory; -use crate::openhuman::providers::{ChatMessage, Provider}; +use crate::openhuman::providers::Provider; use crate::openhuman::skills::Skill; use crate::openhuman::tools::{Tool, ToolSpec}; use std::path::PathBuf; @@ -152,61 +145,3 @@ where { PARENT_CONTEXT.scope(ctx, future).await } - -// ───────────────────────────────────────────────────────────────────────────── -// Fork context -// ───────────────────────────────────────────────────────────────────────────── - -/// Captures the parent's exact rendered prompt + tool schemas + message -/// prefix so a forked sub-agent can replay them byte-for-byte. -/// -/// **Why this matters**: OpenAI-compatible inference backends apply -/// automatic prefix caching server-side based on stable byte sequences. -/// If the forked child's request shares an identical prefix with the -/// parent's previous request, the prefix is served from cache and only -/// the diverging tail is billed. Forking this way is the biggest -/// token-saving mechanism OpenHuman has for parallel sub-agent work. -/// -/// To preserve byte stability we hold: -/// - `system_prompt` as a pre-rendered `String` (not the builder). -/// - `tool_specs` as already-serialised `ToolSpec` values. -/// - `message_prefix` as the parent's `ChatMessage` history *up to and -/// including* the assistant message that issued the `spawn_subagent` -/// tool call. -#[derive(Clone)] -pub struct ForkContext { - /// Parent's rendered system prompt. Becomes message[0] of the child. - pub system_prompt: Arc, - - /// Parent's tool schemas. The child's `ChatRequest.tools` borrows from - /// this slice unchanged. - pub tool_specs: Arc>, - - /// Parent's message history prefix that the child should replay - /// verbatim. Includes the system message at index 0. - pub message_prefix: Arc>, - - /// The actual instruction the model issued for *this* fork — appears - /// as the new user message appended after `message_prefix`. - pub fork_task_prompt: String, -} - -tokio::task_local! { - /// Fork context, scoped per `spawn_subagent { mode: "fork", … }` - /// invocation. The runner reads it when the requested definition has - /// `uses_fork_context = true`. - pub static FORK_CONTEXT: ForkContext; -} - -/// Returns a clone of the current fork context, if one is set. -pub fn current_fork() -> Option { - FORK_CONTEXT.try_with(|ctx| ctx.clone()).ok() -} - -/// Run `future` with `ctx` installed as the active fork context. -pub async fn with_fork_context(ctx: ForkContext, future: F) -> R -where - F: std::future::Future, -{ - FORK_CONTEXT.scope(ctx, future).await -} diff --git a/src/openhuman/agent/harness/mod.rs b/src/openhuman/agent/harness/mod.rs index f317f3bb8..c2b591723 100644 --- a/src/openhuman/agent/harness/mod.rs +++ b/src/openhuman/agent/harness/mod.rs @@ -1,4 +1,4 @@ -//! Multi-agent harness — sub-agent dispatch and fork-cache support. +//! Multi-agent harness — sub-agent dispatch and parent-context plumbing. //! //! The harness provides the infrastructure for an agent to delegate work to //! specialized sub-agents. It manages the lifecycle of these sub-agents, @@ -10,10 +10,8 @@ //! in the global [`AgentDefinitionRegistry`] and runs a dedicated tool loop. //! //! ## Token Optimization -//! - **Typed Sub-agents**: Skips unnecessary system prompt sections (e.g., +//! - **Typed Sub-agents**: Skip unnecessary system prompt sections (e.g., //! identity, global skills) to keep sub-agent prompts small. -//! - **Fork Mode**: Allows sub-agents to replay the parent's exact context -//! to leverage KV-cache reuse on the inference backend. //! //! ## Key Sub-modules //! - **[`subagent_runner`]**: The core logic for executing a sub-agent. @@ -44,10 +42,7 @@ pub use definition::{ AgentDefinition, AgentDefinitionRegistry, DefinitionSource, ModelSpec, PromptSource, SandboxMode, ToolScope, }; -pub use fork_context::{ - current_fork, current_parent, with_fork_context, with_parent_context, ForkContext, - ParentExecutionContext, -}; +pub use fork_context::{current_parent, with_parent_context, ParentExecutionContext}; pub use interrupt::{check_interrupt, InterruptFence, InterruptedError}; pub use sandbox_context::{current_sandbox_mode, with_current_sandbox_mode}; pub use subagent_runner::{run_subagent, SubagentRunError, SubagentRunOptions}; diff --git a/src/openhuman/agent/harness/payload_summarizer.rs b/src/openhuman/agent/harness/payload_summarizer.rs index 9923b7738..d3ce41242 100644 --- a/src/openhuman/agent/harness/payload_summarizer.rs +++ b/src/openhuman/agent/harness/payload_summarizer.rs @@ -362,7 +362,6 @@ mod tests { timeout_secs: None, sandbox_mode: SandboxMode::None, background: false, - uses_fork_context: false, subagents: vec![], delegate_name: None, source: DefinitionSource::Builtin, diff --git a/src/openhuman/agent/harness/sandbox_context.rs b/src/openhuman/agent/harness/sandbox_context.rs index 12ce99483..e0f502cf7 100644 --- a/src/openhuman/agent/harness/sandbox_context.rs +++ b/src/openhuman/agent/harness/sandbox_context.rs @@ -2,12 +2,12 @@ //! implementations can enforce sandbox semantics at execution time without //! widening the [`crate::openhuman::tools::Tool`] trait signature. //! -//! Sibling of the existing [`super::fork_context`] task-locals but serves -//! a different concept: `PARENT_CONTEXT` / `FORK_CONTEXT` carry the -//! *parent agent's* runtime context so that `spawn_subagent` can inherit -//! it, whereas [`CURRENT_AGENT_SANDBOX_MODE`] carries the *currently- -//! executing agent's* sandbox mode so that any tool it invokes can gate -//! on that mode. +//! Sibling of the existing [`super::fork_context`] task-local but serves +//! a different concept: `PARENT_CONTEXT` carries the *parent agent's* +//! runtime context so that `spawn_subagent` can inherit it, whereas +//! [`CURRENT_AGENT_SANDBOX_MODE`] carries the *currently-executing +//! agent's* sandbox mode so that any tool it invokes can gate on that +//! mode. //! //! Why a task-local instead of an argument on [`Tool::execute`]: the tool //! trait is called from many places (CLI, JSON-RPC, tests, agent loops). diff --git a/src/openhuman/agent/harness/session/tests.rs b/src/openhuman/agent/harness/session/tests.rs index e551cd5c6..627cf4c7b 100644 --- a/src/openhuman/agent/harness/session/tests.rs +++ b/src/openhuman/agent/harness/session/tests.rs @@ -480,114 +480,6 @@ async fn turn_dispatches_spawn_subagent_through_full_path() { ); } -/// Fork-mode variant of `turn_dispatches_spawn_subagent_through_full_path`. -/// -/// Exercises the prefix-replay path: the parent issues -/// `spawn_subagent { mode: "fork", … }`, the runner resolves the `fork` -/// built-in definition, pulls the parent's exact rendered prompt + tool -/// schemas + message prefix out of the `ForkContext` task-local, and -/// runs the inner loop on the parent's own provider. -/// -/// From the provider's perspective the response queue is consumed in -/// the same fixed sequence as the typed test — parent tool_call → sub- -/// agent reply → parent folded reply — which is the invariant that -/// makes KV-cache reuse possible on the real backend. -#[tokio::test] -async fn turn_dispatches_spawn_subagent_in_fork_mode() { - use crate::openhuman::agent::harness::AgentDefinitionRegistry; - use crate::openhuman::tools::SpawnSubagentTool; - - // Idempotent — other tests may have already initialised it. - AgentDefinitionRegistry::init_global_builtins().unwrap(); - - let workspace = tempfile::TempDir::new().expect("temp workspace"); - let workspace_path = workspace.path().to_path_buf(); - - // Provider response queue, consumed in order: - // 1. Parent turn iter 0 — emit spawn_subagent with mode=fork. - // 2. Fork sub-agent iter 0 — return "X is Y" (no tool calls). - // 3. Parent turn iter 1 — fold the forked result into the final - // text the user sees. - let provider = Box::new(MockProvider { - responses: Mutex::new(vec![ - crate::openhuman::providers::ChatResponse { - text: Some(String::new()), - tool_calls: vec![crate::openhuman::providers::ToolCall { - id: "call-fork".into(), - name: "spawn_subagent".into(), - arguments: serde_json::json!({ - // agent_id is still required by the schema even - // though `mode=fork` overrides the lookup to the - // synthetic `fork` definition. - "agent_id": "researcher", - "mode": "fork", - "prompt": "analyse branch X" - }) - .to_string(), - }], - usage: None, - }, - crate::openhuman::providers::ChatResponse { - text: Some("X is Y".into()), - tool_calls: vec![], - usage: None, - }, - crate::openhuman::providers::ChatResponse { - text: Some("Based on the research, X is Y.".into()), - tool_calls: vec![], - usage: None, - }, - ]), - }); - - let memory_cfg = crate::openhuman::config::MemoryConfig { - backend: "none".into(), - ..crate::openhuman::config::MemoryConfig::default() - }; - let mem: Arc = - Arc::from(crate::openhuman::memory::create_memory(&memory_cfg, &workspace_path).unwrap()); - - let tools: Vec> = vec![Box::new(SpawnSubagentTool::new())]; - - let mut agent = Agent::builder() - .provider(provider) - .tools(tools) - .memory(mem) - .tool_dispatcher(Box::new(NativeToolDispatcher)) - .workspace_dir(workspace_path) - .build() - .unwrap(); - - let response = agent.turn("tell me about X").await.unwrap(); - assert_eq!(response, "Based on the research, X is Y."); - - // Same history assertions as the typed path — the fork runner - // still threads its compact output back through the parent's tool - // result pipeline. - let has_spawn_call = agent.history().iter().any(|msg| match msg { - ConversationMessage::AssistantToolCalls { tool_calls, .. } => { - tool_calls.iter().any(|c| c.name == "spawn_subagent") - } - _ => false, - }); - assert!( - has_spawn_call, - "parent history should contain the spawn_subagent assistant tool call" - ); - - let tool_result_contains_subagent_output = agent.history().iter().any(|msg| match msg { - ConversationMessage::ToolResults(results) => { - results.iter().any(|r| r.content.contains("X is Y")) - } - ConversationMessage::Chat(chat) if chat.role == "tool" => chat.content.contains("X is Y"), - _ => false, - }); - assert!( - tool_result_contains_subagent_output, - "parent history should contain a tool-result entry with the fork sub-agent's output" - ); -} - /// KV-cache invariant: across multiple turns in the same session, the /// system-prompt bytes submitted to the provider must be byte-identical, /// and the model name must not flip. Both are required for the backend's diff --git a/src/openhuman/agent/harness/session/turn.rs b/src/openhuman/agent/harness/session/turn.rs index 1f230c7b4..b234b19a5 100644 --- a/src/openhuman/agent/harness/session/turn.rs +++ b/src/openhuman/agent/harness/session/turn.rs @@ -8,11 +8,9 @@ //! the context pipeline (tool-result budget → microcompact → //! autocompact signal → session-memory extraction trigger). //! - [`Agent::execute_tool_call`] / [`Agent::execute_tools`] — the -//! per-call runners, including the fork-cache `ForkContext` stash -//! for `spawn_subagent { mode: "fork" }` invocations. -//! - [`Agent::build_parent_execution_context`] / -//! [`Agent::build_fork_context`] — snapshot helpers for sub-agent -//! task-locals. +//! per-call runners. +//! - [`Agent::build_parent_execution_context`] — snapshot helper for +//! the parent-context task-local that sub-agents read. //! - [`Agent::trim_history`], [`Agent::fetch_learned_context`], //! [`Agent::build_system_prompt`] — the small helpers `turn()` leans //! on every call. @@ -862,25 +860,6 @@ impl Agent { log::info!("[agent] executing tool: {}", call.name); log::info!("[agent_loop] tool start name={}", call.name); - // Special-case `spawn_subagent { mode: "fork", … }`: stash a - // ForkContext task-local so the sub-agent runner can replay the - // parent's exact rendered prompt + tool schemas + message prefix - // for backend prefix-cache reuse. The branch is taken before - // executing the tool so the task-local is visible inside - // `tool.execute(...)`. - let fork_context_for_call = if call.name == "spawn_subagent" - && call - .arguments - .get("mode") - .and_then(|v| v.as_str()) - .map(|s| s.eq_ignore_ascii_case("fork")) - .unwrap_or(false) - { - Some(self.build_fork_context(call)) - } else { - None - }; - let (raw_result, success) = if !self.visible_tool_names.is_empty() && !self.visible_tool_names.contains(&call.name) { @@ -900,12 +879,9 @@ impl Agent { // implementation which forwards to `execute`. let prefer_markdown = self.context.prefer_markdown_tool_output(); let options = ToolCallOptions { prefer_markdown }; - let exec = tool.execute_with_options(call.arguments.clone(), options); - let outcome = if let Some(fork_ctx) = fork_context_for_call { - harness::with_fork_context(fork_ctx, exec).await - } else { - exec.await - }; + let outcome = tool + .execute_with_options(call.arguments.clone(), options) + .await; match outcome { Ok(r) => { if !r.is_error { @@ -1094,39 +1070,6 @@ impl Agent { } } - /// Build a [`harness::ForkContext`] capturing the parent's - /// rendered system prompt + tool schemas + message prefix at the - /// moment a `spawn_subagent { mode: "fork", … }` call fires. - /// - /// The system prompt is pulled from `history[0]` (the agent always - /// stores its rendered system prompt as the first message). The - /// message prefix is the entire current history rendered through - /// the dispatcher — the *same* sequence the parent's next call - /// would send, except the new fork directive replaces the parent's - /// next continuation. - pub(super) fn build_fork_context(&self, call: &ParsedToolCall) -> harness::ForkContext { - let messages = self.tool_dispatcher.to_provider_messages(&self.history); - let system_prompt: String = messages - .first() - .filter(|m| m.role == "system") - .map(|m| m.content.clone()) - .unwrap_or_default(); - - let fork_task_prompt = call - .arguments - .get("prompt") - .and_then(|v| v.as_str()) - .unwrap_or_default() - .to_string(); - - harness::ForkContext { - system_prompt: Arc::new(system_prompt), - tool_specs: Arc::clone(&self.visible_tool_specs), - message_prefix: Arc::new(messages), - fork_task_prompt, - } - } - // ───────────────────────────────────────────────────────────────── // History & prompt helpers // ───────────────────────────────────────────────────────────────── diff --git a/src/openhuman/agent/harness/session/turn_tests.rs b/src/openhuman/agent/harness/session/turn_tests.rs index ad4c17ec7..f5618e2db 100644 --- a/src/openhuman/agent/harness/session/turn_tests.rs +++ b/src/openhuman/agent/harness/session/turn_tests.rs @@ -234,24 +234,6 @@ fn trim_history_preserves_system_and_keeps_latest_non_system_entries() { .any(|msg| matches!(msg, ConversationMessage::Chat(chat) if chat.content == "a2"))); } -#[test] -fn build_fork_context_uses_visible_specs_and_prompt_argument() { - let mut visible = HashSet::new(); - visible.insert("echo".to_string()); - let agent = make_agent(Some(visible)); - let call = ParsedToolCall { - name: "spawn_subagent".into(), - arguments: serde_json::json!({ "prompt": "fork task" }), - tool_call_id: None, - }; - - let fork = agent.build_fork_context(&call); - assert_eq!(fork.fork_task_prompt, "fork task"); - assert_eq!(fork.tool_specs.len(), 1); - assert_eq!(fork.tool_specs[0].name, "echo"); - assert_eq!(fork.message_prefix.len(), 0); -} - #[test] fn build_parent_context_and_sanitize_helpers_cover_snapshot_paths() { let mut agent = make_agent(None); diff --git a/src/openhuman/agent/harness/subagent_runner/ops.rs b/src/openhuman/agent/harness/subagent_runner/ops.rs index fac0c8be6..bc7499d00 100644 --- a/src/openhuman/agent/harness/subagent_runner/ops.rs +++ b/src/openhuman/agent/harness/subagent_runner/ops.rs @@ -1,22 +1,17 @@ //! Sub-agent execution entry points and the inner tool-call loop. //! //! The public runner lives in [`run_subagent`]. It dispatches to -//! [`run_typed_mode`] (narrow prompt + filtered tools) or -//! [`run_fork_mode`] (prefix-replay) depending on the -//! [`super::types::SubagentMode`] implied by the -//! [`crate::openhuman::agent::harness::definition::AgentDefinition`]. -//! -//! Both modes delegate to [`run_inner_loop`] which drives provider -//! calls and tool execution until the model returns without further -//! tool calls (or the iteration budget is exhausted). +//! [`run_typed_mode`] (narrow prompt + filtered tools) which builds a +//! brand-new system prompt and a filtered tool list for the requested +//! archetype, then drives provider calls and tool execution until the +//! model returns without further tool calls (or the iteration budget +//! is exhausted). use std::collections::HashSet; use std::sync::Arc; use std::time::Instant; -use super::super::fork_context::{ - current_fork, current_parent, ForkContext, ParentExecutionContext, -}; +use super::super::fork_context::{current_parent, ParentExecutionContext}; use super::super::session::transcript; use super::extract_tool::ExtractFromResultTool; use super::handoff::{ @@ -74,7 +69,7 @@ impl LazyToolkitResolver { /// This is the primary entry point for agent delegation. It performs the following: /// 1. Resolves the [`ParentExecutionContext`] task-local. /// 2. Generates a unique `task_id` if one wasn't provided. -/// 3. Dispatches to either `run_fork_mode` or `run_typed_mode` based on the definition. +/// 3. Dispatches to `run_typed_mode`. /// /// On success returns a [`SubagentRunOutcome`] whose `output` is the /// final assistant text. On failure the error is suitable for stringifying @@ -105,12 +100,7 @@ pub async fn run_subagent( // Write/Admin slugs under `ReadOnly`) read it via // `current_sandbox_mode()`; tools that don't care just ignore it. let outcome = with_current_sandbox_mode(definition.sandbox_mode, async { - if definition.uses_fork_context { - let fork = current_fork().ok_or(SubagentRunError::NoForkContext)?; - run_fork_mode(definition, task_prompt, &options, &parent, &fork, &task_id).await - } else { - run_typed_mode(definition, task_prompt, &options, &parent, &task_id).await - } + run_typed_mode(definition, task_prompt, &options, &parent, &task_id).await }) .await?; @@ -755,112 +745,6 @@ async fn run_typed_mode( }) } -// ───────────────────────────────────────────────────────────────────────────── -// Fork mode — replay parent's bytes for prefix-cache reuse -// ───────────────────────────────────────────────────────────────────────────── - -/// Execute a sub-agent in "Fork" mode. -/// -/// This mode is an optimization. It replays the parent's EXACT rendered prompt -/// and history prefix up to the point of delegation. This allows the inference -/// server to reuse its existing KV-cache for the prefix, drastically reducing -/// first-token latency and token costs for parallel delegation. -async fn run_fork_mode( - definition: &AgentDefinition, - _task_prompt: &str, - _options: &SubagentRunOptions, - parent: &ParentExecutionContext, - fork: &ForkContext, - task_id: &str, -) -> Result { - let started = Instant::now(); - - // The fork's task prompt comes from the ForkContext (set by the - // parent's tool-dispatch site), not from the spawn_subagent args - // directly. This guarantees the bytes the parent committed to are - // what the child sees. - let fork_task_prompt = fork.fork_task_prompt.clone(); - - tracing::debug!( - agent_id = %definition.id, - prefix_len = fork.message_prefix.len(), - "[subagent_runner:fork] replaying parent prefix" - ); - - // History = parent's exact prefix (which already starts with the - // parent's system message), then the new fork directive as a user - // message. The system_prompt arc is unused here because the prefix - // already contains the system message at index 0 — but we sanity- - // check that invariant. - debug_assert!( - fork.message_prefix - .first() - .map(|m| m.role == "system") - .unwrap_or(false), - "fork message_prefix must start with the parent's system message" - ); - let mut history: Vec = (*fork.message_prefix).clone(); - history.push(ChatMessage::user(fork_task_prompt)); - - // Fork mode keeps the parent's exact tool schema snapshot so the - // request body matches the prefix the backend has already cached. - // Runtime execution still resolves against the parent's live tool - // registry. - // - // Sub-agents (including fork-mode ones) must not spawn their own - // sub-agents — the rule that applies in `run_typed_mode`'s filter - // applies here too. We keep `spawn_subagent` / `delegate_*` in - // `fork.tool_specs` so the prefix bytes still match the parent's - // cached body (mutating the specs would defeat the whole point of - // fork mode), and instead drop them from `allowed_names` so the - // runtime rejects any attempt to call them with the usual - // "not in allowlist" path. - let allowed_names: HashSet = parent - .all_tools - .iter() - .map(|t| t.name().to_string()) - .filter(|name| !is_subagent_spawn_tool(name) && name != "spawn_worker_thread") - .collect(); - - let model = parent.model_name.clone(); - let temperature = parent.temperature; - // Use the parent's iteration cap, not the synthetic fork definition's. - let max_iterations = parent.agent_config.max_tool_iterations.max(1); - - // Fork mode replays the parent's exact tool list — no dynamic - // toolkit-scoped tools, so `extra_tools` is empty. - let fork_extra_tools: Vec> = Vec::new(); - // Transcript persistence happens per-iteration inside - // `run_inner_loop`; no post-loop write needed. - let (output, iterations, _agg_usage) = run_inner_loop( - parent.provider.as_ref(), - &mut history, - &parent.all_tools, - fork_extra_tools, - fork.tool_specs.as_slice(), - allowed_names, - None, - &model, - temperature, - max_iterations, - task_id, - &definition.id, - None, - None, - parent, - ) - .await?; - - Ok(SubagentRunOutcome { - task_id: task_id.to_string(), - agent_id: definition.id.clone(), - output, - iterations, - elapsed: started.elapsed(), - mode: SubagentMode::Fork, - }) -} - // ───────────────────────────────────────────────────────────────────────────── // Inner tool-call loop (slim version of agent::loop_::tool_loop) // ───────────────────────────────────────────────────────────────────────────── @@ -986,7 +870,7 @@ async fn run_inner_loop( if force_text_mode { // Append the XML tool protocol + available-tool list to the // existing system prompt. `history[0]` is the system message - // built by `run_typed_mode` / `run_fork_mode` upstream; we + // built by `run_typed_mode` upstream; we // augment it in-place so the model learns the call format for // this session without an extra message round-trip. if let Some(sys) = history.iter_mut().find(|m| m.role == "system") { diff --git a/src/openhuman/agent/harness/subagent_runner/ops_tests.rs b/src/openhuman/agent/harness/subagent_runner/ops_tests.rs index b260a4f20..87abd6cab 100644 --- a/src/openhuman/agent/harness/subagent_runner/ops_tests.rs +++ b/src/openhuman/agent/harness/subagent_runner/ops_tests.rs @@ -23,7 +23,6 @@ fn make_def_named_tools(names: &[&str]) -> AgentDefinition { timeout_secs: None, sandbox_mode: crate::openhuman::agent::harness::definition::SandboxMode::None, background: false, - uses_fork_context: false, subagents: vec![], delegate_name: None, source: crate::openhuman::agent::harness::definition::DefinitionSource::Builtin, @@ -114,12 +113,11 @@ fn filter_skill_filter_combined_with_named_scope() { #[test] fn subagent_mode_as_str_roundtrip() { assert_eq!(SubagentMode::Typed.as_str(), "typed"); - assert_eq!(SubagentMode::Fork.as_str(), "fork"); } // ── End-to-end runner tests with mock provider ──────────────────────── -use crate::openhuman::agent::harness::fork_context::{with_fork_context, with_parent_context}; +use crate::openhuman::agent::harness::fork_context::with_parent_context; use crate::openhuman::providers::{ChatRequest as PChatRequest, ChatResponse, Provider, ToolCall}; use parking_lot::Mutex; use std::sync::Arc; @@ -536,76 +534,6 @@ async fn typed_mode_blocks_unallowed_tool_calls() { ); } -#[tokio::test] -async fn fork_mode_replays_parent_prefix_bytes() { - // Construct a fake fork context with a known message prefix. - // The runner should replay it byte-for-byte plus a single - // appended user message carrying the fork directive. - let provider = ScriptedProvider::new(vec![text_response("fork done")]); - let parent = make_parent(provider.clone(), vec![stub("file_read"), stub("shell")]); - - let prefix = vec![ - crate::openhuman::providers::ChatMessage::system("PARENT_SYSTEM_PROMPT_BYTES"), - crate::openhuman::providers::ChatMessage::user("first user msg"), - crate::openhuman::providers::ChatMessage::assistant("parent assistant"), - ]; - - let fork = ForkContext { - system_prompt: Arc::new("PARENT_SYSTEM_PROMPT_BYTES".into()), - tool_specs: Arc::new(vec![parent.all_tool_specs[0].clone()]), - message_prefix: Arc::new(prefix.clone()), - fork_task_prompt: "ANALYSE THIS BRANCH".into(), - }; - - let def = crate::openhuman::agent::harness::builtin_definitions::fork_definition(); - - let outcome = with_parent_context(parent, async move { - with_fork_context(fork, async { - run_subagent( - &def, - "ignored — fork uses fork_task_prompt", - SubagentRunOptions::default(), - ) - .await - }) - .await - }) - .await - .expect("fork runner should succeed"); - - assert_eq!(outcome.mode, SubagentMode::Fork); - assert_eq!(outcome.output, "fork done"); - - // Verify the request that hit the provider replays the parent - // prefix exactly and appends only the fork directive. - let captured = provider.captured.lock(); - let first_call = &captured[0]; - assert_eq!(first_call.messages.len(), prefix.len() + 1); - for (i, msg) in prefix.iter().enumerate() { - assert_eq!(first_call.messages[i].role, msg.role); - assert_eq!(first_call.messages[i].content, msg.content); - } - // The appended user message carries the fork directive. - let appended = first_call.messages.last().unwrap(); - assert_eq!(appended.role, "user"); - assert_eq!(appended.content, "ANALYSE THIS BRANCH"); - assert_eq!(first_call.tool_count, 1); -} - -#[tokio::test] -async fn fork_mode_errors_when_no_fork_context() { - let provider = ScriptedProvider::new(vec![text_response("unused")]); - let parent = make_parent(provider, vec![stub("file_read")]); - let def = crate::openhuman::agent::harness::builtin_definitions::fork_definition(); - - let result = with_parent_context(parent, async { - run_subagent(&def, "x", SubagentRunOptions::default()).await - }) - .await; - - assert!(matches!(result, Err(SubagentRunError::NoForkContext))); -} - #[tokio::test] async fn runner_errors_outside_parent_context() { let def = make_def_named_tools(&[]); diff --git a/src/openhuman/agent/harness/subagent_runner/types.rs b/src/openhuman/agent/harness/subagent_runner/types.rs index 6b933f345..6d5958138 100644 --- a/src/openhuman/agent/harness/subagent_runner/types.rs +++ b/src/openhuman/agent/harness/subagent_runner/types.rs @@ -56,19 +56,21 @@ pub struct SubagentRunOutcome { } /// Which prompt-construction path the runner took for a sub-agent. +/// +/// Currently the only supported mode is `Typed` (narrow, archetype-specific +/// prompt with filtered tools). Kept as an enum so future modes (e.g. +/// background/swarm) can land without churning every call site that records +/// the mode for telemetry. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum SubagentMode { /// Built a narrow, archetype-specific prompt with filtered tools. Typed, - /// Replayed the parent's exact rendered prompt and history prefix. - Fork, } impl SubagentMode { pub fn as_str(self) -> &'static str { match self { Self::Typed => "typed", - Self::Fork => "fork", } } } @@ -80,12 +82,6 @@ pub enum SubagentRunError { #[error("spawn_subagent called outside of an agent turn — no parent context available")] NoParentContext, - #[error( - "fork-mode sub-agent requested but no ForkContext is set on the task-local. \ - Did the parent agent forget to call `Agent::turn` with fork support?" - )] - NoForkContext, - #[error("agent definition '{0}' not found in registry")] DefinitionNotFound(String), diff --git a/src/openhuman/agent/harness/tool_loop.rs b/src/openhuman/agent/harness/tool_loop.rs index e16f9ede9..f10dda515 100644 --- a/src/openhuman/agent/harness/tool_loop.rs +++ b/src/openhuman/agent/harness/tool_loop.rs @@ -1,5 +1,7 @@ +use crate::openhuman::agent::cost::TurnCost; use crate::openhuman::agent::multimodal; use crate::openhuman::agent::progress::AgentProgress; +use crate::openhuman::agent::stop_hooks::{current_stop_hooks, StopDecision, TurnState}; use crate::openhuman::approval::{ApprovalManager, ApprovalRequest, ApprovalResponse}; use crate::openhuman::providers::{ ChatMessage, ChatRequest, Provider, ProviderCapabilityError, ProviderDelta, @@ -151,6 +153,7 @@ pub(crate) async fn run_tool_call_loop( ); let mut context_guard = ContextGuard::new(); + let mut turn_cost = TurnCost::new(); // Announce turn start to progress subscribers (if any). We use // `send().await` for lifecycle (turn/iteration) events so they @@ -163,6 +166,7 @@ pub(crate) async fn run_tool_call_loop( } } + let stop_hooks = current_stop_hooks(); for iteration in 0..max_iterations { if let Some(ref sink) = on_progress { if let Err(e) = sink @@ -175,6 +179,31 @@ pub(crate) async fn run_tool_call_loop( log::warn!("[agent_loop] progress sink closed at IterationStarted: {e}"); } } + + // ── Stop hooks: policy check before the next LLM call ── + if !stop_hooks.is_empty() { + let state = TurnState { + iteration: (iteration + 1) as u32, + max_iterations: max_iterations as u32, + cost: &turn_cost, + model, + }; + for hook in &stop_hooks { + match hook.check(&state).await { + StopDecision::Continue => {} + StopDecision::Stop { reason } => { + tracing::warn!( + iteration = (iteration + 1), + hook = hook.name(), + reason = %reason, + "[agent_loop] stop hook triggered — aborting turn" + ); + anyhow::bail!("Agent turn stopped by hook '{}': {reason}", hook.name()); + } + } + } + } + // ── Context guard: check utilization before each LLM call ── match context_guard.check() { ContextCheckResult::Ok => {} @@ -311,13 +340,30 @@ pub(crate) async fn run_tool_call_loop( // Update context guard with token usage from this response. if let Some(ref usage) = resp.usage { context_guard.update_usage(usage); + turn_cost.add_call(model, usage); tracing::debug!( iteration, input_tokens = usage.input_tokens, output_tokens = usage.output_tokens, context_window = usage.context_window, + cumulative_usd = turn_cost.total_usd(), "[agent_loop] LLM response received" ); + if let Some(ref sink) = on_progress { + let event = AgentProgress::TurnCostUpdated { + model: model.to_string(), + iteration: (iteration + 1) as u32, + input_tokens: turn_cost.input_tokens, + output_tokens: turn_cost.output_tokens, + cached_input_tokens: turn_cost.cached_input_tokens, + total_usd: turn_cost.total_usd(), + }; + if let Err(e) = sink.send(event).await { + log::warn!( + "[agent_loop] progress sink closed at TurnCostUpdated: {e}" + ); + } + } } else { tracing::debug!( iteration, @@ -407,6 +453,15 @@ pub(crate) async fn run_tool_call_loop( } } history.push(ChatMessage::assistant(response_text.clone())); + log::info!( + "[agent_loop] turn complete: iters={} provider_calls={} tokens_in={} tokens_out={} cached_in={} usd={:.4}", + (iteration + 1), + turn_cost.call_count, + turn_cost.input_tokens, + turn_cost.output_tokens, + turn_cost.cached_input_tokens, + turn_cost.total_usd(), + ); if let Some(ref sink) = on_progress { if let Err(e) = sink .send(AgentProgress::TurnCompleted { @@ -599,38 +654,69 @@ pub(crate) async fn run_tool_call_loop( ); scrubbed = compacted; } - if let Some(summarizer) = payload_summarizer { - log::debug!( - "[agent_loop] payload_summarizer intercepting tool={} bytes={}", - call.name, - scrubbed.len() - ); - match summarizer - .maybe_summarize(&call.name, None, &scrubbed) - .await - { - Ok(Some(payload)) => { - log::info!( - "[agent_loop] payload_summarizer compressed tool={} {}->{} bytes", - call.name, - payload.original_bytes, - payload.summary_bytes - ); - scrubbed = payload.summary; - } - Ok(None) => { - log::debug!( - "[agent_loop] payload_summarizer pass-through tool={} bytes={}", - call.name, - scrubbed.len() - ); - } - Err(e) => { - log::warn!( - "[agent_loop] payload_summarizer error tool={} err={} (passing raw payload through)", - call.name, - e - ); + + // Per-tool max_result_size_chars cap. When + // a tool sets it and the (post-tokenjuice) + // body still exceeds the cap, truncate + // here and skip the global payload + // summarizer for this call — the cap is + // fast and deterministic, the summarizer + // is the fallback for tools that don't + // know their own size budget. + let mut hit_per_tool_cap = false; + if let Some(cap) = tool.max_result_size_chars() { + let char_count = scrubbed.chars().count(); + if char_count > cap { + let truncated: String = scrubbed.chars().take(cap).collect(); + let dropped = char_count - cap; + log::info!( + "[agent_loop] per-tool cap applied tool={} cap_chars={} original_chars={} dropped_chars={}", + call.name, + cap, + char_count, + dropped, + ); + scrubbed = format!( + "{truncated}\n\n[truncated by tool cap: {dropped} more chars not shown]" + ); + hit_per_tool_cap = true; + } + } + + if !hit_per_tool_cap { + if let Some(summarizer) = payload_summarizer { + log::debug!( + "[agent_loop] payload_summarizer intercepting tool={} bytes={}", + call.name, + scrubbed.len() + ); + match summarizer + .maybe_summarize(&call.name, None, &scrubbed) + .await + { + Ok(Some(payload)) => { + log::info!( + "[agent_loop] payload_summarizer compressed tool={} {}->{} bytes", + call.name, + payload.original_bytes, + payload.summary_bytes + ); + scrubbed = payload.summary; + } + Ok(None) => { + log::debug!( + "[agent_loop] payload_summarizer pass-through tool={} bytes={}", + call.name, + scrubbed.len() + ); + } + Err(e) => { + log::warn!( + "[agent_loop] payload_summarizer error tool={} err={} (passing raw payload through)", + call.name, + e + ); + } } } } diff --git a/src/openhuman/agent/harness/tool_loop_tests.rs b/src/openhuman/agent/harness/tool_loop_tests.rs index 87389b5d2..df93325e6 100644 --- a/src/openhuman/agent/harness/tool_loop_tests.rs +++ b/src/openhuman/agent/harness/tool_loop_tests.rs @@ -667,3 +667,222 @@ async fn run_tool_call_loop_propagates_provider_errors_and_max_iteration_failure .to_string() .contains("Agent exceeded maximum tool iterations (1)")); } + +#[tokio::test] +async fn run_tool_call_loop_aborts_when_stop_hook_returns_stop() { + use crate::openhuman::agent::stop_hooks::{with_stop_hooks, StopDecision, StopHook, TurnState}; + use std::sync::atomic::{AtomicU32, Ordering}; + use std::sync::Arc; + + /// Stops the loop on the second iteration (1-based). + struct StopOnIteration(Arc); + + #[async_trait] + impl StopHook for StopOnIteration { + fn name(&self) -> &str { + "test-iter-cap" + } + async fn check(&self, ctx: &TurnState<'_>) -> StopDecision { + self.0.store(ctx.iteration, Ordering::Relaxed); + if ctx.iteration >= 2 { + StopDecision::Stop { + reason: "tripped on iter 2".into(), + } + } else { + StopDecision::Continue + } + } + } + + // Provider would happily loop forever — first response asks for a + // tool, second response would too (we never reach it because the + // stop hook fires at the top of iteration 2). + let provider = ScriptedProvider { + responses: Mutex::new(vec![ + Ok(ChatResponse { + text: Some("{\"name\":\"echo\",\"arguments\":{}}".into()), + tool_calls: vec![], + usage: None, + }), + Ok(ChatResponse { + text: Some("{\"name\":\"echo\",\"arguments\":{}}".into()), + tool_calls: vec![], + usage: None, + }), + ]), + native_tools: false, + vision: false, + }; + let mut history = vec![ChatMessage::user("loop me")]; + let tools: Vec> = vec![Box::new(EchoTool)]; + let last_seen = Arc::new(AtomicU32::new(0)); + let hook: Arc = Arc::new(StopOnIteration(last_seen.clone())); + + let err = with_stop_hooks(vec![hook], async { + run_tool_call_loop( + &provider, + &mut history, + &tools, + "test-provider", + "model", + 0.0, + true, + None, + "channel", + &crate::openhuman::config::MultimodalConfig::default(), + 10, + None, + None, + &[], + None, + None, + ) + .await + }) + .await + .expect_err("stop hook should abort the loop"); + + assert!( + err.to_string().contains("stopped by hook 'test-iter-cap'"), + "got: {err}" + ); + assert!( + err.to_string().contains("tripped on iter 2"), + "stop reason should be propagated, got: {err}" + ); + assert_eq!( + last_seen.load(Ordering::Relaxed), + 2, + "hook should have observed iteration 2" + ); +} + +#[tokio::test] +async fn run_tool_call_loop_runs_unchanged_when_no_stop_hooks_installed() { + // Sanity: with no `with_stop_hooks` scope, the loop behaves + // identically to before this feature landed. + let provider = ScriptedProvider { + responses: Mutex::new(vec![Ok(ChatResponse { + text: Some("done".into()), + tool_calls: vec![], + usage: None, + })]), + native_tools: false, + vision: false, + }; + let mut history = vec![ChatMessage::user("hi")]; + let result = run_tool_call_loop( + &provider, + &mut history, + &[], + "test-provider", + "model", + 0.0, + true, + None, + "channel", + &crate::openhuman::config::MultimodalConfig::default(), + 1, + None, + None, + &[], + None, + None, + ) + .await + .expect("loop should succeed without stop hooks"); + assert_eq!(result, "done"); +} + +#[tokio::test] +async fn run_tool_call_loop_applies_per_tool_max_result_size_cap() { + /// Tool that emits a 200k-char body and declares a 100-char cap + /// via `max_result_size_chars`. The loop should truncate before + /// threading the body into history. + struct CappedHugeTool; + + #[async_trait] + impl Tool for CappedHugeTool { + fn name(&self) -> &str { + "capped_huge" + } + fn description(&self) -> &str { + "emits a giant body but caps itself" + } + fn parameters_schema(&self) -> serde_json::Value { + serde_json::json!({"type": "object"}) + } + async fn execute(&self, _args: serde_json::Value) -> Result { + Ok(ToolResult::success("Z".repeat(200_000))) + } + fn permission_level(&self) -> crate::openhuman::tools::PermissionLevel { + crate::openhuman::tools::PermissionLevel::ReadOnly + } + fn max_result_size_chars(&self) -> Option { + Some(100) + } + } + + let provider = ScriptedProvider { + responses: Mutex::new(vec![ + // Round 1: ask for the tool. + Ok(ChatResponse { + text: Some( + "{\"name\":\"capped_huge\",\"arguments\":{}}".into(), + ), + tool_calls: vec![], + usage: None, + }), + // Round 2: stop. + Ok(ChatResponse { + text: Some("done".into()), + tool_calls: vec![], + usage: None, + }), + ]), + native_tools: false, + vision: false, + }; + let mut history = vec![ChatMessage::user("call the tool")]; + let tools: Vec> = vec![Box::new(CappedHugeTool)]; + + let result = run_tool_call_loop( + &provider, + &mut history, + &tools, + "test-provider", + "model", + 0.0, + true, + None, + "channel", + &crate::openhuman::config::MultimodalConfig::default(), + 2, + None, + None, + &[], + None, + None, + ) + .await + .expect("loop with capped tool should succeed"); + assert_eq!(result, "done"); + + // Tool-results message should contain the truncation marker and + // be far smaller than the 200k raw body (the 100-char cap plus a + // small marker, well under 1k bytes total for this one call). + let tool_results = history + .iter() + .find(|msg| msg.role == "user" && msg.content.contains("[Tool results]")) + .expect("tool results should be appended to history"); + assert!( + tool_results.content.contains("[truncated by tool cap:"), + "expected truncation marker, got body: {}", + &tool_results.content[..tool_results.content.len().min(200)] + ); + assert!( + tool_results.content.len() < 1_000, + "raw 200k payload should not appear in history (got {} bytes)", + tool_results.content.len() + ); +} diff --git a/src/openhuman/agent/mod.rs b/src/openhuman/agent/mod.rs index 103a3cc6e..f5d5aef48 100644 --- a/src/openhuman/agent/mod.rs +++ b/src/openhuman/agent/mod.rs @@ -20,6 +20,7 @@ pub mod agents; pub mod bus; +pub mod cost; pub mod debug; pub mod dispatcher; pub mod error; @@ -37,6 +38,7 @@ pub mod progress; /// a thin re-export shim for now. pub mod prompts; mod schemas; +pub mod stop_hooks; pub mod tree_loader; pub mod triage; pub use schemas::{ diff --git a/src/openhuman/agent/progress.rs b/src/openhuman/agent/progress.rs index 362fd363c..3450114d5 100644 --- a/src/openhuman/agent/progress.rs +++ b/src/openhuman/agent/progress.rs @@ -56,9 +56,9 @@ pub enum AgentProgress { SubagentSpawned { agent_id: String, task_id: String, - /// Resolved spawn mode — `"typed"` or `"fork"`. The UI uses this - /// to distinguish narrow-prompt delegations from prefix-replay - /// forks when labelling the live subagent block. + /// Resolved spawn mode — currently always `"typed"`. Kept as a + /// string so future modes (e.g. background/swarm) can land + /// without changing the event shape. mode: String, /// `true` when the spawn was requested with /// `dedicated_thread: true`. The UI links the inline subagent @@ -165,6 +165,30 @@ pub enum AgentProgress { iteration: u32, }, + /// Cumulative cost / token tally for the current turn, emitted + /// after each provider response that carried a usage block. + /// Consumers can render a live "$0.04 · 1.2k in / 480 out" line in + /// the UI without subscribing to provider-level events. + /// + /// `total_usd` prefers backend-reported `charged_amount_usd` + /// (sum of authoritative figures) and falls back to a tier-based + /// token-rate estimate for calls that didn't carry one — see + /// [`crate::openhuman::agent::cost::TurnCost::total_usd`]. + TurnCostUpdated { + /// Last model that contributed to this update. + model: String, + /// 1-based iteration index this update belongs to. + iteration: u32, + /// Cumulative input tokens across the turn. + input_tokens: u64, + /// Cumulative output tokens across the turn. + output_tokens: u64, + /// Cumulative cached prefix input tokens across the turn. + cached_input_tokens: u64, + /// Best-available USD total for the turn so far. + total_usd: f64, + }, + /// The turn completed with a final text response. TurnCompleted { /// Total iterations used. diff --git a/src/openhuman/agent/stop_hooks.rs b/src/openhuman/agent/stop_hooks.rs new file mode 100644 index 000000000..f3ad0d759 --- /dev/null +++ b/src/openhuman/agent/stop_hooks.rs @@ -0,0 +1,294 @@ +//! Mid-turn stop hooks — policy-driven halt of an in-flight agent +//! turn. +//! +//! Distinct from [`super::harness::interrupt::InterruptFence`], which +//! handles user-driven cancellation (Ctrl+C / `/stop`). Stop hooks are +//! the policy lever: budget caps, rate limits, custom kill switches. +//! They run between iterations of the tool-call loop so a runaway +//! turn can be cut short before the next provider call rather than +//! after the fact. +//! +//! ## Wiring +//! +//! Hooks ride on a task-local rather than a parameter on +//! [`crate::openhuman::agent::harness::tool_loop::run_tool_call_loop`] +//! — that signature already takes 16 args and the function is invoked +//! from a dozen+ call sites. The task-local mirrors how +//! [`super::harness::fork_context::PARENT_CONTEXT`] and +//! [`super::harness::sandbox_context::CURRENT_AGENT_SANDBOX_MODE`] are +//! threaded. +//! +//! Callers register hooks via [`with_stop_hooks`] around their +//! [`Agent::run_single`] / `run_interactive` invocation; the loop +//! reads them via [`current_stop_hooks`] and fires them at the top of +//! each iteration. A hook returning [`StopDecision::Stop`] aborts the +//! loop with a [`StoppedByHookError`]-shaped `anyhow` error so the +//! caller can surface the reason to the user. +//! +//! ## Built-in hooks +//! +//! - [`BudgetStopHook`] — caps cumulative turn cost in USD using the +//! [`super::cost::TurnCost`] accumulator. +//! - [`MaxIterationsStopHook`] — caps iteration count from outside the +//! `max_tool_iterations` config (useful for ad-hoc per-call limits +//! without mutating the agent's persistent config). + +use crate::openhuman::agent::cost::TurnCost; +use async_trait::async_trait; +use std::sync::Arc; + +/// A policy hook fired between iterations of the tool-call loop. +#[async_trait] +pub trait StopHook: Send + Sync { + /// Stable name for tracing / error messages (e.g. `"budget"`). + fn name(&self) -> &str; + + /// Inspect the current turn state and decide whether to continue. + async fn check(&self, ctx: &TurnState<'_>) -> StopDecision; +} + +/// Outcome of a single hook check. +#[derive(Debug, Clone)] +pub enum StopDecision { + /// Keep the loop running. + Continue, + /// Stop the loop. `reason` is propagated to the caller. + Stop { reason: String }, +} + +/// Snapshot of the turn at the moment a hook fires. References are +/// borrowed from the loop's locals so hooks pay no allocation cost on +/// the hot path; clone fields out if you need to keep them. +pub struct TurnState<'a> { + /// 1-based iteration index that's about to start. + pub iteration: u32, + /// Configured iteration cap for this turn. + pub max_iterations: u32, + /// Cumulative cost / token tally so far. + pub cost: &'a TurnCost, + /// Model name passed to this turn's provider calls. + pub model: &'a str, +} + +tokio::task_local! { + /// Active stop hooks. `None` (the task-local-not-set state) is + /// treated as "no hooks" — see [`current_stop_hooks`]. + pub static CURRENT_STOP_HOOKS: Vec>; +} + +/// Returns a clone of the currently-installed hook list, or an empty +/// vec when no scope has been entered. +pub fn current_stop_hooks() -> Vec> { + CURRENT_STOP_HOOKS + .try_with(|hooks| hooks.clone()) + .unwrap_or_default() +} + +/// Run `future` with `hooks` installed as the active stop-hook list. +pub async fn with_stop_hooks(hooks: Vec>, future: F) -> R +where + F: std::future::Future, +{ + CURRENT_STOP_HOOKS.scope(hooks, future).await +} + +// ───────────────────────────────────────────────────────────────────────────── +// Built-in hooks +// ───────────────────────────────────────────────────────────────────────────── + +/// Stop the turn once cumulative cost reaches `max_usd`. +/// +/// Uses [`TurnCost::total_usd`] which prefers the backend's +/// `charged_amount_usd` and falls back to a tier-keyed estimate. +#[derive(Debug, Clone, Copy)] +pub struct BudgetStopHook { + pub max_usd: f64, +} + +impl BudgetStopHook { + pub fn new(max_usd: f64) -> Self { + Self { max_usd } + } +} + +#[async_trait] +impl StopHook for BudgetStopHook { + fn name(&self) -> &str { + "budget" + } + + async fn check(&self, ctx: &TurnState<'_>) -> StopDecision { + // Fail closed on a malformed cap: NaN, non-finite, or + // non-positive `max_usd` should *stop* rather than silently + // disable the guard (NaN comparisons always return false, so + // `spent >= NaN` would otherwise let the loop run forever). + if !self.max_usd.is_finite() || self.max_usd <= 0.0 { + return StopDecision::Stop { + reason: format!("invalid budget cap configured: max_usd={}", self.max_usd), + }; + } + let spent = ctx.cost.total_usd(); + if spent >= self.max_usd { + StopDecision::Stop { + reason: format!( + "turn cost ${spent:.4} reached cap ${cap:.4}", + cap = self.max_usd + ), + } + } else { + StopDecision::Continue + } + } +} + +/// Stop the turn at a hard iteration ceiling. +/// +/// Sibling of `max_tool_iterations` on `AgentConfig`; this hook is +/// useful when callers want to lower the limit for one specific turn +/// without mutating the agent's persistent config. +#[derive(Debug, Clone, Copy)] +pub struct MaxIterationsStopHook { + pub cap: u32, +} + +impl MaxIterationsStopHook { + pub fn new(cap: u32) -> Self { + Self { cap } + } +} + +#[async_trait] +impl StopHook for MaxIterationsStopHook { + fn name(&self) -> &str { + "max_iterations" + } + + async fn check(&self, ctx: &TurnState<'_>) -> StopDecision { + if ctx.iteration > self.cap { + StopDecision::Stop { + reason: format!( + "turn reached iteration cap {} (about to start iteration {})", + self.cap, ctx.iteration + ), + } + } else { + StopDecision::Continue + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::openhuman::providers::UsageInfo; + + fn cost_with_usd(usd: f64) -> TurnCost { + let mut tc = TurnCost::new(); + tc.add_call( + "agentic-v1", + &UsageInfo { + charged_amount_usd: usd, + ..Default::default() + }, + ); + tc + } + + #[tokio::test] + async fn budget_hook_continues_under_cap() { + let cost = cost_with_usd(0.10); + let hook = BudgetStopHook::new(1.00); + let ctx = TurnState { + iteration: 1, + max_iterations: 10, + cost: &cost, + model: "agentic-v1", + }; + assert!(matches!(hook.check(&ctx).await, StopDecision::Continue)); + } + + #[tokio::test] + async fn budget_hook_stops_at_cap() { + let cost = cost_with_usd(1.50); + let hook = BudgetStopHook::new(1.00); + let ctx = TurnState { + iteration: 2, + max_iterations: 10, + cost: &cost, + model: "agentic-v1", + }; + match hook.check(&ctx).await { + StopDecision::Stop { reason } => { + assert!(reason.contains("$1.5000")); + assert!(reason.contains("$1.0000")); + } + other => panic!("expected Stop, got {other:?}"), + } + } + + #[tokio::test] + async fn budget_hook_fails_closed_on_nan_cap() { + // NaN comparisons always return false, so without the guard + // `spent >= NaN` would silently disable the cap forever. + let cost = cost_with_usd(1.0); + let hook = BudgetStopHook::new(f64::NAN); + let ctx = TurnState { + iteration: 1, + max_iterations: 10, + cost: &cost, + model: "agentic-v1", + }; + match hook.check(&ctx).await { + StopDecision::Stop { reason } => assert!(reason.contains("invalid budget cap")), + other => panic!("expected Stop on NaN cap, got {other:?}"), + } + } + + #[tokio::test] + async fn budget_hook_fails_closed_on_non_positive_cap() { + let cost = TurnCost::new(); + let ctx = TurnState { + iteration: 1, + max_iterations: 10, + cost: &cost, + model: "agentic-v1", + }; + for bad in [0.0, -1.0, f64::NEG_INFINITY, f64::INFINITY] { + let hook = BudgetStopHook::new(bad); + assert!( + matches!(hook.check(&ctx).await, StopDecision::Stop { .. }), + "cap {bad} should stop" + ); + } + } + + #[tokio::test] + async fn max_iterations_hook_stops_when_exceeded() { + let cost = TurnCost::new(); + let hook = MaxIterationsStopHook::new(3); + let ctx = TurnState { + iteration: 4, + max_iterations: 10, + cost: &cost, + model: "agentic-v1", + }; + assert!(matches!(hook.check(&ctx).await, StopDecision::Stop { .. })); + } + + #[tokio::test] + async fn current_stop_hooks_returns_empty_outside_scope() { + assert!(current_stop_hooks().is_empty()); + } + + #[tokio::test] + async fn with_stop_hooks_installs_visible_within_scope() { + let hooks: Vec> = vec![Arc::new(BudgetStopHook::new(0.5))]; + with_stop_hooks(hooks, async { + let visible = current_stop_hooks(); + assert_eq!(visible.len(), 1); + assert_eq!(visible[0].name(), "budget"); + }) + .await; + assert!(current_stop_hooks().is_empty()); + } +} diff --git a/src/openhuman/channels/providers/web.rs b/src/openhuman/channels/providers/web.rs index f351807c5..0e7c0ea11 100644 --- a/src/openhuman/channels/providers/web.rs +++ b/src/openhuman/channels/providers/web.rs @@ -992,6 +992,23 @@ fn spawn_progress_bridge( client_id={client_id} thread_id={thread_id} request_id={request_id}" ); } + AgentProgress::TurnCostUpdated { + model, + iteration, + input_tokens, + output_tokens, + cached_input_tokens, + total_usd, + } => { + // Cost telemetry — not surfaced to the UI yet, but + // logged at debug for now and ready for a future + // socket payload. + log::debug!( + "[web_channel] turn cost update model={model} iter={iteration} \ + in={input_tokens} out={output_tokens} cached_in={cached_input_tokens} \ + total_usd={total_usd:.4} client_id={client_id} thread_id={thread_id}" + ); + } } } turn_state.finish(); diff --git a/src/openhuman/channels/runtime/dispatch.rs b/src/openhuman/channels/runtime/dispatch.rs index 9ab09df75..310cf876e 100644 --- a/src/openhuman/channels/runtime/dispatch.rs +++ b/src/openhuman/channels/runtime/dispatch.rs @@ -574,7 +574,6 @@ mod scoping_tests { timeout_secs: None, sandbox_mode: SandboxMode::None, background: false, - uses_fork_context: false, subagents: vec![], delegate_name: None, source: DefinitionSource::Builtin, diff --git a/src/openhuman/threads/turn_state/mirror.rs b/src/openhuman/threads/turn_state/mirror.rs index 3798ecb22..0d4ef2601 100644 --- a/src/openhuman/threads/turn_state/mirror.rs +++ b/src/openhuman/threads/turn_state/mirror.rs @@ -316,6 +316,14 @@ impl TurnStateMirror { } true } + AgentProgress::TurnCostUpdated { .. } => { + // Cost updates don't change the turn-state snapshot + // shape (lifecycle / phase / active tool / etc.), so + // we just acknowledge them without flushing. Surfacing + // cost in the persisted snapshot would force a disk + // flush per LLM call — not worth it for telemetry. + false + } } } diff --git a/src/openhuman/tools/impl/agent/spawn_subagent.rs b/src/openhuman/tools/impl/agent/spawn_subagent.rs index f34d4eb37..d0eeb20cd 100644 --- a/src/openhuman/tools/impl/agent/spawn_subagent.rs +++ b/src/openhuman/tools/impl/agent/spawn_subagent.rs @@ -9,14 +9,8 @@ //! into a single text result that the parent receives as a normal //! `tool_result`. //! -//! Modes: -//! - `"typed"` (default) — narrow prompt + filtered tools + cheaper -//! model. Use for delegated work where the parent doesn't need to -//! share its full context. -//! - `"fork"` — replay the parent's *exact* rendered prompt + tool -//! schemas + message prefix. Use for parallel decomposition of a -//! homogeneous task; relies on the inference backend's automatic -//! prefix caching for token savings. +//! Sub-agents always run in "typed" mode: a narrow archetype-specific +//! prompt with a filtered tool list, on a cheaper model where applicable. //! use crate::core::event_bus::{publish_global, DomainEvent}; use crate::openhuman::agent::harness::definition::AgentDefinitionRegistry; @@ -100,7 +94,7 @@ impl Tool for SpawnSubagentTool { let agent_id_schema = if agent_ids.is_empty() { json!({ "type": "string", - "description": "Sub-agent id (e.g. code_executor, researcher, critic, fork)." + "description": "Sub-agent id (e.g. code_executor, researcher, critic)." }) } else { json!({ @@ -132,11 +126,6 @@ impl Tool for SpawnSubagentTool { "type": "string", "description": "Composio toolkit slug to scope this spawn to — e.g. `gmail`, `notion`, `slack`. REQUIRED when `agent_id = \"integrations_agent\"`. Narrows the sub-agent's visible Composio actions AND its Connected Integrations prompt section to only that toolkit's catalogue, so the sub-agent's context window only carries the platform it was asked to operate on. Must match a currently-connected integration (see the Delegation Guide)." }, - "mode": { - "type": "string", - "enum": ["typed", "fork"], - "description": "`typed` (default) builds a narrow prompt + filtered tools. `fork` replays the parent's exact prompt for prefix-cache reuse on the inference backend." - }, "dedicated_thread": { "type": "boolean", "description": "Default `false`. Set `true` ONLY for long, complex sub-tasks where the parent thread should not be flooded with sub-agent output. The sub-agent's prompt and final summary land in a fresh worker-labeled thread the user can open from the thread list, and the parent receives a compact reference (worker thread id + brief summary) instead of the full transcript. Worker threads cannot themselves spawn another worker (sub-agents never see this tool), so this is a one-level-deep escape hatch." @@ -177,8 +166,6 @@ impl Tool for SpawnSubagentTool { .map(|s| s.trim().to_string()) .filter(|s| !s.is_empty()); - let mode = args.get("mode").and_then(|v| v.as_str()).unwrap_or("typed"); - let dedicated_thread = args .get("dedicated_thread") .and_then(|v| v.as_bool()) @@ -205,19 +192,12 @@ impl Tool for SpawnSubagentTool { } }; - // Resolve `mode` against the definition. Explicit `mode` argument - // wins; otherwise we infer from the definition itself. - let lookup_id = if mode == "fork" { - "fork" - } else { - agent_id.as_str() - }; - let definition = match registry.get(lookup_id) { + let definition = match registry.get(agent_id.as_str()) { Some(def) => def, None => { let available: Vec<&str> = registry.list().iter().map(|d| d.id.as_str()).collect(); return Ok(ToolResult::error(format!( - "spawn_subagent: unknown agent_id '{lookup_id}'. Available: {}", + "spawn_subagent: unknown agent_id '{agent_id}'. Available: {}", available.join(", ") ))); } @@ -376,7 +356,7 @@ impl Tool for SpawnSubagentTool { publish_global(DomainEvent::SubagentSpawned { parent_session: parent_session.clone(), agent_id: definition.id.clone(), - mode: mode.to_string(), + mode: "typed".to_string(), task_id: task_id.clone(), prompt_chars: prompt.chars().count(), }); @@ -391,7 +371,7 @@ impl Tool for SpawnSubagentTool { .send(AgentProgress::SubagentSpawned { agent_id: definition.id.clone(), task_id: task_id.clone(), - mode: mode.to_string(), + mode: "typed".to_string(), dedicated_thread, prompt_chars: prompt.chars().count(), }) diff --git a/src/openhuman/tools/impl/filesystem/file_read.rs b/src/openhuman/tools/impl/filesystem/file_read.rs index acbd934fc..3a692c4e6 100644 --- a/src/openhuman/tools/impl/filesystem/file_read.rs +++ b/src/openhuman/tools/impl/filesystem/file_read.rs @@ -40,6 +40,11 @@ impl Tool for FileReadTool { }) } + /// Pure read — safe to fan out across parallel `file_read` calls. + fn is_concurrency_safe(&self, _args: &serde_json::Value) -> bool { + true + } + async fn execute(&self, args: serde_json::Value) -> anyhow::Result { let path = args .get("path") diff --git a/src/openhuman/tools/impl/filesystem/glob_search.rs b/src/openhuman/tools/impl/filesystem/glob_search.rs index ba8304bdc..bf41e04ee 100644 --- a/src/openhuman/tools/impl/filesystem/glob_search.rs +++ b/src/openhuman/tools/impl/filesystem/glob_search.rs @@ -58,6 +58,11 @@ impl Tool for GlobTool { PermissionLevel::ReadOnly } + /// Pure read — safe to fan out across parallel `glob` calls. + fn is_concurrency_safe(&self, _args: &serde_json::Value) -> bool { + true + } + async fn execute(&self, args: serde_json::Value) -> anyhow::Result { let pattern_str = args .get("pattern") diff --git a/src/openhuman/tools/impl/filesystem/grep.rs b/src/openhuman/tools/impl/filesystem/grep.rs index 002f3c8f0..aafad3f54 100644 --- a/src/openhuman/tools/impl/filesystem/grep.rs +++ b/src/openhuman/tools/impl/filesystem/grep.rs @@ -71,6 +71,11 @@ impl Tool for GrepTool { PermissionLevel::ReadOnly } + /// Pure read — safe to fan out across parallel `grep` calls. + fn is_concurrency_safe(&self, _args: &serde_json::Value) -> bool { + true + } + async fn execute(&self, args: serde_json::Value) -> anyhow::Result { let pattern = args .get("pattern") diff --git a/src/openhuman/tools/impl/network/web_fetch.rs b/src/openhuman/tools/impl/network/web_fetch.rs index ad86757fb..50fd1e0d9 100644 --- a/src/openhuman/tools/impl/network/web_fetch.rs +++ b/src/openhuman/tools/impl/network/web_fetch.rs @@ -71,6 +71,22 @@ impl Tool for WebFetchTool { PermissionLevel::ReadOnly } + /// Idempotent GET — safe to fan out across parallel `web_fetch` + /// calls. Targets that throttle aggressively are the user's + /// concern; we don't try to second-guess at the tool layer. + fn is_concurrency_safe(&self, _args: &serde_json::Value) -> bool { + true + } + + /// Cap web_fetch results at ~50k chars before they reach the + /// model. The tool itself already truncates byte-wise via + /// `max_bytes` (default 1MB), but a 1MB HTML page is still tens + /// of thousands of tokens — the agent rarely needs that much, and + /// when it does, `read_file` on a saved copy is the right tool. + fn max_result_size_chars(&self) -> Option { + Some(50_000) + } + async fn execute(&self, args: serde_json::Value) -> anyhow::Result { let raw_url = args .get("url") diff --git a/src/openhuman/tools/impl/system/shell.rs b/src/openhuman/tools/impl/system/shell.rs index 114fe4f28..2aa2be663 100644 --- a/src/openhuman/tools/impl/system/shell.rs +++ b/src/openhuman/tools/impl/system/shell.rs @@ -83,6 +83,15 @@ impl Tool for ShellTool { }) } + /// Cap shell output at ~30k chars before threading into history. + /// Verbose commands (`find /`, dependency installs, log dumps) + /// can otherwise blow past 100k chars in one call. The agent + /// rarely needs the full firehose — a head/tail/grep follow-up is + /// the right move when it does. + fn max_result_size_chars(&self) -> Option { + Some(30_000) + } + async fn execute(&self, args: serde_json::Value) -> anyhow::Result { let command = args .get("command") diff --git a/src/openhuman/tools/orchestrator_tools.rs b/src/openhuman/tools/orchestrator_tools.rs index fc1c61763..2f4dfbdd5 100644 --- a/src/openhuman/tools/orchestrator_tools.rs +++ b/src/openhuman/tools/orchestrator_tools.rs @@ -216,7 +216,6 @@ mod tests { timeout_secs: None, sandbox_mode: SandboxMode::None, background: false, - uses_fork_context: false, subagents: vec![], delegate_name: delegate_name.map(String::from), source: DefinitionSource::Builtin, diff --git a/src/openhuman/tools/traits.rs b/src/openhuman/tools/traits.rs index 16663ed13..620eeec8b 100644 --- a/src/openhuman/tools/traits.rs +++ b/src/openhuman/tools/traits.rs @@ -170,6 +170,48 @@ pub trait Tool: Send + Sync { ToolCategory::System } + /// Whether two concurrent invocations of this tool are safe to + /// run in parallel inside a single LLM iteration. + /// + /// Read-only tools that touch no shared mutable state should + /// return `true` (the agent's tool loop can then `join_all` a + /// batch of read calls instead of awaiting them serially). Tools + /// that mutate the workspace, write to disk, or interact with + /// external services that throttle by caller should leave the + /// default `false`. + /// + /// The argument is provided so a tool can refine the answer per + /// call (e.g. a generic `bash` tool could allow parallel `ls` / + /// `cat` invocations and reject parallel `npm install`s) — most + /// tools will ignore it. + /// + /// **Wiring note:** the parallel dispatcher in + /// `harness::tool_loop` currently runs tool calls serially + /// regardless of this flag. Annotating tools is still load- + /// bearing: it lets the dispatch refactor land without + /// coordinating with every tool author. See the parallel-tool + /// dispatch follow-up issue. + fn is_concurrency_safe(&self, _args: &serde_json::Value) -> bool { + false + } + + /// Per-tool cap on the character length of the result body sent + /// back to the model. + /// + /// When `Some(cap)` and the tool's `output_for_llm` exceeds it, + /// the agent's tool loop truncates the body and appends a marker + /// before threading the value into history — protecting the + /// context window from one chatty tool. When `None` (the + /// default), no per-tool cap applies and the global + /// `PayloadSummarizer` (if any) handles oversize bodies. + /// + /// Set this on tools whose output is *bounded but unpredictable* + /// (`bash`, `web_fetch`, etc.); leave it unset on tools where + /// callers genuinely want full content (`read_file`, `grep`). + fn max_result_size_chars(&self) -> Option { + None + } + /// Get the full spec for LLM registration fn spec(&self) -> ToolSpec { ToolSpec { @@ -269,6 +311,18 @@ mod tests { assert_eq!(tool.category(), ToolCategory::System); } + #[test] + fn default_is_concurrency_safe_is_false() { + let tool = DummyTool; + assert!(!tool.is_concurrency_safe(&serde_json::Value::Null)); + } + + #[test] + fn default_max_result_size_chars_is_none() { + let tool = DummyTool; + assert!(tool.max_result_size_chars().is_none()); + } + // ── PermissionLevel ordering ─────────────────────────────────── #[test] diff --git a/tests/agent_harness_public.rs b/tests/agent_harness_public.rs index 8242e016d..7d7045147 100644 --- a/tests/agent_harness_public.rs +++ b/tests/agent_harness_public.rs @@ -1,8 +1,7 @@ use anyhow::Result; use async_trait::async_trait; use openhuman_core::openhuman::agent::harness::{ - check_interrupt, current_fork, current_parent, with_fork_context, with_parent_context, - ForkContext, InterruptFence, ParentExecutionContext, + check_interrupt, current_parent, with_parent_context, InterruptFence, ParentExecutionContext, }; use openhuman_core::openhuman::agent::hooks::{ fire_hooks, sanitize_tool_output, PostTurnHook, ToolCallRecord, TurnContext, @@ -195,24 +194,8 @@ async fn interrupt_signal_handler_is_installable() { } #[tokio::test] -async fn fork_and_parent_contexts_are_visible_only_within_scope() { +async fn parent_context_is_visible_only_within_scope() { assert!(current_parent().is_none()); - assert!(current_fork().is_none()); - - let fork = ForkContext { - system_prompt: Arc::new("hello".into()), - tool_specs: Arc::new(vec![]), - message_prefix: Arc::new(vec![ChatMessage::system("hello")]), - fork_task_prompt: "do thing".into(), - }; - - with_fork_context(fork, async { - let inner = current_fork().expect("fork context should be visible"); - assert_eq!(*inner.system_prompt, "hello"); - assert_eq!(inner.fork_task_prompt, "do thing"); - assert_eq!(inner.message_prefix.len(), 1); - }) - .await; let parent = stub_parent_context(); with_parent_context(parent, async { @@ -225,7 +208,6 @@ async fn fork_and_parent_contexts_are_visible_only_within_scope() { .await; assert!(current_parent().is_none()); - assert!(current_fork().is_none()); } #[test]