diff --git a/internal/agent/canvas/compile.go b/internal/agent/canvas/compile.go index 50c6c1beed..cf53976064 100644 --- a/internal/agent/canvas/compile.go +++ b/internal/agent/canvas/compile.go @@ -75,15 +75,15 @@ type CompileOptions struct { // graph does not pause on completion and force an extra, needless // ResumeWithData round. InterruptAfterNonTerminal bool - // SetupOverrides is a run-level override map keyed by cpnID. Each - // component's `params["setups"]` is merged only with its own entry + // OverrideParams is a run-level override map keyed by cpnID. Each + // component's `params` is merged only with its own entry // (an arbitrary string-keyed map); the override wins on top-level key // collision (see node_body.go mergeSetups). Components absent from the // map are left untouched. Used by the ingestion pipeline so a single // Pipeline.Run can override the DSL-baked component setups without - // mutating the shared *Canvas (see node_body.go applySetupOverrides / + // mutating the shared *Canvas (see node_body.go applyOverrideParams / // mergeSetups). - SetupOverrides map[string]any + OverrideParams map[string]any } // CompileOption mutates a CompileOptions before the compile runs. @@ -130,12 +130,12 @@ func WithInterruptAfterNonTerminalCpn() CompileOption { return func(o *CompileOptions) { o.InterruptAfterNonTerminal = true } } -// WithSetupOverrides attaches a run-level setups override map (keyed by +// WithOverrideParams attaches a run-level setups override map (keyed by // cpnID) to the compile. Each component's `params["setups"]` is merged with // its own entry at compile time (run-level wins on key collision, see // node_body.go mergeSetups). Passing nil is a no-op. -func WithSetupOverrides(m map[string]any) CompileOption { - return func(o *CompileOptions) { o.SetupOverrides = m } +func WithOverrideParams(m map[string]any) CompileOption { + return func(o *CompileOptions) { o.OverrideParams = m } } // Compile builds the eino Workflow from the Canvas and returns the @@ -217,8 +217,8 @@ func Compile(ctx context.Context, c *Canvas, opts ...CompileOption) (*CompiledCa // component's `params["setups"]` is merged with its own entry inside // buildNodeBody. The override is keyed by cpnID; the canvas package // never imports ingestion. - if cfg.SetupOverrides != nil { - ctx = withSetupOverrides(ctx, cfg.SetupOverrides) + if cfg.OverrideParams != nil { + ctx = withOverrideParams(ctx, cfg.OverrideParams) } wf, err := BuildWorkflow(ctx, c) diff --git a/internal/agent/canvas/compile_setup_override_test.go b/internal/agent/canvas/compile_setup_override_test.go index b3c1de7bc0..dfea2edef6 100644 --- a/internal/agent/canvas/compile_setup_override_test.go +++ b/internal/agent/canvas/compile_setup_override_test.go @@ -22,16 +22,16 @@ import ( "ragflow/internal/agent/runtime" ) -// TestCompile_SetupOverrides exercises the full canvas-level wiring: +// TestCompile_OverrideParams exercises the full canvas-level wiring: // -// canvas.Compile(ctx, dsl, WithSetupOverrides(override)) +// canvas.Compile(ctx, dsl, WithOverrideParams(override)) // // threads the cpnID-keyed override through ctx into -// BuildWorkflow → buildNodeBody → applySetupOverrides → mergeSetups, so +// BuildWorkflow → buildNodeBody → applyOverrideParams → mergeSetups, so // each component's factory receives its own merged params["setups"]. Only // the entry for a component's own cpnID applies; components absent from the // override map keep their base params (no spurious "setups" injected). -func TestCompile_SetupOverrides(t *testing.T) { +func TestCompile_OverrideParams(t *testing.T) { captured := map[string]map[string]any{} // component_name -> params received by factory factory := func(name string, params map[string]any) (runtime.Component, error) { // Deep-shallow copy so later mutations don't hide what the @@ -92,7 +92,7 @@ func TestCompile_SetupOverrides(t *testing.T) { } ctx := WithComponentFactory(context.Background(), factory) - if _, err := Compile(ctx, dsl, WithSetupOverrides(override)); err != nil { + if _, err := Compile(ctx, dsl, WithOverrideParams(override)); err != nil { t.Fatalf("Compile: %v", err) } diff --git a/internal/agent/canvas/node_body.go b/internal/agent/canvas/node_body.go index cffd5d67fd..7d74711274 100644 --- a/internal/agent/canvas/node_body.go +++ b/internal/agent/canvas/node_body.go @@ -74,7 +74,7 @@ type nodeBodyFn = func(ctx context.Context, in map[string]any) (map[string]any, // Outputs bucket. UserFillUpNodeBody tags its output itself so the // interrupt-driven branch still attributes the resume payload to the // right cpn. -// ctxKeySetupOverrides carries the run-level setups override map into +// ctxKeyOverrideParams carries the run-level setups override map into // BuildWorkflow so a component's `params["setups"]` can be merged with it // at compile time. The map is keyed by cpnID; each component only sees the // entry for its own id (an arbitrary string-keyed map). It mirrors the ctx @@ -82,30 +82,30 @@ type nodeBodyFn = func(ctx context.Context, in map[string]any) (map[string]any, // (componentFactoryFromContext): the override is threaded through // canvas.Compile → BuildWorkflow → buildNodeBody without the canvas // package ever importing the ingestion layer. -const ctxKeySetupOverrides ctxKey = "canvas_setup_overrides" +const ctxKeyOverrideParams ctxKey = "canvas_override_params" -// withSetupOverrides attaches a run-level setups override map to ctx. It is +// withOverrideParams attaches a run-level setups override map to ctx. It is // a no-op when m is nil so callers can pass a possibly-nil run parameter // straight through. -func withSetupOverrides(ctx context.Context, m map[string]any) context.Context { +func withOverrideParams(ctx context.Context, m map[string]any) context.Context { if m == nil { return ctx } - return context.WithValue(ctx, ctxKeySetupOverrides, m) + return context.WithValue(ctx, ctxKeyOverrideParams, m) } -func setupOverridesFromContext(ctx context.Context) map[string]any { - m, _ := ctx.Value(ctxKeySetupOverrides).(map[string]any) +func overrideParamsFromContext(ctx context.Context) map[string]any { + m, _ := ctx.Value(ctxKeyOverrideParams).(map[string]any) return m } -// applySetupOverrides returns a clone of params with the per-component +// applyOverrideParams returns a clone of params with the per-component // setups override (already resolved for this cpnID by the caller) merged // into params["setups"]. The override wins on top-level key collisions. The // original params map is never mutated — the merge result is a fresh map — // because the params come from the shared *Canvas and a per-run override // must not leak into the next Run on the same Pipeline. -func applySetupOverrides(params, cpnOverride map[string]any) map[string]any { +func applyOverrideParams(params, cpnOverride map[string]any) map[string]any { if len(cpnOverride) == 0 { return params } @@ -136,11 +136,11 @@ func mergeSetups(base, override map[string]any) map[string]any { } func buildNodeBody(ctx context.Context, cpnID, name string, params map[string]any) (nodeBodyFn, error) { - if overrides := setupOverridesFromContext(ctx); len(overrides) > 0 { + if overrides := overrideParamsFromContext(ctx); len(overrides) > 0 { // overrides is keyed by cpnID; a component only sees its own // entry. Components absent from the map are left untouched. if cpnOverride, ok := overrides[cpnID].(map[string]any); ok && len(cpnOverride) > 0 { - params = applySetupOverrides(params, cpnOverride) + params = applyOverrideParams(params, cpnOverride) } } if isLegacyNoOp(name) { diff --git a/internal/agent/canvas/node_body_setup_override_test.go b/internal/agent/canvas/node_body_setup_override_test.go index c7ab5b68f2..5a90b4db6e 100644 --- a/internal/agent/canvas/node_body_setup_override_test.go +++ b/internal/agent/canvas/node_body_setup_override_test.go @@ -33,13 +33,13 @@ func (s *stubComponent) Invoke(_ context.Context, in map[string]any) (map[string return in, nil } -// TestBuildNodeBody_SetupOverrides asserts that a run-level setups override +// TestBuildNodeBody_OverrideParams asserts that a run-level setups override // (threaded via ctx, keyed by cpnID) is merged into the component's // `params["setups"]` before the factory is called. Only the entry for the // component's own cpnID applies. The merge is shallow: a top-level key // present in the override fully replaces the base entry for that key // (no inner deep-merge), while base keys absent from the override survive. -func TestBuildNodeBody_SetupOverrides(t *testing.T) { +func TestBuildNodeBody_OverrideParams(t *testing.T) { captured := map[string]any{} factory := func(name string, params map[string]any) (runtime.Component, error) { // Deep-shallow copy so later mutations by the builder don't @@ -84,7 +84,7 @@ func TestBuildNodeBody_SetupOverrides(t *testing.T) { } ctx := WithComponentFactory(context.Background(), factory) - ctx = withSetupOverrides(ctx, override) + ctx = withOverrideParams(ctx, override) body, err := buildNodeBody(ctx, "cpn-parser", "Parser", baseParams) if err != nil { @@ -120,10 +120,10 @@ func TestBuildNodeBody_SetupOverrides(t *testing.T) { } } -// TestBuildNodeBody_SetupOverridesNilIsNoOp asserts that with no override in +// TestBuildNodeBody_OverrideParamsNilIsNoOp asserts that with no override in // ctx the component receives exactly its base params (no spurious setups key // injected, and the original map is untouched). -func TestBuildNodeBody_SetupOverridesNilIsNoOp(t *testing.T) { +func TestBuildNodeBody_OverrideParamsNilIsNoOp(t *testing.T) { captured := map[string]any{} factory := func(name string, params map[string]any) (runtime.Component, error) { cp := make(map[string]any, len(params)) diff --git a/internal/ingestion/pipeline/pipeline.go b/internal/ingestion/pipeline/pipeline.go index 3e2259a628..e1c1d7acbf 100644 --- a/internal/ingestion/pipeline/pipeline.go +++ b/internal/ingestion/pipeline/pipeline.go @@ -212,23 +212,13 @@ var defaultCheckpointTTL = 24 * time.Hour // There is no pipeline-layer partial resume entry point: execution always // starts from the graph entry and component-level replay decisions belong to // the components themselves. -func (p *Pipeline) Run(ctx context.Context, inputs map[string]any, setups ...map[string]any) (map[string]any, error) { +func (p *Pipeline) Run(ctx context.Context, inputs map[string]any, override_params map[string]any) (map[string]any, error) { if p == nil { return nil, fmt.Errorf("pipeline: Run on nil pipeline") } if p.canvas == nil { return nil, fmt.Errorf("pipeline: canvas is nil") } - // runSetups, when non-nil, overrides components' DSL-baked - // `params["setups"]` at compile time. It is keyed by cpnID; each - // component is merged only with its own entry, and within that entry a - // top-level key fully replaces the base entry for that key (see - // canvas.mergeSetups). It is variadic so existing callers that pass - // only (ctx, inputs) keep working. - var runSetups map[string]any - if len(setups) > 0 { - runSetups = setups[0] - } if runtime.DefaultFactory() == nil { runtime.InstallDefaultRegistryFactory() } @@ -269,8 +259,10 @@ func (p *Pipeline) Run(ctx context.Context, inputs map[string]any, setups ...map ) } // Run-level setups (keyed by cpnID) override the DSL-baked component - // setups at compile time (higher priority; see canvas.WithSetupOverrides). - compileOpts = append(compileOpts, canvas.WithSetupOverrides(runSetups)) + // setups at compile time (higher priority; see canvas.WithOverrideParams). + if override_params != nil { + compileOpts = append(compileOpts, canvas.WithOverrideParams(override_params)) + } compiled, err := canvas.Compile(compileCtx, p.canvas, compileOpts...) if err != nil { return nil, fmt.Errorf("pipeline: Run: compile canvas: %w", err) diff --git a/internal/ingestion/pipeline/pipeline_test.go b/internal/ingestion/pipeline/pipeline_test.go index d1da9bb83d..a6c79f93cc 100644 --- a/internal/ingestion/pipeline/pipeline_test.go +++ b/internal/ingestion/pipeline/pipeline_test.go @@ -98,7 +98,7 @@ func TestPipelineRunHappyPath(t *testing.T) { t.Fatalf("NewPipelineFromDSL: %v", err) } - out, err := pipe.Run(context.Background(), map[string]any{"name": "doc-canvas"}) + out, err := pipe.Run(context.Background(), map[string]any{"name": "doc-canvas"}, nil) if err != nil { t.Fatalf("Run: %v", err) } @@ -119,7 +119,7 @@ func TestPipelineRunHappyPath(t *testing.T) { func TestPipelineRunNilPipeline(t *testing.T) { var p *Pipeline - if _, err := p.Run(context.Background(), nil); err == nil { + if _, err := p.Run(context.Background(), nil, nil); err == nil { t.Fatal("expected error for nil pipeline") } } @@ -144,7 +144,7 @@ func TestPipelineRunStageErrorBubbles(t *testing.T) { t.Fatalf("NewPipelineFromDSL: %v", err) } - if _, err := pipe.Run(context.Background(), map[string]any{"name": "x"}); err == nil { + if _, err := pipe.Run(context.Background(), map[string]any{"name": "x"}, nil); err == nil { t.Fatal("expected stage error") } } @@ -250,7 +250,7 @@ func TestPipelineRun_InstanceFactoryOverridesDefaultFactory(t *testing.T) { return &factorySentinelStage{marker: "instance"}, nil }) - out, err := pipe.Run(context.Background(), map[string]any{"name": "doc"}) + out, err := pipe.Run(context.Background(), map[string]any{"name": "doc"}, nil) if err != nil { t.Fatalf("Run: %v", err) } @@ -304,7 +304,7 @@ func TestPipelineRun_TaskScopedFactoriesDoNotLeakAcrossConcurrentPipelines(t *te results := make(chan result, 2) run := func(pipe *Pipeline) { defer wg.Done() - out, err := pipe.Run(context.Background(), map[string]any{"name": "doc"}) + out, err := pipe.Run(context.Background(), map[string]any{"name": "doc"}, nil) if err != nil { results <- result{err: err} return @@ -364,7 +364,7 @@ func TestPipelineRunResumableAutoResumes(t *testing.T) { t.Fatalf("NewPipelineFromDSL: %v", err) } - out, err := pipe.Run(context.Background(), map[string]any{"name": "doc-resume"}) + out, err := pipe.Run(context.Background(), map[string]any{"name": "doc-resume"}, nil) if err != nil { t.Fatalf("Run: %v", err) } @@ -427,7 +427,7 @@ func TestPipelineRunResumableCrossRunResume(t *testing.T) { // Run 1: terminal B errors (oneShotErrStage n=1), simulating a crash. // Non-terminal A's checkpoint + interrupt persist because the error path // does not call ClearInterruptID or store.Delete. - _, err = pipe.Run(context.Background(), map[string]any{"name": "doc-cross-run"}) + _, err = pipe.Run(context.Background(), map[string]any{"name": "doc-cross-run"}, nil) if err == nil { t.Fatal("Run 1: expected error from simulated crash, got nil") } @@ -441,7 +441,7 @@ func TestPipelineRunResumableCrossRunResume(t *testing.T) { // Run 2: resume from after A via tracker.GetInterruptID. A is skipped; // B's oneShotErrStage (n=2) delegates to its embedded mock successfully. - _, err = pipe.Run(context.Background(), map[string]any{"name": "doc-cross-run"}) + _, err = pipe.Run(context.Background(), map[string]any{"name": "doc-cross-run"}, nil) if err != nil { t.Fatalf("Run 2: expected recovery success, got error: %v", err) } @@ -473,7 +473,7 @@ func TestPipelineRun_RequireResumeRejectsWithoutStore(t *testing.T) { if err != nil { t.Fatalf("NewPipelineFromDSL: %v", err) } - _, err = pipe.Run(context.Background(), map[string]any{"name": "doc"}) + _, err = pipe.Run(context.Background(), map[string]any{"name": "doc"}, nil) if !errors.Is(err, ErrResumeUnavailable) { t.Fatalf("expected ErrResumeUnavailable, got %v", err) } @@ -534,7 +534,7 @@ func TestPipelineRunForwardsProgressToSink(t *testing.T) { if err != nil { t.Fatalf("NewPipelineFromDSL: %v", err) } - if _, err := pipe.Run(context.Background(), map[string]any{"name": "doc-sink"}); err != nil { + if _, err := pipe.Run(context.Background(), map[string]any{"name": "doc-sink"}, nil); err != nil { t.Fatalf("Run: %v", err) } @@ -626,7 +626,7 @@ func TestRunPlain_WithTracker_Success(t *testing.T) { t.Fatalf("NewPipelineFromDSL: %v", err) } - _, err = pipe.Run(context.Background(), map[string]any{"name": "doc"}) + _, err = pipe.Run(context.Background(), map[string]any{"name": "doc"}, nil) if err != nil { t.Fatalf("Run: %v", err) } @@ -657,7 +657,7 @@ func TestRunPlain_WithTracker_Error(t *testing.T) { t.Fatalf("NewPipelineFromDSL: %v", err) } - _, err = pipe.Run(context.Background(), map[string]any{"name": "doc"}) + _, err = pipe.Run(context.Background(), map[string]any{"name": "doc"}, nil) if err == nil { t.Fatal("expected stage error, got nil") } diff --git a/internal/ingestion/pipeline/real_storage_integration_test.go b/internal/ingestion/pipeline/real_storage_integration_test.go index 71e5751649..977f908f5c 100644 --- a/internal/ingestion/pipeline/real_storage_integration_test.go +++ b/internal/ingestion/pipeline/real_storage_integration_test.go @@ -95,7 +95,7 @@ func TestPipelineRun_TemplateGeneral_RealMySQLMinIO_OutputShape(t *testing.T) { } out, err := pipe.Run(context.Background(), map[string]any{ "doc_id": docID, - }) + }, nil) if err != nil { t.Fatalf("Run: %v", err) } diff --git a/internal/ingestion/pipeline/template_integration_test.go b/internal/ingestion/pipeline/template_integration_test.go index 2b90f79b58..7dc5215b7a 100644 --- a/internal/ingestion/pipeline/template_integration_test.go +++ b/internal/ingestion/pipeline/template_integration_test.go @@ -87,7 +87,7 @@ func TestPipelineRun_TemplateGeneral_RealComponents(t *testing.T) { attachFixedEmbedderFactory(t, pipe) out, err := pipe.Run(context.Background(), map[string]any{ "doc_id": docID, - }) + }, nil) if err != nil { t.Fatalf("Run: %v", err) } @@ -178,7 +178,7 @@ func TestPipelineRun_TemplateOne_RealComponents(t *testing.T) { attachFixedEmbedderFactory(t, pipe) out, err := pipe.Run(context.Background(), map[string]any{ "doc_id": docID, - }) + }, nil) if err != nil { t.Fatalf("Run: %v", err) } @@ -266,7 +266,7 @@ func TestPipelineRun_TemplateOne_RealComponents_PDFDeepdocChunking(t *testing.T) attachFixedEmbedderFactory(t, pipe) out, err := pipe.Run(context.Background(), map[string]any{ "doc_id": docID, - }) + }, nil) if err != nil { t.Fatalf("Run: %v", err) } @@ -364,7 +364,7 @@ func TestPipelineRun_TemplateManual_RealComponents(t *testing.T) { attachFixedEmbedderFactory(t, pipe) out, err := pipe.Run(context.Background(), map[string]any{ "doc_id": docID, - }) + }, nil) if err != nil { t.Fatalf("Run: %v", err) } @@ -461,7 +461,7 @@ func TestPipelineRun_TemplateLaws_RealComponents(t *testing.T) { attachFixedEmbedderFactory(t, pipe) out, err := pipe.Run(context.Background(), map[string]any{ "doc_id": docID, - }) + }, nil) if err != nil { t.Fatalf("Run: %v", err) } @@ -543,7 +543,7 @@ func TestPipelineRun_TemplatePaper_RealComponents(t *testing.T) { attachFixedEmbedderFactory(t, pipe) out, err := pipe.Run(context.Background(), map[string]any{ "doc_id": docID, - }) + }, nil) if err != nil { t.Fatalf("Run: %v", err) } @@ -623,7 +623,7 @@ func TestPipelineRun_TemplateBook_RealComponents(t *testing.T) { attachFixedEmbedderFactory(t, pipe) out, err := pipe.Run(context.Background(), map[string]any{ "doc_id": docID, - }) + }, nil) if err != nil { t.Fatalf("Run: %v", err) } @@ -745,7 +745,7 @@ func TestPipelineRun_TemplateResume_RealComponents(t *testing.T) { out, err := pipe.Run(context.Background(), map[string]any{ "doc_id": docID, "llm_id": model + "@openai", - }) + }, nil) if err != nil { t.Fatalf("Run: %v", err) } @@ -822,7 +822,7 @@ func TestPipelineRun_AllIngestionTemplates_RealComponentsSmoke(t *testing.T) { attachFixedEmbedderFactory(t, pipe) out, err := pipe.Run(context.Background(), map[string]any{ "doc_id": docID, - }) + }, nil) if err != nil { t.Fatalf("Run: %v", err) } diff --git a/internal/ingestion/task/pipeline_executor_defaults.go b/internal/ingestion/task/pipeline_executor_defaults.go index eb39367a03..519cfe8c7d 100644 --- a/internal/ingestion/task/pipeline_executor_defaults.go +++ b/internal/ingestion/task/pipeline_executor_defaults.go @@ -82,7 +82,7 @@ func (s *PipelineExecutor) defaultRunPipeline(ctx context.Context, dsl string) ( inputs["tenant_id"] = s.taskCtx.Tenant.ID inputs["kb_id"] = s.taskCtx.KB.ID - output, err := pipe.Run(ctx, inputs) + output, err := pipe.Run(ctx, inputs, nil) if err != nil { return nil, dsl, err } diff --git a/internal/ingestion/task/pipeline_real_integration_test.go b/internal/ingestion/task/pipeline_real_integration_test.go index 24746eac56..9ba4890442 100644 --- a/internal/ingestion/task/pipeline_real_integration_test.go +++ b/internal/ingestion/task/pipeline_real_integration_test.go @@ -354,7 +354,7 @@ func TestRunPipeline_RealPipelineOutput_ProducesIndexFields(t *testing.T) { pipelineOut, err := pipe.Run(context.Background(), map[string]any{ "doc_id": docID, - }) + }, nil) if err != nil { t.Fatalf("pipeline Run: %v", err) }