feat(transcript): emit thread_id + fix orchestrator missing cost (#1169)

src/openhuman/agent/harness/session/transcript.rs

@@ -34,7 +34,7 @@
 //!
 //! **Line 1 (meta):**
 //! ```json
-//! {"_meta":{"agent":"code_executor","dispatcher":"native","created":"...","updated":"...","turn_count":3,"input_tokens":5000,"output_tokens":1200,"cached_input_tokens":3500,"charged_amount_usd":0.0045}}
+//! {"_meta":{"agent":"code_executor","dispatcher":"native","created":"...","updated":"...","turn_count":3,"input_tokens":5000,"output_tokens":1200,"cached_input_tokens":3500,"charged_amount_usd":0.0045,"thread_id":"thr_abc123"}}
 //! ```
 //!
 //! **Message lines:**
@@ -93,6 +93,13 @@ pub struct TranscriptMeta {
     pub cached_input_tokens: u64,
     /// Cumulative amount charged in USD.
     pub charged_amount_usd: f64,
+    /// Backend-side LLM thread identifier (the `thread_id` forwarded on
+    /// `/openai/v1/chat/completions` so the OpenHuman backend can group
+    /// `InferenceLog` entries and align KV-cache keys with the same logical
+    /// chat thread the user sees in the UI). `None` for runs that don't
+    /// originate from a thread-scoped channel (e.g. CLI-only sessions).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub thread_id: Option<String>,
 }
 
 /// A parsed session transcript: metadata + exact message array.
@@ -122,6 +129,8 @@ struct MetaPayload {
     output_tokens: u64,
     cached_input_tokens: u64,
     charged_amount_usd: f64,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    thread_id: Option<String>,
 }
 
 /// One message line in the JSONL — only `role` and `content` are required.
@@ -176,6 +185,7 @@ pub fn write_transcript(
             output_tokens: meta.output_tokens,
             cached_input_tokens: meta.cached_input_tokens,
             charged_amount_usd: meta.charged_amount_usd,
+            thread_id: meta.thread_id.clone(),
         },
     };
     let meta_json =
@@ -338,6 +348,7 @@ fn read_transcript_jsonl(path: &Path) -> Result<SessionTranscript> {
                 output_tokens: mp.output_tokens,
                 cached_input_tokens: mp.cached_input_tokens,
                 charged_amount_usd: mp.charged_amount_usd,
+                thread_id: mp.thread_id,
             });
             continue;
         }
@@ -502,6 +513,9 @@ fn render_markdown(
     let _ = writeln!(buf, "# Session transcript — {}", meta.agent_name);
     buf.push('\n');
     let _ = writeln!(buf, "- Dispatcher: {}", meta.dispatcher);
+    if let Some(tid) = meta.thread_id.as_deref() {
+        let _ = writeln!(buf, "- Thread: `{tid}`");
+    }
     let _ = writeln!(buf, "- Turns: {}", meta.turn_count);
     if meta.input_tokens > 0 || meta.output_tokens > 0 {
         let cache_pct = if meta.input_tokens > 0 {
@@ -615,6 +629,7 @@ fn parse_legacy_meta(raw: &str) -> Result<TranscriptMeta> {
         charged_amount_usd: get("charged_usd")
             .and_then(|s| s.trim_start_matches('$').parse().ok())
             .unwrap_or(0.0),
+        thread_id: get("thread_id").filter(|s| !s.is_empty()),
     })
 }
 

src/openhuman/agent/harness/session/transcript_tests.rs

@@ -24,6 +24,7 @@ fn sample_meta() -> TranscriptMeta {
         output_tokens: 1200,
         cached_input_tokens: 3500,
         charged_amount_usd: 0.0045,
+        thread_id: None,
     }
 }
 
@@ -442,3 +443,52 @@ fn latest_in_dir_prefers_jsonl_over_md() {
         "should prefer .jsonl when both exist at same index"
     );
 }
