diff --git a/docs/configure.md b/docs/configure.md index 2196dfe7..8123e3a3 100644 --- a/docs/configure.md +++ b/docs/configure.md @@ -187,6 +187,43 @@ Keys mirror the `pipeline-health` page's queue cell labels (drop the the others are per-pipeline. Bump the queue named by the first cell that reddens; lower them if memory is tight. +## `[attribution]` — explicit identity labels + +Heron records an attribution label and confidence on every reconstructed +span: + +- `high` — the label came from an eBPF process attribution or from a + configured trusted request header. +- `ambiguous` — Heron reconstructed the call, but the capture point did + not expose a reliable identity label. + +eBPF process attribution wins when present. For passive packet capture, +gateway capture, or cloud-probe traffic where Heron cannot see the local +process, you can ask Heron to trust one or more HTTP request headers as +the attribution label: + +```toml +[attribution] +request_headers = ["x-agent-id", "x-workspace-id"] +``` + +Header matching is case-insensitive and order-preserving; the first +configured header with a non-empty value becomes the span's label. For an +mTLS gateway, configure the gateway to expose the client identity/SAN as a +header first — Heron can only promote identity that is visible in the +captured HTTP metadata. + +SFT export can use this field to avoid training on ambiguous-source +records: + +```bash +curl "http://localhost:3000/api/export/trajectories?start=0&end=9999999999999&min_attribution_confidence=high" +``` + +Accepted filter values are `high`, `medium`, and `ambiguous`. Today Heron +emits `high` and `ambiguous`; `medium` is reserved for future +source-specific correlation strategies. + ## `[storage]` — backend selection ```toml diff --git a/docs/design/07-schema.md b/docs/design/07-schema.md index c9ac634a..4a11cc02 100644 --- a/docs/design/07-schema.md +++ b/docs/design/07-schema.md @@ -62,10 +62,18 @@ spans │ ├── request_body: string? # Complete request JSON │ ├── response_body: string? # Complete response JSON │ ├── request_headers: string? # JSON array of [key, value] pairs -│ └── response_headers: string? # JSON array of [key, value] pairs +│ ├── response_headers: string? # JSON array of [key, value] pairs +│ └── body_bytes_dropped: u64 # Bytes elided by the stored-body cap │ └── Metadata - └── server_ip: string + ├── kind: string # llm today; reserved for future span kinds + ├── server_ip: string + ├── process_pid: u32? # eBPF process attribution + ├── process_comm: string? + ├── process_exe: string? + ├── attribution_label: string? # eBPF process or configured gateway identity + ├── attribution_source: string # ebpf_process / request_header: / unknown + └── attribution_confidence: string # high / medium / ambiguous Indexes: - request_time @@ -79,6 +87,12 @@ Indexes: - **Full body storage**: `request_body` and `response_body` store complete JSON. For streaming responses, `response_body` contains the concatenated final content. - **Headers storage**: `request_headers` and `response_headers` store complete HTTP headers as JSON arrays of `[key, value]` pairs, preserving order and allowing duplicate keys. Rate limit info, request IDs, processing time, etc. can be queried from stored headers without top-level extraction. - **`response_id`**: Wire API's response/message ID (e.g., OpenAI `chatcmpl-xxx`, Anthropic `msg_xxx`). Promoted to top-level for fast cross-referencing with vendor logs. +- **Attribution fields**: `process_*` is native OS process attribution from the + eBPF capture path. `attribution_*` is the export-facing identity label and + provenance: eBPF process attribution wins when available; otherwise a + configured trusted request header may supply a high-confidence gateway + identity. Passive capture without visible identity remains + `attribution_confidence='ambiguous'`. ### `finish_reason` vocabulary @@ -326,6 +340,15 @@ traces). canonical `/api/traces*` and `/api/spans*`. Retention config keys `calls`/`turns` remain accepted as serde aliases for `spans`/`traces`. +### Attribution metadata columns + +DuckDB and ClickHouse add `spans.attribution_label`, +`spans.attribution_source`, and `spans.attribution_confidence` +idempotently on init. Existing rows default to no label, +`source='unknown'`, and `confidence='ambiguous'`, so historical exports +can be filtered conservatively with +`min_attribution_confidence=high`. + ### `AgentTurn` rename (`LlmTurn` → `AgentTurn`) - Table `llm_turns` → `agent_turns` (now further renamed to `traces` — see above) diff --git a/server/app/heron/src/bin/storage_bench.rs b/server/app/heron/src/bin/storage_bench.rs index 33382a8a..7809d33d 100644 --- a/server/app/heron/src/bin/storage_bench.rs +++ b/server/app/heron/src/bin/storage_bench.rs @@ -110,6 +110,7 @@ fn make_call(i: usize, body: &str) -> LlmCall { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, } } diff --git a/server/app/heron/src/main.rs b/server/app/heron/src/main.rs index a0ad26e3..ea077c18 100644 --- a/server/app/heron/src/main.rs +++ b/server/app/heron/src/main.rs @@ -9,10 +9,6 @@ use tracing::level_filters::LevelFilter; use tracing_subscriber::fmt::time::UtcTime; use tracing_subscriber::FmtSubscriber; -use heron::create_backend; -use heron::Pipeline; -use tokio::task::JoinSet; -use tokio_util::sync::CancellationToken; use h_common::config::{ config_search_paths, discover_config_path, AppConfig, CaptureSourceConfig, PipelineDef, }; @@ -20,6 +16,10 @@ use h_common::internal_metrics::{ AggregateHistory, HistoryRecorder, Metric, MetricsReporter, MetricsSystem, }; use h_llm::agent_classifier::ClassifierConfig; +use heron::create_backend; +use heron::Pipeline; +use tokio::task::JoinSet; +use tokio_util::sync::CancellationToken; mod cmd; @@ -298,16 +298,15 @@ async fn run_pipeline(cli: Cli) { config.pipelines.clone() }; - let pcap_extract_roots: std::sync::Arc> = - std::sync::Arc::new( - effective_pipelines - .iter() - .map(|def| h_pcap_extract::PipelineRoot { - name: def.name.clone(), - dump_dir: std::path::PathBuf::from(&def.pcap_dump.dir), - }) - .collect(), - ); + let pcap_extract_roots: std::sync::Arc> = std::sync::Arc::new( + effective_pipelines + .iter() + .map(|def| h_pcap_extract::PipelineRoot { + name: def.name.clone(), + dump_dir: std::path::PathBuf::from(&def.pcap_dump.dir), + }) + .collect(), + ); // Validate no duplicate source_ids across all pipeline sources. { @@ -335,6 +334,7 @@ async fn run_pipeline(cli: Cli) { internal_metrics: config.internal_metrics.clone(), api: config.api.clone(), agent_classifier: config.agent_classifier.clone(), + attribution: config.attribution.clone(), body_cap: config.body_cap, }), config_path: config_path @@ -583,6 +583,7 @@ async fn run_pipeline(cli: Cli) { active_turns.clone(), classifier_cfg, config.body_cap, + config.attribution.clone(), ); // Start each per-pipeline MetricsSystem and, if enabled, one diff --git a/server/app/heron/src/pipeline.rs b/server/app/heron/src/pipeline.rs index 102660aa..2abbdb31 100644 --- a/server/app/heron/src/pipeline.rs +++ b/server/app/heron/src/pipeline.rs @@ -45,6 +45,7 @@ use tokio::sync::mpsc::{self, WeakSender}; use tokio::task::{JoinHandle, JoinSet}; use h_capture::{RawPacket, RoutingSender}; +use h_common::attribution::AttributionConfig; use h_common::config::{CaptureSourceConfig, PipelineDef}; use h_common::internal_metrics::{Metric, MetricsSystem}; use h_llm::agent_classifier::ClassifierConfig; @@ -130,6 +131,7 @@ impl Pipeline { active_turns: h_turn::ActiveTraceRegistry, classifier_cfg: ClassifierConfig, body_cap: h_common::config::BodyCapConfig, + attribution_cfg: AttributionConfig, ) -> Self { // ---- Shared sinks (fan-in across every pipeline) ---- // Each shared channel takes the max of its dedicated config across @@ -342,6 +344,7 @@ impl Pipeline { metrics_sys, classifier_cfg.clone(), body_cap, + attribution_cfg.clone(), ); debug_assert_eq!(llm_handles.len(), flow_shards); for (j, h) in llm_handles.into_iter().enumerate() { diff --git a/server/app/heron/tests/pipeline_e2e.rs b/server/app/heron/tests/pipeline_e2e.rs index c4268fbb..cc15ebcf 100644 --- a/server/app/heron/tests/pipeline_e2e.rs +++ b/server/app/heron/tests/pipeline_e2e.rs @@ -15,8 +15,6 @@ use duckdb::Connection; use tempfile::TempDir; use tokio_util::sync::CancellationToken; -use heron::create_backend; -use heron::Pipeline; use h_capture::{CaptureSource, PcapFileSource}; use h_common::config::{ CaptureSourceConfig, DuckDbConfig, PipelineDef, RetentionConfig, StorageConfig, @@ -24,6 +22,8 @@ use h_common::config::{ }; use h_common::internal_metrics::{Metric, MetricsSystem}; use h_llm::wire_apis as wa; +use heron::create_backend; +use heron::Pipeline; fn fixture(name: &str) -> Option { let root = PathBuf::from(env!("CARGO_MANIFEST_DIR")) @@ -128,6 +128,7 @@ async fn run_pipeline_multi(fixture_names: &[&str]) -> Option<(TempDir, PathBuf) h_turn::new_active_trace_registry(), h_llm::agent_classifier::ClassifierConfig::default(), h_common::config::BodyCapConfig::default(), + h_common::attribution::AttributionConfig::default(), ); let _metrics_svcs: Vec<_> = per_pipeline_metrics .into_iter() diff --git a/server/config/default.toml b/server/config/default.toml index f5766743..ed130e13 100644 --- a/server/config/default.toml +++ b/server/config/default.toml @@ -194,6 +194,17 @@ mcp_tool_prefixes = ["mcp__"] # single source of truth and can't drift again. known_tool_registry = [] +# --------------------------------------------------------------------------- +# Attribution — optional trusted identity labels for passive/gateway capture. +# The first configured request header with a non-empty value becomes the span's +# attribution label with high confidence. For mTLS, configure the gateway to +# expose the client identity/SAN as a header first; Heron can only read identity +# that is visible in captured HTTP metadata. +# --------------------------------------------------------------------------- +[attribution] +# request_headers = ["x-agent-id", "x-workspace-id"] +request_headers = [] + # --------------------------------------------------------------------------- # Stored-body cap — bounds the size of request/response bodies kept for # display and re-classification. Usage / model / agent classification are diff --git a/server/h-api/src/routes/export.rs b/server/h-api/src/routes/export.rs index 789ddac1..f83bcdcb 100644 --- a/server/h-api/src/routes/export.rs +++ b/server/h-api/src/routes/export.rs @@ -44,6 +44,9 @@ pub struct ExportTrajectoryParams { pub source_id: Option, #[serde(default)] pub session_id: Option, + /// Optional filter: high | medium | ambiguous. + #[serde(default)] + pub min_attribution_confidence: Option, } /// Mirrors `traces::TracesParams` (the list filters) so a batch export @@ -71,6 +74,9 @@ pub struct ExportBatchParams { /// Max turns to export (default 1000, capped at `MAX_BATCH_TURNS`). #[serde(default)] pub limit: Option, + /// Optional filter: high | medium | ambiguous. + #[serde(default)] + pub min_attribution_confidence: Option, } /// Export a single turn or session as one JSONL line. @@ -78,6 +84,7 @@ pub async fn single( State(storage): State>, Query(p): Query, ) -> Result, ApiError> { + let min_confidence = parse_min_attribution_confidence(p.min_attribution_confidence.as_deref())?; let (detail, scope_label, filename) = match p.scope.as_str() { "turn" => { let turn_id = p @@ -110,7 +117,7 @@ pub async fn single( other => return Err(ApiError::InvalidParam(format!("invalid scope: {other}"))), }; - match line_from_turn_detail(&storage, &detail, scope_label).await? { + match line_from_turn_detail(&storage, &detail, scope_label, min_confidence).await? { Ok(line) => ndjson_response(line, &filename, None), Err(reason) => Err(ApiError::InvalidParam(reason)), } @@ -121,6 +128,7 @@ pub async fn batch( State(storage): State>, Query(p): Query, ) -> Result, ApiError> { + let min_confidence = parse_min_attribution_confidence(p.min_attribution_confidence.as_deref())?; let server_ports = parse_csv(&p.server_port) .iter() .map(|s| { @@ -152,7 +160,7 @@ pub async fn batch( let Some(detail) = storage.query_trace_by_id(&item.turn_id).await? else { continue; }; - match line_from_turn_detail(&storage, &detail, "turn").await { + match line_from_turn_detail(&storage, &detail, "turn", min_confidence).await { Ok(Ok(line)) => lines.push(line), // skip unsupported wire / unavailable body — don't fail the batch Ok(Err(_reason)) => {} @@ -199,10 +207,24 @@ async fn line_from_turn_detail( storage: &Arc, detail: &TraceDetail, scope_label: &str, + min_confidence: Option, ) -> Result, ApiError> { let Some(final_call_id) = detail.final_call_id.clone() else { return Ok(Err("turn has no terminal call".to_string())); }; + + if let Some(min) = min_confidence { + let calls = storage.query_spans_by_ids(&detail.span_ids, false).await?; + if let Some(call) = calls.iter().find(|call| { + attribution_confidence_rank(&call.attribution_confidence).unwrap_or(0) < min + }) { + return Ok(Err(format!( + "span {} attribution confidence '{}' is below requested minimum", + call.id, call.attribution_confidence + ))); + } + } + let calls = storage .query_spans_by_ids(&[final_call_id.clone()], true) .await?; @@ -240,6 +262,11 @@ async fn line_from_turn_detail( "final_finish_reason": detail.final_finish_reason, "total_input_tokens": detail.total_input_tokens, "total_output_tokens": detail.total_output_tokens, + "attribution": { + "label": call.attribution_label, + "source": call.attribution_source, + "confidence": call.attribution_confidence, + }, "captured_at_ms": detail.end_time, }); @@ -248,6 +275,26 @@ async fn line_from_turn_detail( Ok(Ok(format!("{line}\n"))) } +fn parse_min_attribution_confidence(value: Option<&str>) -> Result, ApiError> { + let Some(value) = value.map(str::trim).filter(|s| !s.is_empty()) else { + return Ok(None); + }; + attribution_confidence_rank(value).map(Some).ok_or_else(|| { + ApiError::InvalidParam(format!( + "invalid min_attribution_confidence: {value} (expected high, medium, ambiguous)" + )) + }) +} + +fn attribution_confidence_rank(value: &str) -> Option { + match value { + v if v.eq_ignore_ascii_case("high") => Some(3), + v if v.eq_ignore_ascii_case("medium") => Some(2), + v if v.eq_ignore_ascii_case("ambiguous") => Some(1), + _ => None, + } +} + /// Build a raw NDJSON attachment response. `counts = (total, written, skipped)` /// adds the `X-Export-*` headers (batch only). fn ndjson_response( @@ -328,6 +375,24 @@ mod tests { assert_eq!(resp.status(), StatusCode::BAD_REQUEST); } + #[test] + fn parses_min_attribution_confidence_filter() { + assert_eq!(super::parse_min_attribution_confidence(None).unwrap(), None); + assert_eq!( + super::parse_min_attribution_confidence(Some("HIGH")).unwrap(), + Some(3) + ); + assert_eq!( + super::parse_min_attribution_confidence(Some("medium")).unwrap(), + Some(2) + ); + assert_eq!( + super::parse_min_attribution_confidence(Some("ambiguous")).unwrap(), + Some(1) + ); + assert!(super::parse_min_attribution_confidence(Some("low")).is_err()); + } + #[tokio::test] async fn batch_over_empty_db_is_empty_ok() { let resp = app() diff --git a/server/h-api/tests/duckdb_routes.rs b/server/h-api/tests/duckdb_routes.rs index e19ab532..dfe5e525 100644 --- a/server/h-api/tests/duckdb_routes.rs +++ b/server/h-api/tests/duckdb_routes.rs @@ -8,12 +8,12 @@ use axum::body::Body; use axum::http::{Request, StatusCode}; -use http_body_util::BodyExt; -use serde_json::Value; -use tower::ServiceExt; use h_api::{router, ApiHealthContext, ApiMetricsContext, ApiRuntimeConfigContext}; use h_metrics::model::LlmFinishMetric; use h_storage_duckdb::DuckDbBackend; +use http_body_util::BodyExt; +use serde_json::Value; +use tower::ServiceExt; fn test_metrics_context() -> ApiMetricsContext { let sys = h_common::internal_metrics::MetricsSystem::new(); @@ -32,6 +32,7 @@ fn test_runtime_config_context() -> ApiRuntimeConfigContext { internal_metrics: h_common::config::InternalMetricsConfig::default(), api: h_common::config::ApiConfig::default(), agent_classifier: h_common::config::ClassifierConfigToml::default(), + attribution: h_common::attribution::AttributionConfig::default(), body_cap: h_common::config::BodyCapConfig::default(), }), config_path: "test".to_string(), @@ -579,9 +580,15 @@ async fn deprecated_aliases_work_and_carry_deprecation_header() { .oneshot(Request::builder().uri(alias).body(Body::empty()).unwrap()) .await .unwrap(); - assert_eq!(resp.status(), StatusCode::OK, "alias {alias} must still serve"); assert_eq!( - resp.headers().get("deprecation").map(|v| v.to_str().unwrap()), + resp.status(), + StatusCode::OK, + "alias {alias} must still serve" + ); + assert_eq!( + resp.headers() + .get("deprecation") + .map(|v| v.to_str().unwrap()), Some("true"), "alias {alias} must carry Deprecation: true", ); diff --git a/server/h-common/src/attribution.rs b/server/h-common/src/attribution.rs new file mode 100644 index 00000000..fc5a7433 --- /dev/null +++ b/server/h-common/src/attribution.rs @@ -0,0 +1,89 @@ +//! Attribution labels and confidence for reconstructed spans. +//! +//! Process attribution is only available from attribution-capable capture +//! sources such as eBPF. Passive packet/gateway capture can still carry an +//! explicit identity when operators expose one in an HTTP header at the capture +//! point. These types keep that provenance explicit so downstream export paths +//! can filter out ambiguous training data. + +use std::fmt; + +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum AttributionConfidence { + High, + Medium, + Ambiguous, +} + +impl AttributionConfidence { + pub fn as_str(self) -> &'static str { + match self { + Self::High => "high", + Self::Medium => "medium", + Self::Ambiguous => "ambiguous", + } + } + + pub fn rank(self) -> u8 { + match self { + Self::High => 3, + Self::Medium => 2, + Self::Ambiguous => 1, + } + } +} + +impl fmt::Display for AttributionConfidence { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +impl Default for AttributionConfidence { + fn default() -> Self { + Self::Ambiguous + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct AttributionInfo { + pub label: Option, + pub source: String, + pub confidence: AttributionConfidence, +} + +impl AttributionInfo { + pub fn ambiguous() -> Self { + Self { + label: None, + source: "unknown".to_string(), + confidence: AttributionConfidence::Ambiguous, + } + } + + pub fn high(label: impl Into, source: impl Into) -> Self { + Self { + label: Some(label.into()), + source: source.into(), + confidence: AttributionConfidence::High, + } + } +} + +impl Default for AttributionInfo { + fn default() -> Self { + Self::ambiguous() + } +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +#[serde(default)] +pub struct AttributionConfig { + /// Request header names whose first non-empty value is trusted as an + /// attribution label for passive/gateway capture. Header matching is + /// case-insensitive and order-preserving. + pub request_headers: Vec, +} diff --git a/server/h-common/src/config.rs b/server/h-common/src/config.rs index 7161170f..80583a50 100644 --- a/server/h-common/src/config.rs +++ b/server/h-common/src/config.rs @@ -4,6 +4,7 @@ use std::path::{Path, PathBuf}; use config::Config; use serde::{Deserialize, Serialize}; +use crate::attribution::AttributionConfig; use crate::error::AppError; /// The ordered list of paths searched for a configuration file when the user @@ -125,6 +126,8 @@ pub struct AppConfig { pub api: ApiConfig, #[serde(default)] pub agent_classifier: ClassifierConfigToml, + /// Explicit attribution controls for passive/gateway capture. + pub attribution: AttributionConfig, /// Stored-body size cap. See [`BodyCapConfig`]. pub body_cap: BodyCapConfig, } @@ -191,6 +194,8 @@ struct RawAppConfig { #[serde(default)] agent_classifier: ClassifierConfigToml, #[serde(default)] + attribution: AttributionConfig, + #[serde(default)] body_cap: BodyCapConfig, } @@ -245,6 +250,7 @@ impl RawAppConfig { internal_metrics: self.internal_metrics, api: self.api, agent_classifier: self.agent_classifier, + attribution: self.attribution, body_cap: self.body_cap, } } diff --git a/server/h-common/src/lib.rs b/server/h-common/src/lib.rs index a081b2f5..e3a523cf 100644 --- a/server/h-common/src/lib.rs +++ b/server/h-common/src/lib.rs @@ -12,6 +12,7 @@ //! compile-time coupling surface small. pub mod agent; +pub mod attribution; pub mod config; pub mod config_edit; pub mod error; diff --git a/server/h-llm/src/agents/claude_cli.rs b/server/h-llm/src/agents/claude_cli.rs index ca0bb208..8e9e2db6 100644 --- a/server/h-llm/src/agents/claude_cli.rs +++ b/server/h-llm/src/agents/claude_cli.rs @@ -15,7 +15,8 @@ const UA_PREFIX: &str = "claude-cli/"; /// background `/v1/messages` call that feeds the running transcript to a /// supervisor prompt and returns a tiny `yes/no` verdict (no `tools` /// field, `stop_sequences=[""]`). It is housekeeping, not conversation. -const SECURITY_MONITOR_SYSTEM_SIG: &str = "You are a security monitor for autonomous AI coding agents"; +const SECURITY_MONITOR_SYSTEM_SIG: &str = + "You are a security monitor for autonomous AI coding agents"; /// Tool-call anchor for session synthesis when the legacy /// `x-claude-code-session-id` header is absent (Claude Code stopped sending it @@ -380,6 +381,7 @@ mod tests { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, }) } diff --git a/server/h-llm/src/agents/codex_cli.rs b/server/h-llm/src/agents/codex_cli.rs index a5f7bcca..66e3b774 100644 --- a/server/h-llm/src/agents/codex_cli.rs +++ b/server/h-llm/src/agents/codex_cli.rs @@ -303,6 +303,7 @@ mod tests { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, }) } diff --git a/server/h-llm/src/agents/generic.rs b/server/h-llm/src/agents/generic.rs index ab23395d..3bba2105 100644 --- a/server/h-llm/src/agents/generic.rs +++ b/server/h-llm/src/agents/generic.rs @@ -326,6 +326,7 @@ mod tests { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, }) } diff --git a/server/h-llm/src/agents/hermes.rs b/server/h-llm/src/agents/hermes.rs index 56f19a86..a8cd704c 100644 --- a/server/h-llm/src/agents/hermes.rs +++ b/server/h-llm/src/agents/hermes.rs @@ -280,6 +280,7 @@ mod tests { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, }) } diff --git a/server/h-llm/src/agents/mod.rs b/server/h-llm/src/agents/mod.rs index f1073909..6a3eeef5 100644 --- a/server/h-llm/src/agents/mod.rs +++ b/server/h-llm/src/agents/mod.rs @@ -100,6 +100,7 @@ mod priority_tests { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, } } diff --git a/server/h-llm/src/agents/openclaw.rs b/server/h-llm/src/agents/openclaw.rs index 61a5aa85..7b186941 100644 --- a/server/h-llm/src/agents/openclaw.rs +++ b/server/h-llm/src/agents/openclaw.rs @@ -380,6 +380,7 @@ mod tests { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, }) } diff --git a/server/h-llm/src/agents/opencode.rs b/server/h-llm/src/agents/opencode.rs index a44886be..22e9db78 100644 --- a/server/h-llm/src/agents/opencode.rs +++ b/server/h-llm/src/agents/opencode.rs @@ -190,6 +190,7 @@ mod tests { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, }) } diff --git a/server/h-llm/src/model.rs b/server/h-llm/src/model.rs index 642210ae..91305c1a 100644 --- a/server/h-llm/src/model.rs +++ b/server/h-llm/src/model.rs @@ -75,6 +75,10 @@ pub struct LlmCall { /// Extraction (usage/model) always runs on the full body upstream, so a /// non-zero value never implies lost metrics — only a truncated stored body. pub body_bytes_dropped: u64, + /// Explicit attribution label and confidence for training/export consumers. + /// eBPF process attribution and configured gateway headers can set this to + /// high confidence; packet-tap rows without such identity remain ambiguous. + pub attribution: h_common::attribution::AttributionInfo, /// Owning process (pid / comm / exe), when the capture source attributes it /// (eBPF). `None` for passive taps (pcap / cloud-probe). Copied from the /// request's process attribution (falling back to the response's) in the @@ -399,6 +403,7 @@ mod extension_tests { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, }; let arc = Arc::new(call); @@ -464,6 +469,7 @@ mod extension_tests { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, }; assert!(call.finish_reason.is_none()); diff --git a/server/h-llm/src/processor.rs b/server/h-llm/src/processor.rs index 972219eb..64c82f46 100644 --- a/server/h-llm/src/processor.rs +++ b/server/h-llm/src/processor.rs @@ -1,10 +1,12 @@ use std::sync::Arc; -use serde_json::Value; +use h_common::attribution::{AttributionConfig, AttributionInfo}; use h_common::config::BodyCapConfig; use h_common::internal_metrics::{Metric, MetricsWorker}; +use h_common::process::ProcessInfo; use h_protocol::joiner::HttpJoinerEvent; use h_protocol::model::{HttpRequestData, HttpResponseData, SseEventData}; +use serde_json::Value; use uuid::Uuid; use crate::agent_classifier::{classify, ClassifierConfig}; @@ -54,6 +56,39 @@ fn cap_body(body: Option, cap: &BodyCapConfig) -> (Option, u64) (Some(out), dropped) } +fn derive_attribution( + process: Option<&ProcessInfo>, + request_headers: &[(String, String)], + cfg: &AttributionConfig, +) -> AttributionInfo { + if let Some(process) = process { + let label = process + .exe + .as_deref() + .filter(|s| !s.trim().is_empty()) + .unwrap_or(process.comm.as_str()); + return AttributionInfo::high(label.to_string(), "ebpf_process"); + } + + for configured in &cfg.request_headers { + let configured = configured.trim(); + if configured.is_empty() { + continue; + } + if let Some((name, value)) = request_headers + .iter() + .find(|(name, value)| name.eq_ignore_ascii_case(configured) && !value.trim().is_empty()) + { + return AttributionInfo::high( + value.trim().to_string(), + format!("request_header:{}", name.trim()), + ); + } + } + + AttributionInfo::ambiguous() +} + /// Processes `HttpJoinerEvent`s and extracts `LlmCall` records. Stateless — /// request/response pairing now lives in `h_protocol::joiner::HttpJoiner`. pub struct LlmProcessor { @@ -70,6 +105,10 @@ pub struct LlmProcessor { /// extraction, so it never affects usage/model/agent accuracy — only how /// much body is persisted. Wired from `AppConfig.body_cap`. body_cap: BodyCapConfig, + /// Explicit attribution config. Used only after a wire API is identified so + /// raw packet capture can promote trusted gateway identity headers to + /// exportable attribution labels. + attribution_cfg: AttributionConfig, } impl LlmProcessor { @@ -100,6 +139,7 @@ impl LlmProcessor { estimator, classifier_cfg: ClassifierConfig::default(), body_cap: BodyCapConfig::default(), + attribution_cfg: AttributionConfig::default(), } } @@ -117,6 +157,12 @@ impl LlmProcessor { self } + /// Builder-style override for attribution config. + pub fn with_attribution_cfg(mut self, cfg: AttributionConfig) -> Self { + self.attribution_cfg = cfg; + self + } + /// Process a single joiner event. Returns LlmEvents (Start and/or /// Complete and/or Heartbeat). pub fn process(&mut self, event: HttpJoinerEvent) -> Vec { @@ -307,6 +353,10 @@ impl LlmProcessor { let (response_body, resp_body_dropped) = cap_body(resp_info.response_body, &self.body_cap); let body_bytes_dropped = req_body_dropped.saturating_add(resp_body_dropped); + let process = request.process.clone().or_else(|| response.process.clone()); + let attribution = + derive_attribution(process.as_ref(), &request.headers, &self.attribution_cfg); + let call = LlmCall { source_id: request.flow_key.source_id.clone(), id: Uuid::now_v7().to_string(), @@ -344,13 +394,11 @@ impl LlmProcessor { tool_call_count: 0, tool_names: vec![], body_bytes_dropped, + attribution, // Process attribution rides on the request (the SSL_write side that // the eBPF probe sees first); fall back to the response in case the // flow only learned it on the inbound direction. - process: request - .process - .clone() - .or_else(|| response.process.clone()), + process, }; let agent = build_agent_call_info_with_parsed( @@ -475,11 +523,11 @@ mod tests { use super::*; use crate::wire_apis as wa; use bytes::Bytes; - use std::net::IpAddr; - use std::sync::Arc; use h_common::internal_metrics::MetricsSystem; use h_protocol::model::{HttpRequestData, HttpResponseData, SseEventData}; use h_protocol::net::FlowKey; + use std::net::IpAddr; + use std::sync::Arc; fn empty_registry() -> Arc { Arc::new(AgentProfileRegistry::new()) @@ -587,6 +635,65 @@ mod tests { } } + #[test] + fn attribution_prefers_ebpf_process_over_configured_header() { + let process = + ProcessInfo::new(42, "node").with_exe(Some("/usr/local/bin/node".to_string())); + let cfg = AttributionConfig { + request_headers: vec!["x-agent-id".to_string()], + }; + let info = derive_attribution( + Some(&process), + &[("x-agent-id".to_string(), "gateway-agent".to_string())], + &cfg, + ); + + assert_eq!(info.label.as_deref(), Some("/usr/local/bin/node")); + assert_eq!(info.source, "ebpf_process"); + assert_eq!( + info.confidence, + h_common::attribution::AttributionConfidence::High + ); + } + + #[test] + fn attribution_uses_configured_header_case_insensitively() { + let cfg = AttributionConfig { + request_headers: vec!["x-agent-id".to_string()], + }; + let info = derive_attribution( + None, + &[("X-Agent-ID".to_string(), "agent-a".to_string())], + &cfg, + ); + + assert_eq!(info.label.as_deref(), Some("agent-a")); + assert_eq!(info.source, "request_header:X-Agent-ID"); + assert_eq!( + info.confidence, + h_common::attribution::AttributionConfidence::High + ); + } + + #[test] + fn attribution_is_ambiguous_without_process_or_configured_header() { + let cfg = AttributionConfig { + request_headers: vec!["x-agent-id".to_string()], + }; + let info = derive_attribution( + None, + &[("x-workspace-id".to_string(), "workspace-a".to_string())], + &cfg, + ); + + assert_eq!(info.label, None); + assert_eq!(info.source, "unknown"); + assert_eq!( + info.confidence, + h_common::attribution::AttributionConfidence::Ambiguous + ); + } + #[test] fn request_detects_and_emits_start() { use serde_json::json; @@ -735,8 +842,14 @@ mod tests { let out = out.unwrap(); assert!(out.starts_with("HEAD"), "head retained: {out}"); assert!(out.ends_with("TAIL"), "tail retained: {out}"); - assert!(out.contains("100 bytes elided"), "elision marker present: {out}"); - assert!(out.len() < 108, "stored body must be smaller than the original"); + assert!( + out.contains("100 bytes elided"), + "elision marker present: {out}" + ); + assert!( + out.len() < 108, + "stored body must be smaller than the original" + ); } #[test] @@ -812,20 +925,33 @@ mod tests { LlmEvent::Complete { call, .. } => { // Wire usage extracted exactly (estimator did NOT run, and the // cap did not corrupt extraction). - assert_eq!(call.input_tokens, Some(1234), "usage survives request-body cap"); + assert_eq!( + call.input_tokens, + Some(1234), + "usage survives request-body cap" + ); assert_eq!(call.output_tokens, Some(56)); assert_eq!(call.total_tokens, Some(1290)); // Request body was sampled. - assert!(call.body_bytes_dropped > 0, "huge request body must be capped"); + assert!( + call.body_bytes_dropped > 0, + "huge request body must be capped" + ); let stored = call.request_body.as_ref().expect("request_body stored"); assert!( stored.len() < original_len, "stored body ({}) must be smaller than original ({original_len})", stored.len() ); - assert!(stored.contains("bytes elided"), "stored body carries the cap marker"); + assert!( + stored.contains("bytes elided"), + "stored body carries the cap marker" + ); // Small response stayed verbatim → all dropped bytes are the request's. - assert_eq!(call.response_body.as_deref(), Some(resp_body.to_string().as_str())); + assert_eq!( + call.response_body.as_deref(), + Some(resp_body.to_string().as_str()) + ); } _ => panic!("expected Complete"), } @@ -1182,6 +1308,7 @@ mod tests { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, } } diff --git a/server/h-llm/src/profile.rs b/server/h-llm/src/profile.rs index d47e1dc4..e7723afc 100644 --- a/server/h-llm/src/profile.rs +++ b/server/h-llm/src/profile.rs @@ -250,6 +250,7 @@ mod tests { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, } } diff --git a/server/h-llm/src/stage.rs b/server/h-llm/src/stage.rs index c66baeb1..f7ef9e95 100644 --- a/server/h-llm/src/stage.rs +++ b/server/h-llm/src/stage.rs @@ -17,6 +17,7 @@ use std::sync::Arc; use tokio::sync::mpsc; use tokio::task::JoinHandle; +use h_common::attribution::AttributionConfig; use h_common::config::BodyCapConfig; use h_common::internal_metrics::{Metric, MetricsSystem}; @@ -73,6 +74,7 @@ pub fn spawn_llm_stage( metrics_sys: &mut MetricsSystem, classifier_cfg: ClassifierConfig, body_cap: BodyCapConfig, + attribution_cfg: AttributionConfig, ) -> Vec> { assert!( !metrics_shard_txs.is_empty(), @@ -93,12 +95,14 @@ pub fn spawn_llm_stage( let metrics_txs = metrics_shard_txs.clone(); let calls_tx = calls_tx.clone(); let classifier_cfg = classifier_cfg.clone(); + let attribution_cfg = attribution_cfg.clone(); let worker_metrics = metrics_sys.register_worker(&format!("llm.{i}"), LLM_WORKER_METRICS); handles.push(tokio::spawn(async move { let shard = i; let mut processor = LlmProcessor::new(wire_apis, reg, worker_metrics.clone()) .with_classifier_cfg(classifier_cfg) - .with_body_cap(body_cap); + .with_body_cap(body_cap) + .with_attribution_cfg(attribution_cfg); let reason = 'main: loop { let event = match rx.recv().await { Some(e) => e, @@ -216,11 +220,11 @@ fn metrics_shard_index(event: &LlmEvent, n: usize) -> usize { mod tests { use super::*; use bytes::Bytes; - use std::net::IpAddr; - use std::sync::Arc; use h_protocol::joiner::HttpJoiner; use h_protocol::model::{HttpParseEvent, HttpRequestData, HttpResponseData}; use h_protocol::net::FlowKey; + use std::net::IpAddr; + use std::sync::Arc; use crate::agents::build_default_registry; use crate::model::{LlmCall, TurnShardInput}; @@ -367,6 +371,7 @@ mod tests { &mut metrics_sys, ClassifierConfig::default(), BodyCapConfig::default(), + AttributionConfig::default(), ); let _svc = metrics_sys.start(); @@ -427,6 +432,7 @@ mod tests { &mut metrics_sys, ClassifierConfig::default(), BodyCapConfig::default(), + AttributionConfig::default(), ); let _svc = metrics_sys.start(); @@ -482,6 +488,7 @@ mod tests { &mut metrics_sys, ClassifierConfig::default(), BodyCapConfig::default(), + AttributionConfig::default(), ); let _svc = metrics_sys.start(); diff --git a/server/h-llm/src/test_support.rs b/server/h-llm/src/test_support.rs index 1eec9585..cd65666d 100644 --- a/server/h-llm/src/test_support.rs +++ b/server/h-llm/src/test_support.rs @@ -44,6 +44,7 @@ pub fn empty_llm_call() -> LlmCall { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, } } diff --git a/server/h-metrics/src/aggregator.rs b/server/h-metrics/src/aggregator.rs index e8b7554f..4eda18b1 100644 --- a/server/h-metrics/src/aggregator.rs +++ b/server/h-metrics/src/aggregator.rs @@ -362,10 +362,10 @@ fn dimension_keys( #[cfg(test)] mod tests { use super::*; - use std::net::IpAddr; - use std::sync::Arc; use h_llm::model::{ApiType, LlmCall, LlmCallStart}; use h_llm::wire_apis as wa; + use std::net::IpAddr; + use std::sync::Arc; fn test_metrics() -> MetricsWorker { use h_common::internal_metrics::MetricsSystem; @@ -444,6 +444,7 @@ mod tests { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, }), agent: None, diff --git a/server/h-metrics/src/bucket.rs b/server/h-metrics/src/bucket.rs index a42749e2..1ab75f85 100644 --- a/server/h-metrics/src/bucket.rs +++ b/server/h-metrics/src/bucket.rs @@ -1,7 +1,7 @@ use std::collections::BTreeMap; -use tdigest::TDigest; use h_llm::model::LlmCall; +use tdigest::TDigest; use crate::model::{LlmFinishMetric, LlmMetric, LlmMetricsBatch}; @@ -319,9 +319,9 @@ impl WindowBucket { #[cfg(test)] mod tests { use super::*; - use std::net::IpAddr; use h_llm::model::{ApiType, LlmCall}; use h_llm::wire_apis as wa; + use std::net::IpAddr; #[test] fn digest_empty() { @@ -401,6 +401,7 @@ mod tests { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, } } diff --git a/server/h-metrics/src/stage.rs b/server/h-metrics/src/stage.rs index beddc0f4..be6deb51 100644 --- a/server/h-metrics/src/stage.rs +++ b/server/h-metrics/src/stage.rs @@ -108,9 +108,9 @@ pub fn spawn_metrics_stage( #[cfg(test)] mod tests { use super::*; - use std::net::IpAddr; use h_llm::model::{AgentCallInfo, ApiType, LlmCall, LlmCallStart}; use h_llm::wire_apis as wa; + use std::net::IpAddr; fn start_event(ts_us: i64, model: &str) -> LlmEvent { LlmEvent::Start(LlmCallStart { @@ -160,6 +160,7 @@ mod tests { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, }), agent: Some(AgentCallInfo { diff --git a/server/h-metrics/tests/dimensions.rs b/server/h-metrics/tests/dimensions.rs index 46380ca9..c9c395e7 100644 --- a/server/h-metrics/tests/dimensions.rs +++ b/server/h-metrics/tests/dimensions.rs @@ -62,6 +62,7 @@ fn make_completed_call(request_time: i64, tool_surface: Option) -> tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, }), agent: None, diff --git a/server/h-storage-clickhouse/src/calls.rs b/server/h-storage-clickhouse/src/calls.rs index 3a9c2bbf..99102a06 100644 --- a/server/h-storage-clickhouse/src/calls.rs +++ b/server/h-storage-clickhouse/src/calls.rs @@ -56,6 +56,9 @@ struct CallListRow { process_pid: Option, process_comm: Option, process_exe: Option, + attribution_label: Option, + attribution_source: String, + attribution_confidence: String, } #[derive(Row, Deserialize)] @@ -94,6 +97,9 @@ struct SpanDetailRow { process_pid: Option, process_comm: Option, process_exe: Option, + attribution_label: Option, + attribution_source: String, + attribution_confidence: String, } #[derive(Row, Deserialize)] @@ -120,6 +126,9 @@ struct TurnCallRow { response_body: Option, request_headers: Option, response_headers: Option, + attribution_label: Option, + attribution_source: String, + attribution_confidence: String, } impl ClickHouseBackend { @@ -145,19 +154,31 @@ impl ClickHouseBackend { // Time-range + filter WHERE. Timestamps and numeric ports are // interpolated (values we control); user string lists go through // `sql_in_list`'s single-quote escaping, valid in ClickHouse. - let mut where_parts = - vec![time_where("request_time", query.time_range.start_us, query.time_range.end_us)]; + let mut where_parts = vec![time_where( + "request_time", + query.time_range.start_us, + query.time_range.end_us, + )]; if !query.filter.wire_apis.is_empty() { - where_parts.push(format!("wire_api IN ({})", sql_in_list(&query.filter.wire_apis))); + where_parts.push(format!( + "wire_api IN ({})", + sql_in_list(&query.filter.wire_apis) + )); } if !query.filter.models.is_empty() { where_parts.push(format!("model IN ({})", sql_in_list(&query.filter.models))); } if !query.filter.server_ips.is_empty() { - where_parts.push(format!("server_ip IN ({})", sql_in_list(&query.filter.server_ips))); + where_parts.push(format!( + "server_ip IN ({})", + sql_in_list(&query.filter.server_ips) + )); } if !query.status_codes.is_empty() { - where_parts.push(format!("status_code IN ({})", join_nums(&query.status_codes))); + where_parts.push(format!( + "status_code IN ({})", + join_nums(&query.status_codes) + )); } if !query.finish_reasons.is_empty() { where_parts.push(format!( @@ -169,7 +190,10 @@ impl ClickHouseBackend { where_parts.push(format!("client_ip IN ({})", sql_in_list(&query.client_ips))); } if !query.server_ports.is_empty() { - where_parts.push(format!("server_port IN ({})", join_nums(&query.server_ports))); + where_parts.push(format!( + "server_port IN ({})", + join_nums(&query.server_ports) + )); } if let Some(substr) = query .request_path_contains @@ -198,7 +222,8 @@ impl ClickHouseBackend { wire_api, model, status_code, is_stream, finish_reason, ttft_ms, e2e_latency_ms, \ input_tokens, output_tokens, client_ip, server_ip, server_port, request_path, \ response_body, is_agent_request, tool_surface, agent_topology, tool_call_count, \ - tool_names_json, process_pid, process_comm, process_exe \ + tool_names_json, process_pid, process_comm, process_exe, \ + attribution_label, attribution_source, attribution_confidence \ FROM spans WHERE {where_sql} \ ORDER BY {} {sort_order} LIMIT {} OFFSET {offset}", query.sort_by, query.page_size, @@ -224,7 +249,8 @@ impl ClickHouseBackend { input_tokens, output_tokens, total_tokens, ttft_ms, e2e_latency_ms, response_id, \ client_ip, client_port, server_ip, server_port, request_body, response_body, \ request_headers, response_headers, is_agent_request, tool_surface, agent_topology, \ - tool_call_count, tool_names_json, process_pid, process_comm, process_exe \ + tool_call_count, tool_names_json, process_pid, process_comm, process_exe, \ + attribution_label, attribution_source, attribution_confidence \ FROM spans WHERE id = '{}' LIMIT 1", escape_str(id) ); @@ -314,7 +340,7 @@ impl ClickHouseBackend { toUnixTimestamp64Milli(complete_time) AS complete_time_ms, \ wire_api, model, status_code, is_stream, finish_reason, ttft_ms, e2e_latency_ms, \ input_tokens, output_tokens, request_path, client_ip, client_port, server_ip, \ - server_port, {body_columns} \ + server_port, {body_columns}, attribution_label, attribution_source, attribution_confidence \ FROM spans WHERE id IN ({}) \ ORDER BY request_time ASC, complete_time ASC", sql_in_list(span_ids), @@ -360,6 +386,9 @@ impl ClickHouseBackend { response_body: r.response_body, request_headers: r.request_headers, response_headers: r.response_headers, + attribution_label: r.attribution_label, + attribution_source: r.attribution_source, + attribution_confidence: r.attribution_confidence, } }) .collect(); @@ -415,6 +444,9 @@ fn call_list_item(r: CallListRow) -> SpanListItem { tool_call_count: r.tool_call_count, tool_names, process, + attribution_label: r.attribution_label, + attribution_source: r.attribution_source, + attribution_confidence: r.attribution_confidence, } } @@ -457,5 +489,8 @@ fn call_detail(r: SpanDetailRow) -> SpanDetail { tool_call_count: r.tool_call_count, tool_names, process, + attribution_label: r.attribution_label, + attribution_source: r.attribution_source, + attribution_confidence: r.attribution_confidence, } } diff --git a/server/h-storage-clickhouse/src/it.rs b/server/h-storage-clickhouse/src/it.rs index c991b20b..c70c0abf 100644 --- a/server/h-storage-clickhouse/src/it.rs +++ b/server/h-storage-clickhouse/src/it.rs @@ -146,6 +146,7 @@ pub(crate) mod fixtures { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, } } diff --git a/server/h-storage-clickhouse/src/rows.rs b/server/h-storage-clickhouse/src/rows.rs index f104684b..b60f75c1 100644 --- a/server/h-storage-clickhouse/src/rows.rs +++ b/server/h-storage-clickhouse/src/rows.rs @@ -61,6 +61,9 @@ pub(crate) struct CallRow { /// column is forward-looking for wire-visible tool spans. Tail field to /// match the `spans` table column order (and the DuckDB layout). pub kind: String, + pub attribution_label: Option, + pub attribution_source: String, + pub attribution_confidence: String, } impl From for CallRow { @@ -106,6 +109,9 @@ impl From for CallRow { process_comm: c.process.as_ref().map(|p| p.comm.clone()), process_exe: c.process.as_ref().and_then(|p| p.exe.clone()), kind: "llm".into(), + attribution_label: c.attribution.label, + attribution_source: c.attribution.source, + attribution_confidence: c.attribution.confidence.to_string(), } } } diff --git a/server/h-storage-clickhouse/src/schema.rs b/server/h-storage-clickhouse/src/schema.rs index 171ad0f4..273065c9 100644 --- a/server/h-storage-clickhouse/src/schema.rs +++ b/server/h-storage-clickhouse/src/schema.rs @@ -67,7 +67,10 @@ CREATE TABLE IF NOT EXISTS spans ( process_pid Nullable(UInt32), process_comm Nullable(String), process_exe Nullable(String), - kind String DEFAULT 'llm' + kind String DEFAULT 'llm', + attribution_label Nullable(String), + attribution_source String DEFAULT 'unknown', + attribution_confidence String DEFAULT 'ambiguous' ) ENGINE = MergeTree ORDER BY (request_time, id) "; @@ -250,6 +253,15 @@ pub(crate) async fn init(backend: &ClickHouseBackend) -> Result<()> { backend .exec("ALTER TABLE spans ADD COLUMN IF NOT EXISTS kind String DEFAULT 'llm'") .await?; + backend + .exec("ALTER TABLE spans ADD COLUMN IF NOT EXISTS attribution_label Nullable(String)") + .await?; + backend + .exec("ALTER TABLE spans ADD COLUMN IF NOT EXISTS attribution_source String DEFAULT 'unknown'") + .await?; + backend + .exec("ALTER TABLE spans ADD COLUMN IF NOT EXISTS attribution_confidence String DEFAULT 'ambiguous'") + .await?; } // Step 2: create every table on the db-scoped client. Idempotent. diff --git a/server/h-storage-duckdb/src/calls.rs b/server/h-storage-duckdb/src/calls.rs index 1c50d5ef..7538626d 100644 --- a/server/h-storage-duckdb/src/calls.rs +++ b/server/h-storage-duckdb/src/calls.rs @@ -53,6 +53,9 @@ struct PreparedCall { process_pid: Option, process_comm: Option, process_exe: Option, + attribution_label: Option, + attribution_source: String, + attribution_confidence: String, } fn prepare_call(call: LlmCall) -> PreparedCall { @@ -99,6 +102,9 @@ fn prepare_call(call: LlmCall) -> PreparedCall { process_pid: call.process.as_ref().map(|p| p.pid), process_comm: call.process.as_ref().map(|p| p.comm.clone()), process_exe: call.process.as_ref().and_then(|p| p.exe.clone()), + attribution_label: call.attribution.label, + attribution_source: call.attribution.source, + attribution_confidence: call.attribution.confidence.to_string(), } } @@ -165,7 +171,10 @@ fn read_calls_by_ids_sync( input_tokens, output_tokens, request_path, client_ip, client_port, server_ip, server_port, - {body_columns} + {body_columns}, + attribution_label, + attribution_source, + attribution_confidence FROM spans WHERE id IN ({placeholders}) ORDER BY request_time ASC, complete_time ASC" @@ -260,6 +269,17 @@ fn read_calls_by_ids_sync( response_headers: row .get::<_, Option>(21) .map_err(|e| AppError::Storage(format!("read error: {e}")))?, + attribution_label: row + .get::<_, Option>(22) + .map_err(|e| AppError::Storage(format!("read error: {e}")))?, + attribution_source: row + .get::<_, Option>(23) + .map_err(|e| AppError::Storage(format!("read error: {e}")))? + .unwrap_or_else(|| "unknown".to_string()), + attribution_confidence: row + .get::<_, Option>(24) + .map_err(|e| AppError::Storage(format!("read error: {e}")))? + .unwrap_or_else(|| "ambiguous".to_string()), }); let last = items.last_mut().expect("just pushed"); last.tokens_estimated = derive_tokens_estimated( @@ -342,6 +362,9 @@ impl DuckDbBackend { // LLM call today; written as a literal so the domain // model stays free of a field that has one value. "llm", + p.attribution_label, + p.attribution_source, + p.attribution_confidence, ]) .map_err(|e| AppError::Storage(format!("failed to append call: {e}")))?; } @@ -479,7 +502,8 @@ impl DuckDbBackend { finish_reason, ttft_ms, e2e_latency_ms, input_tokens, output_tokens, \ client_ip, server_ip, server_port, request_path, response_body, \ is_agent_request, tool_surface, agent_topology, tool_call_count, tool_names_json, \ - process_pid, process_comm, process_exe \ + process_pid, process_comm, process_exe, \ + attribution_label, attribution_source, attribution_confidence \ FROM spans WHERE {where_sql} \ ORDER BY {sort_by} {sort_order} \ LIMIT {limit} OFFSET {offset}" @@ -518,6 +542,15 @@ impl DuckDbBackend { .unwrap_or_default(); let process = read_process(row, 22, 23, 24) .map_err(|e| AppError::Storage(format!("read error: {e}")))?; + let attribution_label: Option = row + .get(25) + .map_err(|e| AppError::Storage(format!("read error: {e}")))?; + let attribution_source: Option = row + .get(26) + .map_err(|e| AppError::Storage(format!("read error: {e}")))?; + let attribution_confidence: Option = row + .get(27) + .map_err(|e| AppError::Storage(format!("read error: {e}")))?; items.push(SpanListItem { id: row .get(0) @@ -580,6 +613,10 @@ impl DuckDbBackend { .unwrap_or(0), tool_names, process, + attribution_label, + attribution_source: attribution_source.unwrap_or_else(|| "unknown".to_string()), + attribution_confidence: attribution_confidence + .unwrap_or_else(|| "ambiguous".to_string()), }); } @@ -610,7 +647,8 @@ impl DuckDbBackend { request_headers, response_headers, is_agent_request, tool_surface, agent_topology, tool_call_count, tool_names_json, - process_pid, process_comm, process_exe + process_pid, process_comm, process_exe, + attribution_label, attribution_source, attribution_confidence FROM spans WHERE id = ? LIMIT 1 @@ -631,6 +669,8 @@ impl DuckDbBackend { .as_deref() .and_then(|s| serde_json::from_str(s).ok()) .unwrap_or_default(); + let attribution_source: Option = row.get(35)?; + let attribution_confidence: Option = row.get(36)?; Ok(SpanDetail { id: row.get(0)?, source_id: row.get(1)?, @@ -665,6 +705,10 @@ impl DuckDbBackend { tool_call_count: row.get::<_, Option>(29)?.unwrap_or(0), tool_names, process: read_process(row, 31, 32, 33)?, + attribution_label: row.get(34)?, + attribution_source: attribution_source.unwrap_or_else(|| "unknown".to_string()), + attribution_confidence: attribution_confidence + .unwrap_or_else(|| "ambiguous".to_string()), }) }); @@ -741,11 +785,11 @@ impl DuckDbBackend { #[cfg(test)] mod tests { use crate::DuckDbBackend; - use std::net::IpAddr; use h_llm::model::{ApiType, LlmCall}; use h_llm::wire_apis as wa; use h_storage::query::*; use h_storage::StorageBackend; + use std::net::IpAddr; fn in_memory() -> DuckDbBackend { DuckDbBackend::open(":memory:").unwrap() @@ -793,6 +837,7 @@ mod tests { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, } } diff --git a/server/h-storage-duckdb/src/concurrent_tests.rs b/server/h-storage-duckdb/src/concurrent_tests.rs index 03b783cd..4667d6e9 100644 --- a/server/h-storage-duckdb/src/concurrent_tests.rs +++ b/server/h-storage-duckdb/src/concurrent_tests.rs @@ -50,6 +50,7 @@ fn mk_call(i: usize) -> LlmCall { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, } } diff --git a/server/h-storage-duckdb/src/retention.rs b/server/h-storage-duckdb/src/retention.rs index 130ad1fa..56c4396d 100644 --- a/server/h-storage-duckdb/src/retention.rs +++ b/server/h-storage-duckdb/src/retention.rs @@ -106,14 +106,14 @@ impl DuckDbBackend { #[cfg(test)] mod tests { use crate::DuckDbBackend; - use std::net::IpAddr; - use std::time::{Duration, SystemTime}; use h_llm::model::{ApiType, LlmCall}; use h_llm::wire_apis as wa; use h_metrics::model::{LlmFinishMetric, LlmMetric}; use h_storage::retention::RetentionPolicy; use h_storage::StorageBackend; use h_turn::{Trace, TraceStatus}; + use std::net::IpAddr; + use std::time::{Duration, SystemTime}; fn mk_call(id: &str, request_time_us: i64) -> LlmCall { LlmCall { @@ -151,6 +151,7 @@ mod tests { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, } } diff --git a/server/h-storage-duckdb/src/schema.rs b/server/h-storage-duckdb/src/schema.rs index af06fb94..b7c6e35b 100644 --- a/server/h-storage-duckdb/src/schema.rs +++ b/server/h-storage-duckdb/src/schema.rs @@ -1,8 +1,8 @@ //! DDL constants for the DuckDB-backed schema, plus the `init()` bootstrap //! that creates every table and runs forward-compatible migrations. -use tracing::info; use h_common::error::{AppError, Result}; +use tracing::info; use crate::DuckDbBackend; @@ -45,7 +45,10 @@ CREATE TABLE IF NOT EXISTS spans ( process_pid UINTEGER, process_comm VARCHAR, process_exe VARCHAR, - kind VARCHAR NOT NULL DEFAULT 'llm' + kind VARCHAR NOT NULL DEFAULT 'llm', + attribution_label VARCHAR, + attribution_source VARCHAR NOT NULL DEFAULT 'unknown', + attribution_confidence VARCHAR NOT NULL DEFAULT 'ambiguous' ); "; @@ -464,6 +467,26 @@ pub(crate) async fn init(backend: &DuckDbBackend) -> Result<()> { Err(e) => tracing::info!("phase8 migration: spans.kind add skipped: {e}"), } + // Phase 9: explicit attribution labels and confidence for SFT/export + // consumers. Appended after `kind` in both fresh CREATE and ALTER + // migrations so positional appender writes stay aligned. + let attribution_columns = [ + "ALTER TABLE spans ADD COLUMN IF NOT EXISTS attribution_label VARCHAR;", + "ALTER TABLE spans ADD COLUMN IF NOT EXISTS attribution_source VARCHAR DEFAULT 'unknown';", + "ALTER TABLE spans ADD COLUMN IF NOT EXISTS attribution_confidence VARCHAR DEFAULT 'ambiguous';", + ]; + for stmt in attribution_columns { + match conn.execute_batch(stmt) { + Ok(()) => tracing::debug!( + sql = stmt, + "phase9 migration: spans attribution column added (or already present)" + ), + Err(e) => tracing::info!( + "phase9 migration: spans attribution column add skipped: {e} (sql: {stmt})" + ), + } + } + info!("storage tables initialized"); Ok(()) }) diff --git a/server/h-storage-duckdb/src/turns.rs b/server/h-storage-duckdb/src/turns.rs index f999385d..cefb08bb 100644 --- a/server/h-storage-duckdb/src/turns.rs +++ b/server/h-storage-duckdb/src/turns.rs @@ -782,12 +782,12 @@ impl DuckDbBackend { #[cfg(test)] mod tests { use crate::DuckDbBackend; - use std::net::IpAddr; use h_llm::model::{ApiType, LlmCall}; use h_llm::wire_apis as wa; use h_storage::query::*; use h_storage::StorageBackend; use h_turn::{Trace, TraceStatus}; + use std::net::IpAddr; fn sample_turn( turn_id: &str, @@ -870,6 +870,7 @@ mod tests { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, } } diff --git a/server/h-storage-duckdb/src/util.rs b/server/h-storage-duckdb/src/util.rs index edda61eb..62260c38 100644 --- a/server/h-storage-duckdb/src/util.rs +++ b/server/h-storage-duckdb/src/util.rs @@ -156,6 +156,7 @@ pub(crate) fn extract_full_text( tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, }; let (req, resp) = parse_bodies(&call); @@ -268,6 +269,7 @@ pub(crate) fn extract_full_text_batch( tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, }; let (req, resp) = parse_bodies(&call); diff --git a/server/h-storage-duckdb/tests/migrations.rs b/server/h-storage-duckdb/tests/migrations.rs index 6efda8f7..9a47bbb5 100644 --- a/server/h-storage-duckdb/tests/migrations.rs +++ b/server/h-storage-duckdb/tests/migrations.rs @@ -71,6 +71,7 @@ fn sample_call(id: &str, tool_call_count: u32, body_bytes_dropped: u64) -> LlmCa tool_call_count, tool_names: vec![], body_bytes_dropped, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, } } @@ -718,6 +719,9 @@ async fn phase8_otel_rename_migrates_tables_columns_and_backfills_kind() { // (3) `kind` column added and backfilled to 'llm' on the pre-existing row. let span_cols = column_names(&conn, "spans"); assert!(span_cols.iter().any(|c| c == "kind"), "spans must have kind, got: {span_cols:?}"); + assert!(span_cols.iter().any(|c| c == "attribution_label"), "spans must have attribution_label, got: {span_cols:?}"); + assert!(span_cols.iter().any(|c| c == "attribution_source"), "spans must have attribution_source, got: {span_cols:?}"); + assert!(span_cols.iter().any(|c| c == "attribution_confidence"), "spans must have attribution_confidence, got: {span_cols:?}"); let kind: String = conn .query_row("SELECT kind FROM spans WHERE id = 'call-1'", [], |r| r.get(0)) .expect("migrated span row must be present"); diff --git a/server/h-storage/src/query.rs b/server/h-storage/src/query.rs index 3051462b..44fe01a9 100644 --- a/server/h-storage/src/query.rs +++ b/server/h-storage/src/query.rs @@ -313,6 +313,10 @@ pub struct SpanListItem { /// Owning process (eBPF attribution), or `None` for passive-tap sources. #[serde(default)] pub process: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub attribution_label: Option, + pub attribution_source: String, + pub attribution_confidence: String, } #[derive(Debug, Clone, Serialize)] @@ -564,6 +568,10 @@ pub struct TraceSpanItem { /// the Raw HTTP drawer — no extra fetch needed. pub request_headers: Option, pub response_headers: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub attribution_label: Option, + pub attribution_source: String, + pub attribution_confidence: String, } /// Detail view of an `http_exchanges` row — used by `GET /api/http-exchanges/:id` @@ -927,6 +935,10 @@ pub struct SpanDetail { /// Owning process (eBPF attribution), or `None` for passive-tap sources. #[serde(default)] pub process: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub attribution_label: Option, + pub attribution_source: String, + pub attribution_confidence: String, } #[cfg(test)] diff --git a/server/h-storage/src/sink.rs b/server/h-storage/src/sink.rs index d98d31f4..d7399dfb 100644 --- a/server/h-storage/src/sink.rs +++ b/server/h-storage/src/sink.rs @@ -559,6 +559,7 @@ mod tests { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, } } diff --git a/server/h-turn/src/stage.rs b/server/h-turn/src/stage.rs index 6922654b..f1dbc277 100644 --- a/server/h-turn/src/stage.rs +++ b/server/h-turn/src/stage.rs @@ -133,12 +133,12 @@ pub fn spawn_turn_stage( #[cfg(test)] mod tests { use super::*; - use std::net::IpAddr; - use std::sync::Arc; use h_llm::agents::build_default_registry; use h_llm::model::{AgentCall, AgentCallInfo, ApiType, LlmCall}; use h_llm::wire_apis as wa; use h_llm::wire_apis::build_default_wire_api_registry; + use std::net::IpAddr; + use std::sync::Arc; fn llm_test_metrics() -> h_common::internal_metrics::MetricsWorker { let mut sys = MetricsSystem::new(); @@ -216,6 +216,7 @@ mod tests { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, } } diff --git a/server/h-turn/src/test_support.rs b/server/h-turn/src/test_support.rs index bfb68f04..c640d9cf 100644 --- a/server/h-turn/src/test_support.rs +++ b/server/h-turn/src/test_support.rs @@ -81,6 +81,7 @@ pub fn make_call( tool_call_count, tool_names: tool_names.iter().map(|s| s.to_string()).collect(), body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, }; let agent = AgentCallInfo { diff --git a/server/h-turn/src/tracker.rs b/server/h-turn/src/tracker.rs index 29dfcaca..4f60d389 100644 --- a/server/h-turn/src/tracker.rs +++ b/server/h-turn/src/tracker.rs @@ -1129,6 +1129,7 @@ mod tests { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, } } @@ -1179,6 +1180,7 @@ mod tests { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, } } diff --git a/server/h-turn/tests/corpus_golden.rs b/server/h-turn/tests/corpus_golden.rs index 125b490d..462b4d42 100644 --- a/server/h-turn/tests/corpus_golden.rs +++ b/server/h-turn/tests/corpus_golden.rs @@ -141,6 +141,7 @@ async fn run_pcap_collecting_calls( &mut metrics_sys, h_llm::agent_classifier::ClassifierConfig::default(), h_common::config::BodyCapConfig::default(), + h_common::attribution::AttributionConfig::default(), ); h_turn::spawn_turn_stage( diff --git a/server/h-turn/tests/integration.rs b/server/h-turn/tests/integration.rs index abba191f..3edfe862 100644 --- a/server/h-turn/tests/integration.rs +++ b/server/h-turn/tests/integration.rs @@ -107,6 +107,7 @@ async fn run_pcap_full_sharded( &mut metrics_sys, h_llm::agent_classifier::ClassifierConfig::default(), h_common::config::BodyCapConfig::default(), + h_common::attribution::AttributionConfig::default(), ); h_turn::spawn_turn_stage( @@ -238,6 +239,7 @@ async fn run_pcap_collecting_calls( &mut metrics_sys, h_llm::agent_classifier::ClassifierConfig::default(), h_common::config::BodyCapConfig::default(), + h_common::attribution::AttributionConfig::default(), ); h_turn::spawn_turn_stage( @@ -813,7 +815,7 @@ async fn generic_profile_anthropic_two_call_session() { use h_llm::build_agent_call_info; use h_llm::model::{AgentCall, ApiType, LlmCall}; use h_turn::tracker::{TrackerConfig, TurnTracker}; - use h_turn::{TurnEvent, TraceStatus}; + use h_turn::{TraceStatus, TurnEvent}; fn make_call(req: &str, resp: &str, ts_us: i64, finish: Option<&str>) -> LlmCall { LlmCall { @@ -852,6 +854,7 @@ async fn generic_profile_anthropic_two_call_session() { tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, } } diff --git a/server/h-turn/tests/reorder.rs b/server/h-turn/tests/reorder.rs index d96c490d..345dc1a3 100644 --- a/server/h-turn/tests/reorder.rs +++ b/server/h-turn/tests/reorder.rs @@ -126,6 +126,7 @@ fn anthropic_call( tool_call_count: 0, tool_names: vec![], body_bytes_dropped: 0, + attribution: h_common::attribution::AttributionInfo::ambiguous(), process: None, } }