Integrate chunk plan caching into the runner

2026-07-18 00:20:56 +00:00
parent ebd449d847
commit 51a36efb6b
15 changed files with 590 additions and 441 deletions
--- a/internal/framework/pipeline/checkpoint.go
+++ b/internal/framework/pipeline/checkpoint.go
@@ -20,10 +20,6 @@ type CheckpointRecorder interface {
 	SourceRunning(moduleKey string) error
 	SourceSucceeded(moduleKey string, doc *source.SourceDocument) error
 	SourceFailed(moduleKey string, err error) error
-	ChunkRunning(moduleKey string, sourceDigest string) error
-	ChunkSucceeded(moduleKey string, sourceDigest string, chunks []source.Chunk, warnings []contracts.Warning) error
-	ChunkRejected(moduleKey string, sourceDigest string, rejected contracts.RejectedOutput) error
-	ChunkFailed(moduleKey string, sourceDigest string, err error) error
 	ExtractRunning(laneID string, moduleKey string, dependencies []CheckpointFingerprint) error
 	ExtractSucceeded(laneID string, moduleKey string, dependencies []CheckpointFingerprint, outputs []CheckpointArtifact, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error
 	ExtractFailed(laneID string, moduleKey string, dependencies []CheckpointFingerprint, err error) error
@@ -54,11 +50,6 @@ type SourceCheckpoint struct {
 	Document *source.SourceDocument
 }

-type ChunkCheckpoint struct {
-	Chunks   []source.Chunk
-	Warnings []contracts.Warning
-}
-
 // CheckpointArtifact is the durable, domain-neutral value stored at a lane
 // checkpoint boundary.
 type CheckpointArtifact struct {
@@ -90,7 +81,6 @@ type NormalizeCheckpoint struct {
 type CheckpointLoader interface {
 	Enabled() bool
 	Source(moduleKey string) (SourceCheckpoint, CheckpointDecision)
-	Chunk(moduleKey string, sourceDigest string) (ChunkCheckpoint, CheckpointDecision)
 	Extract(laneID string, moduleKey string, dependencies []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision)
 	Merge(laneID string, moduleKey string, dependencies []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision)
 	Normalize(laneID string, moduleKey string, dependencies []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision)
@@ -105,14 +95,6 @@ func NoopCheckpointLoader() CheckpointLoader     { return noopCheckpointLoader{}
 func (noopCheckpointRecorder) SourceRunning(string) error                           { return nil }
 func (noopCheckpointRecorder) SourceSucceeded(string, *source.SourceDocument) error { return nil }
 func (noopCheckpointRecorder) SourceFailed(string, error) error                     { return nil }
-func (noopCheckpointRecorder) ChunkRunning(string, string) error                    { return nil }
-func (noopCheckpointRecorder) ChunkSucceeded(string, string, []source.Chunk, []contracts.Warning) error {
-	return nil
-}
-func (noopCheckpointRecorder) ChunkRejected(string, string, contracts.RejectedOutput) error {
-	return nil
-}
-func (noopCheckpointRecorder) ChunkFailed(string, string, error) error { return nil }
 func (noopCheckpointRecorder) ExtractRunning(string, string, []CheckpointFingerprint) error {
 	return nil
 }
@@ -149,9 +131,6 @@ func (noopCheckpointLoader) Enabled() bool { return false }
 func (noopCheckpointLoader) Source(string) (SourceCheckpoint, CheckpointDecision) {
 	return SourceCheckpoint{}, CheckpointDecision{Reason: "checkpoint loading disabled"}
 }
-func (noopCheckpointLoader) Chunk(string, string) (ChunkCheckpoint, CheckpointDecision) {
-	return ChunkCheckpoint{}, CheckpointDecision{Reason: "checkpoint loading disabled"}
-}
 func (noopCheckpointLoader) Extract(string, string, []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) {
 	return ExtractCheckpoint{}, CheckpointDecision{Reason: "checkpoint loading disabled"}
 }
--- a/internal/framework/pipeline/chunk_plan_store.go
+++ b/internal/framework/pipeline/chunk_plan_store.go
@@ -8,6 +8,8 @@ import (
 	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
 )

+const ChunkPlanSchemaVersion = "notarius.chunk-plan.v1"
+
 type ChunkPlanProducer struct {
 	InputModule string                          `json:"input_module"`
 	ChunkModule string                          `json:"chunk_module"`
--- a/internal/framework/pipeline/runner.go
+++ b/internal/framework/pipeline/runner.go
@@ -38,19 +38,21 @@ func New() *Runner {
 }

 type RunInput struct {
-	Prepared    *PreparedPipeline
-	SourceID    string
-	Path        string
-	RawInput    []byte
-	SessionID   string
-	RunID       string
-	StartedAt   time.Time
-	LLMProfiles []artifacts.LLMProfileManifest
-	Metadata    map[string]any
-	Warnings    []contracts.Warning
-	Checkpoints CheckpointRecorder
-	Checkpoint  CheckpointLoader
-	Debug       DebugRecorder
+	Prepared       *PreparedPipeline
+	SourceID       string
+	Path           string
+	RawInput       []byte
+	SessionID      string
+	RunID          string
+	StartedAt      time.Time
+	LLMProfiles    []artifacts.LLMProfileManifest
+	Metadata       map[string]any
+	Warnings       []contracts.Warning
+	ChunkCacheMode ChunkCacheMode
+	ChunkPlans     ChunkPlanStore
+	Checkpoints    CheckpointRecorder
+	Checkpoint     CheckpointLoader
+	Debug          DebugRecorder
 	// ExtractWorkers bounds run-wide extract jobs. Values less than one use a
 	// single worker so direct framework callers retain deterministic behavior.
 	ExtractWorkers int
@@ -174,19 +176,14 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err

 	chunker := input.Prepared.chunker
 	attachModuleManifestMetadata(&output, "chunker", chunker)
-	var canonicalChunks []source.Chunk
-	var generatedPlan *source.ChunkPlan
-	var chunkWarnings []contracts.Warning
-	chunkCheckpoint, chunkDecision := checkpointLoader.Chunk(chunker.Key(), doc.Digest)
-	recordCheckpointEvent(&output, checkpointLoader, string(StageChunk), "", chunker.Key(), chunkDecision)
 	chunkStarted := time.Now().UTC()
+	chunkMode := effectiveChunkCacheMode(input.ChunkCacheMode)
 	if err := writeDebugTimed(debugRecorder, "chunk/input.json", debugTimedEnvelope{
 		Stage:     string(StageChunk),
 		ModuleKey: chunker.Key(),
 		StartedAt: chunkStarted,
 		Payload: map[string]any{
-			"reused":       chunkDecision.Reused,
-			"decision":     chunkDecision,
+			"cache_mode":   chunkMode,
 			"source":       debugSourceDocumentEnvelope(doc),
 			"source_input": debugContentEnvelope(sourceInput.Content, sourceInput.MediaType, nil, nil),
 			"options":      redactSensitiveMap(input.pipeline.Chunk.Options),
@@ -195,85 +192,28 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
 	}); err != nil {
 		return failOutput(output), fmt.Errorf("write chunk debug artifact: %w", err)
 	}
-	chunksAccepted := chunkDecision.Reused
-	var chunkRejection *contracts.RejectedOutput
-	if chunkDecision.Reused {
-		canonicalChunks = cloneSourceChunks(chunkCheckpoint.Chunks)
-		chunkWarnings = cloneWarnings(chunkCheckpoint.Warnings)
-		output.Warnings = append(output.Warnings, chunkWarnings...)
-	} else {
-		if err := checkpoints.ChunkRunning(chunker.Key(), doc.Digest); err != nil {
-			return failOutput(output), fmt.Errorf("write chunk checkpoint: %w", err)
-		}
-		chunksAccepted, chunkRejection, err = runWithRetry(ctx, input.pipeline.Chunk.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) {
-			attemptStarted := time.Now().UTC()
-			attemptPath := path.Join("chunk", fmt.Sprintf("attempt-%02d", attempt))
-			attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
-			terminal := newAttemptTerminalRecorder(debugRecorder, attemptPath, "chunk", llmScope, debugTimedEnvelope{Stage: string(StageChunk), ModuleKey: chunker.Key(), Attempt: attempt, StartedAt: attemptStarted})
-			chunkResult, err := chunker.Plan(attemptCtx, contracts.ChunkRequest{
-				Source:      doc,
-				SourceInput: sourceInput.Clone(),
-				SessionID:   sessionID,
-				References:  CloneReferenceSet(input.pipeline.ChunkReferences.ReferenceSet),
-				LLMProfile:  input.pipeline.Chunk.LLMProfile,
-				Metadata:    input.Metadata,
-			})
-			if err != nil {
-				attemptErr := fmt.Errorf("chunk source with chunker %q: %w", chunker.Key(), err)
-				return false, nil, terminal.record(nil, attemptErr)
-			}
-			plan, chunks, err := validateAndMaterializeChunkPlan(doc, chunkResult.Plan)
-			if err != nil {
-				attemptErr := fmt.Errorf("validate chunk plan from chunker %q: %w", chunker.Key(), err)
-				payload := map[string]any{"plan": debugChunkPlanEnvelope(chunkResult.Plan), "warnings": debugWarningEnvelopes(chunkResult.Warnings)}
-				return false, nil, terminal.record(payload, attemptErr)
-			}
-			validationWarnings, rejection, err := r.validateChunks(attemptCtx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.pipeline.ChunkReferences.ReferenceSet, input.Metadata, input.Prepared.chunkValidators, attempt, input.Debug)
-			attemptWarnings := append(cloneWarnings(chunkResult.Warnings), validationWarnings...)
-			payload := map[string]any{
-				"plan":                debugChunkPlanEnvelope(plan),
-				"materialized_chunks": debugSourceChunkEnvelopes(chunks),
-				"warnings":            debugWarningEnvelopes(attemptWarnings),
-				"rejection":           debugRejectedOutputPtr(rejection),
-			}
-			if err != nil || rejection != nil {
-				return false, rejection, terminal.record(payload, err)
-			}
-			canonicalChunks = chunks
-			generatedPlan = &plan
-			chunkWarnings = attemptWarnings
-			if err := terminal.record(payload, nil); err != nil {
-				return false, nil, err
-			}
-			return true, nil, nil
-		})
-		if err != nil {
-			_ = checkpoints.ChunkFailed(chunker.Key(), doc.Digest, err)
-			return failOutput(output), err
-		}
-		if !chunksAccepted {
-			output.Rejected = append(output.Rejected, *chunkRejection)
-			if err := checkpoints.ChunkRejected(chunker.Key(), doc.Digest, *chunkRejection); err != nil {
-				return failOutput(output), fmt.Errorf("write chunk checkpoint: %w", err)
-			}
-		} else {
-			output.Warnings = append(output.Warnings, chunkWarnings...)
-			if err := checkpoints.ChunkSucceeded(chunker.Key(), doc.Digest, canonicalChunks, chunkWarnings); err != nil {
-				return failOutput(output), fmt.Errorf("write chunk checkpoint: %w", err)
-			}
-		}
+	chunkResult, err := r.runChunkPlan(ctx, input, doc, sourceInput, sessionID)
+	if err != nil {
+		return failOutput(output), err
+	}
+	if chunkResult.rejection != nil {
+		output.Rejected = append(output.Rejected, *chunkResult.rejection)
+	}
+	if chunkResult.accepted || chunkResult.lookup.Status == ChunkPlanHit {
+		output.Warnings = append(output.Warnings, chunkResult.warnings...)
 	}
 	chunkDebugPayload := map[string]any{
-		"reused":              chunkDecision.Reused,
-		"accepted":            chunksAccepted,
-		"materialized_chunks": debugSourceChunkEnvelopes(canonicalChunks),
-		"warnings":            chunkWarnings,
+		"cache_mode":          chunkMode,
+		"lookup":              chunkResult.lookup,
+		"accepted":            chunkResult.accepted,
+		"materialized_chunks": debugSourceChunkEnvelopes(chunkResult.chunks),
+		"warnings":            chunkResult.warnings,
 	}
-	if generatedPlan != nil {
-		chunkDebugPayload["plan"] = debugChunkPlanEnvelope(*generatedPlan)
+	if chunkResult.plan != nil {
+		chunkDebugPayload["plan"] = debugChunkPlanEnvelope(*chunkResult.plan)
 	}
-	if chunkRejection != nil {
-		chunkDebugPayload["rejection"] = debugRejectedOutputEnvelope(*chunkRejection)
+	if chunkResult.rejection != nil {
+		chunkDebugPayload["rejection"] = debugRejectedOutputEnvelope(*chunkResult.rejection)
 	}
 	if err := writeDebugTimed(debugRecorder, "chunk/output.json", debugTimedEnvelope{
 		Stage:     string(StageChunk),
@@ -284,8 +224,8 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
 		return failOutput(output), fmt.Errorf("write chunk debug artifact: %w", err)
 	}

-	if chunksAccepted {
-		laneOutput, laneErr := r.runLanes(ctx, input, checkpoints, checkpointLoader, doc, sourceInput, sessionID, canonicalChunks)
+	if chunkResult.accepted {
+		laneOutput, laneErr := r.runLanes(ctx, input, checkpoints, checkpointLoader, doc, sourceInput, sessionID, chunkResult.chunks)
 		mergeLaneOutput(&output, laneOutput)
 		if laneErr != nil {
 			return failOutput(output), laneErr
@@ -473,6 +413,13 @@ func validateRunInput(input RunInput) error {
 	if input.Prepared == nil {
 		return fmt.Errorf("prepared pipeline must not be nil")
 	}
+	mode := effectiveChunkCacheMode(input.ChunkCacheMode)
+	if err := mode.Validate(); err != nil {
+		return err
+	}
+	if mode != ChunkCacheBypass && input.ChunkPlans == nil {
+		return fmt.Errorf("chunk plan store is required for %q mode", mode)
+	}
 	return validateResolvedPipeline(input.Prepared.resolved)
 }

@@ -546,10 +493,6 @@ func manifestFromPipeline(input RunInput) artifacts.RunManifest {
 		References:      ReferenceProvenance(pipeline),
 		LLMProfiles:     cloneLLMProfiles(input.LLMProfiles),
 	}
-	// The runner does not currently maintain a cache or idempotency key. Reference
-	// digests are recorded in manifest provenance and intentionally kept separate
-	// from source_digests.
-
 	for _, lane := range pipeline.ArtifactLanes {
 		laneManifest := artifacts.ArtifactLaneManifest{
 			ID:         lane.ID,
--- a/internal/framework/pipeline/runner_candidate_encoding_test.go
+++ b/internal/framework/pipeline/runner_candidate_encoding_test.go
@@ -209,7 +209,7 @@ func TestRunnerFinalEncodesAcceptedCandidatesOnce(t *testing.T) {
 			for i, event := range output.CheckpointEvents {
 				gotStages[i] = event.Stage
 			}
-			wantStages := []string{"source", string(StageChunk), string(StageExtract), string(StageMerge), string(StageNormalize)}
+			wantStages := []string{"source", string(StageExtract), string(StageMerge), string(StageNormalize)}
 			if !reflect.DeepEqual(gotStages, wantStages) {
 				t.Fatalf("checkpoint event stages = %#v, want %#v", gotStages, wantStages)
 			}
--- a/internal/framework/pipeline/runner_chunk_plan.go
+++ b/internal/framework/pipeline/runner_chunk_plan.go
@@ -0,0 +1,137 @@
+package pipeline
+
+import (
+	"context"
+	"fmt"
+	"path"
+	"time"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
+	"gitea.maximumdirect.net/eric/notarius/internal/core/source"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+)
+
+type chunkPlanExecution struct {
+	accepted  bool
+	chunks    []source.Chunk
+	plan      *source.ChunkPlan
+	warnings  []contracts.Warning
+	rejection *contracts.RejectedOutput
+	lookup    ChunkPlanDecision
+}
+
+func effectiveChunkCacheMode(mode ChunkCacheMode) ChunkCacheMode {
+	if mode == "" {
+		return ChunkCacheBypass
+	}
+	parsed, _ := ParseChunkCacheMode(string(mode))
+	return parsed
+}
+
+// Chunk-plan lookup and publication happen serially before lane workers start.
+// The runner therefore does not add synchronization around the store.
+func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string) (chunkPlanExecution, error) {
+	mode := effectiveChunkCacheMode(input.ChunkCacheMode)
+	chunker := input.Prepared.chunker
+	result := chunkPlanExecution{lookup: ChunkPlanDecision{Reason: "chunk plan lookup skipped"}}
+
+	if mode == ChunkCacheAuto {
+		record, decision, err := input.ChunkPlans.Load(doc.Digest)
+		result.lookup = decision
+		if err != nil {
+			return result, fmt.Errorf("load chunk plan: %w", err)
+		}
+		switch decision.Status {
+		case ChunkPlanHit:
+			plan, chunks, validationErr := validateAndMaterializeChunkPlan(doc, record.Plan)
+			if validationErr == nil {
+				validationWarnings, rejection, err := r.validateChunks(ctx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.pipeline.ChunkReferences.ReferenceSet, input.Metadata, input.Prepared.chunkValidators, 1, input.Debug)
+				result.plan = &plan
+				result.chunks = chunks
+				result.warnings = append(cloneWarnings(record.Warnings), validationWarnings...)
+				result.rejection = rejection
+				result.accepted = rejection == nil && err == nil
+				return result, err
+			}
+			result.lookup = ChunkPlanDecision{Status: ChunkPlanInvalid, Reason: "stored chunk plan cannot be materialized against the current source"}
+		case ChunkPlanMissing, ChunkPlanInvalid:
+			// Generate below.
+		default:
+			return result, fmt.Errorf("load chunk plan returned unsupported status %q", decision.Status)
+		}
+	}
+
+	var producerWarnings []contracts.Warning
+	accepted, rejection, err := runWithRetry(ctx, input.pipeline.Chunk.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) {
+		attemptStarted := time.Now().UTC()
+		attemptPath := path.Join("chunk", fmt.Sprintf("attempt-%02d", attempt))
+		attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath)
+		terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "chunk", llmScope, debugTimedEnvelope{Stage: string(StageChunk), ModuleKey: chunker.Key(), Attempt: attempt, StartedAt: attemptStarted})
+		chunkResult, callErr := chunker.Plan(attemptCtx, contracts.ChunkRequest{
+			Source: doc, SourceInput: sourceInput.Clone(), SessionID: sessionID,
+			References: CloneReferenceSet(input.pipeline.ChunkReferences.ReferenceSet),
+			LLMProfile: input.pipeline.Chunk.LLMProfile, Metadata: input.Metadata,
+		})
+		if callErr != nil {
+			return false, nil, terminal.record(nil, fmt.Errorf("chunk source with chunker %q: %w", chunker.Key(), callErr))
+		}
+		plan, chunks, validationErr := validateAndMaterializeChunkPlan(doc, chunkResult.Plan)
+		if validationErr != nil {
+			attemptErr := fmt.Errorf("validate chunk plan from chunker %q: %w", chunker.Key(), validationErr)
+			payload := map[string]any{"plan": debugChunkPlanEnvelope(chunkResult.Plan), "warnings": debugWarningEnvelopes(chunkResult.Warnings)}
+			return false, nil, terminal.record(payload, attemptErr)
+		}
+		validationWarnings, rejected, validationErr := r.validateChunks(attemptCtx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.pipeline.ChunkReferences.ReferenceSet, input.Metadata, input.Prepared.chunkValidators, attempt, input.Debug)
+		attemptWarnings := append(cloneWarnings(chunkResult.Warnings), validationWarnings...)
+		payload := map[string]any{
+			"plan": debugChunkPlanEnvelope(plan), "materialized_chunks": debugSourceChunkEnvelopes(chunks),
+			"warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected),
+		}
+		if validationErr != nil || rejected != nil {
+			return false, rejected, terminal.record(payload, validationErr)
+		}
+		result.chunks = chunks
+		result.plan = &plan
+		result.warnings = attemptWarnings
+		producerWarnings = cloneWarnings(chunkResult.Warnings)
+		if debugErr := terminal.record(payload, nil); debugErr != nil {
+			return false, nil, debugErr
+		}
+		return true, nil, nil
+	})
+	if err != nil {
+		return result, err
+	}
+	result.accepted = accepted
+	result.rejection = rejection
+	if !accepted {
+		return result, nil
+	}
+
+	if mode == ChunkCacheAuto || mode == ChunkCacheRefresh {
+		planDigest, err := source.DigestChunkPlan(*result.plan)
+		if err != nil {
+			return result, fmt.Errorf("digest generated chunk plan: %w", err)
+		}
+		producerMetadata, _ := moduleManifestMetadata(chunker)
+		record := ChunkPlanRecord{
+			SchemaVersion: ChunkPlanSchemaVersion,
+			SourceDigest:  doc.Digest,
+			PlanDigest:    planDigest,
+			Plan:          source.CloneChunkPlan(*result.plan),
+			Producer: ChunkPlanProducer{
+				InputModule: input.Prepared.input.Key(),
+				ChunkModule: chunker.Key(),
+				LLMProfile:  input.pipeline.Chunk.LLMProfile,
+				References:  append([]artifacts.ReferenceProvenance(nil), referenceTargetProvenance(input.pipeline.ChunkReferences)...),
+				Metadata:    cloneMetadata(producerMetadata),
+			},
+			Warnings:  cloneWarnings(producerWarnings),
+			CreatedAt: time.Now().UTC(),
+		}
+		if err := input.ChunkPlans.Save(record); err != nil {
+			return result, fmt.Errorf("save chunk plan: %w", err)
+		}
+	}
+	return result, nil
+}
--- a/internal/framework/pipeline/runner_chunk_plan_test.go
+++ b/internal/framework/pipeline/runner_chunk_plan_test.go
@@ -0,0 +1,368 @@
+package pipeline
+
+import (
+	"context"
+	"errors"
+	"reflect"
+	"strings"
+	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/core/source"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+)
+
+type recordingChunkPlanStore struct {
+	record   ChunkPlanRecord
+	decision ChunkPlanDecision
+	loadErr  error
+	saveErr  error
+	loads    int
+	saves    int
+	saved    ChunkPlanRecord
+}
+
+func (s *recordingChunkPlanStore) Load(string) (ChunkPlanRecord, ChunkPlanDecision, error) {
+	s.loads++
+	return s.record, s.decision, s.loadErr
+}
+
+func (s *recordingChunkPlanStore) Save(record ChunkPlanRecord) error {
+	s.saves++
+	s.saved = record
+	return s.saveErr
+}
+
+type countingChunkValidator struct {
+	calls  int
+	result contracts.ValidationResult
+	err    error
+}
+
+func (*countingChunkValidator) Name() string { return "test/counting-chunks" }
+func (*countingChunkValidator) ExecutionClass() contracts.ExecutionClass {
+	return contracts.ExecutionClassDeterministic
+}
+func (v *countingChunkValidator) Validate(context.Context, contracts.ChunkValidationRequest) (contracts.ValidationResult, error) {
+	v.calls++
+	return v.result, v.err
+}
+
+type manifestChunker struct {
+	terminalChunker
+	metadata map[string]any
+}
+
+func (c manifestChunker) ManifestMetadata() map[string]any { return c.metadata }
+
+type llmCountingChunker struct {
+	terminalChunker
+	llmCalls *int
+}
+
+func (c llmCountingChunker) Plan(ctx context.Context, request contracts.ChunkRequest) (contracts.ChunkPlanResult, error) {
+	(*c.llmCalls)++
+	return c.terminalChunker.Plan(ctx, request)
+}
+
+type retryingChunker struct {
+	key   string
+	plan  source.ChunkPlan
+	calls int
+}
+
+func (c *retryingChunker) Key() string                             { return c.key }
+func (*retryingChunker) ReferenceSlots() []contracts.ReferenceSlot { return nil }
+func (c *retryingChunker) Plan(context.Context, contracts.ChunkRequest) (contracts.ChunkPlanResult, error) {
+	c.calls++
+	if c.calls == 1 {
+		return contracts.ChunkPlanResult{Warnings: []contracts.Warning{{Scope: "discarded", ReasonCode: "retry", Message: "discarded warning"}}}, errors.New("retry generation")
+	}
+	return contracts.ChunkPlanResult{Plan: source.CloneChunkPlan(c.plan), Warnings: []contracts.Warning{{Scope: "accepted", ReasonCode: "observed", Message: "accepted warning"}}}, nil
+}
+
+type dependencyLoader struct {
+	CheckpointLoader
+	extractDependencies []CheckpointFingerprint
+}
+
+func (l *dependencyLoader) Extract(_ string, _ string, dependencies []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) {
+	l.extractDependencies = append([]CheckpointFingerprint(nil), dependencies...)
+	return ExtractCheckpoint{}, CheckpointDecision{Reason: "not found"}
+}
+
+func TestRunnerChunkPlanModeMatrix(t *testing.T) {
+	tests := []struct {
+		name      string
+		mode      ChunkCacheMode
+		decision  ChunkPlanDecision
+		wantLoads int
+		wantSaves int
+		wantCalls int
+	}{
+		{name: "auto hit", mode: ChunkCacheAuto, decision: ChunkPlanDecision{Status: ChunkPlanHit}, wantLoads: 1},
+		{name: "auto missing", mode: ChunkCacheAuto, decision: ChunkPlanDecision{Status: ChunkPlanMissing}, wantLoads: 1, wantSaves: 1, wantCalls: 1},
+		{name: "auto invalid", mode: ChunkCacheAuto, decision: ChunkPlanDecision{Status: ChunkPlanInvalid}, wantLoads: 1, wantSaves: 1, wantCalls: 1},
+		{name: "bypass", mode: ChunkCacheBypass, wantCalls: 1},
+		{name: "empty bypass", wantCalls: 1},
+		{name: "refresh", mode: ChunkCacheRefresh, wantSaves: 1, wantCalls: 1},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			prepared, plan := preparedTerminalDebugPipeline(t)
+			calls := 0
+			llmCalls := 0
+			prepared.chunker = llmCountingChunker{terminalChunker: terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan, calls: &calls}, llmCalls: &llmCalls}
+			store := &recordingChunkPlanStore{record: chunkPlanRecord(t, prepared, plan), decision: tc.decision}
+			if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: tc.mode, ChunkPlans: store}); err != nil {
+				t.Fatalf("Run() error = %v", err)
+			}
+			if store.loads != tc.wantLoads || store.saves != tc.wantSaves || calls != tc.wantCalls {
+				t.Fatalf("calls = load %d save %d module %d, want %d %d %d", store.loads, store.saves, calls, tc.wantLoads, tc.wantSaves, tc.wantCalls)
+			}
+		})
+	}
+}
+
+func TestRunnerValidatesChunkPlanPolicyInputs(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		mode ChunkCacheMode
+		want string
+	}{
+		{name: "auto requires store", mode: ChunkCacheAuto, want: "store is required"},
+		{name: "refresh requires store", mode: ChunkCacheRefresh, want: "store is required"},
+		{name: "invalid mode", mode: "sometimes", want: "not supported"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			prepared, _ := preparedTerminalDebugPipeline(t)
+			_, err := New().Run(context.Background(), RunInput{Prepared: prepared, ChunkCacheMode: tc.mode})
+			if err == nil || !strings.Contains(err.Error(), tc.want) {
+				t.Fatalf("Run() error = %v, want %q", err, tc.want)
+			}
+		})
+	}
+}
+
+func TestRunnerChunkPlanStoreErrorsAreFrameworkErrors(t *testing.T) {
+	for _, tc := range []struct {
+		name      string
+		mode      ChunkCacheMode
+		decision  ChunkPlanDecision
+		loadErr   error
+		saveErr   error
+		wantError string
+		wantCalls int
+	}{
+		{name: "load", mode: ChunkCacheAuto, loadErr: errors.New("read failed"), wantError: "load chunk plan", wantCalls: 0},
+		{name: "save", mode: ChunkCacheRefresh, saveErr: errors.New("write failed"), wantError: "save chunk plan", wantCalls: 1},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			prepared, plan := preparedTerminalDebugPipeline(t)
+			calls := 0
+			llmCalls := 0
+			prepared.chunker = llmCountingChunker{terminalChunker: terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan, calls: &calls}, llmCalls: &llmCalls}
+			store := &recordingChunkPlanStore{decision: tc.decision, loadErr: tc.loadErr, saveErr: tc.saveErr}
+			_, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: tc.mode, ChunkPlans: store})
+			if err == nil || !strings.Contains(err.Error(), tc.wantError) || calls != tc.wantCalls {
+				t.Fatalf("Run() error = %v, module calls = %d, want %q and %d", err, calls, tc.wantError, tc.wantCalls)
+			}
+		})
+	}
+}
+
+func TestRunnerRegeneratesStructurallyInvalidHit(t *testing.T) {
+	prepared, plan := preparedTerminalDebugPipeline(t)
+	calls := 0
+	prepared.chunker = terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan, calls: &calls}
+	invalid := chunkPlanRecord(t, prepared, plan)
+	invalid.Plan.Ranges[0].StartUnitID = 999
+	store := &recordingChunkPlanStore{record: invalid, decision: ChunkPlanDecision{Status: ChunkPlanHit}}
+	if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: ChunkCacheAuto, ChunkPlans: store}); err != nil {
+		t.Fatal(err)
+	}
+	if calls != 1 || store.loads != 1 || store.saves != 1 {
+		t.Fatalf("calls = module %d load %d save %d, want 1 1 1", calls, store.loads, store.saves)
+	}
+}
+
+func TestRunnerRetriesBeforePublishingAcceptedPlan(t *testing.T) {
+	prepared, plan := preparedTerminalDebugPipeline(t)
+	prepared.resolved.Chunk.Retries = 1
+	chunker := &retryingChunker{key: prepared.resolved.Chunk.Module, plan: plan}
+	prepared.chunker = chunker
+	store := &recordingChunkPlanStore{decision: ChunkPlanDecision{Status: ChunkPlanMissing}}
+	output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: ChunkCacheAuto, ChunkPlans: store})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if chunker.calls != 2 || store.saves != 1 {
+		t.Fatalf("module calls = %d saves = %d, want 2 and 1", chunker.calls, store.saves)
+	}
+	if len(output.Warnings) != 1 || output.Warnings[0].Scope != "accepted" || len(store.saved.Warnings) != 1 || store.saved.Warnings[0].Scope != "accepted" {
+		t.Fatalf("output warnings = %#v stored warnings = %#v", output.Warnings, store.saved.Warnings)
+	}
+}
+
+func TestRunnerAutoHitValidatesOnceWithoutRegenerationOrMutation(t *testing.T) {
+	tests := []struct {
+		name         string
+		result       contracts.ValidationResult
+		validatorErr error
+		wantError    string
+		wantReject   bool
+	}{
+		{name: "warning", result: contracts.ValidationResult{Approved: true, Warnings: []contracts.Warning{{Scope: "current", ReasonCode: "observed", Message: "current warning"}}}},
+		{name: "rejection", result: contracts.ValidationResult{Approved: false, ReasonCode: "rejected", Message: "rejected hit"}, wantReject: true},
+		{name: "error", validatorErr: errors.New("validator failed"), wantError: "validator failed"},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			prepared, plan := preparedTerminalDebugPipeline(t)
+			calls := 0
+			llmCalls := 0
+			prepared.chunker = llmCountingChunker{terminalChunker: terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan, calls: &calls}, llmCalls: &llmCalls}
+			validator := &countingChunkValidator{result: tc.result, err: tc.validatorErr}
+			prepared.chunkValidators.validators = []preparedValidator{{resolved: ResolvedValidator{Binding: Binding(validator.Name()), Target: ValidatorTargetChunk}, chunk: validator}}
+			record := chunkPlanRecord(t, prepared, plan)
+			record.Warnings = []contracts.Warning{{Scope: "stored", ReasonCode: "observed", Message: "stored warning"}}
+			store := &recordingChunkPlanStore{record: record, decision: ChunkPlanDecision{Status: ChunkPlanHit}}
+			debug := newCapturedDebugRecorder()
+			output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: ChunkCacheAuto, ChunkPlans: store, Debug: debug})
+			if tc.wantError != "" {
+				if err == nil || !strings.Contains(err.Error(), tc.wantError) {
+					t.Fatalf("Run() error = %v, want %q", err, tc.wantError)
+				}
+			} else if err != nil {
+				t.Fatalf("Run() error = %v", err)
+			}
+			if validator.calls != 1 || calls != 0 || llmCalls != 0 || store.loads != 1 || store.saves != 0 {
+				t.Fatalf("calls = validator %d module %d llm %d load %d save %d", validator.calls, calls, llmCalls, store.loads, store.saves)
+			}
+			assertAttemptEnvelopeSequence(t, debug, "chunk")
+			if tc.wantError == "" {
+				encoded := string(debug.json["chunk/output.json"])
+				if !strings.Contains(encoded, `"status":"hit"`) || !strings.Contains(encoded, `"plan":`) || !strings.Contains(encoded, `"materialized_chunks":`) {
+					t.Fatalf("chunk hit debug = %s", encoded)
+				}
+			}
+			if tc.wantReject && (len(output.Rejected) != 1 || output.Rejected[0].ReasonCode != "rejected") {
+				t.Fatalf("rejected = %#v", output.Rejected)
+			}
+			if tc.wantError == "" && len(output.Warnings) != 1+len(tc.result.Warnings) {
+				t.Fatalf("warnings = %#v, want stored warning once plus current warnings", output.Warnings)
+			}
+		})
+	}
+}
+
+func TestRunnerPublishesOnlyAcceptedGeneratedPlans(t *testing.T) {
+	for _, tc := range []struct {
+		name       string
+		moduleErr  error
+		validator  contracts.ValidationResult
+		cancel     bool
+		wantCalls  int
+		wantReject bool
+	}{
+		{name: "module error", moduleErr: errors.New("generation failed"), wantCalls: 2},
+		{name: "validator rejection", validator: contracts.ValidationResult{Approved: false, ReasonCode: "rejected", Message: "rejected generated plan"}, wantCalls: 2, wantReject: true},
+		{name: "cancellation", cancel: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			prepared, plan := preparedTerminalDebugPipeline(t)
+			prepared.resolved.Chunk.Retries = 1
+			calls := 0
+			prepared.chunker = terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan, calls: &calls, err: tc.moduleErr}
+			if tc.wantReject {
+				validator := &countingChunkValidator{result: tc.validator}
+				prepared.chunkValidators.validators = []preparedValidator{{resolved: ResolvedValidator{Binding: Binding(validator.Name()), Target: ValidatorTargetChunk}, chunk: validator}}
+			}
+			ctx := context.Background()
+			if tc.cancel {
+				cancelled, cancel := context.WithCancel(ctx)
+				cancel()
+				ctx = cancelled
+			}
+			store := &recordingChunkPlanStore{decision: ChunkPlanDecision{Status: ChunkPlanInvalid}}
+			output, err := New().Run(ctx, RunInput{Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: ChunkCacheAuto, ChunkPlans: store})
+			if tc.wantReject {
+				if err != nil || len(output.Rejected) != 1 {
+					t.Fatalf("Run() error = %v rejected = %#v", err, output.Rejected)
+				}
+			} else if err == nil {
+				t.Fatal("Run() error = nil")
+			}
+			if calls != tc.wantCalls || store.saves != 0 {
+				t.Fatalf("module calls = %d, saves = %d; want %d, 0", calls, store.saves, tc.wantCalls)
+			}
+		})
+	}
+}
+
+func TestRunnerStoresProducerProvenanceAndProducerWarnings(t *testing.T) {
+	prepared, plan := preparedTerminalDebugPipeline(t)
+	prepared.resolved.Chunk.LLMProfile = "chunk-profile"
+	prepared.resolved.ChunkReferences = ResolvedReferenceTarget{
+		Stage: StageChunk,
+		ReferenceSet: contracts.ReferenceSet{Slots: map[string]contracts.ResolvedReferenceSlot{
+			"guide": {Items: []contracts.ReferenceItem{{SlotName: "guide", Digest: "sha256:guide", Origin: contracts.ReferenceOrigin{Type: "file", URI: "file:///guide.txt"}, Content: []byte("sensitive")}}},
+		}},
+	}
+	prepared.chunker = manifestChunker{terminalChunker: terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan, warnings: []contracts.Warning{{Scope: "producer", ReasonCode: "observed", Message: "producer warning"}}}, metadata: map[string]any{"prompt_id": "chunk/prompt"}}
+	validator := &countingChunkValidator{result: contracts.ValidationResult{Approved: true, Warnings: []contracts.Warning{{Scope: "validator", ReasonCode: "observed", Message: "validator warning"}}}}
+	prepared.chunkValidators.validators = []preparedValidator{{resolved: ResolvedValidator{Binding: Binding(validator.Name()), Target: ValidatorTargetChunk}, chunk: validator}}
+	store := &recordingChunkPlanStore{}
+	if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: ChunkCacheRefresh, ChunkPlans: store}); err != nil {
+		t.Fatal(err)
+	}
+	producer := store.saved.Producer
+	if producer.InputModule != prepared.input.Key() || producer.ChunkModule != prepared.chunker.Key() || producer.LLMProfile != "chunk-profile" || producer.Metadata["prompt_id"] != "chunk/prompt" {
+		t.Fatalf("producer = %#v", producer)
+	}
+	if len(producer.References) != 1 || producer.References[0].Digest != "sha256:guide" {
+		t.Fatalf("producer references = %#v", producer.References)
+	}
+	if len(store.saved.Warnings) != 1 || store.saved.Warnings[0].Scope != "producer" {
+		t.Fatalf("stored warnings = %#v, want only producer warning", store.saved.Warnings)
+	}
+}
+
+func TestRunnerRefreshChangesDownstreamChunkFingerprint(t *testing.T) {
+	doc := typedTestDocumentWithUnits(2)
+	prepared := preparedConcurrentPipeline(t, 1)
+	prepared.lanes = prepared.lanes[:1]
+	prepared.resolved.ArtifactLanes = prepared.resolved.ArtifactLanes[:1]
+	prepared.ArtifactLanes = prepared.ArtifactLanes[:1]
+	prepared.input = &typedTestInput{key: prepared.resolved.Input.Module, doc: doc}
+
+	run := func(plan source.ChunkPlan) []CheckpointFingerprint {
+		prepared.chunker = terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan}
+		loader := &dependencyLoader{CheckpointLoader: NoopCheckpointLoader()}
+		if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, ChunkCacheMode: ChunkCacheRefresh, ChunkPlans: &recordingChunkPlanStore{}, Checkpoint: loader}); err != nil {
+			t.Fatal(err)
+		}
+		return loader.extractDependencies
+	}
+	separate := typedTestPlan(doc)
+	combined := source.ChunkPlan{SourceDigest: doc.Digest, Ranges: []source.ChunkRange{{StartUnitID: 1, EndUnitID: 2}}}
+	if first, second := run(separate), run(combined); reflect.DeepEqual(first, second) {
+		t.Fatalf("downstream fingerprints did not change: %#v", first)
+	}
+}
+
+func chunkPlanRecord(t *testing.T, prepared *PreparedPipeline, plan source.ChunkPlan) ChunkPlanRecord {
+	t.Helper()
+	digest, err := source.DigestChunkPlan(plan)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return ChunkPlanRecord{
+		SchemaVersion: ChunkPlanSchemaVersion,
+		SourceDigest:  plan.SourceDigest,
+		PlanDigest:    digest,
+		Plan:          source.CloneChunkPlan(plan),
+		Producer:      ChunkPlanProducer{InputModule: prepared.input.Key(), ChunkModule: prepared.chunker.Key()},
+		CreatedAt:     time.Now().UTC(),
+	}
+}
--- a/internal/framework/pipeline/synchronized_collaborators.go
+++ b/internal/framework/pipeline/synchronized_collaborators.go
@@ -61,11 +61,6 @@ func (l *lockedCheckpointLoader) Source(key string) (SourceCheckpoint, Checkpoin
 	defer l.mu.Unlock()
 	return l.inner.Source(key)
 }