+
+/// `thread_id` (the backend-side LLM thread identifier) must be both
+/// emitted in the JSONL `_meta` header and surfaced in the `.md`
+/// companion so a human reading the transcript can correlate it with
+/// `InferenceLog` rows on the backend. Sessions without an ambient
+/// thread (CLI, tests) keep `thread_id = None` and neither field
+/// appears — the absence is intentional, not a missing feature.
+#[test]
+fn thread_id_round_trips_and_appears_in_md_when_present() {
+    let dir = TempDir::new().unwrap();
+    let path = dir.path().join("thread.jsonl");
+    let messages = sample_messages();
+    let mut meta = sample_meta();
+    meta.thread_id = Some("thread-xyz-42".into());
+
+    write_transcript(&path, &messages, &meta, None).unwrap();
+
+    // JSONL round-trip preserves the field.
+    let loaded = read_transcript(&path).unwrap();
+    assert_eq!(loaded.meta.thread_id.as_deref(), Some("thread-xyz-42"));
+
+    // Markdown companion exposes it under the header.
+    let md = fs::read_to_string(path.with_extension("md")).unwrap();
+    assert!(
+        md.contains("- Thread: `thread-xyz-42`"),
+        "thread id should be rendered in md header, got:\n{md}"
+    );
+}
+
+#[test]
+fn thread_id_absent_omits_md_line_and_jsonl_field() {
+    let dir = TempDir::new().unwrap();
+    let path = dir.path().join("no_thread.jsonl");
+    let messages = sample_messages();
+    let meta = sample_meta(); // thread_id = None
+
+    write_transcript(&path, &messages, &meta, None).unwrap();
+
+    let raw_jsonl = fs::read_to_string(&path).unwrap();
+    assert!(
+        !raw_jsonl.contains("\"thread_id\""),
+        "absent thread_id must be skipped in JSONL so the field doesn't show up as `null`"
+    );
+    let md = fs::read_to_string(path.with_extension("md")).unwrap();
+    assert!(
+        !md.contains("- Thread:"),
+        "no `- Thread:` line should appear when thread_id is None, got:\n{md}"
+    );
+}

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

