fix(observability): populate Langfuse generation/tool I/O and trace metadata (#6)

docs/modules/harness/observability.md

@@ -320,6 +320,35 @@ Redaction policies should handle:
 Redacted events should preserve structural fields such as ids, event kinds,
 timings, and counters so traces remain useful.
 
+## Payload Capture
+
+The event stream is **payload-free by default**: `AgentEvent::ModelCompleted`
+and `AgentEvent::ToolCompleted` carry only ids and usage, never the prompt,
+completion, tool arguments, or tool result. This keeps journals and exporters
+safe for privacy-sensitive deployments without any configuration.
+
+`RunPolicy::capture` (a `PayloadCapture { model_io, tool_io }`) opts a run into
+carrying those payloads onto the completion events:
+
+- `model_io` — attaches the request messages (`input`) and the model completion
+  (`output`) to every `ModelCompleted`.
+- `tool_io` — attaches the tool arguments (`input`) and the tool result
+  (`output`) to every `ToolCompleted`.
+
+Capture is snapshotted inside the agent loop before the request/call value is
+moved into the model or tool wrap onion, so it never perturbs execution. Because
+captured payloads ride the ordinary event pipeline, a `RedactingSink` still
+masks them before they reach a journal or exporter, and the durable journal
+replays them verbatim.
+
+Downstream, the Langfuse exporter reads these fields to populate the
+Input/Output panels of a generation (`generation-create`) and a tool
+observation (`tool-create`). With capture disabled the fields are absent and the
+observation renders with ids, timing, and usage only. The harness exporter also
+defaults the trace-level metadata from the run lineage (root/first run ids and
+parent), mirroring the graph exporter, so a trace is correlatable even when the
+caller passes no metadata.
+
 ## Graph Events
 
 State-graph event payloads should be rich enough for a UI to render run

src/harness/agent_loop/mod.rs

@@ -444,6 +444,14 @@ impl<State: Send + Sync, Ctx: Send + Sync> AgentHarness<State, Ctx> {
                 model: binding.model,
                 streaming,
             };
+            // Snapshot the request messages for observability before `request`
+            // is moved into the model-wrap onion, gated by the capture policy so
+            // payload-free runs never serialize prompt text.
+            let captured_input = self
+                .policy
+                .capture
+                .model_io
+                .then(|| serde_json::to_value(&request.messages).unwrap_or(Value::Null));
             let mut response = self
                 .middleware
                 .run_wrapped_model(ctx, state, request, &base)
@@ -466,9 +474,16 @@ impl<State: Send + Sync, Ctx: Send + Sync> AgentHarness<State, Ctx> {
                 let record = ctx.emit(AgentEvent::UsageRecorded { usage });
                 status.set_last_event(record.id);
             }
+            let captured_output = self
+                .policy
+                .capture
+                .model_io
+                .then(|| serde_json::to_value(&response.message).unwrap_or(Value::Null));
             let record = ctx.emit(AgentEvent::ModelCompleted {
                 call_id,
                 usage: response.usage,
+                input: captured_input,
+                output: captured_output,
             });
             status.set_last_event(record.id);
 
@@ -628,6 +643,10 @@ impl<State: Send + Sync, Ctx: Send + Sync> AgentHarness<State, Ctx> {
                 });
                 status.set_last_event(record.id);
 
+                // Snapshot the arguments for observability before `call` is
+                // moved into the tool-wrap onion, gated by the capture policy.
+                let captured_input = self.policy.capture.tool_io.then(|| call.arguments.clone());
+
                 // The real tool call is the innermost base of the tool-wrap
                 // onion (same before -> wrap -> after ordering as the model
                 // path): lifecycle `before_tool` ran above, the wrap onion runs
