From 2a9d0704105242118ea24a949a8e79152c3f5ee2 Mon Sep 17 00:00:00 2001
From: pixie-agent <croedig@sba-research.org>
Date: Thu, 2 Jul 2026 08:36:58 +0000
Subject: [PATCH 1/3] =?UTF-8?q?refactor(ae):=20rename=20dx=5Fattack=5Fgrap?=
 =?UTF-8?q?h=20=E2=86=92=20dx=5Fevidence=5Fgraph=20(malignant=20view)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Restacks the dx graph rename onto the #73 /query-runner tip. Renames
across the AE forensic path:

- table  forensic_db.dx_attack_graph          → dx_evidence_graph
- view   dx_attack_graph_malicious            → dx_evidence_graph_malignant
         (medical terminology: malignant, not malicious)
- endpoint /dx/attack_graph                    → /dx/evidence_graph
- Go     WriteAttackGraph/handleDXAttackGraph  → *EvidenceGraph

Done mechanically on #73 so it also covers order_query_test.go (absent
from the original rename branch's lineage). Also fixes two pre-existing
#73 lint nits surfaced by touching the package: gofumpt blank lines in
order_query_test.go and the missing order_query_test.go entry in
controller/BUILD.bazel's controller_test srcs (gazelle).
---
 .../services/adaptive_export/cmd/main.go      |  4 ++--
 .../internal/clickhouse/apply.go              | 14 ++++++-------
 .../internal/clickhouse/apply_test.go         |  4 ++--
 .../internal/clickhouse/ddl.go                |  8 ++++----
 .../internal/clickhouse/schema.sql            | 10 +++++-----
 .../internal/control/server.go                | 20 +++++++++----------
 .../internal/controller/BUILD.bazel           |  5 ++++-
 .../internal/controller/order_query_test.go   |  4 +++-
 8 files changed, 37 insertions(+), 32 deletions(-)

