refactor(ingestion/task): extract index-doc mapping into task/indexdoc package (#17749)

## Summary
Extract the pipeline-output → search-engine index document mapping
helpers out of the `task` package into a dedicated, dependency-light
leaf package `internal/ingestion/task/indexdoc`.

These functions are pure transforms (they only depend on
`common`/`utility`) and are not task-orchestration concerns:
- `NormalizeChunks`, `DeepCopyChunks` (was unexported `deepCopyChunks`),
`toChunkMaps` → `indexdoc/normalize.go`
- `ProcessChunksForPipeline`, `RenameTextToContentWithWeight`,
`GetEmbeddingTokenConsumption`, `cleanupConsumedChunkFields`,
`mergeChunkMetadata`, `processChunkPositions`,
`AggregateTableDocMetadata`, `resolveTableColumnConfig` →
`indexdoc/process.go`
- `AddPositions` → `indexdoc/position.go`
- `EmbeddingTokenConsumptionKey` constant → `indexdoc/constants.go`
(task/constants.go keeps only `GRAPH_RAPTOR_FAKE_DOC_ID`)

Call sites in `pipeline_executor.go` and `golden_compare.go` now
reference the `indexdoc` package; package-task tests qualify the moved
symbols.

## Why
The `task` package had grown into a "orchestration + pure mapping +
debug" mix. Splitting the pure mapping helpers into a leaf package
sharpens package boundaries, removes a misleading top-level
`ingestion/chunk` candidate (there are already `parser/chunk` and
`service/chunk`), and lets the golden tool / future reuse pull in the
mapping logic without dragging in `task`'s `dao`/`engine`/`service`
dependency graph (Go subpackage import does not pull in the parent).

## Test plan
- `build.sh --test ./internal/ingestion/task/...` — **green** (task
4.7s, indexdoc 0.007s), matching the pre-change baseline.
- `gofmt` clean; `build.sh` builds both `ragflow-cli` and
`ragflow_server` successfully.
- Integration/E2E tiers are delegated to CI (need real MySQL/MinIO/ES
services).

Note: `pipeline_e2e_test.go` has a **pre-existing** compile error
(`server.ElasticsearchConfig` / `server.InfinityConfig` are now defined
under `internal/server/config/`, not re-exported by `internal/server`).
This is unrelated to this change — the diff to that file is only the
added `indexdoc` import and the qualified `EmbeddingTokenConsumptionKey`
reference.

internal/ingestion/task/constants.go

@@ -20,7 +20,4 @@ package task
 const (
 	// GRAPH_RAPTOR_FAKE_DOC_ID is the fake doc_id used for RAPTOR-generated chunks.
 	GRAPH_RAPTOR_FAKE_DOC_ID = "graph_raptor_fake_doc"
-
-	// EmbeddingTokenConsumptionKey is the key in pipeline output for embedding token count.
-	EmbeddingTokenConsumptionKey = "embedding_token_consumption"
 )

internal/ingestion/task/golden_compare.go

@@ -18,6 +18,8 @@ package task
 
 import (
 	"time"
+
+	indexdoc "ragflow/internal/ingestion/task/indexdoc"
 )
 
 // GoldenCompareResult is the structured output used by the local golden tools.
@@ -35,13 +37,13 @@ func ProcessPipelineOutputForGolden(
 	kbID string,
 	docName string,
 ) (GoldenCompareResult, error) {
-	normalized := NormalizeChunks(pipelineOutput)
+	normalized := indexdoc.NormalizeChunks(pipelineOutput)
 	if normalized == nil {
 		normalized = []map[string]any{}
 	}
 
-	processed := deepCopyChunks(normalized)
-	metadata, err := ProcessChunksForPipeline(processed, docID, kbID, docName, time.Now())
+	processed := indexdoc.DeepCopyChunks(normalized)
+	metadata, err := indexdoc.ProcessChunksForPipeline(processed, docID, kbID, docName, time.Now())
 	if err != nil {
 		return GoldenCompareResult{}, err
 	}

internal/ingestion/task/golden_compare_test.go

@@ -3,6 +3,8 @@ package task
 import (
 	"testing"
 	"time"
+
+	indexdoc "ragflow/internal/ingestion/task/indexdoc"
 )
 
 func TestProcessPipelineOutputForGolden_Markdown(t *testing.T) {
@@ -44,7 +46,7 @@ func TestProcessChunksForPipeline_StableFields(t *testing.T) {
 		},
 	}
 
-	meta, err := ProcessChunksForPipeline(chunks, "doc-1", "kb-1", "sample.md", now)
+	meta, err := indexdoc.ProcessChunksForPipeline(chunks, "doc-1", "kb-1", "sample.md", now)
 	if err != nil {
 		t.Fatalf("ProcessChunksForPipeline: %v", err)
 	}

internal/ingestion/task/indexdoc/constants.go

@@ -0,0 +1,20 @@
+//
+//  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 indexdoc
+
+// EmbeddingTokenConsumptionKey is the key in pipeline output for embedding token count.
+const EmbeddingTokenConsumptionKey = "embedding_token_consumption"

internal/ingestion/task/indexdoc/normalize.go

@@ -14,7 +14,7 @@
 //  limitations under the License.
 //
 
-package task
+package indexdoc
 
 import (
 	"fmt"
@@ -30,16 +30,16 @@ func NormalizeChunks(output map[string]any) []map[string]any {
 	}
 
 	if chunks, ok := output["chunks"].([]map[string]any); ok {
-		return deepCopyChunks(chunks)
+		return DeepCopyChunks(chunks)
 	}
 	if chunks, ok := toChunkMaps(output["chunks"]); ok {
-		return deepCopyChunks(chunks)
+		return DeepCopyChunks(chunks)
 	}
 	if json, ok := output["json"].([]map[string]any); ok {
-		return deepCopyChunks(json)
+		return DeepCopyChunks(json)
 	}
 	if json, ok := toChunkMaps(output["json"]); ok {
-		return deepCopyChunks(json)
+		return DeepCopyChunks(json)
 	}
 	if md, ok := output["markdown"].(string); ok && md != "" {
 		return []map[string]any{{"text": md}}
@@ -69,10 +69,16 @@ func toChunkMaps(v any) ([]map[string]any, bool) {
 	return out, true
 }
 
-// deepCopyChunks returns a deep copy of the chunk slice and each chunk map.
-// Slice values (e.g. []float64 vectors) are fully copied, not shared.
-// Mirrors Python: copy.deepcopy()
-func deepCopyChunks(chunks []map[string]any) []map[string]any {
+// DeepCopyChunks returns a copy of the chunk slice and each chunk map.
+// The chunk maps themselves are freshly allocated, and the value types that
+// the pipeline actually emits — []float64 (vectors), []int, and []string — are
+// element-wise copied so callers cannot mutate the originals through them.
+// Other value types (nested maps, [][]float64 positions, etc.) are shared by
+// reference, not recursively deep-copied; positions are later flattened and
+// copied independently by processChunkPositions. It is therefore NOT a full
+// recursive deep copy (unlike Python's copy.deepcopy), only the copy needed
+// for the post-processing pass over pipeline output.
+func DeepCopyChunks(chunks []map[string]any) []map[string]any {
 	if chunks == nil {
 		return nil
 	}

internal/ingestion/task/indexdoc/normalize_test.go

@@ -1,4 +1,4 @@
-package task
+package indexdoc
 
 import (
 	"testing"

internal/ingestion/task/indexdoc/position.go

@@ -14,7 +14,7 @@
 //  limitations under the License.
 //
 
-package task
+package indexdoc
 
 // AddPositions adds position fields to a chunk map.
 // Input positions is a flat []float64 grouped as [pn, left, right, top, bottom]

internal/ingestion/task/indexdoc/position_test.go

@@ -1,4 +1,4 @@
-package task
+package indexdoc
 
 import (
 	"testing"

internal/ingestion/task/indexdoc/process.go

@@ -14,7 +14,7 @@
 //  limitations under the License.
 //
 
-package task
+package indexdoc
 
 import (
 	"fmt"

internal/ingestion/task/indexdoc/process_test.go

@@ -1,4 +1,4 @@
-package task
+package indexdoc
 
 import (
 	"testing"

internal/ingestion/task/pipeline_e2e_test.go

@@ -30,6 +30,7 @@ import (
 	"ragflow/internal/engine"
 	"ragflow/internal/engine/elasticsearch"
 	"ragflow/internal/engine/infinity"
+	indexdoc "ragflow/internal/ingestion/task/indexdoc"
 	"ragflow/internal/ingestion/testutil"
 	"ragflow/internal/server"
 	"ragflow/internal/service"
@@ -247,7 +248,7 @@ func TestPipelineE2E_PipelineExecutor(t *testing.T) {
 							"q_2_vec": []float64{0.3, 0.4}, // Pre-vectorized to skip embedding
 						},
 					},
-					EmbeddingTokenConsumptionKey: 100,
+					indexdoc.EmbeddingTokenConsumptionKey: 100,
 				}, dsl, nil
 			})
 

internal/ingestion/task/pipeline_executor.go

@@ -31,6 +31,7 @@ import (
 	"ragflow/internal/ingestion/component"
 	"ragflow/internal/ingestion/knowledge_compile"
 	pipelinepkg "ragflow/internal/ingestion/pipeline"
+	indexdoc "ragflow/internal/ingestion/task/indexdoc"
 
 	"gorm.io/gorm"
 )
@@ -198,13 +199,13 @@ func (s *PipelineExecutor) Execute(ctx context.Context) (*PipelineResult, error)
 // performs no DB/index writes — the embedding vectors already computed by the
 // pipeline run are left on the chunks. This keeps debug runs side-effect free.
 func (s *PipelineExecutor) collectDebugOutput(ctx context.Context, pipelineOutput map[string]any, start time.Time) (*PipelineResult, error) {
-	chunks := NormalizeChunks(pipelineOutput)
+	chunks := indexdoc.NormalizeChunks(pipelineOutput)
 	return &PipelineResult{
 		DocID:            s.taskCtx.Doc.ID,
 		KbID:             s.taskCtx.Doc.KbID,
 		Chunks:           chunks,
 		ChunkCount:       countDistinctChunkIDs(chunks),
-		TokenConsumption: GetEmbeddingTokenConsumption(pipelineOutput),
+		TokenConsumption: indexdoc.GetEmbeddingTokenConsumption(pipelineOutput),
 		Duration:         time.Since(start).Seconds(),
 	}, nil
 }
@@ -217,13 +218,13 @@ func (s *PipelineExecutor) processOutput(ctx context.Context, pipelineOutput map
 		return nil, err
 	}
 
-	chunks := NormalizeChunks(pipelineOutput)
+	chunks := indexdoc.NormalizeChunks(pipelineOutput)
 	if len(chunks) == 0 {
 		return nil, nil
 	}
 
-	embeddingTokenConsumption := GetEmbeddingTokenConsumption(pipelineOutput)
-	metadata, err := ProcessChunksForPipeline(
+	embeddingTokenConsumption := indexdoc.GetEmbeddingTokenConsumption(pipelineOutput)
+	metadata, err := indexdoc.ProcessChunksForPipeline(
 		chunks,
 		s.taskCtx.Doc.ID,
 		s.taskCtx.Doc.KbID,
@@ -234,7 +235,7 @@ func (s *PipelineExecutor) processOutput(ctx context.Context, pipelineOutput map
 		return nil, err
 	}
 
-	tableMeta := AggregateTableDocMetadata(chunks, map[string]interface{}(s.taskCtx.Doc.ParserConfig))
+	tableMeta := indexdoc.AggregateTableDocMetadata(chunks, map[string]interface{}(s.taskCtx.Doc.ParserConfig))
 	if tableMeta != nil {
 		if metadata == nil {
 			metadata = make(map[string]any)

internal/ingestion/task/pipeline_executor_test.go

@@ -14,6 +14,7 @@ import (
 	"ragflow/internal/dao"
 	"ragflow/internal/entity"
 	pipelinepkg "ragflow/internal/ingestion/pipeline"
+	indexdoc "ragflow/internal/ingestion/task/indexdoc"
 )
 
 // =============================================================================
@@ -201,7 +202,7 @@ func TestKB_Doc_Tenant_Accessors(t *testing.T) {
 func TestPipelineExecutor_ProcessChunks_WrapsProcessChunksForPipeline(t *testing.T) {
 	svc := mustNewPipelineExecutor(t, makeTaskCtx(), "flow-1", 0)
 	chunks := []map[string]any{{"text": "hello world"}}
-	meta, err := ProcessChunksForPipeline(chunks, svc.taskCtx.Doc.ID, svc.taskCtx.Doc.KbID, *svc.taskCtx.Doc.Name, time.Now())
+	meta, err := indexdoc.ProcessChunksForPipeline(chunks, svc.taskCtx.Doc.ID, svc.taskCtx.Doc.KbID, *svc.taskCtx.Doc.Name, time.Now())
 	if err != nil {
 		t.Fatalf("ProcessChunksForPipeline: %v", err)
 	}