diff --git a/src/harness/agent_loop/test.rs b/src/harness/agent_loop/test.rs index 1177491..a179486 100644 --- a/src/harness/agent_loop/test.rs +++ b/src/harness/agent_loop/test.rs @@ -2165,3 +2165,76 @@ async fn mock_streaming_script_invoke_folds_items_to_response() { .expect("fold succeeds"); assert_eq!(response.text(), "abcd"); } + +#[tokio::test] +async fn tool_completed_event_carries_outcome() { + use crate::harness::events::RecordingListener; + + // A tool that fails: its ToolCompleted event must carry the failure message, + // a real duration, and the output size — from the event itself, not a + // side-channel — so journal-backed exporters can render the outcome. + struct FailTool; + #[async_trait] + impl Tool<()> for FailTool { + fn name(&self) -> &str { + "boom" + } + fn description(&self) -> &str { + "always fails" + } + fn schema(&self) -> ToolSchema { + ToolSchema::new("boom", "always fails", json!({ "type": "object" })) + } + async fn call(&self, _state: &(), call: ToolCall) -> Result { + Ok(ToolResult { + call_id: call.id, + name: "boom".to_string(), + content: "nope".to_string(), + raw: None, + error: Some("kaboom".to_string()), + elapsed_ms: 0, + }) + } + } + + let mut harness: AgentHarness<()> = AgentHarness::new(); + harness.register_model( + "mock", + Arc::new(MockModel::with_responses(vec![ + tool_call_response("c1", "boom", json!({})), + text_response("done", 1, 1), + ])), + ); + harness.register_tool(Arc::new(FailTool)); + + let recorder = Arc::new(RecordingListener::new()); + let ctx: RunContext<()> = RunContext::new(RunConfig::new("run-tc"), ()); + ctx.events.subscribe(recorder.clone()); + harness + .invoke_in_context(&(), ctx, vec![Message::user("go")]) + .await + .expect("run succeeds"); + + let (error, duration_ms, output_bytes) = recorder + .events() + .into_iter() + .find_map(|record| match record.event { + AgentEvent::ToolCompleted { + tool_name, + error, + duration_ms, + output_bytes, + .. + } if tool_name == "boom" => Some((error, duration_ms, output_bytes)), + _ => None, + }) + .expect("a ToolCompleted event for `boom`"); + + assert_eq!( + error.as_deref(), + Some("kaboom"), + "failure message on the event" + ); + assert!(duration_ms.is_some(), "wall-clock duration present"); + assert_eq!(output_bytes, Some(4), "\"nope\".len() == 4"); +} diff --git a/src/harness/agent_loop/tools.rs b/src/harness/agent_loop/tools.rs index b35c9b5..cf7e563 100644 --- a/src/harness/agent_loop/tools.rs +++ b/src/harness/agent_loop/tools.rs @@ -260,12 +260,22 @@ impl AgentHarness { .capture .tool_io .then(|| Value::String(result.content.clone())); + // Outcome fields carried on the event itself (not a side-channel) so + // journal-backed exporters render duration/size/success without the + // live run's state. Duration is wall-clock (completion minus start); + // `error` mirrors `ToolResult::error` (`None` == success). + let duration_ms = crate::harness::ids::now_ms().saturating_sub(prepared.started_at_ms); + let output_bytes = result.content.len() as u64; + let error = result.error.clone(); let record = ctx.emit(AgentEvent::ToolCompleted { call_id: prepared.call_id, tool_name: prepared.tool_name, started_at_ms: Some(prepared.started_at_ms), input: prepared.captured_input, output: captured_output, + duration_ms: Some(duration_ms), + output_bytes: Some(output_bytes), + error, }); status.set_last_event(record.id); diff --git a/src/harness/events/test.rs b/src/harness/events/test.rs index 0ad048f..ee69e17 100644 --- a/src/harness/events/test.rs +++ b/src/harness/events/test.rs @@ -147,6 +147,9 @@ fn completed_events_deserialize_without_started_at_ms() { started_at_ms: Some(1_704_067_199_000), input: None, output: None, + duration_ms: Some(12), + output_bytes: Some(5), + error: None, }; let json = serde_json::to_value(&event).unwrap(); assert_eq!(json["started_at_ms"], 1_704_067_199_000u64); diff --git a/src/harness/events/types.rs b/src/harness/events/types.rs index 030e0b9..27d9c47 100644 --- a/src/harness/events/types.rs +++ b/src/harness/events/types.rs @@ -143,6 +143,22 @@ pub enum AgentEvent { /// exporter can render the result in a tool observation's Output panel. #[serde(default, skip_serializing_if = "Option::is_none")] output: Option, + /// Wall-clock duration of the call in milliseconds (completion minus + /// [`started_at_ms`]). Present regardless of payload capture, so an + /// exporter renders a real duration without a side-channel. `None` for + /// events serialized before this field existed. + #[serde(default, skip_serializing_if = "Option::is_none")] + duration_ms: Option, + /// Size, in bytes, of the tool's textual result content. Present even in + /// payload-free mode (unlike [`output`]), so an exporter can show result + /// size without capturing the body. `None` for older events. + #[serde(default, skip_serializing_if = "Option::is_none")] + output_bytes: Option, + /// Failure message when the tool call failed; `None` on success. Lets an + /// exporter render success/failure and a reason from the journalled + /// event itself rather than a live outcome side-channel. + #[serde(default, skip_serializing_if = "Option::is_none")] + error: Option, }, /// The model called a tool that is not registered, and the run's diff --git a/src/harness/observability/langfuse/mod.rs b/src/harness/observability/langfuse/mod.rs index 4453ca8..527bcc0 100644 --- a/src/harness/observability/langfuse/mod.rs +++ b/src/harness/observability/langfuse/mod.rs @@ -316,26 +316,43 @@ fn observation_event(trace_id: &str, obs: &AgentObservation) -> Value { started_at_ms, input, output, - } => json!({ - "id": obs.event_id.as_str(), - "timestamp": timestamp, - // A tool call is modelled as a span. `tool-create` is not a valid - // Langfuse ingestion observation type — older/self-hosted Langfuse - // rejects it, silently dropping every tool observation. - "type": "span-create", - "body": clean_nulls(json!({ - "id": call_id.as_str(), - "traceId": trace_id, - "name": tool_name, - // Loop-captured start time when available (see the - // generation branch above). - "startTime": started_at_ms.map(iso_ms).unwrap_or_else(|| timestamp.clone()), - "endTime": timestamp, - "input": input, - "output": output, - "metadata": metadata, - })), - }), + duration_ms, + output_bytes, + error, + } => { + // Prefer the loop-captured start + real duration for the end time; + // fall back to the journal timestamp. A failed call is marked ERROR + // with the failure reason, and result size rides metadata so the + // span is informative even in payload-free mode. + let end_time = match (started_at_ms, duration_ms) { + (Some(start), Some(dur)) => iso_ms(start.saturating_add(*dur)), + _ => timestamp.clone(), + }; + let mut tool_metadata = metadata.clone(); + if let (Some(map), Some(bytes)) = (tool_metadata.as_object_mut(), output_bytes) { + map.insert("output_bytes".into(), json!(bytes)); + } + json!({ + "id": obs.event_id.as_str(), + "timestamp": timestamp, + // A tool call is modelled as a span. `tool-create` is not a + // valid Langfuse ingestion observation type — older/self-hosted + // Langfuse rejects it, silently dropping every tool observation. + "type": "span-create", + "body": clean_nulls(json!({ + "id": call_id.as_str(), + "traceId": trace_id, + "name": tool_name, + "startTime": started_at_ms.map(iso_ms).unwrap_or_else(|| timestamp.clone()), + "endTime": end_time, + "input": input, + "output": output, + "level": error.as_ref().map(|_| "ERROR"), + "statusMessage": error, + "metadata": tool_metadata, + })), + }) + } AgentEvent::RunFailed { error, .. } => json!({ "id": obs.event_id.as_str(), "timestamp": timestamp, diff --git a/src/harness/observability/langfuse/test.rs b/src/harness/observability/langfuse/test.rs index c3a3de5..c062215 100644 --- a/src/harness/observability/langfuse/test.rs +++ b/src/harness/observability/langfuse/test.rs @@ -116,6 +116,9 @@ fn populates_generation_and_tool_io_when_captured() { started_at_ms: Some(1_704_067_199_500), input: Some(json!({ "query": "weather" })), output: Some(json!("sunny")), + duration_ms: Some(250), + output_bytes: Some(5), + error: None, }, ), ], @@ -135,7 +138,12 @@ fn populates_generation_and_tool_io_when_captured() { assert_eq!(events[1]["body"]["startTime"], iso_ms(1_704_067_199_000)); assert_eq!(events[1]["body"]["endTime"], iso_ms(1_704_067_200_000)); assert_eq!(events[2]["body"]["startTime"], iso_ms(1_704_067_199_500)); - assert_eq!(events[2]["body"]["endTime"], iso_ms(1_704_067_200_001)); + // The tool span's end time is now start + the event's own duration_ms + // (1_704_067_199_500 + 250), a real execution window rather than the + // journal-append timestamp. + assert_eq!(events[2]["body"]["endTime"], iso_ms(1_704_067_199_750)); + // Result size rides metadata even though this fixture also captured output. + assert_eq!(events[2]["body"]["metadata"]["output_bytes"], 5); // Observation metadata carries only lineage + event kind, not the whole // event payload (which would duplicate input/output already in `body`). diff --git a/src/harness/observability/test.rs b/src/harness/observability/test.rs index 40b6129..3d2ed68 100644 --- a/src/harness/observability/test.rs +++ b/src/harness/observability/test.rs @@ -140,6 +140,9 @@ fn agent_latency_metrics_include_model_tool_and_run_elapsed() { started_at_ms: None, input: None, output: None, + duration_ms: None, + output_bytes: None, + error: None, }, ), obs( diff --git a/src/harness/testkit/test.rs b/src/harness/testkit/test.rs index 0ebeaf6..875a58a 100644 --- a/src/harness/testkit/test.rs +++ b/src/harness/testkit/test.rs @@ -341,6 +341,9 @@ fn make_trajectory() -> Vec { started_at_ms: None, input: None, output: None, + duration_ms: None, + output_bytes: None, + error: None, }, AgentEvent::ModelStarted { call_id: CallId::new("c2"), diff --git a/tests/e2e_orchestrator_subagents.rs b/tests/e2e_orchestrator_subagents.rs index d29559a..4d13d0b 100644 --- a/tests/e2e_orchestrator_subagents.rs +++ b/tests/e2e_orchestrator_subagents.rs @@ -174,6 +174,9 @@ async fn orchestrator_resolves_and_runs_only_the_chosen_subagents() -> Result<() started_at_ms: None, input: None, output: None, + duration_ms: None, + output_bytes: None, + error: None, }); Ok::<(String, String), TinyAgentsError>((name, result.content)) } diff --git a/tests/e2e_registry_observability_contracts.rs b/tests/e2e_registry_observability_contracts.rs index fde5b14..97d403f 100644 --- a/tests/e2e_registry_observability_contracts.rs +++ b/tests/e2e_registry_observability_contracts.rs @@ -186,6 +186,9 @@ fn component_metadata_and_event_kinds_are_stable_serializable_contracts() { started_at_ms: None, input: None, output: None, + duration_ms: None, + output_bytes: None, + error: None, }, AgentEvent::StateUpdate, AgentEvent::MiddlewareStarted { name: "mw".into() }, @@ -310,6 +313,9 @@ async fn event_sinks_journals_and_status_stores_preserve_run_lineage() { started_at_ms: None, input: None, output: None, + duration_ms: None, + output_bytes: None, + error: None, }); assert_eq!(journal.len(), 2); assert_eq!(journal.replay_from(1)[0].event.kind(), "tool.completed");