diff --git a/internal/ingestion/task/debug_log_sink.go b/internal/ingestion/task/debug_log_sink.go index 0bef558c43..5d99709278 100644 --- a/internal/ingestion/task/debug_log_sink.go +++ b/internal/ingestion/task/debug_log_sink.go @@ -271,7 +271,7 @@ func (s *DebugLogSink) Flush(ctx context.Context, finalErr error) { // `json.dumps(self.get_component_obj(self.path[-1]).output())` // (rag/flow/pipeline.py:171). The last executed component is the sink's final // entry (path[-1] always emits its exit progress last). Raw embedding vectors -// are stripped (deepCopyStrip) so the Redis log stays at Python-scale size. +// are stripped (deepCopy(out, true)) so the Redis log stays at Python-scale size. // Returns "" when no output is available; the caller then keeps the // plain-text fallback and the front-end export button stays disabled (empty // output, matching Python's isEmpty check). @@ -284,7 +284,7 @@ func endOutputMessage(entries []debugLogEntry, runOutput map[string]any) string if !ok || len(out) == 0 { return "" } - b, err := json.Marshal(deepCopyStrip(out)) + b, err := json.Marshal(deepCopy(out, true)) if err != nil { return "" } diff --git a/internal/ingestion/task/debug_log_sink_test.go b/internal/ingestion/task/debug_log_sink_test.go index 1bb048ba62..49520a74a6 100644 --- a/internal/ingestion/task/debug_log_sink_test.go +++ b/internal/ingestion/task/debug_log_sink_test.go @@ -264,7 +264,7 @@ type traceChunkComponent struct{} func (traceChunkComponent) Invoke(_ context.Context, _ *gorm.DB, _ map[string]any) (map[string]any, error) { // Use the real component output shape ([]map[string]any as produced by // ChunkDocsToMaps) so the end-to-end strip/recurse path for []map[string]any - // is actually exercised — a []any stub hides the deepCopyStrip type gap. + // is actually exercised — a []any stub hides the deepCopy type gap. return map[string]any{ "chunks": []map[string]any{ {"text": "real chunk", "vector": []float64{0.5}}, diff --git a/internal/ingestion/task/debug_result_dsl.go b/internal/ingestion/task/debug_result_dsl.go index c6ebbda85e..01034c70ab 100644 --- a/internal/ingestion/task/debug_result_dsl.go +++ b/internal/ingestion/task/debug_result_dsl.go @@ -58,13 +58,16 @@ var vectorKeys = map[string]struct{}{ "feature": {}, } -// isVectorKey reports whether k is a raw embedding-vector key that must be +// IsVectorKey reports whether k is a raw embedding-vector key that must be // stripped from debug payloads. It matches the fixed legacy keys // (vector/embedding/feature) AND the dimension-scoped pattern q__vec that // the tokenizer actually emits (see hasEmbeddingVector, tokenizer.go:828, e.g. // q_4_vec, q_1024_vec). The literal "q_vec" entry never matches those, so a // bare map lookup would let real vectors leak into the Redis log. -func isVectorKey(k string) bool { +// +// Exported so the golden-compare tool (internal/ingestion/task/tool) reuses the +// exact same stripping rule instead of re-implementing a weaker copy. +func IsVectorKey(k string) bool { if _, ok := vectorKeys[k]; ok { return true } @@ -121,12 +124,12 @@ func BuildDebugResultDSL(dsl string, output map[string]any) (map[string]any, err // (setups/field_name/...), then inject the runtime outputs wrapper. mergedParams := map[string]any{} for k, v := range staticParams { - mergedParams[k] = deepCopy(v) + mergedParams[k] = deepCopy(v, false) } if format, payload := detectFormat(lookupComponentOutput(output, id)); format != "" { mergedParams["outputs"] = map[string]any{ format: map[string]any{ - "value": deepCopyStrip(payload), + "value": deepCopy(payload, true), "type": "", }, "output_format": map[string]any{"value": format}, @@ -145,7 +148,7 @@ func BuildDebugResultDSL(dsl string, output map[string]any) (map[string]any, err result := map[string]any{ "components": built, - "graph": deepCopy(root["graph"]), + "graph": deepCopy(root["graph"], false), } return result, nil } @@ -170,15 +173,16 @@ func lookupComponentOutput(output map[string]any, id string) any { // map[string]any-shaped state, so accept BOTH concrete types — a // single-type assertion would silently fail the real shape and fall through // to the (usually empty) top-level lookup. + var found any + var ok bool switch state := output["state"].(type) { case map[string]map[string]any: - if v, ok := state[id]; ok { - return v - } + found, ok = state[id] case map[string]any: - if v, ok := state[id]; ok { - return v - } + found, ok = state[id] + } + if ok { + return found } // Fallback: flat shape (tests / non-Snapshot producers). return output[id] @@ -202,55 +206,30 @@ func detectFormat(out any) (string, any) { } // deepCopy returns a JSON-compatible deep copy of v (maps/slices/primitives), -// preserving structure but sharing nothing mutable with the source. -func deepCopy(v any) any { +// preserving structure but sharing nothing mutable with the source. When +// stripVector is true it additionally drops vector keys (see IsVectorKey) from +// every map it visits, so raw embedding vectors never reach the debug log. +func deepCopy(v any, stripVector bool) any { switch val := v.(type) { case map[string]any: cp := make(map[string]any, len(val)) for k, vv := range val { - cp[k] = deepCopy(vv) - } - return cp - case []map[string]any: - cp := make([]any, len(val)) - for i, vv := range val { - cp[i] = deepCopy(vv) - } - return cp - case []any: - cp := make([]any, len(val)) - for i, vv := range val { - cp[i] = deepCopy(vv) - } - return cp - default: - return v - } -} - -// deepCopyStrip is deepCopy plus dropping vectorKeys from every map it visits. -// Used for component payloads so raw embedding vectors never reach the log. -func deepCopyStrip(v any) any { - switch val := v.(type) { - case map[string]any: - cp := make(map[string]any, len(val)) - for k, vv := range val { - if isVectorKey(k) { + if stripVector && IsVectorKey(k) { continue } - cp[k] = deepCopyStrip(vv) + cp[k] = deepCopy(vv, stripVector) } return cp case []map[string]any: cp := make([]any, len(val)) for i, vv := range val { - cp[i] = deepCopyStrip(vv) + cp[i] = deepCopy(vv, stripVector) } return cp case []any: cp := make([]any, len(val)) for i, vv := range val { - cp[i] = deepCopyStrip(vv) + cp[i] = deepCopy(vv, stripVector) } return cp default: diff --git a/internal/ingestion/task/debug_result_dsl_test.go b/internal/ingestion/task/debug_result_dsl_test.go index f443fba339..15ea52cd6e 100644 --- a/internal/ingestion/task/debug_result_dsl_test.go +++ b/internal/ingestion/task/debug_result_dsl_test.go @@ -315,13 +315,13 @@ func TestDeepCopyStrip_StripsVectorsFromMapSlice(t *testing.T) { {"text": "second", "feature": []float64{0.4}}, } - got := deepCopyStrip(src) + got := deepCopy(src, true) // Returned slice must be a deep copy (new []any holding new maps), not the // original slice/map identity. cp, ok := got.([]any) if !ok { - t.Fatalf("deepCopyStrip returned %T, want []any", got) + t.Fatalf("deepCopy(src, true) returned %T, want []any", got) } if len(cp) != 2 { t.Fatalf("len=%d want 2", len(cp)) @@ -378,7 +378,7 @@ func TestDeepCopyStrip_StripsVectorsFromMapSlice(t *testing.T) { // (structure preservation, no vector-strip concern for plain deepCopy). func TestDeepCopy_MapSliceIsDeep(t *testing.T) { src := []map[string]any{{"text": "a", "n": map[string]any{"v": 1}}} - got := deepCopy(src) + got := deepCopy(src, false) cp, ok := got.([]any) if !ok { t.Fatalf("deepCopy returned %T, want []any", got) diff --git a/internal/ingestion/task/tool/compare_pipeline_golden.go b/internal/ingestion/task/tool/compare_pipeline_golden.go index 1dc85f949a..decb80c880 100644 --- a/internal/ingestion/task/tool/compare_pipeline_golden.go +++ b/internal/ingestion/task/tool/compare_pipeline_golden.go @@ -101,7 +101,10 @@ func sanitizeProcessed(chunks []map[string]any) []map[string]any { delete(cp, "create_time") delete(cp, "create_timestamp_flt") for k := range cp { - if strings.HasPrefix(k, "q_") && strings.HasSuffix(k, "_vec") { + // Use the production stripping rule so the golden compare never + // disagrees with the debug payload: fixed legacy keys + // (vector/embedding/feature/q_vec) AND q__vec are all dropped. + if task.IsVectorKey(k) { delete(cp, k) } } diff --git a/internal/ingestion/task/tool/compare_pipeline_golden_test.go b/internal/ingestion/task/tool/compare_pipeline_golden_test.go new file mode 100644 index 0000000000..e2d9e3979c --- /dev/null +++ b/internal/ingestion/task/tool/compare_pipeline_golden_test.go @@ -0,0 +1,66 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package main + +import ( + "testing" +) + +// TestSanitizeProcessed_StripsVectorKeys locks that the golden-compare tool +// drops the SAME vector keys the production debug payload strips, via the +// shared task.IsVectorKey. The previous inline check +// (strings.HasPrefix(k,"q_") && strings.HasSuffix(k,"_vec")) only caught the +// dimension-scoped q__vec pattern and let the fixed legacy keys +// (vector/embedding/feature/q_vec) leak — causing spurious diffs against the +// expected output that already excludes them. This test pins the full set. +func TestSanitizeProcessed_StripsVectorKeys(t *testing.T) { + in := []map[string]any{ + { + "text": "hello", + "vector": []float64{0.1}, + "embedding": []float64{0.2}, + "q_1024_vec": []float64{0.3}, + "q_vec": []float64{0.4}, + "create_time": "2026", + "create_timestamp_flt": 1.0, + }, + } + + got := sanitizeProcessed(in) + if len(got) != 1 { + t.Fatalf("sanitizeProcessed returned %d chunks, want 1", len(got)) + } + cp := got[0] + + // All vector keys (fixed legacy + dimension-scoped) must be removed. + for _, k := range []string{"vector", "embedding", "q_1024_vec", "q_vec"} { + if _, ok := cp[k]; ok { + t.Errorf("vector key %q must be stripped, but present: %#v", k, cp) + } + } + // create_time / create_timestamp_flt are unrelated bookkeeping keys the + // tool always drops; assert they stay dropped. + for _, k := range []string{"create_time", "create_timestamp_flt"} { + if _, ok := cp[k]; ok { + t.Errorf("bookkeeping key %q must be stripped, but present: %#v", k, cp) + } + } + // Non-vector payload must survive. + if cp["text"] != "hello" { + t.Errorf("text=%v want hello", cp["text"]) + } +}