feat: parser pages range and parse type validation for dataset/document (#17293)

## Summary Adds page-range parsing support to the Go-native pipeline path and introduces strict `parse_type` validation for both dataset and document update endpoints. ## What changed ### Pages range parsing - **`internal/utility/pdf_pages.go`** — `NormalizePDFPages`: normalizes raw page ranges (list of `[from,to]` 1-indexed inclusive ranges) into sorted, merged, deduplicated `[][]int`. Invalid ranges are dropped. - **`internal/ingestion/pipeline/pdf_pages.go`** — `NormalizeParserConfigPages`: walks any parser_config map and normalizes `"pages"` values under every component → filetype setup, so the persisted config always carries clean, merged ranges. - **`internal/deepdoc/parser/pdf/parser.go`** — integrates `resolvePagesToProcess` to filter parsed PDF pages by the configured ranges. - Pipeline integration (parser pages): `internal/parser/parser/pdf_parser_common.go`, `chunk_process.go`, plus associated e2e and unit tests. ### Parse type validation (shared logic) - **`internal/service/parser_mode.go`** (new) — `ValidateParseTypeMode`: shared function that validates `parse_type` (1=BuiltIn/parser_id, 2=Pipeline/pipeline_id) and ensures the corresponding field is present. Used by both dataset and document update endpoints. - **`internal/service/dataset/crud.go`** / `update.go` — replaces inline `isPipelineMode`/`isBuiltinMode` computation with the shared `service.ValidateParseTypeMode`. - **`internal/service/document/document_dataset_update.go`** — adds strict `parse_type` validation in `validateDatasetDocumentUpdate`, simplifies the reparse logic to a two-way switch (isBuiltin/isPipeline) now that parse_type is always valid. - **`internal/service/document/document.go`** — adds `ParseType` field to `UpdateDatasetDocumentRequest`. - **`internal/service/document/document_dataset_update.go`** — `updateDocumentParserConfig` fallback path when DSL loading fails. - **`internal/service/parser_mode_test.go`** (new) — test coverage for nil, invalid, and missing-field scenarios. ### Frontend - **`web/src/interfaces/request/document.ts`** — adds `parseType` to `IChangeParserRequestBody`. - **`web/src/hooks/use-document-request.ts`** — `useSetDocumentPipelineParser` sends `parse_type` in the PATCH payload. - **`web/src/pages/dataset/dataset/use-change-document-parser.ts`** — Go/Python branching for the document parser config dialog. - **`web/src/components/document-pipeline-dialog/use-document-pipeline-form.ts`** — `buildSubmitData` returns `parseType` (bugfix: was dropped from the return value). ### Test changes - **Removed**: 2 tests that verified the old "mutually exclusive" error (replaced by `ValidateParseTypeMode` coverage). - **Modified**: 6 tests across document and dataset packages to include `ParseType` in request structs. - **Added**: new e2e tests for pages parsing (`pages_e2e_test.go`, `pdf_parser_pages_e2e_test.go`) and unit tests for `NormalizePDFPages`, `NormalizeParserConfigPages`, `resolvePagesToProcess`. ## Backward compatibility - The `parse_type` field is **required** when `parser_id` or `pipeline_id` is sent. This changes the contract for both dataset and document PATCH endpoints, but aligns the Go backend with the existing frontend behavior (the frontend already sends `parse_type`). Callers that omit `parse_type` when updating parser/pipeline selections will receive a clear error message. - Existing callers that only update fields like `name`, `enabled`, or `meta_fields` are unaffected. - Test updates ensure all known call sites are compliant.
2026-07-24 17:36:47 +08:00 · 2026-07-23 19:57:27 +08:00
parent fce2a94fe0
commit d12fd3b79d
45 changed files with 1819 additions and 496 deletions
--- a/internal/admin/service.go
+++ b/internal/admin/service.go
@@ -241,7 +241,7 @@ func (s *Service) CreateUser(username, password, role string) (map[string]interf
 	asrModel := ""
 	vlmModel := ""
 	rerankModel := ""
-	parserIDs := "naive:General,qa:Q&A,resume:Resume,manual:Manual,table:Table,paper:Paper,book:Book,laws:Laws,presentation:Presentation,picture:Picture,one:One,audio:Audio,email:Email,tag:Tag"
+	parserIDs := "naive:General,qa:Q&A,resume:Resume,manual:Manual,table:Table,paper:Paper,book:Book,laws:Laws,presentation:Presentation,picture:Picture,one:One,audio:Audio,email:Email"

 	if cfg != nil {
 		chatModel = cfg.UserDefaultLLM.DefaultModels.ChatModel.Name
--- a/internal/deepdoc/parser/pdf/ocr_merge_test.go
+++ b/internal/deepdoc/parser/pdf/ocr_merge_test.go
@@ -6,6 +6,7 @@ import (
 	"context"
 	"image/png"
 	"os"
+	"ragflow/internal/common"
 	inf "ragflow/internal/deepdoc/parser/pdf/inference"
 	pdftype "ragflow/internal/deepdoc/parser/pdf/type"
 	util "ragflow/internal/deepdoc/parser/pdf/util"
--- a/internal/deepdoc/parser/pdf/pages_e2e_test.go
+++ b/internal/deepdoc/parser/pdf/pages_e2e_test.go
@@ -0,0 +1,119 @@
+package pdf
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+	"strings"
+	"testing"
+
+	pdf "ragflow/internal/deepdoc/parser/pdf/type"
+	"ragflow/internal/utility"
+)
+
+// TestPagesEndToEnd_NormalizeThenParse verifies the full pages path: a
+// frontend-submitted (JSON-decoded, possibly dirty) pages value is normalized
+// by utility.NormalizePDFPages, fed into ParserConfig.Pages, and the deepdoc
+// parser only processes the resulting page ranges.
+//
+// This ties together step 3 (API-layer normalization) and step 1 (deepdoc
+// page filtering) end-to-end at the parser level.
+func TestPagesEndToEnd_NormalizeThenParse(t *testing.T) {
+	// jsonPages builds a JSON-decoded-style []any of []any of float64, mimicking
+	// what arrives over the wire from the frontend.
+	jsonPages := func(ranges ...[2]float64) any {
+		out := make([]any, 0, len(ranges))
+		for _, r := range ranges {
+			out = append(out, []any{r[0], r[1]})
+		}
+		return out
+	}
+
+	t.Run("overlapping and unsorted ranges normalized then filtered", func(t *testing.T) {
+		// Frontend submitted [1,3],[2,5],[8,10] (overlap 1-3 & 2-5, unsorted).
+		raw := jsonPages([2]float64{1, 3}, [2]float64{2, 5}, [2]float64{8, 10})
+		normalized, err := utility.NormalizePDFPages(raw) // -> [[1,5],[8,10]]
+		if err != nil {
+			t.Fatalf("NormalizePDFPages: %v", err)
+		}
+
+		wantNorm := [][]int{{1, 5}, {8, 10}}
+		if !reflect.DeepEqual(normalized, wantNorm) {
+			t.Fatalf("NormalizePDFPages = %v, want %v", normalized, wantNorm)
+		}
+
+		cfg := pdf.DefaultParserConfig()
+		cfg.Pages = normalized
+		p := NewParser(cfg)
+
+		eng := makePageTaggedEngine(10)
+		result, err := p.ParseRaw(context.Background(), eng, &MockDocAnalyzer{Healthy: true})
+		if err != nil {
+			t.Fatalf("ParseRaw: %v", err)
+		}
+
+		// [1,5] -> 0-based 0..4 ; [8,10] -> 7..9
+		wantPages := map[int]struct{}{0: {}, 1: {}, 2: {}, 3: {}, 4: {}, 7: {}, 8: {}, 9: {}}
+		if got := pageHeightKeys(result.PageHeight); !reflect.DeepEqual(got, wantPages) {
+			t.Errorf("PageHeight keys = %v, want %v", got, wantPages)
+		}
+
+		// Sections must not carry text from skipped pages (5, 6).
+		combined := combineSectionText(result.Sections)
+		for _, pg := range []int{5, 6} {
+			if strings.Contains(combined, fmt.Sprintf("p%d", pg)) {
+				t.Errorf("expected p%d to be skipped; combined=%q", pg, combined)
+			}
+		}
+	})
+
+	t.Run("invalid range rejected during normalization (fail-fast)", func(t *testing.T) {
+		// [1,3],[3,1](from>to),[8,10] -> fail-fast rejects the whole input.
+		raw := jsonPages([2]float64{1, 3}, [2]float64{3, 1}, [2]float64{8, 10})
+		normalized, err := utility.NormalizePDFPages(raw)
+		if err == nil {
+			t.Fatalf("NormalizePDFPages(%v) expected error, got %v", raw, normalized)
+		}
+		if normalized != nil {
+			t.Fatalf("NormalizePDFPages(%v) = %v, want nil on error", raw, normalized)
+		}
+	})
+
+	t.Run("all invalid -> error (fail-fast, no parse-all fallback)", func(t *testing.T) {
+		// [3,1],[0,2] both invalid -> error. The request layer rejects this
+		// before it ever reaches the parser; normalize surfaces the error
+		// rather than silently degrading to "parse all pages".
+		raw := jsonPages([2]float64{3, 1}, [2]float64{0, 2})
+		normalized, err := utility.NormalizePDFPages(raw)
+		if err == nil {
+			t.Fatalf("NormalizePDFPages(%v) expected error, got %v", raw, normalized)
+		}
+		if normalized != nil {
+			t.Fatalf("NormalizePDFPages(%v) = %v, want nil on error", raw, normalized)
+		}
+	})
+
+	t.Run("range clamped to document page count", func(t *testing.T) {
+		// [1,1000000] on a 5-page doc -> clamped to all 5 pages.
+		raw := jsonPages([2]float64{1, 1000000})
+		normalized, err := utility.NormalizePDFPages(raw)
+		if err != nil {
+			t.Fatalf("NormalizePDFPages: %v", err)
+		}
+
+		cfg := pdf.DefaultParserConfig()
+		cfg.Pages = normalized
+		p := NewParser(cfg)
+
+		eng := makePageTaggedEngine(5)
+		result, err := p.ParseRaw(context.Background(), eng, &MockDocAnalyzer{Healthy: true})
+		if err != nil {
+			t.Fatalf("ParseRaw: %v", err)
+		}
+
+		wantPages := map[int]struct{}{0: {}, 1: {}, 2: {}, 3: {}, 4: {}}
+		if got := pageHeightKeys(result.PageHeight); !reflect.DeepEqual(got, wantPages) {
+			t.Errorf("PageHeight keys = %v, want all 5 pages", got)
+		}
+	})
+}
--- a/internal/deepdoc/parser/pdf/pagespdfiumtest/pages_pdfium_e2e_test.go
+++ b/internal/deepdoc/parser/pdf/pagespdfiumtest/pages_pdfium_e2e_test.go
@@ -0,0 +1,132 @@
+//go:build cgo && manual
+
+// Package pagespdfiumtest exercises Config.Pages against a real PDF via the
+// pdfium CGO engine. It lives in a separate subpackage so that it does not
+// share the build of the parent package's manual-tag test files (some of
+// which do not currently compile on main).
+package pagespdfiumtest
+
+import (
+	"context"
+	"image"
+	"os"
+	"path/filepath"
+	"reflect"
+	"testing"
+
+	deepdocpdf "ragflow/internal/deepdoc/parser/pdf"
+	deepdoctype "ragflow/internal/deepdoc/parser/pdf/type"
+)
+
+// noopDocAnalyzer reports unhealthy and returns empty results, forcing the
+// parser onto the charsToBoxes path so no DeepDoc model service is required.
+type noopDocAnalyzer struct{}
+
+func (noopDocAnalyzer) DLA(context.Context, image.Image) ([]deepdoctype.DLARegion, error) {
+	return nil, nil
+}
+func (noopDocAnalyzer) TSR(context.Context, image.Image) ([]deepdoctype.TSRCell, error) {
+	return nil, nil
+}
+func (noopDocAnalyzer) OCRDetect(context.Context, image.Image) ([]deepdoctype.OCRBox, error) {
+	return nil, nil
+}
+func (noopDocAnalyzer) OCRRecognize(context.Context, image.Image) ([]deepdoctype.OCRText, error) {
+	return nil, nil
+}
+func (noopDocAnalyzer) Health() bool { return false }
+
+// pageKeys extracts the set of processed page numbers from PageHeight.
+func pageKeys(m map[int]float64) map[int]struct{} {
+	out := make(map[int]struct{}, len(m))
+	for k := range m {
+		out[k] = struct{}{}
+	}
+	return out
+}
+
+// TestPagesRealPdf_PdfiumFilter verifies Config.Pages end-to-end against a
+// real multi-page PDF through the pdfium CGO engine (NewEngine ->
+// PageCount/ExtractChars/RenderPage call the pdfium C library). DeepDoc
+// DLA/TSR/OCR are stubbed by noopDocAnalyzer.
+//
+// Run with:
+//
+//	./build.sh --test -tags manual -v -run TestPagesRealPdf_PdfiumFilter ./internal/deepdoc/parser/pdf/pagespdfiumtest/
+func TestPagesRealPdf_PdfiumFilter(t *testing.T) {
+	t.Setenv("BATCH_SKIP_DEEPDOC", "1")
+
+	pdfPath := filepath.Join("..", "testdata", "pdfs", "03_multipage.pdf")
+	data, err := os.ReadFile(pdfPath)
+	if err != nil {
+		t.Fatalf("read %s: %v", pdfPath, err)
+	}
+
+	// NewEngine calls into pdfium (CGO). Failure here means the real C path
+	// was reached but pdfium could not open the document.
+	eng, err := deepdocpdf.NewEngine(data)
+	if err != nil {
+		t.Fatalf("NewEngine (pdfium CGO): %v", err)
+	}
+	defer eng.Close()
+
+	total, err := eng.PageCount()
+	if err != nil {
+		t.Fatalf("eng.PageCount: %v", err)
+	}
+	t.Logf("03_multipage.pdf: pdfium PageCount() = %d (real CGO call)", total)
+	if total < 2 {
+		t.Skipf("need >=2 pages to verify filtering, got %d", total)
+	}
+
+	mock := noopDocAnalyzer{}
+
+	// Group 1: restrict to page 1 (1-indexed) -> only 0-based page 0.
+	t.Run("filter to page 1 only", func(t *testing.T) {
+		cfg := deepdoctype.DefaultParserConfig()
+		cfg.Pages = [][]int{{1, 1}}
+		p := deepdocpdf.NewParser(cfg)
+		result, err := p.ParseRaw(context.Background(), eng, mock)
+		if err != nil {
+			t.Fatalf("ParseRaw: %v", err)
+		}
+		if len(result.PageHeight) != 1 {
+			t.Errorf("expected 1 page parsed, got %d (keys=%v)",
+				len(result.PageHeight), pageKeys(result.PageHeight))
+		}
+		if _, ok := result.PageHeight[0]; !ok {
+			t.Errorf("expected page 0 in PageHeight, got keys %v",
+				pageKeys(result.PageHeight))
+		}
+	})
+
+	// Group 2: control — no Pages restriction -> all pages.
+	t.Run("no pages -> all pages (control)", func(t *testing.T) {
+		p := deepdocpdf.NewParser(deepdoctype.DefaultParserConfig())
+		result, err := p.ParseRaw(context.Background(), eng, mock)
+		if err != nil {
+			t.Fatalf("ParseRaw: %v", err)
+		}
+		if len(result.PageHeight) != total {
+			t.Errorf("expected %d pages (all), got %d (keys=%v)",
+				total, len(result.PageHeight), pageKeys(result.PageHeight))
+		}
+	})
+
+	// Group 3: multi-range — first and last page only.
+	if total >= 3 {
+		t.Run("multi-range first and last page", func(t *testing.T) {
+			cfg := deepdoctype.DefaultParserConfig()
+			cfg.Pages = [][]int{{1, 1}, {total, total}}
+			p := deepdocpdf.NewParser(cfg)
+			result, err := p.ParseRaw(context.Background(), eng, mock)
+			if err != nil {
+				t.Fatalf("ParseRaw: %v", err)
+			}
+			want := map[int]struct{}{0: {}, total - 1: {}}
+			if got := pageKeys(result.PageHeight); !reflect.DeepEqual(got, want) {
+				t.Errorf("PageHeight keys = %v, want %v", got, want)
+			}
+		})
+	}
+}
--- a/internal/deepdoc/parser/pdf/parser.go
+++ b/internal/deepdoc/parser/pdf/parser.go
@@ -6,6 +6,7 @@ import (
 	"image"
 	"log/slog"
 	"math"
+	"sort"
 	"sync"

 	lyt "ragflow/internal/deepdoc/parser/pdf/layout"
@@ -108,6 +109,46 @@ func documentPages(pageCount int) []int {
 	return pages
 }

+// resolvePagesToProcess converts the 1-indexed inclusive Config.Pages ranges
+// into a sorted, de-duplicated slice of 0-indexed page numbers clamped to
+// [0, pageCount-1]. Empty/nil ranges fall back to all pages (the historical
+// behavior), so callers that leave Pages unset are unaffected.
+func resolvePagesToProcess(ranges [][]int, pageCount int) []int {
+	if len(ranges) == 0 {
+		return documentPages(pageCount)
+	}
+	seen := make(map[int]struct{}, pageCount)
+	out := make([]int, 0, pageCount)
+	for _, r := range ranges {
+		if len(r) != 2 {
+			continue
+		}
+		from0 := r[0] - 1
+		to0 := r[1] - 1
+		if from0 < 0 {
+			from0 = 0
+		}
+		if from0 > pageCount-1 {
+			continue
+		}
+		if to0 > pageCount-1 {
+			to0 = pageCount - 1
+		}
+		if to0 < from0 {
+			continue
+		}
+		for pg := from0; pg <= to0; pg++ {
+			if _, dup := seen[pg]; dup {
+				continue
+			}
+			seen[pg] = struct{}{}
+			out = append(out, pg)
+		}
+	}
+	sort.Ints(out)
+	return out
+}
+
 // extractOutlines extracts the PDF outlines, returning nil on error.
 func (p *Parser) extractOutlines(engine pdf.PDFEngine) []pdf.Outline {
 	outlines, outlineErr := engine.Outlines()
@@ -497,7 +538,15 @@ func (p *Parser) processPages(ctx context.Context, engine pdf.PDFEngine, docAnal
 	}

 	tb := NewTableBuilderFor(docAnalyzer)
-	pages := documentPages(pageCount)
+	pages := resolvePagesToProcess(p.Config.Pages, pageCount)
+	if len(p.Config.Pages) > 0 {
+		slog.Info("deepdoc pdf parse: page ranges applied",
+			"configured_ranges", p.Config.Pages,
+			"page_count", pageCount,
+			"pages_to_parse", pages)
+	} else {
+		slog.Debug("deepdoc pdf parse: parsing all pages", "page_count", pageCount)
+	}

 	pageResults, pageErr := p.runPageWorkers(ctx, engine, pages, docAnalyzer, tb)
 	if pageErr != nil {
--- a/internal/deepdoc/parser/pdf/parser_pages_test.go
+++ b/internal/deepdoc/parser/pdf/parser_pages_test.go
@@ -0,0 +1,153 @@
+package pdf
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+	"slices"
+	"strings"
+	"testing"
+
+	pdf "ragflow/internal/deepdoc/parser/pdf/type"
+)
+
+// TestResolvePagesToProcess verifies the 1-indexed inclusive ranges are
+// converted to a sorted, de-duplicated, clamped 0-indexed page list.
+func TestResolvePagesToProcess(t *testing.T) {
+	// allPages returns [0..n-1] for assertion convenience.
+	allPages := func(n int) []int {
+		out := make([]int, n)
+		for i := range out {
+			out[i] = i
+		}
+		return out
+	}
+
+	cases := []struct {
+		name      string
+		ranges    [][]int
+		pageCount int
+		want      []int
+	}{
+		{"nil ranges -> all pages", nil, 10, allPages(10)},
+		{"empty ranges -> all pages", [][]int{}, 10, allPages(10)},
+		{"single range", [][]int{{1, 3}}, 10, []int{0, 1, 2}},
+		{"two disjoint ranges", [][]int{{1, 3}, {8, 10}}, 10, []int{0, 1, 2, 7, 8, 9}},
+		{"unbounded upper clamped to page count", [][]int{{1, 1000000}}, 5, allPages(5)},
+		{"range entirely beyond doc -> empty", [][]int{{8, 10}}, 5, []int{}},
+		{"overlapping ranges de-duplicated", [][]int{{1, 3}, {2, 4}}, 10, []int{0, 1, 2, 3}},
+		{"from > to skipped", [][]int{{3, 1}}, 10, []int{}},
+		{"wrong arity skipped", [][]int{{1}}, 10, []int{}},
+		{"unsorted input sorted", [][]int{{8, 10}, {1, 3}}, 10, []int{0, 1, 2, 7, 8, 9}},
+		{"single page range", [][]int{{5, 5}}, 10, []int{4}},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			got := resolvePagesToProcess(c.ranges, c.pageCount)
+			if !slices.Equal(got, c.want) {
+				t.Errorf("resolvePagesToProcess(%v, %d) = %v, want %v", c.ranges, c.pageCount, got, c.want)
+			}
+		})
+	}
+}
+
+// makePageTaggedEngine builds a MockEngine with `numPages` pages, each carrying
+// a single char whose text is "pN" (N = 0-based page number), so parsed pages
+// can be identified in the resulting sections.
+func makePageTaggedEngine(numPages int) *MockEngine {
+	chars := make(map[int][]pdf.TextChar, numPages)
+	for i := 0; i < numPages; i++ {
+		chars[i] = []pdf.TextChar{
+			{Text: fmt.Sprintf("p%d", i), X0: 10, X1: 50, Top: 10, Bottom: 30, PageNumber: i},
+		}
+	}
+	return &MockEngine{NumPages: numPages, Chars: chars, RenderW: 100, RenderH: 100}
+}
+
+// pageHeightKeys returns the set of page numbers that were actually processed,
+// extracted from result.PageHeight (keyed by 0-based page number).
+func pageHeightKeys(m map[int]float64) map[int]struct{} {
+	out := make(map[int]struct{}, len(m))
+	for k := range m {
+		out[k] = struct{}{}
+	}
+	return out
+}
+
+// TestParseRaw_PageRanges_Applied verifies that Config.Pages restricts parsing
+// to the specified pages (1-indexed inclusive) and that pages outside the
+// ranges are not processed.
+func TestParseRaw_PageRanges_Applied(t *testing.T) {
+	eng := makePageTaggedEngine(10)
+	p := NewParser(pdf.ParserConfig{Pages: [][]int{{1, 3}, {8, 10}}})
+
+	result, err := p.ParseRaw(context.Background(), eng, &MockDocAnalyzer{Healthy: true})
+	if err != nil {
+		t.Fatalf("ParseRaw: %v", err)
+	}
+
+	wantPages := map[int]struct{}{0: {}, 1: {}, 2: {}, 7: {}, 8: {}, 9: {}}
+	if got := pageHeightKeys(result.PageHeight); !reflect.DeepEqual(got, wantPages) {
+		t.Errorf("PageHeight keys = %v, want %v", got, wantPages)
+	}
+
+	// Sections must reference only parsed pages and must not contain text from
+	// skipped pages (p3..p6).
+	combined := combineSectionText(result.Sections)
+	for _, pg := range []int{0, 1, 2, 7, 8, 9} {
+		if !strings.Contains(combined, fmt.Sprintf("p%d", pg)) {
+			t.Errorf("expected section text to contain p%d; combined=%q", pg, combined)
+		}
+	}
+	for _, pg := range []int{3, 4, 5, 6} {
+		if strings.Contains(combined, fmt.Sprintf("p%d", pg)) {
+			t.Errorf("expected section text NOT to contain p%d (page skipped); combined=%q", pg, combined)
+		}
+	}
+}
+
+// TestParseRaw_NoPages_AllPagesParsed is the regression guard: with Pages nil
+// (the default), every page is parsed exactly as before this feature.
+func TestParseRaw_NoPages_AllPagesParsed(t *testing.T) {
+	eng := makePageTaggedEngine(10)
+	p := NewParser(pdf.DefaultParserConfig())
+
+	result, err := p.ParseRaw(context.Background(), eng, &MockDocAnalyzer{Healthy: true})
+	if err != nil {
+		t.Fatalf("ParseRaw: %v", err)
+	}
+
+	wantPages := map[int]struct{}{}
+	for i := 0; i < 10; i++ {
+		wantPages[i] = struct{}{}
+	}
+	if got := pageHeightKeys(result.PageHeight); !reflect.DeepEqual(got, wantPages) {
+		t.Errorf("PageHeight keys = %v, want all 10 pages %v", got, wantPages)
+	}
+}
+
+// TestParseRaw_PageRanges_BeyondDoc verifies that a range fully beyond the
+// document page count yields an empty parse (no pages processed).
+func TestParseRaw_PageRanges_BeyondDoc(t *testing.T) {
+	eng := makePageTaggedEngine(5)
+	p := NewParser(pdf.ParserConfig{Pages: [][]int{{8, 10}}})
+
+	result, err := p.ParseRaw(context.Background(), eng, &MockDocAnalyzer{Healthy: true})
+	if err != nil {
+		t.Fatalf("ParseRaw: %v", err)
+	}
+	if len(result.PageHeight) != 0 {
+		t.Errorf("expected 0 pages parsed for fully out-of-range pages, got %d (%v)",
+			len(result.PageHeight), pageHeightKeys(result.PageHeight))
+	}
+}
+
+// combineSectionText concatenates all section text for content assertions.
+func combineSectionText(sections []pdf.Section) string {
+	var b strings.Builder
+	for _, s := range sections {
+		b.WriteString(s.Text)
+	}
+	return b.String()
+}
--- a/internal/deepdoc/parser/pdf/parser_pipeline_integration_test.go
+++ b/internal/deepdoc/parser/pdf/parser_pipeline_integration_test.go
@@ -721,10 +721,3 @@ func TestE2E_ParseAndPostProcess(t *testing.T) {
 	figs := result.Figures()
 	t.Logf("figures: %d", len(figs))
 }
-
-func truncate(s string, n int) string {
-	if len(s) <= n {
-		return s
-	}
-	return s[:n] + "..."
-}
--- a/internal/deepdoc/parser/pdf/pipeline_parity_test.go
+++ b/internal/deepdoc/parser/pdf/pipeline_parity_test.go
@@ -10,6 +10,7 @@ import (
 	"strings"
 	"testing"

+	"ragflow/internal/common"
 	lyt "ragflow/internal/deepdoc/parser/pdf/layout"
 	"ragflow/internal/deepdoc/parser/pdf/tool"
 	pdf "ragflow/internal/deepdoc/parser/pdf/type"
--- a/internal/deepdoc/parser/pdf/table_rotate_integration_test.go
+++ b/internal/deepdoc/parser/pdf/table_rotate_integration_test.go
@@ -6,6 +6,7 @@ import (
 	"context"
 	"os"
 	"path/filepath"
+	"ragflow/internal/common"
 	inf "ragflow/internal/deepdoc/parser/pdf/inference"
 	tbl "ragflow/internal/deepdoc/parser/pdf/table"
 	pdf "ragflow/internal/deepdoc/parser/pdf/type"
--- a/internal/deepdoc/parser/pdf/text_dump_test.go
+++ b/internal/deepdoc/parser/pdf/text_dump_test.go
@@ -6,6 +6,7 @@ import (
 	"context"
 	"os"
 	"path/filepath"
+	"ragflow/internal/common"
 	pdf "ragflow/internal/deepdoc/parser/pdf/type"
 	"strings"
 	"testing"
--- a/internal/deepdoc/parser/type/types.go
+++ b/internal/deepdoc/parser/type/types.go
@@ -213,6 +213,10 @@ type ParserConfig struct {
 	SeparateTablesFigs bool
 	SortByTop          bool
 	SkipOCR            bool
+	// Pages restricts parsing to these 1-indexed inclusive page ranges.
+	// nil/empty means parse all pages. Ranges beyond the document are clamped
+	// at parse time; fully out-of-range ranges are skipped.
+	Pages [][]int
 }

 // DefaultParserConfig returns a ParserConfig with sensible defaults.
--- a/internal/entity/dataset.go
+++ b/internal/entity/dataset.go
@@ -58,7 +58,6 @@ const (
 	ParserTypeOne          ParserType = "one"
 	ParserTypeAudio        ParserType = "audio"
 	ParserTypeEmail        ParserType = "email"
-	ParserTypeTag          ParserType = "tag"
 	ParserTypeKG           ParserType = "knowledge_graph"
 )

--- a/internal/ingestion/component/tokenizer_test.go
+++ b/internal/ingestion/component/tokenizer_test.go
@@ -387,7 +387,7 @@ type countMismatchedEmbedder struct{ want int }

 func (c *countMismatchedEmbedder) MaxTokens() int { return 0 }

-func (c *countMismatchedEmbedder) Encode(texts []string) ([]EmbeddingResult, error) {
+func (c *countMismatchedEmbedder) Encode(ctx context.Context, texts []string) ([]EmbeddingResult, error) {
 	out := make([]EmbeddingResult, c.want)
 	for i := range out {
 		out[i] = EmbeddingResult{Vector: make([]float64, 4), TokenCount: 1}
--- a/internal/ingestion/pipeline/builtin_registry_test.go
+++ b/internal/ingestion/pipeline/builtin_registry_test.go
@@ -90,7 +90,6 @@ func TestRegistryVsHardcodedList(t *testing.T) {
 		"qa":           true,
 		"resume":       true,
 		"table":        true,
-		"tag":          true,
 	}
 	for h := range hardcoded {
 		t.Logf("Hardcoded has: %s", h)
--- a/internal/ingestion/pipeline/pdf_pages.go
+++ b/internal/ingestion/pipeline/pdf_pages.go
@@ -0,0 +1,49 @@
+package pipeline
+
+import "ragflow/internal/utility"
+
+// NormalizeParserConfigPages walks cfg and normalizes the "pages" field under
+// every component's filetype setup under fail-fast semantics.
+//
+// When normalization succeeds, the normalized value is written back. When any
+// "pages" value is invalid (non-list, or contains an invalid range), an error
+// is returned immediately and the request should be rejected — no partial
+// dropping, no保留 of malformed values.
+//
+// The walk is generic (not hardcoded to "pdf"): any filetype setup carrying a
+// "pages" key is normalized. Setups without "pages" are skipped, so this is
+// safe to run on any parser_config and never creates or deletes keys other
+// than overwriting "pages" in place. Structural mismatches (non-map cpnID
+// value, non-map setup) are silently skipped — they are not "pages" errors.
+func NormalizeParserConfigPages(cfg map[string]any) error {
+	if cfg == nil {
+		return nil
+	}
+	for _, v := range cfg {
+		params, ok := v.(map[string]any)
+		if !ok {
+			continue
+		}
+		for _, fv := range params {
+			setup, ok := fv.(map[string]any)
+			if !ok {
+				continue
+			}
+			raw, ok := setup["pages"]
+			if !ok {
+				continue
+			}
+			normalized, err := utility.NormalizePDFPages(raw)
+			if err != nil {
+				return err
+			}
+			if normalized != nil {
+				setup["pages"] = normalized
+			}
+			// nil (no value): leave the key as-is. Empty/null pages are
+			// equivalent to "parse all pages"; overwriting with nil would
+			// not change behavior but would mutate the map unnecessarily.
+		}
+	}
+	return nil
+}
--- a/internal/ingestion/pipeline/pdf_pages_test.go
+++ b/internal/ingestion/pipeline/pdf_pages_test.go
@@ -0,0 +1,161 @@
+package pipeline
+
+import (
+	"reflect"
+	"testing"
+)
+
+// TestNormalizeParserConfigPages verifies the walker normalizes the "pages"
+// field under every component's filetype setup under fail-fast semantics:
+// valid ranges are written back normalized; any invalid range returns an
+// error and rejects the whole config.
+func TestNormalizeParserConfigPages(t *testing.T) {
+	// f64 builds a JSON-decoded-style []any of []any of float64.
+	f64 := func(ranges ...[2]float64) any {
+		out := make([]any, 0, len(ranges))
+		for _, r := range ranges {
+			out = append(out, []any{r[0], r[1]})
+		}
+		return out
+	}
+
+	t.Run("overlap merged and written back", func(t *testing.T) {
+		cfg := map[string]any{
+			"Parser:X": map[string]any{
+				"pdf": map[string]any{"pages": f64([2]float64{1, 200}, [2]float64{111, 333})},
+			},
+		}
+		if err := NormalizeParserConfigPages(cfg); err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+		pdf := cfg["Parser:X"].(map[string]any)["pdf"].(map[string]any)
+		if want := [][]int{{1, 333}}; !reflect.DeepEqual(pdf["pages"], want) {
+			t.Errorf("pages = %v, want %v", pdf["pages"], want)
+		}
+	})
+
+	t.Run("unsorted sorted and written back", func(t *testing.T) {
+		cfg := map[string]any{
+			"Parser:X": map[string]any{
+				"pdf": map[string]any{"pages": f64([2]float64{400, 500}, [2]float64{1, 100})},
+			},
+		}
+		if err := NormalizeParserConfigPages(cfg); err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+		pdf := cfg["Parser:X"].(map[string]any)["pdf"].(map[string]any)
+		if want := [][]int{{1, 100}, {400, 500}}; !reflect.DeepEqual(pdf["pages"], want) {
+			t.Errorf("pages = %v, want %v", pdf["pages"], want)
+		}
+	})
+
+	t.Run("all invalid -> error", func(t *testing.T) {
+		cfg := map[string]any{
+			"Parser:X": map[string]any{
+				"pdf": map[string]any{"pages": f64([2]float64{3, 1})},
+			},
+		}
+		err := NormalizeParserConfigPages(cfg)
+		if err == nil {
+			t.Fatalf("expected error for all-invalid pages, got nil")
+		}
+	})
+
+	t.Run("non-list pages -> error", func(t *testing.T) {
+		cfg := map[string]any{
+			"Parser:X": map[string]any{
+				"pdf": map[string]any{"pages": "1-100"},
+			},
+		}
+		err := NormalizeParserConfigPages(cfg)
+		if err == nil {
+			t.Fatalf("expected error for non-list pages, got nil")
+		}
+	})
+
+	t.Run("no pages key -> unchanged", func(t *testing.T) {
+		cfg := map[string]any{
+			"Parser:X": map[string]any{
+				"pdf": map[string]any{"output_format": "json"},
+			},
+		}
+		if err := NormalizeParserConfigPages(cfg); err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+		pdf := cfg["Parser:X"].(map[string]any)["pdf"].(map[string]any)
+		if _, ok := pdf["pages"]; ok {
+			t.Error("pages key should not exist")
+		}
+		if pdf["output_format"] != "json" {
+			t.Error("other fields should not be touched")
+		}
+	})
+
+	t.Run("non-pdf filetype also normalized (generic walk)", func(t *testing.T) {
+		cfg := map[string]any{
+			"Parser:X": map[string]any{
+				"docx": map[string]any{"pages": f64([2]float64{1, 100})},
+			},
+		}
+		if err := NormalizeParserConfigPages(cfg); err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+		docx := cfg["Parser:X"].(map[string]any)["docx"].(map[string]any)
+		if want := [][]int{{1, 100}}; !reflect.DeepEqual(docx["pages"], want) {
+			t.Errorf("docx.pages = %v, want %v", docx["pages"], want)
+		}
+	})
+
+	t.Run("non-map cpnID value -> unchanged", func(t *testing.T) {
+		cfg := map[string]any{"Parser:X": "not a map"}
+		if err := NormalizeParserConfigPages(cfg); err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+		if cfg["Parser:X"] != "not a map" {
+			t.Errorf("non-map cpnID value should not be touched, got %v", cfg["Parser:X"])
+		}
+	})
+
+	t.Run("multiple cpnIDs each normalized", func(t *testing.T) {
+		cfg := map[string]any{
+			"Parser:A": map[string]any{
+				"pdf": map[string]any{"pages": f64([2]float64{1, 200}, [2]float64{111, 333})},
+			},
+			"Parser:B": map[string]any{
+				"pdf": map[string]any{"pages": f64([2]float64{400, 500}, [2]float64{1, 100})},
+			},
+		}
+		if err := NormalizeParserConfigPages(cfg); err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+		pdfA := cfg["Parser:A"].(map[string]any)["pdf"].(map[string]any)
+		pdfB := cfg["Parser:B"].(map[string]any)["pdf"].(map[string]any)
+		if want := [][]int{{1, 333}}; !reflect.DeepEqual(pdfA["pages"], want) {
+			t.Errorf("Parser:A pages = %v, want %v", pdfA["pages"], want)
+		}
+		if want := [][]int{{1, 100}, {400, 500}}; !reflect.DeepEqual(pdfB["pages"], want) {
+			t.Errorf("Parser:B pages = %v, want %v", pdfB["pages"], want)
+		}
+	})
+
+	t.Run("invalid pages in one cpnID -> error", func(t *testing.T) {
+		cfg := map[string]any{
+			"Parser:A": map[string]any{
+				"pdf": map[string]any{"pages": f64([2]float64{1, 100})},
+			},
+			"Parser:B": map[string]any{
+				"pdf": map[string]any{"pages": f64([2]float64{0, 5})},
+			},
+		}
+		err := NormalizeParserConfigPages(cfg)
+		if err == nil {
+			t.Fatalf("expected error for invalid pages in Parser:B, got nil")
+		}
+	})
+
+	t.Run("nil cfg -> no error", func(t *testing.T) {
+		if err := NormalizeParserConfigPages(nil); err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+	})
+}
--- a/internal/ingestion/pipeline/template/ingestion_pipeline_general.json
+++ b/internal/ingestion/pipeline/template/ingestion_pipeline_general.json
@@ -102,7 +102,13 @@
                            "suffix": [
                                "pdf"
                            ],
-                            "vlm": {}
+                            "vlm": {},
+                            "pages": [
+                                [
+                                    1,
+                                    100000
+                                ]
+                            ]
                        },
                        "spreadsheet": {
                            "flatten_media_to_text": false,
--- a/internal/ingestion/pipeline/template/ingestion_pipeline_tag.json
+++ b/internal/ingestion/pipeline/template/ingestion_pipeline_tag.json
--- a/internal/ingestion/task/chunk_process.go
+++ b/internal/ingestion/task/chunk_process.go
@@ -91,16 +91,10 @@ func ProcessChunksForPipeline(
 		metadata = mergeChunkMetadata(metadata, ck)
 		RenameTextToContentWithWeight(ck)
 		processChunkPositions(ck)
-		removeInternalChunkFields(ck)
 	}
 	return metadata, nil
 }

-func removeInternalChunkFields(ck map[string]any) {
-	delete(ck, "_pdf_positions")
-	delete(ck, "image")
-}
-
 // cleanupConsumedChunkFields materializes the stored array form of the
 // questions/keywords fields (when the upstream Tokenizer did not already set
 // them) and strips the consumed source fields (questions/keywords/summary)
@@ -139,7 +133,11 @@ func mergeChunkMetadata(metadata map[string]any, ck map[string]any) map[string]a

 // processChunkPositions converts the raw "positions" field into indexable
 // position fields (page_num_int, top_int, position_int) via AddPositions,
-// then removes the raw field.
+// then removes the raw "positions" and the sibling "_pdf_positions" internal
+// field. _pdf_positions is the parser-emitted position matrix consumed by
+// the chunker's image-crop pass (pdfcrop_cgo.go); after the chunker it is
+// dead weight with no index column, so it is pruned unconditionally —
+// independent of "positions" and not gated on the early-return below.
 //
 // Two source types reach this point:
 //   - []float64 — flat array of 5-tuples [page,left,right,top,bottom,…] from
@@ -151,6 +149,7 @@ func mergeChunkMetadata(metadata map[string]any, ck map[string]any) map[string]a
 // Both are flattened into a single []float64 for AddPositions, which groups
 // by 5 internally. Unexpected types are logged and discarded.
 func processChunkPositions(ck map[string]any) {
+	delete(ck, "_pdf_positions")
 	poss, exists := ck["positions"]
 	if !exists {
 		return
--- a/internal/ingestion/task/chunk_process_test.go
+++ b/internal/ingestion/task/chunk_process_test.go
@@ -126,11 +126,16 @@ func TestProcessChunksForPipeline_GeneratesIDOnNonStringText(t *testing.T) {
 	}
 }

+// TestProcessChunksForPipeline_RemovesInternalPipelineFields pins that
+// processChunkPositions prunes the _pdf_positions internal field (the
+// parser-emitted position matrix) before indexing. The "image" field is
+// no longer dropped here — its lifecycle is owned by the chunker's
+// imageUploadDecorator (register.go + image_upload.go), which uploads and
+// deletes it at the chunker stage.
 func TestProcessChunksForPipeline_RemovesInternalPipelineFields(t *testing.T) {
 	chunks := []map[string]any{{
 		"text":           "hello",
 		"_pdf_positions": []any{[]any{0, 1, 2, 3, 4}},
-		"image":          "data:image/png;base64,abc",
 	}}

 	_, err := ProcessChunksForPipeline(chunks, "doc-1", "kb-1", "test-doc.pdf", time.Now())
@@ -140,9 +145,6 @@ func TestProcessChunksForPipeline_RemovesInternalPipelineFields(t *testing.T) {
 	if _, exists := chunks[0]["_pdf_positions"]; exists {
 		t.Fatalf("_pdf_positions should be removed before indexing: %v", chunks[0]["_pdf_positions"])
 	}
-	if _, exists := chunks[0]["image"]; exists {
-		t.Fatalf("image should be removed before indexing: %v", chunks[0]["image"])
-	}
 }

 func TestProcessChunksForPipeline_PreservesExistingID(t *testing.T) {
@@ -330,9 +332,17 @@ func TestProcessChunkPositions_2DFloat64(t *testing.T) {
 }

 func TestProcessChunkPositions_NoPositions(t *testing.T) {
-	chunk := map[string]any{"text": "hello"}
+	// _pdf_positions is pruned unconditionally, even on the early-return path
+	// where "positions" is absent, since the two fields are independent.
+	chunk := map[string]any{
+		"text":           "hello",
+		"_pdf_positions": []any{[]any{0, 1, 2, 3, 4}},
+	}
 	processChunkPositions(chunk)
 	if _, exists := chunk["page_num_int"]; exists {
 		t.Error("page_num_int must not be set when positions is missing")
 	}
+	if _, exists := chunk["_pdf_positions"]; exists {
+		t.Error("_pdf_positions must be pruned even when positions is missing")
+	}
 }
--- a/internal/ingestion/task/pipeline_executor_defaults_test.go
+++ b/internal/ingestion/task/pipeline_executor_defaults_test.go
--- a/internal/parser/parser/pdf_parser_cgo.go
+++ b/internal/parser/parser/pdf_parser_cgo.go
@@ -35,6 +35,7 @@ func (p *PDFParser) ParseWithResult(ctx context.Context, filename string, data [
 	}
 	cfg := deepdoctype.DefaultParserConfig()
 	cfg.SkipOCR = false
+	cfg.Pages = p.Pages
 	parser := deepdocpdf.NewParser(cfg)
 	res := parsePDFWithDeepDocOptions(ctx, filename, data, pdfPostProcessOptions{
 		outputFormat:       p.OutputFormat,
--- a/internal/parser/parser/pdf_parser_common.go
+++ b/internal/parser/parser/pdf_parser_common.go
@@ -33,6 +33,7 @@ import (
 	pdflayout "ragflow/internal/deepdoc/parser/pdf/layout"
 	"ragflow/internal/deepdoc/parser/pdf/util"
 	deepdoctype "ragflow/internal/deepdoc/parser/type"
+	"ragflow/internal/utility"
 )

 // ErrPDFEngineUnavailable is returned by PDFParser.ParseWithResult
@@ -58,12 +59,16 @@ type PDFParser struct {
 	Model      string // DeepDoc@buildin@ragflow
 	LibType    string // pdf_oxide, used by DeepDoc

-	FlattenMediaToText                bool
-	RemoveTOC                         bool
-	RemoveHeaderFooter                bool
-	EnableMultiColumn                 bool
-	OutputFormat                      string
-	ParseMethod                       string
+	FlattenMediaToText bool
+	RemoveTOC          bool
+	RemoveHeaderFooter bool
+	EnableMultiColumn  bool
+	OutputFormat       string
+	ParseMethod        string
+	// Pages restricts parsing to these 1-indexed inclusive page ranges.
+	// nil/empty means parse all pages. Populated by ConfigureFromSetup from
+	// the filetype setup map and forwarded to the deepdoc ParserConfig.
+	Pages                             [][]int
 	MinerUAPIServer                   string
 	MinerUAPIKey                      string
 	MinerUBackend                     string
@@ -250,6 +255,18 @@ func (p *PDFParser) ConfigureFromSetup(setup map[string]any) {
 	if v, ok := setup["markdown_image_response_type"].(string); ok && v != "" {
 		p.TCADPMarkdownImageResponseType = v
 	}
+	if raw, ok := setup["pages"]; ok {
+		// Request-layer validation (NormalizeParserConfigPages) already
+		// rejects invalid ranges at the API boundary. At parse time the input
+		// should already be normalized; degrade to "parse all pages" rather
+		// than failing the parse if an unexpected shape slips through.
+		if pages, err := utility.NormalizePDFPages(raw); err != nil {
+			slog.Warn("ConfigureFromSetup: invalid pages range, falling back to all pages",
+				"raw", raw, "err", err)
+		} else {
+			p.Pages = pages
+		}
+	}
 }

 func normalizePDFParseMethod(raw string) string {
--- a/internal/parser/parser/pdf_parser_pages_e2e_test.go
+++ b/internal/parser/parser/pdf_parser_pages_e2e_test.go
@@ -0,0 +1,85 @@
+package parser
+
+import (
+	"context"
+	"fmt"
+	"image"
+	"reflect"
+	"testing"
+
+	deepdocpdf "ragflow/internal/deepdoc/parser/pdf"
+	deepdoctype "ragflow/internal/deepdoc/parser/pdf/type"
+)
+
+// noopDocAnalyzer is a DocAnalyzer that reports unhealthy and returns empty
+// results, forcing the parser onto the charsToBoxes path. It lets the parser
+// package exercise the deepdoc parser without the real DeepDoc models.
+type noopDocAnalyzer struct{}
+
+func (noopDocAnalyzer) DLA(context.Context, image.Image) ([]deepdoctype.DLARegion, error) {
+	return nil, nil
+}
+func (noopDocAnalyzer) TSR(context.Context, image.Image) ([]deepdoctype.TSRCell, error) {
+	return nil, nil
+}
+func (noopDocAnalyzer) OCRDetect(context.Context, image.Image) ([]deepdoctype.OCRBox, error) {
+	return nil, nil
+}
+func (noopDocAnalyzer) OCRRecognize(context.Context, image.Image) ([]deepdoctype.OCRText, error) {
+	return nil, nil
+}
+func (noopDocAnalyzer) Health() bool { return false }
+
+// TestPDFParser_PagesEndToEnd_FromConfigureFromSetup verifies the full path
+// from a filetype setup map (as delivered by the pipeline override_params)
+// through ConfigureFromSetup -> PDFParser.Pages -> deepdoc ParserConfig.Pages
+// -> resolvePagesToProcess, asserting only the configured page ranges are
+// parsed.
+//
+// This ties together step 2 (ConfigureFromSetup + cfg.Pages plumbing) and
+// step 1 (deepdoc page filtering) at the PDFParser level.
+func TestPDFParser_PagesEndToEnd_FromConfigureFromSetup(t *testing.T) {
+	// Build a 10-page mock engine where page N carries the text "pN".
+	chars := make(map[int][]deepdoctype.TextChar, 10)
+	for i := 0; i < 10; i++ {
+		chars[i] = []deepdoctype.TextChar{
+			{Text: fmt.Sprintf("p%d", i), X0: 10, X1: 50, Top: 10, Bottom: 30, PageNumber: i},
+		}
+	}
+	eng := &deepdocpdf.MockEngine{NumPages: 10, Chars: chars, RenderW: 100, RenderH: 100}
+
+	// 1. ConfigureFromSetup reads "pages" exactly as the pipeline would pass
+	//    it (JSON-decoded []any of []any of float64).
+	p := &PDFParser{}
+	p.ConfigureFromSetup(map[string]any{
+		"pages": []any{
+			[]any{float64(1), float64(3)},
+			[]any{float64(8), float64(10)},
+		},
+	})
+	wantPages := [][]int{{1, 3}, {8, 10}}
+	if !reflect.DeepEqual(p.Pages, wantPages) {
+		t.Fatalf("PDFParser.Pages = %v, want %v", p.Pages, wantPages)
+	}
+
+	// 2. Build the deepdoc config the same way ParseWithResult does
+	//    (cfg.Pages = p.Pages) and run the parser.
+	cfg := deepdoctype.DefaultParserConfig()
+	cfg.Pages = p.Pages
+	docParser := deepdocpdf.NewParser(cfg)
+
+	result, err := docParser.ParseRaw(context.Background(), eng, noopDocAnalyzer{})
+	if err != nil {
+		t.Fatalf("ParseRaw: %v", err)
+	}
+
+	// 3. [1,3] -> 0-based 0..2 ; [8,10] -> 7..9
+	gotPages := make(map[int]struct{}, len(result.PageHeight))
+	for k := range result.PageHeight {
+		gotPages[k] = struct{}{}
+	}
+	want := map[int]struct{}{0: {}, 1: {}, 2: {}, 7: {}, 8: {}, 9: {}}
+	if !reflect.DeepEqual(gotPages, want) {
+		t.Errorf("PageHeight keys = %v, want %v", gotPages, want)
+	}
+}
--- a/internal/parser/parser/pdf_parser_pages_test.go
+++ b/internal/parser/parser/pdf_parser_pages_test.go
@@ -0,0 +1,87 @@
+package parser
+
+import (
+	"reflect"
+	"testing"
+)
+
+// TestConfigureFromSetup_Pages verifies ConfigureFromSetup reads the "pages"
+// field from the filetype setup map, normalizes it, and assigns it to
+// PDFParser.Pages.
+func TestConfigureFromSetup_Pages(t *testing.T) {
+	t.Run("reads and normalizes pages", func(t *testing.T) {
+		p := &PDFParser{}
+		setup := map[string]any{
+			"pages": []any{
+				[]any{float64(1), float64(3)},
+				[]any{float64(8), float64(10)},
+			},
+		}
+		p.ConfigureFromSetup(setup)
+		want := [][]int{{1, 3}, {8, 10}}
+		if !reflect.DeepEqual(p.Pages, want) {
+			t.Errorf("Pages = %v, want %v", p.Pages, want)
+		}
+	})
+
+	t.Run("overlapping ranges merged", func(t *testing.T) {
+		p := &PDFParser{}
+		setup := map[string]any{
+			"pages": []any{
+				[]any{float64(1), float64(200)},
+				[]any{float64(111), float64(333)},
+			},
+		}
+		p.ConfigureFromSetup(setup)
+		want := [][]int{{1, 333}}
+		if !reflect.DeepEqual(p.Pages, want) {
+			t.Errorf("Pages = %v, want %v", p.Pages, want)
+		}
+	})
+
+	t.Run("all invalid -> nil", func(t *testing.T) {
+		p := &PDFParser{}
+		setup := map[string]any{
+			"pages": []any{[]any{float64(3), float64(1)}},
+		}
+		p.ConfigureFromSetup(setup)
+		if p.Pages != nil {
+			t.Errorf("Pages = %v, want nil", p.Pages)
+		}
+	})
+
+	t.Run("missing pages key -> nil", func(t *testing.T) {
+		p := &PDFParser{}
+		setup := map[string]any{"flatten_media_to_text": true}
+		p.ConfigureFromSetup(setup)
+		if p.Pages != nil {
+			t.Errorf("Pages = %v, want nil", p.Pages)
+		}
+	})
+
+	// Regression guard: reading pages must not break other fields.
+	t.Run("other fields still read (no regression)", func(t *testing.T) {
+		p := &PDFParser{}
+		setup := map[string]any{
+			"flatten_media_to_text": true,
+			"parse_method":          "DeepDOC",
+			"output_format":         "json",
+			"pages": []any{
+				[]any{float64(1), float64(100)},
+			},
+		}
+		p.ConfigureFromSetup(setup)
+		if !p.FlattenMediaToText {
+			t.Error("FlattenMediaToText not read")
+		}
+		if p.ParseMethod != "DeepDOC" {
+			t.Errorf("ParseMethod = %q, want DeepDOC", p.ParseMethod)
+		}
+		if p.OutputFormat != "json" {
+			t.Errorf("OutputFormat = %q, want json", p.OutputFormat)
+		}
+		if want := [][]int{{1, 100}}; !reflect.DeepEqual(p.Pages, want) {
+			t.Errorf("Pages = %v, want %v", p.Pages, want)
+		}
+	})
+}
--- a/internal/service/dataset/create_test.go
+++ b/internal/service/dataset/create_test.go
@@ -44,9 +44,11 @@ func TestCreateDataset_NoComponentParams(t *testing.T) {
 	insertCreateDatasetTenant(t, "tenant-1")

 	chunkMethod := "naive"
+	parseType := 1
 	result, code, err := testDatasetCreateService(t).CreateDataset(&service.CreateDatasetRequest{
-		Name:     "ds-no-cp",
-		ParserID: &chunkMethod,
+		Name:      "ds-no-cp",
+		ParserID:  &chunkMethod,
+		ParseType: &parseType,
 	}, "tenant-1")
 	if err != nil {
 		t.Fatalf("CreateDataset failed: %v", err)
@@ -65,9 +67,11 @@ func TestCreateDataset_ComponentParamsPopulated(t *testing.T) {
 	insertCreateDatasetTenant(t, "tenant-1")

 	chunkMethod := "general"
+	parseType := 1
 	result, code, err := testDatasetCreateService(t).CreateDataset(&service.CreateDatasetRequest{
-		Name:     "ds-with-cp",
-		ParserID: &chunkMethod,
+		Name:      "ds-with-cp",
+		ParserID:  &chunkMethod,
+		ParseType: &parseType,
 	}, "tenant-1")
 	if err != nil {
 		t.Fatalf("CreateDataset failed: %v", err)
@@ -218,23 +222,3 @@ func TestCreateDataset_RejectsInvalidEmbeddingModel(t *testing.T) {
 		})
 	}
 }
-
-func TestCreateDataset_RejectsBothWithoutParseType(t *testing.T) {
-	db := setupServiceTestDB(t)
-	pushServiceDB(t, db)
-	insertCreateDatasetTenant(t, "tenant-1")
-
-	pipelineID := "0123456789abcdef0123456789abcdef"
-	chunkMethod := "naive"
-	_, code, err := testDatasetCreateService(t).CreateDataset(&service.CreateDatasetRequest{
-		Name:       "ds-both",
-		ParserID:   &chunkMethod,
-		PipelineID: &pipelineID,
-	}, "tenant-1")
-	if err == nil {
-		t.Fatal("expected error when both parser_id and pipeline_id are provided without parse_type")
-	}
-	if code != common.CodeDataError {
-		t.Fatalf("expected CodeDataError, got %d", code)
-	}
-}
--- a/internal/service/dataset/crud.go
+++ b/internal/service/dataset/crud.go
@@ -34,18 +34,17 @@ func (d *DatasetService) CreateDataset(req *service.CreateDatasetRequest, tenant
 		return nil, common.CodeDataError, errors.New("Tenant not found.")
 	}

-	isPipelineMode := req.ParseType != nil && *req.ParseType == 2
-	isBuiltinMode := req.ParseType != nil && *req.ParseType == 1
-
-	if isBuiltinMode && req.PipelineID != nil {
-		req.PipelineID = nil
-	}
-	if isPipelineMode && req.ParserID != nil {
-		req.ParserID = nil
-	}
-
-	if req.ParseType == nil && req.ParserID != nil && req.PipelineID != nil {
-		return nil, common.CodeDataError, errors.New("parser_id and pipeline_id are mutually exclusive")
+	if req.ParserID != nil || req.PipelineID != nil || req.ParseType != nil {
+		isBuiltin, isPipeline, err := service.ValidateParseTypeMode(req.ParseType, req.ParserID, req.PipelineID)
+		if err != nil {
+			return nil, common.CodeDataError, err
+		}
+		if isBuiltin && req.PipelineID != nil {
+			req.PipelineID = nil
+		}
+		if isPipeline && req.ParserID != nil {
+			req.ParserID = nil
+		}
 	}

 	parserID := ""
--- a/internal/service/dataset/helpers_test.go
+++ b/internal/service/dataset/helpers_test.go
@@ -10,7 +10,7 @@ import (
 // TestValidateParserID_AcceptsRegistryRefs verifies that every
 // canonical builtin pipeline id passes validation.
 func TestValidateParserID_AcceptsRegistryRefs(t *testing.T) {
-	for _, id := range []string{"general", "book", "audio", "qa", "table", "tag"} {
+	for _, id := range []string{"general", "book", "audio", "qa", "table"} {
 		if err := validateParserID(id); err != nil {
 			t.Errorf("validateParserID(%q) = %v, want nil", id, err)
 		}
--- a/internal/service/dataset/update.go
+++ b/internal/service/dataset/update.go
@@ -98,14 +98,17 @@ func (d *DatasetService) UpdateDataset(datasetID, tenantID string, req service.U
 		simpleUpdates["permission"] = permission
 	}

-	isPipelineMode := req.ParseType != nil && *req.ParseType == 2
-	isBuiltinMode := req.ParseType != nil && *req.ParseType == 1
-
-	if isBuiltinMode && req.PipelineID != nil {
-		req.PipelineID = nil
-	}
-	if isPipelineMode && req.ParserID != nil {
-		req.ParserID = nil
+	if req.ParserID != nil || req.PipelineID != nil || req.ParseType != nil {
+		isBuiltin, isPipeline, modeErr := service.ValidateParseTypeMode(req.ParseType, req.ParserID, req.PipelineID)
+		if modeErr != nil {
+			return nil, common.CodeDataError, modeErr
+		}
+		if isBuiltin && req.PipelineID != nil {
+			req.PipelineID = nil
+		}
+		if isPipeline && req.ParserID != nil {
+			req.ParserID = nil
+		}
 	}

 	var pipelineID *string
@@ -122,10 +125,6 @@ func (d *DatasetService) UpdateDataset(datasetID, tenantID string, req service.U
 		return nil, common.CodeDataError, err
 	}

-	if req.ParseType == nil && parserIDProvided && req.PipelineID != nil {
-		return nil, common.CodeDataError, errors.New("parser_id and pipeline_id are mutually exclusive")
-	}
-
 	embdID, embdIDProvided, err := datasetUpdateEmbeddingID(req)
 	if err != nil {
 		return nil, common.CodeDataError, err
@@ -135,6 +134,9 @@ func (d *DatasetService) UpdateDataset(datasetID, tenantID string, req service.U
 		if err := validateDatasetParserConfigSize(req.ParserConfig); err != nil {
 			return nil, common.CodeDataError, err
 		}
+		if err := pipelinepkg.NormalizeParserConfigPages(map[string]any(req.ParserConfig)); err != nil {
+			return nil, common.CodeDataError, err
+		}
 	}

 	var requestedPagerank int64
@@ -222,21 +224,17 @@ func (d *DatasetService) UpdateDataset(datasetID, tenantID string, req service.U
 		}

 		if req.ParserConfig != nil && len(req.ParserConfig) > 0 {
-			effectiveParserID := lockedKB.ParserID
-			if parserIDProvided {
-				effectiveParserID = parserID
-			}
-			effectivePipelineID := lockedKB.PipelineID
-			if pipelineID != nil {
-				effectivePipelineID = pipelineID
-			} else if parserIDProvided && lockedKB.PipelineID != nil {
-				effectivePipelineID = nil
-			}
-			isCanvas := effectivePipelineID != nil && strings.TrimSpace(*effectivePipelineID) != ""
-			dslJSON, dslErr := service.LoadPipelineDSL(isCanvas, effectiveParserID, effectivePipelineID)
+			// Resolve effective mode/IDs once via the shared helper. parse_type
+			// is authoritative; the per-mode req IDs were already cleaned above,
+			// but ResolveParseMode does not rely on that — it ignores the
+			// non-applicable ID for the selected mode.
+			isPipeline, effParserID, effPipelineID := service.ResolveParseMode(
+				req.ParseType, req.ParserID, req.PipelineID,
+				service.ParseModeState{ParserID: lockedKB.ParserID, PipelineID: lockedKB.PipelineID})
+			dslJSON, dslErr := service.LoadPipelineDSL(isPipeline, effParserID, effPipelineID)
 			if dslErr != nil {
 				common.Warn("failed to load pipeline DSL for building parser_config",
-					zap.String("parserID", effectiveParserID), zap.Error(dslErr))
+					zap.String("parserID", effParserID), zap.Error(dslErr))
 			}
 			if dslJSON != nil {
 				updates["parser_config"] = pipelinepkg.BuildParserConfig(dslJSON, map[string]interface{}(req.ParserConfig))
--- a/internal/service/dataset/update_test.go
+++ b/internal/service/dataset/update_test.go
@@ -41,6 +41,7 @@ func TestDatasetServiceUpdateDatasetUpdatesFields(t *testing.T) {
 	permission := string(entity.TenantPermissionTeam)
 	chunkMethod := string(entity.ParserTypeBook)
 	embeddingModel := "BAAI/bge-large-zh-v1.5@Builtin"
+	parseType := 1

 	result, code, err := testDatasetUpdateService(t).UpdateDataset("kb-1", "tenant-1", service.UpdateDatasetRequest{
 		Name:           &name,
@@ -48,6 +49,7 @@ func TestDatasetServiceUpdateDatasetUpdatesFields(t *testing.T) {
 		Language:       &language,
 		Permission:     &permission,
 		ParserID:       &chunkMethod,
+		ParseType:      &parseType,
 		EmbeddingModel: &embeddingModel,
 	})
 	if err != nil {
@@ -94,29 +96,6 @@ func TestDatasetServiceUpdateDatasetUpdatesFields(t *testing.T) {
 	}
 }

-func TestUpdateDataset_RejectsSimultaneousParserIDAndPipelineID(t *testing.T) {
-	db := setupDatasetUpdateTestDB(t)
-	pushServiceDB(t, db)
-	insertDatasetUpdateKB(t, "kb-1", "tenant-1", "Original")
-
-	chunkMethod := "book"
-	pipelineID := "abcdef0123456789abcdef0123456789"
-
-	_, code, err := testDatasetUpdateService(t).UpdateDataset("kb-1", "tenant-1", service.UpdateDatasetRequest{
-		ParserID:   &chunkMethod,
-		PipelineID: &pipelineID,
-	})
-	if err == nil {
-		t.Fatal("expected mutual-exclusivity error when both parser_id and pipeline_id are set")
-	}
-	if code != common.CodeDataError {
-		t.Fatalf("expected data error code, got %d", code)
-	}
-	if !strings.Contains(err.Error(), "mutually exclusive") {
-		t.Fatalf("expected error to mention 'mutually exclusive', got: %v", err)
-	}
-}
-
 func TestUpdateDataset_ParseTypeBuiltinClearsPipelineID(t *testing.T) {
 	db := setupDatasetUpdateTestDB(t)
 	pushServiceDB(t, db)
@@ -177,6 +156,90 @@ func TestUpdateDataset_ParseTypePipelineIgnoresParserID(t *testing.T) {
 	}
 }

+// TestUpdateDataset_ParseTypePipelineCleansConfigAgainstCanvas verifies that
+// with parse_type=2 and a dirty parser_id, parser_config is still cleaned
+// against the canvas DSL (not the builtin DSL). This locks in the
+// ResolveParseMode contract on the dataset path: mode is authoritative and
+// ignores the conflicting parser_id.
+func TestUpdateDataset_ParseTypePipelineCleansConfigAgainstCanvas(t *testing.T) {
+	db := setupDatasetUpdateTestDB(t)
+	pushServiceDB(t, db)
+	insertDatasetUpdateCanvasKB(t, "kb-1", "tenant-1", "Original",
+		"abcdef0123456789abcdef0123456789")
+	seedDatasetUpdateCanvas(t, "abcdef0123456789abcdef0123456789", "tenant-1",
+		datasetUpdateCanvasDSL("Parser:CustomP", "chunk_token_num"))
+
+	// Dirty parser_id that the builtin branch would otherwise use to load a
+	// builtin DSL. parse_type=2 must ignore it.
+	chunkMethod := "book"
+	pipelineID := "abcdef0123456789abcdef0123456789"
+	parseType := 2
+	override := map[string]interface{}{
+		"Parser:CustomP": map[string]interface{}{"chunk_token_num": float64(256)},
+	}
+
+	_, code, err := testDatasetUpdateService(t).UpdateDataset("kb-1", "tenant-1", service.UpdateDatasetRequest{
+		ParserID:     &chunkMethod,
+		PipelineID:   &pipelineID,
+		ParseType:    &parseType,
+		ParserConfig: override,
+	})
+	if err != nil {
+		t.Fatalf("UpdateDataset failed: %v", err)
+	}
+	if code != common.CodeSuccess {
+		t.Fatalf("expected success code, got %d", code)
+	}
+
+	// Read back the persisted KB and verify parser_config was cleaned against
+	// the canvas DSL: it must carry the canvas component (Parser:CustomP), not
+	// a builtin-book component structure.
+	var kb entity.Knowledgebase
+	if err := dao.DB.First(&kb, "id = ?", "kb-1").Error; err != nil {
+		t.Fatalf("read back kb: %v", err)
+	}
+	cleaned, ok := kb.ParserConfig["Parser:CustomP"]
+	if !ok {
+		t.Fatalf("parser_config = %v, want Parser:CustomP component from canvas DSL", kb.ParserConfig)
+	}
+	cpn, ok := cleaned.(map[string]any)
+	if !ok {
+		t.Fatalf("Parser:CustomP = %T, want map", cleaned)
+	}
+	if v, _ := cpn["chunk_token_num"].(float64); v != 256 {
+		t.Fatalf("chunk_token_num = %v, want 256 (override applied on canvas defaults)", cpn["chunk_token_num"])
+	}
+}
+
+// TestUpdateDatasetRejectsInvalidPages verifies the fail-fast contract for the
+// "pages" range: an invalid range (from>to) aborts the request with
+// CodeDataError instead of being silently dropped or persisted.
+func TestUpdateDatasetRejectsInvalidPages(t *testing.T) {
+	db := setupDatasetUpdateTestDB(t)
+	pushServiceDB(t, db)
+	insertDatasetUpdateKB(t, "kb-1", "tenant-1", "Original")
+
+	chunkMethod := "manual"
+	parseType := 1
+	_, code, err := testDatasetUpdateService(t).UpdateDataset("kb-1", "tenant-1", service.UpdateDatasetRequest{
+		ParserID:  &chunkMethod,
+		ParseType: &parseType,
+		ParserConfig: map[string]interface{}{
+			"Parser:HipSignsRhyme": map[string]interface{}{
+				"pdf": map[string]interface{}{
+					"pages": []interface{}{[]interface{}{float64(5), float64(3)}}, // from>to invalid
+				},
+			},
+		},
+	})
+	if err == nil {
+		t.Fatalf("expected error for invalid pages range, got nil")
+	}
+	if code != common.CodeDataError {
+		t.Fatalf("code = %v, want CodeDataError", code)
+	}
+}
+
 func TestDatasetServiceGetDatasetReturnsEmptyConnectorList(t *testing.T) {
 	db := setupDatasetUpdateTestDB(t)
 	pushServiceDB(t, db)
@@ -1008,8 +1071,10 @@ func TestUpdateDataset_SwitchCanvasToBuiltinValidatesAgainstBuiltin(t *testing.T
 	insertDatasetUpdateCanvasKB(t, "kb-1", "tenant-1", "Original", "canvas-1")

 	chunkMethod := "naive"
+	parseType := 1
 	_, code, err := testDatasetUpdateService(t).UpdateDataset("kb-1", "tenant-1", service.UpdateDatasetRequest{
-		ParserID: &chunkMethod,
+		ParserID:  &chunkMethod,
+		ParseType: &parseType,
 		ParserConfig: map[string]interface{}{
 			// Valid for the "general" builtin template, not the canvas.
 			"Parser:HipSignsRhyme": map[string]interface{}{
--- a/internal/service/dataset_types.go
+++ b/internal/service/dataset_types.go
@@ -129,7 +129,7 @@ type CreateDatasetRequest struct {
 	ParserID       *string `json:"parser_id,omitempty"`
 	PipelineID     *string `json:"pipeline_id,omitempty"`
 	// ParseType indicates pipeline selection mode: 1 = BuiltIn (parser_id),
-	// 2 = Pipeline (pipeline_id). nil means unspecified (backward compat).
+	// 2 = Pipeline (pipeline_id). nil means unspecified.
 	ParseType *int `json:"parse_type,omitempty"`
 }

@@ -154,7 +154,7 @@ type UpdateDatasetRequest struct {
 	ParserConfig   map[string]interface{}     `json:"parser_config,omitempty"`
 	PipelineID     *string                    `json:"pipeline_id,omitempty"`
 	// ParseType indicates pipeline selection mode: 1 = BuiltIn (parser_id),
-	// 2 = Pipeline (pipeline_id). nil means unspecified (backward compat).
+	// 2 = Pipeline (pipeline_id). nil means unspecified.
 	ParseType *int `json:"parse_type,omitempty"`

 	// ParserConfigProvided reports whether the raw request body contained a
--- a/internal/service/document/document.go
+++ b/internal/service/document/document.go
@@ -122,6 +122,9 @@ type UpdateDatasetDocumentRequest struct {
 	Progress     *float64       `json:"progress"`
 	ParserConfig map[string]any `json:"parser_config"`
 	MetaFields   map[string]any `json:"meta_fields"`
+	// ParseType indicates pipeline selection mode: 1 = BuiltIn (parser_id),
+	// 2 = Pipeline (pipeline_id). nil means unspecified.
+	ParseType *int `json:"parse_type,omitempty"`
 }

 // PATCH /api/v1/datasets/:dataset_id/documents/:document_id.
--- a/internal/service/document/document_dataset_update.go
+++ b/internal/service/document/document_dataset_update.go
@@ -142,28 +142,21 @@ func (s *DocumentService) UpdateDatasetDocument(userID, datasetID, documentID st
 		}
 	}

-	if present["parser_config"] && req.ParserConfig != nil {
-		// Resolve effective pipeline to load the DSL for cleaning.
-		isCanvas := kb.PipelineID != nil && strings.TrimSpace(*kb.PipelineID) != ""
-		if req.PipelineID != nil {
-			isCanvas = strings.TrimSpace(*req.PipelineID) != ""
-		}
-		if req.ParserID != nil {
-			isCanvas = false
-		}
-		effParserID := kb.ParserID
-		if req.ParserID != nil {
-			effParserID = strings.TrimSpace(*req.ParserID)
-		}
-		effPipelineID := kb.PipelineID
-		if req.PipelineID != nil {
-			effPipelineID = req.PipelineID
-		}
-		if req.ParserID != nil && req.PipelineID == nil && kb.PipelineID != nil {
-			effPipelineID = nil
-		}
+	// Resolve the effective parse mode once: parse_type is authoritative when
+	// present, otherwise inherit the dataset's current mode. Both the
+	// parser_config cleaning and the reparse targeting derive from this single
+	// resolution so they can never disagree. (See service.ResolveParseMode.)
+	isPipeline, effParserID, effPipelineID := service.ResolveParseMode(
+		req.ParseType, req.ParserID, req.PipelineID,
+		service.ParseModeState{ParserID: kb.ParserID, PipelineID: kb.PipelineID})

-		dslJSON, err := service.LoadPipelineDSL(isCanvas, effParserID, effPipelineID)
+	if present["parser_config"] && req.ParserConfig != nil {
+		// Normalize "pages" ranges before persistence. Invalid ranges are
+		// rejected (fail-fast) — the request is aborted with a clear error.
+		if err := pipelinepkg.NormalizeParserConfigPages(map[string]any(req.ParserConfig)); err != nil {
+			return nil, common.CodeDataError, err
+		}
+		dslJSON, err := service.LoadPipelineDSL(isPipeline, effParserID, effPipelineID)
 		if err != nil {
 			common.Warn("cleanAndUpdateDocumentParserConfig: failed to load DSL, falling back to merge",
 				zap.Error(err))
@@ -180,22 +173,30 @@ func (s *DocumentService) UpdateDatasetDocument(userID, datasetID, documentID st
 		}
 	}

-	if present["pipeline_id"] {
-		if req.PipelineID != nil && strings.TrimSpace(*req.PipelineID) != "" {
-			if err := s.resetDocumentForReparse(doc, kb.TenantID, nil, req.PipelineID); err != nil {
-				return nil, common.CodeDataError, err
+	// Apply parser_id / pipeline_id changes. parse_type is validated earlier
+	// (in validateDatasetDocumentUpdate); reparse only fires when parse_type
+	// explicitly selects a mode. The mode direction comes from the same
+	// isPipeline resolution above.
+	var reparseParserID, reparsePipelineID *string
+	if req.ParseType != nil {
+		switch {
+		case !isPipeline: // BuiltIn
+			if present["parser_id"] && req.ParserID != nil {
+				if p := strings.TrimSpace(*req.ParserID); p != "" {
+					reparseParserID = &p
+				}
 			}
-		} else {
-			// Explicitly cleared: drop the custom canvas so the worker falls
-			// back to the built-in template, matching validation.
+			// Drop any prior canvas so the worker falls back to the builtin template.
 			empty := ""
-			if err := s.resetDocumentForReparse(doc, kb.TenantID, nil, &empty); err != nil {
-				return nil, common.CodeDataError, err
+			reparsePipelineID = &empty
+		case isPipeline: // Pipeline
+			if present["pipeline_id"] && req.PipelineID != nil {
+				reparsePipelineID = req.PipelineID
 			}
 		}
-	} else if present["parser_id"] && req.ParserID != nil && strings.TrimSpace(*req.ParserID) != "" {
-		parserID := strings.TrimSpace(*req.ParserID)
-		if err := s.resetDocumentForReparse(doc, kb.TenantID, &parserID, nil); err != nil {
+	}
+	if reparseParserID != nil || reparsePipelineID != nil {
+		if err := s.resetDocumentForReparse(doc, kb.TenantID, reparseParserID, reparsePipelineID); err != nil {
 			return nil, common.CodeDataError, err
 		}
 	}
@@ -247,10 +248,18 @@ func (s *DocumentService) validateDatasetDocumentUpdate(datasetID, documentID, u
 		}
 	}

-	if present["parser_id"] && req.ParserID != nil {
-		parserID := strings.TrimSpace(*req.ParserID)
-		if (doc.Type == "visual" && parserID != "picture") || (isPresentationFile(doc.Name) && parserID != "presentation") {
-			return common.CodeDataError, errors.New("Not supported yet!")
+	if present["parse_type"] || present["parser_id"] || present["pipeline_id"] {
+		isBuiltin, _, err := service.ValidateParseTypeMode(req.ParseType, req.ParserID, req.PipelineID)
+		if err != nil {
+			return common.CodeDataError, err
+		}
+		// The parser_id type constraint (visual/presentation) only applies in
+		// builtin mode — in pipeline mode parser_id is not applicable.
+		if isBuiltin && present["parser_id"] && req.ParserID != nil {
+			parserID := strings.TrimSpace(*req.ParserID)
+			if (doc.Type == "visual" && parserID != "picture") || (isPresentationFile(doc.Name) && parserID != "presentation") {
+				return common.CodeDataError, errors.New("Not supported yet!")
+			}
 		}
 	}
 	if present["name"] && req.Name != nil {
--- a/internal/service/document/document_test.go
+++ b/internal/service/document/document_test.go
@@ -1645,10 +1645,12 @@ func TestUpdateDatasetDocumentRejectsUnsupportedParserIDForVisualDoc(t *testing.
 	}

 	parserID := "naive"
+	parseType := 1
 	svc := testDocumentService(t)
 	_, code, err := svc.UpdateDatasetDocument("tenant-1", "kb-1", "doc-1", &UpdateDatasetDocumentRequest{
-		ParserID: &parserID,
-	}, map[string]bool{"parser_id": true})
+		ParserID:  &parserID,
+		ParseType: &parseType,
+	}, map[string]bool{"parser_id": true, "parse_type": true})
 	if err == nil {
 		t.Fatal("expected parser_id visual error")
 	}
@@ -1696,11 +1698,13 @@ func TestUpdateDatasetDocumentParserIDResetsForReparse(t *testing.T) {
 	insertTestKB(t, "kb-1", "tenant-1", 1, 10, 5)
 	insertNamedTestDoc(t, "doc-1", "kb-1", "doc.txt", 10, 5)

+	parseType := 1
 	chunkMethod := "manual"
 	svc := testDocumentService(t)
 	resp, code, err := svc.UpdateDatasetDocument("tenant-1", "kb-1", "doc-1", &UpdateDatasetDocumentRequest{
-		ParserID: &chunkMethod,
-	}, map[string]bool{"parser_id": true})
+		ParserID:  &chunkMethod,
+		ParseType: &parseType,
+	}, map[string]bool{"parser_id": true, "parse_type": true})
 	if err != nil {
 		t.Fatalf("UpdateDatasetDocument failed: code=%v err=%v", code, err)
 	}
@@ -2270,11 +2274,13 @@ func TestUpdateDatasetDocumentPipelineIDTakesPrecedenceOverParserID(t *testing.T

 	pipelineID := "1234567890abcdef1234567890abcdef"
 	chunkMethod := "manual"
+	parseType := 2
 	svc := testDocumentService(t)
 	resp, code, err := svc.UpdateDatasetDocument("tenant-1", "kb-1", "doc-1", &UpdateDatasetDocumentRequest{
 		PipelineID: &pipelineID,
 		ParserID:   &chunkMethod,
-	}, map[string]bool{"pipeline_id": true, "parser_id": true})
+		ParseType:  &parseType,
+	}, map[string]bool{"pipeline_id": true, "parser_id": true, "parse_type": true})
 	if err != nil {
 		t.Fatalf("UpdateDatasetDocument failed: code=%v err=%v", code, err)
 	}
@@ -2286,6 +2292,151 @@ func TestUpdateDatasetDocumentPipelineIDTakesPrecedenceOverParserID(t *testing.T
 	}
 }

+func TestUpdateDatasetDocumentParseTypeBuiltin(t *testing.T) {
+	db := setupServiceTestDB(t)
+	pushServiceDB(t, db)
+	insertTestKB(t, "kb-1", "tenant-1", 1, 10, 5)
+	insertNamedTestDoc(t, "doc-1", "kb-1", "doc.txt", 10, 5)
+
+	// Seed a prior canvas-backed document so the builtin transition must
+	// actively clear the persisted pipeline_id. Without this, a regression in
+	// the builtin branch's canvas-clearing behavior would go undetected.
+	priorPipeline := "1234567890abcdef1234567890abcdef"
+	if err := db.Model(&entity.Document{}).Where("id = ?", "doc-1").
+		Update("pipeline_id", priorPipeline).Error; err != nil {
+		t.Fatalf("seed prior pipeline: %v", err)
+	}
+
+	parseType := 1
+	parserID := "manual"
+	pipelineIDEmpty := ""
+	svc := testDocumentService(t)
+	resp, code, err := svc.UpdateDatasetDocument("tenant-1", "kb-1", "doc-1", &UpdateDatasetDocumentRequest{
+		ParseType:  &parseType,
+		ParserID:   &parserID,
+		PipelineID: &pipelineIDEmpty,
+	}, map[string]bool{"parser_id": true, "pipeline_id": true, "parse_type": true})
+	if err != nil {
+		t.Fatalf("UpdateDatasetDocument failed: code=%v err=%v", code, err)
+	}
+	if resp.ParserID != "manual" {
+		t.Fatalf("parser_id = %q, want %q", resp.ParserID, "manual")
+	}
+	if resp.PipelineID != nil && *resp.PipelineID != "" {
+		t.Fatalf("pipeline_id = %v, want empty", resp.PipelineID)
+	}
+
+	// Assert the persisted pipeline_id was cleared, not merely the response.
+	var updated entity.Document
+	if err := db.First(&updated, "id = ?", "doc-1").Error; err != nil {
+		t.Fatalf("read back doc: %v", err)
+	}
+	if updated.PipelineID != nil && *updated.PipelineID != "" {
+		t.Fatalf("persisted pipeline_id = %q, want cleared", *updated.PipelineID)
+	}
+}
+
+func TestUpdateDatasetDocumentParseTypePipeline(t *testing.T) {
+	db := setupServiceTestDB(t)
+	pushServiceDB(t, db)
+	insertTestKB(t, "kb-1", "tenant-1", 1, 10, 5)
+	insertNamedTestDoc(t, "doc-1", "kb-1", "doc.txt", 10, 5)
+
+	parseType := 2
+	pipelineID := "1234567890abcdef1234567890abcdef"
+	svc := testDocumentService(t)
+	resp, code, err := svc.UpdateDatasetDocument("tenant-1", "kb-1", "doc-1", &UpdateDatasetDocumentRequest{
+		ParseType:  &parseType,
+		PipelineID: &pipelineID,
+	}, map[string]bool{"parser_id": false, "pipeline_id": true, "parse_type": true})
+	if err != nil {
+		t.Fatalf("UpdateDatasetDocument failed: code=%v err=%v", code, err)
+	}
+	if resp.PipelineID == nil || *resp.PipelineID != pipelineID {
+		t.Fatalf("pipeline_id = %v, want %q", resp.PipelineID, pipelineID)
+	}
+	if resp.ParserID != "naive" {
+		t.Fatalf("parser_id = %q, want original naive", resp.ParserID)
+	}
+}
+
+// TestUpdateDatasetDocumentParseTypePipelineIgnoresDirtyParserID reproduces
+// the comment-3 bug: parse_type=2 (pipeline) with a dirty req.ParserID must
+// still resolve to pipeline mode so parser_config is cleaned against the
+// canvas DSL (fallback when the canvas is absent), not the builtin DSL.
+//
+// With the bug, req.ParserID != nil flipped isCanvas=false inside the
+// parser_config block, so the config was rebuilt against the builtin "manual"
+// DSL and unknown fields were dropped. After the fix, the canvas load fails
+// (no such pipeline in the test DB) and the original config is persisted as-is.
+func TestUpdateDatasetDocumentParseTypePipelineIgnoresDirtyParserID(t *testing.T) {
+	db := setupServiceTestDB(t)
+	pushServiceDB(t, db)
+	insertTestKB(t, "kb-1", "tenant-1", 1, 10, 5)
+	insertNamedTestDoc(t, "doc-1", "kb-1", "doc.txt", 10, 5)
+
+	parseType := 2
+	// Dirty parser_id that the builtin branch would otherwise use to load a
+	// builtin DSL and strip unknown fields.
+	parserID := "manual"
+	pipelineID := "1234567890abcdef1234567890abcdef"
+	svc := testDocumentService(t)
+	resp, code, err := svc.UpdateDatasetDocument("tenant-1", "kb-1", "doc-1", &UpdateDatasetDocumentRequest{
+		ParseType:  &parseType,
+		ParserID:   &parserID,
+		PipelineID: &pipelineID,
+		ParserConfig: map[string]any{
+			"nonexistent_field": "value",
+		},
+	}, map[string]bool{"parser_id": true, "pipeline_id": true, "parse_type": true, "parser_config": true})
+	if err != nil {
+		t.Fatalf("UpdateDatasetDocument failed: code=%v err=%v", code, err)
+	}
+	if resp.PipelineID == nil || *resp.PipelineID != pipelineID {
+		t.Fatalf("pipeline_id = %v, want %q", resp.PipelineID, pipelineID)
+	}
+	if resp.ParserID != "naive" {
+		t.Fatalf("parser_id = %q, want original naive (parser_id must not apply in pipeline mode)", resp.ParserID)
+	}
+	// Pipeline mode with an absent canvas must fall back to persisting the
+	// original config verbatim. If the builtin path ran instead, the unknown
+	// field would have been stripped during DSL cleaning.
+	if v, ok := resp.ParserConfig["nonexistent_field"]; !ok || v != "value" {
+		t.Fatalf("parser_config = %v, want nonexistent_field=value preserved (canvas fallback)", resp.ParserConfig)
+	}
+}
+
+// TestUpdateDatasetDocumentRejectsInvalidPages verifies the fail-fast contract
+// for the "pages" range: an invalid range (from<1) aborts the request with
+// CodeDataError instead of being silently dropped or persisted.
+func TestUpdateDatasetDocumentRejectsInvalidPages(t *testing.T) {
+	db := setupServiceTestDB(t)
+	pushServiceDB(t, db)
+	insertTestKB(t, "kb-1", "tenant-1", 1, 10, 5)
+	insertNamedTestDoc(t, "doc-1", "kb-1", "doc.txt", 10, 5)
+
+	parseType := 1
+	parserID := "manual"
+	svc := testDocumentService(t)
+	_, code, err := svc.UpdateDatasetDocument("tenant-1", "kb-1", "doc-1", &UpdateDatasetDocumentRequest{
+		ParseType: &parseType,
+		ParserID:  &parserID,
+		ParserConfig: map[string]any{
+			"Parser:HipSignsRhyme": map[string]any{
+				"pdf": map[string]any{
+					"pages": []any{[]any{float64(0), float64(5)}}, // from=0 invalid
+				},
+			},
+		},
+	}, map[string]bool{"parser_id": true, "parse_type": true, "parser_config": true})
+	if err == nil {
+		t.Fatalf("expected error for invalid pages range, got nil")
+	}
+	if code != common.CodeDataError {
+		t.Fatalf("code = %v, want CodeDataError", code)
+	}
+}
+
 func TestUpdateDatasetDocumentEnabledUpdatesStatus(t *testing.T) {
 	db := setupServiceTestDB(t)
 	pushServiceDB(t, db)
--- a/internal/service/parser_mode.go
+++ b/internal/service/parser_mode.go
@@ -0,0 +1,89 @@
+package service
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+)
+
+// ValidateParseTypeMode validates parse_type and ensures the corresponding
+// identifier field is present. Returns (isBuiltin, isPipeline, error).
+//   - parse_type must be 1 (BuiltIn) or 2 (Pipeline); nil is an error.
+//   - BuiltIn mode requires parserID to be present and non-empty.
+//   - Pipeline mode requires pipelineID to be present and non-empty.
+func ValidateParseTypeMode(parseType *int, parserID, pipelineID *string) (isBuiltin bool, isPipeline bool, err error) {
+	if parseType == nil {
+		return false, false, errors.New("parse_type is required")
+	}
+	switch *parseType {
+	case 1:
+		if parserID == nil || strings.TrimSpace(*parserID) == "" {
+			return false, false, errors.New("parser_id is required when parse_type is BuiltIn")
+		}
+		return true, false, nil
+	case 2:
+		if pipelineID == nil || strings.TrimSpace(*pipelineID) == "" {
+			return false, false, errors.New("pipeline_id is required when parse_type is Pipeline")
+		}
+		return false, true, nil
+	default:
+		return false, false, fmt.Errorf("invalid parse_type: %d (must be 1 or 2)", *parseType)
+	}
+}
+
+// ParseModeState carries the persisted IDs that ResolveParseMode falls back to
+// when the request does not switch modes.
+type ParseModeState struct {
+	ParserID   string
+	PipelineID *string
+}
+
+// ResolveParseMode returns the effective (isPipeline, parserID, pipelineID)
+// after applying an update. It is the single source of truth for "which DSL
+// should parser_config be cleaned against" and "which mode should reparse
+// target", shared by the dataset and document update paths.
+//
+// parseType is authoritative when present:
+//   - 1 (BuiltIn): effParserID = reqParserID (or current); effPipelineID = nil
+//     so any prior canvas is cleared. reqPipelineID is ignored.
+//   - 2 (Pipeline): effPipelineID = reqPipelineID (or current); effParserID =
+//     current.ParserID (parser_id is not applicable in pipeline mode).
+//     reqParserID is ignored.
+//
+// When parseType is nil (no mode switch, e.g. only parser_config changed),
+// inherit current mode and apply any incremental ID updates from the request;
+// isPipeline is derived from whether effPipelineID is non-empty.
+func ResolveParseMode(parseType *int, reqParserID, reqPipelineID *string,
+	current ParseModeState) (isPipeline bool, effParserID string, effPipelineID *string) {
+	if parseType != nil {
+		switch *parseType {
+		case 1: // BuiltIn
+			effParserID = current.ParserID
+			if reqParserID != nil {
+				if p := strings.TrimSpace(*reqParserID); p != "" {
+					effParserID = p
+				}
+			}
+			return false, effParserID, nil
+		case 2: // Pipeline
+			effPipelineID = current.PipelineID
+			if reqPipelineID != nil {
+				effPipelineID = reqPipelineID
+			}
+			return true, current.ParserID, effPipelineID
+		}
+	}
+	// No mode switch — inherit current state, apply incremental ID updates.
+	effParserID = current.ParserID
+	if reqParserID != nil {
+		if p := strings.TrimSpace(*reqParserID); p != "" {
+			effParserID = p
+		}
+	}
+	effPipelineID = current.PipelineID
+	if reqPipelineID != nil {
+		effPipelineID = reqPipelineID
+	}
+	isPipeline = effPipelineID != nil && strings.TrimSpace(*effPipelineID) != ""
+	return isPipeline, effParserID, effPipelineID
+}
--- a/internal/service/parser_mode_test.go
+++ b/internal/service/parser_mode_test.go
@@ -0,0 +1,208 @@
+package service
+
+import (
+	"testing"
+)
+
+func TestValidateParseTypeMode_Nil(t *testing.T) {
+	_, _, err := ValidateParseTypeMode(nil, nil, nil)
+	if err == nil || err.Error() != "parse_type is required" {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}
+
+func TestValidateParseTypeMode_InvalidZero(t *testing.T) {
+	v := 0
+	_, _, err := ValidateParseTypeMode(&v, nil, nil)
+	if err == nil || err.Error() != "invalid parse_type: 0 (must be 1 or 2)" {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}
+
+func TestValidateParseTypeMode_InvalidThree(t *testing.T) {
+	v := 3
+	_, _, err := ValidateParseTypeMode(&v, nil, nil)
+	if err == nil || err.Error() != "invalid parse_type: 3 (must be 1 or 2)" {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}
+
+func TestValidateParseTypeMode_BuiltinMissingParserID(t *testing.T) {
+	t.Run("nil parserID", func(t *testing.T) {
+		pt := 1
+		_, _, err := ValidateParseTypeMode(&pt, nil, nil)
+		if err == nil || err.Error() != "parser_id is required when parse_type is BuiltIn" {
+			t.Fatalf("unexpected error: %v", err)
+		}
+	})
+	t.Run("empty parserID", func(t *testing.T) {
+		pt := 1
+		empty := ""
+		_, _, err := ValidateParseTypeMode(&pt, &empty, nil)
+		if err == nil || err.Error() != "parser_id is required when parse_type is BuiltIn" {
+			t.Fatalf("unexpected error: %v", err)
+		}
+	})
+	t.Run("whitespace parserID", func(t *testing.T) {
+		pt := 1
+		ws := "  "
+		_, _, err := ValidateParseTypeMode(&pt, &ws, nil)
+		if err == nil || err.Error() != "parser_id is required when parse_type is BuiltIn" {
+			t.Fatalf("unexpected error: %v", err)
+		}
+	})
+}
+
+func TestValidateParseTypeMode_BuiltinOK(t *testing.T) {
+	pt := 1
+	pid := "laws"
+	builtin, pipeline, err := ValidateParseTypeMode(&pt, &pid, nil)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !builtin || pipeline {
+		t.Fatalf("expected builtin=true pipeline=false, got builtin=%v pipeline=%v", builtin, pipeline)
+	}
+}
+
+func TestValidateParseTypeMode_PipelineMissingPipelineID(t *testing.T) {
+	t.Run("nil pipelineID", func(t *testing.T) {
+		pt := 2
+		_, _, err := ValidateParseTypeMode(&pt, nil, nil)
+		if err == nil || err.Error() != "pipeline_id is required when parse_type is Pipeline" {
+			t.Fatalf("unexpected error: %v", err)
+		}
+	})
+	t.Run("empty pipelineID", func(t *testing.T) {
+		pt := 2
+		empty := ""
+		_, _, err := ValidateParseTypeMode(&pt, nil, &empty)
+		if err == nil || err.Error() != "pipeline_id is required when parse_type is Pipeline" {
+			t.Fatalf("unexpected error: %v", err)
+		}
+	})
+}
+
+func TestValidateParseTypeMode_PipelineOK(t *testing.T) {
+	pt := 2
+	pipe := "abc123"
+	builtin, pipeline, err := ValidateParseTypeMode(&pt, nil, &pipe)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if builtin || !pipeline {
+		t.Fatalf("expected builtin=false pipeline=true, got builtin=%v pipeline=%v", builtin, pipeline)
+	}
+}
+
+func TestResolveParseMode_BuiltinIgnoresPipelineID(t *testing.T) {
+	pt := 1
+	parserID := "manual"
+	pipelineID := "should-be-ignored"
+	cur := ParseModeState{ParserID: "naive", PipelineID: strPtr("prior-canvas")}
+	isPipeline, effParserID, effPipelineID := ResolveParseMode(&pt, &parserID, &pipelineID, cur)
+	if isPipeline {
+		t.Fatalf("isPipeline = true, want false (builtin mode)")
+	}
+	if effParserID != "manual" {
+		t.Fatalf("effParserID = %q, want manual", effParserID)
+	}
+	if effPipelineID != nil {
+		t.Fatalf("effPipelineID = %v, want nil (canvas cleared)", effPipelineID)
+	}
+}
+
+// TestResolveParseMode_PipelineIgnoresParserID reproduces the comment-3 bug
+// contract: parse_type=2 must ignore a dirty req.ParserID and keep isPipeline
+// true so parser_config is cleaned against the canvas DSL, not the builtin DSL.
+func TestResolveParseMode_PipelineIgnoresParserID(t *testing.T) {
+	pt := 2
+	parserID := "should-be-ignored"
+	pipelineID := "1234567890abcdef1234567890abcdef"
+	cur := ParseModeState{ParserID: "naive", PipelineID: nil}
+	isPipeline, effParserID, effPipelineID := ResolveParseMode(&pt, &parserID, &pipelineID, cur)
+	if !isPipeline {
+		t.Fatalf("isPipeline = false, want true (pipeline mode)")
+	}
+	if effParserID != "naive" {
+		t.Fatalf("effParserID = %q, want current naive (parser_id not applicable in pipeline mode)", effParserID)
+	}
+	if effPipelineID == nil || *effPipelineID != pipelineID {
+		t.Fatalf("effPipelineID = %v, want %q", effPipelineID, pipelineID)
+	}
+}
+
+func TestResolveParseMode_BuiltinFallsBackToCurrentParserID(t *testing.T) {
+	pt := 1
+	cur := ParseModeState{ParserID: "naive", PipelineID: strPtr("prior-canvas")}
+	isPipeline, effParserID, effPipelineID := ResolveParseMode(&pt, nil, nil, cur)
+	if isPipeline {
+		t.Fatalf("isPipeline = true, want false")
+	}
+	if effParserID != "naive" {
+		t.Fatalf("effParserID = %q, want naive", effParserID)
+	}
+	if effPipelineID != nil {
+		t.Fatalf("effPipelineID = %v, want nil", effPipelineID)
+	}
+}
+
+func TestResolveParseMode_PipelineFallsBackToCurrentPipelineID(t *testing.T) {
+	pt := 2
+	cur := ParseModeState{ParserID: "naive", PipelineID: strPtr("prior-canvas")}
+	isPipeline, effParserID, effPipelineID := ResolveParseMode(&pt, nil, nil, cur)
+	if !isPipeline {
+		t.Fatalf("isPipeline = false, want true")
+	}
+	if effParserID != "naive" {
+		t.Fatalf("effParserID = %q, want naive", effParserID)
+	}
+	if effPipelineID == nil || *effPipelineID != "prior-canvas" {
+		t.Fatalf("effPipelineID = %v, want prior-canvas", effPipelineID)
+	}
+}
+
+// TestResolveParseMode_NilParseTypeInheritsCurrent covers the "only
+// parser_config changed" path: no mode switch, inherit current state.
+func TestResolveParseMode_NilParseTypeInheritsCurrent(t *testing.T) {
+	t.Run("current is pipeline", func(t *testing.T) {
+		cur := ParseModeState{ParserID: "naive", PipelineID: strPtr("canvas-1")}
+		isPipeline, effParserID, effPipelineID := ResolveParseMode(nil, nil, nil, cur)
+		if !isPipeline {
+			t.Fatalf("isPipeline = false, want true")
+		}
+		if effParserID != "naive" {
+			t.Fatalf("effParserID = %q, want naive", effParserID)
+		}
+		if effPipelineID == nil || *effPipelineID != "canvas-1" {
+			t.Fatalf("effPipelineID = %v, want canvas-1", effPipelineID)
+		}
+	})
+	t.Run("current is builtin", func(t *testing.T) {
+		cur := ParseModeState{ParserID: "manual", PipelineID: nil}
+		isPipeline, effParserID, effPipelineID := ResolveParseMode(nil, nil, nil, cur)
+		if isPipeline {
+			t.Fatalf("isPipeline = true, want false")
+		}
+		if effParserID != "manual" {
+			t.Fatalf("effParserID = %q, want manual", effParserID)
+		}
+		if effPipelineID != nil {
+			t.Fatalf("effPipelineID = %v, want nil", effPipelineID)
+		}
+	})
+	t.Run("incremental parser_id update applied", func(t *testing.T) {
+		cur := ParseModeState{ParserID: "naive", PipelineID: nil}
+		newParser := "laws"
+		isPipeline, effParserID, effPipelineID := ResolveParseMode(nil, &newParser, nil, cur)
+		if isPipeline {
+			t.Fatalf("isPipeline = true, want false")
+		}
+		if effParserID != "laws" {
+			t.Fatalf("effParserID = %q, want laws", effParserID)
+		}
+		if effPipelineID != nil {
+			t.Fatalf("effPipelineID = %v, want nil", effPipelineID)
+		}
+	})
+}
--- a/internal/service/pipeline_params.go
+++ b/internal/service/pipeline_params.go
@@ -37,10 +37,10 @@ func loadCanvasDSLJSON(canvasID string) ([]byte, error) {
 }

 // LoadPipelineDSL loads the DSL JSON for a pipeline identified by parserID
-// (built-in) or pipelineID (custom canvas). When both are provided, isCanvas
+// (built-in) or pipelineID (custom canvas). When both are provided, isPipeline
 // selects which one to use.
-func LoadPipelineDSL(isCanvas bool, parserID string, pipelineID *string) ([]byte, error) {
-	if isCanvas {
+func LoadPipelineDSL(isPipeline bool, parserID string, pipelineID *string) ([]byte, error) {
+	if isPipeline {
 		return loadCanvasDSLJSON(strings.TrimSpace(*pipelineID))
 	}
 	registry, err := pipelinepkg.DefaultRegistry()
@@ -62,10 +62,10 @@ func LoadPipelineDSL(isCanvas bool, parserID string, pipelineID *string) ([]byte
 // For builtin templates the DSL is loaded from the embedded registry; for custom
 // canvas pipelines it is loaded from the canvas row in the database.
 func ResolveComponentParamsDefaults(parserID string, pipelineID *string) (entity.JSONMap, error) {
-	isCanvas := pipelineID != nil && strings.TrimSpace(*pipelineID) != ""
+	isPipeline := pipelineID != nil && strings.TrimSpace(*pipelineID) != ""
 	var cp map[string]map[string]any
 	var err error
-	if isCanvas {
+	if isPipeline {
 		dslJSON, lerr := loadCanvasDSLJSON(strings.TrimSpace(*pipelineID))
 		if lerr != nil {
 			return nil, fmt.Errorf("load canvas DSL: %w", lerr)
@@ -134,18 +134,3 @@ func getEmbdIDs(kbs []*entity.Knowledgebase) []string {
 	}
 	return ids
 }
-
-// Backward-compat lowercase aliases for callers within the service package.
-// These will be removed when callers are updated to use the exported names.
-
-func loadPipelineDSL(isCanvas bool, parserID string, pipelineID *string) ([]byte, error) {
-	return LoadPipelineDSL(isCanvas, parserID, pipelineID)
-}
-
-func resolveComponentParamsDefaults(parserID string, pipelineID *string) (entity.JSONMap, error) {
-	return ResolveComponentParamsDefaults(parserID, pipelineID)
-}
-
-func validateDatasetEmbeddingModels(kbs []*entity.Knowledgebase) error {
-	return ValidateDatasetEmbeddingModels(kbs)
-}
--- a/internal/utility/pdf_pages.go
+++ b/internal/utility/pdf_pages.go
@@ -0,0 +1,102 @@
+package utility
+
+import (
+	"fmt"
+	"sort"
+)
+
+// NormalizePDFPages normalizes a raw "pages" value (list[list[int]], 1-indexed
+// inclusive ranges, JSON-decoded as []any of []any of float64) into a sorted,
+// merged, de-duplicated [][]int.
+//
+// Semantics (fail-fast):
+//   - nil or empty list → (nil, nil): "no value", callers treat as "parse all
+//     pages".
+//   - Any invalid range → (nil, error): the whole input is rejected; callers
+//     should surface the error and abort the request. No partial dropping.
+//   - All ranges valid → (normalized, nil).
+//
+// Validation per range [from, to]:
+//   - both values must be integers (int, int64, or integral float64);
+//   - from >= 1 (1-indexed);
+//   - from <= to.
+//
+// Surviving ranges are sorted by `from` then merged when overlapping or
+// adjacent (next.from <= cur.to + 1).
+func NormalizePDFPages(raw any) ([][]int, error) {
+	list, ok := raw.([]any)
+	if !ok {
+		// nil raw (no key / JSON null) is "no value"; a non-list raw is a type
+		// error. raw==nil falls through the !ok branch because nil does not
+		// satisfy []any.
+		if raw == nil {
+			return nil, nil
+		}
+		return nil, fmt.Errorf("pages must be a list of [from,to] ranges, got %T", raw)
+	}
+	if len(list) == 0 {
+		return nil, nil
+	}
+
+	ranges := make([][]int, 0, len(list))
+	for _, item := range list {
+		pair, ok := item.([]any)
+		if !ok || len(pair) != 2 {
+			return nil, fmt.Errorf("invalid page range %v: must be a [from,to] pair", item)
+		}
+		from, ok := toInt(pair[0])
+		if !ok {
+			return nil, fmt.Errorf("invalid page range [%v,%v]: from must be an integer", pair[0], pair[1])
+		}
+		to, ok := toInt(pair[1])
+		if !ok {
+			return nil, fmt.Errorf("invalid page range [%v,%v]: to must be an integer", pair[0], pair[1])
+		}
+		if from < 1 {
+			return nil, fmt.Errorf("invalid page range [%d,%d]: from must be >= 1", from, to)
+		}
+		if from > to {
+			return nil, fmt.Errorf("invalid page range [%d,%d]: from must be <= to", from, to)
+		}
+		ranges = append(ranges, []int{from, to})
+	}
+
+	sort.Slice(ranges, func(i, j int) bool {
+		if ranges[i][0] != ranges[j][0] {
+			return ranges[i][0] < ranges[j][0]
+		}
+		return ranges[i][1] < ranges[j][1]
+	})
+
+	merged := ranges[:1]
+	for _, r := range ranges[1:] {
+		last := merged[len(merged)-1]
+		if r[0] <= last[1]+1 { // overlap or adjacent
+			if r[1] > last[1] {
+				last[1] = r[1]
+			}
+		} else {
+			merged = append(merged, r)
+		}
+	}
+	return merged, nil
+}
+
+// toInt coerces a JSON-decoded numeric value to int. Accepts int, int64, and
+// integral float64; rejects non-numeric or non-integral values.
+func toInt(v any) (int, bool) {
+	switch x := v.(type) {
+	case int:
+		return x, true
+	case int64:
+		return int(x), true
+	case float64:
+		// Reject non-integral floats (e.g. 1.5) — pages must be integers.
+		if x != float64(int(x)) {
+			return 0, false
+		}
+		return int(x), true
+	default:
+		return 0, false
+	}
+}
--- a/internal/utility/pdf_pages_test.go
+++ b/internal/utility/pdf_pages_test.go
@@ -0,0 +1,115 @@
+package utility
+
+import (
+	"reflect"
+	"testing"
+)
+
+// TestNormalizePDFPages verifies normalization of the 1-indexed inclusive
+// "pages" ranges under fail-fast semantics: valid ranges are normalized
+// (sorted/merged/deduplicated), any invalid range rejects the whole input,
+// and nil/empty means "no value" (parse all pages).
+func TestNormalizePDFPages(t *testing.T) {
+	// f64 builds a JSON-decoded-style []any of []any of float64.
+	f64 := func(ranges ...[2]float64) any {
+		out := make([]any, 0, len(ranges))
+		for _, r := range ranges {
+			out = append(out, []any{r[0], r[1]})
+		}
+		return out
+	}
+
+	// valid cases: expect (normalized, nil)
+	validCases := []struct {
+		name string
+		raw  any
+		want [][]int
+	}{
+		{"single range float64", f64([2]float64{1, 100}), [][]int{{1, 100}}},
+		{"two disjoint ranges", f64([2]float64{1, 100}, [2]float64{400, 500}), [][]int{{1, 100}, {400, 500}}},
+		{"overlap merged", f64([2]float64{1, 200}, [2]float64{111, 333}), [][]int{{1, 333}}},
+		{"adjacent merged", f64([2]float64{1, 100}, [2]float64{101, 200}), [][]int{{1, 200}}},
+		{"unsorted input sorted", f64([2]float64{400, 500}, [2]float64{1, 100}), [][]int{{1, 100}, {400, 500}}},
+		{"int accepted", []any{[]any{int(1), int(100)}}, [][]int{{1, 100}}},
+		{"int64 accepted", []any{[]any{int64(1), int64(100)}}, [][]int{{1, 100}}},
+		{"single page range", f64([2]float64{5, 5}), [][]int{{5, 5}}},
+	}
+	for _, c := range validCases {
+		t.Run("valid/"+c.name, func(t *testing.T) {
+			got, err := NormalizePDFPages(c.raw)
+			if err != nil {
+				t.Fatalf("NormalizePDFPages(%v) unexpected error: %v", c.raw, err)
+			}
+			if !reflect.DeepEqual(got, c.want) {
+				t.Errorf("NormalizePDFPages(%v) = %v, want %v", c.raw, got, c.want)
+			}
+		})
+	}
+
+	// no-value cases: expect (nil, nil)
+	noValueCases := []struct {
+		name string
+		raw  any
+	}{
+		{"nil", nil},
+		{"empty slice", []any{}},
+	}
+	for _, c := range noValueCases {
+		t.Run("no-value/"+c.name, func(t *testing.T) {
+			got, err := NormalizePDFPages(c.raw)
+			if err != nil {
+				t.Fatalf("NormalizePDFPages(%v) unexpected error: %v", c.raw, err)
+			}
+			if got != nil {
+				t.Errorf("NormalizePDFPages(%v) = %v, want nil", c.raw, got)
+			}
+		})
+	}
+
+	// invalid cases: expect (nil, error) — fail-fast, no partial dropping
+	invalidCases := []struct {
+		name string
+		raw  any
+	}{
+		{"not a slice", "1-100"},
+		{"from>to", f64([2]float64{5, 3})},
+		{"from<1", f64([2]float64{0, 5})},
+		{"from<1 negative", f64([2]float64{-3, 5})},
+		{"mixed valid and invalid", f64([2]float64{1, 10}, [2]float64{9, 2}, [2]float64{20, 30})},
+		{"wrong arity", []any{[]any{float64(1)}}},
+		{"non-numeric", []any{[]any{"a", "b"}}},
+		{"non-integral float", f64([2]float64{1.5, 3})},
+		{"all invalid", f64([2]float64{5, 3}, [2]float64{0, 2})},
+	}
+	for _, c := range invalidCases {
+		t.Run("invalid/"+c.name, func(t *testing.T) {
+			got, err := NormalizePDFPages(c.raw)
+			if err == nil {
+				t.Fatalf("NormalizePDFPages(%v) expected error, got nil (result=%v)", c.raw, got)
+			}
+			if got != nil {
+				t.Errorf("NormalizePDFPages(%v) = %v, want nil on error", c.raw, got)
+			}
+		})
+	}
+
+	// Idempotency: feeding the output back in yields the same result.
+	t.Run("idempotent", func(t *testing.T) {
+		once, err := NormalizePDFPages(f64([2]float64{1, 200}, [2]float64{111, 333}))
+		if err != nil {
+			t.Fatalf("first pass: %v", err)
+		}
+		// Convert [][]int to []any of []any for the second pass.
+		raw2 := make([]any, 0, len(once))
+		for _, r := range once {
+			raw2 = append(raw2, []any{r[0], r[1]})
+		}
+		twice, err := NormalizePDFPages(raw2)
+		if err != nil {
+			t.Fatalf("second pass: %v", err)
+		}
+		if !reflect.DeepEqual(once, twice) {
+			t.Errorf("not idempotent: once=%v twice=%v", once, twice)
+		}
+	})
+}
--- a/test/testcases/restful_api/test_datasets.py
+++ b/test/testcases/restful_api/test_datasets.py
@@ -36,6 +36,18 @@ def _skip_go_ignored_null(payload, field):
        pytest.skip(f"Go dataset update ignores an explicit null {field}")


+def _parser_id_fields(chunk_method):
+    """Build parser_id/chunk_method fields with parse_type=1 for Go proxy.
+
+    Go proxy requires parse_type alongside parser_id (fail-fast by design);
+    Python mode uses chunk_method alone.
+    """
+    fields = {PARSER_ID_FIELD: chunk_method}
+    if IS_GO_PROXY:
+        fields["parse_type"] = 1
+    return fields
+
+
 def _is_infinity_doc_engine(rest_client: RestClient) -> bool:
    env_engine = (os.getenv("DOC_ENGINE") or "").strip().lower()
    if env_engine:
@@ -160,7 +172,7 @@ def test_dataset_update_language_connectors_avatar_and_description_contract(rest
        json={
            "name": "dataset_update_lang_connectors",
            "description": "",
-            PARSER_ID_FIELD: "naive",
+            **_parser_id_fields("naive"),
            "language": "English",
            "connectors": [],
            "avatar": avatar_value,
@@ -198,9 +210,8 @@ def test_dataset_update_language_connectors_avatar_and_description_contract(rest
        "presentation",
        "qa",
        "table",
-        "tag",
    ],
-    ids=["naive", "book", "email", "laws", "manual", "one", "paper", "picture", "presentation", "qa", "table", "tag"],
+    ids=["naive", "book", "email", "laws", "manual", "one", "paper", "picture", "presentation", "qa", "table"],
 )
 def test_dataset_update_chunk_method_contract(rest_client, clear_datasets, chunk_method):
    create_res = rest_client.post("/datasets", json={"name": f"dataset_update_chunk_{chunk_method}"})
@@ -211,7 +222,7 @@ def test_dataset_update_chunk_method_contract(rest_client, clear_datasets, chunk

    update_res = rest_client.put(
        f"/datasets/{dataset_id}",
-        json={PARSER_ID_FIELD: chunk_method},
+        json=_parser_id_fields(chunk_method),
    )
    assert update_res.status_code == 200
    update_payload = update_res.json()
@@ -363,9 +374,9 @@ def test_dataset_update_parser_config_valid_matrix_contract(rest_client, clear_d
@pytest.mark.parametrize(
    "name, update_payload",
    [
-        ("parser_config_empty", {PARSER_ID_FIELD: "qa", "parser_config": {}}),
-        ("parser_config_none", {PARSER_ID_FIELD: "qa", "parser_config": None}),
-        ("parser_config_unset", {PARSER_ID_FIELD: "qa"}),
+        ("parser_config_empty", {**_parser_id_fields("qa"), "parser_config": {}}),
+        ("parser_config_none", {**_parser_id_fields("qa"), "parser_config": None}),
+        ("parser_config_unset", _parser_id_fields("qa")),
    ],
    ids=["parser_config_empty", "parser_config_none", "parser_config_unset"],
 )
@@ -896,14 +907,16 @@ def test_dataset_update_chunk_method_invalid_contract(rest_client, clear_dataset

    expected_chunk_message = "Input should be 'naive', 'book', 'email', 'laws', 'manual', 'one', 'paper', 'picture', 'presentation', 'qa', 'table', 'tag' or 'resume'"
    for chunk_method in ("", "unknown", []):
-        res = rest_client.put(f"/datasets/{dataset_id}", json={PARSER_ID_FIELD: chunk_method})
+        res = rest_client.put(f"/datasets/{dataset_id}", json=_parser_id_fields(chunk_method))
        assert res.status_code == 200
        payload = res.json()
        assert payload["code"] == ARGUMENT_ERROR_CODE, payload
        if IS_GO_PROXY and not isinstance(chunk_method, str):
            assert "cannot unmarshal" in payload["message"] and f".{PARSER_ID_FIELD}" in payload["message"], payload
+        elif IS_GO_PROXY and chunk_method == "":
+            assert payload["message"] == "parser_id is required when parse_type is BuiltIn", payload
        elif IS_GO_PROXY:
-            assert payload["message"].startswith("Input should be 'audio', 'book'") and payload["message"].endswith("or 'tag'"), payload
+            assert payload["message"].startswith("Input should be 'audio', 'book'") and payload["message"].endswith("or 'table'"), payload
        else:
            assert expected_chunk_message in payload["message"], payload

@@ -913,7 +926,7 @@ def test_dataset_update_chunk_method_invalid_contract(rest_client, clear_dataset
    _skip_go_ignored_null(none_payload, PARSER_ID_FIELD)
    assert none_payload["code"] == ARGUMENT_ERROR_CODE, none_payload
    if IS_GO_PROXY:
-        assert none_payload["message"].startswith("Input should be 'audio', 'book'") and none_payload["message"].endswith("or 'tag'"), none_payload
+        assert none_payload["message"].startswith("Input should be 'audio', 'book'") and none_payload["message"].endswith("or 'table'"), none_payload
    else:
        assert expected_chunk_message in none_payload["message"], none_payload

@@ -1174,12 +1187,11 @@ def test_dataset_create_avatar_and_description_contract(rest_client, clear_datas
        ("presentation", "presentation"),
        ("qa", "qa"),
        ("table", "table"),
-        ("tag", "tag"),
    ],
-    ids=["naive", "book", "email", "laws", "manual", "one", "paper", "picture", "presentation", "qa", "table", "tag"],
+    ids=["naive", "book", "email", "laws", "manual", "one", "paper", "picture", "presentation", "qa", "table"],
 )
 def test_dataset_create_chunk_method_contract(rest_client, clear_datasets, name, chunk_method):
-    res = rest_client.post("/datasets", json={"name": name, PARSER_ID_FIELD: chunk_method})
+    res = rest_client.post("/datasets", json={"name": name, **_parser_id_fields(chunk_method)})
    assert res.status_code == 200
    payload = res.json()
    assert payload["code"] == 0, payload
@@ -1672,23 +1684,25 @@ def test_dataset_create_permission_and_chunk_method_contract(rest_client, clear_
    ]
    expected_chunk_message = "Input should be 'naive', 'book', 'email', 'laws', 'manual', 'one', 'paper', 'picture', 'presentation', 'qa', 'table', 'tag' or 'resume'"
    for name, chunk_method in chunk_method_invalid_cases:
-        res = rest_client.post("/datasets", json={"name": name, PARSER_ID_FIELD: chunk_method})
+        res = rest_client.post("/datasets", json={"name": name, **_parser_id_fields(chunk_method)})
        assert res.status_code == 200
        payload = res.json()
        assert payload["code"] == ARGUMENT_ERROR_CODE, payload
        if IS_GO_PROXY and not isinstance(chunk_method, str):
            assert "cannot unmarshal" in payload["message"] and f".{PARSER_ID_FIELD}" in payload["message"], payload
+        elif IS_GO_PROXY and chunk_method == "":
+            assert payload["message"] == "parser_id is required when parse_type is BuiltIn", payload
        elif IS_GO_PROXY:
-            assert payload["message"].startswith("Input should be 'audio', 'book'") and payload["message"].endswith("or 'tag'"), payload
+            assert payload["message"].startswith("Input should be 'audio', 'book'") and payload["message"].endswith("or 'table'"), payload
        else:
            assert expected_chunk_message in payload["message"], payload

-    chunk_method_none_res = rest_client.post("/datasets", json={"name": "chunk_method_none", PARSER_ID_FIELD: None})
+    chunk_method_none_res = rest_client.post("/datasets", json={"name": "chunk_method_none", **_parser_id_fields(None)})
    assert chunk_method_none_res.status_code == 200
    chunk_method_none_payload = chunk_method_none_res.json()
    assert chunk_method_none_payload["code"] == ARGUMENT_ERROR_CODE, chunk_method_none_payload
    if IS_GO_PROXY:
-        assert chunk_method_none_payload["message"].startswith("Input should be 'audio', 'book'") and chunk_method_none_payload["message"].endswith("or 'tag'"), chunk_method_none_payload
+        assert chunk_method_none_payload["message"] == "parser_id is required when parse_type is BuiltIn", chunk_method_none_payload
    else:
        assert expected_chunk_message in chunk_method_none_payload["message"], chunk_method_none_payload

--- a/web/src/components/chunk-method-dialog/index.tsx
+++ b/web/src/components/chunk-method-dialog/index.tsx
@@ -141,6 +141,20 @@ export function ChunkMethodDialog({
        entity_types: z.array(z.string()).optional(),
        pages: z
          .array(z.object({ from: z.coerce.number(), to: z.coerce.number() }))
+          .refine(
+            (ranges) =>
+              ranges.every(
+                (r) =>
+                  Number.isInteger(r.from) &&
+                  Number.isInteger(r.to) &&
+                  r.from >= 1 &&
+                  r.from <= r.to,
+              ),
+            {
+              message:
+                'page range invalid: from/to must be integers, from >= 1, from <= to',
+            },
+          )
          .optional(),
        metadata: z.any().optional(),
        built_in_metadata: z
--- a/web/src/components/document-pipeline-dialog/use-document-pipeline-form.ts
+++ b/web/src/components/document-pipeline-dialog/use-document-pipeline-form.ts
@@ -185,6 +185,7 @@ export function useDocumentPipelineForm({
        parser_id: isPipeline ? '' : data.parser_id,
        pipeline_id: isPipeline ? data.pipeline_id : '',
        parser_config: transformedConfig,
+        parseType: data.parseType,
      };
    },
    [],
--- a/web/src/hooks/use-document-request.ts
+++ b/web/src/hooks/use-document-request.ts
@@ -524,12 +524,14 @@ export const useSetDocumentPipelineParser = () => {
    mutationFn: async ({
      parserId,
      pipelineId,
+      parseType,
      documentId,
      datasetId,
      parserConfig,
    }: {
      parserId: string;
      pipelineId: string;
+      parseType?: number;
      documentId: string;
      datasetId: string;
      parserConfig?: IChangeParserConfigRequestBody;
@@ -539,6 +541,10 @@ export const useSetDocumentPipelineParser = () => {
        pipeline_id: pipelineId,
      };

+      if (parseType !== undefined) {
+        updateData.parse_type = parseType;
+      }
+
      if (parserConfig) {
        updateData.parser_config = isPipelineParserConfig(parserConfig)
          ? parserConfig
--- a/web/src/interfaces/request/document.ts
+++ b/web/src/interfaces/request/document.ts
@@ -40,6 +40,8 @@ export interface IChangeParserConfigRequestBody {
 export interface IChangeParserRequestBody {
  parser_id: string;
  pipeline_id?: string;
+  // 1 = BuiltIn (parser_id), 2 = Pipeline (pipeline_id); omitted = unspecified.
+  parseType?: number;
  doc_id?: string;
  parser_config: IChangeParserConfigRequestBody;
 }
--- a/web/src/pages/dataset/dataset/use-change-document-parser.ts
+++ b/web/src/pages/dataset/dataset/use-change-document-parser.ts
@@ -25,16 +25,19 @@ export const useChangeDocumentParser = () => {
      if (record?.id && record?.dataset_id) {
        // The Go document endpoint takes `parser_id` and a pipeline-shaped
        // parser_config; the Python one keeps the legacy payload shape.
-        const setParser = isGoBackend()
-          ? setDocumentPipelineParser
-          : setDocumentParser;
-        const ret = await setParser({
+        const common = {
          parserId: parserConfigInfo.parser_id,
          pipelineId: parserConfigInfo.pipeline_id || '',
          documentId: record?.id,
          datasetId: record?.dataset_id,
          parserConfig: parserConfigInfo.parser_config,
-        });
+        };
+        const ret = isGoBackend()
+          ? await setDocumentPipelineParser({
+              ...common,
+              parseType: parserConfigInfo.parseType,
+            })
+          : await setDocumentParser(common);
        if (ret === 0) {
          hideChangeParserModal();
        }