From ef6926322de146717dfcf637600ad71835d8ad5b Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Fri, 17 Jul 2026 14:05:21 +0000 Subject: [PATCH] Remove obsolete sequential extraction path --- internal/framework/pipeline/runner.go | 5 +- .../framework/pipeline/runner_concurrent.go | 55 ++++-- .../pipeline/runner_extract_handoff_test.go | 180 ++++++++++++++++++ internal/framework/pipeline/runner_typed.go | 124 +----------- 4 files changed, 226 insertions(+), 138 deletions(-) create mode 100644 internal/framework/pipeline/runner_extract_handoff_test.go diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 3f9c395..3794cc0 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -54,9 +54,8 @@ type RunInput struct { // single worker so direct framework callers retain deterministic behavior. ExtractWorkers int - pipeline ResolvedPipeline - llmClient contracts.StructuredLLMClient - extractDecision *CheckpointDecision + pipeline ResolvedPipeline + llmClient contracts.StructuredLLMClient } type RunOutput struct { diff --git a/internal/framework/pipeline/runner_concurrent.go b/internal/framework/pipeline/runner_concurrent.go index e2c9da6..b82b31c 100644 --- a/internal/framework/pipeline/runner_concurrent.go +++ b/internal/framework/pipeline/runner_concurrent.go @@ -28,6 +28,22 @@ type laneExtractState struct { failed bool } +type finalizedExtractResults struct { + accepted []erasedExtractArtifact + serialized []CheckpointArtifact + warnings []contracts.Warning + rejected []contracts.RejectedOutput + decision CheckpointDecision +} + +func loadExtract(loader CheckpointLoader, laneID, moduleKey string, deps []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) { + return loader.Extract(laneID, moduleKey, deps) +} + +func recordExtract(recorder CheckpointRecorder, laneID, moduleKey string, deps []CheckpointFingerprint, outputs []CheckpointArtifact, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error { + return recorder.ExtractSucceeded(laneID, moduleKey, deps, outputs, rejected, warnings) +} + type extractJob struct { lane *laneExtractState chunk source.Chunk @@ -56,19 +72,6 @@ type orderedRunError struct { err error } -type completedExtractLoader struct { - CheckpointLoader - laneID string - checkpoint ExtractCheckpoint -} - -func (l completedExtractLoader) Extract(laneID, _ string, _ []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) { - if laneID == l.laneID { - return l.checkpoint, CheckpointDecision{Reused: true, Reason: "coordinated extract result"} - } - return ExtractCheckpoint{}, CheckpointDecision{Reason: "extract result unavailable"} -} - func (r *Runner) runLanes(parent context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []source.Chunk) (RunOutput, error) { output := RunOutput{Manifest: manifestFromPipeline(input)} states := make([]*laneExtractState, len(input.Prepared.lanes)) @@ -85,6 +88,8 @@ func (r *Runner) runLanes(parent context.Context, input RunInput, checkpoints Ch if err := checkpoints.ExtractRunning(prepared.resolved.ID, prepared.resolved.Extract.Module, state.deps); err != nil { return output, fmt.Errorf("write extract checkpoint for lane %q: %w", prepared.resolved.ID, err) } + } else if err := finalizeLaneExtract(checkpoints, state); err != nil { + return output, err } states[i] = state } @@ -325,10 +330,26 @@ func finalizeLaneExtract(checkpoints CheckpointRecorder, state *laneExtractState func (r *Runner) continueLane(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []source.Chunk, state *laneExtractState) (RunOutput, error) { lane := state.prepared.resolved local := RunOutput{Manifest: manifestFromPipeline(input)} - checkpoint := ExtractCheckpoint{Outputs: state.serialized, Rejected: state.rejected, Warnings: state.warnings} - coordinatedLoader := completedExtractLoader{CheckpointLoader: loader, laneID: lane.ID, checkpoint: checkpoint} - input.extractDecision = &state.decision - err := r.runTypedLane(ctx, input, checkpoints, coordinatedLoader, doc, sourceInput, sessionID, chunks, state.prepared, &local) + results := finalizedExtractResults{ + accepted: state.values, + serialized: state.serialized, + warnings: state.warnings, + rejected: state.rejected, + decision: state.decision, + } + local.Warnings = append(local.Warnings, cloneWarnings(results.warnings)...) + local.Rejected = append(local.Rejected, cloneRejectedOutputs(results.rejected)...) + recordCheckpointEvent(&local, loader, string(StageExtract), lane.ID, lane.Extract.Module, results.decision) + if err := writeDebugTimed(input.Debug, path.Join("extract", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": results.decision.Reused, "decision": results.decision, "source": debugSourceDocumentEnvelope(doc), "chunks": debugSourceChunkEnvelopes(chunks), "options": redactSensitiveMap(lane.Extract.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil { + return local, &laneRunError{stage: StageExtract, err: err} + } + if err := writeDebugTimed(input.Debug, path.Join("extract", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": results.decision.Reused, "outputs": debugCheckpointArtifacts(results.serialized), "rejected": debugRejectedOutputEnvelopes(results.rejected), "warnings": debugWarningEnvelopes(results.warnings)}}); err != nil { + return local, &laneRunError{stage: StageExtract, err: err} + } + if len(results.accepted) == 0 { + return local, nil + } + err := r.continueTypedLane(ctx, input, checkpoints, loader, doc, sourceInput, sessionID, state.prepared, results, &local) return local, err } diff --git a/internal/framework/pipeline/runner_extract_handoff_test.go b/internal/framework/pipeline/runner_extract_handoff_test.go new file mode 100644 index 0000000..8705d56 --- /dev/null +++ b/internal/framework/pipeline/runner_extract_handoff_test.go @@ -0,0 +1,180 @@ +package pipeline + +import ( + "context" + "reflect" + "strings" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +type extractCaptureRecorder struct { + CheckpointRecorder + checkpoint ExtractCheckpoint +} + +func (r *extractCaptureRecorder) ExtractSucceeded(_ string, _ string, _ []CheckpointFingerprint, outputs []CheckpointArtifact, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error { + r.checkpoint = ExtractCheckpoint{ + Outputs: cloneCheckpointArtifacts(outputs), + Rejected: cloneRejectedOutputs(rejected), + Warnings: cloneWarnings(warnings), + } + return nil +} + +type extractResultLoader struct { + CheckpointLoader + checkpoint ExtractCheckpoint + decision CheckpointDecision +} + +func (l *extractResultLoader) Enabled() bool { return true } + +func (l *extractResultLoader) Extract(string, string, []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) { + return ExtractCheckpoint{ + Outputs: cloneCheckpointArtifacts(l.checkpoint.Outputs), + Rejected: cloneRejectedOutputs(l.checkpoint.Rejected), + Warnings: cloneWarnings(l.checkpoint.Warnings), + }, l.decision +} + +func cloneCheckpointArtifacts(values []CheckpointArtifact) []CheckpointArtifact { + if len(values) == 0 { + return nil + } + cloned := make([]CheckpointArtifact, len(values)) + for i := range values { + cloned[i] = cloneCheckpointArtifact(values[i]) + } + return cloned +} + +func TestRunnerContinuesFromFreshAndReusedExtractResults(t *testing.T) { + prepared := preparedAttemptDebugPipeline(t) + extractCalls := 0 + installExtractOperation(prepared, 0, func(_ context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) { + extractCalls++ + return erasedTypedResult{ + Value: typedValueForLane(0, request.Chunk.Index), + Warnings: []contracts.Warning{{Scope: "extract", ReasonCode: "observed", Message: "accepted extract"}}, + }, nil + }) + + freshDebug := newCapturedDebugRecorder() + recorder := &extractCaptureRecorder{CheckpointRecorder: NoopCheckpointRecorder()} + freshLoader := &extractResultLoader{ + CheckpointLoader: NoopCheckpointLoader(), + decision: CheckpointDecision{Reason: "extract checkpoint not found"}, + } + fresh, err := New().Run(context.Background(), RunInput{ + Prepared: prepared, + RawInput: []byte("input"), + Checkpoints: recorder, + Checkpoint: freshLoader, + Debug: freshDebug, + }) + if err != nil { + t.Fatalf("fresh Run() error = %v, want nil", err) + } + if extractCalls != 1 { + t.Fatalf("fresh extract calls = %d, want 1", extractCalls) + } + assertExtractDecision(t, fresh.CheckpointEvents, "executed", "extract checkpoint not found") + assertExtractDebugPaths(t, freshDebug, true) + + reusedDebug := newCapturedDebugRecorder() + reusedLoader := &extractResultLoader{ + CheckpointLoader: NoopCheckpointLoader(), + checkpoint: recorder.checkpoint, + decision: CheckpointDecision{Reused: true, Reason: "extract checkpoint matched"}, + } + reused, err := New().Run(context.Background(), RunInput{ + Prepared: prepared, + RawInput: []byte("input"), + Checkpoint: reusedLoader, + Debug: reusedDebug, + }) + if err != nil { + t.Fatalf("reused Run() error = %v, want nil", err) + } + if extractCalls != 1 { + t.Fatalf("extract calls after reuse = %d, want 1", extractCalls) + } + assertExtractDecision(t, reused.CheckpointEvents, "reused", "extract checkpoint matched") + assertExtractDebugPaths(t, reusedDebug, false) + if !reflect.DeepEqual(reused.NormalizeOutputs, fresh.NormalizeOutputs) { + t.Fatalf("reused normalize outputs = %#v, want fresh outputs %#v", reused.NormalizeOutputs, fresh.NormalizeOutputs) + } + if !reflect.DeepEqual(reused.Warnings, fresh.Warnings) { + t.Fatalf("reused warnings = %#v, want fresh warnings %#v", reused.Warnings, fresh.Warnings) + } +} + +func TestRunnerPromotesOnlyAcceptedExtractRetryWarnings(t *testing.T) { + prepared := preparedAttemptDebugPipeline(t) + prepared.lanes[0].resolved.Extract.Retries = 1 + attempts := 0 + installExtractOperation(prepared, 0, func(_ context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) { + attempts++ + scope := "discarded" + if attempts == 2 { + scope = "accepted" + } + return erasedTypedResult{ + Value: typedValueForLane(0, request.Chunk.Index), + Warnings: []contracts.Warning{{Scope: scope, ReasonCode: "observed", Message: scope}}, + }, nil + }) + validatorCalls := 0 + prepared.lanes[0].extractValidators.validators[0].typedValidate = func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) { + validatorCalls++ + return contracts.ValidationResult{Approved: validatorCalls == 2, ReasonCode: "retry", Message: "retry extract"}, nil + } + debug := newCapturedDebugRecorder() + output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug}) + if err != nil { + t.Fatalf("Run() error = %v, want nil", err) + } + if attempts != 2 { + t.Fatalf("extract attempts = %d, want 2", attempts) + } + if len(output.Warnings) != 1 || output.Warnings[0].Scope != "accepted" { + t.Fatalf("promoted warnings = %#v, want accepted attempt only", output.Warnings) + } + name := "extract/notes/chunk-000001/attempt-02.json" + if !debug.has(name) { + t.Fatalf("debug artifact %q is missing; names = %#v", name, debug.names()) + } +} + +func assertExtractDecision(t *testing.T, events []CheckpointEvent, action string, reason string) { + t.Helper() + for _, event := range events { + if event.Stage == string(StageExtract) { + if event.Action != action || event.Reason != reason { + t.Fatalf("extract checkpoint event = %#v, want action %q reason %q", event, action, reason) + } + return + } + } + t.Fatalf("extract checkpoint event missing from %#v", events) +} + +func assertExtractDebugPaths(t *testing.T, debug *capturedDebugRecorder, wantAttempt bool) { + t.Helper() + for _, name := range []string{"extract/notes/input.json", "extract/notes/output.json"} { + if !debug.has(name) { + t.Fatalf("debug artifact %q is missing; names = %#v", name, debug.names()) + } + } + attemptPath := "extract/notes/chunk-000001/attempt-01.json" + if debug.has(attemptPath) != wantAttempt { + t.Fatalf("attempt debug path present = %t, want %t; names = %#v", debug.has(attemptPath), wantAttempt, debug.names()) + } + input := debug.envelope(t, "extract/notes/input.json") + wantReuse := !wantAttempt + if !strings.Contains(string(debug.json["extract/notes/input.json"]), `"reused":`+map[bool]string{true: "true", false: "false"}[wantReuse]) { + t.Fatalf("extract input payload = %#v, want reused %t", input.Payload, wantReuse) + } +} diff --git a/internal/framework/pipeline/runner_typed.go b/internal/framework/pipeline/runner_typed.go index 098fd1f..fbb922b 100644 --- a/internal/framework/pipeline/runner_typed.go +++ b/internal/framework/pipeline/runner_typed.go @@ -6,25 +6,18 @@ import ( "encoding/hex" "fmt" "path" - "sort" "time" "gitea.maximumdirect.net/eric/notarius/internal/core/source" "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" ) -func loadExtract(loader CheckpointLoader, laneID, moduleKey string, deps []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) { - return loader.Extract(laneID, moduleKey, deps) -} func loadMerge(loader CheckpointLoader, laneID, moduleKey string, deps []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) { return loader.Merge(laneID, moduleKey, deps) } func loadNormalize(loader CheckpointLoader, laneID, moduleKey string, deps []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) { return loader.Normalize(laneID, moduleKey, deps) } -func recordExtract(recorder CheckpointRecorder, laneID, moduleKey string, deps []CheckpointFingerprint, outputs []CheckpointArtifact, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error { - return recorder.ExtractSucceeded(laneID, moduleKey, deps, outputs, rejected, warnings) -} func recordMerge(recorder CheckpointRecorder, laneID, moduleKey string, deps []CheckpointFingerprint, output CheckpointArtifact, warnings []contracts.Warning) error { return recorder.MergeSucceeded(laneID, moduleKey, deps, output, warnings) } @@ -126,8 +119,8 @@ type laneRunError struct { func (e *laneRunError) Error() string { return e.err.Error() } func (e *laneRunError) Unwrap() error { return e.err } -func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []source.Chunk, prepared preparedLaneExecutor, output *RunOutput) (err error) { - activeStage := StageExtract +func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, prepared preparedLaneExecutor, extracts finalizedExtractResults, output *RunOutput) (err error) { + activeStage := StageMerge defer func() { if err != nil { err = &laneRunError{stage: activeStage, err: err} @@ -139,116 +132,11 @@ func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints C } setTypedLaneManifestMetadata(output, lane.ID, typed.extractor, typed.merger, typed.normalizer) - values := make([]erasedExtractArtifact, 0, len(chunks)) - serializedExtracts := make([]CheckpointArtifact, 0, len(chunks)) - extractWarnings := []contracts.Warning{} - rejectedStart := len(output.Rejected) - chunksDigest, err := joinedChunkDigest(chunks) - if err != nil { - return fmt.Errorf("digest chunks for lane %q: %w", lane.ID, err) - } - extractDeps := digestFingerprints("chunks", chunksDigest) - cp, decision := loadExtract(loader, lane.ID, lane.Extract.Module, extractDeps) - if decision.Reused { - for _, stored := range cp.Outputs { - if _, decodeErr := decodeCheckpointArtifact(typed.codec, stored); decodeErr != nil { - decision = CheckpointDecision{Reason: "extract artifact checkpoint codec is incompatible: " + decodeErr.Error()} - break - } - } - } - reportedDecision := decision - if input.extractDecision != nil { - reportedDecision = *input.extractDecision - } - recordCheckpointEvent(output, loader, string(StageExtract), lane.ID, lane.Extract.Module, reportedDecision) - if err := writeDebugTimed(input.Debug, path.Join("extract", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": reportedDecision.Reused, "decision": reportedDecision, "source": debugSourceDocumentEnvelope(doc), "chunks": debugSourceChunkEnvelopes(chunks), "options": redactSensitiveMap(lane.Extract.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil { - return err - } - if decision.Reused { - for _, stored := range cp.Outputs { - value, decodeErr := decodeCheckpointArtifact(typed.codec, stored) - if decodeErr != nil { - return fmt.Errorf("decode extract checkpoint for lane %q: %w", lane.ID, decodeErr) - } - stored = hydrateCheckpointArtifact(typed.codec, stored, value) - artifact := erasedExtractArtifact{LaneID: lane.ID, ExtractorKey: lane.Extract.Module, SourceID: doc.ID, ChunkID: stored.ChunkID, ChunkIndex: stored.ChunkIndex, ChunkRef: stored.ChunkRef, Value: value} - if stored.ChunkIndex >= 0 && stored.ChunkIndex < len(chunks) && artifact.ChunkRef == (source.SourceRef{}) { - artifact.ChunkRef = chunks[stored.ChunkIndex].Ref - } - values = append(values, artifact) - serializedExtracts = append(serializedExtracts, cloneCheckpointArtifact(stored)) - } - extractWarnings = cloneWarnings(cp.Warnings) - output.Warnings = append(output.Warnings, extractWarnings...) - output.Rejected = append(output.Rejected, cloneRejectedOutputs(cp.Rejected)...) - } else { - if err := checkpoints.ExtractRunning(lane.ID, lane.Extract.Module, extractDeps); err != nil { - return fmt.Errorf("write extract checkpoint for lane %q: %w", lane.ID, err) - } - for i := range chunks { - chunk := chunks[i] - var accepted erasedExtractArtifact - var serializedAccepted CheckpointArtifact - var acceptedWarnings []contracts.Warning - ok, rejection, runErr := runWithRetry(ctx, lane.Extract.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { - started := time.Now().UTC() - attemptPath := path.Join("extract", debugPathComponent(lane.ID), fmt.Sprintf("chunk-%06d", chunk.Index+1), fmt.Sprintf("attempt-%02d", attempt)) - attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) - result, callErr := typed.extract(attemptCtx, typed.extractor, contracts.TypedExtractionRequest{Source: doc, Chunk: &chunk, SourceInput: chunkInputMaterial(sourceInput, chunk), SessionID: sessionID, References: CloneReferenceSet(lane.ExtractReferences.ReferenceSet), LLMProfile: lane.Extract.LLMProfile, Metadata: cloneMetadata(input.Metadata)}) - if callErr != nil { - _ = writeDebugAttempt(input.Debug, attemptPath, debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started, Error: callErr.Error()}, llmScope) - return false, nil, fmt.Errorf("extract lane %q chunk %q with extractor %q: %w", lane.ID, chunk.ID, lane.Extract.Module, callErr) - } - artifact := erasedExtractArtifact{LaneID: lane.ID, ExtractorKey: lane.Extract.Module, SourceID: doc.ID, ChunkID: chunk.ID, ChunkIndex: chunk.Index, ChunkRef: chunk.Ref, Value: result.Value} - warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageExtract, laneID: lane.ID, moduleKey: lane.Extract.Module, source: doc, sourceID: doc.ID, sourceInput: chunkInputMaterial(sourceInput, chunk), sessionID: sessionID, references: lane.ExtractReferences.ReferenceSet, metadata: input.Metadata, chunk: &chunk, ref: chunk.Ref, value: result.Value}, prepared.extractValidators, attempt, input.Debug) - if validateErr != nil || rejected != nil { - return false, rejected, validateErr - } - stored, encodeErr := checkpointArtifact(typed.codec, artifact.LaneID, artifact.ExtractorKey, artifact.SourceID, artifact.Value) - if encodeErr != nil { - return false, nil, encodeErr - } - stored.ChunkID, stored.ChunkIndex, stored.ChunkRef = artifact.ChunkID, artifact.ChunkIndex, artifact.ChunkRef - accepted, serializedAccepted = artifact, stored - acceptedWarnings = append(cloneWarnings(result.Warnings), warnings...) - if debugErr := writeDebugAttempt(input.Debug, attemptPath, debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started, Payload: map[string]any{"output": debugCheckpointArtifact(stored), "warnings": debugWarningEnvelopes(acceptedWarnings)}}, llmScope); debugErr != nil { - return false, nil, debugErr - } - return true, nil, nil - }) - if runErr != nil { - _ = checkpoints.ExtractFailed(lane.ID, lane.Extract.Module, extractDeps, runErr) - return runErr - } - if !ok { - output.Rejected = append(output.Rejected, *rejection) - continue - } - values = append(values, accepted) - serializedExtracts = append(serializedExtracts, serializedAccepted) - extractWarnings = append(extractWarnings, acceptedWarnings...) - output.Warnings = append(output.Warnings, acceptedWarnings...) - } - if err := recordExtract(checkpoints, lane.ID, lane.Extract.Module, extractDeps, serializedExtracts, cloneRejectedOutputs(output.Rejected[rejectedStart:]), extractWarnings); err != nil { - return fmt.Errorf("write extract checkpoint for lane %q: %w", lane.ID, err) - } - } - sort.SliceStable(values, func(i, j int) bool { return values[i].ChunkIndex < values[j].ChunkIndex }) - sort.SliceStable(serializedExtracts, func(i, j int) bool { return serializedExtracts[i].ChunkIndex < serializedExtracts[j].ChunkIndex }) - if err := writeDebugTimed(input.Debug, path.Join("extract", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": reportedDecision.Reused, "outputs": debugCheckpointArtifacts(serializedExtracts), "rejected": debugRejectedOutputEnvelopes(output.Rejected[rejectedStart:]), "warnings": debugWarningEnvelopes(extractWarnings)}}); err != nil { - return err - } - if len(values) == 0 { - return nil - } - activeStage = StageMerge - - mergeInputs := make([]contracts.ExtractArtifact[any], len(values)) - for i, value := range values { + mergeInputs := make([]contracts.ExtractArtifact[any], len(extracts.accepted)) + for i, value := range extracts.accepted { mergeInputs[i] = contracts.ExtractArtifact[any]{LaneID: value.LaneID, ExtractorKey: value.ExtractorKey, SourceID: value.SourceID, ChunkID: value.ChunkID, ChunkIndex: value.ChunkIndex, ChunkRef: value.ChunkRef, Value: value.Value} } - mergeDeps := artifactCheckpointDigests(serializedExtracts) + mergeDeps := artifactCheckpointDigests(extracts.serialized) mergeCP, mergeDecision := loadMerge(loader, lane.ID, lane.Merge.Module, mergeDeps) if mergeDecision.Reused { if _, decodeErr := decodeCheckpointArtifact(typed.codec, mergeCP.Output); decodeErr != nil { @@ -256,7 +144,7 @@ func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints C } } recordCheckpointEvent(output, loader, string(StageMerge), lane.ID, lane.Merge.Module, mergeDecision) - if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageMerge), LaneID: lane.ID, ModuleKey: lane.Merge.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": mergeDecision.Reused, "decision": mergeDecision, "source": debugSourceDocumentEnvelope(doc), "extract_outputs": debugCheckpointArtifacts(serializedExtracts), "options": redactSensitiveMap(lane.Merge.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil { + if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageMerge), LaneID: lane.ID, ModuleKey: lane.Merge.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": mergeDecision.Reused, "decision": mergeDecision, "source": debugSourceDocumentEnvelope(doc), "extract_outputs": debugCheckpointArtifacts(extracts.serialized), "options": redactSensitiveMap(lane.Merge.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil { return err } var merged erasedMergeArtifact