diff --git a/internal/framework/pipeline/checkpoint.go b/internal/framework/pipeline/checkpoint.go index a126ed5..743d52a 100644 --- a/internal/framework/pipeline/checkpoint.go +++ b/internal/framework/pipeline/checkpoint.go @@ -112,6 +112,30 @@ func requireReusableCheckpoint(policy CheckpointExecutionPolicy, stepID, laneID return nil } +// resolveCheckpointDecision applies runner policy and canonical payload +// validation at the single point where a stage's observable decision is made. +func resolveCheckpointDecision(output *RunOutput, loader CheckpointLoader, policy CheckpointExecutionPolicy, stage ModuleStage, stepID, laneID, moduleKey string, decision CheckpointDecision, codec artifactCodecEntry, artifacts []CheckpointArtifact) (CheckpointDecision, error) { + decision = forceCheckpointDecision(policy, stepID, laneID, decision) + if err := requireReusableCheckpoint(policy, stepID, laneID, decision); err != nil { + return decision, err + } + if decision.Reused { + for _, artifact := range artifacts { + if _, _, err := decodeCanonicalCheckpointArtifact(codec, artifact); err != nil { + decision = CheckpointDecision{Category: "executed", ReasonCode: "artifact_not_canonical", Detail: "stored " + string(stage) + " artifact failed canonical codec validation", Reason: string(stage) + " artifact checkpoint is not canonical"} + break + } + } + } + if err := requireReusableCheckpoint(policy, stepID, laneID, decision); err != nil { + return decision, err + } + if output != nil { + recordCheckpointEvent(output, loader, string(stage), stepID, laneID, moduleKey, decision) + } + return decision, nil +} + type SourceCheckpoint struct { Document *source.SourceDocument } diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index f5ad699..30f6867 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -252,22 +252,8 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err } if chunkResult.accepted { - for _, step := range input.Prepared.Steps { - stepInput := input - stepInput.stepID = step.ID - stepReferences, referenceProvenance, handoffErr := buildStepReferenceSets(input, step, output.NormalizeOutputs) - if handoffErr != nil { - return failOutput(output), fmt.Errorf("prepare generated references for pipeline step %q: %w", step.ID, handoffErr) - } - stepInput.references = stepReferences - output.Manifest.References = append(output.Manifest.References, referenceProvenance...) - laneOutput, laneErr := r.runLanes(ctx, stepInput, step, checkpoints, checkpointLoader, doc, sourceInput, sessionID, chunkResult.chunks) - if err := mergeLaneOutput(&output, laneOutput); err != nil { - return failOutput(output), err - } - if laneErr != nil { - return failOutput(output), fmt.Errorf("execute pipeline step %q: %w", step.ID, laneErr) - } + if err := r.runPreparedSteps(ctx, input, checkpoints, checkpointLoader, doc, sourceInput, sessionID, chunkResult.chunks, &output); err != nil { + return failOutput(output), err } } @@ -335,6 +321,27 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err return output, nil } +func (r *Runner) runPreparedSteps(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []source.Chunk, output *RunOutput) error { + for _, step := range input.Prepared.Steps { + stepInput := input + stepInput.stepID = step.ID + stepReferences, referenceProvenance, err := buildStepReferenceSets(input, step, output.NormalizeOutputs) + if err != nil { + return fmt.Errorf("prepare generated references for pipeline step %q: %w", step.ID, err) + } + stepInput.references = stepReferences + output.Manifest.References = append(output.Manifest.References, referenceProvenance...) + laneOutput, laneErr := r.runLanes(ctx, stepInput, step, checkpoints, loader, doc, sourceInput, sessionID, chunks) + if err := mergeLaneOutput(output, laneOutput); err != nil { + return err + } + if laneErr != nil { + return fmt.Errorf("execute pipeline step %q: %w", step.ID, laneErr) + } + } + return nil +} + func runWithRetry(ctx context.Context, retries int, run func(attempt int) (bool, *contracts.RejectedOutput, error)) (bool, *contracts.RejectedOutput, error) { attempts := retries + 1 var last *contracts.RejectedOutput diff --git a/internal/framework/pipeline/runner_concurrent.go b/internal/framework/pipeline/runner_concurrent.go index 341e2f5..01e8c3b 100644 --- a/internal/framework/pipeline/runner_concurrent.go +++ b/internal/framework/pipeline/runner_concurrent.go @@ -77,28 +77,46 @@ type orderedRunError struct { func (r *Runner) runLanes(parent context.Context, input RunInput, step PreparedPipelineStep, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []source.Chunk) (RunOutput, error) { output := RunOutput{Manifest: manifestFromPipeline(input)} + states, err := initializeLaneStates(input, step, checkpoints, loader, doc, chunks, &output) + if err != nil { + return output, err + } + completedOutputs, runErrors := r.runLaneEngine(parent, input, checkpoints, loader, doc, sourceInput, sessionID, chunks, states) + if err := mergeCompletedLanes(&output, completedOutputs); err != nil { + return output, err + } + if err := selectRunError(parent, runErrors); err != nil { + return output, err + } + return output, nil +} + +func initializeLaneStates(input RunInput, step PreparedPipelineStep, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, chunks []source.Chunk, output *RunOutput) ([]*laneExtractState, error) { states := make([]*laneExtractState, len(step.lanes)) for i, prepared := range step.lanes { if prepared.typed == nil { - return output, fmt.Errorf("typed lane %q executor is not prepared", prepared.resolved.ID) + return nil, fmt.Errorf("typed lane %q executor is not prepared", prepared.resolved.ID) } - if err := setTypedLaneManifestMetadata(&output, prepared.resolved.ID, prepared.typed.extractor, prepared.typed.merger, prepared.typed.normalizer); err != nil { - return output, err + if err := setTypedLaneManifestMetadata(output, prepared.resolved.ID, prepared.typed.extractor, prepared.typed.merger, prepared.typed.normalizer); err != nil { + return nil, err } state, err := prepareLaneExtract(input, loader, doc, chunks, i, prepared) if err != nil { - return output, err + return nil, err } if !state.decision.Reused { if err := checkpointExtractRunning(checkpoints, input.stepID, 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) + return nil, fmt.Errorf("write extract checkpoint for lane %q: %w", prepared.resolved.ID, err) } } else if err := finalizeLaneExtract(checkpoints, input.stepID, state); err != nil { - return output, err + return nil, err } states[i] = state } + return states, nil +} +func (r *Runner) runLaneEngine(parent context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []source.Chunk, states []*laneExtractState) ([]RunOutput, []orderedRunError) { workerCount := input.ExtractWorkers if workerCount < 1 { workerCount = 1 @@ -216,15 +234,16 @@ func (r *Runner) runLanes(parent context.Context, input RunInput, step PreparedP } close(continuations) continuationWorkers.Wait() + return completedOutputs, runErrors +} + +func mergeCompletedLanes(output *RunOutput, completedOutputs []RunOutput) error { for i := range completedOutputs { - if err := mergeLaneOutput(&output, completedOutputs[i]); err != nil { - return output, err + if err := mergeLaneOutput(output, completedOutputs[i]); err != nil { + return err } } - if err := selectRunError(parent, runErrors); err != nil { - return output, err - } - return output, nil + return nil } func prepareLaneExtract(input RunInput, loader CheckpointLoader, doc *source.SourceDocument, chunks []source.Chunk, index int, prepared preparedLaneExecutor) (*laneExtractState, error) { @@ -237,19 +256,8 @@ func prepareLaneExtract(input RunInput, loader CheckpointLoader, doc *source.Sou deps := append(digestFingerprints("chunks", digest), generatedReferenceDependencies(extractReferences)...) state := &laneExtractState{index: index, prepared: prepared, deps: normalizeCheckpointFingerprints(deps), remaining: len(chunks), results: make(map[int]extractJobResult, len(chunks))} cp, decision := loadExtract(loader, input.stepID, lane.ID, lane.Extract.Module, state.deps) - decision = forceCheckpointDecision(input.CheckpointPolicy, input.stepID, lane.ID, decision) - if err := requireReusableCheckpoint(input.CheckpointPolicy, input.stepID, lane.ID, decision); err != nil { - return nil, err - } - if decision.Reused { - for _, stored := range cp.Outputs { - if _, _, decodeErr := decodeCanonicalCheckpointArtifact(typed.codec, stored); decodeErr != nil { - decision = CheckpointDecision{Category: "executed", ReasonCode: "artifact_not_canonical", Detail: "stored extract artifact failed canonical codec validation", Reason: "extract artifact checkpoint is not canonical"} - break - } - } - } - if err := requireReusableCheckpoint(input.CheckpointPolicy, input.stepID, lane.ID, decision); err != nil { + decision, err = resolveCheckpointDecision(nil, loader, input.CheckpointPolicy, StageExtract, input.stepID, lane.ID, lane.Extract.Module, decision, typed.codec, cp.Outputs) + if err != nil { return nil, err } state.decision = decision diff --git a/internal/framework/pipeline/runner_typed.go b/internal/framework/pipeline/runner_typed.go index 53c79eb..3bd9b98 100644 --- a/internal/framework/pipeline/runner_typed.go +++ b/internal/framework/pipeline/runner_typed.go @@ -160,13 +160,13 @@ type laneRunError struct { func (e *laneRunError) Error() string { return e.err.Error() } func (e *laneRunError) Unwrap() error { return e.err } -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} - } - }() +type mergeStageResult struct { + artifact erasedMergeArtifact + serialized CheckpointArtifact + terminal bool +} + +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) error { lane, typed := prepared.resolved, prepared.typed if typed == nil { return fmt.Errorf("typed lane %q executor is not prepared", lane.ID) @@ -174,7 +174,26 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi if err := setTypedLaneManifestMetadata(output, lane.ID, typed.extractor, typed.merger, typed.normalizer); err != nil { return err } + merged, err := r.runMergeStage(ctx, input, checkpoints, loader, doc, sourceInput, sessionID, prepared, extracts, output) + if err != nil { + return &laneRunError{stage: StageMerge, err: err} + } + if merged.terminal { + return nil + } + normalized, err := r.runNormalizeStage(ctx, input, checkpoints, loader, doc, sourceInput, sessionID, prepared, merged.artifact, merged.serialized, output) + if err != nil { + return &laneRunError{stage: StageNormalize, err: err} + } + if normalized.accepted { + output.NormalizeOutputs = append(output.NormalizeOutputs, contracts.SerializedOutput{StepID: input.stepID, LaneID: lane.ID, NormalizerKey: lane.Normalize.Module, SourceID: doc.ID, Artifact: contracts.CloneSerializedArtifact(normalized.serialized.Artifact)}) + } + return nil +} +func (r *Runner) runMergeStage(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, prepared preparedLaneExecutor, extracts finalizedExtractResults, output *RunOutput) (mergeStageResult, error) { + var stageResult mergeStageResult + lane, typed := prepared.resolved, prepared.typed 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} @@ -183,21 +202,12 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi mergeDeps := append(artifactCheckpointDigests(extracts.serialized), generatedReferenceDependencies(mergeReferences)...) mergeDeps = normalizeCheckpointFingerprints(mergeDeps) mergeCP, mergeDecision := loadMerge(loader, input.stepID, lane.ID, lane.Merge.Module, mergeDeps) - mergeDecision = forceCheckpointDecision(input.CheckpointPolicy, input.stepID, lane.ID, mergeDecision) - if err := requireReusableCheckpoint(input.CheckpointPolicy, input.stepID, lane.ID, mergeDecision); err != nil { - return err + mergeDecision, err := resolveCheckpointDecision(output, loader, input.CheckpointPolicy, StageMerge, input.stepID, lane.ID, lane.Merge.Module, mergeDecision, typed.codec, []CheckpointArtifact{mergeCP.Output}) + if err != nil { + return stageResult, err } - if mergeDecision.Reused { - if _, _, decodeErr := decodeCanonicalCheckpointArtifact(typed.codec, mergeCP.Output); decodeErr != nil { - mergeDecision = CheckpointDecision{Category: "executed", ReasonCode: "artifact_not_canonical", Detail: "stored merge artifact failed canonical codec validation", Reason: "merge artifact checkpoint is not canonical"} - } - } - if err := requireReusableCheckpoint(input.CheckpointPolicy, input.stepID, lane.ID, mergeDecision); err != nil { - return err - } - recordCheckpointEvent(output, loader, string(StageMerge), input.stepID, lane.ID, lane.Merge.Module, mergeDecision) if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageMerge), StepID: input.stepID, 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 + return stageResult, err } var merged erasedMergeArtifact var serializedMerge CheckpointArtifact @@ -205,7 +215,7 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi if mergeDecision.Reused { value, hydrated, decodeErr := decodeCanonicalCheckpointArtifact(typed.codec, mergeCP.Output) if decodeErr != nil { - return fmt.Errorf("decode merge checkpoint for lane %q: %w", lane.ID, decodeErr) + return stageResult, fmt.Errorf("decode merge checkpoint for lane %q: %w", lane.ID, decodeErr) } merged = erasedMergeArtifact{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: value} serializedMerge = hydrated @@ -213,7 +223,7 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi output.Warnings = append(output.Warnings, mergeWarnings...) } else { if err := checkpointMergeRunning(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps); err != nil { - return err + return stageResult, err } ok, rejection, runErr := runWithRetry(ctx, lane.Merge.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { started := time.Now().UTC() @@ -256,58 +266,62 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi }) if runErr != nil { _ = checkpointMergeFailed(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, runErr) - return runErr + return stageResult, runErr } if !ok { output.Rejected = append(output.Rejected, *rejection) if err := checkpointMergeRejected(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, *rejection); err != nil { - return err + return stageResult, err } - return nil + stageResult.terminal = true + return stageResult, nil } output.Warnings = append(output.Warnings, mergeWarnings...) if err := recordMerge(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, serializedMerge, mergeWarnings); err != nil { - return err + return stageResult, err } } if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageMerge), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Merge.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": mergeDecision.Reused, "accepted": true, "output": debugCheckpointArtifact(serializedMerge), "warnings": debugWarningEnvelopes(mergeWarnings)}}); err != nil { - return err + return stageResult, err } + stageResult.artifact = merged + stageResult.serialized = serializedMerge + return stageResult, nil +} - activeStage = StageNormalize +type normalizeStageResult struct { + serialized CheckpointArtifact + warnings []contracts.Warning + accepted bool +} + +func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, prepared preparedLaneExecutor, merged erasedMergeArtifact, serializedMerge CheckpointArtifact, output *RunOutput) (normalizeStageResult, error) { + var stageResult normalizeStageResult + lane, typed := prepared.resolved, prepared.typed normalizeReferences := operationReferenceSet(input, lane.NormalizeReferences) normalizeDeps := append(artifactCheckpointDigests([]CheckpointArtifact{serializedMerge}), generatedReferenceDependencies(normalizeReferences)...) normalizeDeps = normalizeCheckpointFingerprints(normalizeDeps) normalizeCP, normalizeDecision := loadNormalize(loader, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps) - normalizeDecision = forceCheckpointDecision(input.CheckpointPolicy, input.stepID, lane.ID, normalizeDecision) - if err := requireReusableCheckpoint(input.CheckpointPolicy, input.stepID, lane.ID, normalizeDecision); err != nil { - return err + normalizeDecision, err := resolveCheckpointDecision(output, loader, input.CheckpointPolicy, StageNormalize, input.stepID, lane.ID, lane.Normalize.Module, normalizeDecision, typed.codec, []CheckpointArtifact{normalizeCP.Output}) + if err != nil { + return stageResult, err } - if normalizeDecision.Reused { - if _, _, decodeErr := decodeCanonicalCheckpointArtifact(typed.codec, normalizeCP.Output); decodeErr != nil { - normalizeDecision = CheckpointDecision{Category: "executed", ReasonCode: "artifact_not_canonical", Detail: "stored normalize artifact failed canonical codec validation", Reason: "normalize artifact checkpoint is not canonical"} - } - } - if err := requireReusableCheckpoint(input.CheckpointPolicy, input.stepID, lane.ID, normalizeDecision); err != nil { - return err - } - recordCheckpointEvent(output, loader, string(StageNormalize), input.stepID, lane.ID, lane.Normalize.Module, normalizeDecision) if err := writeDebugTimed(input.Debug, path.Join("normalize", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageNormalize), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": normalizeDecision.Reused, "decision": normalizeDecision, "source": debugSourceDocumentEnvelope(doc), "merge_output": debugCheckpointArtifact(serializedMerge), "options": redactSensitiveMap(lane.Normalize.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil { - return err + return stageResult, err } var serializedNormalize CheckpointArtifact var normalizeWarnings []contracts.Warning if normalizeDecision.Reused { _, hydrated, decodeErr := decodeCanonicalCheckpointArtifact(typed.codec, normalizeCP.Output) if decodeErr != nil { - return fmt.Errorf("decode normalize checkpoint for lane %q: %w", lane.ID, decodeErr) + return stageResult, fmt.Errorf("decode normalize checkpoint for lane %q: %w", lane.ID, decodeErr) } serializedNormalize = hydrated normalizeWarnings = cloneWarnings(normalizeCP.Warnings) output.Warnings = append(output.Warnings, normalizeWarnings...) } else { if err := checkpointNormalizeRunning(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps); err != nil { - return err + return stageResult, err } ok, rejection, runErr := runWithRetry(ctx, lane.Normalize.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { started := time.Now().UTC() @@ -349,25 +363,27 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi }) if runErr != nil { _ = checkpointNormalizeFailed(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, runErr) - return runErr + return stageResult, runErr } if !ok { output.Rejected = append(output.Rejected, *rejection) if err := checkpointNormalizeRejected(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, *rejection); err != nil { - return err + return stageResult, err } - return nil + return stageResult, nil } output.Warnings = append(output.Warnings, normalizeWarnings...) if err := recordNormalize(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, serializedNormalize, normalizeWarnings); err != nil { - return err + return stageResult, err } } if err := writeDebugTimed(input.Debug, path.Join("normalize", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageNormalize), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": normalizeDecision.Reused, "accepted": true, "output": debugCheckpointArtifact(serializedNormalize), "warnings": debugWarningEnvelopes(normalizeWarnings)}}); err != nil { - return err + return stageResult, err } - output.NormalizeOutputs = append(output.NormalizeOutputs, contracts.SerializedOutput{StepID: input.stepID, LaneID: lane.ID, NormalizerKey: lane.Normalize.Module, SourceID: doc.ID, Artifact: contracts.CloneSerializedArtifact(serializedNormalize.Artifact)}) - return nil + stageResult.serialized = serializedNormalize + stageResult.warnings = normalizeWarnings + stageResult.accepted = true + return stageResult, nil } func setTypedLaneManifestMetadata(output *RunOutput, laneID string, extractor, merger, normalizer any) error {