fix: count distinct chunk IDs so doc.chunk_num matches actual indexed chunks (#17182)

internal/ingestion/task/pipeline_executor.go

@@ -224,16 +224,36 @@ func (s *PipelineExecutor) processOutput(ctx context.Context, pipelineOutput map
 		return nil, err
 	}
 
+	chunkCount := countDistinctChunkIDs(chunks)
+
 	return &PipelineResult{
 		DocID:            s.taskCtx.Doc.ID,
 		KbID:             s.taskCtx.Doc.KbID,
 		Metadata:         metadata,
-		ChunkCount:       len(chunks),
+		ChunkCount:       chunkCount,
 		TokenConsumption: embeddingTokenConsumption,
 		Duration:         time.Since(start).Seconds(),
 	}, nil
 }
 
+// countDistinctChunkIDs returns the number of distinct chunk IDs in the slice.
+// After ProcessChunksForPipeline, every chunk carries an "id" field computed
+// from xxhash(text+docID). Chunks with identical text share the same id and
+// the search engine treats them as upserts, so the effective stored chunk count
+// is the number of unique ids — not len(chunks). Mirrors the index-side
+// deduplication that happens at write time.
+func countDistinctChunkIDs(chunks []map[string]any) int {
+	seen := make(map[string]struct{}, len(chunks))
+	for _, ck := range chunks {
+		id, _ := ck["id"].(string)
+		if id == "" {
+			continue
+		}
+		seen[id] = struct{}{}
+	}
+	return len(seen)
+}
+
 func (s *PipelineExecutor) recordPipelineLog(docID, dsl, status string) {
 	var dslMap entity.JSONMap
 	if err := json.Unmarshal([]byte(dsl), &dslMap); err != nil {

internal/ingestion/task/pipeline_executor_test.go

@@ -472,3 +472,42 @@ func TestPipelineExecutorRunPipelineWithDSLForwardsSink(t *testing.T) {
 		}
 	}
 }
+
+func TestCountDistinctChunkIDs(t *testing.T) {
+	// Empty list
+	if n := countDistinctChunkIDs(nil); n != 0 {
+		t.Fatalf("nil chunks: got %d, want 0", n)
+	}
+
+	// All unique
+	chunks := []map[string]any{
+		{"id": "a"},
+		{"id": "b"},
+		{"id": "c"},
+	}
+	if n := countDistinctChunkIDs(chunks); n != 3 {
+		t.Fatalf("all unique: got %d, want 3", n)
+	}
+
+	// Duplicates present — this is the key case
+	chunks = []map[string]any{
+		{"id": "x"},
+		{"id": "y"},
+		{"id": "x"}, // duplicate of [0]
+		{"id": "z"},
+		{"id": "y"}, // duplicate of [1]
+	}
+	if n := countDistinctChunkIDs(chunks); n != 3 {
+		t.Fatalf("with duplicates: got %d, want 3", n)
+	}
+
+	// Missing id fields are skipped
+	chunks = []map[string]any{
+		{"id": "one"},
+		{"text": "no id"},
+		{"id": "two"},
+	}
+	if n := countDistinctChunkIDs(chunks); n != 2 {
+		t.Fatalf("mixed present/absent ids: got %d, want 2", n)
+	}
+}