feat(tracing): Langfuse prompt/reply/cost + one-trace-per-turn (repairs #4399 build break) (#4434)

src/bin/harness_subagent_audit.rs

@@ -636,7 +636,8 @@ async fn drain_progress(
             | AgentProgress::ToolCallArgsDelta { .. }
             | AgentProgress::TaskBoardUpdated { .. }
             | AgentProgress::TurnCostUpdated { .. }
-            | AgentProgress::TurnStarted => {}
+            | AgentProgress::TurnStarted
+            | AgentProgress::TurnContent { .. } => {}
         }
     }
 }

src/openhuman/agent/harness/session/turn/core.rs

@@ -1063,6 +1063,16 @@ impl Agent {
         )
         .await;
 
+        // Content (prompt + reply) rides its own event so a tracing consumer can
+        // attach it to the turn span. Transmission off-device stays gated by the
+        // exporter's opt-in `capture_content` flag; here it is only surfaced onto
+        // the in-memory span.
+        self.emit_progress(AgentProgress::TurnContent {
+            input: Some(user_message.to_string()),
+            output: Some(reply.clone()),
+        })
+        .await;
+
         self.emit_progress(AgentProgress::TurnCompleted {
             iterations: outcome.model_calls as u32,
         })

src/openhuman/agent/progress.rs

@@ -283,4 +283,16 @@ pub enum AgentProgress {
         /// Total iterations used.
         iterations: u32,
     },
+
+    /// The turn's content: the user's prompt and the model's final reply.
+    /// Emitted just before [`Self::TurnCompleted`] so a tracing consumer can
+    /// attach `input`/`output` to the turn span. Carries prompt/reply text, so
+    /// exporters must honor the opt-in `observability.agent_tracing.capture_content`
+    /// gate before transmitting it off-device.
+    TurnContent {
+        /// The user's prompt for this turn.
+        input: Option<String>,
+        /// The model's final reply for this turn.
+        output: Option<String>,
+    },
 }

src/openhuman/agent/progress_tracing.rs

@@ -52,11 +52,16 @@ pub(crate) mod langfuse;
 /// Trace-level correlation context, stamped onto the root span.
 #[derive(Debug, Clone)]
 pub struct TraceContext {
-    /// Trace id — the agent session id. All spans of a run share it.
+    /// Trace id — unique per turn. Every span of a single turn shares it, so
+    /// each turn becomes its own Langfuse trace.
     pub session_id: String,
     /// User attribution (e.g. the broadcast client id / "system" for
     /// autonomous runs). `None` when the caller is anonymous.
     pub user_id: Option<String>,
+    /// Grouping key (the thread/conversation id) exported as the Langfuse
+    /// `sessionId` so per-turn traces still group under one session. `None`
+    /// leaves the trace ungrouped.
+    pub session_group: Option<String>,
 }
 
 impl TraceContext {
@@ -64,8 +69,16 @@ impl TraceContext {
         Self {
             session_id: session_id.into(),
             user_id,
+            session_group: None,
         }
     }
+
+    /// Set the grouping key (thread/conversation id) for the Langfuse
+    /// `sessionId`, so a conversation's per-turn traces group together.
+    pub fn with_session_group(mut self, group: impl Into<String>) -> Self {
+        self.session_group = Some(group.into());
+        self
+    }
 }
 
 /// Derive the trace id (session id) for a run: prefer the UI session id when
