diff --git a/src/bin/harness_subagent_audit.rs b/src/bin/harness_subagent_audit.rs index 53a6320e2d..1e333a7ecb 100644 --- a/src/bin/harness_subagent_audit.rs +++ b/src/bin/harness_subagent_audit.rs @@ -447,6 +447,7 @@ async fn drain_progress( prompt_chars, worker_thread_id, display_name, + .. } => { eprintln!( "[harness_subagent_audit] progress turn={} subagent_spawned agent_id={} task_id={} mode={} dedicated_thread={} prompt_chars={} worker_thread_id={}", @@ -586,7 +587,7 @@ async fn drain_progress( output: _, elapsed_ms, iteration, - failure: _, + .. } => { eprintln!( "[harness_subagent_audit] progress turn={} subagent_tool_completed agent_id={} task_id={} tool={} call_id={} success={} output_chars={} elapsed_ms={} iteration={}", diff --git a/src/openhuman/agent/harness/session/tool_progress.rs b/src/openhuman/agent/harness/session/tool_progress.rs index 83861fa091..3515be0437 100644 --- a/src/openhuman/agent/harness/session/tool_progress.rs +++ b/src/openhuman/agent/harness/session/tool_progress.rs @@ -176,6 +176,10 @@ impl ProgressReporter for TurnProgress { tool_name: tool_name.to_string(), success, output_chars: output.chars().count(), + output: output.to_string(), + // The legacy reporter path has no captured argument + // payload at completion time. + arguments: None, elapsed_ms, iteration, failure: None, diff --git a/src/openhuman/agent/progress.rs b/src/openhuman/agent/progress.rs index a45cb9a596..f6f132ff35 100644 --- a/src/openhuman/agent/progress.rs +++ b/src/openhuman/agent/progress.rs @@ -54,6 +54,18 @@ pub enum AgentProgress { tool_name: String, success: bool, output_chars: usize, + /// Full text the tool returned. Mirrors + /// [`Self::SubagentToolCallCompleted::output`] so trace exporters can + /// record the parent-scope tool result as the span output (truncated + + /// content-gated at the collector). Empty when the harness ran with + /// payload capture off. + output: String, + /// The arguments the tool was invoked with, when the harness captured + /// them (`PayloadCapture::tool_io`). The crate emits arguments on the + /// *completed* event, so trace exporters use this to backfill the tool + /// span's input (the started event's `arguments` is `Null` on the + /// tinyagents path). + arguments: Option, elapsed_ms: u64, /// 1-based iteration index. iteration: u32, @@ -89,6 +101,11 @@ pub enum AgentProgress { /// "Researcher", "Coding Agent"). Falls back to `agent_id` in /// the UI when absent. display_name: Option, + /// The full delegated prompt text. `prompt_chars` stays as the cheap + /// size hint; trace exporters record this (truncated + content-gated) + /// as the subagent span's input so a delegation is inspectable + /// end-to-end in Langfuse. + prompt: String, }, /// A sub-agent completed successfully. @@ -103,6 +120,9 @@ pub enum AgentProgress { iterations: u32, /// Character length of the sub-agent's final assistant text. output_chars: usize, + /// The sub-agent's full final assistant text. Trace exporters record + /// this (truncated + content-gated) as the subagent span's output. + output: String, /// Absolute path to the worker's isolated `git worktree` checkout, /// when it ran with `isolation = "worktree"` (#3376). `None` for /// non-isolated (read-only / shared-workspace) workers. @@ -187,6 +207,11 @@ pub enum AgentProgress { /// actual result/output. `output_chars` is kept as a cheap size hint /// for consumers that only want the length. output: String, + /// The arguments the tool was invoked with, when the harness captured + /// them (`PayloadCapture::tool_io`). Backfills the tool span's input in + /// trace exports (the started event's `arguments` is `Null` on the + /// tinyagents path). + arguments: Option, elapsed_ms: u64, /// 1-based child iteration index. iteration: u32, @@ -292,12 +317,27 @@ pub enum AgentProgress { /// [`Self::TurnCostUpdated`] (cumulative rollup), every field here is /// **per-call**, so trace exporters can render each model invocation as /// its own Langfuse generation with exact model + token + cost figures. - /// Emitted once per parent-scope model call, right after the usage block - /// is recorded; child (subagent) calls stay inside the cumulative rollup - /// because this event carries no task attribution. + /// Emitted once per model call — parent-scope calls carry + /// `subagent_task_id: None`; child (subagent) calls carry the owning + /// task id so exporters can nest the generation under the subagent span. ModelCallCompleted { /// Model that served this call (tier handle or concrete model id). model: String, + /// Provider that served this call (`"managed"`, `"openai"`, + /// `"ollama"`, …). Trace exporters render the Langfuse model as + /// `{provider_id}.{model}` (e.g. `managed.chat-v1`). + provider_id: String, + /// Owning subagent task id when this call ran inside a child run + /// (`spawn_subagent` / Context Scout). `None` for parent-scope calls. + subagent_task_id: Option, + /// The request messages sent to the model (including the system + /// prompt), when the harness captured them + /// (`PayloadCapture::model_io`). Trace exporters record this as the + /// generation's input, gated on `capture_content`. + input: Option, + /// The model completion (assistant message), when captured. Recorded + /// as the generation's output, gated on `capture_content`. + output: Option, /// 1-based iteration index (one model call per iteration). iteration: u32, /// Input/prompt tokens for this call. diff --git a/src/openhuman/agent/progress_tracing.rs b/src/openhuman/agent/progress_tracing.rs index 4681418274..9efa86db20 100644 --- a/src/openhuman/agent/progress_tracing.rs +++ b/src/openhuman/agent/progress_tracing.rs @@ -24,9 +24,12 @@ //! ## Privacy //! //! Spans always carry *metadata* — span names, counts, timings, and -//! token/cost figures. While `observability.agent_tracing.capture_content` is -//! on, the turn's prompt/reply and **truncated** tool arguments/results are -//! additionally recorded as span `input`/`output`; with the flag off (the +//! token/cost figures (model labels are `{provider_id}.{model}`, e.g. +//! `managed.chat-v1`). While `observability.agent_tracing.capture_content` is +//! on, content is additionally recorded as span `input`/`output` — the turn's +//! prompt/reply, each generation's **truncated** request messages (system +//! prompt included) + completion, **truncated** tool arguments/results, and +//! each subagent's delegated prompt + final output. With the flag off (the //! default — #4454), none of that content ever reaches the in-memory span, so //! no exporter (NDJSON file, app log, or Langfuse) can leak it. //! Streamed text/thinking deltas (`TextDelta`, `ThinkingDelta`, @@ -315,15 +318,6 @@ pub struct SpanCollector { /// fresh nonce makes every span id globally unique. id_prefix: String, - /// Storage-level privacy gate (mirrors - /// `observability.agent_tracing.capture_content`). When `false` (the - /// default), prompt/reply content from [`AgentProgress::TurnContent`] is - /// **never attached to a span** — so no exporter (NDJSON file, app log, or - /// Langfuse push) can ever serialize it. This is the single choke point - /// referenced in the module docs: gating at storage protects every present - /// and future exporter, not just the transmission path. - capture_content: bool, - turn_span_id: Option, turn_span_index: Option, current_iteration_span_id: Option, @@ -342,8 +336,6 @@ impl SpanCollector { spans: Vec::new(), next_span_seq: 0, id_prefix: uuid::Uuid::new_v4().simple().to_string(), - // Metadata-only by default; opt in via `with_content_capture`. - capture_content: false, turn_span_id: None, turn_span_index: None, current_iteration_span_id: None, @@ -353,12 +345,14 @@ impl SpanCollector { } } - /// Opt into attaching prompt/reply content to spans (from - /// [`AgentProgress::TurnContent`]). Wire this to - /// `observability.agent_tracing.capture_content`. Left off, content is - /// dropped at storage time so it can never reach any exporter. + /// Opt into attaching content to spans (prompt/reply, generation + /// request/completion, tool + subagent I/O). Wire this to + /// `observability.agent_tracing.capture_content`. Equivalent to setting + /// [`TraceContext::with_capture_content`] before construction — there is a + /// single storage-level gate (`ctx.capture_content`), so content dropped + /// here can never reach any exporter. pub fn with_content_capture(mut self, capture_content: bool) -> Self { - self.capture_content = capture_content; + self.ctx.capture_content = capture_content; self } @@ -549,11 +543,20 @@ impl SpanCollector { /// Fold a per-call `ModelCallCompleted` into the tree: /// /// 1. emit a closed [`SpanKind::Generation`] span (name `llm.`) - /// parented under the current iteration, carrying exact per-call - /// model/usage/cost plus provenance (`gen_ai.provider`) and the pricing - /// basis the local estimator would use; - /// 2. accumulate reasoning / cache-creation tokens onto the root turn - /// span, which `TurnCostUpdated` (cumulative rollup) does not carry. + /// parented under the current iteration — or, for a child call + /// (`subagent_task_id` set), under the owning subagent's current + /// iteration — carrying exact per-call model/usage/cost plus provenance + /// (`gen_ai.provider`) and the pricing basis the local estimator uses; + /// 2. record the captured request messages (incl. the system prompt) and + /// completion as the generation's input/output, gated on + /// `capture_content` and truncated to [`MAX_MODEL_CONTENT_CHARS`]; + /// 3. accumulate reasoning / cache-creation tokens onto the root turn + /// span, which `TurnCostUpdated` (cumulative rollup) does not carry — + /// and, for child calls, roll model + usage onto the subagent span so + /// a delegation (e.g. the Context Scout) surfaces its model natively. + /// + /// The Langfuse-facing model label is `{provider_id}.{model}` (e.g. + /// `managed.chat-v1`, `openai.gpt-4o`). /// /// Generation start is approximated by the enclosing iteration span's /// start (the iteration opens on `ModelStarted`); end is the observation @@ -562,6 +565,10 @@ impl SpanCollector { fn record_model_call( &mut self, model: &str, + provider_id: &str, + subagent_task_id: Option<&str>, + input: Option<&serde_json::Value>, + output: Option<&serde_json::Value>, iteration: u32, input_tokens: u64, output_tokens: u64, @@ -571,24 +578,37 @@ impl SpanCollector { cost_usd: f64, now_unix_ms: u64, ) { - let start_unix_ms = self - .current_iteration_index + // Resolve the parent + start basis: a child call nests under its + // subagent's current iteration (else the subagent span itself); a + // parent call nests under the turn's current iteration (else root). + let subagent_state = subagent_task_id.and_then(|id| self.subagents.get(id)); + let (parent, start_basis_index) = match subagent_state { + Some(state) => match &state.current_iteration_span_id { + Some(id) => { + let idx = self.span_index_by_id(id); + (id.clone(), idx) + } + None => ( + self.spans[state.span_index].span_id.clone(), + Some(state.span_index), + ), + }, + None => { + let parent = self.active_parent_id(now_unix_ms); + (parent, self.current_iteration_index) + } + }; + let start_unix_ms = start_basis_index .and_then(|idx| self.spans.get(idx)) .map(|span| span.start_unix_ms) .unwrap_or(now_unix_ms); - let parent = self.active_parent_id(now_unix_ms); - // Model provenance: managed OpenHuman tier vs custom/BYO model. - let provider_source = if crate::openhuman::agent::cost::is_managed_tier(model) { - "managed" - } else { - "custom" - }; + let labeled_model = format!("{provider_id}.{model}"); let pricing = crate::openhuman::agent::cost::lookup_pricing(model); let mut attrs = BTreeMap::new(); - attrs.insert("gen_ai.request.model".to_string(), json_str(model)); - attrs.insert("gen_ai.provider".to_string(), json_str(provider_source)); + attrs.insert("gen_ai.request.model".to_string(), json_str(&labeled_model)); + attrs.insert("gen_ai.provider".to_string(), json_str(provider_id)); attrs.insert("agent.iteration".to_string(), json_u32(iteration)); attrs.insert( "gen_ai.usage.input_tokens".to_string(), @@ -632,8 +652,12 @@ impl SpanCollector { ); log::debug!( - "[agent-tracing] generation span model={model} provider={provider_source} \ - iteration={iteration} in={input_tokens} out={output_tokens} cost_usd={cost_usd:.6}" + "[agent-tracing] generation span model={labeled_model} \ + iteration={iteration} child={} in={input_tokens} out={output_tokens} \ + cost_usd={cost_usd:.6} input_captured={} output_captured={}", + subagent_task_id.is_some(), + input.is_some(), + output.is_some(), ); let (_, index) = self.open_span( SpanKind::Generation, @@ -642,10 +666,63 @@ impl SpanCollector { start_unix_ms, attrs, ); + // Captured request messages (incl. system prompt) + completion become + // the generation's input/output — only while content capture is on, + // truncated so one huge context window can't bloat the trace batch. + if self.ctx.capture_content { + if let Some(span) = self.spans.get_mut(index) { + if let Some(value) = input { + span.input = Some(capture_model_content(value)); + } + if let Some(value) = output { + span.output = Some(capture_model_content(value)); + } + } + } self.close_span(index, now_unix_ms, SpanStatus::Ok, BTreeMap::new()); + // Child call: roll model + usage + cost onto the owning subagent span + // so the delegation row (e.g. `subagent.Context Scout`) natively shows + // which model served it and what it cost. + if let Some(state_index) = subagent_task_id + .and_then(|id| self.subagents.get(id)) + .map(|state| state.span_index) + { + if let Some(span) = self.spans.get_mut(state_index) { + span.attributes + .insert("gen_ai.request.model".to_string(), json_str(&labeled_model)); + span.attributes + .insert("gen_ai.provider".to_string(), json_str(provider_id)); + for (key, add) in [ + ("gen_ai.usage.input_tokens", input_tokens), + ("gen_ai.usage.output_tokens", output_tokens), + ("gen_ai.usage.cached_input_tokens", cached_input_tokens), + ] { + let prior = span + .attributes + .get(key) + .and_then(serde_json::Value::as_u64) + .unwrap_or(0); + span.attributes + .insert(key.to_string(), json_u64(prior.saturating_add(add))); + } + let prior_cost = span + .attributes + .get("gen_ai.usage.cost_usd") + .and_then(serde_json::Value::as_f64) + .unwrap_or(0.0); + span.attributes.insert( + "gen_ai.usage.cost_usd".to_string(), + json_f64(prior_cost + cost_usd), + ); + } + return; + } + // Root rollup for the usage dimensions the cumulative TurnCostUpdated - // event does not carry (reasoning / cache-creation), plus provenance. + // event does not carry (reasoning / cache-creation), plus provenance + // and the provider-labeled model (TurnCostUpdated only knows the raw + // model handle and can fire before the first per-call event). let root = match self.turn_span_index { Some(idx) => idx, None => { @@ -655,7 +732,9 @@ impl SpanCollector { }; if let Some(span) = self.spans.get_mut(root) { span.attributes - .insert("gen_ai.provider".to_string(), json_str(provider_source)); + .insert("gen_ai.provider".to_string(), json_str(provider_id)); + span.attributes + .insert("gen_ai.request.model".to_string(), json_str(&labeled_model)); for (key, add) in [ ("gen_ai.usage.reasoning_tokens", reasoning_tokens), ("gen_ai.usage.cache_creation_tokens", cache_creation_tokens), @@ -739,11 +818,22 @@ impl SpanCollector { call_id, success, output_chars, + output, + arguments, elapsed_ms, failure, .. } => { if let Some(index) = self.open_tools.remove(call_id) { + // The tinyagents path emits `Null` arguments on the started + // event and the real captured arguments on completion — + // backfill the span input when it's still empty. + if self.spans[index].input.is_none() { + if let Some(arguments) = arguments { + self.capture_tool_arguments(index, arguments); + } + } + self.capture_tool_output(index, output); let start = self.spans[index].start_unix_ms; let mut extra = BTreeMap::new(); extra.insert( @@ -772,6 +862,10 @@ impl SpanCollector { AgentProgress::ModelCallCompleted { model, + provider_id, + subagent_task_id, + input, + output, iteration, input_tokens, output_tokens, @@ -782,6 +876,10 @@ impl SpanCollector { } => { self.record_model_call( model, + provider_id, + subagent_task_id.as_deref(), + input.as_ref(), + output.as_ref(), *iteration, *input_tokens, *output_tokens, @@ -799,6 +897,7 @@ impl SpanCollector { mode, dedicated_thread, prompt_chars, + prompt, display_name, .. } => { @@ -826,6 +925,16 @@ impl SpanCollector { now_unix_ms, attrs, ); + // The delegated prompt is the subagent span's input (gated + + // truncated like model content — scout prompts run 10k+ chars). + if self.ctx.capture_content && !prompt.is_empty() { + if let Some(span) = self.spans.get_mut(index) { + span.input = Some(serde_json::Value::String(truncate_chars( + prompt, + MAX_MODEL_CONTENT_CHARS, + ))); + } + } self.subagents.insert( task_id.clone(), SubagentState { @@ -917,6 +1026,7 @@ impl SpanCollector { success, output_chars, output, + arguments, elapsed_ms, .. } => { @@ -927,6 +1037,11 @@ impl SpanCollector { else { return; }; + if self.spans[index].input.is_none() { + if let Some(arguments) = arguments { + self.capture_tool_arguments(index, arguments); + } + } self.capture_tool_output(index, output); let start = self.spans[index].start_unix_ms; let mut extra = BTreeMap::new(); @@ -944,6 +1059,7 @@ impl SpanCollector { elapsed_ms, iterations, output_chars, + output, .. } => { let Some(state) = self.subagents.remove(task_id) else { @@ -954,6 +1070,16 @@ impl SpanCollector { self.close_span(idx, now_unix_ms, SpanStatus::Ok, BTreeMap::new()); } } + // The subagent's final assistant text is the span's output + // (same gate + cap as its prompt input). + if self.ctx.capture_content && !output.is_empty() { + if let Some(span) = self.spans.get_mut(state.span_index) { + span.output = Some(serde_json::Value::String(truncate_chars( + output, + MAX_MODEL_CONTENT_CHARS, + ))); + } + } let start = self.spans[state.span_index].start_unix_ms; let mut extra = BTreeMap::new(); extra.insert("subagent.iterations".to_string(), json_u32(*iterations)); @@ -1034,7 +1160,7 @@ impl SpanCollector { // exporter — NDJSON file, app log, or Langfuse push — can ever // serialize it. This is the single choke point; the exporters // deliberately do not re-check the flag. - if !self.capture_content { + if !self.ctx.capture_content { log::debug!( target: "agent-tracing", "[agent-tracing] TurnContent dropped at storage (capture_content=false)" @@ -1110,6 +1236,25 @@ const MAX_TOOL_CONTENT_CHARS: usize = 4_000; /// Cap on captured error text (Langfuse observation `statusMessage`). const MAX_ERROR_MESSAGE_CHARS: usize = 500; +/// Cap on captured model request/completion content and subagent +/// prompt/output. Larger than the tool cap because a generation's input is the +/// full message array (system prompt included) — but still bounded so a +/// 100k-token context can't push the ingestion batch past Langfuse's event +/// size limits. +const MAX_MODEL_CONTENT_CHARS: usize = 25_000; + +/// Capture a model-payload JSON value for a span: kept structured when it +/// serializes within [`MAX_MODEL_CONTENT_CHARS`], else degraded to a truncated +/// string (readable in Langfuse, bounded in size). +fn capture_model_content(value: &serde_json::Value) -> serde_json::Value { + let serialized = value.to_string(); + if serialized.chars().count() <= MAX_MODEL_CONTENT_CHARS { + value.clone() + } else { + serde_json::Value::String(truncate_chars(&serialized, MAX_MODEL_CONTENT_CHARS)) + } +} + /// Truncate `text` to `max` characters, appending an explicit truncation /// marker (with the omitted char count) when content was dropped. Returns the /// input unchanged when it already fits. Slices on char boundaries, so it diff --git a/src/openhuman/agent/progress_tracing/tests.rs b/src/openhuman/agent/progress_tracing/tests.rs index fbb8c4ab88..66ca35944e 100644 --- a/src/openhuman/agent/progress_tracing/tests.rs +++ b/src/openhuman/agent/progress_tracing/tests.rs @@ -88,6 +88,8 @@ fn tool_completed( tool_name: tool.to_string(), success, output_chars: chars, + output: String::new(), + arguments: None, elapsed_ms: elapsed, iteration: 1, failure: None, @@ -237,6 +239,7 @@ fn spawn(task: &str, display: &str) -> AgentProgress { mode: "typed".to_string(), dedicated_thread: true, prompt_chars: 256, + prompt: "delegated prompt".to_string(), worker_thread_id: Some("worker-abc".to_string()), display_name: Some(display.to_string()), } @@ -286,6 +289,7 @@ fn subagent_lifecycle_nests_under_the_turn() { success: true, output_chars: 99, output: "file contents".to_string(), + arguments: None, elapsed_ms: 40, iteration: 1, failure: None, @@ -299,6 +303,7 @@ fn subagent_lifecycle_nests_under_the_turn() { elapsed_ms: 500, iterations: 3, output_chars: 1024, + output: String::new(), worktree_path: Some("/private/should/not/leak".to_string()), changed_files: vec!["secret_file.rs".to_string()], dirty_status: Some(true), @@ -387,6 +392,7 @@ fn unknown_subagent_task_ids_are_ignored() { elapsed_ms: 1, iterations: 1, output_chars: 1, + output: String::new(), worktree_path: None, changed_files: vec![], dirty_status: None, @@ -790,6 +796,7 @@ fn tool_io_is_captured_when_capture_content_is_on() { success: true, output_chars: 13, output: "file contents".to_string(), + arguments: None, elapsed_ms: 4, iteration: 1, failure: None, @@ -838,6 +845,7 @@ fn tool_io_is_never_recorded_when_capture_content_is_off() { success: true, output_chars: 6, output: "sekrit".to_string(), + arguments: None, elapsed_ms: 4, iteration: 1, failure: None, @@ -888,6 +896,10 @@ fn captured_tool_io_is_truncated_with_marker() { fn model_call(model: &str, reasoning: u64, cache_write: u64) -> AgentProgress { AgentProgress::ModelCallCompleted { model: model.to_string(), + provider_id: "managed".to_string(), + subagent_task_id: None, + input: None, + output: None, iteration: 1, input_tokens: 1_000, output_tokens: 200, @@ -928,7 +940,11 @@ fn model_call_completed_emits_generation_span_with_usage_cost_and_pricing() { assert_eq!(generation.status, SpanStatus::Ok); let a = &generation.attributes; - assert_eq!(a["gen_ai.request.model"], serde_json::json!("agentic-v1")); + // Provider-labeled model: `{provider_id}.{model}`. + assert_eq!( + a["gen_ai.request.model"], + serde_json::json!("managed.agentic-v1") + ); assert_eq!(a["gen_ai.usage.input_tokens"], serde_json::json!(1_000)); assert_eq!(a["gen_ai.usage.output_tokens"], serde_json::json!(200)); // Cache reads always flow, even when other calls happen to be zero. @@ -952,16 +968,33 @@ fn model_call_completed_emits_generation_span_with_usage_cost_and_pricing() { #[test] fn custom_model_generation_is_stamped_custom_provenance() { - let mut c = collect(&[ - (AgentProgress::TurnStarted, 0), - (model_call("claude-imaginary-9", 0, 0), 10), - ]); + // A BYO model rides whatever provider id the event carries ("openai", + // "ollama", or the "custom" default) — both on the generation and the root. + let event = AgentProgress::ModelCallCompleted { + model: "claude-imaginary-9".to_string(), + provider_id: "custom".to_string(), + subagent_task_id: None, + input: None, + output: None, + iteration: 1, + input_tokens: 10, + output_tokens: 2, + cached_input_tokens: 0, + cache_creation_tokens: 0, + reasoning_tokens: 0, + cost_usd: 0.0001, + }; + let mut c = collect(&[(AgentProgress::TurnStarted, 0), (event, 10)]); c.finish(20); let generation = find(c.spans(), "llm.claude-imaginary-9"); assert_eq!( generation.attributes["gen_ai.provider"], serde_json::json!("custom") ); + assert_eq!( + generation.attributes["gen_ai.request.model"], + serde_json::json!("custom.claude-imaginary-9") + ); // Provenance also lands on the root turn span (→ trace metadata). let turn = find(c.spans(), "agent.turn"); assert_eq!( @@ -1100,6 +1133,8 @@ fn failed_tool_records_classified_cause_only_when_capture_on() { tool_name: "shell".to_string(), success: false, output_chars: 0, + output: String::new(), + arguments: None, elapsed_ms: 5, iteration: 1, failure: Some(ClassifiedFailure { @@ -1167,3 +1202,270 @@ fn span_ids_are_unique_across_turns() { "span ids must be globally unique across turns" ); } + +// ── content-bearing wiring (system prompt / tool IO / subagent IO) ────────── + +fn capture_ctx() -> TraceContext { + ctx().with_capture_content(true) +} + +fn collect_with_capture(events: &[(AgentProgress, u64)]) -> SpanCollector { + let mut c = SpanCollector::new(capture_ctx()); + for (event, ts) in events { + c.record(event, *ts); + } + c +} + +fn model_call_with_content(subagent_task_id: Option<&str>) -> AgentProgress { + AgentProgress::ModelCallCompleted { + model: "chat-v1".to_string(), + provider_id: "managed".to_string(), + subagent_task_id: subagent_task_id.map(str::to_string), + input: Some(serde_json::json!([ + {"role": "system", "content": "You are OpenHuman."}, + {"role": "user", "content": "hi"} + ])), + output: Some(serde_json::json!({"role": "assistant", "content": "hello"})), + iteration: 1, + input_tokens: 100, + output_tokens: 10, + cached_input_tokens: 0, + cache_creation_tokens: 0, + reasoning_tokens: 0, + cost_usd: 0.001, + } +} + +#[test] +fn generation_records_request_messages_and_completion_when_capture_on() { + let mut c = collect_with_capture(&[ + (AgentProgress::TurnStarted, 0), + (model_call_with_content(None), 10), + ]); + c.finish(20); + let generation = find(c.spans(), "llm.chat-v1"); + let input = generation.input.as_ref().expect("generation input"); + assert!( + input.to_string().contains("You are OpenHuman."), + "system prompt must land in the generation input: {input}" + ); + assert!(generation + .output + .as_ref() + .expect("generation output") + .to_string() + .contains("hello")); +} + +#[test] +fn generation_withholds_content_when_capture_off() { + let mut c = collect(&[ + (AgentProgress::TurnStarted, 0), + (model_call_with_content(None), 10), + ]); + c.finish(20); + let generation = find(c.spans(), "llm.chat-v1"); + assert!(generation.input.is_none(), "capture off → no prompt"); + assert!(generation.output.is_none(), "capture off → no completion"); +} + +#[test] +fn subagent_model_call_nests_generation_and_stamps_model_on_subagent_span() { + let mut c = collect_with_capture(&[ + (AgentProgress::TurnStarted, 0), + (spawn("task-9", "Context Scout"), 5), + ( + AgentProgress::SubagentIterationStarted { + agent_id: "context_scout".to_string(), + task_id: "task-9".to_string(), + iteration: 1, + max_iterations: 8, + extended_policy: true, + }, + 6, + ), + (model_call_with_content(Some("task-9")), 10), + ]); + c.finish(20); + let spans = c.spans(); + + // Generation nests under the subagent's iteration, not the parent turn. + let generation = find(spans, "llm.chat-v1"); + let child_iter = find(spans, "subagent.iteration#1"); + assert_eq!( + generation.parent_span_id.as_deref(), + Some(child_iter.span_id.as_str()) + ); + assert!(generation.input.is_some(), "child generation carries input"); + + // The subagent span itself surfaces the provider-labeled model + usage. + let sub = find(spans, "subagent.Context Scout"); + assert_eq!( + sub.attributes["gen_ai.request.model"], + serde_json::json!("managed.chat-v1") + ); + assert_eq!( + sub.attributes["gen_ai.usage.input_tokens"], + serde_json::json!(100) + ); + assert_eq!( + sub.attributes["gen_ai.usage.cost_usd"], + serde_json::json!(0.001) + ); + + // The parent turn's rollup is NOT polluted by the child call. + let turn = find(spans, "agent.turn"); + assert!(turn.attributes.get("gen_ai.request.model").is_none()); +} + +#[test] +fn parent_tool_completion_backfills_arguments_and_records_output() { + let mut c = collect_with_capture(&[ + (AgentProgress::TurnStarted, 0), + ( + // tinyagents path: Started carries Null arguments. + AgentProgress::ToolCallStarted { + call_id: "c1".to_string(), + tool_name: "web_search".to_string(), + arguments: serde_json::Value::Null, + iteration: 1, + display_label: None, + display_detail: None, + }, + 5, + ), + ( + AgentProgress::ToolCallCompleted { + call_id: "c1".to_string(), + tool_name: "web_search".to_string(), + success: true, + output_chars: 7, + output: "results".to_string(), + arguments: Some(serde_json::json!({"query": "weather"})), + elapsed_ms: 40, + iteration: 1, + failure: None, + }, + 45, + ), + ]); + c.finish(50); + let tool = find(c.spans(), "tool.web_search"); + assert!( + tool.input + .as_ref() + .expect("tool input backfilled from completion") + .to_string() + .contains("weather"), + "arguments from the completion event must backfill the span input" + ); + assert_eq!( + tool.output, + Some(serde_json::Value::String("results".to_string())) + ); +} + +#[test] +fn subagent_span_records_prompt_and_final_output_when_capture_on() { + let mut c = collect_with_capture(&[ + (AgentProgress::TurnStarted, 0), + (spawn("task-1", "Researcher"), 5), + ( + AgentProgress::SubagentCompleted { + agent_id: "researcher".to_string(), + task_id: "task-1".to_string(), + elapsed_ms: 100, + iterations: 2, + output_chars: 12, + output: "final answer".to_string(), + worktree_path: None, + changed_files: vec![], + dirty_status: None, + }, + 105, + ), + ]); + c.finish(110); + let sub = find(c.spans(), "subagent.Researcher"); + assert_eq!( + sub.input, + Some(serde_json::Value::String("delegated prompt".to_string())) + ); + assert_eq!( + sub.output, + Some(serde_json::Value::String("final answer".to_string())) + ); +} + +#[test] +fn subagent_content_is_withheld_when_capture_off() { + let mut c = collect(&[ + (AgentProgress::TurnStarted, 0), + (spawn("task-1", "Researcher"), 5), + ( + AgentProgress::SubagentCompleted { + agent_id: "researcher".to_string(), + task_id: "task-1".to_string(), + elapsed_ms: 100, + iterations: 2, + output_chars: 12, + output: "final answer".to_string(), + worktree_path: None, + changed_files: vec![], + dirty_status: None, + }, + 105, + ), + ]); + c.finish(110); + let sub = find(c.spans(), "subagent.Researcher"); + assert!(sub.input.is_none()); + assert!(sub.output.is_none()); +} + +#[test] +fn oversized_model_content_degrades_to_truncated_string() { + let big = "x".repeat(MAX_MODEL_CONTENT_CHARS + 100); + let captured = capture_model_content(&serde_json::json!({ "content": big })); + let rendered = match &captured { + serde_json::Value::String(s) => s.clone(), + other => other.to_string(), + }; + assert!(rendered.chars().count() <= MAX_MODEL_CONTENT_CHARS + 64); + assert!(rendered.contains("truncated")); +} + +#[test] +fn turn_content_respects_the_trace_context_capture_gate() { + // Regression (PR #4506 review): the collector briefly carried TWO capture + // gates — a collector-level flag (checked by the TurnContent arm) and the + // TraceContext flag (checked everywhere else). The web progress bridge only + // sets the TraceContext flag, so TurnContent silently dropped the turn's + // prompt/reply even with capture_content enabled. There is now a single + // gate: both construction styles must attach TurnContent. + for collector in [ + SpanCollector::new(ctx().with_capture_content(true)), + SpanCollector::new(ctx()).with_content_capture(true), + ] { + let mut c = collector; + c.record(&AgentProgress::TurnStarted, 0); + c.record( + &AgentProgress::TurnContent { + input: Some("the prompt".to_string()), + output: Some("the reply".to_string()), + }, + 5, + ); + c.finish(10); + let turn = find(c.spans(), "agent.turn"); + assert_eq!( + turn.input, + Some(serde_json::Value::String("the prompt".to_string())) + ); + assert_eq!( + turn.output, + Some(serde_json::Value::String("the reply".to_string())) + ); + } +} diff --git a/src/openhuman/agent_orchestration/ops.rs b/src/openhuman/agent_orchestration/ops.rs index 614b182dc8..a898fb37bd 100644 --- a/src/openhuman/agent_orchestration/ops.rs +++ b/src/openhuman/agent_orchestration/ops.rs @@ -449,6 +449,7 @@ impl AgentOrchestrationSession { mode: "typed".to_string(), dedicated_thread: false, prompt_chars: prompt.chars().count(), + prompt: prompt.clone(), worker_thread_id: None, display_name: resolved_display_name, }) @@ -554,6 +555,7 @@ impl AgentOrchestrationSession { elapsed_ms: outcome.elapsed.as_millis() as u64, iterations: outcome.iterations as u32, output_chars: outcome.output.chars().count(), + output: outcome.output.clone(), worktree_path: None, changed_files: Vec::new(), dirty_status: None, diff --git a/src/openhuman/agent_orchestration/spawn_parallel_graph.rs b/src/openhuman/agent_orchestration/spawn_parallel_graph.rs index 94b44a24f3..09fd3c1a33 100644 --- a/src/openhuman/agent_orchestration/spawn_parallel_graph.rs +++ b/src/openhuman/agent_orchestration/spawn_parallel_graph.rs @@ -701,7 +701,7 @@ async fn stage_spawn_parallel_workers_from_defs( progress_sink, &definition, &task_id, - prompt.chars().count(), + &prompt, task.ownership .as_deref() .map(str::trim) @@ -968,9 +968,10 @@ async fn project_spawn_parallel_spawned( progress_sink: Option<&Sender>, definition: &AgentDefinition, task_id: &str, - prompt_chars: usize, + prompt: &str, has_ownership: bool, ) { + let prompt_chars = prompt.chars().count(); tracing::debug!( parent_session = %parent_session, task_id = %task_id, @@ -994,6 +995,7 @@ async fn project_spawn_parallel_spawned( mode: "typed".to_string(), dedicated_thread: false, prompt_chars, + prompt: prompt.to_string(), worker_thread_id: None, display_name: Some(definition.display_name().to_string()), }) @@ -1052,6 +1054,7 @@ async fn project_spawn_parallel_result( elapsed_ms: *elapsed_ms, iterations: *iterations, output_chars: output.as_ref().map(|s| s.chars().count()).unwrap_or(0), + output: output.clone().unwrap_or_default(), worktree_path: worktree_path.clone(), changed_files: changed_files.clone(), dirty_status: *dirty_status, diff --git a/src/openhuman/agent_orchestration/tools/agent_prepare_context.rs b/src/openhuman/agent_orchestration/tools/agent_prepare_context.rs index dfa020e8b4..6ce07d9029 100644 --- a/src/openhuman/agent_orchestration/tools/agent_prepare_context.rs +++ b/src/openhuman/agent_orchestration/tools/agent_prepare_context.rs @@ -229,6 +229,7 @@ async fn run_context_scout_with_catalog_and_workspace( mode: "typed".to_string(), dedicated_thread: false, prompt_chars: scout_prompt.chars().count(), + prompt: scout_prompt.clone(), worker_thread_id: None, display_name: Some(definition.display_name().to_string()), }) @@ -288,6 +289,7 @@ async fn run_context_scout_with_catalog_and_workspace( elapsed_ms: outcome.elapsed.as_millis() as u64, iterations: outcome.iterations as u32, output_chars: 0, + output: String::new(), worktree_path: None, changed_files: Vec::new(), dirty_status: None, @@ -327,6 +329,7 @@ async fn run_context_scout_with_catalog_and_workspace( elapsed_ms: outcome.elapsed.as_millis() as u64, iterations: outcome.iterations as u32, output_chars: bundle.chars().count(), + output: bundle.clone(), worktree_path: None, changed_files: Vec::new(), dirty_status: None, @@ -412,6 +415,7 @@ async fn run_context_scout_with_catalog_and_workspace( elapsed_ms: outcome.elapsed.as_millis() as u64, iterations: outcome.iterations as u32, output_chars: 0, + output: String::new(), worktree_path: None, changed_files: Vec::new(), dirty_status: None, @@ -450,6 +454,7 @@ async fn run_context_scout_with_catalog_and_workspace( elapsed_ms: outcome.elapsed.as_millis() as u64, iterations: outcome.iterations as u32, output_chars: 0, + output: String::new(), worktree_path: None, changed_files: Vec::new(), dirty_status: None, diff --git a/src/openhuman/agent_orchestration/tools/continue_subagent.rs b/src/openhuman/agent_orchestration/tools/continue_subagent.rs index a14fc452bb..fefcf59e42 100644 --- a/src/openhuman/agent_orchestration/tools/continue_subagent.rs +++ b/src/openhuman/agent_orchestration/tools/continue_subagent.rs @@ -222,6 +222,7 @@ impl Tool for ContinueSubagentTool { mode: "typed".to_string(), dedicated_thread: false, prompt_chars: message.chars().count(), + prompt: message.to_string(), worker_thread_id: checkpoint.worker_thread_id.clone(), display_name: definition.display_name.clone(), }) @@ -328,6 +329,7 @@ impl Tool for ContinueSubagentTool { elapsed_ms: outcome.elapsed.as_millis() as u64, iterations: outcome.iterations as u32, output_chars: outcome.output.chars().count(), + output: outcome.output.clone(), worktree_path: None, changed_files: Vec::new(), dirty_status: None, @@ -372,6 +374,7 @@ impl Tool for ContinueSubagentTool { elapsed_ms: outcome.elapsed.as_millis() as u64, iterations: outcome.iterations as u32, output_chars: outcome.output.chars().count(), + output: outcome.output.clone(), worktree_path: None, changed_files: Vec::new(), dirty_status: None, diff --git a/src/openhuman/agent_orchestration/tools/dispatch.rs b/src/openhuman/agent_orchestration/tools/dispatch.rs index 384b57a952..be577eb465 100644 --- a/src/openhuman/agent_orchestration/tools/dispatch.rs +++ b/src/openhuman/agent_orchestration/tools/dispatch.rs @@ -114,6 +114,7 @@ pub(crate) async fn dispatch_subagent( mode: "typed".to_string(), dedicated_thread: false, prompt_chars: prompt.chars().count(), + prompt: prompt.to_string(), worker_thread_id: None, display_name: Some(definition.display_name().to_string()), }) diff --git a/src/openhuman/agent_orchestration/tools/spawn_async_subagent.rs b/src/openhuman/agent_orchestration/tools/spawn_async_subagent.rs index 148f945bd6..55433fd134 100644 --- a/src/openhuman/agent_orchestration/tools/spawn_async_subagent.rs +++ b/src/openhuman/agent_orchestration/tools/spawn_async_subagent.rs @@ -445,6 +445,7 @@ impl Tool for SpawnAsyncSubagentTool { mode: "async".to_string(), dedicated_thread: worker_thread_id.is_some(), prompt_chars: prompt.chars().count(), + prompt: prompt.clone(), worker_thread_id: worker_thread_id.clone(), display_name: Some(definition.display_name().to_string()), }) @@ -567,6 +568,7 @@ impl Tool for SpawnAsyncSubagentTool { elapsed_ms: outcome.elapsed.as_millis() as u64, iterations: outcome.iterations as u32, output_chars: outcome.output.chars().count(), + output: outcome.output.clone(), worktree_path: None, changed_files: Vec::new(), dirty_status: None, @@ -626,6 +628,7 @@ impl Tool for SpawnAsyncSubagentTool { elapsed_ms: outcome.elapsed.as_millis() as u64, iterations: outcome.iterations as u32, output_chars: outcome.output.chars().count(), + output: outcome.output.clone(), worktree_path: None, changed_files: Vec::new(), dirty_status: None, diff --git a/src/openhuman/agent_orchestration/tools/spawn_subagent.rs b/src/openhuman/agent_orchestration/tools/spawn_subagent.rs index cbf9b85e33..a90f57556b 100644 --- a/src/openhuman/agent_orchestration/tools/spawn_subagent.rs +++ b/src/openhuman/agent_orchestration/tools/spawn_subagent.rs @@ -493,6 +493,7 @@ impl Tool for SpawnSubagentTool { mode: "typed".to_string(), dedicated_thread, prompt_chars: prompt.chars().count(), + prompt: prompt.clone(), worker_thread_id: worker_thread_id.clone(), display_name: Some(definition.display_name().to_string()), }) @@ -582,6 +583,7 @@ impl Tool for SpawnSubagentTool { elapsed_ms: outcome.elapsed.as_millis() as u64, iterations: outcome.iterations as u32, output_chars: outcome.output.chars().count(), + output: outcome.output.clone(), worktree_path: None, changed_files: Vec::new(), dirty_status: None, @@ -653,6 +655,7 @@ impl Tool for SpawnSubagentTool { elapsed_ms: outcome.elapsed.as_millis() as u64, iterations: outcome.iterations as u32, output_chars: outcome.output.chars().count(), + output: outcome.output.clone(), worktree_path: None, changed_files: Vec::new(), dirty_status: None, diff --git a/src/openhuman/channels/providers/web/progress_bridge.rs b/src/openhuman/channels/providers/web/progress_bridge.rs index 3071197a93..296a472c95 100644 --- a/src/openhuman/channels/providers/web/progress_bridge.rs +++ b/src/openhuman/channels/providers/web/progress_bridge.rs @@ -113,6 +113,22 @@ fn subagent_worktree_detail( } } +/// Trace user attribution for a turn whose `auth_get_me` cache is cold +/// (headless / autonomous / freshly booted cores): read the on-disk +/// app-session profile and return the user's email (preferred) or backend +/// user id. `None` when signed out or the profile is unreadable — the caller +/// then falls back to the transport client id. +fn session_profile_user_attribution(config: &crate::openhuman::config::Config) -> Option { + let state = crate::openhuman::credentials::session_support::build_session_state(config).ok()?; + state + .user + .as_ref() + .and_then(|u| u.get("email")) + .and_then(serde_json::Value::as_str) + .map(str::to_string) + .or(state.user_id) +} + /// Spawn a background task that reads [`AgentProgress`] events from the /// agent turn loop and translates them into [`WebChannelEvent`]s tagged /// with the correct client/thread/request IDs. The task runs until the @@ -175,6 +191,7 @@ pub(crate) fn spawn_progress_bridge( let user_attributed = identity.is_some(); let user_id = identity .and_then(|i| i.id.or(i.email)) + .or_else(|| session_profile_user_attribution(&config)) .unwrap_or_else(|| client_id.clone()); // Run origin for trace metadata: the request's source tag // ("ptt"/"dictation"/"type"/"agentbox"/"autonomous"/…), else a @@ -456,6 +473,7 @@ pub(crate) fn spawn_progress_bridge( elapsed_ms, iteration, failure, + .. } => { // Serialize the classified failure (if any) for the UI + ledger. let failure_json = failure.as_ref().and_then(|f| serde_json::to_value(f).ok()); @@ -501,6 +519,7 @@ pub(crate) fn spawn_progress_bridge( prompt_chars, worker_thread_id, display_name, + .. } => { let label = display_name.as_deref().unwrap_or(&agent_id); let kind = if worker_thread_id.is_some() { @@ -585,6 +604,7 @@ pub(crate) fn spawn_progress_bridge( worktree_path, changed_files, dirty_status, + .. } => { let completed_at = chrono::Utc::now(); ledger_upsert_agent_run( @@ -901,6 +921,7 @@ pub(crate) fn spawn_progress_bridge( elapsed_ms, iteration, failure, + .. } => { // Serialize the classified failure (if any) so a failed // sub-agent tool row carries its "why + next" copy on the @@ -1214,6 +1235,52 @@ pub(crate) fn spawn_progress_bridge( #[cfg(test)] mod tests { + use super::session_profile_user_attribution; + + #[test] + fn session_profile_attribution_none_when_signed_out() { + let tmp = tempfile::TempDir::new().unwrap(); + let config = crate::openhuman::config::Config { + workspace_dir: tmp.path().join("workspace"), + action_dir: tmp.path().join("workspace"), + config_path: tmp.path().join("config.toml"), + ..Default::default() + }; + assert!(session_profile_user_attribution(&config).is_none()); + } + + #[test] + fn session_profile_attribution_prefers_email_from_stored_session() { + let tmp = tempfile::TempDir::new().unwrap(); + let config = crate::openhuman::config::Config { + workspace_dir: tmp.path().join("workspace"), + action_dir: tmp.path().join("workspace"), + config_path: tmp.path().join("config.toml"), + ..Default::default() + }; + let service = crate::openhuman::credentials::AuthService::from_config(&config); + let mut metadata = std::collections::HashMap::new(); + metadata.insert( + "user_json".to_string(), + "{\"email\": \"steven@example.test\", \"_id\": \"u-1\"}".to_string(), + ); + metadata.insert("user_id".to_string(), "u-1".to_string()); + service + .store_provider_token( + crate::openhuman::credentials::APP_SESSION_PROVIDER, + crate::openhuman::credentials::DEFAULT_AUTH_PROFILE_NAME, + "session-token", + metadata, + true, + ) + .expect("store session profile"); + assert_eq!( + session_profile_user_attribution(&config).as_deref(), + Some("steven@example.test"), + "cold-cache attribution must resolve the on-disk session email" + ); + } + use super::*; #[test] diff --git a/src/openhuman/inference/provider/claude_code/mod.rs b/src/openhuman/inference/provider/claude_code/mod.rs index 023bf60872..9697ce98f9 100644 --- a/src/openhuman/inference/provider/claude_code/mod.rs +++ b/src/openhuman/inference/provider/claude_code/mod.rs @@ -206,6 +206,10 @@ fn thread_key_from_messages(messages: &[ChatMessage]) -> String { #[async_trait] impl Provider for ClaudeCodeProvider { + fn telemetry_provider_id(&self) -> String { + "claude-code".to_string() + } + fn capabilities(&self) -> ProviderCapabilities { ProviderCapabilities { native_tool_calling: true, diff --git a/src/openhuman/inference/provider/compatible_provider_impl.rs b/src/openhuman/inference/provider/compatible_provider_impl.rs index 64313bfb38..a88ac85ac0 100644 --- a/src/openhuman/inference/provider/compatible_provider_impl.rs +++ b/src/openhuman/inference/provider/compatible_provider_impl.rs @@ -16,6 +16,12 @@ use super::{AuthStyle, OpenAiCompatibleProvider}; #[async_trait] impl Provider for OpenAiCompatibleProvider { + fn telemetry_provider_id(&self) -> String { + // The configured provider slug ("openai", "groq", "venice", ...), + // normalized for stable telemetry labels. + self.name.trim().to_ascii_lowercase().replace(' ', "-") + } + fn capabilities(&self) -> crate::openhuman::inference::provider::traits::ProviderCapabilities { crate::openhuman::inference::provider::traits::ProviderCapabilities { native_tool_calling: self.native_tool_calling, diff --git a/src/openhuman/inference/provider/openhuman_backend.rs b/src/openhuman/inference/provider/openhuman_backend.rs index d49586e952..3892afa9df 100644 --- a/src/openhuman/inference/provider/openhuman_backend.rs +++ b/src/openhuman/inference/provider/openhuman_backend.rs @@ -127,6 +127,10 @@ impl OpenHumanBackendProvider { #[async_trait] impl Provider for OpenHumanBackendProvider { + fn telemetry_provider_id(&self) -> String { + "managed".to_string() + } + fn capabilities(&self) -> ProviderCapabilities { ProviderCapabilities { native_tool_calling: true, diff --git a/src/openhuman/inference/provider/reliable.rs b/src/openhuman/inference/provider/reliable.rs index 411c2eb6cf..822b1c7f03 100644 --- a/src/openhuman/inference/provider/reliable.rs +++ b/src/openhuman/inference/provider/reliable.rs @@ -133,6 +133,15 @@ impl ReliableProvider { #[async_trait] impl Provider for ReliableProvider { + fn telemetry_provider_id(&self) -> String { + // Delegate to the primary (first) upstream - the one that serves the + // call unless a failover kicks in. + self.providers + .first() + .map(|(_, p)| p.telemetry_provider_id()) + .unwrap_or_else(|| "custom".to_string()) + } + async fn warmup(&self) -> anyhow::Result<()> { for (name, provider) in &self.providers { tracing::info!(provider = name, "Warming up provider connection pool"); diff --git a/src/openhuman/inference/provider/router.rs b/src/openhuman/inference/provider/router.rs index 38fb7ee41a..f4d46ace92 100644 --- a/src/openhuman/inference/provider/router.rs +++ b/src/openhuman/inference/provider/router.rs @@ -164,6 +164,13 @@ impl RouterProvider { #[async_trait] impl Provider for RouterProvider { + fn telemetry_provider_id(&self) -> String { + self.providers + .get(self.default_index) + .map(|(_, p)| p.telemetry_provider_id()) + .unwrap_or_else(|| "custom".to_string()) + } + async fn chat_with_system( &self, system_prompt: Option<&str>, diff --git a/src/openhuman/inference/provider/traits.rs b/src/openhuman/inference/provider/traits.rs index 5594963c76..21e50eff01 100644 --- a/src/openhuman/inference/provider/traits.rs +++ b/src/openhuman/inference/provider/traits.rs @@ -421,6 +421,17 @@ fn format_prompt_messages(messages: &[ChatMessage]) -> String { #[async_trait] pub trait Provider: Send + Sync { + /// Stable provider identifier for telemetry/tracing. Rendered by trace + /// exporters as the Langfuse `gen_ai.provider` and the `{provider}.{model}` + /// model label (e.g. `managed.chat-v1`, `openai.gpt-4o`). + /// + /// Defaults to `"custom"`; concrete providers override with their slug + /// (the managed backend returns `"managed"`, an OpenAI-compatible BYOK + /// provider its configured name, wrappers delegate to their active inner). + fn telemetry_provider_id(&self) -> String { + "custom".to_string() + } + /// Query provider capabilities. /// /// Default implementation returns minimal capabilities (no native tool calling). diff --git a/src/openhuman/routing/provider.rs b/src/openhuman/routing/provider.rs index d26f733edb..6794f3c4c5 100644 --- a/src/openhuman/routing/provider.rs +++ b/src/openhuman/routing/provider.rs @@ -377,6 +377,12 @@ impl IntelligentRoutingProvider { #[async_trait] impl Provider for IntelligentRoutingProvider { + fn telemetry_provider_id(&self) -> String { + // Attribute to the remote provider; local-routing decisions are + // per-call and the remote is the configured default. + self.remote.telemetry_provider_id() + } + fn capabilities(&self) -> ProviderCapabilities { self.remote.capabilities() } diff --git a/src/openhuman/threads/turn_state/mirror_tests.rs b/src/openhuman/threads/turn_state/mirror_tests.rs index 04ab9d6e80..04f72814ea 100644 --- a/src/openhuman/threads/turn_state/mirror_tests.rs +++ b/src/openhuman/threads/turn_state/mirror_tests.rs @@ -128,6 +128,8 @@ fn tool_call_start_and_complete_track_timeline() { tool_name: "shell".into(), success: true, output_chars: 12, + output: String::new(), + arguments: None, elapsed_ms: 50, iteration: 1, failure: None, @@ -202,6 +204,8 @@ fn tool_call_started_reuses_args_delta_placeholder_for_same_call_id() { tool_name: "shell".into(), success: true, output_chars: 1, + output: String::new(), + arguments: None, elapsed_ms: 5, iteration: 1, failure: None, @@ -314,6 +318,7 @@ fn subagent_lifecycle_records_and_clears_active() { mode: "typed".into(), dedicated_thread: false, prompt_chars: 42, + prompt: String::new(), worker_thread_id: None, display_name: Some("Researcher".into()), }); @@ -344,6 +349,7 @@ fn subagent_lifecycle_records_and_clears_active() { elapsed_ms: 1234, iterations: 2, output_chars: 80, + output: String::new(), worktree_path: None, changed_files: Vec::new(), dirty_status: None, @@ -366,6 +372,7 @@ fn subagent_transcript_persists_interleaved_prose_and_tools() { mode: "typed".into(), dedicated_thread: false, prompt_chars: 10, + prompt: String::new(), worker_thread_id: None, display_name: Some("Researcher".into()), }); @@ -409,6 +416,7 @@ fn subagent_transcript_persists_interleaved_prose_and_tools() { success: true, output_chars: 5, output: String::new(), + arguments: None, elapsed_ms: 12, iteration: 1, failure: None, diff --git a/src/openhuman/tinyagents/journal.rs b/src/openhuman/tinyagents/journal.rs index c7b1552ec7..aa057676f1 100644 --- a/src/openhuman/tinyagents/journal.rs +++ b/src/openhuman/tinyagents/journal.rs @@ -419,8 +419,12 @@ mod tests { let journal: Arc = Arc::new(StoreEventJournal::new(stores.journal)); let sink = EventSink::with_stream_id(run_id.as_str()); - let journal_sink = JournalSink::new(journal, run_id.clone()); - let redacting = RedactingSink::new(Arc::new(journal_sink), vec!["sk-super-secret".into()]); + // Keep a handle to the JournalSink: persistence became asynchronous + // (background `AppendWorker` drain, tinyagents v1.5 audit remediation), + // so the test must `flush()` before reading the journal back — exactly + // the contract the crate documents for read-after-write. + let journal_sink = Arc::new(JournalSink::new(journal, run_id.clone())); + let redacting = RedactingSink::new(journal_sink.clone(), vec!["sk-super-secret".into()]); sink.subscribe(Arc::new(FanOutSink::new().with(Arc::new(redacting)))); sink.emit(AgentEvent::ModelStarted { @@ -431,6 +435,9 @@ mod tests { call_id: "c1".into(), tool_name: "echo".to_string(), }); + // Drain the async persistence worker so the durable log has caught up + // (flush blocks on the drain thread's ack, not on this runtime). + journal_sink.flush(); // Reconstruct from the durable store alone. let replayed = read_run_events_at(&tmp, run_id.as_str(), 0).await; diff --git a/src/openhuman/tinyagents/mod.rs b/src/openhuman/tinyagents/mod.rs index c1df7d9d33..38d6a172eb 100644 --- a/src/openhuman/tinyagents/mod.rs +++ b/src/openhuman/tinyagents/mod.rs @@ -157,6 +157,16 @@ fn run_policy_for(max_iterations: usize, response_cache_enabled: bool) -> RunPol // pass `false` here AND attach no cache, so a live user turn can never be // served a cached model response (double fail-safe). policy.cache.response_cache_enabled = response_cache_enabled; + // Payload capture ON: the loop stamps request messages + completion onto + // `ModelCompleted` and tool arguments + result onto `ToolCompleted`, which + // the `OpenhumanEventBridge` projects into content-bearing `AgentProgress` + // events (generation/tool span input+output in trace exports). Privacy + // posture is unchanged off-device: the durable journal passes through a + // `RedactingSink` (on-device, same data class as the threads DB, which + // already persists full conversations + tool output), and the Langfuse + // exporter withholds all content unless + // `observability.agent_tracing.capture_content` is on. + policy.capture = tinyagents::harness::runtime::PayloadCapture::all(); policy } @@ -450,6 +460,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(); let AssembledTurnHarness { harness, cursor, @@ -645,6 +659,7 @@ pub(crate) async fn run_turn_via_tinyagents_shared( let bridge = OpenhumanEventBridge::with_scope( on_progress, model, + provider_id.clone(), max_iterations, subagent_scope.clone(), cursor.clone(), diff --git a/src/openhuman/tinyagents/observability.rs b/src/openhuman/tinyagents/observability.rs index 6cd3e4c732..912a64e903 100644 --- a/src/openhuman/tinyagents/observability.rs +++ b/src/openhuman/tinyagents/observability.rs @@ -143,11 +143,27 @@ struct BridgeState { cache_misses: u64, } +/// Per-model-call figures `record_usage` resolved from the provider-usage +/// carry (charged>estimate cost precedence + cache/reasoning breakdown), +/// keyed by iteration so the subsequent `ModelCompleted` projection reports +/// the same numbers as the wallet accounting. +#[derive(Clone, Copy, Debug)] +struct ResolvedCallFigures { + cost_usd: f64, + cache_creation_tokens: u64, + reasoning_tokens: u64, +} + /// An [`EventListener`] that mirrors harness events onto openhuman's progress /// sink and cost tracker. pub(crate) struct OpenhumanEventBridge { on_progress: Option>, model: String, + /// Telemetry provider id (`"managed"`, `"openai"`, …) — from + /// [`Provider::telemetry_provider_id`](crate::openhuman::inference::provider::Provider::telemetry_provider_id). + /// Rides on `ModelCallCompleted` so trace exporters render the Langfuse + /// model as `{provider_id}.{model}`. + provider_id: String, max_iterations: u32, /// `None` for a parent turn; `Some` to emit child-scoped `Subagent*` events. scope: Option, @@ -175,6 +191,13 @@ pub(crate) struct OpenhumanEventBridge { /// iteration cursor, bumped once per `ModelStarted`) so a given call's usage /// is recorded exactly once. See [`OpenhumanEventBridge::record_usage`]. recorded_iterations: Mutex>, + /// Per-iteration figures resolved by `record_usage` (see + /// [`ResolvedCallFigures`]); taken by the `ModelCompleted` arm. + resolved_calls: Mutex>, + /// `call_id → start instant` for in-flight tool calls, written on + /// `ToolStarted` and taken on `ToolCompleted` so the projected completion + /// event carries a real `elapsed_ms` (the crate event has no timing). + tool_started_at: Mutex>, state: Mutex, /// Ordered overflow buffer for progress events that hit backpressure /// (channel `Full`). Once ANY event spills here, `draining` stays set and @@ -204,6 +227,7 @@ impl OpenhumanEventBridge { Self::with_scope( on_progress, model, + "custom", max_iterations, None, Arc::default(), @@ -219,6 +243,7 @@ impl OpenhumanEventBridge { pub(crate) fn with_scope( on_progress: Option>, model: impl Into, + provider_id: impl Into, max_iterations: usize, scope: Option, cursor: IterationCursor, @@ -229,6 +254,7 @@ impl OpenhumanEventBridge { Arc::new(Self { on_progress, model: model.into(), + provider_id: provider_id.into(), max_iterations: max_iterations as u32, scope, cursor, @@ -236,6 +262,8 @@ impl OpenhumanEventBridge { failure_map, usage_carry, recorded_iterations: Mutex::new(std::collections::HashSet::new()), + resolved_calls: Mutex::new(std::collections::HashMap::new()), + tool_started_at: Mutex::new(std::collections::HashMap::new()), state: Mutex::new(BridgeState::default()), overflow: Arc::default(), }) @@ -387,16 +415,13 @@ impl OpenhumanEventBridge { .unwrap_or_else(|p| p.into_inner()) .pop_front(); - // Estimate from catalogued per-MTok rates as the floor; prefer the - // provider's own charged amount when it reported one (charged > estimate - // precedence — restores the legacy observer's behaviour, so credit-metered - // backends surface real billing rather than a token-rate estimate). - let estimate = crate::openhuman::cost::catalog::estimate_cost_usd( - &self.model, - usage.input_tokens, - usage.output_tokens, - usage.cache_read_tokens, - ); + // Estimate as the floor via the tier-aware `agent::cost` table (managed + // handles like `chat-v1`/`burst-v1` + the vendor catalog + heuristics — + // the catalog-only lookup priced every managed-tier call as $0); prefer + // the provider's own charged amount when it reported one (charged > + // estimate precedence, so credit-metered backends surface real billing + // rather than a token-rate estimate). + let estimate = Self::estimate_call_cost(&self.model, usage); let call_cost = carried .as_ref() .map(|u| u.charged_amount_usd) @@ -435,6 +460,21 @@ impl OpenhumanEventBridge { context_window, "[cost] recording per-call usage (charged>estimate precedence via provider carry)" ); + // Stash the resolved per-call figures so the `ModelCompleted` arm (which + // fires right after this event and emits the `ModelCallCompleted` + // generation telemetry) reports the SAME cost/cache/reasoning numbers as + // the wallet accounting, instead of re-deriving a bare estimate. + self.resolved_calls + .lock() + .unwrap_or_else(|p| p.into_inner()) + .insert( + iteration, + ResolvedCallFigures { + cost_usd: call_cost, + cache_creation_tokens, + reasoning_tokens, + }, + ); let (input, output, cached, charged) = { let mut s = self.state.lock().unwrap(); s.input_tokens += usage.input_tokens; @@ -473,38 +513,12 @@ impl OpenhumanEventBridge { // The cost footer is a top-level surface; for a child run the global // cost tracker feed above is the authoritative accounting and the parent // emits its own footer, so suppress the per-child `TurnCostUpdated`. + // Per-call generation telemetry (`ModelCallCompleted`) is emitted from + // the `AgentEvent::ModelCompleted` arm instead — that event fires after + // `UsageRecorded` and is the only one carrying the captured request + // messages + completion, so the generation gets usage AND content in + // one shot (for parent and child scopes alike). if self.scope.is_none() { - // Per-call telemetry first (exact model/usage/cost for THIS call — - // trace exporters turn it into a Langfuse generation), then the - // cumulative footer rollup. Child-scoped calls carry no task - // attribution on this event, so they stay cumulative-only. - log::debug!( - "[tinyagents][usage] model_call_completed model={} iteration={} in={} out={} \ - cache_read={} cache_write={} reasoning={} cost_usd={:.6}", - self.model, - iteration, - usage.input_tokens, - usage.output_tokens, - usage.cache_read_tokens, - usage.cache_creation_tokens, - usage.reasoning_tokens, - call_cost, - ); - self.send(AgentProgress::ModelCallCompleted { - model: self.model.clone(), - iteration, - input_tokens: usage.input_tokens, - output_tokens: usage.output_tokens, - cached_input_tokens: usage.cache_read_tokens, - // Use the carried-or-crate breakdown (computed above): for - // providers where cache-creation/reasoning tokens ride only on - // the carried UsageInfo, the crate `usage.*` fields are 0, which - // would leave per-call progress/Langfuse telemetry at zero even - // though the cost tracker got the right values (#4467). - cache_creation_tokens, - reasoning_tokens, - cost_usd: call_cost, - }); self.send(AgentProgress::TurnCostUpdated { model: self.model.clone(), iteration, @@ -515,6 +529,26 @@ impl OpenhumanEventBridge { }); } } + + /// Estimate one call's USD cost. Uses the tier-aware + /// [`agent::cost`](crate::openhuman::agent::cost) table (managed handles + /// like `chat-v1`/`burst-v1` + the vendor catalog + heuristics) — the + /// previous `cost::catalog::estimate_cost_usd` only knew concrete vendor + /// ids, so every managed-tier call priced as $0 in traces and the footer. + fn estimate_call_cost(model: &str, usage: &Usage) -> f64 { + crate::openhuman::agent::cost::estimate_call_cost_usd( + model, + &UsageInfo { + input_tokens: usage.input_tokens, + output_tokens: usage.output_tokens, + context_window: 0, + cached_input_tokens: usage.cache_read_tokens, + cache_creation_tokens: usage.cache_creation_tokens, + reasoning_tokens: usage.reasoning_tokens, + charged_amount_usd: 0.0, + }, + ) + } } impl EventListener for OpenhumanEventBridge { @@ -611,6 +645,76 @@ impl EventListener for OpenhumanEventBridge { // exactly once per model call; prefer it over `ModelCompleted`'s // optional usage to avoid double counting. AgentEvent::UsageRecorded { usage } => self.record_usage(usage), + // Per-call generation telemetry. `ModelCompleted` fires exactly once + // per model call, after `UsageRecorded`, and is the only event + // carrying the captured request messages (incl. the system prompt) + // + completion (`RunPolicy.capture.model_io`, enabled in + // `run_policy_for`). Emitted for parent AND child scopes — the + // child call carries its owning `subagent_task_id` so the trace + // exporter nests the generation under the subagent span (this is + // what makes the Context Scout's model calls visible in Langfuse). + AgentEvent::ModelCompleted { + usage, + input, + output, + .. + } => { + let iteration = self.iteration(); + let usage = usage.unwrap_or_default(); + // Prefer the figures `record_usage` resolved for this call + // (charged>estimate cost + carried cache/reasoning breakdown — + // `UsageRecorded` fires before `ModelCompleted`), so the + // generation telemetry matches the wallet accounting exactly; + // fall back to a bare tier-aware estimate when absent. + let resolved = self + .resolved_calls + .lock() + .unwrap_or_else(|p| p.into_inner()) + .remove(&iteration); + let call_cost = resolved + .as_ref() + .map(|r| r.cost_usd) + .unwrap_or_else(|| Self::estimate_call_cost(&self.model, &usage)); + let cache_creation_tokens = resolved + .as_ref() + .map(|r| r.cache_creation_tokens) + .unwrap_or(usage.cache_creation_tokens); + let reasoning_tokens = resolved + .as_ref() + .map(|r| r.reasoning_tokens) + .unwrap_or(usage.reasoning_tokens); + log::debug!( + "[tinyagents][usage] model_call_completed model={} provider={} iteration={} \ + child={} in={} out={} cache_read={} cache_write={} reasoning={} \ + cost_usd={:.6} input_captured={} output_captured={}", + self.model, + self.provider_id, + iteration, + self.scope.is_some(), + usage.input_tokens, + usage.output_tokens, + usage.cache_read_tokens, + cache_creation_tokens, + reasoning_tokens, + call_cost, + input.is_some(), + output.is_some(), + ); + self.send(AgentProgress::ModelCallCompleted { + model: self.model.clone(), + provider_id: self.provider_id.clone(), + subagent_task_id: self.scope.as_ref().map(|s| s.task_id.clone()), + input: input.clone(), + output: output.clone(), + iteration, + input_tokens: usage.input_tokens, + output_tokens: usage.output_tokens, + cached_input_tokens: usage.cache_read_tokens, + cache_creation_tokens, + reasoning_tokens, + cost_usd: call_cost, + }); + } AgentEvent::CostRecorded { cost } => { tracing::debug!( cost = ?cost, @@ -737,6 +841,8 @@ impl EventListener for OpenhumanEventBridge { tool_name: requested_name.clone(), success: false, output_chars: 0, + output: String::new(), + arguments: Some(arguments.clone()), elapsed_ms: 0, iteration, failure, @@ -761,6 +867,7 @@ impl EventListener for OpenhumanEventBridge { success: false, output_chars: 0, output: String::new(), + arguments: Some(arguments.clone()), elapsed_ms: 0, iteration, failure, @@ -777,6 +884,12 @@ impl EventListener for OpenhumanEventBridge { // instead of a rewritten ToolStarted. So this arm fires only for // real, model-visible tools and needs no sentinel guard. let iteration = self.iteration(); + // Stamp the start instant so the completion event carries a real + // elapsed_ms (the crate's ToolCompleted has no timing payload). + self.tool_started_at + .lock() + .unwrap_or_else(|p| p.into_inner()) + .insert(call_id.as_str().to_string(), std::time::Instant::now()); match &self.scope { None => self.send(AgentProgress::ToolCallStarted { call_id: call_id.as_str().to_string(), @@ -799,7 +912,10 @@ impl EventListener for OpenhumanEventBridge { } } AgentEvent::ToolCompleted { - call_id, tool_name, .. + call_id, + tool_name, + input, + output, } => { let iteration = self.iteration(); // The crate event carries no success/error, so read what the @@ -813,10 +929,34 @@ impl EventListener for OpenhumanEventBridge { .and_then(|mut m| m.remove(call_id.as_str())); let success = outcome.as_ref().map(|(ok, ..)| *ok).unwrap_or(true); // Real execution duration + output size the capture middleware - // recorded off the `ToolResult` (#4467, item 4). Absent (event - // projected before the middleware ran) → 0, as before. - let elapsed_ms = outcome.as_ref().map(|(_, _, e, _)| *e).unwrap_or(0); - let output_chars = outcome.as_ref().map(|(_, _, _, c)| *c).unwrap_or(0); + // recorded off the `ToolResult` (#4467, item 4). Fall back to + // the bridge's own ToolStarted stamp for duration, and to the + // captured payload for size, when the middleware ran late. + let stamped_elapsed = self + .tool_started_at + .lock() + .unwrap_or_else(|p| p.into_inner()) + .remove(call_id.as_str()) + .map(|t| t.elapsed().as_millis() as u64) + .unwrap_or(0); + let elapsed_ms = outcome + .as_ref() + .map(|(_, _, e, _)| *e) + .filter(|e| *e > 0) + .unwrap_or(stamped_elapsed); + // Tool result text, captured by the harness when + // `RunPolicy.capture.tool_io` is on (the loop emits it as a + // JSON string). Empty when capture is off. + let output_text = match output { + Some(serde_json::Value::String(s)) => s.clone(), + Some(v) => v.to_string(), + None => String::new(), + }; + let output_chars = outcome + .as_ref() + .map(|(_, _, _, c)| *c) + .filter(|c| *c > 0) + .unwrap_or_else(|| output_text.chars().count()); // Carry the classified failure onto whichever completion event // this projects — main-agent OR sub-agent (#4459). Previously // the sub-agent branch dropped it on the floor. @@ -827,6 +967,8 @@ impl EventListener for OpenhumanEventBridge { tool_name: tool_name.clone(), success, output_chars, + output: output_text, + arguments: input.clone(), elapsed_ms, iteration, failure, @@ -838,7 +980,8 @@ impl EventListener for OpenhumanEventBridge { tool_name: tool_name.clone(), success, output_chars, - output: String::new(), + output: output_text, + arguments: input.clone(), elapsed_ms, iteration, failure, @@ -980,6 +1123,151 @@ mod tests { assert_eq!((input, output), (100, 40)); } + #[tokio::test] + async fn model_completed_projects_generation_with_content_and_provider() { + let (tx, mut rx) = tokio::sync::mpsc::channel(64); + let bridge = OpenhumanEventBridge::with_scope( + Some(tx), + "chat-v1", + "managed", + 10, + None, + Arc::default(), + Arc::default(), + Arc::default(), + Arc::default(), + ); + let sink = EventSink::new(); + sink.subscribe(bridge.clone()); + + sink.emit(AgentEvent::ModelStarted { + call_id: "m1".into(), + model: "chat-v1".to_string(), + }); + sink.emit(AgentEvent::ModelCompleted { + call_id: "m1".into(), + usage: Some(Usage::new(1_000, 50)), + input: Some(serde_json::json!([ + {"role": "system", "content": "You are OpenHuman."} + ])), + output: Some(serde_json::json!({"role": "assistant", "content": "hi"})), + }); + + let mut seen = None; + while let Ok(p) = rx.try_recv() { + if let AgentProgress::ModelCallCompleted { + model, + provider_id, + subagent_task_id, + input, + output, + input_tokens, + cost_usd, + .. + } = p + { + seen = Some(( + model, + provider_id, + subagent_task_id, + input, + output, + input_tokens, + cost_usd, + )); + } + } + let (model, provider_id, task, input, output, input_tokens, cost_usd) = + seen.expect("ModelCallCompleted projected from ModelCompleted"); + assert_eq!(model, "chat-v1"); + assert_eq!(provider_id, "managed"); + assert!(task.is_none(), "parent scope carries no task id"); + assert!(input.unwrap().to_string().contains("You are OpenHuman.")); + assert!(output.unwrap().to_string().contains("hi")); + assert_eq!(input_tokens, 1_000); + // chat-v1 is a managed tier handle — the tier-aware estimator must + // price it (> $0); the old catalog-only lookup returned exactly 0. + assert!(cost_usd > 0.0, "managed tier call must not price as $0"); + } + + #[tokio::test] + async fn subagent_model_completed_carries_task_attribution() { + let (tx, mut rx) = tokio::sync::mpsc::channel(64); + let bridge = OpenhumanEventBridge::with_scope( + Some(tx), + "burst-v1", + "managed", + 8, + Some(SubagentScope { + agent_id: "context_scout".to_string(), + task_id: "ctx-1".to_string(), + extended_policy: true, + }), + Arc::default(), + Arc::default(), + Arc::default(), + Arc::default(), + ); + let sink = EventSink::new(); + sink.subscribe(bridge.clone()); + sink.emit(AgentEvent::ModelCompleted { + call_id: "m1".into(), + usage: Some(Usage::new(10, 5)), + input: None, + output: None, + }); + let mut task = None; + while let Ok(p) = rx.try_recv() { + if let AgentProgress::ModelCallCompleted { + subagent_task_id, .. + } = p + { + task = subagent_task_id; + } + } + assert_eq!( + task.as_deref(), + Some("ctx-1"), + "child model calls must carry the owning subagent task id" + ); + } + + #[tokio::test] + async fn tool_completed_projects_output_arguments_and_elapsed() { + let (tx, mut rx) = tokio::sync::mpsc::channel(64); + let bridge = OpenhumanEventBridge::new(Some(tx), "mock-model", 10); + let sink = EventSink::new(); + sink.subscribe(bridge.clone()); + + sink.emit(AgentEvent::ToolStarted { + call_id: "t1".into(), + tool_name: "echo".to_string(), + }); + sink.emit(AgentEvent::ToolCompleted { + call_id: "t1".into(), + tool_name: "echo".to_string(), + input: Some(serde_json::json!({"text": "ping"})), + output: Some(serde_json::Value::String("pong".to_string())), + }); + + let mut seen = None; + while let Ok(p) = rx.try_recv() { + if let AgentProgress::ToolCallCompleted { + output, + output_chars, + arguments, + .. + } = p + { + seen = Some((output, output_chars, arguments)); + } + } + let (output, output_chars, arguments) = seen.expect("tool completion projected"); + assert_eq!(output, "pong"); + assert_eq!(output_chars, 4); + assert!(arguments.unwrap().to_string().contains("ping")); + } + #[tokio::test] async fn unknown_tool_call_projects_attempted_name_as_failed_timeline_row() { // #4118: the crate recovers an unavailable-tool call via ReturnToolError diff --git a/tests/memory_core_threads_raw_coverage_e2e.rs b/tests/memory_core_threads_raw_coverage_e2e.rs index 7f9bc83f90..4622795415 100644 --- a/tests/memory_core_threads_raw_coverage_e2e.rs +++ b/tests/memory_core_threads_raw_coverage_e2e.rs @@ -559,6 +559,7 @@ async fn thread_ops_welcome_migration_and_turn_state_cover_error_and_cleanup_pat mode: "typed".into(), dedicated_thread: true, prompt_chars: 42, + prompt: String::new(), worker_thread_id: None, display_name: Some("Researcher".into()), })); @@ -568,6 +569,7 @@ async fn thread_ops_welcome_migration_and_turn_state_cover_error_and_cleanup_pat elapsed_ms: 50, iterations: 2, output_chars: 100, + output: String::new(), worktree_path: None, changed_files: vec![], dirty_status: None, diff --git a/tests/memory_raw_coverage_e2e.rs b/tests/memory_raw_coverage_e2e.rs index 95e9fa0317..bfbda1bc1a 100644 --- a/tests/memory_raw_coverage_e2e.rs +++ b/tests/memory_raw_coverage_e2e.rs @@ -588,6 +588,7 @@ fn threads_turn_state_store_skips_corrupt_entries_and_marks_interrupted() { name: "memory.search".into(), round: 1, status: ToolTimelineStatus::Running, + failure: None, args_buffer: Some("{\"query\":\"coverage\"}".into()), display_name: Some("Search Memory".into()), detail: None, @@ -613,6 +614,7 @@ fn threads_turn_state_store_skips_corrupt_entries_and_marks_interrupted() { output_chars: Some(64), display_name: None, detail: None, + failure: None, }], transcript: vec![], }), diff --git a/tests/memory_threads_raw_coverage_e2e.rs b/tests/memory_threads_raw_coverage_e2e.rs index 239739febb..1fccf4c14b 100644 --- a/tests/memory_threads_raw_coverage_e2e.rs +++ b/tests/memory_threads_raw_coverage_e2e.rs @@ -1150,6 +1150,7 @@ fn thread_title_error_and_turn_state_helpers_cover_wire_shapes() { name: "memory.search".into(), round: 1, status: ToolTimelineStatus::Success, + failure: None, args_buffer: Some("{\"q\":\"coverage\"}".into()), display_name: Some("Memory Search".into()), detail: Some("2 results".into()), @@ -3157,6 +3158,8 @@ fn turn_state_mirror_persists_progress_edges_from_public_events() { tool_name: "memory.search".into(), success: false, output_chars: 0, + output: String::new(), + arguments: None, elapsed_ms: 11, iteration: 2, failure: None, @@ -3176,6 +3179,7 @@ fn turn_state_mirror_persists_progress_edges_from_public_events() { mode: "typed".into(), dedicated_thread: true, prompt_chars: 99, + prompt: String::new(), worker_thread_id: None, display_name: Some("Researcher".into()), })); @@ -3209,8 +3213,10 @@ fn turn_state_mirror_persists_progress_edges_from_public_events() { success: true, output_chars: 44, output: "child tool output".into(), + arguments: None, elapsed_ms: 22, iteration: 1, + failure: None, })); assert!(mirror.observe(&AgentProgress::SubagentFailed { agent_id: "researcher".into(), @@ -3451,6 +3457,7 @@ fn turn_state_store_persists_lists_marks_and_clears_snapshots() { name: "subagent:research".into(), round: 2, status: ToolTimelineStatus::Running, + failure: None, args_buffer: None, display_name: Some("Research".into()), detail: None, @@ -3476,6 +3483,7 @@ fn turn_state_store_persists_lists_marks_and_clears_snapshots() { output_chars: Some(10), display_name: None, detail: None, + failure: None, }], transcript: vec![], }), diff --git a/vendor/tinyagents b/vendor/tinyagents index 9442bd2964..357bcc8587 160000 --- a/vendor/tinyagents +++ b/vendor/tinyagents @@ -1 +1 @@ -Subproject commit 9442bd2964479b849ca78286eeb2a4e93a4547b4 +Subproject commit 357bcc85871b40c03a7dd81b61d1b3253bfb33e6