feat(ae): /dx/evidence_manifest ingest → forensic_db.dx_evidence_manifest

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

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)

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 {

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

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;

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"`

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)
+	}
+}