From e0ff95ae79e77a2dbc0426976dd2a4bcdfbf8afc Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Mon, 4 May 2026 00:10:26 -0700 Subject: [PATCH] feat(transcript): emit thread_id + fix orchestrator missing cost (#1169) --- .../agent/harness/session/transcript.rs | 17 ++++++- .../agent/harness/session/transcript_tests.rs | 50 +++++++++++++++++++ src/openhuman/agent/harness/session/turn.rs | 1 + .../harness/subagent_runner/extract_tool.rs | 1 + .../agent/harness/subagent_runner/ops.rs | 1 + src/openhuman/providers/compatible.rs | 13 ++++- src/openhuman/providers/compatible_tests.rs | 49 ++++++++++++++++++ src/openhuman/providers/compatible_types.rs | 17 +++++++ 8 files changed, 146 insertions(+), 3 deletions(-) diff --git a/src/openhuman/agent/harness/session/transcript.rs b/src/openhuman/agent/harness/session/transcript.rs index 1ece5bb19..9fca5dbff 100644 --- a/src/openhuman/agent/harness/session/transcript.rs +++ b/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, } /// 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, } /// 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 { 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 { 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()), }) } diff --git a/src/openhuman/agent/harness/session/transcript_tests.rs b/src/openhuman/agent/harness/session/transcript_tests.rs index e419b71e7..21c586a3a 100644 --- a/src/openhuman/agent/harness/session/transcript_tests.rs +++ b/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}" + ); +} diff --git a/src/openhuman/agent/harness/session/turn.rs b/src/openhuman/agent/harness/session/turn.rs index c9a5b964f..4ead0418e 100644 --- a/src/openhuman/agent/harness/session/turn.rs +++ b/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) { diff --git a/src/openhuman/agent/harness/subagent_runner/extract_tool.rs b/src/openhuman/agent/harness/subagent_runner/extract_tool.rs index fdfa7b6f7..a4a89431a 100644 --- a/src/openhuman/agent/harness/subagent_runner/extract_tool.rs +++ b/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)) { diff --git a/src/openhuman/agent/harness/subagent_runner/ops.rs b/src/openhuman/agent/harness/subagent_runner/ops.rs index 52bae5396..8e5302ac9 100644 --- a/src/openhuman/agent/harness/subagent_runner/ops.rs +++ b/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!( diff --git a/src/openhuman/providers/compatible.rs b/src/openhuman/providers/compatible.rs index 3988b8309..ac3cb1e51 100644 --- a/src/openhuman/providers/compatible.rs +++ b/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); diff --git a/src/openhuman/providers/compatible_tests.rs b/src/openhuman/providers/compatible_tests.rs index f8a4387ce..4fbd894dc 100644 --- a/src/openhuman/providers/compatible_tests.rs +++ b/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; diff --git a/src/openhuman/providers/compatible_types.rs b/src/openhuman/providers/compatible_types.rs index ac80183b8..87f0ac063 100644 --- a/src/openhuman/providers/compatible_types.rs +++ b/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, + /// 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, +} + +/// 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)]