diff --git a/src/vizier/services/adaptive_export/cmd/main.go b/src/vizier/services/adaptive_export/cmd/main.go
index 30db7220f77..97a3b907793 100644
--- a/src/vizier/services/adaptive_export/cmd/main.go
+++ b/src/vizier/services/adaptive_export/cmd/main.go
@@ -576,9 +576,9 @@ func main() {
 		// Wire the controller as the /query runner: dx OrderQuery → one-shot pixie
 		// capture written to forensic_db (write⊇read; entlein/dx#93). When the
 		// operator-side querier is disabled (no PushPixieTables), OrderQuery returns
-		// an error and /query 502s — start/stop + dx_attack_graph still work.
+		// an error and /query 502s — start/stop + dx_evidence_graph still work.
 		ctrlSrv := control.New(activeSet, ctl)
-		ctrlSrv.SetGraphWriter(applier) // dx_attack_graph ingest → ClickHouse
+		ctrlSrv.SetGraphWriter(applier) // dx_evidence_graph ingest → ClickHouse
 		// Bearer-JWT auth on the control surface (CodeRabbit: protect control
 		// endpoints). Same shared lib + signing key the broker/PEM use — dx
 		// attaches the service JWT it already mints. Default-OFF so this can
diff --git a/src/vizier/services/adaptive_export/internal/clickhouse/apply.go b/src/vizier/services/adaptive_export/internal/clickhouse/apply.go
index 84b3afbbcb0..1faeac12bda 100644
--- a/src/vizier/services/adaptive_export/internal/clickhouse/apply.go
+++ b/src/vizier/services/adaptive_export/internal/clickhouse/apply.go
@@ -67,9 +67,9 @@ var OperatorOwnedTables = []string{
 	// dx_evidence_graph UI (px.DataFrame clickhouse_dsn) has a real,
 	// globally-registered table to read. dx emits edges, AE persists.
 	// Not a pixie socket_tracer table → not in PixieTables().
-	"dx_attack_graph",
-	// rule-ins-only VIEW over dx_attack_graph; created AFTER it (depends on it).
-	"dx_attack_graph_malicious",
+	"dx_evidence_graph",
+	// rule-ins-only VIEW over dx_evidence_graph; created AFTER it (depends on it).
+	"dx_evidence_graph_malignant",
 }
 
 // Applier applies operator-owned DDL to a ClickHouse cluster over the
@@ -108,15 +108,15 @@ func (a *Applier) Apply(ctx context.Context) error {
 	return nil
 }
 
-// WriteAttackGraph inserts dx evidence-graph edges into
-// forensic_db.dx_attack_graph. jsonEachRow is newline-delimited JSON objects
+// WriteEvidenceGraph inserts dx evidence-graph edges into
+// forensic_db.dx_evidence_graph. jsonEachRow is newline-delimited JSON objects
 // whose keys are the column names (JSONEachRow; unknown keys are skipped,
 // missing columns default). No-op on empty input.
-func (a *Applier) WriteAttackGraph(ctx context.Context, jsonEachRow []byte) error {
+func (a *Applier) WriteEvidenceGraph(ctx context.Context, jsonEachRow []byte) error {
 	if len(jsonEachRow) == 0 {
 		return nil
 	}
-	_, err := a.c.Insert(ctx, "INSERT INTO forensic_db.dx_attack_graph FORMAT JSONEachRow",
+	_, err := a.c.Insert(ctx, "INSERT INTO forensic_db.dx_evidence_graph FORMAT JSONEachRow",
 		jsonEachRow, chhttp.InsertOptions{})
 	return err
 }
diff --git a/src/vizier/services/adaptive_export/internal/clickhouse/apply_test.go b/src/vizier/services/adaptive_export/internal/clickhouse/apply_test.go
index e108e05540c..d9856d5bb01 100644
--- a/src/vizier/services/adaptive_export/internal/clickhouse/apply_test.go
+++ b/src/vizier/services/adaptive_export/internal/clickhouse/apply_test.go
@@ -59,7 +59,7 @@ func TestApply_ExecutesEveryOperatorOwnedTable(t *testing.T) {
 	}
 	// Spot-check that the SECOND call is for the first OperatorOwnedTables entry,
 	// and that the LAST call is for the last OperatorOwnedTables entry (robust to
-	// new operator-owned tables being appended, e.g. dx_attack_graph).
+	// new operator-owned tables being appended, e.g. dx_evidence_graph).
 	if !strings.Contains(bodies[1], "forensic_db."+OperatorOwnedTables[0]) {
 		t.Fatalf("second DDL not for %s; got: %s", OperatorOwnedTables[0], bodies[1])
 	}
@@ -228,7 +228,7 @@ func TestOperatorOwnedTables_DoesNotIncludeKubescape(t *testing.T) {
 // plugin can auto-DDL them with the wrong schema), then the operator's
 // own write targets in declared order.
 func TestOperatorOwnedTables_TrailingOperatorTables(t *testing.T) {
-	want := []string{"adaptive_attribution", "trigger_watermark", "ae_reconcile", "dx_attack_graph", "dx_attack_graph_malicious"}
+	want := []string{"adaptive_attribution", "trigger_watermark", "ae_reconcile", "dx_evidence_graph", "dx_evidence_graph_malignant"}
 	got := OperatorOwnedTables[len(OperatorOwnedTables)-len(want):]
 	for i, w := range want {
 		if got[i] != w {
diff --git a/src/vizier/services/adaptive_export/internal/clickhouse/ddl.go b/src/vizier/services/adaptive_export/internal/clickhouse/ddl.go
index e4503bb340c..5b658ebcbae 100644
--- a/src/vizier/services/adaptive_export/internal/clickhouse/ddl.go
+++ b/src/vizier/services/adaptive_export/internal/clickhouse/ddl.go
@@ -66,11 +66,11 @@ var KnownTables = []string{
 	"ae_reconcile",
 	// operator-owned dx evidence-graph edge list (read by the Pixie
 	// dx_evidence_graph UI via clickhouse_dsn). NOT a pixie table.
-	"dx_attack_graph",
-	// rule-ins-only VIEW over dx_attack_graph (condition != ''); the
+	"dx_evidence_graph",
+	// rule-ins-only VIEW over dx_evidence_graph (condition != ''); the
 	// dx_evidence_graph UI reads this by default so benign rows are filtered
-	// in ClickHouse, not pulled. Must follow dx_attack_graph (depends on it).
-	"dx_attack_graph_malicious",
+	// in ClickHouse, not pulled. Must follow dx_evidence_graph (depends on it).
+	"dx_evidence_graph_malignant",
 }
 
 // ErrUnknownTable is returned by DDL / Columns when asked for a table
diff --git a/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql b/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql
index 494285b3d12..0f055847e49 100644
--- a/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql
+++ b/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql
@@ -487,7 +487,7 @@ CREATE TABLE IF NOT EXISTS forensic_db.ae_reconcile (
   -- unbounded storage (CodeRabbit). 30d matches the pixie observation tables.
   TTL toDateTime(ts) + INTERVAL 30 DAY DELETE;
 
--- dx_attack_graph — dx evidence-graph edge list: one row per directed hop of an
+-- dx_evidence_graph — dx evidence-graph edge list: one row per directed hop of an
 -- investigation (delivery/egress/execution/exfil/pivot), read by the Pixie
 -- dx_evidence_graph UI via px.DataFrame(clickhouse_dsn=...). Operator-owned
 -- (dx emits the edges, AE persists them); NOT a pixie socket_tracer table.
@@ -498,7 +498,7 @@ CREATE TABLE IF NOT EXISTS forensic_db.ae_reconcile (
 -- event_time". Same convention as kubescape_logs. event_time is nanos, so the
 -- partition/TTL use fromUnixTimestamp64Nano (toDateTime would read ns as seconds
 -- → year ~58e9 → broken partitions; see the soc#225 fix).
-CREATE TABLE IF NOT EXISTS forensic_db.dx_attack_graph (
+CREATE TABLE IF NOT EXISTS forensic_db.dx_evidence_graph (
     investigation_id  String,
     event_time        UInt64,
     hostname          String,
@@ -526,7 +526,7 @@ CREATE TABLE IF NOT EXISTS forensic_db.dx_attack_graph (
   TTL toDateTime(fromUnixTimestamp64Nano(event_time)) + INTERVAL 30 DAY DELETE
   SETTINGS index_granularity = 8192;
 
--- dx_attack_graph_malicious — rule-ins-only view (condition != '') the
+-- dx_evidence_graph_malignant — rule-ins-only view (condition != '') the
 -- dx_evidence_graph UI reads by default so benign rows stay in ClickHouse.
-CREATE VIEW IF NOT EXISTS forensic_db.dx_attack_graph_malicious AS
-  SELECT * FROM forensic_db.dx_attack_graph WHERE `condition` != '';
+CREATE VIEW IF NOT EXISTS forensic_db.dx_evidence_graph_malignant AS
+  SELECT * FROM forensic_db.dx_evidence_graph WHERE `condition` != '';
diff --git a/src/vizier/services/adaptive_export/internal/control/server.go b/src/vizier/services/adaptive_export/internal/control/server.go
index 88e67d76369..1ebce67fd64 100644
--- a/src/vizier/services/adaptive_export/internal/control/server.go
+++ b/src/vizier/services/adaptive_export/internal/control/server.go
@@ -54,16 +54,16 @@ type queryRunner interface {
 }
 
 // graphWriter persists dx evidence-graph edges (newline-delimited JSON,
-// JSONEachRow) to forensic_db.dx_attack_graph. nil → /dx/attack_graph 501s.
+// JSONEachRow) to forensic_db.dx_evidence_graph. nil → /dx/evidence_graph 501s.
 type graphWriter interface {
-	WriteAttackGraph(ctx context.Context, jsonEachRow []byte) error
+	WriteEvidenceGraph(ctx context.Context, jsonEachRow []byte) error
 }
 
 // Server is the control HTTP surface.
 type Server struct {
 	set    exporter
 	runner queryRunner // may be nil; /query then returns 501
-	graph  graphWriter // may be nil; /dx/attack_graph then returns 501
+	graph  graphWriter // may be nil; /dx/evidence_graph then returns 501
 	mux    *http.ServeMux
 	verify func(bearer string) error // nil → auth disabled; set via SetAuth
 }
@@ -76,11 +76,11 @@ func New(set exporter, runner queryRunner) *Server {
 	s.mux.HandleFunc("/export/start", s.handleStart)
 	s.mux.HandleFunc("/export/stop", s.handleStop)
 	s.mux.HandleFunc("/query", s.handleQuery)
-	s.mux.HandleFunc("/dx/attack_graph", s.handleDXAttackGraph)
+	s.mux.HandleFunc("/dx/evidence_graph", s.handleDXEvidenceGraph)
 	return s
 }
 
-// SetGraphWriter wires the dx_attack_graph sink.
+// SetGraphWriter wires the dx_evidence_graph sink.
 func (s *Server) SetGraphWriter(g graphWriter) { s.graph = g }
 
 // SetAuth turns on bearer-JWT auth for the control surface, verified with the
@@ -115,9 +115,9 @@ func (s *Server) Handler() http.Handler {
 	})
 }
 
-// handleDXAttackGraph ingests a JSON array of dx evidence-graph edges and writes
-// them to forensic_db.dx_attack_graph (as JSONEachRow).
-func (s *Server) handleDXAttackGraph(w http.ResponseWriter, r *http.Request) {
+// handleDXEvidenceGraph ingests a JSON array of dx evidence-graph edges and writes
+// them to forensic_db.dx_evidence_graph (as JSONEachRow).
+func (s *Server) handleDXEvidenceGraph(w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodPost {
 		w.WriteHeader(http.StatusMethodNotAllowed)
 		return
@@ -140,7 +140,7 @@ func (s *Server) handleDXAttackGraph(w http.ResponseWriter, r *http.Request) {
 		buf.Write(e)
 		buf.WriteByte('\n')
 	}
-	if err := s.graph.WriteAttackGraph(r.Context(), buf.Bytes()); err != nil {
+	if err := s.graph.WriteEvidenceGraph(r.Context(), buf.Bytes()); err != nil {
 		w.WriteHeader(http.StatusBadGateway)
 		return
 	}
@@ -175,7 +175,7 @@ func (t targetReq) target() anomaly.Target {
 }
 
 // maxControlBodyBytes caps a single control-surface request body. The
-// largest legitimate payload we accept is /dx/attack_graph which is a
+// largest legitimate payload we accept is /dx/evidence_graph which is a
 // JSON array of pre-marshalled JSONEachRow lines — measured live the
 // hottest dx rule-in pass fits in ~256 KiB. 4 MiB is well above that
 // and below the per-pod memory headroom an oversized POST could
diff --git a/src/vizier/services/adaptive_export/internal/controller/BUILD.bazel b/src/vizier/services/adaptive_export/internal/controller/BUILD.bazel
index 5e19fbeaf1e..6024b7ce5c7 100644
--- a/src/vizier/services/adaptive_export/internal/controller/BUILD.bazel
+++ b/src/vizier/services/adaptive_export/internal/controller/BUILD.bazel
@@ -34,7 +34,10 @@ go_library(
 
 pl_go_test(
     name = "controller_test",
-    srcs = ["controller_test.go"],
+    srcs = [
+        "controller_test.go",
+        "order_query_test.go",
+    ],
     embed = [":controller"],
     deps = [
         "//src/vizier/services/adaptive_export/internal/anomaly",
diff --git a/src/vizier/services/adaptive_export/internal/controller/order_query_test.go b/src/vizier/services/adaptive_export/internal/controller/order_query_test.go
index 8ba079c8f13..5519e9d4d25 100644
--- a/src/vizier/services/adaptive_export/internal/controller/order_query_test.go
+++ b/src/vizier/services/adaptive_export/internal/controller/order_query_test.go
@@ -46,6 +46,7 @@ func (s *recordingSink) Write(context.Context, []sink.AttributionRow) error { re
 func (s *recordingSink) QueryActive(context.Context, string) ([]sink.AttributionRow, error) {
 	return nil, nil
 }
+
 func (s *recordingSink) WritePixieRows(_ context.Context, table string, rows []map[string]any) error {
 	if s.werr != nil {
 		return s.werr
@@ -55,6 +56,7 @@ func (s *recordingSink) WritePixieRows(_ context.Context, table string, rows []m
 	s.written[table] += len(rows)
 	return nil
 }
+
 func (s *recordingSink) count(table string) int {
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -104,7 +106,7 @@ func TestOrderQueryWritesRows(t *testing.T) {
 }
 
 // With the operator-side querier disabled, OrderQuery errors (so /query 502s) —
-// start/stop + dx_attack_graph remain usable.
+// start/stop + dx_evidence_graph remain usable.
 func TestOrderQueryNoQuerierErrors(t *testing.T) {
 	snk := newRecordingSink()
 	lo, hi := oqWindow()

From 51d5e18d5157ccb8a2c2f91c42d6b53a401be406 Mon Sep 17 00:00:00 2001
From: pixie-agent <croedig@sba-research.org>
Date: Thu, 2 Jul 2026 14:09:31 +0000
Subject: [PATCH 2/3] =?UTF-8?q?feat(ae):=20add=20/dx/evidence=5Fmanifest?=
 =?UTF-8?q?=20ingest=20=E2=86=92=20forensic=5Fdb.dx=5Fevidence=5Fmanifest?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Closes the AE-side wire gap for dx's §9 completeness contract (stacked on
the dx_evidence_graph rename). dx already POSTs manifests via
aeclient.WriteEvidenceManifest → /dx/evidence_manifest; aeprod26 404s that
path, so manifest_rows_exported=0 / graph_write_failures{kind=manifest}>0.

Adds, mirroring the evidence_graph path:
- control: manifestWriter iface + SetManifestWriter + POST /dx/evidence_manifest
  handler. Accepts one manifest.Manifest per verdict; scalars map to typed
  columns, nested collections (case_window/findings/orders/seeds/chain) are
  rendered as JSON text so the JSONEachRow insert is CH-version independent.
- clickhouse: forensic_db.dx_evidence_manifest table (columns = manifest.Manifest
  JSON tags; event_time nanos + hostname read-path like dx_evidence_graph),
  WriteEvidenceManifest sink, KnownTables + OperatorOwnedTables (created on boot).
- main: wire SetManifestWriter(applier).
- tests: endpoint 501/202/502 + nested-as-text flattening; table-set guard.
---
 .../services/adaptive_export/cmd/main.go      |  3 +-
 .../internal/clickhouse/apply.go              | 18 ++++
 .../internal/clickhouse/apply_test.go         |  2 +-
 .../internal/clickhouse/ddl.go                |  4 +
 .../internal/clickhouse/schema.sql            | 31 ++++++
 .../internal/control/server.go                | 96 ++++++++++++++++++-
 .../internal/control/server_test.go           | 59 ++++++++++++
 7 files changed, 206 insertions(+), 7 deletions(-)

diff --git a/src/vizier/services/adaptive_export/cmd/main.go b/src/vizier/services/adaptive_export/cmd/main.go
index 97a3b907793..dffdf7dbfc8 100644
--- a/src/vizier/services/adaptive_export/cmd/main.go
+++ b/src/vizier/services/adaptive_export/cmd/main.go
@@ -578,7 +578,8 @@ func main() {
 		// operator-side querier is disabled (no PushPixieTables), OrderQuery returns
 		// an error and /query 502s — start/stop + dx_evidence_graph still work.
 		ctrlSrv := control.New(activeSet, ctl)
-		ctrlSrv.SetGraphWriter(applier) // dx_evidence_graph ingest → ClickHouse
+		ctrlSrv.SetGraphWriter(applier)    // dx_evidence_graph ingest → ClickHouse
+		ctrlSrv.SetManifestWriter(applier) // dx_evidence_manifest ingest → ClickHouse
 		// Bearer-JWT auth on the control surface (CodeRabbit: protect control
 		// endpoints). Same shared lib + signing key the broker/PEM use — dx
 		// attaches the service JWT it already mints. Default-OFF so this can
diff --git a/src/vizier/services/adaptive_export/internal/clickhouse/apply.go b/src/vizier/services/adaptive_export/internal/clickhouse/apply.go
index 1faeac12bda..0115795c79a 100644
--- a/src/vizier/services/adaptive_export/internal/clickhouse/apply.go
+++ b/src/vizier/services/adaptive_export/internal/clickhouse/apply.go
@@ -70,6 +70,10 @@ var OperatorOwnedTables = []string{
 	"dx_evidence_graph",
 	// rule-ins-only VIEW over dx_evidence_graph; created AFTER it (depends on it).
 	"dx_evidence_graph_malignant",
+	// dx §9 completeness manifest — one row per verdict naming the evidence rows
+	// dx consulted. Created on boot so POST /dx/evidence_manifest has a target.
+	// Independent of dx_evidence_graph. Not a pixie table → not in PixieTables().
+	"dx_evidence_manifest",
 }
 
 // Applier applies operator-owned DDL to a ClickHouse cluster over the
@@ -121,6 +125,20 @@ func (a *Applier) WriteEvidenceGraph(ctx context.Context, jsonEachRow []byte) er
 	return err
 }
 
+// WriteEvidenceManifest inserts one dx §9 completeness manifest (per verdict)
+// into forensic_db.dx_evidence_manifest. jsonEachRow is a single JSONEachRow
+// object whose keys are the column names; the control handler pre-renders the
+// nested collections (case_window/findings/orders/seeds/chain) as JSON text so
+// the insert is ClickHouse-version independent. No-op on empty input.
+func (a *Applier) WriteEvidenceManifest(ctx context.Context, jsonEachRow []byte) error {
+	if len(jsonEachRow) == 0 {
+		return nil
+	}
+	_, err := a.c.Insert(ctx, "INSERT INTO forensic_db.dx_evidence_manifest FORMAT JSONEachRow",
+		jsonEachRow, chhttp.InsertOptions{})
+	return err
+}
+
 // execute is the DDL primitive — used by Apply for CREATE statements.
 func (a *Applier) execute(ctx context.Context, sql string) error {
 	_, err := a.c.Exec(ctx, sql)
diff --git a/src/vizier/services/adaptive_export/internal/clickhouse/apply_test.go b/src/vizier/services/adaptive_export/internal/clickhouse/apply_test.go
index d9856d5bb01..3c834d11a80 100644
--- a/src/vizier/services/adaptive_export/internal/clickhouse/apply_test.go
+++ b/src/vizier/services/adaptive_export/internal/clickhouse/apply_test.go
@@ -228,7 +228,7 @@ func TestOperatorOwnedTables_DoesNotIncludeKubescape(t *testing.T) {
 // plugin can auto-DDL them with the wrong schema), then the operator's
 // own write targets in declared order.
 func TestOperatorOwnedTables_TrailingOperatorTables(t *testing.T) {
-	want := []string{"adaptive_attribution", "trigger_watermark", "ae_reconcile", "dx_evidence_graph", "dx_evidence_graph_malignant"}
+	want := []string{"adaptive_attribution", "trigger_watermark", "ae_reconcile", "dx_evidence_graph", "dx_evidence_graph_malignant", "dx_evidence_manifest"}
 	got := OperatorOwnedTables[len(OperatorOwnedTables)-len(want):]
 	for i, w := range want {
 		if got[i] != w {
diff --git a/src/vizier/services/adaptive_export/internal/clickhouse/ddl.go b/src/vizier/services/adaptive_export/internal/clickhouse/ddl.go
index 5b658ebcbae..c9513c75b88 100644
--- a/src/vizier/services/adaptive_export/internal/clickhouse/ddl.go
+++ b/src/vizier/services/adaptive_export/internal/clickhouse/ddl.go
@@ -71,6 +71,10 @@ var KnownTables = []string{
 	// dx_evidence_graph UI reads this by default so benign rows are filtered
 	// in ClickHouse, not pulled. Must follow dx_evidence_graph (depends on it).
 	"dx_evidence_graph_malignant",
+	// operator-owned dx §9 completeness manifest — one row per verdict naming
+	// the evidence rows dx consulted (POST /dx/evidence_manifest). NOT a pixie
+	// table. Independent of dx_evidence_graph.
+	"dx_evidence_manifest",
 }
 
 // ErrUnknownTable is returned by DDL / Columns when asked for a table
diff --git a/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql b/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql
index 0f055847e49..92ae3310a76 100644
--- a/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql
+++ b/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql
@@ -530,3 +530,34 @@ CREATE TABLE IF NOT EXISTS forensic_db.dx_evidence_graph (
 -- dx_evidence_graph UI reads by default so benign rows stay in ClickHouse.
 CREATE VIEW IF NOT EXISTS forensic_db.dx_evidence_graph_malignant AS
   SELECT * FROM forensic_db.dx_evidence_graph WHERE `condition` != '';
+
+-- dx_evidence_manifest — the §9 completeness contract: one row per verdict
+-- (ruled_in | metastasis), naming the evidence rows dx consulted so the
+-- validator can join them against what AE persisted (write⊇read, checkable).
+-- Operator-owned (dx emits the manifest via POST /dx/evidence_manifest, AE
+-- persists it); NOT a pixie table. Column names are the manifest.Manifest
+-- JSON tags (dx internal/manifest). Same event_time (unix NANOSECONDS) +
+-- hostname read-path convention as dx_evidence_graph so it is px-readable.
+-- The nested collections (case_window/findings/orders/seeds/chain) are stored
+-- as JSON text in String columns; the control handler pre-renders them so the
+-- JSONEachRow insert is ClickHouse-version independent.
+CREATE TABLE IF NOT EXISTS forensic_db.dx_evidence_manifest (
+    investigation_id  String,
+    event_time        UInt64,
+    hostname          String,
+    `condition`       String,
+    verdict           String,
+    confidence        Float64,
+    posterior         Float64,
+    catalog_version   String,
+    case_window       String,
+    findings          String,
+    orders            String,
+    seeds             String,
+    chain             String,
+    evidence_hash     String
+) ENGINE = MergeTree()
+  ORDER BY (event_time, hostname)
+  PARTITION BY toYYYYMM(fromUnixTimestamp64Nano(event_time))
+  TTL toDateTime(fromUnixTimestamp64Nano(event_time)) + INTERVAL 30 DAY DELETE
+  SETTINGS index_granularity = 8192;
diff --git a/src/vizier/services/adaptive_export/internal/control/server.go b/src/vizier/services/adaptive_export/internal/control/server.go
index 1ebce67fd64..20ae154de20 100644
--- a/src/vizier/services/adaptive_export/internal/control/server.go
+++ b/src/vizier/services/adaptive_export/internal/control/server.go
@@ -59,13 +59,20 @@ type graphWriter interface {
 	WriteEvidenceGraph(ctx context.Context, jsonEachRow []byte) error
 }
 
+// manifestWriter persists one dx §9 completeness manifest per verdict
+// (JSONEachRow) to forensic_db.dx_evidence_manifest. nil → /dx/evidence_manifest 501s.
+type manifestWriter interface {
+	WriteEvidenceManifest(ctx context.Context, jsonEachRow []byte) error
+}
+
 // Server is the control HTTP surface.
 type Server struct {
-	set    exporter
-	runner queryRunner // may be nil; /query then returns 501
-	graph  graphWriter // may be nil; /dx/evidence_graph then returns 501
-	mux    *http.ServeMux
-	verify func(bearer string) error // nil → auth disabled; set via SetAuth
+	set      exporter
+	runner   queryRunner    // may be nil; /query then returns 501
+	graph    graphWriter    // may be nil; /dx/evidence_graph then returns 501
+	manifest manifestWriter // may be nil; /dx/evidence_manifest then returns 501
+	mux      *http.ServeMux
+	verify   func(bearer string) error // nil → auth disabled; set via SetAuth
 }
 
 // New builds the control server. runner may be nil for deployments that
@@ -77,12 +84,16 @@ func New(set exporter, runner queryRunner) *Server {
 	s.mux.HandleFunc("/export/stop", s.handleStop)
 	s.mux.HandleFunc("/query", s.handleQuery)
 	s.mux.HandleFunc("/dx/evidence_graph", s.handleDXEvidenceGraph)
+	s.mux.HandleFunc("/dx/evidence_manifest", s.handleDXEvidenceManifest)
 	return s
 }
 
 // SetGraphWriter wires the dx_evidence_graph sink.
 func (s *Server) SetGraphWriter(g graphWriter) { s.graph = g }
 
+// SetManifestWriter wires the dx_evidence_manifest sink.
+func (s *Server) SetManifestWriter(m manifestWriter) { s.manifest = m }
+
 // SetAuth turns on bearer-JWT auth for the control surface, verified with the
 // SAME shared lib + signing key the vizier broker/PEM use (px.dev/pixie/src/
 // shared/services/utils). dx already mints a service JWT (GenerateJWTForService,
@@ -147,6 +158,81 @@ func (s *Server) handleDXEvidenceGraph(w http.ResponseWriter, r *http.Request) {
 	w.WriteHeader(http.StatusAccepted)
 }
 
+// dxManifest mirrors the wire shape of dx's manifest.Manifest (internal/manifest).
+// Scalars map to typed forensic_db.dx_evidence_manifest columns; the nested
+// collections are held as raw JSON and persisted as JSON text in String columns
+// so the JSONEachRow insert is ClickHouse-version independent.
+type dxManifest struct {
+	InvestigationID string          `json:"investigation_id"`
+	EventTime       int64           `json:"event_time"`
+	Hostname        string          `json:"hostname"`
+	Condition       string          `json:"condition"`
+	Verdict         string          `json:"verdict"`
+	Confidence      float64         `json:"confidence"`
+	Posterior       float64         `json:"posterior"`
+	CatalogVersion  string          `json:"catalog_version"`
+	CaseWindow      json.RawMessage `json:"case_window"`
+	Findings        json.RawMessage `json:"findings"`
+	Orders          json.RawMessage `json:"orders"`
+	Seeds           json.RawMessage `json:"seeds"`
+	Chain           json.RawMessage `json:"chain"`
+	EvidenceHash    string          `json:"evidence_hash"`
+}
+
+// handleDXEvidenceManifest ingests ONE dx completeness manifest (per verdict)
+// and writes it to forensic_db.dx_evidence_manifest as a single JSONEachRow
+// row, with the nested collections rendered as JSON text.
+func (s *Server) handleDXEvidenceManifest(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		w.WriteHeader(http.StatusMethodNotAllowed)
+		return
+	}
+	if s.manifest == nil {
+		w.WriteHeader(http.StatusNotImplemented)
+		return
+	}
+	var m dxManifest
+	if !decode(w, r, &m) {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+	row := map[string]any{
+		"investigation_id": m.InvestigationID,
+		"event_time":       m.EventTime,
+		"hostname":         m.Hostname,
+		"condition":        m.Condition,
+		"verdict":          m.Verdict,
+		"confidence":       m.Confidence,
+		"posterior":        m.Posterior,
+		"catalog_version":  m.CatalogVersion,
+		"case_window":      jsonText(m.CaseWindow),
+		"findings":         jsonText(m.Findings),
+		"orders":           jsonText(m.Orders),
+		"seeds":            jsonText(m.Seeds),
+		"chain":            jsonText(m.Chain),
+		"evidence_hash":    m.EvidenceHash,
+	}
+	line, err := json.Marshal(row)
+	if err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+	if err := s.manifest.WriteEvidenceManifest(r.Context(), append(line, '\n')); err != nil {
+		w.WriteHeader(http.StatusBadGateway)
+		return
+	}
+	w.WriteHeader(http.StatusAccepted)
+}
+
+// jsonText renders a nested JSON value as compact text for a String column;
+// nil/absent/null → "" so the column holds an empty string rather than "null".
+func jsonText(raw json.RawMessage) string {
+	if len(raw) == 0 || string(raw) == "null" {
+		return ""
+	}
+	return string(raw)
+}
+
 // ── wire types ────────────────────────────────────────────────────────
 type targetReq struct {
 	Namespace string `json:"namespace"`
diff --git a/src/vizier/services/adaptive_export/internal/control/server_test.go b/src/vizier/services/adaptive_export/internal/control/server_test.go
index 630e8515f58..02ab9d285e3 100644
--- a/src/vizier/services/adaptive_export/internal/control/server_test.go
+++ b/src/vizier/services/adaptive_export/internal/control/server_test.go
@@ -17,6 +17,8 @@
 package control
 
 import (
+	"context"
+	"encoding/json"
 	"net/http"
 	"net/http/httptest"
 	"strings"
@@ -218,3 +220,60 @@ type fakeErr struct{}
 func (fakeErr) Error() string { return "boom" }
 
 var errFake = fakeErr{}
+
+type fakeManifest struct {
+	got string
+	err error
+}
+
+func (f *fakeManifest) WriteEvidenceManifest(_ context.Context, jsonEachRow []byte) error {
+	f.got = string(jsonEachRow)
+	return f.err
+}
+
+// TestEvidenceManifest: /dx/evidence_manifest is 501 without a writer; with one it
+// persists ONE JSONEachRow row per verdict, scalars as typed columns and the nested
+// collections (findings/case_window/...) rendered as JSON *text* so the insert is
+// ClickHouse-version independent.
+func TestEvidenceManifest(t *testing.T) {
+	srv := New(&fakeExporter{}, nil)
+	if r := do(t, srv, http.MethodPost, "/dx/evidence_manifest", `{"investigation_id":"i1"}`); r.StatusCode != http.StatusNotImplemented {
+		t.Fatalf("no writer: got %d, want 501", r.StatusCode)
+	}
+
+	fm := &fakeManifest{}
+	srv.SetManifestWriter(fm)
+	body := `{"investigation_id":"i1","event_time":1730000000000000000,"hostname":"n1",` +
+		`"verdict":"ruled_in","confidence":0.9,"case_window":[1.0,2.0],` +
+		`"findings":[{"vector":"process"}],"evidence_hash":"h"}`
+	if r := do(t, srv, http.MethodPost, "/dx/evidence_manifest", body); r.StatusCode != http.StatusAccepted {
+		t.Fatalf("got %d, want 202", r.StatusCode)
+	}
+	if !strings.HasSuffix(fm.got, "\n") {
+		t.Fatalf("missing JSONEachRow newline terminator: %q", fm.got)
+	}
+	var row map[string]any
+	if err := json.Unmarshal([]byte(strings.TrimSpace(fm.got)), &row); err != nil {
+		t.Fatalf("row not valid JSON: %v (%s)", err, fm.got)
+	}
+	if row["investigation_id"] != "i1" || row["hostname"] != "n1" || row["verdict"] != "ruled_in" {
+		t.Fatalf("scalar columns wrong: %#v", row)
+	}
+	// event_time is a large nanos int; it must round-trip as an integer, not a float in sci notation.
+	if !strings.Contains(fm.got, `"event_time":1730000000000000000`) {
+		t.Fatalf("event_time not an integer literal: %s", fm.got)
+	}
+	// nested collections persisted as JSON text strings, not raw arrays.
+	if row["findings"] != `[{"vector":"process"}]` {
+		t.Fatalf("findings should be JSON text, got %#v", row["findings"])
+	}
+	if row["case_window"] != `[1.0,2.0]` {
+		t.Fatalf("case_window should be JSON text, got %#v", row["case_window"])
+	}
+
+	// writer failure surfaces as 502.
+	fm.err = errFake
+	if r := do(t, srv, http.MethodPost, "/dx/evidence_manifest", body); r.StatusCode != http.StatusBadGateway {
+		t.Fatalf("writer error: got %d, want 502", r.StatusCode)
+	}
+}

From e77894dbe2c5c1a2e9642c332f52143825d90d31 Mon Sep 17 00:00:00 2001
From: Entlein <eineintlein@gmail.com>
Date: Wed, 8 Jul 2026 11:29:07 +0200
Subject: [PATCH 3/3] feat(ae): reattribute DNS via kubescape process tree
 (pixie#80)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Pixie drops pod attribution for short-lived processes — UDP DNS resolvers
fork, do one lookup, and exit, so upid_to_pod_name yields "" at query time
and dns_events land unattributed and get filtered out of the steered set.
Kubescape captured those pids (and parents) at exec time with the owning
pod; join on it.

- kubescape.PIDIndex: pid -> "<ns>/<pod>" from RuntimeProcessDetails trees
  (top-level + children + ppid), built from the rows the trigger already reads.
- pxl.QueryFor: expose df.pid; for ProcessTreeAttributed tables (dns_events)
  skip the in-pixie pod filter so empty-pod rows survive to the operator.
- controller.enrichProcessTree: reattribute empty-pod rows via the index, then
  target-filter — with no index the unattributed rows are dropped, so a
  pulled-unfiltered DNS window never floods forensic_db. Wired in both the
  OrderQuery (/query) and pushPixieRows paths.
- trigger ProcessTreeIndex(): bounded kubescape_logs read -> BuildPIDIndex;
  wired into controller.Config in main.go.

Validated live on a SOC rig: pixie dns_events pid 882811 (pod="") resolves to
observer via kubescape's tree. Unit + e2e tests cover index, enrichment,
query shape, and no-flood.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 .../services/adaptive_export/cmd/main.go      |  4 +
 .../internal/controller/controller.go         | 36 +++++++
 .../internal/e2e/loadtest_test.go             |  8 +-
 .../internal/kubescape/BUILD.bazel            | 12 ++-
 .../internal/kubescape/enrich.go              | 95 +++++++++++++++++++
 .../internal/kubescape/enrich_test.go         | 65 +++++++++++++
 .../internal/kubescape/pidindex.go            | 94 ++++++++++++++++++
 .../internal/kubescape/pidindex_test.go       | 63 ++++++++++++
 .../adaptive_export/internal/pxl/compile.go   |  4 +
 .../adaptive_export/internal/pxl/queryfor.go  | 47 ++++++---
 .../internal/pxl/queryfor_test.go             |  6 +-
 .../internal/trigger/clickhouse.go            | 44 +++++++++
 12 files changed, 461 insertions(+), 17 deletions(-)
 create mode 100644 src/vizier/services/adaptive_export/internal/kubescape/enrich.go
 create mode 100644 src/vizier/services/adaptive_export/internal/kubescape/enrich_test.go
 create mode 100644 src/vizier/services/adaptive_export/internal/kubescape/pidindex.go
 create mode 100644 src/vizier/services/adaptive_export/internal/kubescape/pidindex_test.go

diff --git a/src/vizier/services/adaptive_export/cmd/main.go b/src/vizier/services/adaptive_export/cmd/main.go
index dffdf7dbfc8..bedec2656b2 100644
--- a/src/vizier/services/adaptive_export/cmd/main.go
+++ b/src/vizier/services/adaptive_export/cmd/main.go
@@ -363,6 +363,10 @@ func main() {
 		MaxInflightQueriesGlobal:  intEnvOrZero(envMaxInflightQueriesGlobal),
 		EmptyResultSkipAfterN:     intEnvOrZero(envEmptyResultSkipAfterN),
 		EmptyResultSkipTTL:        durEnvOrZero(envEmptyResultSkipTTLSec, time.Second),
+		// Source the kubescape process-tree index from the same ClickHouse the
+		// trigger reads, so the operator can reattribute DNS rows pixie leaves
+		// unattributed (short-lived resolver pids). See pixie#80.
+		ProcessTreeIndex: trg.ProcessTreeIndex,
 	}
 	if streamingMode {
 		// Route through the non-blocking notifier — handle() returns
diff --git a/src/vizier/services/adaptive_export/internal/controller/controller.go b/src/vizier/services/adaptive_export/internal/controller/controller.go
index 9045b843fec..986ba19eeca 100644
--- a/src/vizier/services/adaptive_export/internal/controller/controller.go
+++ b/src/vizier/services/adaptive_export/internal/controller/controller.go
@@ -160,6 +160,14 @@ type Config struct {
 	// Used by the rev-3 streaming path to shrink its ActiveSet.
 	// Same contract as OnAttribution: synchronous, non-blocking.
 	OnPrune func(namespace, pod string)
+
+	// ProcessTreeIndex, when non-nil, supplies a pid -> "<ns>/<pod>" index
+	// built from kubescape's recent process trees. It is used to reattribute
+	// rows of ProcessTreeAttributed tables (dns_events) that pixie left with an
+	// empty pod — the short-lived DNS resolver whose pid is unmappable at query
+	// time but which kubescape captured. Nil disables enrichment (rows keep
+	// pixie's own attribution). See kubescape.PIDIndex and pixie#80.
+	ProcessTreeIndex func(context.Context) (kubescape.PIDIndex, error)
 }
 
 func (c *Config) defaulted() Config {
@@ -255,6 +263,30 @@ func (c *Controller) WithPixieQuerier(q PixieQuerier) *Controller {
 // when no kubescape anomaly opened a window for that pod (entlein/dx#93). Reuses the
 // same QueryFor → querier.Query → sink.WritePixieRows path + globalSem + reconcile
 // accounting as the anomaly-driven push. Satisfies control.queryRunner.
+// enrichProcessTree reattributes rows of a ProcessTreeAttributed table (DNS)
+// that pixie left with an empty pod, using kubescape's process-tree index, and
+// keeps only the target pod's rows. For non-attributed tables it is a no-op.
+//
+// It ALWAYS filters attributed tables to the target: those tables are pulled
+// unfiltered from pixie (queryfor skips the in-pixie pod filter), so with no
+// index available (ProcessTreeIndex unset or erroring) the still-unattributed
+// rows are simply dropped here — a pulled-unfiltered DNS window can never flood
+// forensic_db. With an index, the empty-pod rows are reattributed first.
+func (c *Controller) enrichProcessTree(table string, target anomaly.Target, rows []map[string]any) []map[string]any {
+	if !pxl.ProcessTreeAttributed(table) {
+		return rows
+	}
+	var idx kubescape.PIDIndex
+	if c.cfg.ProcessTreeIndex != nil {
+		ictx, icancel := context.WithTimeout(context.Background(), 30*time.Second)
+		if got, ierr := c.cfg.ProcessTreeIndex(ictx); ierr == nil {
+			idx = got
+		}
+		icancel()
+	}
+	return kubescape.EnrichRows(rows, idx, target.Namespace, target.Pod)
+}
+
 func (c *Controller) OrderQuery(target anomaly.Target, table string, start, end time.Time, queryID string) error {
 	if c.querier == nil {
 		return errors.New("controller: no pixie querier (operator-side push disabled)")
@@ -289,6 +321,7 @@ func (c *Controller) OrderQuery(target anomaly.Target, table string, start, end
 		return qerr
 	}
 	readCount = len(rows)
+	rows = c.enrichProcessTree(table, target, rows)
 	if len(rows) == 0 {
 		return nil // nothing to persist; the read/0-wrote reconcile row still records it
 	}
@@ -635,6 +668,9 @@ func (c *Controller) pushPixieRows(ctx context.Context, initial sink.Attribution
 				c.noteQueryResult(initial.Namespace, initial.Pod, table, len(rows))
 				nrows := len(rows)
 				readCount = nrows
+				// Reattribute + target-filter DNS pulled unfiltered from pixie.
+				rows = c.enrichProcessTree(table, target, rows)
+				nrows = len(rows)
 				if nrows > 0 {
 					// Bound the sink write with its own timeout. Without
 					// this, a stalled CH HTTP write would hold the table
diff --git a/src/vizier/services/adaptive_export/internal/e2e/loadtest_test.go b/src/vizier/services/adaptive_export/internal/e2e/loadtest_test.go
index 3f88e90bf96..9824c406580 100644
--- a/src/vizier/services/adaptive_export/internal/e2e/loadtest_test.go
+++ b/src/vizier/services/adaptive_export/internal/e2e/loadtest_test.go
@@ -88,9 +88,13 @@ func (q *cannedQuerier) Query(_ context.Context, pxl string) ([]map[string]any,
 		// Deterministic, fully-specified row. encoding/json sorts map keys,
 		// so the serialized bytes are byte-identical every rep.
 		rows = append(rows, map[string]any{
+			// Attributed to the same pod as the stub's kubescape anomaly
+			// (redis/redis-578d5dc9bd-kjj78), so process-tree-attributed tables
+			// (dns_events), which are now target-filtered in the operator rather
+			// than in pixie, survive the pull. Non-attributed tables ignore it.
 			"time_":     1744477360303026359 + int64(i),
-			"namespace": "aeload",
-			"pod":       "aeload/gen-l1",
+			"namespace": "redis",
+			"pod":       "redis/redis-578d5dc9bd-kjj78",
 			"req_path":  fmt.Sprintf("/ping/%d", i),
 			"table":     m[1],
 		})
diff --git a/src/vizier/services/adaptive_export/internal/kubescape/BUILD.bazel b/src/vizier/services/adaptive_export/internal/kubescape/BUILD.bazel
index 47b9b0b3481..0648ca59c1a 100644
--- a/src/vizier/services/adaptive_export/internal/kubescape/BUILD.bazel
+++ b/src/vizier/services/adaptive_export/internal/kubescape/BUILD.bazel
@@ -19,7 +19,11 @@ load("//bazel:pl_build_system.bzl", "pl_go_test")
 
 go_library(
     name = "kubescape",
-    srcs = ["extract.go"],
+    srcs = [
+        "enrich.go",
+        "extract.go",
+        "pidindex.go",
+    ],
     importpath = "px.dev/pixie/src/vizier/services/adaptive_export/internal/kubescape",
     visibility = ["//src/vizier/services/adaptive_export:__subpackages__"],
     deps = [
@@ -29,7 +33,11 @@ go_library(
 
 pl_go_test(
     name = "kubescape_test",
-    srcs = ["extract_test.go"],
+    srcs = [
+        "enrich_test.go",
+        "extract_test.go",
+        "pidindex_test.go",
+    ],
     embed = [":kubescape"],
     deps = [
         "//src/vizier/services/adaptive_export/internal/anomaly",
diff --git a/src/vizier/services/adaptive_export/internal/kubescape/enrich.go b/src/vizier/services/adaptive_export/internal/kubescape/enrich.go
new file mode 100644
index 00000000000..b219512ff03
--- /dev/null
+++ b/src/vizier/services/adaptive_export/internal/kubescape/enrich.go
@@ -0,0 +1,95 @@
+// Copyright 2018- The Pixie Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// SPDX-License-Identifier: Apache-2.0
+
+// Package kubescape parses the Kubescape-shaped fields of a
+// forensic_db.kubescape_logs row into the source-agnostic types used
+// downstream:
+//   - anomaly.Target — workload identity (used to compute the hash)
+//   - Event          — Target plus event-specific fields (event_time,
+//     rule id, hostname) needed for window math + persistence
+
+package kubescape
+
+import (
+	"encoding/json"
+	"strconv"
+	"strings"
+)
+
+// asUint64 coerces the many numeric shapes a pixie/ClickHouse row column can
+// carry (the querier returns map[string]any) into a PID. Unknown shapes -> 0.
+func asUint64(v any) uint64 {
+	switch n := v.(type) {
+	case uint64:
+		return n
+	case int64:
+		if n < 0 {
+			return 0
+		}
+		return uint64(n)
+	case int:
+		if n < 0 {
+			return 0
+		}
+		return uint64(n)
+	case float64:
+		if n < 0 {
+			return 0
+		}
+		return uint64(n)
+	case json.Number:
+		if p, err := n.Int64(); err == nil && p >= 0 {
+			return uint64(p)
+		}
+	case string:
+		if p, err := strconv.ParseUint(n, 10, 64); err == nil {
+			return p
+		}
+	}
+	return 0
+}
+
+// EnrichRows reattributes pixie rows that pixie left with an empty "pod" by
+// looking up their "pid" in the process-tree index, then keeps only rows that
+// belong to the target ("<namespace>/<pod>"). Rows pixie already attributed are
+// kept unchanged. Rows still unresolved after the index (genuine host DNS,
+// pods kubescape does not profile) are NOT evidence for this anomaly's pod and
+// are dropped — preserving the operator's active-set discipline.
+//
+// When targetPod is empty the target filter is skipped (attribute-only mode).
+// The input slice is filtered in place; the returned slice aliases it.
+func EnrichRows(rows []map[string]any, idx PIDIndex, targetNS, targetPod string) []map[string]any {
+	want := targetNS + "/" + targetPod
+	out := rows[:0]
+	for _, row := range rows {
+		pod, _ := row["pod"].(string)
+		if pod == "" {
+			if pid := asUint64(row["pid"]); pid != 0 {
+				if k := idx.Resolve(pid); k != "" {
+					pod = k
+					row["pod"] = k
+					if i := strings.IndexByte(k, '/'); i >= 0 {
+						row["namespace"] = k[:i]
+					}
+				}
+			}
+		}
+		if targetPod == "" || pod == want {
+			out = append(out, row)
+		}
+	}
+	return out
+}
diff --git a/src/vizier/services/adaptive_export/internal/kubescape/enrich_test.go b/src/vizier/services/adaptive_export/internal/kubescape/enrich_test.go
new file mode 100644
index 00000000000..aa9e5876342
--- /dev/null
+++ b/src/vizier/services/adaptive_export/internal/kubescape/enrich_test.go
@@ -0,0 +1,65 @@
+// Copyright 2018- The Pixie Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// SPDX-License-Identifier: Apache-2.0
+
+// Package kubescape parses the Kubescape-shaped fields of a
+// forensic_db.kubescape_logs row into the source-agnostic types used
+// downstream:
+//   - anomaly.Target — workload identity (used to compute the hash)
+//   - Event          — Target plus event-specific fields (event_time,
+//     rule id, hostname) needed for window math + persistence
+
+package kubescape_test
+
+import (
+	"testing"
+
+	"px.dev/pixie/src/vizier/services/adaptive_export/internal/kubescape"
+)
+
+func obsIndex() kubescape.PIDIndex {
+	return kubescape.BuildPIDIndex([]kubescape.Row{{
+		K8sDetails: `{"podName":"observer-x","podNamespace":"java-poc"}`,
+		ProcessDetails: `{"processTree":{"pid":25740,"comm":"sh",` +
+			`"childrenMap":{"getent␟882811":{"pid":882811,"ppid":25740,"comm":"getent"}}}}`,
+	}})
+}
+
+func TestEnrichRows_ReattributesEmptyPodViaPID(t *testing.T) {
+	rows := []map[string]any{
+		{"pid": int64(882811), "pod": ""},               // DNS pixie left blank -> observer
+		{"pid": int64(4242), "pod": ""},                 // host DNS -> unresolved -> dropped
+		{"pid": int64(0), "pod": "java-poc/observer-x"}, // already attributed -> kept
+	}
+	got := kubescape.EnrichRows(rows, obsIndex(), "java-poc", "observer-x")
+	if len(got) != 2 {
+		t.Fatalf("want 2 rows (reattributed + already-attributed), got %d: %v", len(got), got)
+	}
+	for _, r := range got {
+		if r["pod"] != "java-poc/observer-x" {
+			t.Errorf("row not attributed to target: %v", r)
+		}
+	}
+}
+
+func TestEnrichRows_HandlesNumericShapes(t *testing.T) {
+	for _, pidVal := range []any{int64(882811), float64(882811), uint64(882811), "882811"} {
+		rows := []map[string]any{{"pid": pidVal, "pod": ""}}
+		got := kubescape.EnrichRows(rows, obsIndex(), "java-poc", "observer-x")
+		if len(got) != 1 || got[0]["pod"] != "java-poc/observer-x" {
+			t.Errorf("pid shape %T (%v) not resolved: %v", pidVal, pidVal, got)
+		}
+	}
+}
diff --git a/src/vizier/services/adaptive_export/internal/kubescape/pidindex.go b/src/vizier/services/adaptive_export/internal/kubescape/pidindex.go
new file mode 100644
index 00000000000..dd4a8497253
--- /dev/null
+++ b/src/vizier/services/adaptive_export/internal/kubescape/pidindex.go
@@ -0,0 +1,94 @@
+// Copyright 2018- The Pixie Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// SPDX-License-Identifier: Apache-2.0
+
+// Package kubescape parses the Kubescape-shaped fields of a
+// forensic_db.kubescape_logs row into the source-agnostic types used
+// downstream:
+//   - anomaly.Target — workload identity (used to compute the hash)
+//   - Event          — Target plus event-specific fields (event_time,
+//     rule id, hostname) needed for window math + persistence
+
+package kubescape
+
+import "encoding/json"
+
+// PIDIndex maps a host PID to its owning "<namespace>/<pod>", reconstructed
+// from the process trees kubescape records in forensic_db.kubescape_logs.
+//
+// Why this exists: Pixie drops pod attribution for short-lived processes —
+// most importantly UDP DNS resolvers (getent/nslookup/the libc resolver),
+// which are forked from a long-running container process, do one lookup, and
+// exit. By the time upid_to_pod_name runs at query time the PID (and its
+// cgroup) is gone, so dns_events land with an empty pod and get filtered out
+// of the steered set. Kubescape, however, captured that PID — and its parent
+// chain — at exec time, tagged with the owning pod. Joining pixie's ephemeral
+// dns_events.pid against this index recovers the attribution with no pixie
+// change. See k8sstormcenter/pixie#80.
+type PIDIndex map[uint64]string
+
+// ptNode mirrors one node of kubescape's RuntimeProcessDetails.processTree.
+// childrenMap is keyed by "<comm>␟<pid>"; we only need the pids, so the
+// map values recurse into the same shape.
+type ptNode struct {
+	PID         uint64            `json:"pid"`
+	PPID        uint64            `json:"ppid"`
+	ChildrenMap map[string]ptNode `json:"childrenMap"`
+}
+
+// walk records this node's pid (and its parent's, so an unseen child still
+// resolves through the tree) then recurses into every child.
+func (n *ptNode) walk(idx PIDIndex, podKey string) {
+	if n.PID != 0 {
+		idx[n.PID] = podKey
+	}
+	if n.PPID != 0 {
+		// The parent is the long-running process that owns the pod; index it
+		// too so a child pid we never see directly can be reached via ppid.
+		if _, ok := idx[n.PPID]; !ok {
+			idx[n.PPID] = podKey
+		}
+	}
+	for _, c := range n.ChildrenMap {
+		c := c
+		c.walk(idx, podKey)
+	}
+}
+
+// BuildPIDIndex folds a batch of kubescape rows into a pid -> "<ns>/<pod>"
+// index. Rows without a pod (host-pid events) contribute nothing. Pure: no
+// I/O, no clock, safe to call on every refresh with the recent kubescape set.
+func BuildPIDIndex(rows []Row) PIDIndex {
+	idx := PIDIndex{}
+	for _, r := range rows {
+		var k k8sDetails
+		if err := json.Unmarshal([]byte(r.K8sDetails), &k); err != nil || k.PodName == "" {
+			continue
+		}
+		podKey := k.PodNamespace + "/" + k.PodName
+		var pd struct {
+			ProcessTree ptNode `json:"processTree"`
+		}
+		if err := json.Unmarshal([]byte(r.ProcessDetails), &pd); err != nil {
+			continue
+		}
+		pd.ProcessTree.walk(idx, podKey)
+	}
+	return idx
+}
+
+// Resolve returns "<namespace>/<pod>" for pid, or "" if kubescape never saw
+// it (a genuine host process, or a pod kubescape does not profile).
+func (idx PIDIndex) Resolve(pid uint64) string { return idx[pid] }
diff --git a/src/vizier/services/adaptive_export/internal/kubescape/pidindex_test.go b/src/vizier/services/adaptive_export/internal/kubescape/pidindex_test.go
new file mode 100644
index 00000000000..18a8115eb5d
--- /dev/null
+++ b/src/vizier/services/adaptive_export/internal/kubescape/pidindex_test.go
@@ -0,0 +1,63 @@
+// Copyright 2018- The Pixie Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// SPDX-License-Identifier: Apache-2.0
+
+// Package kubescape parses the Kubescape-shaped fields of a
+// forensic_db.kubescape_logs row into the source-agnostic types used
+// downstream:
+//   - anomaly.Target — workload identity (used to compute the hash)
+//   - Event          — Target plus event-specific fields (event_time,
+//     rule id, hostname) needed for window math + persistence
+
+package kubescape_test
+
+import (
+	"testing"
+
+	"px.dev/pixie/src/vizier/services/adaptive_export/internal/kubescape"
+)
+
+// Real-shape kubescape RuntimeProcessDetails: a parent (tini) with a child
+// (getent — the exfil DNS resolver) in childrenMap. Both pids must resolve to
+// the owning pod, which is what recovers DNS attribution pixie drops.
+func TestBuildPIDIndex_ResolvesChildAndParent(t *testing.T) {
+	rows := []kubescape.Row{{
+		K8sDetails: `{"podName":"observer-6fbb545fdd-r4dt8","podNamespace":"java-poc"}`,
+		ProcessDetails: `{"processTree":{"pid":25740,"ppid":25061,"comm":"tini",` +
+			`"childrenMap":{"getent␟882811":{"pid":882811,"ppid":25740,"comm":"getent"}}}}`,
+	}}
+	idx := kubescape.BuildPIDIndex(rows)
+
+	if got := idx.Resolve(882811); got != "java-poc/observer-6fbb545fdd-r4dt8" {
+		t.Errorf("child getent pid 882811: got %q, want java-poc/observer-6fbb545fdd-r4dt8", got)
+	}
+	if got := idx.Resolve(25740); got != "java-poc/observer-6fbb545fdd-r4dt8" {
+		t.Errorf("parent pid 25740: got %q, want java-poc/observer-6fbb545fdd-r4dt8", got)
+	}
+	if got := idx.Resolve(999999); got != "" {
+		t.Errorf("unseen pid: got %q, want empty", got)
+	}
+}
+
+// Host-pid events (no pod) must contribute nothing, not a "/" key.
+func TestBuildPIDIndex_SkipsHostPidRows(t *testing.T) {
+	rows := []kubescape.Row{{
+		K8sDetails:     `{"podName":"","podNamespace":""}`,
+		ProcessDetails: `{"processTree":{"pid":4242,"comm":"kubelet"}}`,
+	}}
+	if got := kubescape.BuildPIDIndex(rows).Resolve(4242); got != "" {
+		t.Errorf("host pid must not resolve, got %q", got)
+	}
+}
diff --git a/src/vizier/services/adaptive_export/internal/pxl/compile.go b/src/vizier/services/adaptive_export/internal/pxl/compile.go
index de3d16d0aad..91790eb1275 100644
--- a/src/vizier/services/adaptive_export/internal/pxl/compile.go
+++ b/src/vizier/services/adaptive_export/internal/pxl/compile.go
@@ -63,6 +63,10 @@ func CompilePassthrough(table string, window time.Duration) (string, error) {
 	b.WriteString("df = df[df.time_ <  px.int64_to_time(%d)]\n")
 	b.WriteString("df.namespace = px.upid_to_namespace(df.upid)\n")
 	b.WriteString("df.pod = px.upid_to_pod_name(df.upid)\n")
+	// Carry the host PID so downstream enrichment can reattribute rows from
+	// short-lived processes (DNS resolvers) via kubescape's process tree —
+	// keeps the firehose column shape identical to QueryFor. See kubescape.PIDIndex.
+	b.WriteString("df.pid = px.upid_to_pid(df.upid)\n")
 	b.WriteString("px.display(df, '" + table + "')\n")
 	return b.String(), nil
 }
diff --git a/src/vizier/services/adaptive_export/internal/pxl/queryfor.go b/src/vizier/services/adaptive_export/internal/pxl/queryfor.go
index 168c54a4722..2bb4d6a1cf4 100644
--- a/src/vizier/services/adaptive_export/internal/pxl/queryfor.go
+++ b/src/vizier/services/adaptive_export/internal/pxl/queryfor.go
@@ -71,24 +71,49 @@ func QueryFor(table string, t anomaly.Target, sliceStart, sliceEnd, now time.Tim
 	// not the bare pod name. Filtering against bare t.Pod would always
 	// miss; build the namespaced key when we have both fields.
 	b.WriteString("df.pod = px.upid_to_pod_name(df.upid)\n")
-	if t.Namespace != "" {
-		b.WriteString("df = df[df.namespace == '" + escapePxL(t.Namespace) + "']\n")
-	}
-	if t.Pod != "" {
+	// Expose the host PID so the operator can recover pod attribution that
+	// kubescape captured for short-lived processes (e.g. UDP DNS resolvers)
+	// which pixie leaves with an empty pod. See kubescape.PIDIndex.
+	b.WriteString("df.pid = px.upid_to_pid(df.upid)\n")
+	// Tables frequently emitted by short-lived, unattributable processes
+	// (DNS resolvers) SKIP the in-pixie pod filter: their empty-pod rows would
+	// be dropped here before the operator can reattribute them from kubescape's
+	// process tree. They are pulled window-scoped and filtered to the target
+	// pod AFTER enrichment (see controller). All other tables filter in-pixie.
+	if !ProcessTreeAttributed(table) {
 		if t.Namespace != "" {
-			// Both fields present — use exact equality on the namespaced key.
-			b.WriteString("df = df[df.pod == '" + escapePxL(t.Namespace+"/"+t.Pod) + "']\n")
-		} else {
-			// Pod-only fallback: df.pod is "<ns>/<pod>", so a bare-pod
-			// equality always misses. Regex-anchor "<any-ns>/<pod>" via
-			// px.regex_match so the defensive path stays functional.
-			b.WriteString("df = df[px.regex_match('^[^/]+/" + escapePxL(regexp.QuoteMeta(t.Pod)) + "$', df.pod)]\n")
+			b.WriteString("df = df[df.namespace == '" + escapePxL(t.Namespace) + "']\n")
+		}
+		if t.Pod != "" {
+			if t.Namespace != "" {
+				// Both fields present — use exact equality on the namespaced key.
+				b.WriteString("df = df[df.pod == '" + escapePxL(t.Namespace+"/"+t.Pod) + "']\n")
+			} else {
+				// Pod-only fallback: df.pod is "<ns>/<pod>", so a bare-pod
+				// equality always misses. Regex-anchor "<any-ns>/<pod>" via
+				// px.regex_match so the defensive path stays functional.
+				b.WriteString("df = df[px.regex_match('^[^/]+/" + escapePxL(regexp.QuoteMeta(t.Pod)) + "$', df.pod)]\n")
+			}
 		}
 	}
 	b.WriteString("px.display(df, '" + table + "')\n")
 	return b.String(), nil
 }
 
+// processTreeAttributedTables are pixie tables whose rows are commonly emitted
+// by short-lived processes pixie cannot attribute to a pod at query time (the
+// classic case is UDP DNS: the resolver forks, does one lookup, and exits, so
+// upid_to_pod_name yields ""). For these we skip the in-pixie pod filter and
+// reattribute downstream from kubescape's process tree (kubescape.PIDIndex).
+var processTreeAttributedTables = map[string]bool{
+	"dns_events": true,
+}
+
+// ProcessTreeAttributed reports whether `table` should be pulled window-scoped
+// (unfiltered by pod in pixie) and reattributed from the kubescape process
+// tree by the operator, rather than filtered on pixie's own pod attribution.
+func ProcessTreeAttributed(table string) bool { return processTreeAttributedTables[table] }
+
 // pxlEscaper turns raw bytes that could break out of a PxL single-quoted
 // string into their Python-style escape sequences. The backslash MUST be
 // mapped FIRST so its own substitution doesn't get double-escaped when
diff --git a/src/vizier/services/adaptive_export/internal/pxl/queryfor_test.go b/src/vizier/services/adaptive_export/internal/pxl/queryfor_test.go
index 562ea794cc0..91c602d7b38 100644
--- a/src/vizier/services/adaptive_export/internal/pxl/queryfor_test.go
+++ b/src/vizier/services/adaptive_export/internal/pxl/queryfor_test.go
@@ -264,7 +264,9 @@ func TestEscapePxL_TableDriven(t *testing.T) {
 //
 // Total: 10 statements + trailing empty == strings.Split == 11 entries.
 func TestQueryFor_RejectsInjectionInTargetFields(t *testing.T) {
-	const wantLines = 11
+	// 12 = the fixed 8-line projection preamble (incl. df.pid) + namespace
+	// filter + pod filter + display. An injected newline would add lines.
+	const wantLines = 12
 
 	cases := []struct {
 		name   string
@@ -336,7 +338,7 @@ func TestQueryFor_PodOnlyRegexEscapesQuoteMetaInjection(t *testing.T) {
 	if err != nil {
 		t.Fatalf("QueryFor: %v", err)
 	}
-	if strings.Contains(q, "exec(") || strings.Count(q, "\n") > 9 {
+	if strings.Contains(q, "exec(") || strings.Count(q, "\n") > 10 {
 		t.Fatalf("pod-only path injection succeeded:\n%s", q)
 	}
 }
diff --git a/src/vizier/services/adaptive_export/internal/trigger/clickhouse.go b/src/vizier/services/adaptive_export/internal/trigger/clickhouse.go
index f040b1db4c0..fb45ee7309e 100644
--- a/src/vizier/services/adaptive_export/internal/trigger/clickhouse.go
+++ b/src/vizier/services/adaptive_export/internal/trigger/clickhouse.go
@@ -425,6 +425,50 @@ func (t *ClickHouseHTTP) fetchSince(ctx context.Context, watermark uint64) ([]ku
 	return parseJSONEachRow(resp.Body)
 }
 
+// processTreeIndexLimit bounds the ProcessTreeIndex read. The most-recent N
+// kubescape process-tree rows cover the live process population for this host
+// generously; the read is cheap and runs only for DNS pulls.
+const processTreeIndexLimit = 5000
+
+// ProcessTreeIndex reads this host's most-recent kubescape process trees and
+// folds them into a pid -> "<namespace>/<pod>" index. The operator uses it to
+// reattribute pixie rows (DNS) whose short-lived resolver pid is unmappable at
+// query time but which kubescape captured at exec time. See kubescape.PIDIndex,
+// controller.enrichProcessTree, pixie#80. Ordering by the normalized event_time
+// DESC + LIMIT keeps the scan bounded without time-unit arithmetic.
+func (t *ClickHouseHTTP) ProcessTreeIndex(ctx context.Context) (kubescape.PIDIndex, error) {
+	q := url.Values{}
+	q.Set("query", fmt.Sprintf(
+		"SELECT RuleID, RuntimeK8sDetails, RuntimeProcessDetails, event_time, hostname "+
+			"FROM %s.%s "+
+			"WHERE hostname = %s AND RuntimeProcessDetails != '' "+
+			"ORDER BY %s DESC LIMIT %d FORMAT JSONEachRow",
+		t.cfg.Database, t.cfg.Table, quoteCH(t.cfg.Hostname),
+		chNormEventTimeNanos, processTreeIndexLimit))
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet,
+		t.cfg.Endpoint+"/?"+q.Encode(), nil)
+	if err != nil {
+		return nil, err
+	}
+	if t.cfg.Username != "" {
+		req.SetBasicAuth(t.cfg.Username, t.cfg.Password)
+	}
+	resp, err := t.client.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode/100 != 2 {
+		body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
+		return nil, fmt.Errorf("ProcessTreeIndex HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body)))
+	}
+	rows, _, err := parseJSONEachRow(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+	return kubescape.BuildPIDIndex(rows), nil
+}
+
 // parseJSONEachRow streams JSONEachRow output line-by-line from r.
 // Streaming (vs io.ReadAll into a []byte) bounds memory at one row
 // regardless of how large the ClickHouse result set is.