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

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,20 +52,33 @@ 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 {
     pub fn new(session_id: impl Into<String>, user_id: Option<String>) -> Self {
         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 {