From ba831160f1642a38d151dc5c35e7ddde63ee5c98 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 3 Jul 2026 22:33:48 +0000 Subject: [PATCH] wip(agent): #4454 partial implementation (agent interrupted by spend limit) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Uncommitted implementation salvaged from the parity workflow worktree after the subagent was killed by the monthly spend limit before it could verify, commit, and open a PR. NOT yet build-verified — needs a cargo check pass and test run before merge. Preserved so the work is not lost. Ref: tinyhumansai/openhuman#4454 Claude-Session: https://claude.ai/code/session_019j5TLsRLHsM3kqAYFyH4hR --- src/openhuman/agent/progress_tracing.rs | 70 +++++++-- .../agent/progress_tracing/langfuse.rs | 93 +++++++++++- src/openhuman/agent/progress_tracing/tests.rs | 138 +++++++++++++++++- .../channels/providers/web/progress_bridge.rs | 4 + 4 files changed, 285 insertions(+), 20 deletions(-) diff --git a/src/openhuman/agent/progress_tracing.rs b/src/openhuman/agent/progress_tracing.rs index 1a0137d98b..21c7aa01a1 100644 --- a/src/openhuman/agent/progress_tracing.rs +++ b/src/openhuman/agent/progress_tracing.rs @@ -142,9 +142,10 @@ pub struct TraceSpan { pub status: SpanStatus, /// Metadata-only attributes (no secrets/PII). pub attributes: BTreeMap, - /// 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`. + /// Optional prompt/input content. Attached **only** when + /// `observability.agent_tracing.capture_content` is enabled — the + /// [`SpanCollector`] gates this at storage time (#4454), so when capture is + /// off (the default) the field stays `None` and no exporter can ever emit it. #[serde(skip_serializing_if = "Option::is_none")] pub input: Option, /// Optional model-reply/output content. Same capture/gating rules as @@ -181,6 +182,13 @@ struct SubagentState { #[derive(Debug)] pub struct SpanCollector { ctx: TraceContext, + /// Privacy gate at the **storage** level: when `false` (the documented + /// default), prompt/reply content from `AgentProgress::TurnContent` is + /// dropped on the floor and never attached to a span — so no exporter + /// (NDJSON file, app log, or Langfuse push) can ever transmit it. This is + /// the single choke point that enforces + /// `observability.agent_tracing.capture_content` for all sinks (#4454). + capture_content: bool, spans: Vec, next_span_seq: u64, /// Per-collector (per-turn) random prefix for minted span ids. Langfuse @@ -202,9 +210,13 @@ pub struct SpanCollector { } impl SpanCollector { - pub fn new(ctx: TraceContext) -> Self { + /// Build a collector for one turn's trace. `capture_content` mirrors + /// `observability.agent_tracing.capture_content`: when `false` (default), + /// prompt/reply text is never stored on any span (see [`Self::capture_content`]). + pub fn new(ctx: TraceContext, capture_content: bool) -> Self { Self { ctx, + capture_content, spans: Vec::new(), next_span_seq: 0, id_prefix: uuid::Uuid::new_v4().simple().to_string(), @@ -625,6 +637,23 @@ impl SpanCollector { } AgentProgress::TurnContent { input, output } => { + // Privacy choke point (#4454): only ever attach prompt/reply to + // the span when content capture is explicitly enabled. When off + // (the default) the content is dropped here so it can NEVER reach + // any exporter — NDJSON file, app log, or Langfuse push. + if !self.capture_content { + log::debug!( + "[agent-tracing] TurnContent dropped at storage \ + (capture_content=false); prompt/reply not attached to span" + ); + return; + } + log::debug!( + "[agent-tracing] TurnContent attached to turn span \ + (capture_content=true, input={}, output={})", + input.is_some(), + output.is_some() + ); // 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 { @@ -715,6 +744,21 @@ fn json_f64(n: f64) -> serde_json::Value { .unwrap_or(serde_json::Value::Null) } +/// Clone `spans` with any prompt/reply content (`input`/`output`) stripped. +/// Used before logging spans to the plaintext app log so PII never lands there, +/// regardless of the `capture_content` gate (#4454). +fn strip_span_content(spans: &[TraceSpan]) -> Vec { + spans + .iter() + .map(|span| { + let mut redacted = span.clone(); + redacted.input = None; + redacted.output = None; + redacted + }) + .collect() +} + /// Serialize spans to NDJSON (one span object per line) in the requested /// backend envelope. Both backends share the [`TraceSpan`] body; Langfuse /// wraps each line with a `{"type":"span-create", ...}` observation envelope @@ -769,13 +813,21 @@ pub(crate) fn export_spans(config: &AgentTracingConfig, spans: &[TraceSpan]) { } } None => { - // No path configured — surface to the log so the export still works - // on read-only / sandboxed deployments. - log::info!( - "[agent-tracing] {} spans (trace_id={}):\n{}", + // No path configured — surface to the app log so the export still + // works on read-only / sandboxed deployments. Two hard rules (#4454): + // 1. NEVER at `info` — span dumps belong at `debug` diagnostics. + // 2. NEVER carry prompt/reply content into the plaintext app log, + // even when `capture_content` is on (the file sink is the + // operator's opt-in content target; the log is not). Strip + // input/output before serializing so no PII can land in logs. + let redacted = strip_span_content(spans); + let metadata_only = spans_to_ndjson(config.backend, &redacted); + log::debug!( + "[agent-tracing] {} spans (trace_id={}) [metadata-only; \ + prompt/reply content withheld from log]:\n{}", spans.len(), spans.first().map(|s| s.trace_id.as_str()).unwrap_or(""), - payload.trim_end() + metadata_only.trim_end() ); } } diff --git a/src/openhuman/agent/progress_tracing/langfuse.rs b/src/openhuman/agent/progress_tracing/langfuse.rs index 2772b3c8c6..deade4f811 100644 --- a/src/openhuman/agent/progress_tracing/langfuse.rs +++ b/src/openhuman/agent/progress_tracing/langfuse.rs @@ -183,15 +183,24 @@ fn apply_usage_fields(body: &mut Value, span: &TraceSpan) -> bool { } let input = input.unwrap_or(0); let output = output.unwrap_or(0); + // Langfuse aggregates `usageDetails` keys as **disjoint** buckets and + // independently trusts the explicit `total`. `input_tokens` is inclusive of + // the cached read tokens (see cost.rs:171), so emitting `input` (full) PLUS + // `cache_read_input_tokens` double-counts the cache and makes the components + // exceed `total`. Split them so the buckets are disjoint and sum to `total`: + // input (uncached) + cache_read_input_tokens + output == total (#4454). + let cached = attrs + .get("gen_ai.usage.cached_input_tokens") + .and_then(Value::as_u64) + .filter(|c| *c > 0); + let uncached_input = input.saturating_sub(cached.unwrap_or(0)); let mut usage = Map::new(); - usage.insert("input".to_string(), json!(input)); + usage.insert("input".to_string(), json!(uncached_input)); usage.insert("output".to_string(), json!(output)); + // `total` stays the full accounting figure (full input + output); the + // disjoint components (uncached input + cached + output) reconstruct it. 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) - { + if let Some(cached) = cached { usage.insert("cache_read_input_tokens".to_string(), json!(cached)); } body["usageDetails"] = Value::Object(usage); @@ -468,6 +477,78 @@ mod tests { assert_eq!(trace["body"]["sessionId"], "thread-abc"); } + #[test] + fn usage_details_are_disjoint_and_sum_to_total() { + // #4454: `input_tokens` is inclusive of cached read tokens, so the + // usageDetails components (input + cache_read_input_tokens + output) + // must be disjoint and reconstruct `total` — no double-count. + 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.usage.input_tokens".into(), json!(1_000)); + turn.attributes + .insert("gen_ai.usage.output_tokens".into(), json!(200)); + turn.attributes + .insert("gen_ai.usage.cached_input_tokens".into(), json!(300)); + let batch = spans_to_langfuse_batch(&[turn], false); + let usage = &batch["batch"][1]["body"]["usageDetails"]; + + let uncached = usage["input"].as_u64().unwrap(); + let cached = usage["cache_read_input_tokens"].as_u64().unwrap(); + let output = usage["output"].as_u64().unwrap(); + let total = usage["total"].as_u64().unwrap(); + + // `input` is the uncached remainder, NOT the full input (which would + // double-count the cache against `cache_read_input_tokens`). + assert_eq!(uncached, 700, "input must exclude the cached tokens"); + assert_eq!(cached, 300); + assert_eq!(output, 200); + assert_eq!(total, 1_200, "total is full input (1000) + output (200)"); + // Disjoint components reconstruct the total exactly. + assert_eq!(uncached + cached + output, total); + } + + #[test] + fn usage_details_without_cache_report_full_input() { + // No cached tokens → `input` is the whole input and there is no cache + // bucket; components still sum to total. + 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.usage.input_tokens".into(), json!(100)); + turn.attributes + .insert("gen_ai.usage.output_tokens".into(), json!(20)); + turn.attributes + .insert("gen_ai.usage.cached_input_tokens".into(), json!(0)); + let batch = spans_to_langfuse_batch(&[turn], false); + let usage = &batch["batch"][1]["body"]["usageDetails"]; + assert_eq!(usage["input"], 100); + assert_eq!(usage["output"], 20); + assert_eq!(usage["total"], 120); + assert!( + usage.get("cache_read_input_tokens").is_none(), + "no cache bucket when cached is zero" + ); + } + #[tokio::test] async fn empty_spans_push_is_ok_noop() { let config = Config::default(); diff --git a/src/openhuman/agent/progress_tracing/tests.rs b/src/openhuman/agent/progress_tracing/tests.rs index 86e5b92d0e..e4089679fc 100644 --- a/src/openhuman/agent/progress_tracing/tests.rs +++ b/src/openhuman/agent/progress_tracing/tests.rs @@ -11,7 +11,14 @@ fn ctx() -> TraceContext { } fn collect(events: &[(AgentProgress, u64)]) -> SpanCollector { - let mut c = SpanCollector::new(ctx()); + // Default helper captures content (capture_content = true) so content-shape + // tests can assert prompt/reply attachment; the gate is exercised explicitly + // by `collect_with_capture` in the privacy tests. + collect_with_capture(events, true) +} + +fn collect_with_capture(events: &[(AgentProgress, u64)], capture_content: bool) -> SpanCollector { + let mut c = SpanCollector::new(ctx(), capture_content); for (event, ts) in events { c.record(event, *ts); } @@ -377,7 +384,7 @@ fn subagent_failure_records_error_without_raw_text() { #[test] fn unknown_subagent_task_ids_are_ignored() { // Completion/tool events with no matching spawn must not panic or spawn. - let mut c = SpanCollector::new(ctx()); + let mut c = SpanCollector::new(ctx(), false); c.record(&AgentProgress::TurnStarted, 0); c.record( &AgentProgress::SubagentCompleted { @@ -453,7 +460,7 @@ fn tool_arguments_are_never_serialized() { #[test] fn first_event_lazily_opens_the_turn_span() { // Stream that begins mid-flight (no TurnStarted) still correlates. - let mut c = SpanCollector::new(ctx()); + let mut c = SpanCollector::new(ctx(), false); c.record( &AgentProgress::IterationStarted { iteration: 4, @@ -493,7 +500,7 @@ fn finish_seals_all_open_spans_idempotently() { #[test] fn cost_update_before_turn_start_lazily_opens_root() { - let mut c = SpanCollector::new(ctx()); + let mut c = SpanCollector::new(ctx(), false); c.record( &AgentProgress::TurnCostUpdated { model: "m".to_string(), @@ -520,7 +527,7 @@ fn trace_session_id_prefers_ui_session_else_thread() { #[test] fn no_user_attribution_omits_user_id() { - let mut c = SpanCollector::new(TraceContext::new("anon-1", None)); + let mut c = SpanCollector::new(TraceContext::new("anon-1", None), false); c.record(&AgentProgress::TurnStarted, 0); let turn = find(c.spans(), "agent.turn"); assert!(turn.attributes.get("user.id").is_none()); @@ -691,6 +698,7 @@ 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"), + false, ); c.record(&AgentProgress::TurnStarted, 1_000); let turn = find(c.spans(), "agent.turn"); @@ -717,3 +725,123 @@ fn span_ids_are_unique_across_turns() { "span ids must be globally unique across turns" ); } + +// ── #4454: capture_content gate at span storage ───────────────────────────── + +const SECRET_PROMPT: &str = "SECRET-PROMPT-my-bank-pin-is-1234"; +const SECRET_REPLY: &str = "SECRET-REPLY-transfer-all-funds"; + +fn turn_with_content(capture_content: bool) -> Vec { + let mut c = collect_with_capture( + &[ + (AgentProgress::TurnStarted, 1_000), + ( + AgentProgress::TurnContent { + input: Some(SECRET_PROMPT.to_string()), + output: Some(SECRET_REPLY.to_string()), + }, + 1_100, + ), + (AgentProgress::TurnCompleted { iterations: 1 }, 1_200), + ], + capture_content, + ); + c.finish(1_200); + c.into_spans() +} + +#[test] +fn capture_content_off_drops_prompt_reply_at_storage() { + // With the default gate off, TurnContent is dropped at the storage choke + // point — the span carries no input/output at all. + let spans = turn_with_content(false); + let turn = find(&spans, "agent.turn"); + assert!( + turn.input.is_none(), + "prompt must not be stored when capture_content is off" + ); + assert!( + turn.output.is_none(), + "reply must not be stored when capture_content is off" + ); +} + +#[test] +fn capture_content_on_stores_prompt_reply() { + // Opt-in: content is stored so the operator's configured file sink / Langfuse + // push can carry it. + let spans = turn_with_content(true); + let turn = find(&spans, "agent.turn"); + assert_eq!( + turn.input.as_ref().and_then(|v| v.as_str()), + Some(SECRET_PROMPT) + ); + assert_eq!( + turn.output.as_ref().and_then(|v| v.as_str()), + Some(SECRET_REPLY) + ); +} + +#[test] +fn ndjson_export_omits_content_when_capture_off() { + // Acceptance criterion: with capture_content off, no prompt/reply text + // appears in the NDJSON export payload (either backend envelope). + let spans = turn_with_content(false); + for backend in [AgentTracingBackend::Otel, AgentTracingBackend::Langfuse] { + let out = spans_to_ndjson(backend, &spans); + assert!( + !out.contains(SECRET_PROMPT), + "prompt text leaked into {backend:?} NDJSON export" + ); + assert!( + !out.contains(SECRET_REPLY), + "reply text leaked into {backend:?} NDJSON export" + ); + } +} + +#[test] +fn ndjson_log_payload_never_carries_content_even_when_captured() { + // The no-path (app-log) sink must NEVER carry content, even when + // capture_content is on and the span itself holds it — content is stripped + // before it is serialized for the log. + let spans = turn_with_content(true); + let turn = find(&spans, "agent.turn"); + assert!(turn.input.is_some(), "precondition: span holds content"); + + let redacted = strip_span_content(&spans); + let logged = spans_to_ndjson(AgentTracingBackend::Otel, &redacted); + assert!( + !logged.contains(SECRET_PROMPT) && !logged.contains(SECRET_REPLY), + "app-log payload must never carry prompt/reply content" + ); + // Metadata survives the strip so the log still correlates. + assert!(logged.contains("agent.turn")); + assert!(logged.contains("sess-42")); +} + +#[test] +fn export_to_file_omits_content_when_capture_off() { + // End-to-end file sink: capture_content off → the written NDJSON file has no + // prompt/reply text (the acceptance criterion for the file export path). + let dir = std::env::temp_dir().join(format!("oh-trace-nopii-{}", std::process::id())); + let _ = std::fs::create_dir_all(&dir); + let path = dir.join("spans.ndjson"); + let cfg = AgentTracingConfig { + enabled: true, + backend: AgentTracingBackend::Langfuse, + export_path: Some(path.to_string_lossy().to_string()), + capture_content: false, + }; + export_spans(&cfg, &turn_with_content(false)); + let body = std::fs::read_to_string(&path).unwrap(); + assert!( + !body.contains(SECRET_PROMPT), + "prompt leaked into export file" + ); + assert!( + !body.contains(SECRET_REPLY), + "reply leaked into export file" + ); + let _ = std::fs::remove_dir_all(&dir); +} diff --git a/src/openhuman/channels/providers/web/progress_bridge.rs b/src/openhuman/channels/providers/web/progress_bridge.rs index c41645ae23..365c7c4922 100644 --- a/src/openhuman/channels/providers/web/progress_bridge.rs +++ b/src/openhuman/channels/providers/web/progress_bridge.rs @@ -165,9 +165,13 @@ pub(crate) fn spawn_progress_bridge( // 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}"); + // Content capture is gated at the span-storage level: when + // `capture_content` is off (default), prompt/reply text is never + // attached to a span, so no exporter can transmit it (#4454). Some(SpanCollector::new( TraceContext::new(trace_id, Some(client_id.clone())) .with_session_group(thread_id.clone()), + config.observability.agent_tracing.capture_content, )) } else { None