fix(ingestion): discard stale checkpoint when the DSL is edited before resume (#17351)

## Summary - Root cause: Pipeline.Run keys the eino checkpoint by taskID only. Resuming a failed run after the user edits the pipeline DSL recompiles a graph with different topology, but the old checkpoint (bound to the previous graph's node ids / wiring) is restored, causing eino to error. - Fix: fingerprint the DSL file (full canvas DSL) and the runtime override_params, persisted next to the eino checkpoint. On resume, if either fingerprint differs, discard the stale checkpoint + interrupt marker and re-run from scratch. The warning log distinguishes a DSL-file edit from a runtime-override edit. ## Test plan - TestPipelineRunResumableDSLChanged: editing the DSL between runs discards the checkpoint and re-runs from scratch. - TestPipelineRunResumableOverrideChanged: editing only the runtime override does the same. - TestClassifyDSLChange: unit-tests the mismatch-reason classifier. - bash build.sh --test ./internal/ingestion/pipeline/... passes. ## Notes - Component-code changes (e.g. a component's output contract) are not covered by the DSL fingerprint; that is a separate, smaller-blast-radius gap noted in code comments. --------- Signed-off-by: xugangqiang <xugangqiang@hotmail.com> Co-authored-by: CodeBuddy <noreply@tencent.com>
2026-07-25 09:53:29 +08:00 · 2026-07-24 19:18:07 +08:00
parent a154d751b4
commit 75c9af361f
2 changed files with 440 additions and 3 deletions
--- a/internal/ingestion/pipeline/pipeline.go
+++ b/internal/ingestion/pipeline/pipeline.go
@@ -18,6 +18,8 @@ package pipeline

 import (
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -36,7 +38,13 @@ import (

 // Pipeline is a compiled ingestion canvas plus task-scoped metadata.
 type Pipeline struct {
-	taskID     string
+	taskID string
+	// rawDSL holds the canonical (key-sorted) JSON of the canvas DSL this
+	// pipeline was compiled from. It is the basis for the resume-time DSL
+	// fingerprint (guardDSLChange): any edit to the template DSL changes
+	// these bytes, so a stale checkpoint can be detected and discarded
+	// instead of being resumed against an incompatible graph.
+	rawDSL     []byte
 	documentID string // owning document; progress is mirrored back to the
 	// document table so the existing GET /api/v1/datasets/{dataset_id}/documents
 	// endpoint (which reads document.progress/run/progress_msg) reflects the
@@ -144,9 +152,17 @@ func NewPipelineFromDSL(dsl []byte, taskID string, opts ...PipelineOption) (*Pip
 	if err != nil {
 		return nil, fmt.Errorf("pipeline: decode canvas DSL: %w", err)
 	}
+	// Capture the canonical canvas DSL bytes for the resume-time DSL
+	// fingerprint. json.Marshal sorts map keys, so this is stable across
+	// re-decodes of the same logical DSL (formatting-independent).
+	rawBytes, err := json.Marshal(canvasDSL)
+	if err != nil {
+		return nil, fmt.Errorf("pipeline: canonicalize DSL: %w", err)
+	}
 	p := &Pipeline{
 		taskID: taskID,
 		canvas: cnv,
+		rawDSL: rawBytes,
 	}
 	for _, o := range opts {
 		o(p)
@@ -208,6 +224,165 @@ func cloneMapOrEmpty(m map[string]any) map[string]any {
 // the TTL only guards against leaks from crashed runs that never clean up.
 var defaultCheckpointTTL = 24 * time.Hour

+// dslKeySuffix / ovfKeySuffix derive the two DSL-fingerprint keys from a
+// checkpoint id. Both live in the same CheckPointStore as the eino payload
+// (under cpID+dslKeySuffix / cpID+ovfKeySuffix) so a resume can compare the
+// current DSL against the one that wrote the checkpoint — without depending
+// on the RunTracker being injected. eino only ever reads/writes the exact
+// cpID, so these sibling keys never collide with eino's checkpoint.
+//
+// The split is deliberate: dslKeySuffix fingerprints the DSL FILE (the full
+// canvas DSL — topology, component params, everything the user edits in the
+// template), while ovfKeySuffix fingerprints only the runtime override_params
+// (Doc.ParserConfig + injected LLM id). This lets the mismatch warning say
+// whether the DSL file changed or the runtime override changed, instead of an
+// ambiguous "params changed".
+const (
+	dslKeySuffix = ":dsl"
+	ovfKeySuffix = ":ovf"
+)
+
+// dslFileFingerprint returns a stable hash of the FULL canvas DSL this
+// pipeline was compiled from. It captures topology (components, edges, graph
+// grouping) AND the per-component params/defaults — i.e. everything the user
+// can edit in the DSL template file. The raw DSL bytes are key-sorted at
+// decode time, so logical edits (not whitespace) drive the value.
+//
+// It returns an error (instead of a silent fallback) when rawDSL is unset:
+// NewPipelineFromDSL always populates rawDSL, so an empty value means the
+// Pipeline was built through an unexpected path. A fallback that re-encodes
+// p.canvas would produce a different hash (Canvas.NodeParents is json:"-",
+// and struct field order differs from the map form), silently breaking the
+// change-detection contract.
+func (p *Pipeline) dslFileFingerprint() (string, error) {
+	if len(p.rawDSL) == 0 {
+		return "", fmt.Errorf("rawDSL not set (Pipeline built without NewPipelineFromDSL?)")
+	}
+	h := sha256.New()
+	h.Write(p.rawDSL)
+	return hex.EncodeToString(h.Sum(nil)), nil
+}
+
+// overrideFingerprint returns a stable hash of the run-level override_params
+// (Doc.ParserConfig + injected LLM id). It is tracked separately from the DSL
+// file so a change limited to the runtime override is distinguishable (in the
+// warning log) from an edit to the DSL template file.
+func overrideFingerprint(override map[string]any) string {
+	h := sha256.New()
+	_ = json.NewEncoder(h).Encode(override)
+	return hex.EncodeToString(h.Sum(nil))
+}
+
+// classifyDSLChange decides why a stored fingerprint pair no longer matches
+// the current run, for the mismatch warning log. It returns "" when neither
+// changed (no stale checkpoint to discard). A DSL file change takes precedence
+// in the label because an edited template is the more common/expected case;
+// when both changed the label still says DSL. The caller only invokes this
+// when a checkpoint actually exists, so an empty stored fingerprint is treated
+// as "changed" (safe: discard and re-run from scratch).
+func classifyDSLChange(storedDsl, storedOvf, dslFP, ovfFP string) string {
+	dslChanged := storedDsl != dslFP
+	ovfChanged := storedOvf != ovfFP
+	if !dslChanged && !ovfChanged {
+		return ""
+	}
+	if dslChanged {
+		return "DSL/template changed"
+	}
+	return "runtime override/parser-config changed"
+}
+
+// guardDSLChange prevents resuming a checkpoint written by a DIFFERENT DSL
+// than the one currently compiled. eino checkpoints are bound to the graph's
+// node ids / wiring; resuming against a modified graph restores state for
+// nodes that no longer exist (or have different wiring) and makes eino error
+// out. When either the DSL file fingerprint or the runtime override
+// fingerprint differs we drop the stale checkpoint (and the persisted
+// interrupt marker) so the run starts fresh from the entry node. When no
+// checkpoint exists this is simply a fresh run, and we record both
+// fingerprints so a later resume can detect edits. All store writes are
+// best-effort: a write failure must not abort the run.
+func (p *Pipeline) guardDSLChange(ctx context.Context, store canvas.CheckPointStore, tracker *canvas.RunTracker, cpID string, override map[string]any) {
+	// Only meaningful when a checkpoint store is wired (the resumable path).
+	// The non-resumable runPlain path never calls here, but guard anyway so a
+	// future caller cannot nil-panic on store.Get.
+	if store == nil {
+		return
+	}
+	dslKey := cpID + dslKeySuffix
+	ovfKey := cpID + ovfKeySuffix
+
+	dslFP, err := p.dslFileFingerprint()
+	if err != nil {
+		// Without a DSL fingerprint we cannot detect edits. Skip the guard
+		// rather than risk an empty hash that silently mismatches every run.
+		// The run still proceeds (best-effort); resume safety is unchanged
+		// only because rawDSL is always set via NewPipelineFromDSL.
+		common.Error(fmt.Sprintf("pipeline: skip DSL-change guard for %s: %v", p.taskID, err), err)
+		return
+	}
+	ovfFP := overrideFingerprint(override)
+
+	// A checkpoint from a previous run exists → this is a resume candidate.
+	// A read error means we cannot trust the existence check, so bail out
+	// entirely: do NOT overwrite cpID:dsl / cpID:ovf. Writing current
+	// fingerprints after a failed lookup could mask a real stale-checkpoint
+	// mismatch on a later resume (the old checkpoint would survive, but its
+	// fingerprint would be overwritten to match the new DSL).
+	_, found, cpErr := store.Get(ctx, cpID)
+	if cpErr != nil {
+		common.Error(fmt.Sprintf("pipeline: lookup checkpoint for %s failed, skip DSL-change guard: %v", p.taskID, cpErr), cpErr)
+		return
+	}
+	if found {
+		storedDsl, dOk, dErr := store.Get(ctx, dslKey)
+		storedOvf, oOk, oErr := store.Get(ctx, ovfKey)
+		if dErr != nil || oErr != nil {
+			common.Error(fmt.Sprintf("pipeline: read DSL fingerprints for %s failed, skip DSL-change guard: %v", p.taskID, coalesceErr(dErr, oErr)), coalesceErr(dErr, oErr))
+			return
+		}
+		reason := classifyDSLChange(orEmpty(dOk, storedDsl), orEmpty(oOk, storedOvf), dslFP, ovfFP)
+		if reason != "" {
+			common.Warn(fmt.Sprintf("pipeline: DSL for task %s changed since checkpoint was written (%s); discarding stale checkpoint and re-running from scratch", p.taskID, reason))
+			if err := store.Delete(ctx, cpID); err != nil {
+				common.Error(fmt.Sprintf("pipeline: delete stale checkpoint for %s failed: %v", p.taskID, err), err)
+			}
+			if tracker != nil {
+				if err := tracker.ClearInterruptID(ctx, cpID); err != nil {
+					common.Error(fmt.Sprintf("pipeline: clear interrupt id for %s failed: %v", p.taskID, err), err)
+				}
+			}
+		}
+	}
+	// Record the fingerprints of the DSL + override that produced this run.
+	if err := store.Set(ctx, dslKey, []byte(dslFP)); err != nil {
+		common.Error(fmt.Sprintf("pipeline: persist DSL fingerprint for %s failed: %v", p.taskID, err), err)
+	}
+	if err := store.Set(ctx, ovfKey, []byte(ovfFP)); err != nil {
+		common.Error(fmt.Sprintf("pipeline: persist override fingerprint for %s failed: %v", p.taskID, err), err)
+	}
+}
+
+// orEmpty returns the byte slice when ok, else an empty string, so a missing
+// stored fingerprint is treated as "" (classifyDSLChange then sees a change).
+func orEmpty(ok bool, b []byte) string {
+	if !ok {
+		return ""
+	}
+	return string(b)
+}
+
+// coalesceErr returns the first non-nil error, so a single message can report
+// whichever of two reads failed without dereferencing nil.
+func coalesceErr(errs ...error) error {
+	for _, e := range errs {
+		if e != nil {
+			return e
+		}
+	}
+	return nil
+}
+
 // Run executes the full ingestion graph described by the canonical DSL.
 // There is no pipeline-layer partial resume entry point: execution always
 // starts from the graph entry and component-level replay decisions belong to
@@ -300,6 +475,10 @@ func (p *Pipeline) Run(ctx context.Context, inputs map[string]any, override_para
 		return p.runPlain(runCtx, current, compiled, tracker, runState)
 	}

+	// Resumable path: detect DSL / override edits since the checkpoint was
+	// written and discard a stale checkpoint before resuming (see guardDSLChange).
+	p.guardDSLChange(ctx, store, tracker, p.taskID, override_params)
+
 	// Resumable path: record the run, then loop Invoke until the graph
 	// completes or a non-resumable error surfaces.
 	if tracker != nil {
@@ -393,6 +572,8 @@ func (p *Pipeline) runResumable(ctx context.Context, runCtx context.Context, cur
 			}
 			if store != nil {
 				utility.BestEffort(fmt.Sprintf("delete checkpoint for %s", p.taskID), func() error { return store.Delete(ctx, cpID) })
+				utility.BestEffort(fmt.Sprintf("delete DSL fingerprint for %s", p.taskID), func() error { return store.Delete(ctx, cpID+dslKeySuffix) })
+				utility.BestEffort(fmt.Sprintf("delete override fingerprint for %s", p.taskID), func() error { return store.Delete(ctx, cpID+ovfKeySuffix) })
 			}
 			return finalizeResult(current, out, runState), nil
 		}
@@ -434,6 +615,16 @@ func (p *Pipeline) cleanupCheckpoint(ctx context.Context, store canvas.CheckPoin
 		if err := store.Delete(ctx, cpID); err != nil {
 			common.Error(fmt.Sprintf("pipeline: delete checkpoint %s failed: %v", cpID, err), err)
 		}
+		// Drop the DSL / override fingerprints alongside the checkpoint so
+		// they share one lifecycle on cancellation (otherwise the fingerprint
+		// keys linger up to TTL while the checkpoint is gone — harmless, but
+		// inconsistent). A later re-run overwrites them anyway.
+		if err := store.Delete(ctx, cpID+dslKeySuffix); err != nil {
+			common.Error(fmt.Sprintf("pipeline: delete DSL fingerprint %s failed: %v", cpID, err), err)
+		}
+		if err := store.Delete(ctx, cpID+ovfKeySuffix); err != nil {
+			common.Error(fmt.Sprintf("pipeline: delete override fingerprint %s failed: %v", cpID, err), err)
+		}
 	}
 	if tracker != nil {
 		_ = tracker.ClearInterruptID(ctx, cpID)
--- a/internal/ingestion/pipeline/pipeline_test.go
+++ b/internal/ingestion/pipeline/pipeline_test.go
@@ -453,6 +453,239 @@ func TestPipelineRunResumableCrossRunResume(t *testing.T) {
 	}
 }

+// TestPipelineRunResumableDSLChanged discards a stale checkpoint when the DSL
+// is edited between the failed run and the resume. Run 1 fails on the terminal
+// stage, persisting a checkpoint keyed by taskID. Run 2 re-runs the SAME
+// taskID but with a modified DSL (a param edit, enough to change the
+// fingerprint). The guard must detect the mismatch, discard the checkpoint,
+// and start fresh — so every stage re-executes (A runs on both runs) instead
+// of erroring while resuming an incompatible graph (the original bug).
+func TestPipelineRunResumableDSLChanged(t *testing.T) {
+	mockA := &mockCanvasStage{output: map[string]any{"a": 1}}
+	termStage := &oneShotErrStage{mockCanvasStage: mockCanvasStage{output: map[string]any{"b": 2}}}
+
+	const (
+		nameA = "p.DSLChangedA"
+		nameB = "p.DSLChangedB"
+	)
+	runtime.MustRegister(nameA, runtime.CategoryIngestion,
+		func(_ string, _ map[string]any) (runtime.Component, error) { return mockA, nil },
+		runtime.Metadata{Version: "1.0.0"})
+	runtime.MustRegister(nameB, runtime.CategoryIngestion,
+		func(_ string, _ map[string]any) (runtime.Component, error) { return termStage, nil },
+		runtime.Metadata{Version: "1.0.0"})
+
+	const taskID = "task-dsl-changed"
+
+	// Shared store + tracker across both runs (same taskID).
+	store := newMemCheckpointStore()
+	mr := miniredis.RunT(t)
+	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	t.Cleanup(func() { client.Close() })
+	tracker := canvas.NewRunTrackerWithClient(client, time.Hour)
+
+	// Run 1: original DSL; terminal B errors (oneShotErrStage n=1), leaving a
+	// checkpoint + interrupt for the original DSL fingerprint.
+	pipe1, err := NewPipelineFromDSL([]byte(`{
+		"dsl": {
+			"components": {
+				"begin": {"obj": {"component_name": "Begin", "params": {}}, "downstream": ["a"]},
+				"a": {"obj": {"component_name": "`+nameA+`", "params": {}}, "upstream": ["begin"], "downstream": ["b"]},
+				"b": {"obj": {"component_name": "`+nameB+`", "params": {}}, "upstream": ["a"]}
+			},
+			"path": ["begin", "a", "b"],
+			"graph": {"nodes": []}
+		}
+	}`), taskID, WithCheckPointStore(store), WithRunTracker(tracker))
+	if err != nil {
+		t.Fatalf("NewPipelineFromDSL (run 1): %v", err)
+	}
+	if _, err := pipe1.Run(context.Background(), map[string]any{"name": "doc-dsl-changed"}, nil); err == nil {
+		t.Fatal("Run 1: expected error from simulated crash, got nil")
+	}
+	if mockA.calls != 1 {
+		t.Fatalf("Run 1: expected A to run once, got %d", mockA.calls)
+	}
+
+	// Run 2: SAME taskID, but a modified DSL (param added to "a" → different
+	// fingerprint). Without the guard this would resume the stale checkpoint
+	// against an incompatible graph and error; with the guard it discards and
+	// re-runs from scratch.
+	pipe2, err := NewPipelineFromDSL([]byte(`{
+		"dsl": {
+			"components": {
+				"begin": {"obj": {"component_name": "Begin", "params": {}}, "downstream": ["a"]},
+				"a": {"obj": {"component_name": "`+nameA+`", "params": {"chunk_size": 128}}, "upstream": ["begin"], "downstream": ["b"]},
+				"b": {"obj": {"component_name": "`+nameB+`", "params": {}}, "upstream": ["a"]}
+			},
+			"path": ["begin", "a", "b"],
+			"graph": {"nodes": []}
+		}
+	}`), taskID, WithCheckPointStore(store), WithRunTracker(tracker))
+	if err != nil {
+		t.Fatalf("NewPipelineFromDSL (run 2): %v", err)
+	}
+	if _, err := pipe2.Run(context.Background(), map[string]any{"name": "doc-dsl-changed"}, nil); err != nil {
+		t.Fatalf("Run 2: expected fresh run to succeed after DSL edit, got error: %v", err)
+	}
+
+	// Fresh run re-executed A (and B, which now delegates successfully at n=2).
+	if mockA.calls != 2 {
+		t.Fatalf("Run 2: expected A to run again (fresh run), got A.calls=%d", mockA.calls)
+	}
+	if termStage.calls != 1 {
+		t.Fatalf("Run 2: expected B to delegate once, got %d", termStage.calls)
+	}
+	if store.deleteCount() < 1 {
+		t.Fatalf("expected the stale checkpoint to be deleted on DSL change, got deleteCount=%d", store.deleteCount())
+	}
+}
+
+// TestClassifyDSLChange is the table-driven unit test for the mismatch-reason
+// classifier used by the warning log. It isolates the "DSL file changed vs
+// runtime override changed" decision from the Run/resume plumbing so the
+// diagnostic label can be verified directly (the log text itself is not
+// asserted by capture).
+func TestClassifyDSLChange(t *testing.T) {
+	const (
+		dsl = "dslfp"
+		ovf = "ovrfp"
+	)
+	cases := []struct {
+		name       string
+		storedDsl  string
+		storedOvf  string
+		dslFP      string
+		ovfFP      string
+		wantReason string
+	}{
+		{"no change", dsl, ovf, dsl, ovf, ""},
+		{"dsl file changed", dsl + "x", ovf, dsl, ovf, "DSL/template changed"},
+		{"override changed", dsl, ovf + "x", dsl, ovf, "runtime override/parser-config changed"},
+		{"both changed", dsl + "x", ovf + "x", dsl, ovf, "DSL/template changed"},
+		// Caller only invokes classify when a checkpoint exists; a missing
+		// fingerprint is treated as "changed" (safe: discard + fresh run).
+		{"missing stored fingerprint", "", "", dsl, ovf, "DSL/template changed"},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			got := classifyDSLChange(c.storedDsl, c.storedOvf, c.dslFP, c.ovfFP)
+			if got != c.wantReason {
+				t.Fatalf("classifyDSLChange = %q, want %q", got, c.wantReason)
+			}
+		})
+	}
+}
+
+// errCheckpointStore fails every Get with getErr and records whether Set was
+// ever called. It lets us assert that guardDSLChange does NOT overwrite the
+// DSL/override fingerprints after a failed checkpoint lookup (which would mask
+// a real stale-checkpoint mismatch on a later resume).
+type errCheckpointStore struct {
+	getErr    error
+	setCalled bool
+}
+
+func (s *errCheckpointStore) Get(_ context.Context, _ string) ([]byte, bool, error) {
+	return nil, false, s.getErr
+}
+func (s *errCheckpointStore) Set(_ context.Context, _ string, _ []byte) error {
+	s.setCalled = true
+	return nil
+}
+func (s *errCheckpointStore) Delete(_ context.Context, _ string) error { return nil }
+
+// TestGuardDSLChange_CheckpointLookupErrorSkipsOverwrite verifies the
+// CodeRabbit finding: when store.Get(cpID) returns an error, guardDSLChange
+// bails out entirely and must NOT call Set on the fingerprint keys.
+func TestGuardDSLChange_CheckpointLookupErrorSkipsOverwrite(t *testing.T) {
+	const taskID = "task-lookup-err"
+	store := &errCheckpointStore{getErr: errors.New("redis temporarily unavailable")}
+	p := &Pipeline{
+		taskID: taskID,
+		rawDSL: []byte(`{"dsl":{"components":{"begin":{"obj":{"component_name":"Begin","params":{}}}}}}`),
+	}
+	// tracker is nil-safe (only used on the delete branch, which is skipped).
+	p.guardDSLChange(context.Background(), store, nil, taskID, map[string]any{"k": "v"})
+	if store.setCalled {
+		t.Fatal("guardDSLChange must not overwrite fingerprints after a failed checkpoint lookup")
+	}
+}
+
+// TestPipelineRunResumableOverrideChanged verifies that editing ONLY the
+// runtime override_params (same DSL file) between the failed run and the
+// resume is detected as a change and forces a fresh run (the checkpoint is
+// discarded) — it must NOT resume against stale override state. This is the
+// "runtime override/parser-config changed" branch of guardDSLChange.
+func TestPipelineRunResumableOverrideChanged(t *testing.T) {
+	mockA := &mockCanvasStage{output: map[string]any{"a": 1}}
+	termStage := &oneShotErrStage{mockCanvasStage: mockCanvasStage{output: map[string]any{"b": 2}}}
+
+	const (
+		nameA = "p.OvfChangedA"
+		nameB = "p.OvfChangedB"
+	)
+	runtime.MustRegister(nameA, runtime.CategoryIngestion,
+		func(_ string, _ map[string]any) (runtime.Component, error) { return mockA, nil },
+		runtime.Metadata{Version: "1.0.0"})
+	runtime.MustRegister(nameB, runtime.CategoryIngestion,
+		func(_ string, _ map[string]any) (runtime.Component, error) { return termStage, nil },
+		runtime.Metadata{Version: "1.0.0"})
+
+	const taskID = "task-override-changed"
+
+	store := newMemCheckpointStore()
+	mr := miniredis.RunT(t)
+	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	t.Cleanup(func() { client.Close() })
+	tracker := canvas.NewRunTrackerWithClient(client, time.Hour)
+
+	dsl := `{
+		"dsl": {
+			"components": {
+				"begin": {"obj": {"component_name": "Begin", "params": {}}, "downstream": ["a"]},
+				"a": {"obj": {"component_name": "` + nameA + `", "params": {}}, "upstream": ["begin"], "downstream": ["b"]},
+				"b": {"obj": {"component_name": "` + nameB + `", "params": {}}, "upstream": ["a"]}
+			},
+			"path": ["begin", "a", "b"],
+			"graph": {"nodes": []}
+		}
+	}`
+
+	// Run 1: override v1; terminal B errors (oneShotErrStage n=1), leaving a
+	// checkpoint + interrupt. The DSL file fingerprint is recorded too.
+	pipe1, err := NewPipelineFromDSL([]byte(dsl), taskID, WithCheckPointStore(store), WithRunTracker(tracker))
+	if err != nil {
+		t.Fatalf("NewPipelineFromDSL (run 1): %v", err)
+	}
+	if _, err := pipe1.Run(context.Background(), map[string]any{"name": "doc-ovf-changed"}, map[string]any{"k": "v1"}); err == nil {
+		t.Fatal("Run 1: expected error from simulated crash, got nil")
+	}
+	if mockA.calls != 1 {
+		t.Fatalf("Run 1: expected A to run once, got %d", mockA.calls)
+	}
+
+	// Run 2: SAME DSL file, but override changed to v2. The guard must detect
+	// the override change, discard the stale checkpoint, and re-run from
+	// scratch (A runs again).
+	pipe2, err := NewPipelineFromDSL([]byte(dsl), taskID, WithCheckPointStore(store), WithRunTracker(tracker))
+	if err != nil {
+		t.Fatalf("NewPipelineFromDSL (run 2): %v", err)
+	}
+	if _, err := pipe2.Run(context.Background(), map[string]any{"name": "doc-ovf-changed"}, map[string]any{"k": "v2"}); err != nil {
+		t.Fatalf("Run 2: expected fresh run to succeed after override edit, got error: %v", err)
+	}
+	if mockA.calls != 2 {
+		t.Fatalf("Run 2: expected A to run again (fresh run on override change), got A.calls=%d", mockA.calls)
+	}
+	if termStage.calls != 1 {
+		t.Fatalf("Run 2: expected B to delegate once, got %d", termStage.calls)
+	}
+	if store.deleteCount() < 1 {
+		t.Fatalf("expected the stale checkpoint to be deleted on override change, got deleteCount=%d", store.deleteCount())
+	}
+}
+
 // TestPipelineRun_RequireResumeRejectsWithoutStore verifies that with
 // WithRequireResume set and no checkpoint store resolvable (no
 // injected store, no global Redis in unit scope), Run must refuse to start
@@ -576,6 +809,13 @@ func TestCleanupCheckpoint_DeletesStoreAndClearsTracker(t *testing.T) {
 	if err := store.Set(context.Background(), "cp-1", []byte("data")); err != nil {
 		t.Fatalf("store.Set: %v", err)
 	}
+	// Fingerprint keys must share the checkpoint's lifecycle on cleanup.
+	if err := store.Set(context.Background(), "cp-1"+dslKeySuffix, []byte("dslfp")); err != nil {
+		t.Fatalf("store.Set dsl: %v", err)
+	}
+	if err := store.Set(context.Background(), "cp-1"+ovfKeySuffix, []byte("ovrfp")); err != nil {
+		t.Fatalf("store.Set ovf: %v", err)
+	}
 	tracker := canvas.NewRunTrackerWithClient(client, time.Hour)
 	if err := tracker.AttachInterrupt(context.Background(), "cp-1", "interrupt-1"); err != nil {
 		t.Fatalf("AttachInterrupt: %v", err)
@@ -584,8 +824,14 @@ func TestCleanupCheckpoint_DeletesStoreAndClearsTracker(t *testing.T) {
 	p := &Pipeline{}
 	p.cleanupCheckpoint(context.Background(), store, tracker, "cp-1")

-	if store.deleteCount() != 1 {
-		t.Fatalf("store.Delete was not called")
+	// checkpoint + dsl fingerprint + ovf fingerprint = 3 deletes.
+	if store.deleteCount() != 3 {
+		t.Fatalf("expected 3 store deletes (checkpoint + 2 fingerprints), got %d", store.deleteCount())
+	}
+	for _, k := range []string{"cp-1", "cp-1" + dslKeySuffix, "cp-1" + ovfKeySuffix} {
+		if _, found, _ := store.Get(context.Background(), k); found {
+			t.Fatalf("expected key %q to be deleted", k)
+		}
 	}
 	id, ok, err := tracker.GetInterruptID(context.Background(), "cp-1")
 	if err != nil {