Add attribution confidence for SFT exports

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

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:<name> / 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)

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,
     }
 }

server/app/heron/src/main.rs

@@ -9,17 +9,17 @@ 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,
 };
 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<Vec<h_pcap_extract::PipelineRoot>> =
-        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<Vec<h_pcap_extract::PipelineRoot>> = 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

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() {

server/app/heron/tests/pipeline_e2e.rs

@@ -15,15 +15,15 @@ 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,
     StorageSinkConfig,
 };
 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<PathBuf> {
     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()

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

server/h-api/src/routes/export.rs

@@ -44,6 +44,9 @@ pub struct ExportTrajectoryParams {
     pub source_id: Option<String>,
     #[serde(default)]
     pub session_id: Option<String>,
+    /// Optional filter: high | medium | ambiguous.
+    #[serde(default)]
+    pub min_attribution_confidence: Option<String>,
 }
 
 /// Mirrors `traces::TracesParams` (the list filters) so a batch export
@@ -71,13 +74,17 @@ pub struct ExportBatchParams {
     /// Max turns to export (default 1000, capped at `MAX_BATCH_TURNS`).
     #[serde(default)]
     pub limit: Option<u32>,
+    /// Optional filter: high | medium | ambiguous.
+    #[serde(default)]
+    pub min_attribution_confidence: Option<String>,
 }
 
 /// Export a single turn or session as one JSONL line.
 pub async fn single(
     State(storage): State<Arc<dyn StorageBackend>>,
     Query(p): Query<ExportTrajectoryParams>,
 ) -> Result<Response<Body>, 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<Arc<dyn StorageBackend>>,
     Query(p): Query<ExportBatchParams>,
 ) -> Result<Response<Body>, 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<dyn StorageBackend>,
     detail: &TraceDetail,
     scope_label: &str,
+    min_confidence: Option<u8>,
 ) -> Result<Result<String, String>, 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<Option<u8>, 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<u8> {
+    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()