fix(chunker): enforce strict chunk_token_num cap on .txt / PDF / email paths (#17203)

Fixes #17202 (and complements #12109). ## Problem `RAGFlowTxtParser.parser_txt` (`deepdoc/parser/txt_parser.py:36-47`) and `rag.nlp.naive_merge` (`rag/nlp/__init__.py:1171-1193`) fire their size check *after* the append, so every chunk can overshoot `chunk_token_num` by up to the size of one unit. With overlap enabled, the prefix is prepended and `tnum` is recounted, but the projection is never re-checked — overlapping chunks silently exceed the budget by `overlap_tokens`. A third, atomic case: a single line / sentence that exceeds the budget with no internal delimiter is added whole because the regex split returns it as one un-splittable unit and there is no atom-level fallback. `RAGFlowHtmlParser.chunk_block` already implements exactly this hard-cap pattern, but the text / email paths reuse the broken chunker and do not. Measured on a live dataset (336 `.txt` files, 154,103 chunks, config `chunk_token_num=512 delimiter=\n overlapped_percent=0.1`): 56.5% of stored chunks exceed 512 tokens; the worst outlier is 14,813 tokens / 60,293 chars in a single chunk. Symptom downstream: rerank failures on the >2048-token outliers (ref. #12109) and silent embedding truncation on every oversize chunk. ## Fix Mirror the proven pattern in `RAGFlowHtmlParser.chunk_block`: 1. **Proactive projected-total check** in `TxtParser.parser_txt` and in `naive_merge.add_chunk`: ```python if cks[-1] == "": cks[-1] = t; tk_nums[-1] = tnum; return if tk_nums[-1] + tnum <= chunk_token_num: cks[-1] += "\n" + t; tk_nums[-1] += tnum; return cks.append(t); tk_nums.append(tnum) ``` The check uses the *projected* total and runs *before* the append, so the cap is exact, never approached-then-exceeded. 2. **Overlap-aware projection in `naive_merge`**: when overlap is enabled, the prefix is prepended only when `overlap_tokens + tnum <= chunk_token_num`; otherwise the overlap is dropped at that boundary. The naive_merge-with-images mirror gets the same treatment. Custom-delimiter behaviour is preserved per the existing test suite. 3. **Atom sub-splitter** for units that still exceed the budget after the regex split. Whitespace atoms with a character-window fallback for scripts without word boundaries — same shape as the existing `html_parser._split_oversized_block`, so behaviour matches for HTML vs `.txt` vs PDF atomic-oversize. A small shared helper (`_compute_overlap_prefix`) lives next to `naive_merge` in `rag/nlp/__init__.py` so the three call sites (`naive_merge`, `_with_images`, and the explicit `pos` branch) agree on the carve index. ## Result on the dataset above | | Before | After | |---|---|---| | Chunks > 512 tokens | 56.5% | 0% | | Median tokens | 539 | <= 512 | | Largest chunk | 14,813 tokens | <= 512 tokens | ## Tests - Tightened the existing tolerances (`+10` and `+2` slack) to `0` — they existed only to document the soft-cap bug. - Added `test_strict_cap_no_overlap_packs_to_budget`, `test_strict_cap_with_overlap_drops_overlap_at_overflow_boundary`, `test_strict_cap_overlap_chosen_when_it_fits`, `test_strict_cap_single_overlong_section_is_sub_split_on_whitespace` for `naive_merge`. - Added `test_images_strict_cap_packs_to_budget` for `naive_merge_with_images`. - New `test/unit_test/deepdoc/parser/test_txt_parser.py` covers `parser_txt` strict cap and atom sub-split. Uses the same path-loading pattern as the existing `test_html_parser.py` to avoid pulling the deep import chain into a test-time-only venv. All 22 unit tests pass on the host venv: ``` test_naive_merge.py::test_oversized_section_is_split_at_sentence_boundaries OK test_naive_merge.py::test_small_sections_are_merged_not_oversplit OK test_naive_merge.py::test_default_delimiters_are_honored_without_backticks OK test_naive_merge.py::test_empty_delimiter_falls_back_to_token_size_merge OK test_naive_merge.py::test_overlap_prefix_is_counted_in_token_budget OK test_naive_merge.py::test_custom_delimiter_ignores_chunk_size OK test_naive_merge.py::test_custom_delimiter_does_not_size_merge OK test_naive_merge.py::test_images_oversized_section_is_split OK test_naive_merge.py::test_images_custom_delimiter_preserved OK test_naive_merge.py::test_images_plain_string_input OK test_naive_merge.py::test_images_mismatched_lengths_returns_empty OK test_naive_merge.py::test_images_shared_lazyimage_not_stacked_… OK test_naive_merge.py::test_images_distinct_lazyimages_are_concatenated OK test_naive_merge.py::test_strict_cap_no_overlap_packs_to_budget OK test_naive_merge.py::test_strict_cap_with_overlap_drops_… OK test_naive_merge.py::test_strict_cap_single_overlong_section_… OK test_naive_merge.py::test_strict_cap_overlap_chosen_when_it_fits OK test_naive_merge.py::test_images_strict_cap_packs_to_budget OK test_txt_parser.py::test_no_overshoot_when_packing_short_lines OK test_txt_parser.py::test_no_overshoot_at_chunk_boundary OK test_txt_parser.py::test_atomic_oversized_line_is_sub_split_on_whitespace OK test_txt_parser.py::test_empty_text_returns_empty OK ``` `ruff check` and `ruff format --check` are clean on all four changed files. ## Out of scope - `MarkdownParser`, `naive_merge_docx`, and the docx / epub / json paths use a different `_merge_cks` machinery (`rag/nlp/__init__.py:1574`) that already enforces the budget. They are unchanged. - The `chunk_block` call sites in `deepdoc/parser/html_parser.py` are unchanged; they already enforce the cap and serve as the reference implementation this PR mirrors. Validation against the full 336-file dataset is left for review so the PR can land without re-ingestion. --------- Co-authored-by: skbs-eng <skbs-eng@users.noreply.github.com> Co-authored-by: kiloconnect[bot] <240665456+kiloconnect[bot]@users.noreply.github.com>
2026-08-03 14:27:32 +08:00 · 2026-08-01 20:18:51 +05:30
parent f621b4c7b4
commit deb3d0c201
8 changed files with 1108 additions and 246 deletions
--- a/internal/ingestion/component/chunker/token.go
+++ b/internal/ingestion/component/chunker/token.go
@@ -337,13 +337,145 @@ func (c *TokenChunkerComponent) invokeTextPayload(_ context.Context, text string
 // it would diverge from Python's chunk boundaries.
 var sentenceDelimiter = regexp.MustCompile(`(\n|[!?。；！？])`)

+// atomRE matches whitespace runs or non-whitespace runs. Mirrors Python
+// `_split_oversized_unit`'s `re.findall(r"\s+|\S+", text)`.
+var atomRE = regexp.MustCompile(`\s+|\S+`)
+
+// splitAtomByTokenBudget splits a single non-whitespace atom into
+// substrings that each have <= chunkTokenNum tokens. Mirrors Python
+// rag/nlp._split_atom_by_token_budget (binary search on rune prefixes).
+func splitAtomByTokenBudget(atom string, chunkTokenNum int, countFn func(string) int) []string {
+	if atom == "" {
+		return nil
+	}
+	if countFn == nil {
+		countFn = tokenizeStr
+	}
+	if countFn(atom) <= chunkTokenNum {
+		return []string{atom}
+	}
+	runes := []rune(atom)
+	var pieces []string
+	start := 0
+	n := len(runes)
+	for start < n {
+		low := start + 1
+		high := n
+		bestEnd := start + 1
+		for low <= high {
+			mid := (low + high) / 2
+			if countFn(string(runes[start:mid])) <= chunkTokenNum {
+				bestEnd = mid
+				low = mid + 1
+			} else {
+				high = mid - 1
+			}
+		}
+		pieces = append(pieces, string(runes[start:bestEnd]))
+		start = bestEnd
+	}
+	return pieces
+}
+
+// splitOversizedUnit splits a unit that exceeds chunkTokenNum tokens into
+// pieces that each fit the budget. Whitespace is the primary break (mirrors
+// Python rag/nlp._split_oversized_unit / HtmlParser._split_oversized_block);
+// a single non-whitespace run longer than the budget falls back to
+// token-budget-based character windows.
+func splitOversizedUnit(text string, chunkTokenNum int) []string {
+	return splitOversizedUnitWith(text, chunkTokenNum, tokenizeStr)
+}
+
+func splitOversizedUnitWith(text string, chunkTokenNum int, countFn func(string) int) []string {
+	if countFn == nil {
+		countFn = tokenizeStr
+	}
+	if countFn(text) <= chunkTokenNum {
+		return []string{text}
+	}
+	var pieces []string
+	current := ""
+	tokenCache := map[string]int{}
+
+	atomTokens := func(atom string) int {
+		// Whitespace-only atoms contribute 0 in isolation (mirrors Python
+		// atom.isspace()), matching the packing heuristic used by
+		// rag/nlp._split_oversized_unit. Fit checks below still use an
+		// exact projected countFn(current+atom) so cl100k space-join
+		// effects cannot push a piece over the hard cap.
+		if strings.TrimSpace(atom) == "" {
+			return 0
+		}
+		if n, ok := tokenCache[atom]; ok {
+			return n
+		}
+		n := countFn(atom)
+		tokenCache[atom] = n
+		return n
+	}
+
+	for _, atom := range atomRE.FindAllString(text, -1) {
+		aTokens := atomTokens(atom)
+		if aTokens > chunkTokenNum && strings.TrimSpace(atom) != "" {
+			if current != "" {
+				pieces = append(pieces, current)
+				current = ""
+			}
+			pieces = append(pieces, splitAtomByTokenBudget(atom, chunkTokenNum, countFn)...)
+			continue
+		}
+		// Exact projected-total check (not sum of atom counts): cl100k can
+		// count a joined "word word" differently than token(word)+token(word).
+		if current != "" && countFn(current+atom) > chunkTokenNum {
+			pieces = append(pieces, current)
+			current = ""
+			// Leading whitespace after a flush has no content value; drop it
+			// so the next piece does not start with a pure-space prefix that
+			// would never fit usefully on its own.
+			if strings.TrimSpace(atom) == "" {
+				continue
+			}
+			// If the atom alone still exceeds (pathological), carve it.
+			if atomTokens(atom) > chunkTokenNum {
+				pieces = append(pieces, splitAtomByTokenBudget(atom, chunkTokenNum, countFn)...)
+				continue
+			}
+		}
+		current += atom
+	}
+	if current != "" {
+		pieces = append(pieces, current)
+	}
+	return pieces
+}
+
+// computeOverlapPrefix returns (overlapText, overlapTokenCount) carved from
+// the tail of prevText after stripping parser tags. overlappedPct is a
+// percentage in [0, 100]. Mirrors Python rag/nlp._compute_overlap_prefix.
+func computeOverlapPrefix(prevText string, overlappedPct float64) (string, int) {
+	visible := removeTag(prevText)
+	if visible == "" {
+		return "", 0
+	}
+	runes := []rune(visible)
+	cut := int(float64(len(runes)) * (100 - overlappedPct) / 100.0)
+	if cut < 0 {
+		cut = 0
+	}
+	if cut >= len(runes) {
+		return "", 0
+	}
+	overlap := string(runes[cut:])
+	return overlap, tokenizeStr(overlap)
+}
+
 // mergeByTokenSize implements exact token-based chunk merging that mirrors
-// Python's naive_merge (rag/nlp/__init__.py:1156). It uses
-// tokenizeStr (= tokenizer.NumTokensFromString, cl100k_base BPE) for
-// precise token counting, treats the payload as a single section, splits
-// oversized sections on sentence delimiters (dropping the delimiter, as
-// Python does), and greedily merges into chunks of approximately
-// chunk_token_size tokens with optional overlap from the previous chunk.
+// Python's naive_merge (rag/nlp/__init__.py) after the strict chunk_token_num
+// hard-cap fix. It uses tokenizeStr for precise token counting, treats the
+// payload as a single section, splits oversized sections on production sentence
+// delimiters, hard-caps atomic oversize units via splitOversizedUnit, and merges
+// only when the projected total stays within chunk_token_size. Overlap is
+// applied only when the resulting chunk still fits the budget.
 func (c *TokenChunkerComponent) mergeByTokenSize(text string, childrenPattern *regexp.Regexp) map[string]any {
 	target := c.param.ChunkTokenSize
 	overlapPct := c.param.OverlappedPercent
@@ -359,61 +491,58 @@ func (c *TokenChunkerComponent) mergeByTokenSize(text string, childrenPattern *r
 	}

 	// Normalize line endings to LF before any splitting. Python's
-	// naive_merge (rag/nlp/__init__.py:1166) runs
-	//   text = text.replace("\r\n", "\n").replace("\r", "\n")
-	// so CRLF/CR input must segment and split exactly like LF input.
-	// Without this, stray "\r" would survive inside chunks, diverging
-	// from Python.
+	// naive_merge runs text.replace("\r\n", "\n").replace("\r", "\n"),
+	// then treats the input string as one section.
 	text = strings.ReplaceAll(strings.ReplaceAll(text, "\r\n", "\n"), "\r", "\n")
-
-	// Treat the whole payload as a single section, mirroring Python's
-	// naive_merge (rag/nlp/__init__.py:1157) which wraps the input string
-	// as a one-element list. naive_merge does NOT pre-split on blank
-	// lines, and because "\n" is itself a delimiter it is dropped (blank
-	// lines collapse), exactly as Python does. CRLF/CR normalization
-	// already happened above.
 	sections := []string{text}
 	if len(sections) == 0 {
 		return emptyOutputs()
 	}

-	// Sentence/clause-boundary regex for splitting oversized sections.
-	// Mirrors Python's production delimiter (rag/app/naive.py:1285 passes
-	// "\n!?。；！？") — ASCII "!" and "?" plus the CJK punctuation, with no
-	// English ". " fallback.
-	sentenceDelim := sentenceDelimiter
+	var cks []string
+	var tkns []int

-	var cks []string // chunk texts
-	var tkns []int   // token counts per chunk
-
-	// mergeOrNew mirrors Python add_chunk in naive_merge:
-	//   - If the current chunk is empty or would overflow the
-	//     threshold → start a new chunk (with optional overlap prefix).
-	//   - Otherwise → merge into the current chunk.
-	mergeOrNew := func(segment string, tokens int) {
-		threshold := float64(target) * (100 - overlapPct) / 100.0
-		if len(cks) == 0 || float64(tkns[len(tkns)-1]) > threshold {
-			seg := segment
-			segTokens := tokens
-			if overlapPct > 0 && len(cks) > 0 {
-				// Strip parser tags before computing the overlap suffix,
-				// matching Python nlp/__init__.py:1181
-				prev := removeTag(cks[len(cks)-1])
-				// Take the last overlapped_percent of the previous chunk
-				// (in runes, matching Python's len(overlapped) * ratio).
-				prevRunes := []rune(prev)
-				cut := int(float64(len(prevRunes)) * (100 - overlapPct) / 100.0)
-				if cut < len(prevRunes) {
-					suffix := string(prevRunes[cut:])
-					seg = suffix + segment
-					segTokens = tokenizeStr(seg)
+	// addChunk applies the projected-total merge and optional-overlap decision
+	// to one unit that already fits target.
+	addChunk := func(segment string) {
+		tnum := tokenizeStr(segment)
+		if len(cks) == 0 {
+			cks = append(cks, segment)
+			tkns = append(tkns, tnum)
+			return
+		}
+		merged := cks[len(cks)-1] + segment
+		mergedN := tokenizeStr(merged)
+		if mergedN <= target {
+			cks[len(cks)-1] = merged
+			tkns[len(tkns)-1] = mergedN
+			return
+		}
+		newText := segment
+		newTokens := tnum
+		if overlapPct > 0 {
+			overlapText, _ := computeOverlapPrefix(cks[len(cks)-1], overlapPct)
+			if overlapText != "" {
+				candidate := overlapText + segment
+				if candidateTokens := tokenizeStr(candidate); candidateTokens <= target {
+					newText = candidate
+					newTokens = candidateTokens
 				}
 			}
-			cks = append(cks, seg)
-			tkns = append(tkns, segTokens)
-		} else {
-			cks[len(cks)-1] += segment
-			tkns[len(tkns)-1] += tokens
+		}
+		cks = append(cks, newText)
+		tkns = append(tkns, newTokens)
+	}
+
+	addUnit := func(unit string) {
+		if tokenizeStr(unit) <= target {
+			addChunk(unit)
+			return
+		}
+		slog.Debug("TokenChunker: splitting oversized unit via splitOversizedUnit",
+			"len", len(unit), "tokens", tokenizeStr(unit), "chunk_token_size", target)
+		for _, piece := range splitOversizedUnit(unit, target) {
+			addChunk(piece)
 		}
 	}

@@ -423,46 +552,24 @@ func (c *TokenChunkerComponent) mergeByTokenSize(text string, childrenPattern *r
 			continue
 		}
 		t := "\n" + sec
-		tn := tokenizeStr(t)
-
-		if tn < 8 {
-			// Tiny section — always merge into the previous chunk.
-			if len(cks) > 0 {
-				cks[len(cks)-1] += t
-				tkns[len(tkns)-1] += tn
-			} else {
-				cks = append(cks, t)
-				tkns = append(tkns, tn)
-			}
+		if tokenizeStr(t) <= target {
+			addChunk(t)
 			continue
 		}
-
-		if tn <= target {
-			mergeOrNew(t, tn)
-			continue
-		}
-
-		// Oversized section: split on sentence delimiters. Python's
-		// naive_merge (rag/nlp/__init__.py:1216-1225) splits with a
-		// capturing-group re.split but then SKIPS any segment that is a
-		// pure delimiter (re.fullmatch(dels, sub_sec)), so the delimiter
-		// character (\n / 。 / ！ / ？) is DROPPED from the chunk text
-		// rather than retained. We mirror that by using regexp.Split
-		// (which discards the delimiter) and prepending a single "\n" to
-		// each segment, matching Python's add_chunk("\n"+sub_sec).
-		parts := sentenceDelim.Split(sec, -1)
+		// Oversized section: split on production sentence delimiters, then
+		// hard-cap any unit that still exceeds the budget (unbroken atoms).
+		parts := sentenceDelimiter.Split(sec, -1)
+		hadPart := false
 		for _, part := range parts {
 			part = strings.TrimSpace(part)
 			if part == "" {
 				continue
 			}
-			// Route every segment — including tiny <8-token fragments —
-			// through mergeOrNew so it honours the token threshold,
-			// mirroring Python's add_chunk. The old shortcut appended
-			// unconditionally, merging fragments into an already-overfull
-			// chunk (review #2).
-			p := "\n" + part
-			mergeOrNew(p, tokenizeStr(p))
+			hadPart = true
+			addUnit("\n" + part)
+		}
+		if !hadPart {
+			addUnit(t)
 		}
 	}

@@ -764,75 +871,99 @@ func takeFromStart(text string, tokens int) string {
 	return best
 }

-// mergeByTokenSizeFromJSON mirrors `naive_merge` at
-// rag/nlp/__init__.py:1156.
+// mergeByTokenSizeFromJSON mirrors Python naive_merge's projected-total
+// hard cap (rag/nlp/__init__.py after the strict chunk_token_num fix).
+// Oversized text units are sub-split via splitOversizedUnit before merge;
+// overlap is applied only when overlap+segment still fits the budget.
 func mergeByTokenSizeFromJSON(perItem [][]schema.ChunkDoc, chunkTokens int, overlappedPct float64) [][]schema.ChunkDoc {
-	// overlappedPct is a [0,100] percentage. Clamp so the merge math below
-	// never yields a negative/inverted threshold for out-of-range input
-	// (review: yuzhichang, PR #17396).
+	// overlappedPct is a [0,100] percentage. Clamp defensively because this
+	// helper is also exercised directly by tests.
 	if overlappedPct < 0 {
 		overlappedPct = 0
 	} else if overlappedPct > 100 {
 		overlappedPct = 100
 	}
-	threshold := float64(chunkTokens) * (100 - overlappedPct) / 100.0
 	for idx := range perItem {
 		chunks := perItem[idx]
 		if len(chunks) == 0 {
 			continue
 		}
 		var merged []schema.ChunkDoc
+
+		// addTextChunk applies the projected-total merge / overlap-drop
+		// decision for one text unit that already fits chunkTokens.
+		addTextChunk := func(ck schema.ChunkDoc) {
+			tk := intValue(ck.TKNums)
+			if tk <= 0 {
+				tk = tokenizeStr(ck.Text)
+				ck.TKNums = intPtr(tk)
+			}
+			if len(merged) == 0 || merged[len(merged)-1].CKType != "text" {
+				// First text chunk, or first text after a non-text chunk:
+				// no prior text to overlap with.
+				merged = append(merged, cloneChunkDoc(ck))
+				return
+			}
+			prev := &merged[len(merged)-1]
+			// Empty previous text: assign incoming text directly
+			// (diff Chunker-2.11 / token_chunker.py:236-239).
+			if prev.Text == "" {
+				prev.Text = ck.Text
+				prev.TKNums = intPtr(tk)
+				prev.PDFPositions = extendRawJSONArray(prev.PDFPositions, ck.PDFPositions)
+				prev.Positions = extendRawJSONArray(prev.Positions, ck.Positions)
+				return
+			}
+			// Proactive projected-total merge (joined with "\n").
+			joined := prev.Text + "\n" + ck.Text
+			joinedN := tokenizeStr(joined)
+			if joinedN <= chunkTokens {
+				prev.Text = joined
+				prev.TKNums = intPtr(joinedN)
+				prev.PDFPositions = extendRawJSONArray(prev.PDFPositions, ck.PDFPositions)
+				prev.Positions = extendRawJSONArray(prev.Positions, ck.Positions)
+				return
+			}
+			// Start a new chunk; apply overlap only when it still fits.
+			cp := cloneChunkDoc(ck)
+			if overlappedPct > 0 {
+				if overlapText, overlapTokens := computeOverlapPrefix(prev.Text, overlappedPct); overlapTokens > 0 && overlapTokens+tk <= chunkTokens {
+					cp.Text = overlapText + cp.Text
+					cp.TKNums = intPtr(tokenizeStr(cp.Text))
+				}
+			}
+			merged = append(merged, cp)
+		}
+
 		for _, ck := range chunks {
-			ckType := ck.CKType
-			if ckType != "text" {
+			if ck.CKType != "text" {
 				merged = append(merged, cloneChunkDoc(ck))
 				continue
 			}
 			tk := intValue(ck.TKNums)
-			// Mirror Python's naive_merge.add_chunk: start a new chunk
-			// when either (a) this is the first text chunk, or
-			// (b) the currently accumulated chunk already exceeds the
-			// threshold (not the incoming segment).
-			if len(merged) == 0 || merged[len(merged)-1].CKType != "text" ||
-				float64(intValue(merged[len(merged)-1].TKNums)) > threshold {
-				cp := cloneChunkDoc(ck)
-				// Overlap: prepend tail of previous chunk onto the new
-				// chunk, matching Python's
-				//   t = overlapped[overlap_cut:] + t
-				//   tnum = num_tokens_from_string(t)
-				if len(merged) > 0 && merged[len(merged)-1].CKType == "text" && overlappedPct > 0 {
-					// Strip parser tags before computing the overlap
-					// suffix, matching Python nlp/__init__.py:1181
-					//
-					if prevText := removeTag(merged[len(merged)-1].Text); prevText != "" {
-						runes := []rune(prevText)
-						cut := int(float64(len(runes)) * (100 - overlappedPct) / 100.0)
-						if cut < len(runes) {
-							cp.Text = string(runes[cut:]) + cp.Text
-							cp.TKNums = intPtr(tokenizeStr(cp.Text))
-						}
-					}
-				}
-				merged = append(merged, cp)
+			if tk <= 0 {
+				tk = tokenizeStr(ck.Text)
+			}
+			if tk <= chunkTokens {
+				addTextChunk(ck)
 				continue
 			}
-			// Merge into the accumulated text chunk.
-			prev := &merged[len(merged)-1]
-			// Mirror Python token_chunker.py:236-239: when the accumulated
-			// chunk has empty text, assign the incoming text directly instead
-			// of skipping it
-			if prev.Text == "" {
-				prev.Text = ck.Text
-			} else {
-				prev.Text = prev.Text + "\n" + ck.Text
+			// Hard-cap atomic oversize units before merge.
+			slog.Debug("TokenChunker: splitting oversized JSON unit via splitOversizedUnit",
+				"len", len(ck.Text), "tokens", tk, "chunk_token_size", chunkTokens)
+			for _, piece := range splitOversizedUnit(ck.Text, chunkTokens) {
+				if strings.TrimSpace(piece) == "" {
+					continue
+				}
+				cp := cloneChunkDoc(ck)
+				cp.Text = piece
+				cp.TKNums = intPtr(tokenizeStr(piece))
+				// Coordinates stay on the first piece only to avoid duplicating
+				// PDF bboxes across atom slices.
+				addTextChunk(cp)
+				ck.PDFPositions = nil
+				ck.Positions = nil
 			}
-			prev.TKNums = intPtr(intValue(prev.TKNums) + tk)
-			// Preserve PDF coordinates across the merge: extend the
-			// coordinate lists instead of dropping the incoming item's
-			// positions. Mirrors Python token_chunker.py:240
-			// `merged[prev][PDF_POSITIONS_KEY].extend(...)` (diffs 2.5 / 2.3).
-			prev.PDFPositions = extendRawJSONArray(prev.PDFPositions, ck.PDFPositions)
-			prev.Positions = extendRawJSONArray(prev.Positions, ck.Positions)
 		}
 		perItem[idx] = merged
 	}
--- a/internal/ingestion/component/chunker/token_batch1_test.go
+++ b/internal/ingestion/component/chunker/token_batch1_test.go
@@ -54,18 +54,40 @@ func TestSentenceDelimiterMatchesBangAndQuestion(t *testing.T) {
 // from the previous chunk AFTER remove_tag, otherwise parser tags (e.g.
 // "@@1\t2.3##") leak into the overlap region. Mirrors Python
 // nlp/__init__.py:1181 (remove_tag applied before overlap).
+//
+// After the strict-cap fix, a new chunk is started only when the projected
+// join exceeds the budget — so the first unit must already sit near the
+// budget and the second unit must not fit alongside it.
 func TestMergeByTokenSizeFromJSON_OverlapStripsTags(t *testing.T) {
+	// Size a and b so:
+	//   - each unit alone fits the budget (no atom-split),
+	//   - the projected join exceeds the budget (forces a new chunk),
+	//   - overlap+b still fits (so the overlap path is exercised).
 	aText := strings.Repeat("word ", 20) + "@@1\t2.3## tail"
+	bText := "body"
+	aN, bN := tokenizeStr(aText), tokenizeStr(bText)
+	joinedN := tokenizeStr(aText + "\n" + bText)
+	// Budget just below the join so a and b cannot merge, but each alone fits.
+	budget := joinedN - 1
+	if budget < aN {
+		budget = aN
+	}
+	if budget < bN {
+		budget = bN
+	}
+	if joinedN <= budget {
+		t.Fatalf("could not derive tight budget (a=%d b=%d joined=%d budget=%d)", aN, bN, joinedN, budget)
+	}
 	items := [][]schema.ChunkDoc{
 		{
-			{Text: aText, DocType: "text", CKType: "text", TKNums: intPtr(100)},
-			{Text: "body", DocType: "text", CKType: "text", TKNums: intPtr(5)},
+			{Text: aText, DocType: "text", CKType: "text", TKNums: intPtr(aN)},
+			{Text: bText, DocType: "text", CKType: "text", TKNums: intPtr(bN)},
 		},
 	}
-	got := mergeByTokenSizeFromJSON(items, 128, 30.0)
+	got := mergeByTokenSizeFromJSON(items, budget, 30.0)
 	merged := got[0]
 	if len(merged) != 2 {
-		t.Fatalf("want 2 merged chunks (overlap path), got %d", len(merged))
+		t.Fatalf("want 2 merged chunks (overlap path), got %d (a=%d b=%d budget=%d)", len(merged), aN, bN, budget)
 	}
 	// The overlap prefix is prepended to the SECOND chunk. The original
 	// first chunk legitimately keeps its own parser tag; only the overlap
@@ -73,6 +95,9 @@ func TestMergeByTokenSizeFromJSON_OverlapStripsTags(t *testing.T) {
 	if strings.Contains(merged[1].Text, "@@") || strings.Contains(merged[1].Text, "##") {
 		t.Errorf("overlap prefix leaked parser tag into chunk 1: %q", merged[1].Text)
 	}
+	if n := tokenizeStr(merged[1].Text); n > budget {
+		t.Errorf("overlap pushed second chunk over budget: tokens=%d", n)
+	}
 }

 // TestMergeByTokenSizeFromJSON_ClampsOverlappedPct locks the review finding
--- a/internal/ingestion/component/chunker/token_strict_cap_test.go
+++ b/internal/ingestion/component/chunker/token_strict_cap_test.go
@@ -0,0 +1,266 @@
+//
+//  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 chunker
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"unicode/utf8"
+
+	"ragflow/internal/ingestion/component/schema"
+)
+
+// wordCount is a deterministic tokenizer stand-in used only via
+// splitOversizedUnitWith in unit-level helper tests.
+func wordCount(s string) int {
+	s = strings.TrimSpace(s)
+	if s == "" {
+		return 0
+	}
+	return len(strings.Fields(s))
+}
+
+func charCount(s string) int { return utf8.RuneCountInString(s) }
+
+func TestSplitOversizedUnit_WhitespacePacksToBudget(t *testing.T) {
+	// 100 words, budget 30 → must yield multiple pieces, each ≤ 30 words.
+	text := strings.TrimSpace(strings.Repeat("word ", 100))
+	pieces := splitOversizedUnitWith(text, 30, wordCount)
+	if len(pieces) < 2 {
+		t.Fatalf("want multiple pieces, got %d: %#v", len(pieces), pieces)
+	}
+	total := 0
+	for _, p := range pieces {
+		n := wordCount(p)
+		if n > 30 {
+			t.Errorf("piece exceeds budget: tokens=%d text=%q", n, p)
+		}
+		total += n
+	}
+	if total != 100 {
+		t.Errorf("word count not preserved: got %d want 100", total)
+	}
+}
+
+func TestSplitOversizedUnit_UnbrokenAtomFallsBackToCharWindows(t *testing.T) {
+	// Unbroken run with char-as-token counting — must sub-split on runes.
+	atom := strings.Repeat("a", 80)
+	pieces := splitOversizedUnitWith(atom, 50, charCount)
+	if len(pieces) < 2 {
+		t.Fatalf("want >=2 pieces for unbroken atom, got %d", len(pieces))
+	}
+	joined := strings.Join(pieces, "")
+	if joined != atom {
+		t.Errorf("content not preserved: got %q", joined)
+	}
+	for _, p := range pieces {
+		if charCount(p) > 50 {
+			t.Errorf("piece exceeds budget: %d runes in %q", charCount(p), p)
+		}
+	}
+}
+
+func TestSplitOversizedUnit_WithinBudgetUnchanged(t *testing.T) {
+	text := "hello world"
+	pieces := splitOversizedUnitWith(text, 100, wordCount)
+	if len(pieces) != 1 || pieces[0] != text {
+		t.Fatalf("within-budget text must be returned as-is, got %#v", pieces)
+	}
+}
+
+func TestComputeOverlapPrefix_StripsTagsAndCounts(t *testing.T) {
+	prev := strings.Repeat("word ", 20) + "@@1\t2.3## tail"
+	overlap, n := computeOverlapPrefix(prev, 30)
+	if strings.Contains(overlap, "@@") || strings.Contains(overlap, "##") {
+		t.Errorf("overlap must strip parser tags, got %q", overlap)
+	}
+	if n <= 0 {
+		t.Errorf("overlap token count must be >0, got %d", n)
+	}
+	if tokenizeStr(overlap) != n {
+		t.Errorf("reported tokens %d != tokenizeStr(overlap) %d", n, tokenizeStr(overlap))
+	}
+}
+
+func TestMergeByTokenSizeFromJSON_StrictCapNoOvershoot(t *testing.T) {
+	// Eight 25-token-ish sections under a 50-token budget must pack without
+	// any chunk exceeding the budget (Python test_strict_cap_no_overlap).
+	const budget = 50
+	sections := make([]schema.ChunkDoc, 0, 8)
+	for i := 0; i < 8; i++ {
+		text := strings.TrimSpace(strings.Repeat("w ", 25))
+		sections = append(sections, schema.ChunkDoc{
+			Text: text, DocType: "text", CKType: "text", TKNums: intPtr(tokenizeStr(text)),
+		})
+	}
+	got := mergeByTokenSizeFromJSON([][]schema.ChunkDoc{sections}, budget, 0)
+	merged := got[0]
+	if len(merged) < 3 {
+		t.Fatalf("want >=3 chunks, got %d", len(merged))
+	}
+	for i, ck := range merged {
+		n := tokenizeStr(ck.Text)
+		if n > budget {
+			t.Errorf("chunk %d exceeds budget: tokens=%d text_len=%d", i, n, len(ck.Text))
+		}
+	}
+}
+
+func TestMergeByTokenSizeFromJSON_OverlapDroppedAtOverflow(t *testing.T) {
+	// With a tight budget, overlap must never push a chunk over the cap.
+	const budget = 25
+	sections := make([]schema.ChunkDoc, 0, 20)
+	for i := 0; i < 20; i++ {
+		text := strings.TrimSpace(strings.Repeat("w ", 10))
+		sections = append(sections, schema.ChunkDoc{
+			Text: text, DocType: "text", CKType: "text", TKNums: intPtr(tokenizeStr(text)),
+		})
+	}
+	got := mergeByTokenSizeFromJSON([][]schema.ChunkDoc{sections}, budget, 20)
+	for i, ck := range got[0] {
+		if n := tokenizeStr(ck.Text); n > budget {
+			t.Errorf("chunk %d exceeds budget with overlap: tokens=%d", i, n)
+		}
+	}
+}
+
+func TestMergeByTokenSizeFromJSON_OversizedUnitIsSubSplit(t *testing.T) {
+	// A single unit larger than the budget must be atom-split before merge.
+	const budget = 30
+	long := strings.TrimSpace(strings.Repeat("word ", 100))
+	items := [][]schema.ChunkDoc{{
+		{Text: long, DocType: "text", CKType: "text", TKNums: intPtr(tokenizeStr(long))},
+	}}
+	got := mergeByTokenSizeFromJSON(items, budget, 0)
+	if len(got[0]) < 2 {
+		t.Fatalf("oversized unit must yield multiple chunks, got %d", len(got[0]))
+	}
+	for i, ck := range got[0] {
+		if n := tokenizeStr(ck.Text); n > budget {
+			t.Errorf("chunk %d exceeds budget: tokens=%d", i, n)
+		}
+	}
+}
+
+func TestMergeByTokenSize_TextPathStrictCap(t *testing.T) {
+	// End-to-end text path: long multi-paragraph input under a tight budget.
+	const budget = 40
+	var b strings.Builder
+	for i := 0; i < 30; i++ {
+		b.WriteString(strings.TrimSpace(strings.Repeat("word ", 15)))
+		b.WriteString("\n\n")
+	}
+	comp, err := NewTokenChunker(map[string]any{
+		"delimiter_mode":   "token_size",
+		"chunk_token_size": budget,
+	})
+	if err != nil {
+		t.Fatalf("NewTokenChunker: %v", err)
+	}
+	tc := comp.(*TokenChunkerComponent)
+	out := tc.mergeByTokenSize(b.String(), nil)
+	chunks, _ := out["chunks"].([]map[string]any)
+	if len(chunks) < 2 {
+		t.Fatalf("want multiple chunks, got %d", len(chunks))
+	}
+	for i, ck := range chunks {
+		text, _ := ck["text"].(string)
+		if n := tokenizeStr(text); n > budget {
+			t.Errorf("chunk %d exceeds budget: tokens=%d", i, n)
+		}
+	}
+}
+
+func TestMergeByTokenSize_UnbrokenAtomStrictCap(t *testing.T) {
+	// Unbroken dense string (no whitespace / sentence delim) must still
+	// hard-cap via the character-window fallback inside splitOversizedUnit.
+	const budget = 20
+	// Use many distinct ASCII letters so cl100k does not collapse the whole
+	// run into a handful of tokens.
+	var b strings.Builder
+	for i := 0; i < 400; i++ {
+		b.WriteByte(byte('a' + i%26))
+	}
+	text := b.String()
+	if tokenizeStr(text) <= budget {
+		t.Skipf("tokenizer collapsed unbroken atom to %d tokens (<= budget)", tokenizeStr(text))
+	}
+	comp, err := NewTokenChunker(map[string]any{
+		"delimiter_mode":   "token_size",
+		"chunk_token_size": budget,
+	})
+	if err != nil {
+		t.Fatalf("NewTokenChunker: %v", err)
+	}
+	tc := comp.(*TokenChunkerComponent)
+	out := tc.mergeByTokenSize(text, nil)
+	chunks, _ := out["chunks"].([]map[string]any)
+	if len(chunks) < 2 {
+		t.Fatalf("want multiple chunks for unbroken atom, got %d (total_tokens=%d)", len(chunks), tokenizeStr(text))
+	}
+	var joined strings.Builder
+	for i, ck := range chunks {
+		s, _ := ck["text"].(string)
+		joined.WriteString(s)
+		if n := tokenizeStr(s); n > budget {
+			t.Errorf("chunk %d exceeds budget: tokens=%d text=%q", i, n, s)
+		}
+	}
+	// mergeByTokenSize prefixes "\n" on sections; stripping newlines recovers
+	// the original unbroken atom.
+	if strings.ReplaceAll(joined.String(), "\n", "") != text {
+		t.Errorf("content not preserved after stripping newlines: got %q", joined.String())
+	}
+}
+
+func TestInvokeTextPayload_StrictCapEndToEnd(t *testing.T) {
+	const budget = 32
+	var b strings.Builder
+	for i := 0; i < 20; i++ {
+		b.WriteString(strings.TrimSpace(strings.Repeat("alpha ", 12)))
+		b.WriteByte('\n')
+	}
+	comp, err := NewTokenChunker(map[string]any{
+		"delimiter_mode":   "token_size",
+		"chunk_token_size": budget,
+	})
+	if err != nil {
+		t.Fatalf("NewTokenChunker: %v", err)
+	}
+	out, err := comp.Invoke(context.Background(), nil, map[string]any{
+		"name":          "doc.txt",
+		"output_format": "text",
+		"text":          b.String(),
+	})
+	if err != nil {
+		t.Fatalf("Invoke: %v", err)
+	}
+	if errMsg, _ := out["_ERROR"].(string); errMsg != "" {
+		t.Fatalf("Invoke error payload: %s", errMsg)
+	}
+	chunks, _ := out["chunks"].([]map[string]any)
+	if len(chunks) == 0 {
+		t.Fatalf("expected chunks, got %#v", out)
+	}
+	for i, ck := range chunks {
+		text, _ := ck["text"].(string)
+		if n := tokenizeStr(text); n > budget {
+			t.Errorf("chunk %d exceeds budget: tokens=%d", i, n)
+		}
+	}
+}