@@ -646,9 +665,16 @@ impl<State: Send + Sync, Ctx: Send + Sync> AgentHarness<State, Ctx> {
                 run.tool_calls += 1;
                 status.tool_calls = run.tool_calls;
                 status.active_tool_calls.retain(|c| c != &tool_call_id);
+                let captured_output = self
+                    .policy
+                    .capture
+                    .tool_io
+                    .then(|| Value::String(result.content.clone()));
                 let record = ctx.emit(AgentEvent::ToolCompleted {
                     call_id: tool_call_id,
                     tool_name,
+                    input: captured_input,
+                    output: captured_output,
                 });
                 status.set_last_event(record.id);
 

src/harness/events/types.rs

@@ -75,6 +75,18 @@ pub enum AgentEvent {
         /// Token usage reported by the provider, when available.
         #[serde(default, skip_serializing_if = "Option::is_none")]
         usage: Option<Usage>,
+        /// The request messages sent to the model, captured only when
+        /// [`PayloadCapture::model_io`][crate::harness::runtime::PayloadCapture::model_io]
+        /// is enabled. `None` in the default payload-free mode. Populated so an
+        /// exporter can render the prompt in a generation's Input panel.
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        input: Option<serde_json::Value>,
+        /// The model completion (assistant message), captured only when
+        /// [`PayloadCapture::model_io`][crate::harness::runtime::PayloadCapture::model_io]
+        /// is enabled. `None` in the default payload-free mode. Populated so an
+        /// exporter can render the completion in a generation's Output panel.
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        output: Option<serde_json::Value>,
     },
 
     /// A tool-selection middleware filtered the model-visible tool set before a
@@ -103,6 +115,18 @@ pub enum AgentEvent {
         call_id: CallId,
         /// Name of the tool that was invoked.
         tool_name: String,
+        /// The arguments the tool was invoked with, captured only when
+        /// [`PayloadCapture::tool_io`][crate::harness::runtime::PayloadCapture::tool_io]
+        /// is enabled. `None` in the default payload-free mode. Populated so an
+        /// exporter can render the call in a tool observation's Input panel.
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        input: Option<serde_json::Value>,
+        /// The tool result content, captured only when
+        /// [`PayloadCapture::tool_io`][crate::harness::runtime::PayloadCapture::tool_io]
+        /// is enabled. `None` in the default payload-free mode. Populated so an
+        /// exporter can render the result in a tool observation's Output panel.
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        output: Option<serde_json::Value>,
     },
 
     /// The model called a tool that is not registered, and the run's

src/harness/observability/langfuse.rs

@@ -156,6 +156,7 @@ impl LangfuseClient {
             .first()
             .map(|obs| iso_ms(obs.ts_ms))
             .unwrap_or_else(|| iso_ms(now_ms()));
+        let metadata = trace_metadata(&trace, observations);
 
         let mut batch = Vec::with_capacity(observations.len() + 1);
         batch.push(json!({
@@ -172,7 +173,7 @@ impl LangfuseClient {
                 "release": trace.release,
                 "version": trace.version,
                 "tags": if trace.tags.is_empty() { Value::Null } else { json!(trace.tags) },
-                "metadata": if trace.metadata.is_null() { Value::Null } else { trace.metadata },
+                "metadata": metadata,
             })),
         }));
 
@@ -277,6 +278,33 @@ fn resolve_trace_id(
         })
 }
 
+/// Builds the trace-level metadata, defaulting useful run-lineage coordinates
+/// (root/first run ids and thread, mirroring what the graph exporter folds in)
+/// so a harness trace is correlatable even when the caller passes no metadata.
+/// Any caller-supplied [`LangfuseTraceConfig::metadata`] keys are merged on top
+/// and win on collision. Returns [`Value::Null`] only when there is nothing to
+/// attach, so [`clean_nulls`] drops the field entirely.
+fn trace_metadata(trace: &LangfuseTraceConfig, observations: &[AgentObservation]) -> Value {
+    let mut metadata = serde_json::Map::new();
+    if let Some(first) = observations.first() {
+        metadata.insert("root_run_id".to_string(), json!(first.root_run_id.as_str()));
+        metadata.insert("run_id".to_string(), json!(first.run_id.as_str()));
+        if let Some(parent) = &first.parent_run_id {
+            metadata.insert("parent_run_id".to_string(), json!(parent.as_str()));
+        }
+    }
+    if let Value::Object(extra) = &trace.metadata {
+        for (k, v) in extra {
+            metadata.insert(k.clone(), v.clone());
+        }
+    }
+    if metadata.is_empty() {
+        Value::Null
+    } else {
+        Value::Object(metadata)
+    }
+}
+
 fn observation_event(trace_id: &str, obs: &AgentObservation) -> Value {
     let timestamp = iso_ms(obs.ts_ms);
     let metadata = json!({
@@ -287,7 +315,12 @@ fn observation_event(trace_id: &str, obs: &AgentObservation) -> Value {
         "event": obs.event,
     });
     match &obs.event {
-        AgentEvent::ModelCompleted { call_id, usage } => json!({
+        AgentEvent::ModelCompleted {
+            call_id,
+            usage,
+            input,
+            output,
+        } => json!({
             "id": obs.event_id.as_str(),
             "timestamp": timestamp,
             "type": "generation-create",
@@ -298,10 +331,17 @@ fn observation_event(trace_id: &str, obs: &AgentObservation) -> Value {
                 "startTime": timestamp,
                 "endTime": timestamp,
                 "usage": usage.map(langfuse_usage),
+                "input": input,
+                "output": output,
                 "metadata": metadata,
             })),
         }),
-        AgentEvent::ToolCompleted { call_id, tool_name } => json!({
+        AgentEvent::ToolCompleted {
+            call_id,
+            tool_name,
+            input,
+            output,
+        } => json!({
             "id": obs.event_id.as_str(),
             "timestamp": timestamp,
             "type": "tool-create",
@@ -311,6 +351,8 @@ fn observation_event(trace_id: &str, obs: &AgentObservation) -> Value {
                 "name": tool_name,
                 "startTime": timestamp,
                 "endTime": timestamp,
+                "input": input,
+                "output": output,
                 "metadata": metadata,
             })),
         }),
@@ -469,6 +511,8 @@ mod tests {
                                 total_tokens: 7,
                                 ..Default::default()
                             }),
+                            input: None,
+                            output: None,
                         },
                     ),
                 ],