-func (l *lockedCheckpointLoader) Chunk(key, digest string) (ChunkCheckpoint, CheckpointDecision) {
-	l.mu.Lock()
-	defer l.mu.Unlock()
-	return l.inner.Chunk(key, digest)
-}
 func (l *lockedCheckpointLoader) Extract(lane, key string, deps []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) {
 	l.mu.Lock()
 	defer l.mu.Unlock()
@@ -107,18 +102,6 @@ func (r *lockedCheckpointRecorder) SourceSucceeded(key string, doc *source.Sourc
 func (r *lockedCheckpointRecorder) SourceFailed(key string, err error) error {
 	return r.call(func() error { return r.inner.SourceFailed(key, err) })
 }
-func (r *lockedCheckpointRecorder) ChunkRunning(key, digest string) error {
-	return r.call(func() error { return r.inner.ChunkRunning(key, digest) })
-}
-func (r *lockedCheckpointRecorder) ChunkSucceeded(key, digest string, chunks []source.Chunk, warnings []contracts.Warning) error {
-	return r.call(func() error { return r.inner.ChunkSucceeded(key, digest, chunks, warnings) })
-}
-func (r *lockedCheckpointRecorder) ChunkRejected(key, digest string, rejected contracts.RejectedOutput) error {
-	return r.call(func() error { return r.inner.ChunkRejected(key, digest, rejected) })
-}
-func (r *lockedCheckpointRecorder) ChunkFailed(key, digest string, err error) error {
-	return r.call(func() error { return r.inner.ChunkFailed(key, digest, err) })
-}
 func (r *lockedCheckpointRecorder) ExtractRunning(lane, key string, deps []CheckpointFingerprint) error {
 	return r.call(func() error { return r.inner.ExtractRunning(lane, key, deps) })
 }