diff --git a/k8s/vizier/adaptive_export/kustomization.yaml b/k8s/vizier/adaptive_export/kustomization.yaml
index 94bcd745b64..43400cd4d8d 100644
--- a/k8s/vizier/adaptive_export/kustomization.yaml
+++ b/k8s/vizier/adaptive_export/kustomization.yaml
@@ -7,4 +7,4 @@ resources:
 images:
   - name: vizier-adaptive_export_image
     newName: ghcr.io/k8sstormcenter/vizier-adaptive_export_image
-    newTag: 0.14.19-aeprod57
+    newTag: 0.14.19-aeprod58
diff --git a/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql b/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql
index 6fa8ed7f002..51ff1c33a3a 100644
--- a/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql
+++ b/src/vizier/services/adaptive_export/internal/clickhouse/schema.sql
@@ -649,11 +649,23 @@ CREATE TABLE IF NOT EXISTS forensic_db.dx_ptrace (
 
 -- dc_snoop (dentry cache, V1/V2 process+file) — exported via the OTel/ClickHouse
 -- retention plugin (px.export). pid-keyed; t = R (reference) / M (miss).
+-- ppid/pcomm captured inline in the tracepoint (curtask->real_parent) so every
+-- dcache event carries its parent with no join. pid_start/ppid_start are
+-- group_leader->start_time (ns since boot) — a pid-reuse-stable identity for the
+-- process and its parent, and the (pid,pid_start)/(ppid,ppid_start) keys the
+-- per-node process forest walks to correlate evidence to ancestry. Deeper
+-- ancestry (pod attribution of a blank-pod child) is resolved by dx against the
+-- forest; the AE only cuts 1-level own-stack noise here via the ppid->namespace
+-- join.
 -- One column per line (schema-verify parser is line-oriented).
 CREATE TABLE IF NOT EXISTS forensic_db.dc_snoop (
   time_ DateTime64(9, 'UTC'),
   pid Int64,
+  pid_start Int64,
+  ppid Int64,
+  ppid_start Int64,
   comm String,
+  pcomm String,
   t String,
   file String,
   namespace String,
diff --git a/src/vizier/services/adaptive_export/internal/pxl/queryfor.go b/src/vizier/services/adaptive_export/internal/pxl/queryfor.go
index 18c877c9db7..a9d7fdf6bfe 100644
--- a/src/vizier/services/adaptive_export/internal/pxl/queryfor.go
+++ b/src/vizier/services/adaptive_export/internal/pxl/queryfor.go
@@ -90,11 +90,19 @@ func QueryFor(table string, t anomaly.Target, sliceStart, sliceEnd, now time.Tim
 			}
 		}
 	} else if IsDarkVector(table) {
-		// Node-scoped (transient attack pids resolve blank ns, so no pod filter). Drop
-		// own-stack comms before the pid-merge to keep it cheap, then drop infra namespaces.
-		b.WriteString(darkCommExclusion(table))
+		// Node-scoped (transient attack pids resolve blank ns, so no pod filter).
+		// Noise is cut structurally, NOT by a hardcoded comm blocklist (deleted):
+		//   1. child-namespace exclusion drops resolved own-stack processes (vizier,
+		//      pem, node-agent, … resolve to pl/honey/clickhouse via their pod);
+		//   2. parent-ancestry exclusion (ppid -> parent namespace) drops the
+		//      transient children of own-stack pods that resolve blank themselves.
+		// Host-level runtime + kernel threads (containerd-shim/systemd/k3s/kworker)
+		// have a blank namespace AND a blank/kernel-thread parent, so 1-level ancestry
+		// cannot resolve them — they enter the always-on retention look-back and are
+		// relevance-filtered by dx's multi-level process forest, not here.
 		b.WriteString(PodEnrichPxL(table))
 		b.WriteString(darkNamespaceExclusion())
+		b.WriteString(darkParentAncestryExclusion(table))
 	} else {
 		b.WriteString(PodEnrichPxL(table))
 		if t.Namespace != "" {
@@ -132,83 +140,67 @@ func pixieSourceFor(table string) string {
 	return table
 }
 
-// Dark-vector tables carrying a comm column (so the comm exclusion applies).
-var darkVectorHasComm = map[string]bool{
-	"dc_snoop": true, "creds_change": true, "dx_vfs_events": true,
-	"dx_unlink": true, "dx_dlookup": true, "dx_mprotect": true,
-	"dx_bpf": true, "dx_ptrace": true,
+// Dark-vector tables that carry a ppid column (so the parent-ancestry exclusion
+// applies). Only dc_snoop captures ppid today; creds_change and the dx_* bpftrace
+// tables do not, so they get the child-namespace exclusion only.
+var darkVectorHasPpid = map[string]bool{
+	"dc_snoop": true,
 }
 
-// Own-stack + node/system comms dropped from the node-scoped dark capture; workload
-// comms (redis-*, etc.) are never listed. Override via DC_SNOOP_EXCLUDE_COMMS (csv).
-var darkExcludeCommsDefault = []string{
-	"pem", "kelvin", "containerd", "containerd-shim", "runc", "node-agent",
-	"runc:[2:INIT]", "runc:[1:CHILD]",
-	"vizier-query-broker", "vizier-metadata", "nats-server", "k3s-server",
-	"k3s-agent", "systemd", "systemd-journal", "SystemLogFlush", "kubelet",
-	"AsyncInsertQ", "BgSchPool", "Collector", "AsyncMetrics", "MergeMutate",
-	"MergeTreeIndex", "CgrpMemUsgObsr", "coredns", "metadata", "storage",
-	"operator", "iptables", "iptables-save", "iptables-restor", "ip6tables",
-	"ConfigReloader", "clickhouse-oper", "Formatter", "(setup.sh)", "cmd",
-	"vector-worker", "metrics-server", "local-path-prov", "portmap",
-	"(udev-worker)", "systemd-resolve", "systemd-timesyn",
-	"systemd-udevd", "systemd-sysctl", "host-local", "bridge", "flannel",
-	"loopback", "bandwidth", "dbus-daemon", "mount", "umount", "tailscaled",
-	"grpc_health_pro", "kubevuln", "opm", "(spawn)", "kube-proxy",
-	"pause", "systemd-logind",
-}
-
-// Kernel-thread families whose names carry a variable suffix (kworker/u8:3) that
-// exact match misses; dropped via px.contains.
-var darkExcludeCommSubstrings = []string{
-	"kworker", "ksoftirqd", "migration", "rcu_", "kthreadd", "kdevtmpfs",
-	"kcompactd", "khugepaged", "kswapd", "watchdog", "cpuhp", "ksmd", "irq/",
-}
-
-// Infra namespaces dropped from the node-scoped dark capture. Blank-namespace rows
-// (transient attack children) survive. Override via DC_SNOOP_EXCLUDE_NAMESPACES.
+// Infra namespaces dropped from the node-scoped dark capture — applied BOTH to the
+// row's own namespace (resolved own-stack processes) and, for tables with ppid, to
+// the parent's namespace (ancestry). Blank-namespace rows survive (transient attack
+// children; host/kernel processes). Override via DC_SNOOP_EXCLUDE_NAMESPACES.
 var darkExcludeNamespacesDefault = []string{
-	"pl", "honey", "px-operator", "olm", "clickhouse", "socdemo", "socdemo-ch",
+	"pl", "honey", "px-operator", "olm", "clickhouse",
 	"kube-system", "kube-public", "kube-node-lease", "local-path-storage",
 }
 
-func darkCommExclusion(table string) string {
-	if !darkVectorHasComm[table] {
-		return ""
-	}
-	comms := darkExcludeCommsDefault
-	if v := strings.TrimSpace(os.Getenv("DC_SNOOP_EXCLUDE_COMMS")); v != "" {
-		comms = nil
+// darkExcludeNamespaces resolves the effective infra-namespace list (env override
+// or default), shared by the child-namespace and parent-ancestry exclusions so the
+// two can never drift.
+func darkExcludeNamespaces() []string {
+	if v := strings.TrimSpace(os.Getenv("DC_SNOOP_EXCLUDE_NAMESPACES")); v != "" {
+		var nss []string
 		for _, s := range strings.Split(v, ",") {
 			if s = strings.TrimSpace(s); s != "" {
-				comms = append(comms, s)
+				nss = append(nss, s)
 			}
 		}
+		return nss
 	}
+	return darkExcludeNamespacesDefault
+}
+
+func darkNamespaceExclusion() string {
 	var b strings.Builder
-	for _, c := range comms {
-		b.WriteString("df = df[df.comm != '" + escapePxL(c) + "']\n")
-	}
-	for _, s := range darkExcludeCommSubstrings {
-		b.WriteString("df = df[px.logicalNot(px.contains(df.comm, '" + escapePxL(s) + "'))]\n")
+	for _, ns := range darkExcludeNamespaces() {
+		b.WriteString("df = df[df.namespace != '" + escapePxL(ns) + "']\n")
 	}
 	return b.String()
 }
 
-func darkNamespaceExclusion() string {
-	nss := darkExcludeNamespacesDefault
-	if v := strings.TrimSpace(os.Getenv("DC_SNOOP_EXCLUDE_NAMESPACES")); v != "" {
-		nss = nil
-		for _, s := range strings.Split(v, ",") {
-			if s = strings.TrimSpace(s); s != "" {
-				nss = append(nss, s)
-			}
-		}
+// darkParentAncestryExclusion resolves each row's PARENT namespace via a
+// process_stats join on ppid and drops rows whose parent lives in an own-stack
+// namespace — even when the row's own pod is blank (a transient process exec'd by
+// an infra pod). This is the ppid-based ancestry cut that REPLACES the deleted
+// hardcoded comm blocklist: pod/namespace-rooted, not comm-matched. parent_namespace
+// is dropped before display so the sink projection is unchanged. Emits nothing for
+// tables without a ppid column. Mirrors the retention path (dc_snoop.pxl).
+func darkParentAncestryExclusion(table string) string {
+	if !darkVectorHasPpid[table] {
+		return ""
 	}
 	var b strings.Builder
-	for _, ns := range nss {
-		b.WriteString("df = df[df.namespace != '" + escapePxL(ns) + "']\n")
+	b.WriteString("par = px.DataFrame(table='process_stats', start_time='" + darkProcStatsWindow + "')\n")
+	b.WriteString("par.parent_namespace = par.ctx['namespace']\n")
+	b.WriteString("par.ppid = px.upid_to_pid(par.upid)\n")
+	b.WriteString("par = par.groupby(['parent_namespace', 'ppid']).agg()\n")
+	b.WriteString("df = df.merge(par, how='left', left_on=['ppid'], right_on=['ppid'], suffixes=['', '_par'])\n")
+	for _, ns := range darkExcludeNamespaces() {
+		b.WriteString("df = df[df.parent_namespace != '" + escapePxL(ns) + "']\n")
 	}
+	b.WriteString("df = df.drop(['parent_namespace'])\n")
 	return b.String()
 }
 
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 f42c779bd5b..2e828f3fd1c 100644
--- a/src/vizier/services/adaptive_export/internal/pxl/queryfor_test.go
+++ b/src/vizier/services/adaptive_export/internal/pxl/queryfor_test.go
@@ -37,6 +37,30 @@ var (
 	}
 )
 
+// TestQueryFor_CleanupNoCollateral — the pure-ancestry cleanup touched ONLY the
+// dark-vector branch. Every native/protocol table (redis_events, dns_events, …) and
+// stack_trace must be UNAFFECTED: no ancestry join, no parent_namespace, no comm
+// drops, and they must still pin the alert pod (dx-steered pod scope). This is the
+// "other CH tables not affected" guard.
+func TestQueryFor_CleanupNoCollateral(t *testing.T) {
+	native := []string{"redis_events", "dns_events", "http_events", "conn_stats", "pgsql_events", "mysql_events", "stack_trace"}
+	for _, tbl := range native {
+		q, err := QueryFor(tbl, target, fixedStart, fixedEnd, fixedNow)
+		if err != nil {
+			t.Fatalf("QueryFor(%s): %v", tbl, err)
+		}
+		for _, banned := range []string{"parent_namespace", "left_on=['ppid']", "df.comm !=", "px.contains(df.comm", "df = df[df.namespace != '"} {
+			if strings.Contains(q, banned) {
+				t.Errorf("%s (native) must not contain dark-path machinery %q; got:\n%s", tbl, banned, q)
+			}
+		}
+		// still pod-scoped (dx-steered): the alert pod is pinned
+		if !strings.Contains(q, "redis-6fbcfb97c-82qxv") {
+			t.Errorf("%s must still pin the alert pod; got:\n%s", tbl, q)
+		}
+	}
+}
+
 // TestQueryFor_UnknownTable — non-builtin tables wrap ErrUnknownTable.
 func TestQueryFor_UnknownTable(t *testing.T) {
 	_, err := QueryFor("nope_table", target, fixedStart, fixedEnd, fixedNow)
@@ -378,14 +402,14 @@ func TestQueryFor_NoEndTimeAtLiveEdge(t *testing.T) {
 }
 
 // TestQueryFor_DarkNamespaceExclusion — the node-scoped dark capture (dc_snoop)
-// must drop infra namespaces (pl, kube-system, …) while KEEPING blank-namespace
-// transient rows (the attack's short-lived children). Mirrors the shipped preset.
+// must drop infra namespaces by the row's OWN namespace (resolved own-stack
+// processes) while KEEPING blank-namespace transient rows. Mirrors the shipped
+// preset.
 func TestQueryFor_DarkNamespaceExclusion(t *testing.T) {
 	q, err := QueryFor("dc_snoop", target, fixedStart, fixedEnd, fixedNow)
 	if err != nil {
 		t.Fatalf("QueryFor: %v", err)
 	}
-	// namespace drops present for infra
 	for _, ns := range []string{"pl", "kube-system", "clickhouse"} {
 		if !strings.Contains(q, "df = df[df.namespace != '"+ns+"']") {
 			t.Errorf("dark capture must drop infra namespace %q; got:\n%s", ns, q)
@@ -395,47 +419,93 @@ func TestQueryFor_DarkNamespaceExclusion(t *testing.T) {
 	if strings.Contains(q, "df = df[df.namespace == '") {
 		t.Errorf("dark capture must not pin df.namespace ==; got:\n%s", q)
 	}
-	// host/CNI comm drops present
-	for _, c := range []string{"host-local", "systemd-udevd", "tailscaled", "kubevuln"} {
-		if !strings.Contains(q, "df = df[df.comm != '"+c+"']") {
-			t.Errorf("dark capture must drop host/CNI comm %q; got:\n%s", c, q)
-		}
-	}
 }
 
 // TestQueryFor_DarkNamespaceExclusion_EnvOverride — DC_SNOOP_EXCLUDE_NAMESPACES
-// replaces the default list.
+// replaces the default list for BOTH the child-namespace and the parent-ancestry
+// drops (they share one resolved list, so they can never drift).
 func TestQueryFor_DarkNamespaceExclusion_EnvOverride(t *testing.T) {
 	t.Setenv("DC_SNOOP_EXCLUDE_NAMESPACES", "foo,bar")
 	q, err := QueryFor("dc_snoop", target, fixedStart, fixedEnd, fixedNow)
 	if err != nil {
 		t.Fatalf("QueryFor: %v", err)
 	}
-	if !strings.Contains(q, "df = df[df.namespace != 'foo']") || !strings.Contains(q, "df = df[df.namespace != 'bar']") {
-		t.Errorf("env override must emit foo/bar drops; got:\n%s", q)
+	for _, want := range []string{
+		"df = df[df.namespace != 'foo']", "df = df[df.namespace != 'bar']",
+		"df = df[df.parent_namespace != 'foo']", "df = df[df.parent_namespace != 'bar']",
+	} {
+		if !strings.Contains(q, want) {
+			t.Errorf("env override must emit %q; got:\n%s", want, q)
+		}
+	}
+	if strings.Contains(q, "!= 'pl']") {
+		t.Errorf("env override must REPLACE the default (no 'pl' in either child or parent drop); got:\n%s", q)
+	}
+}
+
+// TestQueryFor_DarkNoCommBlocklist — the hardcoded comm blocklist and kernel-thread
+// substring filter were DELETED (pure-ancestry cleanup). The dark capture must emit
+// NO comm-based drops at all: no `df.comm != ...`, no `px.contains(df.comm, ...)`.
+// Noise is cut structurally (own + parent namespace), so host/kernel comms now enter
+// the retention look-back for dx's forest to relevance-filter.
+func TestQueryFor_DarkNoCommBlocklist(t *testing.T) {
+	q, err := QueryFor("dc_snoop", target, fixedStart, fixedEnd, fixedNow)
+	if err != nil {
+		t.Fatalf("QueryFor: %v", err)
+	}
+	if strings.Contains(q, "df.comm !=") {
+		t.Errorf("comm blocklist must be gone (no df.comm != drops); got:\n%s", q)
 	}
-	if strings.Contains(q, "df = df[df.namespace != 'pl']") {
-		t.Errorf("env override must REPLACE the default (no 'pl'); got:\n%s", q)
+	if strings.Contains(q, "px.contains(df.comm") {
+		t.Errorf("kernel-thread substring filter must be gone; got:\n%s", q)
 	}
 }
 
-// dc_snoop drops kernel-thread families (variable suffix) via px.logicalNot(px.contains),
-// keeps workload comms (redis-server), and doesn't pin the alert pod's namespace.
-func TestQueryFor_DarkCommSubstringExclusion(t *testing.T) {
+// TestQueryFor_DarkParentAncestry — for a ppid-bearing dark table (dc_snoop) the
+// query must resolve the PARENT namespace via a process_stats join on ppid, drop
+// own-stack parents (the ancestry cut that replaced the comm blocklist), and DROP
+// parent_namespace before display so the sink projection is unchanged.
+func TestQueryFor_DarkParentAncestry(t *testing.T) {
 	q, err := QueryFor("dc_snoop", target, fixedStart, fixedEnd, fixedNow)
 	if err != nil {
 		t.Fatalf("QueryFor: %v", err)
 	}
-	for _, sub := range []string{"kworker", "ksoftirqd", "rcu_"} {
-		want := "df = df[px.logicalNot(px.contains(df.comm, '" + sub + "'))]"
+	for _, want := range []string{
+		"par = px.DataFrame(table='process_stats'",
+		"par.parent_namespace = par.ctx['namespace']",
+		"par.ppid = px.upid_to_pid(par.upid)",
+		"left_on=['ppid'], right_on=['ppid']",
+		"df = df[df.parent_namespace != 'pl']",
+		"df = df[df.parent_namespace != 'clickhouse']",
+		"df = df.drop(['parent_namespace'])",
+	} {
 		if !strings.Contains(q, want) {
-			t.Errorf("want kernel-thread drop %q; got:\n%s", want, q)
+			t.Errorf("dark ancestry filter missing %q; got:\n%s", want, q)
 		}
 	}
-	if strings.Contains(q, "df.comm != 'redis-server'") || strings.Contains(q, "df.comm, 'redis") {
-		t.Errorf("workload comm redis-* must NOT be excluded; got:\n%s", q)
+	// parent_namespace must be dropped BEFORE px.display (not leak to the sink).
+	di, pi := strings.Index(q, "df.drop(['parent_namespace'])"), strings.Index(q, "px.display")
+	if di < 0 || pi < 0 || di > pi {
+		t.Errorf("parent_namespace must be dropped before px.display; drop@%d display@%d", di, pi)
 	}
-	if !strings.Contains(q, "df = df[df.comm != 'pause']") {
-		t.Errorf("want exact drop of 'pause'; got:\n%s", q)
+}
+
+// TestQueryFor_AncestryOnlyForPpidTables — the ancestry join applies ONLY to dark
+// tables that capture ppid (dc_snoop). creds_change and the dx_* tables have no
+// ppid, so they must NOT get a process_stats-on-ppid parent merge (it would fail to
+// compile), only the child-namespace exclusion.
+func TestQueryFor_AncestryOnlyForPpidTables(t *testing.T) {
+	for _, tbl := range []string{"creds_change", "dx_vfs_events"} {
+		q, err := QueryFor(tbl, target, fixedStart, fixedEnd, fixedNow)
+		if err != nil {
+			t.Fatalf("QueryFor(%s): %v", tbl, err)
+		}
+		if strings.Contains(q, "parent_namespace") || strings.Contains(q, "left_on=['ppid']") {
+			t.Errorf("%s has no ppid — must NOT get the ancestry join; got:\n%s", tbl, q)
+		}
+		// but it still gets the child-namespace exclusion
+		if !strings.Contains(q, "df = df[df.namespace != 'pl']") {
+			t.Errorf("%s must still get child-namespace exclusion; got:\n%s", tbl, q)
+		}
 	}
 }
diff --git a/src/vizier/services/adaptive_export/internal/script/dc_snoop_contract_test.go b/src/vizier/services/adaptive_export/internal/script/dc_snoop_contract_test.go
new file mode 100644
index 00000000000..521bb025176
--- /dev/null
+++ b/src/vizier/services/adaptive_export/internal/script/dc_snoop_contract_test.go
@@ -0,0 +1,156 @@
+// Copyright 2018- The Pixie Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package script
+
+import (
+	"regexp"
+	"sort"
+	"strings"
+	"testing"
+
+	"px.dev/pixie/src/vizier/services/adaptive_export/internal/clickhouse"
+)
+
+var (
+	projRe = regexp.MustCompile(`df = df\[\[([^\]]*)\]\]`)
+	quotRe = regexp.MustCompile(`'([^']+)'`)
+)
+
+// lastProjection returns the column names in the final `df = df[['a','b',...]]`
+// projection of an export script.
+func lastProjection(script string) []string {
+	m := projRe.FindAllStringSubmatch(script, -1)
+	if len(m) == 0 {
+		return nil
+	}
+	var cols []string
+	for _, q := range quotRe.FindAllStringSubmatch(m[len(m)-1][1], -1) {
+		cols = append(cols, q[1])
+	}
+	return cols
+}
+
+// TestDcSnoopExportColumnsMatchSchema is the strict-sink contract: the columns the
+// dc_snoop retention export projects — plus event_time (added via df.event_time =
+// df.time_) — must be EXACTLY the forensic_db.dc_snoop schema columns. A mismatch
+// means the OTel/ClickHouse sink sends an unknown or missing column and the INSERT
+// fails. This is the coupling that adding ppid/pcomm/pid_start/ppid_start could have
+// silently broken, and it guards the steered path too (queryfor auto-carries the
+// tracepoint columns, so schema == export == tracepoint-derived).
+func TestDcSnoopExportColumnsMatchSchema(t *testing.T) {
+	proj := lastProjection(dcSnoopScript)
+	if len(proj) == 0 {
+		t.Fatal("no df[[...]] projection found in dc_snoop.pxl")
+	}
+	got := append(append([]string{}, proj...), "event_time")
+	want, err := clickhouse.Columns("dc_snoop")
+	if err != nil {
+		t.Fatalf("Columns(dc_snoop): %v", err)
+	}
+	sort.Strings(got)
+	sort.Strings(want)
+	if strings.Join(got, ",") != strings.Join(want, ",") {
+		t.Fatalf("dc_snoop export columns != schema columns\n export=%v\n schema=%v", got, want)
+	}
+}
+
+// TestDcSnoopTracepointCapturesParent — the bpftrace program must emit the parent
+// identity inline (curtask->real_parent via a $tk intermediate) so every dcache
+// event carries ppid/pcomm and a pid-reuse-stable start time (group_leader->
+// start_time) for both the process and its parent — the process-forest edge.
+//
+// This is the exact structure of the last image that CAPTURED (9fd0ca430, 49132
+// rows): $tk intermediate, real_parent, raw group_leader->start_time, 9 args incl
+// pcomm. The ONLY change is the u64 start fields use %llu, not %lld: the 64-bit
+// 'lld' verb is not handled by Pixie's tracepoint printf output parser and
+// misaligns every field after it, which is why ppid parsed as 0 despite being a
+// correct %d. %llu is the proven u64 verb (time_ uses it). A prior "simplify" to
+// inline casts / ->parent / a /10000000 divide made the program capture ZERO rows
+// (ae_reconcile read 49132 -> 0), so this test pins the capturing form.
+func TestDcSnoopTracepointCapturesParent(t *testing.T) {
+	for _, tok := range []string{
+		"ppid:", "pcomm:", "pid_start:", "ppid_start:", "group_leader->start_time",
+	} {
+		if !strings.Contains(dcSnoopDeployScript, tok) {
+			t.Errorf("dc_snoop_deploy.pxl missing %q — parent/identity capture incomplete", tok)
+		}
+	}
+	// Both probe blocks (kprobe:lookup_fast + kretprobe:d_lookup) must carry the
+	// parent pid via the capturing real_parent form.
+	if n := strings.Count(dcSnoopDeployScript, "real_parent->pid"); n < 2 {
+		t.Errorf("dc_snoop_deploy.pxl: real_parent->pid must appear in BOTH probe blocks, got %d", n)
+	}
+	// The %lld regression must not creep back — it is not in Pixie's tracepoint
+	// printf output parser and silently zeroes every field after it (that was the
+	// ppid=0 bug). The u64 start fields must use %llu.
+	if strings.Contains(dcSnoopDeployScript, "%lld") {
+		t.Error("dc_snoop_deploy.pxl uses percent-lld — misaligns output; use percent-llu for u64 start fields")
+	}
+	if strings.Count(dcSnoopDeployScript, "pid_start:%llu") < 2 || strings.Count(dcSnoopDeployScript, "ppid_start:%llu") < 2 {
+		t.Error("dc_snoop_deploy.pxl: pid_start/ppid_start must use percent-llu in both probe blocks")
+	}
+	// A /10000000 divide inside the printf args coincided with zero capture — keep
+	// the raw start_time (the proven-capturing form).
+	if strings.Contains(dcSnoopDeployScript, "10000000") {
+		t.Error("dc_snoop_deploy.pxl divides start_time — the proven-capturing form uses raw group_leader->start_time")
+	}
+	// Tracepoint fields must be a superset of the raw (non-enriched) schema columns
+	// the export reads straight from the table.
+	for _, c := range []string{"time_", "pid", "pid_start", "ppid", "ppid_start", "comm", "pcomm", "t", "file"} {
+		if !strings.Contains(dcSnoopDeployScript, c+":") {
+			t.Errorf("dc_snoop_deploy.pxl printf missing field %q", c)
+		}
+	}
+}
+
+// TestDcSnoopAncestryFilterJoinsParentNamespace — the export must resolve the
+// parent's namespace via a process_stats join keyed on ppid, and the resolved
+// parent_namespace must be a TEMP column (used only for the drop, never projected
+// to the sink — else the INSERT gets an unknown column). The actual drops are
+// injected from env by presets.go and asserted in presets_test.go.
+func TestDcSnoopAncestryFilterJoinsParentNamespace(t *testing.T) {
+	for _, tok := range []string{
+		"par = px.DataFrame(table='process_stats'",
+		"par.parent_namespace = par.ctx['namespace']",
+		"par.ppid = px.upid_to_pid(par.upid)",
+		"left_on=['ppid'], right_on=['ppid']",
+		"# __DC_SNOOP_PARENT_EXCLUSION__",
+	} {
+		if !strings.Contains(dcSnoopScript, tok) {
+			t.Errorf("dc_snoop.pxl ancestry filter missing %q", tok)
+		}
+	}
+	// parent_namespace must NOT reach the sink — it is not a schema column.
+	if proj := lastProjection(dcSnoopScript); contains(proj, "parent_namespace") {
+		t.Error("parent_namespace leaked into the final projection — sink would reject the INSERT")
+	}
+}
+
+func contains(ss []string, want string) bool {
+	for _, s := range ss {
+		if s == want {
+			return true
+		}
+	}
+	return false
+}
+
+// TestDcSnoopCollapseKeepsRepeats — the path-walk collapse must (a) exist and be
+// keyed by basename so distinct files never merge, and (b) preserve every real
+// repeat with its exact timestamp: it filters to the max-length row per micro-
+// window (no count/dedup aggregation), so two identical accesses survive as two
+// rows.
+func TestDcSnoopCollapseKeepsRepeats(t *testing.T) {
+	for _, tok := range []string{
+		"px.length(df.file)", "px.replace('.*/'", "px.bin(df.time_", "'leaf'", "flen_max", "df.flen == df.flen_max",
+	} {
+		if !strings.Contains(dcSnoopScript, tok) {
+			t.Errorf("dc_snoop.pxl collapse missing %q", tok)
+		}
+	}
+	// Must NOT fold repeats into counts — that would drop the exact timestamps.
+	if strings.Contains(dcSnoopScript, "px.count") {
+		t.Errorf("dc_snoop.pxl must NOT aggregate repeats into counts (keep exact timestamps)")
+	}
+}
diff --git a/src/vizier/services/adaptive_export/internal/script/presets.go b/src/vizier/services/adaptive_export/internal/script/presets.go
index 212f869b50b..b6c50d34d61 100644
--- a/src/vizier/services/adaptive_export/internal/script/presets.go
+++ b/src/vizier/services/adaptive_export/internal/script/presets.go
@@ -15,14 +15,6 @@ var defaultExcludeNamespaces = []string{
 	"kube-system", "kube-public", "kube-node-lease", "local-path-storage",
 }
 
-var defaultExcludeComms = []string{
-	"k3s-server", "k3s-agent", "containerd", "containerd-shim",
-	"runc", "runc:[2:INIT]", "runc:[1:CHILD]", "node-agent", "kelvin",
-	"vizier-pem", "vizier-query-broker", "vizier-metadata",
-	"systemd", "systemd-journal", "iptables", "ip6tables", "kubelet",
-	"operator", "storage",
-}
-
 func csvEnv(key string, def []string) []string {
 	v := os.Getenv(key)
 	if v == "" {
@@ -37,17 +29,31 @@ func csvEnv(key string, def []string) []string {
 	return out
 }
 
-// dcSnoopExclusion builds the dc_snoop noise filter (namespace + comm drops) from
-// DC_SNOOP_EXCLUDE_NAMESPACES / DC_SNOOP_EXCLUDE_COMMS, substituted into
-// dc_snoop.pxl at # __DC_SNOOP_EXCLUSION__ so a process can be added without a
-// recompile. Kept in sync with dx benchlive.writeSelfExclusion.
+// dcSnoopExclusion builds the dc_snoop CHILD-namespace noise filter from
+// DC_SNOOP_EXCLUDE_NAMESPACES, substituted into dc_snoop.pxl at
+// # __DC_SNOOP_EXCLUSION__ so an infra namespace can be added without a recompile.
+// The hardcoded comm blocklist was DELETED (pure-ancestry cleanup): own-stack pods
+// are dropped here by their resolved namespace, and their transient blank-namespace
+// children by the parent-ancestry filter (dcSnoopParentExclusion). Host/kernel
+// processes (blank namespace, blank/kernel parent) are left to dx's process forest.
 func dcSnoopExclusion() string {
 	var b strings.Builder
 	for _, ns := range csvEnv("DC_SNOOP_EXCLUDE_NAMESPACES", defaultExcludeNamespaces) {
 		fmt.Fprintf(&b, "df = df[df.namespace != '%s']\n", ns)
 	}
-	for _, c := range csvEnv("DC_SNOOP_EXCLUDE_COMMS", defaultExcludeComms) {
-		fmt.Fprintf(&b, "df = df[df.comm != '%s']\n", c)
+	return strings.TrimRight(b.String(), "\n")
+}
+
+// dcSnoopParentExclusion builds the ppid-ancestry filter (drop events whose PARENT
+// resolves to an own-stack namespace) from the SAME namespace list as the self
+// filter, substituted into dc_snoop.pxl at # __DC_SNOOP_PARENT_EXCLUSION__. Rooted
+// on the parent's namespace (via the ppid->process_stats join), not comm, so it
+// catches transient children exec'd by infra pods without touching workload/attack
+// children of shared (blank-namespace) parents like containerd-shim/runc.
+func dcSnoopParentExclusion() string {
+	var b strings.Builder
+	for _, ns := range csvEnv("DC_SNOOP_EXCLUDE_NAMESPACES", defaultExcludeNamespaces) {
+		fmt.Fprintf(&b, "df = df[df.parent_namespace != '%s']\n", ns)
 	}
 	return strings.TrimRight(b.String(), "\n")
 }
@@ -100,7 +106,7 @@ func DesiredTracepoints() []TracepointDef {
 // registers if-not-present. Names are operator-managed (reconciled on boot).
 func DarkVectorPresets() []*ScriptDefinition {
 	return []*ScriptDefinition{
-		{Name: "ch-dc_snoop", Description: "dc_snoop (dentry cache: process+file, V1/V2) → ClickHouse", FrequencyS: 10, Script: strings.Replace(dcSnoopScript, "# __DC_SNOOP_EXCLUSION__", dcSnoopExclusion(), 1)},
+		{Name: "ch-dc_snoop", Description: "dc_snoop (dentry cache: process+file, V1/V2) → ClickHouse", FrequencyS: 10, Script: strings.Replace(strings.Replace(dcSnoopScript, "# __DC_SNOOP_PARENT_EXCLUSION__", dcSnoopParentExclusion(), 1), "# __DC_SNOOP_EXCLUSION__", dcSnoopExclusion(), 1)},
 		{Name: "ch-stack_trace", Description: "stack_traces.beta (continuous profiler, V9) → ClickHouse", FrequencyS: 10, Script: stackTraceScript},
 		{Name: "ch-creds_change", Description: "commit_creds privilege-escalation to root (V7) → ClickHouse", FrequencyS: 10, Script: credsChangeScript},
 	}
diff --git a/src/vizier/services/adaptive_export/internal/script/presets/dc_snoop.pxl b/src/vizier/services/adaptive_export/internal/script/presets/dc_snoop.pxl
index b930f3907ef..490cee24747 100644
--- a/src/vizier/services/adaptive_export/internal/script/presets/dc_snoop.pxl
+++ b/src/vizier/services/adaptive_export/internal/script/presets/dc_snoop.pxl
@@ -24,6 +24,25 @@ import px
 table_name = 'dc_snoop'
 df = px.DataFrame(table=table_name, start_time=px.plugin.start_time, end_time=px.plugin.end_time)
 
+# Collapse dcache path-walk amplification. One path resolution fires the kprobe
+# once per component (nd->last.name is the shrinking *remaining* path), so a single
+# file access lands as N suffix rows (sh, bin/sh, ..., opt/bitnami/common/bin/sh).
+# Keep only the full path (longest fragment) per (pid, comm, t, basename, 50us
+# window): all fragments of one walk share the basename, so grouping on it never
+# merges two DISTINCT files; the tight window separates rare same-basename accesses,
+# and a walk straddling a bin boundary only UNDER-collapses (safe, never drops a
+# real access). Distinct repeats of the same file are preserved (kept as separate
+# rows) — folding those is a separate (pid,comm,file,window)->count pass. ~3-4x.
+# Runs before the pod merge so the join is over the collapsed set. [rig-validate]
+df.flen = px.length(df.file)
+df.leaf = px.replace('.*/', df.file, '')
+df.win = px.bin(df.time_, 50000)
+grp = df.groupby(['pid', 'comm', 't', 'leaf', 'win']).agg(flen_max=('flen', px.max))
+df = df.merge(grp, how='inner',
+              left_on=['pid', 'comm', 't', 'leaf', 'win'],
+              right_on=['pid', 'comm', 't', 'leaf', 'win'], suffixes=['', '_g'])
+df = df[df.flen == df.flen_max]
+
 # pid -> pod/namespace attribution. Dark-vector tracepoints emit a raw kernel pid
 # with NO upid, so px.upid_to_* / ctx['pod'] fail outright; resolve pod by merging
 # process_stats on pid ONLY - the validated join (PodEnrichPxL, dx#126). NOT
@@ -38,9 +57,32 @@ proc.hostname = px.upid_to_node_name(proc.upid)
 proc.pid = px.upid_to_pid(proc.upid)
 proc = proc.groupby(['namespace', 'pod', 'container', 'hostname', 'pid']).agg()
 df = df.merge(proc, how='left', left_on=['pid'], right_on=['pid'], suffixes=['', '_x'])
+
+# Ancestry filter (the ppid-based noise cut). Resolve the PARENT's namespace via
+# the same process_stats join, keyed on ppid, and drop events whose parent lives
+# in an own-stack namespace even when the child's own pod is blank (a transient
+# process exec'd by an infra pod - the dcache firehose we were drowning in). This
+# is pod/namespace-rooted, NOT comm-matched, so a real attack child of a shared
+# parent (containerd-shim/runc carry no namespace) survives. Best-effort left
+# join: parent_namespace stays blank for host/transient/unseen parents (kept - a
+# blank parent is not proven own-stack). The drops are injected from env below by
+# presets.go (DC_SNOOP_EXCLUDE_NAMESPACES, same list as the self-namespace
+# filter). parent_namespace is a temp column dropped by the final
+# projection below, so the schema/sink coupling is unchanged. Multi-level ancestry
+# (walk ppid->parent->... past shared parents) is dx's job on the process forest.
+par = px.DataFrame(table='process_stats', start_time='-5m')
+par.parent_namespace = par.ctx['namespace']
+par.ppid = px.upid_to_pid(par.upid)
+par = par.groupby(['parent_namespace', 'ppid']).agg()
+df = df.merge(par, how='left', left_on=['ppid'], right_on=['ppid'], suffixes=['', '_par'])
+# __DC_SNOOP_PARENT_EXCLUSION__
+
 # Keep exactly the forensic_db.dc_snoop columns (drop the merge's pid_x etc.),
-# else the export sink sends an unknown column and the INSERT fails.
-df = df[['time_', 'pid', 'comm', 't', 'file', 'namespace', 'pod', 'container', 'hostname']]
+# else the export sink sends an unknown column and the INSERT fails. ppid/pcomm/
+# pid_start/ppid_start come straight from the tracepoint (no join). Parent pod
+# resolution + multi-level ancestry is dx's job against the process forest.
+df = df[['time_', 'pid', 'pid_start', 'ppid', 'ppid_start', 'comm', 'pcomm', 't', 'file',
+         'namespace', 'pod', 'container', 'hostname']]
 
 # Drop known infrastructure namespaces + process comms (blank-namespace workload
 # rows are kept). The filter is injected here from env by presets.go
diff --git a/src/vizier/services/adaptive_export/internal/script/presets/dc_snoop_deploy.pxl b/src/vizier/services/adaptive_export/internal/script/presets/dc_snoop_deploy.pxl
index bf8333f0a4b..60a94cca9d0 100644
--- a/src/vizier/services/adaptive_export/internal/script/presets/dc_snoop_deploy.pxl
+++ b/src/vizier/services/adaptive_export/internal/script/presets/dc_snoop_deploy.pxl
@@ -25,6 +25,7 @@ import pxtrace
 
 program = """
 #include <linux/fs.h>
+#include <linux/sched.h>
 
 // from fs/namei.c:
 struct nameidata {
@@ -33,13 +34,32 @@ struct nameidata {
                 // [...]
 };
 
-// comment out this block to avoid showing hits:
+// pid_start/ppid/ppid_start/pcomm come from the current task's real_parent via a
+// $tk intermediate so every dcache event carries its parent inline (curtask struct
+// walks in a kprobe are supported — see pxbeta/vfs_snoop). group_leader->start_time
+// (ns since boot) is a pid-reuse-stable identity for the process and its parent.
+//
+// This is the EXACT structure of the last image that CAPTURED (9fd0ca430: 49132
+// rows) — $tk intermediate, real_parent, raw group_leader->start_time, 9 args
+// including pcomm — with ONE change: the two u64 start fields use the unsigned
+// 64-bit verb, not the signed one. The signed 64-bit verb is not handled by
+// Pixie's tracepoint printf output parser; it misaligns every field after it,
+// which is why ppid parsed as 0 despite being a correct signed-int verb. The
+// unsigned 64-bit verb is proven (time_ uses it; see the connector's u64 tests).
+// Do NOT "simplify" to inline casts, ->parent, or a divided start_time: those
+// changes (my earlier rewrite) made the program capture ZERO rows.
 kprobe:lookup_fast,
 kprobe:lookup_fast.constprop.*
 {
         $nd = (struct nameidata *)arg0;
-        printf("time_:%llu pid:%d comm:%s t:%s file:%s",
-                     nsecs, pid, comm, "R", str($nd->last.name));
+        $tk = (struct task_struct *)curtask;
+        printf("time_:%llu pid:%d pid_start:%llu ppid:%d ppid_start:%llu comm:%s pcomm:%s t:%s file:%s",
+                     nsecs, pid,
+                     $tk->group_leader->start_time,
+                     $tk->real_parent->pid,
+                     $tk->real_parent->group_leader->start_time,
+                     comm, $tk->real_parent->comm,
+                     "R", str($nd->last.name));
 }
 
 kprobe:d_lookup
@@ -51,8 +71,14 @@ kprobe:d_lookup
 kretprobe:d_lookup
 /@fname[tid]/
 {
-        printf("time_:%llu pid:%d comm:%s t:%s file:%s",
-                     nsecs, pid, comm, "M", str(@fname[tid]));
+        $tk = (struct task_struct *)curtask;
+        printf("time_:%llu pid:%d pid_start:%llu ppid:%d ppid_start:%llu comm:%s pcomm:%s t:%s file:%s",
+                     nsecs, pid,
+                     $tk->group_leader->start_time,
+                     $tk->real_parent->pid,
+                     $tk->real_parent->group_leader->start_time,
+                     comm, $tk->real_parent->comm,
+                     "M", str(@fname[tid]));
         delete(@fname[tid]);
 }
 """
diff --git a/src/vizier/services/adaptive_export/internal/script/presets_test.go b/src/vizier/services/adaptive_export/internal/script/presets_test.go
index 2ea0f01ab22..d4f141035bc 100644
--- a/src/vizier/services/adaptive_export/internal/script/presets_test.go
+++ b/src/vizier/services/adaptive_export/internal/script/presets_test.go
@@ -24,27 +24,81 @@ func TestDcSnoopExclusionDefault(t *testing.T) {
 	if strings.Contains(s, "#__DC_SNOOP_EXCLUSION__") {
 		t.Fatal("exclusion placeholder was not substituted")
 	}
+	// Child-namespace drops present for infra namespaces.
 	for _, want := range []string{
-		"df = df[df.comm != 'k3s-server']",
-		"df = df[df.comm != 'runc:[2:INIT]']",
 		"df = df[df.namespace != 'honey']",
+		"df = df[df.namespace != 'pl']",
+		"df = df[df.namespace != 'clickhouse']",
 	} {
 		if !strings.Contains(s, want) {
-			t.Errorf("default filter missing: %s", want)
+			t.Errorf("default namespace filter missing: %s", want)
 		}
 	}
+	// The hardcoded comm blocklist was DELETED (pure-ancestry cleanup): the retention
+	// script must emit NO comm-based drops.
+	if strings.Contains(s, "df.comm !=") {
+		t.Error("comm blocklist must be gone from dc_snoop.pxl (no df.comm != drops)")
+	}
+	if strings.Contains(s, "px.contains(df.comm") {
+		t.Error("kernel-thread substring filter must be gone from dc_snoop.pxl")
+	}
 	if strings.Contains(s, "df = df[df.namespace != '']") {
 		t.Error("must NOT drop blank-namespace rows")
 	}
 }
 
-func TestDcSnoopExclusionConfigurable(t *testing.T) {
-	t.Setenv("DC_SNOOP_EXCLUDE_COMMS", "foo, bar")
+func TestDcSnoopParentExclusion(t *testing.T) {
 	s := chDcSnoop(t)
-	if !strings.Contains(s, "df = df[df.comm != 'foo']") || !strings.Contains(s, "df = df[df.comm != 'bar']") {
-		t.Error("DC_SNOOP_EXCLUDE_COMMS override not applied")
+	if strings.Contains(s, "# __DC_SNOOP_PARENT_EXCLUSION__") {
+		t.Fatal("parent-exclusion placeholder was not substituted")
 	}
-	if strings.Contains(s, "k3s-server") {
-		t.Error("env override should replace, not append to, the default comm list")
+	// The ppid-ancestry filter drops on parent_namespace, using the same namespace
+	// list as the self filter.
+	for _, want := range []string{
+		"df = df[df.parent_namespace != 'pl']",
+		"df = df[df.parent_namespace != 'honey']",
+		"df = df[df.parent_namespace != 'clickhouse']",
+	} {
+		if !strings.Contains(s, want) {
+			t.Errorf("parent ancestry filter missing: %s", want)
+		}
+	}
+	// Never drop rows whose parent namespace is blank (host/transient/unseen
+	// parents are kept — a blank parent is not proven own-stack).
+	if strings.Contains(s, "df = df[df.parent_namespace != '']") {
+		t.Error("must NOT drop blank-parent-namespace rows")
+	}
+	// The parent join must resolve ppid -> parent's namespace.
+	for _, want := range []string{
+		"par.parent_namespace = par.ctx['namespace']",
+		"par.ppid = px.upid_to_pid(par.upid)",
+		"left_on=['ppid'], right_on=['ppid']",
+	} {
+		if !strings.Contains(s, want) {
+			t.Errorf("parent join step missing: %s", want)
+		}
+	}
+}
+
+func TestDcSnoopParentExclusionConfigurable(t *testing.T) {
+	t.Setenv("DC_SNOOP_EXCLUDE_NAMESPACES", "foo, bar")
+	s := chDcSnoop(t)
+	if !strings.Contains(s, "df = df[df.parent_namespace != 'foo']") ||
+		!strings.Contains(s, "df = df[df.parent_namespace != 'bar']") {
+		t.Error("DC_SNOOP_EXCLUDE_NAMESPACES override not applied to parent filter")
+	}
+	if strings.Contains(s, "df = df[df.parent_namespace != 'pl']") {
+		t.Error("env override should replace, not append to, the default namespace list")
+	}
+}
+
+// TestDcSnoopNoCommMachinery — the DC_SNOOP_EXCLUDE_COMMS env override and its
+// backing list are gone; setting the (now-defunct) var must NOT reintroduce any
+// comm drop into the retention script.
+func TestDcSnoopNoCommMachinery(t *testing.T) {
+	t.Setenv("DC_SNOOP_EXCLUDE_COMMS", "foo,bar")
+	s := chDcSnoop(t)
+	if strings.Contains(s, "df.comm !=") {
+		t.Error("DC_SNOOP_EXCLUDE_COMMS must be a no-op now (pure ancestry); got comm drops in script")
 	}
 }
diff --git a/src/vizier/services/adaptive_export/internal/sink/dc_snoop_bench_test.go b/src/vizier/services/adaptive_export/internal/sink/dc_snoop_bench_test.go
new file mode 100644
index 00000000000..530cbcf1d42
--- /dev/null
+++ b/src/vizier/services/adaptive_export/internal/sink/dc_snoop_bench_test.go
@@ -0,0 +1,138 @@
+// Copyright 2018- The Pixie Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package sink
+
+import (
+	"bytes"
+	"testing"
+	"time"
+)
+
+// dcSnoopRow builds a representative dc_snoop event. withParent adds the four
+// fields the ppid change introduces (pid_start/ppid/ppid_start Int64 + pcomm
+// String) — everything else is the pre-change shape.
+func dcSnoopRow(withParent bool) map[string]any {
+	r := map[string]any{
+		"time_":      time.Unix(0, 1_700_000_000_171_199_174),
+		"pid":        int64(159249),
+		"comm":       "sh",
+		"t":          "R",
+		"file":       "opt/bitnami/common/bin/sh",
+		"namespace":  "redis",
+		"pod":        "redis-master-0",
+		"container":  "redis",
+		"hostname":   "node-01",
+		"event_time": time.Unix(0, 1_700_000_000_171_199_174),
+	}
+	if withParent {
+		r["pid_start"] = int64(1_700_000_000_000_000_000)
+		r["ppid"] = int64(90059)
+		r["ppid_start"] = int64(1_699_999_999_000_000_000)
+		r["pcomm"] = "containerd-shim"
+	}
+	return r
+}
+
+// encodeBytesPerRow encodes n identical dc_snoop rows via the same fast path the
+// sink uses (encodePixieRowsFast → appendJSONValue) and returns the wire bytes/row.
+// cols pins the projection so we can measure the old (10-col) vs new (14-col) shape
+// independent of the current schema.
+func encodeBytesPerRow(b *testing.B, withParent bool, cols []string) float64 {
+	const n = 1000
+	rows := make([]map[string]any, n)
+	for i := range rows {
+		rows[i] = dcSnoopRow(withParent)
+	}
+	var buf bytes.Buffer
+	// warm one pass to get the byte size
+	buf.Reset()
+	for _, r := range rows {
+		buf.WriteByte('{')
+		for j, c := range cols {
+			if j > 0 {
+				buf.WriteByte(',')
+			}
+			buf.WriteByte('"')
+			buf.WriteString(c)
+			buf.WriteString(`":`)
+			_ = appendJSONValue(&buf, r[c])
+		}
+		buf.WriteString("}\n")
+	}
+	bytesPerRow := float64(buf.Len()) / n
+
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		buf.Reset()
+		for _, r := range rows {
+			buf.WriteByte('{')
+			for j, c := range cols {
+				if j > 0 {
+					buf.WriteByte(',')
+				}
+				buf.WriteByte('"')
+				buf.WriteString(c)
+				buf.WriteString(`":`)
+				_ = appendJSONValue(&buf, r[c])
+			}
+			buf.WriteString("}\n")
+		}
+	}
+	return bytesPerRow
+}
+
+var (
+	dcSnoopOldCols = []string{"time_", "pid", "comm", "t", "file", "namespace", "pod", "container", "hostname", "event_time"}
+	dcSnoopNewCols = []string{"time_", "pid", "pid_start", "ppid", "ppid_start", "comm", "pcomm", "t", "file", "namespace", "pod", "container", "hostname", "event_time"}
+)
+
+// BenchmarkDCSnoopEncode_Baseline / _WithParent measure the per-event wire bytes
+// before and after the ppid addition. The delta (reported as bytes/row) IS the
+// additional data the change loads per dcache event.
+func BenchmarkDCSnoopEncode_Baseline(b *testing.B) {
+	bpr := encodeBytesPerRow(b, false, dcSnoopOldCols)
+	b.ReportMetric(bpr, "bytes/row")
+}
+
+func BenchmarkDCSnoopEncode_WithParent(b *testing.B) {
+	bpr := encodeBytesPerRow(b, true, dcSnoopNewCols)
+	b.ReportMetric(bpr, "bytes/row")
+	// Columnar (PEM table-store) cost of the 4 added fields: 3×Int64 + comm(16).
+	b.ReportMetric(3*8+16, "columnar_add_bytes/row")
+}
+
+// TestDCSnoopPerEventDataDelta prints the concrete numbers (not just a benchmark
+// metric) so the RCA has a hard figure: added wire bytes/row and the columnar
+// (PEM) add, plus what that is per 1M dcache events.
+func TestDCSnoopPerEventDataDelta(t *testing.T) {
+	oldB := sizeOnce(false, dcSnoopOldCols)
+	newB := sizeOnce(true, dcSnoopNewCols)
+	addWire := newB - oldB
+	addColumnar := 3*8 + 16 // pid_start+ppid+ppid_start (Int64) + pcomm (comm String, 16)
+	t.Logf("dc_snoop per-event data delta:")
+	t.Logf("  wire (JSON) bytes/row:      old=%d  new=%d  +%d", oldB, newB, addWire)
+	t.Logf("  columnar (PEM) bytes/row:   +%d (3xInt64 + comm16)", addColumnar)
+	t.Logf("  per 1,000,000 events:       +%d MB wire, +%d MB columnar", addWire, addColumnar)
+	t.Logf("  NOTE: PEM table-store is CAPPED (PL_TABLE_STORE_DATA_LIMIT_MB) — the +40B/event")
+	t.Logf("        fills the cap faster (shorter lookback) but does NOT raise peak memory.")
+	t.Logf("        collapse also removes ~3.5x rows at export, so net exported data DROPS.")
+}
+
+func sizeOnce(withParent bool, cols []string) int {
+	r := dcSnoopRow(withParent)
+	var buf bytes.Buffer
+	buf.WriteByte('{')
+	for j, c := range cols {
+		if j > 0 {
+			buf.WriteByte(',')
+		}
+		buf.WriteByte('"')
+		buf.WriteString(c)
+		buf.WriteString(`":`)
+		_ = appendJSONValue(&buf, r[c])
+	}
+	buf.WriteString("}\n")
+	return buf.Len()
+}