adaptive_export: /query runner + dx_evidence_graph rename + evidence_manifest (folds #77/#78)

.arclint

@@ -21,6 +21,7 @@
     "(^src/operator/client/versioned/)",
     "(^src/operator/apis/px.dev/v1alpha1/zz_generated.deepcopy.go)",
     "(^src/e2e_test/adaptive_export_loadtest/tools/loadgen/)",
+    "(^src/e2e_test/adaptive_export_loadtest/suite/)",
     "(^src/stirling/bpf_tools/bcc_bpf/system-headers)",
     "(^src/stirling/mysql/testing/.*\\.json$)",
     "(^src/stirling/obj_tools/testdata/go/test_go_binary.go)",

.bazelignore

@@ -10,3 +10,4 @@ src/datagen/pii/privy
 # adaptive_export_loadtest generator is a docker-built test tool (see its README);
 # build-agent to replace with a bazel target. Until then, keep it out of gazelle.
 src/e2e_test/adaptive_export_loadtest/tools/loadgen
+src/e2e_test/adaptive_export_loadtest/suite

.github/workflows/e2e_calibrate_soc.yaml

@@ -16,7 +16,7 @@ permissions:
 
 jobs:
   e2e:
-    runs-on: oracle-vm-16cpu-64gb-x86-64   
+    runs-on: oracle-vm-16cpu-64gb-x86-64
     timeout-minutes: 90
     env:
       KUBECONFIG: /etc/rancher/k3s/k3s.yaml
@@ -36,7 +36,7 @@ jobs:
         PX_CLOUD_ADDR: pixie.austrianopencloudcommunity.org
         PX_DEPLOY_KEY: ${{ secrets.PX_DEPLOY_KEY }}
         PX_API_KEY: ${{ secrets.PX_API_KEY }}
-        TS_AUTHKEY: ${{ secrets.TAILSCALE_AUTH_KEY }}          
+        TS_AUTHKEY: ${{ secrets.TAILSCALE_AUTH_KEY }}
         CLICKHOUSE_ANALYST_PASSWORD: ${{ secrets.CLICKHOUSE_ANALYST_PASSWORD }}
         CLICKHOUSE_INGEST_PASSWORD: ${{ secrets.CLICKHOUSE_INGEST_PASSWORD }}
         CLICKHOUSE_PIXIE_PASSWORD: ${{ secrets.CLICKHOUSE_PIXIE_PASSWORD }}
@@ -48,7 +48,7 @@ jobs:
         make pixie                              # vizier + AE
         make kubescape || true                  # node-agent (netStreaming)
         bash tree/clickhouse-lab/install.sh      # forensic_db
-        make java-poc                          
+        make java-poc
         if [ -n "${{ inputs.dx_image }}" ]; then
           kubectl -n honey set image ds/dx-daemon dx-daemon="${{ inputs.dx_image }}" || true
         fi

src/e2e_test/adaptive_export_loadtest/skaffold.yaml

@@ -22,15 +22,15 @@ kind: Config
 metadata:
   name: e2e-nonpixie
 requires:
-  - git:
-      repo: https://github.com/k8sstormcenter/soc
-      path: skaffold.yaml
-      ref: main
-    configs:
-      - soc-stack
-  - git:
-      repo: https://github.com/k8sstormcenter/bob
-      path: example/java-poc/skaffold.yaml
-      ref: main
-    configs:
-      - java-poc-apps
+- git:
+    repo: https://github.com/k8sstormcenter/soc
+    path: skaffold.yaml
+    ref: main
+  configs:
+  - soc-stack
+- git:
+    repo: https://github.com/k8sstormcenter/bob
+    path: example/java-poc/skaffold.yaml
+    ref: main
+  configs:
+  - java-poc-apps

src/vizier/services/adaptive_export/cmd/main.go

@@ -573,8 +573,13 @@ func main() {
 	// steers this AE's activeSet (Upsert/Remove) over HTTP. Off by default so
 	// the existing trigger→controller→activeSet flow is unchanged.
 	if addr := os.Getenv("CONTROL_ADDR"); addr != "" {
-		ctrlSrv := control.New(activeSet, nil) // OrderQuery runner wired later
-		ctrlSrv.SetGraphWriter(applier)        // dx_attack_graph ingest → ClickHouse
+		// 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_evidence_graph still work.
+		ctrlSrv := control.New(activeSet, ctl)
+		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

src/vizier/services/adaptive_export/internal/clickhouse/apply.go

@@ -67,9 +67,13 @@ 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",
+	// 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
@@ -108,15 +112,29 @@ 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
+}
+
+// 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
 }

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", "dx_evidence_manifest"}
 	got := OperatorOwnedTables[len(OperatorOwnedTables)-len(want):]
 	for i, w := range want {
 		if got[i] != w {

src/vizier/services/adaptive_export/internal/clickhouse/ddl.go

@@ -66,11 +66,15 @@ 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",
+	// 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

src/vizier/services/adaptive_export/internal/clickhouse/schema.sql

@@ -492,7 +492,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.
@@ -503,7 +503,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,
@@ -531,7 +531,38 @@ 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` != '';
+
+-- 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;

src/vizier/services/adaptive_export/internal/control/server.go

@@ -54,18 +54,25 @@ 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
+}
+
+// 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/attack_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
@@ -76,13 +83,17 @@ 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)
+	s.mux.HandleFunc("/dx/evidence_manifest", s.handleDXEvidenceManifest)
 	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 }
 
+// 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,
@@ -115,9 +126,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,13 +151,88 @@ 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
 	}
 	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"`
@@ -175,7 +261,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