Fix Pipeline.Run callers and rename SetupOverrides (#16973)

Adapt Pipeline.Run call sites to the new signature and rename the
SetupOverrides family to OverrideParams for consistency.

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)

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)
 	}
 

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) {

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))

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)

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")
 	}

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)
 	}

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)
 			}

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
 	}

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)
 	}