@@ -1399,6 +1399,7 @@ impl Agent {
             output_tokens,
             cached_input_tokens,
             charged_amount_usd,
+            thread_id: crate::openhuman::providers::thread_context::current_thread_id(),
         };
 
         if let Err(err) = transcript::write_transcript(path, messages, &meta, turn_usage) {

src/openhuman/agent/harness/subagent_runner/extract_tool.rs

@@ -479,6 +479,7 @@ fn write_extract_transcript(
         output_tokens: 0,
         cached_input_tokens: 0,
         charged_amount_usd: 0.0,
+        thread_id: crate::openhuman::providers::thread_context::current_thread_id(),
     };
 
     if let Err(e) = write_transcript(&path, &messages, &meta, Some(&turn_usage)) {

src/openhuman/agent/harness/subagent_runner/ops.rs

@@ -959,6 +959,7 @@ async fn run_inner_loop(
             output_tokens: usage.output_tokens,
             cached_input_tokens: usage.cached_input_tokens,
             charged_amount_usd: usage.charged_amount_usd,
+            thread_id: crate::openhuman::providers::thread_context::current_thread_id(),
         };
         if let Err(err) = transcript::write_transcript(&path, history, &meta, None) {
             tracing::debug!(

src/openhuman/providers/compatible.rs

@@ -34,8 +34,8 @@ use compatible_parse::{
 use compatible_stream::sse_bytes_to_chunks;
 use compatible_types::{
     ApiChatRequest, ApiChatResponse, ApiUsage, Choice, Function, Message, NativeChatRequest,
-    NativeMessage, OpenHumanMeta, ResponseMessage, ResponsesRequest, StreamChunkResponse,
-    StreamingToolCall, ToolCall,
+    NativeMessage, OpenAiStreamOptions, OpenHumanMeta, ResponseMessage, ResponsesRequest,
+    StreamChunkResponse, StreamingToolCall, ToolCall,
 };
 
 /// A provider that speaks the OpenAI-compatible chat completions API.
@@ -1394,6 +1394,14 @@ impl Provider for OpenAiCompatibleProvider {
                 tool_choice: tools.as_ref().map(|_| "auto".to_string()),
                 tools: tools.clone(),
                 thread_id: self.outbound_thread_id(),
+                // Ask the server for a final usage chunk so token
+                // accounting (and `openhuman.billing.charged_amount_usd`
+                // for the OpenHuman backend) makes it back from
+                // streaming responses — orchestrator sessions otherwise
+                // lose the `- Charged: $…` line in their transcripts.
+                stream_options: Some(OpenAiStreamOptions {
+                    include_usage: true,
+                }),
             };
             let stream_dump_seq = reserve_dump_seq();
             dump_prompt_if_enabled(&self.name, model, stream_dump_seq, &native_request);
@@ -1428,6 +1436,7 @@ impl Provider for OpenAiCompatibleProvider {
             tool_choice: tools.as_ref().map(|_| "auto".to_string()),
             tools,
             thread_id,
+            stream_options: None,
         };
         let dump_seq = reserve_dump_seq();
         dump_prompt_if_enabled(&self.name, model, dump_seq, &native_request);

src/openhuman/providers/compatible_tests.rs

@@ -60,6 +60,7 @@ fn native_request_emits_thread_id_when_present() {
         tools: None,
         tool_choice: None,
         thread_id: Some("thread-abc".to_string()),
+        stream_options: None,
     };
     let json = serde_json::to_value(&req).unwrap();
     assert_eq!(
@@ -76,6 +77,7 @@ fn native_request_emits_thread_id_when_present() {
         tools: None,
         tool_choice: None,
         thread_id: None,
+        stream_options: None,
     };
     let json_no_thread = serde_json::to_value(&req_no_thread).unwrap();
     assert!(
@@ -84,6 +86,53 @@ fn native_request_emits_thread_id_when_present() {
     );
 }
 
+/// Streaming responses arrive without `usage` unless the request asks
+/// for `stream_options.include_usage = true` (OpenAI spec). Without it
+/// the OpenHuman backend's `openhuman.billing` block also never lands,
+/// so transcript headers for orchestrator sessions lose the
+/// `- Charged: $…` line. The non-streaming path stays untouched.
+#[test]
+fn streaming_request_sets_stream_options_include_usage() {
+    let req = super::NativeChatRequest {
+        model: "sonnet".to_string(),
+        messages: Vec::new(),
+        temperature: 0.0,
+        stream: Some(true),
+        tools: None,
+        tool_choice: None,
+        thread_id: None,
+        stream_options: Some(super::compatible_types::OpenAiStreamOptions {
+            include_usage: true,
+        }),
+    };
+    let json = serde_json::to_value(&req).unwrap();
+    assert_eq!(
+        json.pointer("/stream_options/include_usage")
+            .and_then(|v| v.as_bool()),
+        Some(true),
+        "streaming requests must opt into the final usage chunk"
+    );
+}
+
+#[test]
+fn non_streaming_request_omits_stream_options() {
+    let req = super::NativeChatRequest {
+        model: "sonnet".to_string(),
+        messages: Vec::new(),
+        temperature: 0.0,
+        stream: Some(false),
+        tools: None,
+        tool_choice: None,
+        thread_id: None,
+        stream_options: None,
+    };
+    let json = serde_json::to_value(&req).unwrap();
+    assert!(
+        json.get("stream_options").is_none(),
+        "non-streaming requests must not emit stream_options (OpenAI rejects it on stream=false)"
+    );
+}
+
 #[tokio::test]
 async fn outbound_thread_id_is_gated_per_provider() {
     use crate::openhuman::providers::thread_context::with_thread_id;

src/openhuman/providers/compatible_types.rs

@@ -47,6 +47,23 @@ pub(crate) struct NativeChatRequest {
     /// set — see `crate::openhuman::providers::thread_context`).
     #[serde(skip_serializing_if = "Option::is_none")]
     pub(crate) thread_id: Option<String>,
+    /// OpenAI streaming `stream_options`. Set to `{"include_usage": true}`
+    /// on streaming requests so the server emits a final usage chunk
+    /// (carrying token counts and `openhuman.billing.charged_amount_usd`
+    /// when the OpenHuman backend is in front). Without this, streaming
+    /// responses arrive with `usage = None`, transcript headers lose the
+    /// `- Charged: $…` line, and per-message cost annotations vanish for
+    /// streamed sessions (typically the orchestrator).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) stream_options: Option<OpenAiStreamOptions>,
+}
+
+/// OpenAI-spec `stream_options` payload (sent on the wire). Distinct from
+/// `crate::openhuman::providers::traits::StreamOptions`, which is the
+/// caller-side knob set on `ChatRequest` to toggle agent streaming.
+#[derive(Debug, Serialize)]
+pub(crate) struct OpenAiStreamOptions {
+    pub(crate) include_usage: bool,
 }
 
 #[derive(Debug, Serialize)]