From bb96bb687dcd4a0e2dd373cafde940ab06fa2d86 Mon Sep 17 00:00:00 2001 From: Jack Date: Thu, 6 Aug 2026 15:50:39 +0800 Subject: [PATCH] refactor(task): sink parser page-cap override into pipeline package (#17905) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Moves the canvas-debug parser page-cap injection out of the `task` orchestrator and into a **debug-agnostic** `pipeline` helper, so `PipelineExecutor` keeps only the orchestration skeleton (resolving one of the P1 review findings: the executor was overloaded with DSL/parser-param assembly). ### Changes - **`pipeline/parser_page_cap.go`** (new): - `BuildParserPageCapOverride(parserConfig, dsl, docType, capPages int, parserComponentName string, familyOf)` — injects the `ParserConfig[cpnID][family]["pages"]` cap through the same `override_params` channel production uses. The cap value and family resolution are injected by the caller, so the function carries no debug semantics and is reusable for any page-cap scenario. - `ExtractParserCpnID(dsl, parserComponentName)` — shared Parser cpnID discovery from (optionally enveloped) DSL. - `UnwrapCanvasDSL(raw []byte)` — exported single source of truth for stripping the `{"dsl": {...}}` canvas envelope. - `pipeline` does **not** import `component` (no reverse dependency); callers inject `component.ComponentNameParser` / `component.ParserFileFamily`. - **`task/pipeline_executor.go`**: removed `injectDebugPageCap` (the `debugPageCapPages = 2` constant stays in the task package). The debug branch now calls `pipeline.BuildParserPageCapOverride(...)`. - **`task/pipeline_executor.go` `warnUnknownComponentParams`**: fixed a production no-op bug — it passed the enveloped DSL straight to `ExtractAllComponentParams`, which silently errored and disabled the unknown-cpnID guard. It now unwraps the envelope first. - **`task/debug_result_dsl.go`**: reuses `pipeline.UnwrapCanvasDSL` instead of a third inline envelope-unwrap copy. ### Behavior No external debug-preview behavior changes. The three original invariants are preserved exactly: 1. explicit `pages` caps under `cpnID+family` are respected (not overwritten), 2. an empty family (unknown docType) is a no-op, 3. the injected shape is `[]any{[]any{1, capPages}}` (the `[]any`-of-`[]any` form `NormalizePDFPages` requires). ## Test plan - New `pipeline/parser_page_cap_test.go`: `BuildParserPageCapOverride` (inject / respect-existing / unknown-family no-op / no-Parser no-op), `ExtractParserCpnID` (enveloped + raw), `UnwrapCanvasDSL`. - `task/debug_test.go`: `TestInjectDebugPageCap` migrated to the new helper; new `TestWarnUnknownComponentParamsDetectsUnknownCPNFromEnvelope` captures the warning via `zaptest/observer` to prove the envelope no-op bug is fixed. - Both `internal/ingestion/pipeline` and `internal/ingestion/task` pass `build.sh --test` (unit tier). ## Notes - `TOKEN_CHUNKER_HANDOFF.md` is an unrelated untracked file and was deliberately **not** included in this PR. --- .../ingestion/pipeline/parser_page_cap.go | 109 ++++++++++++++++ .../pipeline/parser_page_cap_test.go | 120 ++++++++++++++++++ internal/ingestion/pipeline/pipeline.go | 23 +--- .../task/debug_pages_integration_test.go | 31 ++++- internal/ingestion/task/debug_result_dsl.go | 8 +- internal/ingestion/task/debug_test.go | 113 +++++++++++++++-- internal/ingestion/task/pipeline_executor.go | 100 +++------------ 7 files changed, 389 insertions(+), 115 deletions(-) create mode 100644 internal/ingestion/pipeline/parser_page_cap.go create mode 100644 internal/ingestion/pipeline/parser_page_cap_test.go diff --git a/internal/ingestion/pipeline/parser_page_cap.go b/internal/ingestion/pipeline/parser_page_cap.go new file mode 100644 index 0000000000..94cab3eef1 --- /dev/null +++ b/internal/ingestion/pipeline/parser_page_cap.go @@ -0,0 +1,109 @@ +package pipeline + +import ( + "encoding/json" + "fmt" +) + +// UnwrapCanvasDSL decodes a raw pipeline DSL and strips the optional canvas +// envelope {"dsl": {...}}, returning the inner components-carrying map. A raw +// (non-enveloped) DSL is returned unchanged. It is the []byte entry point used +// by helpers that need the inner DSL (ExtractParserCpnID, +// BuildParserPageCapOverride) so envelope-handling lives in exactly one place. +func UnwrapCanvasDSL(raw []byte) (map[string]any, error) { + var top map[string]any + if err := json.Unmarshal(raw, &top); err != nil { + return nil, fmt.Errorf("UnwrapCanvasDSL: decode: %w", err) + } + if top == nil { + return nil, errNilDSL + } + if env, ok := top["dsl"].(map[string]any); ok && len(env) > 0 { + return env, nil + } + return top, nil +} + +// ExtractParserCpnID returns the cpnID of the first component whose +// component_name equals parserComponentName, or "" when no such component +// exists or the DSL cannot be read. dsl may be enveloped; it is unwrapped +// first. +func ExtractParserCpnID(dsl []byte, parserComponentName string) string { + inner, err := UnwrapCanvasDSL(dsl) + if err != nil { + return "" + } + innerJSON, err := json.Marshal(inner) + if err != nil { + return "" + } + schemas, err := ExtractAllComponentParams(innerJSON) + if err != nil { + return "" + } + for _, s := range schemas { + if s.ComponentName == parserComponentName { + return s.CpnID + } + } + return "" +} + +// BuildParserPageCapOverride returns parserConfig with a page cap injected for +// the Parser component, keyed by its cpnID and the document's filetype family. +// +// It is debug-agnostic: capPages and familyOf are supplied by the caller, so +// the function carries no debug-specific semantics and is reusable for any +// page-cap scenario. docType is the uploaded file extension (e.g. "pdf"); +// familyOf maps it to a parser setup family (callers pass +// component.ParserFileFamily); parserComponentName is the Parser component +// name (callers pass component.ComponentNameParser) — kept as a parameter so +// the pipeline package does not import component. +// +// When no Parser component is found or the family is empty, the call is a +// no-op and parserConfig is returned unchanged. An explicit "pages" cap +// already present under cpnID+family is respected and left untouched — the +// cap is only a fallback. The injected shape is []any{[]any{1, capPages}}: a +// JSON-decoded list of 1-indexed inclusive ranges, the exact form the +// deepdoc/pdf parser's NormalizePDFPages consumes via +// ParserConfig[cpnID][family]["pages"]. +func BuildParserPageCapOverride( + parserConfig map[string]any, + dsl []byte, + docType string, + capPages int, + parserComponentName string, + familyOf func(string) string, +) map[string]any { + if parserConfig == nil { + parserConfig = map[string]any{} + } + parserCpnID := ExtractParserCpnID(dsl, parserComponentName) + if parserCpnID == "" { + return parserConfig + } + family := familyOf(docType) + if family == "" { + return parserConfig + } + // Respect an explicit page cap already present under cpnID + family. + if cpnEntry, ok := parserConfig[parserCpnID].(map[string]any); ok { + if famEntry, ok := cpnEntry[family].(map[string]any); ok { + if _, has := famEntry["pages"]; has { + return parserConfig + } + } + } + cpnEntry, ok := parserConfig[parserCpnID].(map[string]any) + if !ok { + cpnEntry = map[string]any{} + parserConfig[parserCpnID] = cpnEntry + } + famEntry, ok := cpnEntry[family].(map[string]any) + if !ok { + famEntry = map[string]any{} + cpnEntry[family] = famEntry + } + famEntry["pages"] = []any{[]any{1, capPages}} + return parserConfig +} diff --git a/internal/ingestion/pipeline/parser_page_cap_test.go b/internal/ingestion/pipeline/parser_page_cap_test.go new file mode 100644 index 0000000000..495fb89cc6 --- /dev/null +++ b/internal/ingestion/pipeline/parser_page_cap_test.go @@ -0,0 +1,120 @@ +package pipeline + +import ( + "reflect" + "testing" +) + +// testParserComponentName mirrors component.ComponentNameParser ("Parser"). +// The pipeline package deliberately does NOT import component (no reverse +// dependency); callers inject the name, so the test uses the literal here. +const testParserComponentName = "Parser" + +// envelopedDSL wraps a components map in the canvas envelope {"dsl": {...}}. +func envelopedDSL(components string) []byte { + return []byte(`{"dsl": {"components": ` + components + `}}`) +} + +// TestExtractParserCpnID verifies the Parser cpnID is discovered from both +// enveloped and raw DSL, and "" is returned when no Parser component exists. +func TestExtractParserCpnID(t *testing.T) { + // enveloped DSL with a Parser component. + dsl := envelopedDSL(`{"Parser:Abc": {"obj": {"component_name": "Parser", "params": {}}}}`) + if got := ExtractParserCpnID(dsl, testParserComponentName); got != "Parser:Abc" { + t.Fatalf("enveloped: want Parser:Abc, got %q", got) + } + + // no Parser component -> "". + dslNo := envelopedDSL(`{"Tokenizer:X": {"obj": {"component_name": "Tokenizer", "params": {}}}}`) + if got := ExtractParserCpnID(dslNo, testParserComponentName); got != "" { + t.Fatalf("no parser: want empty, got %q", got) + } + + // raw (non-enveloped) inner DSL is also accepted. + raw := []byte(`{"components": {"Parser:Z": {"obj": {"component_name": "Parser", "params": {}}}}}`) + if got := ExtractParserCpnID(raw, testParserComponentName); got != "Parser:Z" { + t.Fatalf("raw: want Parser:Z, got %q", got) + } +} + +// TestBuildParserPageCapOverride verifies the debug-agnostic page-cap override: +// normal injection, respect-existing-cap, unknown-family no-op, no-parser no-op. +func TestBuildParserPageCapOverride(t *testing.T) { + familyOf := func(ext string) string { + if ext == "pdf" { + return "pdf" + } + return "" + } + const docType = "pdf" + dsl := envelopedDSL(`{"Parser:Abc": {"obj": {"component_name": "Parser", "params": {}}}}`) + + // 1. normal: enveloped DSL + pdf -> pages cap injected. + pc := map[string]any{} + out := BuildParserPageCapOverride(pc, dsl, docType, 2, testParserComponentName, familyOf) + fam, ok := out["Parser:Abc"].(map[string]any)["pdf"].(map[string]any) + if !ok { + t.Fatalf("missing cpnID/family entry: %#v", out) + } + if !reflect.DeepEqual(fam["pages"], []any{[]any{1, 2}}) { + t.Fatalf("pages shape wrong: %#v", fam["pages"]) + } + + // 2. respect an explicit existing cap (fallback, not overwrite). + pc2 := map[string]any{ + "Parser:Abc": map[string]any{ + "pdf": map[string]any{"pages": []any{[]any{1, 99}}}, + }, + } + out2 := BuildParserPageCapOverride(pc2, dsl, docType, 2, testParserComponentName, familyOf) + fam2 := out2["Parser:Abc"].(map[string]any)["pdf"].(map[string]any) + if !reflect.DeepEqual(fam2["pages"], []any{[]any{1, 99}}) { + t.Fatalf("existing cap must be respected, got %#v", fam2["pages"]) + } + + // 3. unknown docType -> empty family -> no-op. + out3 := BuildParserPageCapOverride(map[string]any{}, dsl, "xyz", 2, testParserComponentName, familyOf) + if len(out3) != 0 { + t.Fatalf("unknown docType should be no-op, got %#v", out3) + } + + // 4. no Parser component -> no-op. + dslNoParser := envelopedDSL(`{"Tokenizer:X": {"obj": {"component_name": "Tokenizer", "params": {}}}}`) + out4 := BuildParserPageCapOverride(map[string]any{}, dslNoParser, docType, 2, testParserComponentName, familyOf) + if len(out4) != 0 { + t.Fatalf("no Parser should be no-op, got %#v", out4) + } +} + +// TestUnwrapCanvasDSL verifies the envelope is stripped and a raw DSL passes +// through unchanged; an empty/invalid payload errors. +func TestUnwrapCanvasDSL(t *testing.T) { + env := []byte(`{"dsl": {"components": {}}}`) + inner, err := UnwrapCanvasDSL(env) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if _, ok := inner["components"]; !ok { + t.Fatalf("envelope not stripped: %#v", inner) + } + + raw := []byte(`{"components": {}}`) + if _, err := UnwrapCanvasDSL(raw); err != nil { + t.Fatalf("raw DSL should pass through, got error: %v", err) + } + + // an empty-but-valid object passes through (matches UnwrapCanvasDSL: only + // a truly nil/unparseable DSL errors). + empty, err := UnwrapCanvasDSL([]byte(`{}`)) + if err != nil { + t.Fatalf("empty object should not error, got: %v", err) + } + if len(empty) != 0 { + t.Fatalf("empty object should yield empty map, got: %#v", empty) + } + + // invalid JSON errors. + if _, err := UnwrapCanvasDSL([]byte(`not json`)); err == nil { + t.Fatalf("invalid JSON should error") + } +} diff --git a/internal/ingestion/pipeline/pipeline.go b/internal/ingestion/pipeline/pipeline.go index 9f8abfb804..e106d81901 100644 --- a/internal/ingestion/pipeline/pipeline.go +++ b/internal/ingestion/pipeline/pipeline.go @@ -140,11 +140,10 @@ func WithProgressSink(s ProgressSink) PipelineOption { // It accepts either the inner canvas DSL or the template wrapper whose // top-level `dsl` field carries that canvas. func NewPipelineFromDSL(dsl []byte, taskID string, opts ...PipelineOption) (*Pipeline, error) { - var raw map[string]any - if err := json.Unmarshal(dsl, &raw); err != nil { - return nil, fmt.Errorf("pipeline: decode DSL: %w", err) - } - canvasDSL, err := unwrapCanvasDSL(raw) + // UnwrapCanvasDSL is the single source of truth for stripping the + // optional {"dsl": {...}} canvas envelope; it also reports a nil/unparseable + // DSL. + canvasDSL, err := UnwrapCanvasDSL(dsl) if err != nil { return nil, err } @@ -181,20 +180,6 @@ func (p *Pipeline) WithComponentFactory(factory runtime.ComponentFactory) *Pipel return p } -func unwrapCanvasDSL(raw map[string]any) (map[string]any, error) { - if len(raw) == 0 { - return nil, errNilDSL - } - if rawDSL, ok := raw["dsl"]; ok { - canvasDSL, ok := rawDSL.(map[string]any) - if !ok || len(canvasDSL) == 0 { - return nil, errNilDSL - } - return canvasDSL, nil - } - return raw, nil -} - func mergeInto(dst, src map[string]any) map[string]any { if src == nil { return dst diff --git a/internal/ingestion/task/debug_pages_integration_test.go b/internal/ingestion/task/debug_pages_integration_test.go index 6d38d2dbc2..d4c2694893 100644 --- a/internal/ingestion/task/debug_pages_integration_test.go +++ b/internal/ingestion/task/debug_pages_integration_test.go @@ -27,6 +27,9 @@ import ( "testing" "time" + "go.uber.org/zap" + "go.uber.org/zap/zaptest/observer" + "ragflow/internal/common" "ragflow/internal/dao" "ragflow/internal/entity" "ragflow/internal/ingestion/component" @@ -42,9 +45,9 @@ import ( // real pdfium parser, in-memory sqlite + storage): // // - Uncapped baseline: an explicit cpnID+family page cap of [[1, 1000000]] -// is supplied in ParserConfig. injectDebugPageCap must RESPECT it, so the +// is supplied in ParserConfig. pipeline.BuildParserPageCapOverride must RESPECT it, so the // parser reads every page. -// - Capped: no page cap is supplied, so injectDebugPageCap injects the +// - Capped: no page cap is supplied, so pipeline.BuildParserPageCapOverride injects the // debug default [[1, debugPageCapPages]], and the parser reads only the // leading pages. // @@ -79,7 +82,7 @@ func TestExecute_DebugViaEntry_HonorsPagesCap_Integration(t *testing.T) { // envelopeDSL is stored verbatim on the canvas (this is what production // persists), so loadDSLFromCanvas marshals the envelope and Run receives - // it. injectDebugPageCap must unwrap it to find the Parser cpnID. + // it. pipeline.BuildParserPageCapOverride must unwrap it to find the Parser cpnID. var envelope struct { DSL json.RawMessage `json:"dsl"` } @@ -129,7 +132,7 @@ func TestExecute_DebugViaEntry_HonorsPagesCap_Integration(t *testing.T) { t.Cleanup(func() { _ = realDB.Where("id = ?", canvasID).Delete(&entity.UserCanvas{}).Error }) // Explicit "parse all pages" cap (JSON-decoded []any form, the shape the - // parser actually consumes). injectDebugPageCap must respect it and leave + // parser actually consumes). pipeline.BuildParserPageCapOverride must respect it and leave // it untouched, so the parser reads every page of the PDF. allPages := map[string]any{ parserCpnID: map[string]any{ @@ -142,7 +145,7 @@ func TestExecute_DebugViaEntry_HonorsPagesCap_Integration(t *testing.T) { newDebugCtx := func(parserConfig map[string]any) *TaskContext { // A canvas-debug (dry-run) context carries no KB: KB.ID == "" is the // single debug signal. The executor then skips the persist stage and - // injects the debug page cap (see injectDebugPageCap, gated on + // injects the debug page cap (see pipeline.BuildParserPageCapOverride, gated on // KB.ID == ""). return &TaskContext{ Doc: entity.Document{ @@ -163,6 +166,15 @@ func TestExecute_DebugViaEntry_HonorsPagesCap_Integration(t *testing.T) { } } + // Capture warnings during the runs so we assert the envelope-unwrap fix + // did not regress into noisy warnings: with a well-formed template the + // parserConfig cpnID matches the DSL, so warnUnknownComponentParams must + // NOT emit its "not present in the pipeline DSL" warning. + core, recorded := observer.New(zap.NewAtomicLevelAt(zap.DebugLevel)) + oldLogger := common.Logger + common.Logger = zap.New(core) + defer func() { common.Logger = oldLogger }() + // Uncapped baseline: explicit pages=all → executor respects override. uncappedExec, err := NewPipelineExecutor(newDebugCtx(allPages), canvasID, 0) if err != nil { @@ -183,6 +195,15 @@ func TestExecute_DebugViaEntry_HonorsPagesCap_Integration(t *testing.T) { t.Fatalf("Execute (capped): %v", err) } + // A well-formed template's parserConfig cpnID matches the DSL, so the + // unknown-cpnID guard must stay silent. If it warned here, the envelope + // was not unwrapped (the pre-fix no-op regression). + for _, e := range recorded.All() { + if strings.Contains(e.Message, "not present in the pipeline DSL") { + t.Errorf("warnUnknownComponentParams emitted an unknown-cpnID warning during a well-formed debug run (envelope-unwrap regression?): %s", e.Message) + } + } + uncappedLen := len(joinedChunks(uncapped.Chunks)) cappedLen := len(joinedChunks(capped.Chunks)) if uncappedLen == 0 { diff --git a/internal/ingestion/task/debug_result_dsl.go b/internal/ingestion/task/debug_result_dsl.go index e9131e65d8..c6ebbda85e 100644 --- a/internal/ingestion/task/debug_result_dsl.go +++ b/internal/ingestion/task/debug_result_dsl.go @@ -30,6 +30,8 @@ import ( "encoding/json" "fmt" "strings" + + pipelinepkg "ragflow/internal/ingestion/pipeline" ) // ResultSink is an OPTIONAL capability a ProgressSink may implement to receive @@ -80,9 +82,11 @@ func BuildDebugResultDSL(dsl string, output map[string]any) (map[string]any, err if err := json.Unmarshal([]byte(dsl), &tpl); err != nil { return nil, fmt.Errorf("BuildDebugResultDSL: unmarshal dsl: %w", err) } + // Unwrap the canvas envelope via the shared helper so envelope handling + // lives in exactly one place (pipeline.UnwrapCanvasDSL). root := tpl - if nested, ok := tpl["dsl"].(map[string]any); ok { - root = nested + if inner, err := pipelinepkg.UnwrapCanvasDSL([]byte(dsl)); err == nil && inner != nil { + root = inner } components, ok := root["components"].(map[string]any) diff --git a/internal/ingestion/task/debug_test.go b/internal/ingestion/task/debug_test.go index d39681f097..ed028d68d0 100644 --- a/internal/ingestion/task/debug_test.go +++ b/internal/ingestion/task/debug_test.go @@ -23,9 +23,13 @@ import ( "os" "path/filepath" "reflect" + "strings" "testing" + "go.uber.org/zap" + "go.uber.org/zap/zaptest/observer" "gorm.io/gorm" + "ragflow/internal/common" "ragflow/internal/entity" "ragflow/internal/ingestion/component" "ragflow/internal/ingestion/pipeline" @@ -37,8 +41,8 @@ import ( // the debug signal used across the pipeline). The parser page cap is no longer // stored as a flat ParserConfig key here — flat keys are dropped by the // override_params merge and never reach the parser. It is injected at run time -// by injectDebugPageCap via Run's override_params channel (see -// TestInjectDebugPageCap). +// by pipeline.BuildParserPageCapOverride via Run's override_params channel +// (see TestInjectDebugPageCap). func TestNewDebugTaskContext_InjectsDebugID(t *testing.T) { taskCtx := NewDebugTaskContext("t1", "canvas-1", "doc.pdf", []byte("page one\fpage two\fpage three")) @@ -127,6 +131,10 @@ func TestExecute_DebugViaEntry(t *testing.T) { // storage JSON round-trip and the shape the deepdoc pdf parser consumes // (NormalizePDFPages requires []any, not a Go [][]int). // +// The DSL/parser-family knowledge now lives in pipeline.BuildParserPageCapOverride +// (debug-agnostic); this test drives that helper directly to pin the behavior +// the executor relies on. +// // It also pins the regression: a flat top-level "pages" key would be dropped // by the override_params merge and never reach the parser, so the cap must be // nested under the cpnID. @@ -159,9 +167,15 @@ func TestInjectDebugPageCap(t *testing.T) { t.Fatal("template has no Parser component") } + apply := func(cfg map[string]any, dslArg string, docType string) map[string]any { + return pipeline.BuildParserPageCapOverride( + cfg, []byte(dslArg), docType, debugPageCapPages, + component.ComponentNameParser, component.ParserFileFamily) + } + t.Run("pdf injects [1,2] under cpnID+family", func(t *testing.T) { parserConfig := map[string]any{} - injectDebugPageCap(dsl, parserConfig, "pdf") + apply(parserConfig, dsl, "pdf") famEntry, ok := parserConfig[parserCpnID].(map[string]any) if !ok { t.Fatalf("parserConfig[%q] = %T, want map[string]any", parserCpnID, parserConfig[parserCpnID]) @@ -177,7 +191,7 @@ func TestInjectDebugPageCap(t *testing.T) { t.Run("docx injects under docx family", func(t *testing.T) { parserConfig := map[string]any{} - injectDebugPageCap(dsl, parserConfig, "docx") + apply(parserConfig, dsl, "docx") famEntry, ok := parserConfig[parserCpnID].(map[string]any) if !ok { t.Fatalf("parserConfig[%q] = %T, want map[string]any", parserCpnID, parserConfig[parserCpnID]) @@ -193,7 +207,7 @@ func TestInjectDebugPageCap(t *testing.T) { t.Run("empty docType is a no-op", func(t *testing.T) { parserConfig := map[string]any{} - injectDebugPageCap(dsl, parserConfig, "") + apply(parserConfig, dsl, "") if len(parserConfig) != 0 { t.Errorf("parserConfig = %v, want empty (no family derivable from empty docType)", parserConfig) } @@ -205,7 +219,7 @@ func TestInjectDebugPageCap(t *testing.T) { "pdf": map[string]any{"parse_method": "deepdoc"}, }, } - injectDebugPageCap(dsl, parserConfig, "pdf") + apply(parserConfig, dsl, "pdf") famEntry := parserConfig[parserCpnID].(map[string]any) pdf := famEntry["pdf"].(map[string]any) if pdf["parse_method"] != "deepdoc" { @@ -218,11 +232,11 @@ func TestInjectDebugPageCap(t *testing.T) { t.Run("envelope dsl form (production shape)", func(t *testing.T) { // In production dsl is the canvas envelope {"dsl": {"components": ...}}, - // not the bare components map. injectDebugPageCap must still find the - // Parser cpnID after unwrapping. + // not the bare components map. The helper must still find the Parser + // cpnID after unwrapping. wrapped := fmt.Sprintf(`{"dsl":%s}`, string(envelope.DSL)) parserConfig := map[string]any{} - injectDebugPageCap(wrapped, parserConfig, "pdf") + apply(parserConfig, wrapped, "pdf") famEntry, ok := parserConfig[parserCpnID].(map[string]any) if !ok { t.Fatalf("parserConfig[%q] = %T, want map[string]any (envelope dsl must unwrap)", parserCpnID, parserConfig[parserCpnID]) @@ -244,7 +258,7 @@ func TestInjectDebugPageCap(t *testing.T) { "pdf": map[string]any{"pages": []any{[]any{1, 1000000}}}, }, } - injectDebugPageCap(dsl, parserConfig, "pdf") + apply(parserConfig, dsl, "pdf") famEntry := parserConfig[parserCpnID].(map[string]any) pdf := famEntry["pdf"].(map[string]any) if !reflect.DeepEqual(pdf["pages"], []any{[]any{1, 1000000}}) { @@ -252,3 +266,82 @@ func TestInjectDebugPageCap(t *testing.T) { } }) } + +// TestWarnUnknownComponentParamsDetectsUnknownCPNFromEnvelope pins the fix for +// the enveloped-DSL no-op bug: warnUnknownComponentParams previously passed the +// raw (enveloped) DSL straight to ExtractAllComponentParams, whose "components" +// key is nested under "dsl", so it errored and silently returned — never +// detecting unknown cpnIDs in production. The helper now unwraps the envelope +// first, so an unknown cpnID in parserConfig is actually surfaced. +func TestWarnUnknownComponentParamsDetectsUnknownCPNFromEnvelope(t *testing.T) { + core, recorded := observer.New(zap.NewAtomicLevelAt(zap.DebugLevel)) + old := common.Logger + common.Logger = zap.New(core) + defer func() { common.Logger = old }() + + // Enveloped DSL (production shape) carrying only a Parser component. + dsl := `{"dsl": {"components": {"Parser:Abc": {"obj": {"component_name": "Parser", "params": {}}}}}}` + // parserConfig references a cpnID NOT present in the DSL -> must be warned. + parserConfig := map[string]any{ + "Parser:Unknown": map[string]any{"pdf": map[string]any{}}, + } + + warnUnknownComponentParams(dsl, parserConfig) + + found := false + for _, e := range recorded.All() { + if strings.Contains(e.Message, "Parser:Unknown") { + found = true + } + } + if !found { + t.Fatalf("expected a warning about unknown cpnID Parser:Unknown (envelope DSL must be unwrapped); got logs: %v", recorded.All()) + } +} + +// TestBuildDebugResultDSL_Envelope pins that BuildDebugResultDSL unwraps the +// canvas envelope before reading "components" — the same shared +// pipeline.UnwrapCanvasDSL the cap override and warnUnknownComponentParams +// use. An enveloped DSL (production shape {"dsl": {...}}) must resolve the +// components map exactly like the equivalent raw (non-enveloped) DSL. +func TestBuildDebugResultDSL_Envelope(t *testing.T) { + const compID = "Parser:Abc" + rawDSL := `{"components": {"` + compID + `": {"obj": {"component_name": "Parser", "params": {"parse_method": "deepdoc"}}}}}` + output := map[string]any{ + "state": map[string]any{ + compID: map[string]any{"chunks": []any{map[string]any{"text": "hi"}}}, + }, + } + + // Raw (non-enveloped) DSL. + rawRes, err := BuildDebugResultDSL(rawDSL, output) + if err != nil { + t.Fatalf("raw DSL: %v", err) + } + rawComps, ok := rawRes["components"].(map[string]any) + if !ok { + t.Fatalf("raw DSL: components missing: %#v", rawRes) + } + if _, ok := rawComps[compID]; !ok { + t.Fatalf("raw DSL: components missing %q: %#v", compID, rawComps) + } + + // Enveloped DSL (production shape) must unwrap to the same result. + envDSL := `{"dsl": ` + rawDSL + `}` + envRes, err := BuildDebugResultDSL(envDSL, output) + if err != nil { + t.Fatalf("enveloped DSL: %v", err) + } + envComps, ok := envRes["components"].(map[string]any) + if !ok { + t.Fatalf("enveloped DSL: components missing (envelope not unwrapped?): %#v", envRes) + } + if _, ok := envComps[compID]; !ok { + t.Fatalf("enveloped DSL: components missing %q (envelope not unwrapped?): %#v", compID, envComps) + } + + // The two shapes must yield an identical component output. + if !reflect.DeepEqual(rawComps[compID], envComps[compID]) { + t.Fatalf("enveloped and raw DSL produced different results:\nraw=%#v\nenv=%#v", rawComps[compID], envComps[compID]) + } +} diff --git a/internal/ingestion/task/pipeline_executor.go b/internal/ingestion/task/pipeline_executor.go index 8316db14f4..c0cbfbc994 100644 --- a/internal/ingestion/task/pipeline_executor.go +++ b/internal/ingestion/task/pipeline_executor.go @@ -374,7 +374,22 @@ func warnUnknownComponentParams(dsl string, parserConfig map[string]any) { if len(parserConfig) == 0 { return } - schemas, err := pipelinepkg.ExtractAllComponentParams([]byte(dsl)) + // dsl arrives as the canvas ENVELOPE ({ "dsl": { "components": ... } }) in + // production, so it must be unwrapped before ExtractAllComponentParams + // runs (that helper expects the inner DSL). The previous direct call + // passed the enveloped DSL, whose "components" key is nested under "dsl", + // so it silently returned an error and made this guard a no-op. + inner, err := pipelinepkg.UnwrapCanvasDSL([]byte(dsl)) + if err != nil { + common.Warn(fmt.Sprintf("warnUnknownComponentParams: cannot parse DSL to validate component params: %v", err)) + return + } + innerJSON, err := json.Marshal(inner) + if err != nil { + common.Warn(fmt.Sprintf("warnUnknownComponentParams: cannot re-encode DSL: %v", err)) + return + } + schemas, err := pipelinepkg.ExtractAllComponentParams(innerJSON) if err != nil { common.Warn(fmt.Sprintf("warnUnknownComponentParams: cannot parse DSL to validate component params: %v", err)) return @@ -472,9 +487,12 @@ func (s *PipelineExecutor) runPipelineWithDSL(ctx context.Context, dsl string) ( // and the document's filetype family. It is NOT passed through pipeline // inputs: the parser selects pages from ParserConfig[cpnID][family] // ["pages"] (a list of 1-indexed inclusive ranges), exactly mirroring - // NormalizeParserConfigPages / pdf_pages_test.go. See injectDebugPageCap. + // NormalizeParserConfigPages / pdf_pages_test.go. The DSL/parser-family + // knowledge now lives in pipeline.BuildParserPageCapOverride. if debug { - injectDebugPageCap(dsl, parserConfig, s.taskCtx.Doc.Type) + parserConfig = pipelinepkg.BuildParserPageCapOverride( + parserConfig, []byte(dsl), s.taskCtx.Doc.Type, + debugPageCapPages, component.ComponentNameParser, component.ParserFileFamily) } // Component params from Doc.ParserConfig — including the tenant LLM id @@ -511,79 +529,3 @@ func (s *PipelineExecutor) runPipelineWithDSL(ctx context.Context, dsl string) ( // inclusive range [1, debugPageCapPages], matching the production // ParserConfig[cpnID][filetype]["pages"] shape (see NormalizeParserConfigPages). const debugPageCapPages = 2 - -// injectDebugPageCap wires the canvas-debug page cap into parserConfig using -// the SAME channel the production ParserConfig travels through: Run's -// override_params (the 3rd argument), keyed by the Parser component's cpnID -// and the document's filetype family. It must NOT be passed as a pipeline -// input — the parser selects pages from ParserConfig[cpnID][family]["pages"], -// which the deepdoc/pdf parser consumes as a list of page ranges. -// -// dsl is the raw pipeline DSL (used only to discover the Parser component's -// cpnID); docType is the uploaded file's extension (e.g. "pdf"). When no -// Parser component can be found, or the docType yields no known family, the -// call is a no-op (the run parses everything, which is safe). -// -// dsl arrives as the canvas ENVELOPE ({ "dsl": { "components": ... } }) in -// production (loadDSLFromCanvas marshals canvas.DSL), so it is unwrapped the -// same way NewPipelineFromDSL does before ExtractAllComponentParams runs. -// An explicit page cap already present in parserConfig (keyed by cpnID + -// family) is respected and left untouched — the debug default is only a -// fallback, so a debug run can honour a narrower or wider caller-supplied cap. -func injectDebugPageCap(dsl string, parserConfig map[string]any, docType string) { - // Unwrap the canvas envelope to the inner components map. - var raw map[string]any - if err := json.Unmarshal([]byte(dsl), &raw); err != nil { - return - } - if env, ok := raw["dsl"].(map[string]any); ok && len(env) > 0 { - raw = env - } - inner, err := json.Marshal(raw) - if err != nil { - return - } - schemas, err := pipelinepkg.ExtractAllComponentParams(inner) - if err != nil { - return - } - var parserCpnID string - for _, s := range schemas { - if s.ComponentName == component.ComponentNameParser { - parserCpnID = s.CpnID - break - } - } - if parserCpnID == "" { - return - } - family := component.ParserFileFamily(docType) - if family == "" { - return - } - // Respect an explicit page cap already present under cpnID + family. - if cpnEntry, ok := parserConfig[parserCpnID].(map[string]any); ok { - if famEntry, ok := cpnEntry[family].(map[string]any); ok { - if _, has := famEntry["pages"]; has { - return - } - } - } - cpnEntry, ok := parserConfig[parserCpnID].(map[string]any) - if !ok { - cpnEntry = map[string]any{} - parserConfig[parserCpnID] = cpnEntry - } - famEntry, ok := cpnEntry[family].(map[string]any) - if !ok { - famEntry = map[string]any{} - cpnEntry[family] = famEntry - } - // pages is delivered as a JSON-decoded map — the very shape a ParserConfig - // arrives in from the API/storage JSON round-trip: a []any of [from,to] - // pairs. The deepdoc/pdf parser's NormalizePDFPages requires this - // []any-of-[]any form (not a Go [][]int), so it is built explicitly here. - // The shallow override_params merge in applyOverrideParams preserves this - // shape all the way to ConfigureFromSetup, so the cap is honoured. - famEntry["pages"] = []any{[]any{1, debugPageCapPages}} -}