fix: Go ingestion migration batch 5 (Parser 1.1/1.7/2.11, Chunker 1.7/1.8/2.6/2.7, Tokenizer 6x fixes) (#17419)

## Summary

Continuation of the Python→Go ingestion pipeline migration (File →
Parser → Chunker → Extractor → Tokenizer). Fixes cover Parser, Chunker,
and Tokenizer gaps identified. Fix page number (0-indexed and 1-index
mixed before fix; use 1-indexed after fix) and chunk order issues.

### Parser
- **Slides TCADP (1.7):** `pptx_tcadp.go` + TCADP branch in
`pptx_parser.go`/`ppt_parser.go` — PowerPoint files now support
`parse_method="tcadp"` via the TCADP cloud service, matching the
spreadsheet-family TCADP pattern. PPT containers pass `"PPT"` as
fileType (not hardcoded `"PPTX"`).
- **Audio default output_format (2.11):** `defaultSetups()` audio
default changed from `"text"` to `"json"`, aligning with Python
`parser.py:232` and `AllowedOutputFormat["audio"]={"json"}`.
- **PDF VLM enhancement (1.1):** `maybeDispatchPDFVisionEnhancement` in
`pdf_vision_dispatch.go` enriches image/table items with IMAGE2TEXT
model descriptions after PDF parsing, mirroring Python
`enhance_media_sections_with_vision`. Semaphore fix: acquire before
goroutine start to prevent unbounded goroutine creation.
- **json family (2.3):** reclassified as Keep Go — `json_parser.go` is a
functional enhancement, not a parity gap.
- **page number:** changed from "mixed use of 1-indexed & 0-indexed" to
"1-indexed"

### Chunker
- **BULLET_PATTERN fallback (1.7):** 4th-level fallback in
`resolveTitleLevels` (`title.go`) detects bullet/numbered-list patterns
(Chinese legal, numbering, English) when outline + regex levels produce
only bodyLevel. Guarded by `allBodyLevel` to never override existing
structure.
- **Tag/One chunker fields (1.8):** `tag.go` sets `TopInt` from source
row index; `one.go` preserves `Positions`/`PDFPositions` from source
items. TSV multi-line RowNum fix: tracks `contentStart` for correct row
attribution.
- **Overlapped_percent normalization (2.6):**
`NormalizeOverlappedPercent` in `schema/chunker.go` mirrors Python
`common/float_utils.py:50-58` — accepts `[0,1)` fraction or `[0,90]`
percent, normalizes to canonical `[0,90]`.
- **Paragraph splitting (2.7):** aligned to Python flow `naive_merge` —
`CRLF` normalization, `splitKeepingDelimiter` preserves sentence
delimiters, single-section merge with token-budget-governed chunking.
- **chunk order:** sort by reading order

### Tokenizer
- **Phantom chunk filtering (Omission 2):** `isPhantomChunk` + filter
loop in `chunksFromTokenizerUpstream` skips zero-value ChunkDocs.
- **Batch size env var (Omission 3):** `embeddingBatchSize()` reads
`TOKENIZER_EMBEDDING_BATCH_SIZE`, defaults to 16.
- **Summary empty check (Diff 5):** `TrimSpace(s) != ""` → `s != ""`,
matching Python truthy check.
- **chunk_order_int all paths (Diff 8):** set unconditionally before
full_text/embedding branching.
- **Timeout default (Diff 10):** `600s` → `60s`, matching Python
`@timeout(60)`.
- **Small maxTokens truncation (Diff 14):** `truncateForEmbedding`
returns `""` when `maxTokens <= 10`, matching Python.

### Code review fixes
- Semaphore acquire moved before goroutine in `pdf_vision_dispatch.go`
(concurrency control)
- Context propagation in `pptx_tcadp.go` (cancellation support)
- Test resolver leak fix in `media_dispatch_test.go` (defer restore)
- Migration history comments removed per AGENTS.md

## Test plan
```
bash build.sh --test ./internal/parser/parser/... ./internal/ingestion/component/...
```

## Notes
- Migration diff tracking: `docs/migration_python_go_diff.md`
- Remaining gaps: Extractor component only (21 items)

Author: xugangqiang

internal/ingestion/component/chunker/crop_title_test.go

@@ -0,0 +1,65 @@
+//go:build cgo
+
+//  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.
+//
+//      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 chunker
+
+import (
+	"context"
+	"strings"
+	"testing"
+)
+
+// TestCropTitleChunks_CropsPositionedChunk pins residual A: chunks
+// produced by the Title/Group/Hierarchy chunkers must be cropped on demand
+// (mirroring the TokenChunker JSON path). The chunk shape mirrors the real
+// Group/Hierarchy output: it carries doc_type_kwd + positions but NO ck_type
+// (buildChunksFromRecordGroups never sets ck_type). cropTitleChunks must
+// derive ck_type from doc_type_kwd so needsCrop (pdfcrop_cgo.go:151) fires.
+// A text chunk with positions gets a rendered preview; a chunk without
+// positions is left untouched; and the derived ck_type must not leak into
+// the returned chunk shape.
+func TestCropTitleChunks_CropsPositionedChunk(t *testing.T) {
+	ctx := context.Background()
+	pos := jsonPositions(t, []float64{1, 10, 100, 10, 100})
+	in := []map[string]any{
+		{"text": "page text", "doc_type_kwd": "text", "positions": pos},
+		{"text": "no coords", "doc_type_kwd": "text"},
+	}
+
+	got := cropTitleChunks(ctx, mockCropEngine{}, in)
+	if len(got) != 2 {
+		t.Fatalf("len = %d, want 2", len(got))
+	}
+	img, _ := got[0]["image"].(string)
+	if !strings.HasPrefix(img, "data:image/png;base64,") {
+		t.Errorf("positioned text chunk: image = %q, want data:image/png;base64, prefix", img)
+	}
+	if got[1]["image"] != nil {
+		t.Errorf("chunk without positions should not be cropped, got %v", got[1]["image"])
+	}
+	if _, ok := got[0]["ck_type"]; ok {
+		t.Errorf("derived ck_type must not leak into output chunk: %v", got[0])
+	}
+}
+
+// TestCropTitleChunks_NilEnginePassthrough asserts the helper is a no-op
+// when no PDF engine is available (best-effort contract).
+func TestCropTitleChunks_NilEnginePassthrough(t *testing.T) {
+	in := []map[string]any{{"text": "hello"}}
+	got := cropTitleChunks(context.Background(), nil, in)
+	if len(got) != 1 || got[0]["text"] != "hello" {
+		t.Fatalf("nil engine should pass through unchanged: %v", got)
+	}
+}

internal/ingestion/component/chunker/group.go

@@ -36,6 +36,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"log/slog"
 	"regexp"
 	"sort"
 	"strings"
@@ -108,7 +109,7 @@ func buildSectionIDs(levels []int, targetLevel int) []int {
 // supplied inputs. Detected headings + adjacent merges happen in two
 // goroutines (heading detection sequential, then a fan-out over
 // record-buckets for the merge pass).
-func invokeGroup(_ context.Context, inputs map[string]any, p *titleChunkerParam) (map[string]any, error) {
+func invokeGroup(parentCtx context.Context, db *gorm.DB, inputs map[string]any, p *titleChunkerParam) (map[string]any, error) {
 	records := extractLineRecords(inputs)
 	common.Debug("chunker stage",
 		zap.String("component", "Chunker"),
@@ -161,6 +162,21 @@ func invokeGroup(_ context.Context, inputs map[string]any, p *titleChunkerParam)
 		zap.Int("chunks", len(chunks)),
 		zap.Bool("plain_text", isPlainTextFormat(inputs)),
 	)
+
+	// On-demand PDF preview cropping for image/table/text chunks,
+	// mirroring the TokenChunker JSON path (token.go:513). Best-effort:
+	// a missing or unreadable PDF simply skips cropping.
+	if upstream, uErr := decodeChunkerFromUpstream(inputs); uErr == nil {
+		engine, eErr := newPDFEngineFromUpstream(parentCtx, db, upstream)
+		if eErr != nil {
+			slog.Warn("GroupTitleChunker: could not open PDF for on-demand cropping", "err", eErr)
+		}
+		if engine != nil {
+			defer engine.Close()
+			chunks = cropTitleChunks(parentCtx, engine, chunks)
+		}
+	}
+
 	if len(chunks) == 0 {
 		return emptyOutputs(), nil
 	}
@@ -484,7 +500,7 @@ func (c *GroupTitleChunkerComponent) Invoke(ctx context.Context, db *gorm.DB, in
 			"_ERROR":        "GroupTitleChunker: missing required upstream field \"name\"",
 		}, nil
 	}
-	return invokeGroup(ctx, withName(inputs, name), &c.param)
+	return invokeGroup(ctx, db, withName(inputs, name), &c.param)
 }
 
 // init registers GroupTitleChunker under CategoryIngestion.

internal/ingestion/component/chunker/hierarchy.go

@@ -41,6 +41,7 @@ package chunker
 import (
 	"context"
 	"fmt"
+	"log/slog"
 
 	"go.uber.org/zap"
 	"gorm.io/gorm"
@@ -139,7 +140,7 @@ func (n *chunkNode) getPaths(paths *[][]int, titles []int, depth int, includeHea
 }
 
 // invokeHierarchy runs the HierarchyTitleChunker strategy.
-func invokeHierarchy(_ context.Context, inputs map[string]any, p *titleChunkerParam) (map[string]any, error) {
+func invokeHierarchy(parentCtx context.Context, db *gorm.DB, inputs map[string]any, p *titleChunkerParam) (map[string]any, error) {
 	records := extractLineRecords(inputs)
 	common.Debug("chunker stage",
 		zap.String("component", "Chunker"),
@@ -246,6 +247,21 @@ func invokeHierarchy(_ context.Context, inputs map[string]any, p *titleChunkerPa
 		zap.Int("chunks", len(chunks)),
 		zap.Bool("plain_text", isPlainTextFormat(inputs)),
 	)
+
+	// On-demand PDF preview cropping for image/table/text chunks,
+	// mirroring the TokenChunker JSON path (token.go:513). Best-effort:
+	// a missing or unreadable PDF simply skips cropping.
+	if upstream, uErr := decodeChunkerFromUpstream(inputs); uErr == nil {
+		engine, eErr := newPDFEngineFromUpstream(parentCtx, db, upstream)
+		if eErr != nil {
+			slog.Warn("HierarchyTitleChunker: could not open PDF for on-demand cropping", "err", eErr)
+		}
+		if engine != nil {
+			defer engine.Close()
+			chunks = cropTitleChunks(parentCtx, engine, chunks)
+		}
+	}
+
 	if len(chunks) == 0 {
 		return emptyOutputs(), nil
 	}
@@ -302,7 +318,7 @@ func (c *HierarchyTitleChunkerComponent) Invoke(ctx context.Context, db *gorm.DB
 			"_ERROR":        "HierarchyTitleChunker: missing required upstream field \"name\"",
 		}, nil
 	}
-	return invokeHierarchy(ctx, withName(inputs, name), &c.param)
+	return invokeHierarchy(ctx, db, withName(inputs, name), &c.param)
 }
 
 // init registers HierarchyTitleChunker under CategoryIngestion.

internal/ingestion/component/chunker/one.go

@@ -141,10 +141,12 @@ func emitOneFromItems(items, chunks []schema.ChunkDoc) map[string]any {
 			return emptyOutputs()
 		}
 		out := schema.ChunkDoc{
-			Text:    text,
-			DocType: docType,
-			CKType:  docType,
-			Image:   it.Image,
+			Text:         text,
+			DocType:      docType,
+			CKType:       docType,
+			Image:        it.Image,
+			Positions:    it.Positions,
+			PDFPositions: it.PDFPositions,
 		}
 		return chunkOutputs([]schema.ChunkDoc{out})
 	}
@@ -164,6 +166,11 @@ func emitOneFromItems(items, chunks []schema.ChunkDoc) map[string]any {
 		return emptyOutputs()
 	}
 	out := schema.ChunkDoc{Text: merged, DocType: "text", CKType: "text"}
+	// Multi-item merge produces a single text-only chunk mirroring Python
+	// one.py:166-168. Per-item Positions/PDFPositions are intentionally
+	// not carried — merging coordinates from different source items would
+	// produce meaningless composite geometry, and the downstream
+	// processChunkPositions would map them to incorrect pages.
 	if img != "" {
 		out.Image = img
 	}

internal/ingestion/component/chunker/one_test.go

@@ -98,3 +98,46 @@ func TestOneChunker_JSONMultipleMerges(t *testing.T) {
 		t.Errorf("image = %q, want first available media context", got)
 	}
 }
+
+// TestOneChunker_PreservesPositions verifies that when a single upstream
+// item carries PDF coordinates, the OneChunker preserves both Positions
+// and PDFPositions (with their coordinate values) on the output chunk.
+func TestOneChunker_PreservesPositions(t *testing.T) {
+	chunks := oneChunksOf(t, map[string]any{
+		"name":          "page.pdf",
+		"output_format": "json",
+		"json": []map[string]any{
+			{
+				"text":           "page text",
+				"positions":      []any{[]any{10.0, 20.0, 30.0, 40.0}},
+				"_pdf_positions": []any{[]any{1.0, 2.0, 3.0, 4.0, 5.0}},
+			},
+		},
+	})
+	if len(chunks) != 1 {
+		t.Fatalf("want 1 chunk, got %d", len(chunks))
+	}
+	assertCoordTuple(t, "positions", chunks[0]["positions"], []float64{10.0, 20.0, 30.0, 40.0})
+	assertCoordTuple(t, "_pdf_positions", chunks[0]["_pdf_positions"], []float64{1.0, 2.0, 3.0, 4.0, 5.0})
+}
+
+// assertCoordTuple verifies a positions/_pdf_positions field round-tripped
+// as a [][]float64 with the expected single-row coordinate tuple.
+func assertCoordTuple(t *testing.T, key string, got any, want []float64) {
+	t.Helper()
+	rows, ok := got.([][]float64)
+	if !ok {
+		t.Fatalf("%s = %v, want [][]float64 (got %T)", key, got, got)
+	}
+	if len(rows) != 1 {
+		t.Fatalf("%s has %d rows, want 1", key, len(rows))
+	}
+	if len(rows[0]) != len(want) {
+		t.Fatalf("%s[0] = %v, want %v (len %d vs %d)", key, rows[0], want, len(rows[0]), len(want))
+	}
+	for i, w := range want {
+		if rows[0][i] != w {
+			t.Errorf("%s[0][%d] = %v, want %v", key, i, rows[0][i], w)
+		}
+	}
+}

internal/ingestion/component/chunker/qa.go

@@ -101,7 +101,7 @@ func (c *QAChunkerComponent) invoke(_ context.Context, inputs map[string]any) (m
 
 	qPrefix, aPrefix := "问题：", "回答："
 	// Python qa.py defaults to Chinese when no language is supplied; only
-	// an explicit "english" switches to English prefixes (diff Chunker-2.13).
+	// an explicit "english" switches to English prefixes
 	eng := strings.EqualFold(c.param.Lang, "english")
 	if eng {
 		qPrefix, aPrefix = "Question: ", "Answer: "
@@ -137,7 +137,7 @@ func (c *QAChunkerComponent) invoke(_ context.Context, inputs map[string]any) (m
 			ContentLtks:       contentLTKS,
 			ContentSmLtks:     contentSMLTKS,
 		}
-		// Restore metadata lost before diff Chunker-1.8: top_int (row
+		//
 		// index), image id + coordinates carried from the source item.
 		if pair.RowNum >= 0 {
 			chunk.TopInt = []int{pair.RowNum}
@@ -172,14 +172,14 @@ type qaPair struct {
 	RowNum int
 	// Image and positions are carried from the upstream item so the QA
 	// chunk preserves metadata that Python sets via beAdocPdf/beAdocDocx
-	// (diff Chunker-1.8).
+	//
 	Image        string
 	PDFPositions json.RawMessage
 	Positions    json.RawMessage
 }
 
 // rmQAPrefixRe mirrors Python qa.py:241 `[\t:： ]+` — one-or-more separator
-// chars, so "Q:: answer" is fully stripped (diff Chunker-2.12).
+// chars, so "Q:: answer" is fully stripped
 var rmQAPrefixRe = regexp.MustCompile(`(?i)^(问题|答案|回答|user|assistant|Q|A|Question|Answer|问|答)[\t:： ]+`)
 
 func rmQAPrefix(txt string) string {
@@ -424,7 +424,7 @@ func extractQAJSON(items []schema.ChunkDoc) []qaPair {
 		}
 		tmp := extractQAText(txt)
 		// Preserve the source item's image id and coordinates on each
-		// extracted pair (diff Chunker-1.8).
+		// extracted pair
 		for _, p := range tmp {
 			p.Image = item.Image
 			p.PDFPositions = item.PDFPositions

internal/ingestion/component/chunker/qa_batch2_test.go

@@ -41,7 +41,7 @@ func qaInvoke(t *testing.T, inputs map[string]any) []map[string]any {
 	return chunks
 }
 
-// TestQAChunker_DefaultLangIsChinese exercises migration diff Chunker-2.13:
+// TestQAChunker_DefaultLangIsChinese exercises migration :
 // when no language is supplied, Python defaults to Chinese prefixes
 // ("问题："/"回答："); the legacy Go code defaulted to English.
 func TestQAChunker_DefaultLangIsChinese(t *testing.T) {
@@ -79,7 +79,7 @@ func TestRmQAPrefixStripsMultipleSeparators(t *testing.T) {
 	}
 }
 
-// TestQAChunker_SetsTopInt exercises migration diff Chunker-1.8 (top_int):
+// TestQAChunker_SetsTopInt exercises migration  (top_int):
 // each QA chunk must carry the source row index in `top_int`, matching
 // Python beAdoc(..., row_num=i).
 func TestQAChunker_SetsTopInt(t *testing.T) {

internal/ingestion/component/chunker/qa_test.go

@@ -213,7 +213,7 @@ func TestQAChunker_PrefixSpaceSeparatorStrips(t *testing.T) {
 	}
 	cww, _ := chunks[0]["content_with_weight"].(string)
 	// Python qa.py:241 uses `[\t:： ]+`, so a space is a valid separator:
-	// a leading "A"/"Q" followed by a space is stripped (diff Chunker-2.12).
+	// a leading "A"/"Q" followed by a space is stripped. .
 	if cww != "Question: language model is useful\tAnswer: How does it work" {
 		t.Fatalf("space-separator prefix not stripped: %q", cww)
 	}

internal/ingestion/component/chunker/tag.go

@@ -125,6 +125,7 @@ func (c *TagChunkerComponent) invoke(_ context.Context, inputs map[string]any) (
 			ContentLtks:       contentLTKS,
 			ContentSmLtks:     contentSMLTKS,
 			TagKwd:            splitTagKwd(pair.Tags),
+			TopInt:            []int{pair.RowNum},
 		}
 		chunks = append(chunks, chunk)
 	}
@@ -133,9 +134,13 @@ func (c *TagChunkerComponent) invoke(_ context.Context, inputs map[string]any) (
 }
 
 // tagPair is a (content, tags) row extracted from the upstream payload.
+// RowNum is the 0-based record START line (the first physical source
+// line of the record), used by every extractor consistently and mapped
+// to Python's top_int in beAdoc(tag.py:33).
 type tagPair struct {
 	Content string
 	Tags    string
+	RowNum  int
 }
 
 // extractTagText ports tag.py:60-89 (txt) and tag.py:91-113 (csv).
@@ -157,18 +162,23 @@ func extractTagText(text string) []tagPair {
 func extractTagTextTab(lines []string) []tagPair {
 	var pairs []tagPair
 	content := ""
-	for _, line := range lines {
+	contentStart := -1
+	for i, line := range lines {
 		if strings.TrimSpace(line) == "" {
 			continue
 		}
+		if contentStart < 0 {
+			contentStart = i
+		}
 		parts := strings.Split(line, "\t")
 		if len(parts) != 2 {
 			content += "\n" + line
 			continue
 		}
 		content += "\n" + parts[0]
-		pairs = append(pairs, tagPair{Content: content, Tags: parts[1]})
+		pairs = append(pairs, tagPair{Content: content, Tags: parts[1], RowNum: contentStart})
 		content = ""
+		contentStart = -1
 	}
 	return pairs
 }
@@ -202,6 +212,7 @@ func extractTagTextCSV(text string, lines []string) []tagPair {
 		for curLine < len(lineStarts) && lineStarts[curLine] < endOff {
 			curLine++
 		}
+		startLine := prevLine
 		raw := strings.Join(lines[prevLine:curLine], "\n")
 		prevLine = curLine
 
@@ -210,7 +221,10 @@ func extractTagTextCSV(text string, lines []string) []tagPair {
 			continue
 		}
 		content += "\n" + record[0]
-		pairs = append(pairs, tagPair{Content: content, Tags: record[1]})
+		// RowNum is the 0-based record START line, kept consistent with
+		// extractTagTextTab (line i) and extractTagTable (<tr> i) so every
+		// tag-pair source uses the same row-index convention.
+		pairs = append(pairs, tagPair{Content: content, Tags: record[1], RowNum: startLine})
 		content = ""
 	}
 	return pairs
@@ -226,7 +240,7 @@ func extractTagTable(htmlStr string) []tagPair {
 	}
 	rows := htmlTR.FindAllStringSubmatch(htmlStr, -1)
 	pairs := make([]tagPair, 0, len(rows))
-	for _, row := range rows {
+	for i, row := range rows {
 		cells := htmlTD.FindAllStringSubmatch(row[1], -1)
 		var texts []string
 		for _, cell := range cells {
@@ -237,7 +251,7 @@ func extractTagTable(htmlStr string) []tagPair {
 			}
 		}
 		if len(texts) >= 2 {
-			pairs = append(pairs, tagPair{Content: texts[0], Tags: texts[1]})
+			pairs = append(pairs, tagPair{Content: texts[0], Tags: texts[1], RowNum: i})
 		}
 	}
 	return pairs