@@ -479,8 +523,82 @@ mod tests {
         assert_eq!(events[0]["type"], "trace-create");
         assert_eq!(events[0]["body"]["id"], "root-1");
         assert_eq!(events[0]["body"]["userId"], "user-1");
+        // Trace metadata is defaulted from run lineage even without a caller value.
+        assert_eq!(events[0]["body"]["metadata"]["root_run_id"], "root-1");
+        assert_eq!(events[0]["body"]["metadata"]["run_id"], "run-1");
         assert_eq!(events[2]["type"], "generation-create");
         assert_eq!(events[2]["body"]["id"], "model-call");
         assert_eq!(events[2]["body"]["usage"]["input"], 3);
+        // Payload-free generation: no input/output body fields.
+        assert!(events[2]["body"].get("input").is_none());
+        assert!(events[2]["body"].get("output").is_none());
+    }
+
+    #[test]
+    fn populates_generation_and_tool_io_when_captured() {
+        let client =
+            LangfuseClient::proxy("https://backend.test/telemetry/langfuse/ingestion", "t")
+                .unwrap();
+        let batch = client
+            .build_ingestion_batch(
+                LangfuseTraceConfig::default(),
+                &[
+                    obs(
+                        0,
+                        AgentEvent::ModelCompleted {
+                            call_id: CallId::new("model-call"),
+                            usage: None,
+                            input: Some(json!([{ "role": "user", "content": "hi" }])),
+                            output: Some(json!({ "content": "hello there" })),
+                        },
+                    ),
+                    obs(
+                        1,
+                        AgentEvent::ToolCompleted {
+                            call_id: CallId::new("tool-call"),
+                            tool_name: "lookup".to_string(),
+                            input: Some(json!({ "query": "weather" })),
+                            output: Some(json!("sunny")),
+                        },
+                    ),
+                ],
+            )
+            .unwrap();
+
+        let events = batch["batch"].as_array().unwrap();
+        assert_eq!(events[1]["type"], "generation-create");
+        assert_eq!(events[1]["body"]["input"][0]["content"], "hi");
+        assert_eq!(events[1]["body"]["output"]["content"], "hello there");
+        assert_eq!(events[2]["type"], "tool-create");
+        assert_eq!(events[2]["body"]["input"]["query"], "weather");
+        assert_eq!(events[2]["body"]["output"], "sunny");
+    }
+
+    #[test]
+    fn merges_caller_trace_metadata_over_defaults() {
+        let client =
+            LangfuseClient::proxy("https://backend.test/telemetry/langfuse/ingestion", "t")
+                .unwrap();
+        let batch = client
+            .build_ingestion_batch(
+                LangfuseTraceConfig {
+                    metadata: json!({ "deployment": "prod", "root_run_id": "override" }),
+                    ..Default::default()
+                },
+                &[obs(
+                    0,
+                    AgentEvent::RunStarted {
+                        run_id: RunId::new("run-1"),
+                        thread_id: None,
+                    },
+                )],
+            )
+            .unwrap();
+
+        let meta = &batch["batch"].as_array().unwrap()[0]["body"]["metadata"];
+        assert_eq!(meta["deployment"], "prod");
+        // Caller keys win on collision with the defaulted lineage.
+        assert_eq!(meta["root_run_id"], "override");
+        assert_eq!(meta["run_id"], "run-1");
     }
 }

src/harness/observability/test.rs

@@ -117,6 +117,8 @@ fn agent_latency_metrics_include_model_tool_and_run_elapsed() {
             AgentEvent::ModelCompleted {
                 call_id: model_id.clone(),
                 usage: None,
+                input: None,
+                output: None,
             },
         ),
         obs(
@@ -133,6 +135,8 @@ fn agent_latency_metrics_include_model_tool_and_run_elapsed() {
             AgentEvent::ToolCompleted {
                 call_id: tool_id.clone(),
                 tool_name: "lookup".to_string(),
+                input: None,
+                output: None,
             },
         ),
         obs(