fix: restrict file agg events to ai agents

Author: kylewanginchina

server/ingester/event/decoder/decoder.go

@@ -402,6 +402,10 @@ func splitFilePath(fullPath string) (string, string) {
 	return "", fullPath
 }
 
+func shouldAggregateFileAggEvent(rootPID uint32) bool {
+	return rootPID != 0
+}
+
 func (d *Decoder) emitFileAggItems(items []*dbwriter.FileAggEventStore) {
 	if len(items) == 0 {
 		return
@@ -456,7 +460,7 @@ func (d *Decoder) writeRawFileEvent(vtapId uint16, e *pb.ProcEvent) {
 
 	d.export(s)
 	d.rawFileWriter().Write(s)
-	if d.fileAggReducer != nil {
+	if d.fileAggReducer != nil && shouldAggregateFileAggEvent(s.RootPID) {
 		d.emitFileAggItems(d.fileAggReducer.Add(s))
 	}
 }

server/ingester/event/decoder/decoder_test.go

@@ -259,3 +259,12 @@ func TestExtractFileMgmtTargets(t *testing.T) {
 		t.Fatalf("chmod targets = (%d,%d,%d)", uid, gid, mode)
 	}
 }
+
+func TestShouldAggregateFileAggEvent(t *testing.T) {
+	if shouldAggregateFileAggEvent(0) {
+		t.Fatalf("expected root pid 0 to skip file_agg_event aggregation")
+	}
+	if !shouldAggregateFileAggEvent(42) {
+		t.Fatalf("expected non-zero root pid to allow file_agg_event aggregation")
+	}
+}

server/ingester/event/decoder/file_agg_reducer.go

@@ -113,6 +113,9 @@ func shouldSplitFileAggByIdleGap(current, next *dbwriter.FileAggEventStore) bool
 }
 
 func (r *FileAggReducer) Add(raw *dbwriter.EventStore) []*dbwriter.FileAggEventStore {
+	if raw == nil || raw.RootPID == 0 {
+		return nil
+	}
 	next := cloneRawFileEventToAgg(raw)
 	if next == nil {
 		return nil

server/ingester/event/decoder/file_agg_reducer_test.go

@@ -195,3 +195,19 @@ func TestFileAggReducerSplitsSameKeyWhenIdleGapExceeded(t *testing.T) {
 		t.Fatalf("remaining event_count = %d, want 1", remaining[0].EventCount)
 	}
 }
+
+func TestFileAggReducerSkipsNonAiAgentEvent(t *testing.T) {
+	reducer := NewFileAggReducer()
+	raw := makeRawFileEvent("bash", "read", "plain.txt", 16)
+	raw.RootPID = 0
+
+	flushed := reducer.Add(raw)
+	if flushed != nil {
+		t.Fatalf("add returned %d flushed items, want nil", len(flushed))
+	}
+
+	remaining := reducer.Flush()
+	if remaining != nil {
+		t.Fatalf("flush returned %d items, want nil", len(remaining))
+	}
+}