@@ -129,6 +142,15 @@ pub struct TraceSpan {
     pub status: SpanStatus,
     /// Metadata-only attributes (no secrets/PII).
     pub attributes: BTreeMap<String, serde_json::Value>,
+    /// Optional prompt/input content. Populated only when content capture is on
+    /// (via `AgentProgress::TurnContent`); the exporter still gates transmission
+    /// behind `observability.agent_tracing.capture_content`.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub input: Option<serde_json::Value>,
+    /// Optional model-reply/output content. Same capture/gating rules as
+    /// [`Self::input`].
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub output: Option<serde_json::Value>,
 }
 
 impl TraceSpan {
@@ -161,6 +183,12 @@ pub struct SpanCollector {
     ctx: TraceContext,
     spans: Vec<TraceSpan>,
     next_span_seq: u64,
+    /// Per-collector (per-turn) random prefix for minted span ids. Langfuse
+    /// dedupes observations by id **globally**, so a bare per-turn sequence
+    /// (`0000…0001`) collides across turns and silently binds later turns'
+    /// observations to whichever trace first claimed the id. Prefixing with a
+    /// fresh nonce makes every span id globally unique.
+    id_prefix: String,
 
     turn_span_id: Option<String>,
     turn_span_index: Option<usize>,
@@ -179,6 +207,7 @@ impl SpanCollector {
             ctx,
             spans: Vec::new(),
             next_span_seq: 0,
+            id_prefix: uuid::Uuid::new_v4().simple().to_string(),
             turn_span_id: None,
             turn_span_index: None,
             current_iteration_span_id: None,
@@ -208,7 +237,9 @@ impl SpanCollector {
     /// and deterministic within a run, which keeps the tests reproducible.
     fn mint_span_id(&mut self) -> String {
         self.next_span_seq += 1;
-        format!("{:016x}", self.next_span_seq)
+        // Nonce prefix keeps the id globally unique across turns (Langfuse
+        // dedupes observations by id project-wide).
+        format!("{}-{:016x}", self.id_prefix, self.next_span_seq)
     }
 
     fn open_span(
@@ -231,6 +262,8 @@ impl SpanCollector {
             end_unix_ms: None,
             status: SpanStatus::Unset,
             attributes,
+            input: None,
+            output: None,
         });
         (span_id, index)
     }
@@ -269,6 +302,12 @@ impl SpanCollector {
                 serde_json::Value::String(user.clone()),
             );
         }
+        if let Some(group) = &self.ctx.session_group {
+            attrs.insert(
+                "thread.id".to_string(),
+                serde_json::Value::String(group.clone()),
+            );
+        }
         let (id, index) = self.open_span(SpanKind::Turn, "agent.turn", None, start_unix_ms, attrs);
         self.turn_span_id = Some(id.clone());
         self.turn_span_index = Some(index);
@@ -585,6 +624,26 @@ impl SpanCollector {
                 }
             }
 
+            AgentProgress::TurnContent { input, output } => {
+                // Attach prompt/reply to the root turn span. Held in-memory only;
+                // the exporter decides whether to transmit it (opt-in gate).
+                let index = match self.turn_span_index {
+                    Some(idx) => idx,
+                    None => {
+                        self.ensure_turn_span(now_unix_ms);
+                        self.turn_span_index.expect("turn span just created")
+                    }
+                };
+                if let Some(span) = self.spans.get_mut(index) {
+                    if let Some(text) = input {
+                        span.input = Some(serde_json::Value::String(text.clone()));
+                    }
+                    if let Some(text) = output {
+                        span.output = Some(serde_json::Value::String(text.clone()));
+                    }
+                }
+            }
+
             AgentProgress::TurnCompleted { iterations } => {
                 self.close_current_iteration(now_unix_ms);
                 if let Some(index) = self.turn_span_index {

src/openhuman/agent/progress_tracing/langfuse.rs

@@ -11,9 +11,12 @@
 //! never hold Langfuse keys and never hit `/api/public/ingestion` directly.
 //!
 //! Best-effort: any failure is logged and swallowed by the caller so tracing
-//! never breaks a turn. Spans carry only metadata (names, kinds, timings,
-//! counts) — never prompt text, tool arguments, or PII — honoring the project's
-//! "never log secrets or full PII" rule.
+//! never breaks a turn. By default spans carry only metadata (names, kinds,
+//! timings, and non-PII token/cost figures — the latter promoted into Langfuse's
+//! native `usageDetails`/`costDetails`). Prompt text and the model's reply are
+//! withheld unless the operator opts in via
+//! `observability.agent_tracing.capture_content`, preserving the project's
+//! "never log secrets or full PII" default.
 
 use std::time::Duration;
 
@@ -83,7 +86,7 @@ fn langfuse_metadata(span: &TraceSpan) -> Value {
 /// a single `trace-create` for the shared trace id followed by one
 /// `span-create` observation per span. Field names are Langfuse's camelCase
 /// (`traceId`, `startTime`, `parentObservationId`); timestamps are ISO strings.
-pub(crate) fn spans_to_langfuse_batch(spans: &[TraceSpan]) -> Value {
+pub(crate) fn spans_to_langfuse_batch(spans: &[TraceSpan], include_content: bool) -> Value {
     let mut batch: Vec<Value> = Vec::with_capacity(spans.len() + 1);
 
     // One trace-create for the run, keyed by the shared trace id. Prefer the
@@ -93,15 +96,25 @@ pub(crate) fn spans_to_langfuse_batch(spans: &[TraceSpan]) -> Value {
         .find(|s| s.parent_span_id.is_none())
         .or_else(|| spans.first())
     {
+        let mut trace_body = json!({
+            "id": root.trace_id,
+            "name": root.name,
+            "timestamp": iso_millis(root.start_unix_ms),
+        });
+        // Attribute the trace to the user and group per-turn traces under the
+        // conversation via Langfuse's native `userId`/`sessionId` (read from the
+        // turn span's stamped attributes).
+        if let Some(user) = root.attributes.get("user.id").and_then(Value::as_str) {
+            trace_body["userId"] = json!(user);
+        }
+        if let Some(group) = root.attributes.get("thread.id").and_then(Value::as_str) {
+            trace_body["sessionId"] = json!(group);
+        }
         batch.push(json!({
             "id": new_event_id(),
             "type": "trace-create",
             "timestamp": iso_millis(root.start_unix_ms),
-            "body": {
-                "id": root.trace_id,
-                "name": root.name,
-                "timestamp": iso_millis(root.start_unix_ms),
-            },
+            "body": trace_body,
         }));
     }
 
@@ -120,9 +133,29 @@ pub(crate) fn spans_to_langfuse_batch(spans: &[TraceSpan]) -> Value {
         if let Some(parent) = &span.parent_span_id {
             body["parentObservationId"] = json!(parent);
         }
+        // Prompt/reply content is transmitted only when the caller opted in
+        // (`observability.agent_tracing.capture_content`); otherwise it never
+        // leaves the device even though it may sit on the in-memory span.
+        if include_content {
+            if let Some(input) = &span.input {
+                body["input"] = input.clone();
+            }
+            if let Some(output) = &span.output {
+                body["output"] = output.clone();
+            }
+        }
+        // A span carrying `gen_ai.usage.*` attributes (today only the root turn
+        // span) is emitted as a Langfuse `generation` so the UI renders native
+        // token usage + cost instead of burying them in metadata. Token counts
+        // and cost are non-PII, so this promotion is unconditional.
+        let event_type = if apply_usage_fields(&mut body, span) {
+            "generation-create"
+        } else {
+            "span-create"
+        };
         batch.push(json!({
             "id": new_event_id(),
-            "type": "span-create",
+            "type": event_type,
             "timestamp": iso_millis(span.start_unix_ms),
             "body": body,
         }));
@@ -131,6 +164,46 @@ pub(crate) fn spans_to_langfuse_batch(spans: &[TraceSpan]) -> Value {
     json!({ "batch": batch })
 }
 
+/// Promote a span's `gen_ai.usage.*` / `gen_ai.request.model` attributes into
+/// Langfuse's native `model` / `usageDetails` / `costDetails` fields so the
+/// trace surfaces real token counts and cost (Langfuse only renders these on
+/// `generation` observations). Returns `true` when usage was found, so the
+/// caller emits the span as a `generation-create`. Only token/cost figures are
+/// touched — never prompt text or PII.
+fn apply_usage_fields(body: &mut Value, span: &TraceSpan) -> bool {
+    let attrs = &span.attributes;
+    let input = attrs
+        .get("gen_ai.usage.input_tokens")
+        .and_then(Value::as_u64);
+    let output = attrs
+        .get("gen_ai.usage.output_tokens")
+        .and_then(Value::as_u64);
+    if input.is_none() && output.is_none() {
+        return false;
+    }
+    let input = input.unwrap_or(0);
+    let output = output.unwrap_or(0);
+    let mut usage = Map::new();
+    usage.insert("input".to_string(), json!(input));
+    usage.insert("output".to_string(), json!(output));
+    usage.insert("total".to_string(), json!(input.saturating_add(output)));
+    if let Some(cached) = attrs
+        .get("gen_ai.usage.cached_input_tokens")
+        .and_then(Value::as_u64)
+        .filter(|c| *c > 0)
+    {
+        usage.insert("cache_read_input_tokens".to_string(), json!(cached));
+    }
+    body["usageDetails"] = Value::Object(usage);
+    if let Some(model) = attrs.get("gen_ai.request.model").and_then(Value::as_str) {
+        body["model"] = json!(model);
+    }
+    if let Some(cost) = attrs.get("gen_ai.usage.cost_usd").and_then(Value::as_f64) {
+        body["costDetails"] = json!({ "total": cost });
+    }
+    true
+}
+
 /// Fresh per-event id. Langfuse dedupes ingestion events by this id, so it must
 /// be unique per event (distinct from the observation/trace id in `body`).
 fn new_event_id() -> String {
@@ -152,7 +225,8 @@ pub(crate) async fn push_spans(config: &Config, spans: &[TraceSpan]) -> Result<(
         ));
     }
     let token = require_live_session_token(config)?;
-    let batch = spans_to_langfuse_batch(spans);
+    let include_content = config.observability.agent_tracing.capture_content;
+    let batch = spans_to_langfuse_batch(spans, include_content);
     let span_count = spans.len();
 
     tracing::debug!(
@@ -173,19 +247,36 @@ pub(crate) async fn push_spans(config: &Config, spans: &[TraceSpan]) -> Result<(
         .map_err(|err| format!("POST {url} failed: {err}"))?;
 
     let status = response.status();
-    // Langfuse returns 207 Multi-Status on partial success; `is_success()`
-    // covers the whole 2xx range including that.
-    if status.is_success() {
+    let body = response.text().await.unwrap_or_default();
+    if !status.is_success() {
+        let excerpt: String = body.chars().take(200).collect();
+        return Err(format!("Langfuse ingestion returned {status}: {excerpt}"));
+    }
+    // Langfuse returns 207 Multi-Status even when individual events are rejected
+    // — the failures live in the response `errors` array, not the HTTP status.
+    // Surface them (a partial rejection is logged but never fails the turn).
+    let rejected = serde_json::from_str::<Value>(&body)
+        .ok()
+        .and_then(|v| v.get("errors").and_then(Value::as_array).cloned())
+        .filter(|errs| !errs.is_empty());
+    if let Some(errs) = rejected {
+        let excerpt: String = serde_json::to_string(&errs)
+            .unwrap_or_default()
+            .chars()
+            .take(400)
+            .collect();
+        tracing::warn!(
+            target: LOG_TARGET,
+            "[agent-tracing] Langfuse ({status}) rejected {} of {span_count} span event(s): {excerpt}",
+            errs.len()
+        );
+    } else {
         tracing::debug!(
             target: LOG_TARGET,
             "[agent-tracing] pushed {span_count} spans to Langfuse ({status})"
         );
-        Ok(())
-    } else {
-        let body = response.text().await.unwrap_or_default();
-        let excerpt: String = body.chars().take(200).collect();
-        Err(format!("Langfuse ingestion returned {status}: {excerpt}"))
     }
+    Ok(())
 }
 
 #[cfg(test)]
@@ -217,6 +308,8 @@ mod tests {
             end_unix_ms: end,
             status,
             attributes,
+            input: None,
+            output: None,
         }
     }
 
@@ -273,7 +366,7 @@ mod tests {
                 Some(1_500),
             ),
         ];
-        let payload = spans_to_langfuse_batch(&spans);
+        let payload = spans_to_langfuse_batch(&spans, false);
         let batch = payload["batch"].as_array().expect("batch array");
         assert_eq!(batch.len(), 3, "one trace-create + two span-create");
 
@@ -300,6 +393,81 @@ mod tests {
         assert_ne!(batch[1]["id"], batch[1]["body"]["id"]);
     }
 
+    #[test]
+    fn usage_span_becomes_generation_and_content_is_gated() {
+        let mut turn = span(
+            "trace-1",
+            "root",
+            None,
+            "agent.turn",
+            SpanKind::Turn,
+            SpanStatus::Ok,
+            1_000,
+            Some(2_000),
+        );
+        turn.attributes.clear();
+        turn.attributes
+            .insert("gen_ai.request.model".into(), json!("claude-x"));
+        turn.attributes
+            .insert("gen_ai.usage.input_tokens".into(), json!(100));
+        turn.attributes
+            .insert("gen_ai.usage.output_tokens".into(), json!(20));
+        turn.attributes
+            .insert("gen_ai.usage.cost_usd".into(), json!(0.0123));
+        turn.input = Some(json!("what is 2+2?"));
+        turn.output = Some(json!("4"));
+        let spans = vec![turn];
+
+        // Content OFF (default): span is promoted to a generation with native
+        // usage + cost, but prompt/reply are withheld.
+        let off = spans_to_langfuse_batch(&spans, false);
+        let obs = &off["batch"][1];
+        assert_eq!(obs["type"], "generation-create");
+        assert_eq!(obs["body"]["model"], "claude-x");
+        assert_eq!(obs["body"]["usageDetails"]["input"], 100);
+        assert_eq!(obs["body"]["usageDetails"]["output"], 20);
+        assert_eq!(obs["body"]["usageDetails"]["total"], 120);
+        assert_eq!(obs["body"]["costDetails"]["total"], 0.0123);
+        assert!(
+            obs["body"].get("input").is_none(),
+            "prompt must be withheld when capture_content is off"
+        );
+        assert!(obs["body"].get("output").is_none());
+
+        // Content ON: prompt/reply included, usage/cost unchanged.
+        let on = spans_to_langfuse_batch(&spans, true);
+        let obs = &on["batch"][1];
+        assert_eq!(obs["type"], "generation-create");
+        assert_eq!(obs["body"]["input"], "what is 2+2?");
+        assert_eq!(obs["body"]["output"], "4");
+        assert_eq!(obs["body"]["costDetails"]["total"], 0.0123);
+    }
+
+    #[test]
+    fn trace_create_carries_user_and_session_grouping() {
+        // The turn span's user.id / thread.id attributes are promoted onto the
+        // trace-create as Langfuse userId / sessionId so per-turn traces group
+        // under one conversation and attribute to a user.
+        let mut turn = span(
+            "trace:req-1",
+            "root",
+            None,
+            "agent.turn",
+            SpanKind::Turn,
+            SpanStatus::Ok,
+            1_000,
+            Some(2_000),
+        );
+        turn.attributes.insert("user.id".into(), json!("client-7"));
+        turn.attributes
+            .insert("thread.id".into(), json!("thread-abc"));
+        let payload = spans_to_langfuse_batch(&[turn], false);
+        let trace = &payload["batch"][0];
+        assert_eq!(trace["type"], "trace-create");
+        assert_eq!(trace["body"]["userId"], "client-7");
+        assert_eq!(trace["body"]["sessionId"], "thread-abc");
+    }
+
     #[tokio::test]
     async fn empty_spans_push_is_ok_noop() {
         let config = Config::default();

src/openhuman/agent/progress_tracing/tests.rs

@@ -52,6 +52,7 @@ fn tracing_config_round_trips_through_json() {
         enabled: true,
         backend: AgentTracingBackend::Langfuse,
         export_path: Some("/tmp/spans.ndjson".to_string()),
+        capture_content: false,
     };
     let s = serde_json::to_string(&cfg).unwrap();
     let back: AgentTracingConfig = serde_json::from_str(&s).unwrap();
@@ -570,6 +571,7 @@ fn export_disabled_is_a_noop_and_writes_nothing() {
         enabled: false,
         backend: AgentTracingBackend::Otel,
         export_path: Some(path.to_string_lossy().to_string()),
+        capture_content: false,
     };
     export_spans(&cfg, &one_turn_spans());
     assert!(
@@ -588,6 +590,7 @@ fn export_appends_ndjson_to_the_configured_file() {
         enabled: true,
         backend: AgentTracingBackend::Otel,
         export_path: Some(path.to_string_lossy().to_string()),
+        capture_content: false,
     };
     let spans = one_turn_spans();
     export_spans(&cfg, &spans);
@@ -611,6 +614,7 @@ fn export_with_no_path_does_not_panic() {
         enabled: true,
         backend: AgentTracingBackend::Langfuse,
         export_path: None,
+        capture_content: false,
     };
     export_spans(&cfg, &one_turn_spans());
     export_spans(&cfg, &[]); // empty slice short-circuits.
@@ -643,6 +647,7 @@ async fn export_run_trace_otel_backend_uses_local_sink() {
         enabled: true,
         backend: AgentTracingBackend::Otel,
         export_path: Some(path.to_string_lossy().to_string()),
+        capture_content: false,
     };
     export_run_trace(&config, &one_turn_spans()).await;
 
@@ -653,3 +658,61 @@ async fn export_run_trace_otel_backend_uses_local_sink() {
     );
     let _ = std::fs::remove_dir_all(&dir);
 }
+
+// ── Route A: content + grouping + span-id uniqueness ────────────────────────
+
+#[test]
+fn turn_content_attaches_input_output_to_turn_span() {
+    let c = collect(&[
+        (AgentProgress::TurnStarted, 1_000),
+        (
+            AgentProgress::TurnContent {
+                input: Some("what is your favorite color?".to_string()),
+                output: Some("i'm partial to teal".to_string()),
+            },
+            1_100,
+        ),
+    ]);
+    let turn = find(c.spans(), "agent.turn");
+    assert_eq!(
+        turn.input.as_ref().and_then(|v| v.as_str()),
+        Some("what is your favorite color?"),
+        "TurnContent input must land on the turn span"
+    );
+    assert_eq!(
+        turn.output.as_ref().and_then(|v| v.as_str()),
+        Some("i'm partial to teal")
+    );
+}
+
+#[test]
+fn turn_span_stamps_user_and_thread_grouping_attributes() {
+    let mut c = SpanCollector::new(
+        TraceContext::new("trace:req-1", Some("client-7".to_string()))
+            .with_session_group("thread-abc"),
+    );
+    c.record(&AgentProgress::TurnStarted, 1_000);
+    let turn = find(c.spans(), "agent.turn");
+    assert_eq!(
+        turn.attributes.get("user.id").and_then(|v| v.as_str()),
+        Some("client-7")
+    );
+    assert_eq!(
+        turn.attributes.get("thread.id").and_then(|v| v.as_str()),
+        Some("thread-abc"),
+        "session_group must be stamped as thread.id for the Langfuse sessionId"
+    );
+}
+
+#[test]
+fn span_ids_are_unique_across_turns() {
+    // Two separate collectors (two turns) must not reuse span ids, or Langfuse
+    // dedupes their observations onto whichever trace claimed the id first.
+    let a = collect(&[(AgentProgress::TurnStarted, 1)]);
+    let b = collect(&[(AgentProgress::TurnStarted, 1)]);
+    assert_ne!(
+        find(a.spans(), "agent.turn").span_id,
+        find(b.spans(), "agent.turn").span_id,
+        "span ids must be globally unique across turns"
+    );
+}

src/openhuman/channels/providers/web/progress_bridge.rs

@@ -160,11 +160,15 @@ pub(crate) fn spawn_progress_bridge(
             use crate::openhuman::agent::progress_tracing::{
                 trace_session_id, SpanCollector, TraceContext,
             };
-            let session_id = trace_session_id(metadata.session_id, &thread_id);
-            Some(SpanCollector::new(TraceContext::new(
-                session_id,
-                Some(client_id.clone()),
-            )))
+            // One trace per turn: the trace id is unique per request, while the
+            // thread id rides along as the Langfuse `sessionId` so a
+            // conversation's per-turn traces still group under one session.
+            let base = trace_session_id(metadata.session_id, &thread_id);
+            let trace_id = format!("{base}:{request_id}");
+            Some(SpanCollector::new(
+                TraceContext::new(trace_id, Some(client_id.clone()))
+                    .with_session_group(thread_id.clone()),
+            ))
         } else {
             None
         };
@@ -1080,6 +1084,10 @@ pub(crate) fn spawn_progress_bridge(
                          total_usd={total_usd:.4} client_id={client_id} thread_id={thread_id}"
                     );
                 }
+                AgentProgress::TurnContent { .. } => {
+                    // Prompt/reply content is attached to the trace span by the
+                    // span collector above; the ledger/telemetry bridge ignores it.
+                }
             }
         }
         turn_state.finish();

src/openhuman/config/schema/load/env_overlay.rs

@@ -531,6 +531,21 @@ impl Config {
                 _ => {}
             }
         }
+
+        // Opt-in: export prompt/reply content on trace spans (default off — a
+        // deliberate PII reversal). Token/cost export is unaffected by this flag.
+        if let Some(flag) = env.get("OPENHUMAN_AGENT_TRACING_CAPTURE_CONTENT") {
+            let normalized = flag.trim().to_ascii_lowercase();
+            match normalized.as_str() {
+                "1" | "true" | "yes" | "on" => {
+                    self.observability.agent_tracing.capture_content = true
+                }
+                "0" | "false" | "no" | "off" => {
+                    self.observability.agent_tracing.capture_content = false
+                }
+                _ => {}
+            }
+        }
     }
 
     fn apply_learning_env<E: super::env::EnvLookup + ?Sized>(&mut self, env: &E) {

src/openhuman/config/schema/load_tests.rs

@@ -524,6 +524,25 @@ impl EnvLookup for HashMapEnv {
     }
 }
 
+#[test]
+fn env_overlay_toggles_agent_tracing_capture_content() {
+    // Off by default; the opt-in env var enables prompt/reply export.
+    let mut cfg = Config::default();
+    assert!(!cfg.observability.agent_tracing.capture_content);
+    cfg.apply_env_overlay_with(
+        &HashMapEnv::new().with("OPENHUMAN_AGENT_TRACING_CAPTURE_CONTENT", "true"),
+    );
+    assert!(cfg.observability.agent_tracing.capture_content);
+
+    // An explicit falsy value turns it back off.
+    let mut cfg = Config::default();
+    cfg.observability.agent_tracing.capture_content = true;
+    cfg.apply_env_overlay_with(
+        &HashMapEnv::new().with("OPENHUMAN_AGENT_TRACING_CAPTURE_CONTENT", "off"),
+    );
+    assert!(!cfg.observability.agent_tracing.capture_content);
+}
+
 #[test]
 fn env_overlay_model_only_honours_namespaced_var() {
     // Both set → OPENHUMAN_MODEL wins; bare MODEL is ignored even when

src/openhuman/config/schema/observability.rs

@@ -81,6 +81,13 @@ pub struct AgentTracingConfig {
     /// spans are emitted to the application log at `info` level instead, so
     /// the export still works on read-only or sandboxed deployments.
     pub export_path: Option<String>,
+
+    /// Opt-in: include the turn's prompt (`input`) and the model's reply
+    /// (`output`) on exported spans. Off by default — this reverses the
+    /// otherwise metadata-only, PII-free posture, so it must be enabled
+    /// deliberately. Token/cost figures are always exported (they carry no PII)
+    /// regardless of this flag.
+    pub capture_content: bool,
 }
 
 impl Default for AgentTracingConfig {
@@ -89,6 +96,7 @@ impl Default for AgentTracingConfig {
             enabled: false,
             backend: AgentTracingBackend::Otel,
             export_path: None,
+            capture_content: false,
         }
     }
 }

src/openhuman/threads/turn_state/mirror.rs

@@ -425,6 +425,11 @@ impl TurnStateMirror {
                 // flush per LLM call — not worth it for telemetry.
                 false
             }
+            AgentProgress::TurnContent { .. } => {
+                // Prompt/reply content is consumed by the tracing exporter, not
+                // the turn-state snapshot; nothing to mirror, no flush.
+                false
+            }
         }
     }
 

src/openhuman/workflows/run_log.rs

@@ -232,7 +232,8 @@ pub fn format_event(ev: &AgentProgress) -> Option<String> {
         | AgentProgress::TaskBoardUpdated { .. }
         | AgentProgress::SubagentTextDelta { .. }
         | AgentProgress::SubagentThinkingDelta { .. }
-        | AgentProgress::SubagentIterationStarted { .. } => return None,
+        | AgentProgress::SubagentIterationStarted { .. }
+        | AgentProgress::TurnContent { .. } => return None,
     };
     Some(format!(
         "{}  {}",
@@ -838,6 +839,12 @@ mod tests {
             iteration: 1
         })
         .is_none());
+        // Content (prompt/reply) rides its own event and is never logged here.
+        assert!(format_event(&AgentProgress::TurnContent {
+            input: Some("secret prompt".into()),
+            output: Some("secret reply".into()),
+        })
+        .is_none());
         let line = format_event(&AgentProgress::ToolCallStarted {
             call_id: "c1".into(),
             tool_name: "codegraph_search".into(),