From 8e669a1f146622f785a1da55dd2fdb3ef01a9100 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sun, 9 Aug 2026 01:19:55 +0000 Subject: [PATCH] Preserve terminal rejection warnings --- .../assembled_spell_pipeline_contract_test.go | 8 +- .../cli/spell_catalog_retry_contract_test.go | 2 +- internal/framework/pipeline/runner.go | 43 +++--- .../framework/pipeline/runner_chunk_plan.go | 36 +++-- .../framework/pipeline/runner_concurrent.go | 33 ++-- .../runner_rejection_warnings_test.go | 142 ++++++++++++++++++ internal/framework/pipeline/runner_typed.go | 66 ++++---- 7 files changed, 254 insertions(+), 76 deletions(-) create mode 100644 internal/framework/pipeline/runner_rejection_warnings_test.go diff --git a/internal/cli/assembled_spell_pipeline_contract_test.go b/internal/cli/assembled_spell_pipeline_contract_test.go index 03d4baf..2891bac 100644 --- a/internal/cli/assembled_spell_pipeline_contract_test.go +++ b/internal/cli/assembled_spell_pipeline_contract_test.go @@ -134,7 +134,7 @@ func TestAssembledSpellPipelineHonorsNormalizeValidatorOverride(t *testing.T) { } } -func TestAssembledSpellPipelineRejectsUnknownSpellWithoutPromotingAttemptWarning(t *testing.T) { +func TestAssembledSpellPipelinePromotesTerminalUnknownSpellWarning(t *testing.T) { registries, resolved, _ := assembledSpellPipeline(t, assembledSpellPipelineOptions{unknownSpell: true}) prepared, err := pipeline.Prepare(resolved, registries, pipeline.ModuleDependencies{}) if err != nil { @@ -161,10 +161,8 @@ func TestAssembledSpellPipelineRejectsUnknownSpellWithoutPromotingAttemptWarning if !reflect.DeepEqual(rejectedFile.Rejected, output.Rejected) { t.Fatalf("rejected file = %#v, run rejections = %#v, want durable rejection diagnostic", rejectedFile.Rejected, output.Rejected) } - for _, warning := range output.Warnings { - if warning.ReasonCode == spellnormalize.ReasonCodeSpellNameUnresolved { - t.Fatalf("warnings = %#v, want rejected-attempt warning to remain non-durable", output.Warnings) - } + if len(output.Warnings) != 1 || output.Warnings[0].ReasonCode != spellnormalize.ReasonCodeSpellNameUnresolved || output.Warnings[0].Scope != "spell_casts[0]" { + t.Fatalf("warnings = %#v, want terminal normalize catalog warning", output.Warnings) } } diff --git a/internal/cli/spell_catalog_retry_contract_test.go b/internal/cli/spell_catalog_retry_contract_test.go index a6c54e6..2a246b4 100644 --- a/internal/cli/spell_catalog_retry_contract_test.go +++ b/internal/cli/spell_catalog_retry_contract_test.go @@ -92,7 +92,7 @@ func TestProductionSpellCatalogValidationRetries(t *testing.T) { t.Fatalf("rejection = %#v, want exhausted unknown-spell rejection", rejection) } if len(output.Warnings) != 0 { - t.Fatalf("warnings = %#v, want no warnings from rejected attempts", output.Warnings) + t.Fatalf("warnings = %#v, want no emitted warnings from rejected attempts", output.Warnings) } return } diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index d74123e..79f7b70 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -240,9 +240,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, 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...) - } + output.Warnings = append(output.Warnings, chunkResult.warnings...) chunkDebugPayload := map[string]any{ "cache_mode": chunkMode, "lookup": chunkResult.lookup, @@ -422,45 +420,52 @@ func (r *Runner) runPreparedSteps(ctx context.Context, input RunInput, checkpoin return nil } -func runWithRetry(ctx context.Context, retries int, run func(attempt int) (bool, *contracts.RejectedOutput, error)) (bool, *contracts.RejectedOutput, error) { +type retryAttemptResult struct { + accepted bool + rejection *contracts.RejectedOutput + warnings []contracts.Warning +} + +func runWithRetry(ctx context.Context, retries int, run func(attempt int) (retryAttemptResult, error)) (retryAttemptResult, error) { attempts := retries + 1 - var last *contracts.RejectedOutput + var last retryAttemptResult for attempt := 1; attempt <= attempts; attempt++ { if err := ctx.Err(); err != nil { - return false, nil, err + return retryAttemptResult{}, err } - accepted, rejection, err := run(attempt) + result, err := run(attempt) if err != nil { var debugErr *attemptDebugPersistenceError if errors.As(err, &debugErr) { - return false, nil, fmt.Errorf("failed after %d attempt(s): %w", attempt, err) + return retryAttemptResult{}, fmt.Errorf("failed after %d attempt(s): %w", attempt, err) } if attempt == attempts { - return false, nil, fmt.Errorf("failed after %d attempt(s): %w", attempt, err) + return retryAttemptResult{}, fmt.Errorf("failed after %d attempt(s): %w", attempt, err) } if ctxErr := ctx.Err(); ctxErr != nil { - return false, nil, ctxErr + return retryAttemptResult{}, ctxErr } continue } - if accepted { - return true, nil, nil + if result.accepted { + return result, nil } - if rejection != nil { + if result.rejection != nil { + rejection := *result.rejection rejection.AttemptCount = attempt - last = rejection + last = retryAttemptResult{rejection: &rejection, warnings: cloneWarnings(result.warnings)} } if ctxErr := ctx.Err(); ctxErr != nil { - return false, nil, ctxErr + return retryAttemptResult{}, ctxErr } if attempt == attempts { - if last == nil { - last = &contracts.RejectedOutput{ReasonCode: "output_rejected", Message: "output rejected", AttemptCount: attempt} + if last.rejection == nil { + last.rejection = &contracts.RejectedOutput{ReasonCode: "output_rejected", Message: "output rejected", AttemptCount: attempt} } - return false, last, nil + return last, nil } } - return false, last, nil + return last, nil } func (r *Runner) validateChunks(ctx context.Context, doc *source.SourceDocument, moduleKey string, chunks []source.Chunk, sourceInput contracts.LLMInputMaterial, sessionID string, references contracts.ReferenceSet, metadata map[string]any, prepared preparedValidatorChain, attempt int, debug DebugRecorder) ([]contracts.Warning, *contracts.RejectedOutput, error) { diff --git a/internal/framework/pipeline/runner_chunk_plan.go b/internal/framework/pipeline/runner_chunk_plan.go index ad6a0a5..2cb21f3 100644 --- a/internal/framework/pipeline/runner_chunk_plan.go +++ b/internal/framework/pipeline/runner_chunk_plan.go @@ -84,14 +84,14 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S } var producerWarnings []contracts.Warning - accepted, rejection, err := runWithRetry(ctx, input.pipeline.Chunk.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { + retryResult, err := runWithRetry(ctx, input.pipeline.Chunk.Retries, func(attempt int) (retryAttemptResult, 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}) requestMetadata, metadataErr := cloneMetadata(input.Metadata) if metadataErr != nil { - return false, nil, terminal.record(nil, fmt.Errorf("clone chunk request metadata: %w", metadataErr)) + return retryAttemptResult{}, terminal.record(nil, fmt.Errorf("clone chunk request metadata: %w", metadataErr)) } chunkResult, callErr := chunker.Plan(attemptCtx, contracts.ChunkRequest{ Source: doc, SourceInput: sourceInput.Clone(), SessionID: sessionID, @@ -99,21 +99,21 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S LLMProfile: input.pipeline.Chunk.LLMProfile, Metadata: requestMetadata, }) if callErr != nil { - return false, nil, terminal.record(nil, fmt.Errorf("chunk source with chunker %q: %w", chunker.Key(), callErr)) + return retryAttemptResult{}, 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) + return retryAttemptResult{}, terminal.record(payload, attemptErr) } planDigest, digestErr := source.DigestChunkPlan(plan) if digestErr != nil { - return false, nil, terminal.record(nil, fmt.Errorf("digest generated chunk plan: %w", digestErr)) + return retryAttemptResult{}, terminal.record(nil, fmt.Errorf("digest generated chunk plan: %w", digestErr)) } producerMetadata, _, metadataErr := moduleManifestMetadata(chunker) if metadataErr != nil { - return false, nil, terminal.record(nil, fmt.Errorf("clone chunker manifest metadata: %w", metadataErr)) + return retryAttemptResult{}, terminal.record(nil, fmt.Errorf("clone chunker manifest metadata: %w", metadataErr)) } profile := "" if input.pipeline.ChunkExecutionClass == contracts.ExecutionClassLLMBacked { @@ -137,7 +137,7 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S action = "bypassed" } if candidateErr := result.setCandidate(candidate, action); candidateErr != nil { - return false, nil, terminal.record(nil, fmt.Errorf("clone generated chunk plan record: %w", candidateErr)) + return retryAttemptResult{}, terminal.record(nil, fmt.Errorf("clone generated chunk plan record: %w", candidateErr)) } 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...) @@ -145,9 +145,16 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S "plan": debugChunkPlanEnvelope(plan), "materialized_chunks": debugSourceChunkEnvelopes(chunks), "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected), } - if validationErr != nil || rejected != nil { + if validationErr != nil { result.setValidation(validationWarnings, rejected, validationErr) - return false, rejected, terminal.record(payload, validationErr) + return retryAttemptResult{}, terminal.record(payload, validationErr) + } + if rejected != nil { + result.setValidation(validationWarnings, rejected, nil) + if debugErr := terminal.record(payload, nil); debugErr != nil { + return retryAttemptResult{}, debugErr + } + return retryAttemptResult{rejection: rejected, warnings: attemptWarnings}, nil } result.chunks = chunks result.plan = &plan @@ -155,9 +162,9 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S producerWarnings = cloneWarnings(chunkResult.Warnings) result.setValidation(validationWarnings, nil, nil) if debugErr := terminal.record(payload, nil); debugErr != nil { - return false, nil, debugErr + return retryAttemptResult{}, debugErr } - return true, nil, nil + return retryAttemptResult{accepted: true}, nil }) if err != nil { if result.summary.ValidationStatus == "not_run" { @@ -165,9 +172,10 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S } return result, err } - result.accepted = accepted - result.rejection = rejection - if !accepted { + result.accepted = retryResult.accepted + result.rejection = retryResult.rejection + if !retryResult.accepted { + result.warnings = cloneWarnings(retryResult.warnings) return result, nil } diff --git a/internal/framework/pipeline/runner_concurrent.go b/internal/framework/pipeline/runner_concurrent.go index e68acee..39781b4 100644 --- a/internal/framework/pipeline/runner_concurrent.go +++ b/internal/framework/pipeline/runner_concurrent.go @@ -254,6 +254,9 @@ func (e *laneEngine) startContinuationWorkers() { } } +// collect is the sole owner of pending, launch, and completion accounting. It +// must drain closed extract results and one completion for every launched +// continuation, even after cancellation. func (e *laneEngine) collect() { resultChannel := (<-chan extractJobResult)(e.results) for resultChannel != nil || len(e.pending) > 0 || e.completed < e.launched { @@ -407,20 +410,20 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source. var accepted erasedExtractArtifact var serialized CheckpointArtifact var acceptedWarnings []contracts.Warning - ok, rejection, err := runWithRetry(ctx, lane.Extract.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { + retryResult, err := runWithRetry(ctx, lane.Extract.Retries, func(attempt int) (retryAttemptResult, error) { started := time.Now().UTC() attemptPath := path.Join("extract", fileio.EncodePathComponent(lane.ID), fmt.Sprintf("chunk-%06d", chunk.Index+1), fmt.Sprintf("attempt-%02d", attempt)) attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "extract", llmScope, debugTimedEnvelope{Stage: string(StageExtract), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started}) requestMetadata, metadataErr := cloneMetadata(input.Metadata) if metadataErr != nil { - return false, nil, terminal.record(nil, fmt.Errorf("clone extract request metadata: %w", metadataErr)) + return retryAttemptResult{}, terminal.record(nil, fmt.Errorf("clone extract request metadata: %w", metadataErr)) } extractReferences := operationReferenceSet(input, lane.ExtractReferences) extracted, callErr := typed.extract(attemptCtx, typed.extractor, contracts.TypedExtractionRequest{Source: doc, Chunk: &chunk, SourceInput: chunkInputMaterial(sourceInput, chunk), SessionID: sessionID, References: CloneReferenceSet(extractReferences), LLMProfile: lane.Extract.LLMProfile, Metadata: requestMetadata}) if callErr != nil { attemptErr := fmt.Errorf("extract lane %q chunk %q with extractor %q: %w", lane.ID, chunk.ID, lane.Extract.Module, callErr) - return false, nil, terminal.record(nil, attemptErr) + return retryAttemptResult{}, terminal.record(nil, attemptErr) } artifact := erasedExtractArtifact{LaneID: lane.ID, ExtractorKey: lane.Extract.Module, SourceID: doc.ID, ChunkID: chunk.ID, ChunkIndex: chunk.Index, ChunkRef: chunk.Ref, Value: extracted.Value} attemptWarnings := cloneWarnings(extracted.Warnings) @@ -428,31 +431,38 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source. if encodeErr != nil { attemptErr := fmt.Errorf("serialize extract candidate for lane %q chunk %q: %w", lane.ID, chunk.ID, encodeErr) payload := map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)} - return false, nil, terminal.record(payload, attemptErr) + return retryAttemptResult{}, terminal.record(payload, attemptErr) } serializedCandidate.ChunkID, serializedCandidate.ChunkIndex, serializedCandidate.ChunkRef = artifact.ChunkID, artifact.ChunkIndex, artifact.ChunkRef warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageExtract, stepID: input.stepID, laneID: lane.ID, moduleKey: lane.Extract.Module, source: doc, sourceID: doc.ID, sourceInput: chunkInputMaterial(sourceInput, chunk), sessionID: sessionID, references: extractReferences, metadata: input.Metadata, chunk: &chunk, ref: chunk.Ref, value: extracted.Value, candidate: &serializedCandidate}, state.prepared.extractValidators, attempt, input.Debug) attemptWarnings = append(attemptWarnings, warnings...) payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected)} - if validateErr != nil || rejected != nil { - return false, rejected, terminal.record(payload, validateErr) + if validateErr != nil { + return retryAttemptResult{}, terminal.record(payload, validateErr) + } + if rejected != nil { + if debugErr := terminal.record(payload, nil); debugErr != nil { + return retryAttemptResult{}, debugErr + } + return retryAttemptResult{rejection: rejected, warnings: attemptWarnings}, nil } stored, encodeErr := checkpointArtifact(typed.codec, artifact.LaneID, artifact.ExtractorKey, artifact.SourceID, artifact.Value) if encodeErr != nil { attemptErr := fmt.Errorf("serialize accepted extract output for lane %q chunk %q: %w", lane.ID, chunk.ID, encodeErr) - return false, nil, terminal.record(payload, attemptErr) + return retryAttemptResult{}, terminal.record(payload, attemptErr) } stored.ChunkID, stored.ChunkIndex, stored.ChunkRef = artifact.ChunkID, artifact.ChunkIndex, artifact.ChunkRef accepted, serialized = artifact, stored acceptedWarnings = attemptWarnings if debugErr := terminal.record(payload, nil); debugErr != nil { - return false, nil, debugErr + return retryAttemptResult{}, debugErr } - return true, nil, nil + return retryAttemptResult{accepted: true}, nil }) result.err = err - if err == nil && !ok { - result.rejected = rejection + if err == nil && !retryResult.accepted { + result.rejected = retryResult.rejection + result.warnings = cloneWarnings(retryResult.warnings) return result } result.value, result.serialized, result.warnings = accepted, serialized, acceptedWarnings @@ -470,6 +480,7 @@ func finalizeLaneExtract(checkpoints CheckpointRecorder, stepID string, state *l result := state.results[index] if result.rejected != nil { state.rejected = append(state.rejected, *result.rejected) + state.warnings = append(state.warnings, result.warnings...) continue } state.values = append(state.values, result.value) diff --git a/internal/framework/pipeline/runner_rejection_warnings_test.go b/internal/framework/pipeline/runner_rejection_warnings_test.go new file mode 100644 index 0000000..becebe0 --- /dev/null +++ b/internal/framework/pipeline/runner_rejection_warnings_test.go @@ -0,0 +1,142 @@ +package pipeline + +import ( + "context" + "fmt" + "reflect" + "strings" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +type warningChunker struct { + key string + plan source.ChunkPlan + calls int +} + +func (c *warningChunker) Key() string { return c.key } +func (*warningChunker) ReferenceSlots() []contracts.ReferenceSlot { return nil } +func (c *warningChunker) Plan(context.Context, contracts.ChunkRequest) (contracts.ChunkPlanResult, error) { + c.calls++ + return contracts.ChunkPlanResult{ + Plan: source.CloneChunkPlan(c.plan), + Warnings: []contracts.Warning{{Scope: fmt.Sprintf("operation-%d", c.calls), ReasonCode: "operation", Message: "operation warning"}}, + }, nil +} + +type chunkValidationFunc struct { + name string + validate func(context.Context, contracts.ChunkValidationRequest) (contracts.ValidationResult, error) +} + +func (v chunkValidationFunc) Name() string { return v.name } +func (chunkValidationFunc) ExecutionClass() contracts.ExecutionClass { + return contracts.ExecutionClassDeterministic +} +func (v chunkValidationFunc) Validate(ctx context.Context, request contracts.ChunkValidationRequest) (contracts.ValidationResult, error) { + return v.validate(ctx, request) +} + +func TestRunnerPromotesOnlyTerminalRejectionWarnings(t *testing.T) { + for _, target := range []ModuleStage{StageChunk, StageExtract, StageMerge, StageNormalize} { + t.Run(string(target), func(t *testing.T) { + prepared := preparedAttemptDebugPipeline(t) + lane := &prepared.Steps[0].lanes[0] + attempts := 0 + first := func() contracts.ValidationResult { + return contracts.ValidationResult{Approved: true, Warnings: []contracts.Warning{{Scope: fmt.Sprintf("validator-%d", attempts), ReasonCode: "validator", Message: "validator warning"}}} + } + reject := func() contracts.ValidationResult { + return contracts.ValidationResult{Approved: false, ReasonCode: "rejected", Message: "rejected"} + } + debug := newCapturedDebugRecorder() + recorder := &extractCaptureRecorder{CheckpointRecorder: NoopCheckpointRecorder()} + + switch target { + case StageChunk: + chunker := prepared.chunker.(*typedTestChunker) + prepared.chunker = &warningChunker{key: prepared.resolved.Chunk.Module, plan: source.CloneChunkPlan(chunker.plan)} + prepared.resolved.Chunk.Retries = 1 + prepared.chunkValidators.validators = []preparedValidator{ + {resolved: ResolvedValidator{Binding: Binding("warning-approval"), Target: ValidatorTargetChunk}, chunk: chunkValidationFunc{name: "warning-approval", validate: func(context.Context, contracts.ChunkValidationRequest) (contracts.ValidationResult, error) { + return first(), nil + }}}, + {resolved: ResolvedValidator{Binding: Binding("warning-rejection"), Target: ValidatorTargetChunk}, chunk: chunkValidationFunc{name: "warning-rejection", validate: func(context.Context, contracts.ChunkValidationRequest) (contracts.ValidationResult, error) { + return reject(), nil + }}}, + } + chunkerWithWarnings := prepared.chunker.(*warningChunker) + first = func() contracts.ValidationResult { + return contracts.ValidationResult{Approved: true, Warnings: []contracts.Warning{{Scope: fmt.Sprintf("validator-%d", chunkerWithWarnings.calls), ReasonCode: "validator", Message: "validator warning"}}} + } + case StageExtract: + lane.resolved.Extract.Retries = 1 + installExtractOperation(prepared, 0, func(context.Context, contracts.TypedExtractionRequest) (erasedTypedResult, error) { + attempts++ + return erasedTypedResult{Value: codecNotes{Items: []string{"extract"}}, Warnings: []contracts.Warning{{Scope: fmt.Sprintf("operation-%d", attempts), ReasonCode: "operation", Message: "operation warning"}}}, nil + }) + lane.extractValidators.validators = rejectionWarningTypedValidators(first, reject) + case StageMerge: + lane.resolved.Merge.Retries = 1 + lane.typed.merge = func(context.Context, any, contracts.TypedMergeRequest[any]) (erasedTypedResult, error) { + attempts++ + return erasedTypedResult{Value: codecNotes{Items: []string{"merge"}}, Warnings: []contracts.Warning{{Scope: fmt.Sprintf("operation-%d", attempts), ReasonCode: "operation", Message: "operation warning"}}}, nil + } + lane.mergeValidators.validators = rejectionWarningTypedValidators(first, reject) + case StageNormalize: + lane.resolved.Normalize.Retries = 1 + lane.typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) { + attempts++ + return erasedTypedResult{Value: codecNotes{Items: []string{"normalize"}}, Warnings: []contracts.Warning{{Scope: fmt.Sprintf("operation-%d", attempts), ReasonCode: "operation", Message: "operation warning"}}}, nil + } + lane.normalizeValidators.validators = rejectionWarningTypedValidators(first, reject) + } + + output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoints: recorder, Debug: debug}) + if err != nil { + t.Fatalf("Run() error = %v, want nil", err) + } + wantScopes := []string{"operation-2", "validator-2"} + if got := rejectionWarningScopes(output.Warnings); !reflect.DeepEqual(got, wantScopes) { + t.Fatalf("published warning scopes = %#v, want %#v", got, wantScopes) + } + if len(output.Rejected) != 1 || output.Rejected[0].AttemptCount != 2 { + t.Fatalf("rejections = %#v, want final rejection after two attempts", output.Rejected) + } + if target == StageExtract && !reflect.DeepEqual(rejectionWarningScopes(recorder.checkpoint.Warnings), wantScopes) { + t.Fatalf("extract checkpoint warnings = %#v, want %#v", recorder.checkpoint.Warnings, wantScopes) + } + attemptPath := fmt.Sprintf("%s/notes/attempt-01.json", target) + if target == StageChunk { + attemptPath = "chunk/attempt-01.json" + } else if target == StageExtract { + attemptPath = "extract/notes/chunk-000001/attempt-01.json" + } + if !strings.Contains(string(debug.json[attemptPath]), "operation-1") || !strings.Contains(string(debug.json[attemptPath]), "validator-1") { + t.Fatalf("first attempt debug = %s, want discarded warnings", debug.json[attemptPath]) + } + }) + } +} + +func rejectionWarningTypedValidators(first func() contracts.ValidationResult, reject func() contracts.ValidationResult) []preparedValidator { + return []preparedValidator{ + {resolved: ResolvedValidator{Binding: Binding("warning-approval"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"}, typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) { + return first(), nil + }}, + {resolved: ResolvedValidator{Binding: Binding("warning-rejection"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"}, typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) { + return reject(), nil + }}, + } +} + +func rejectionWarningScopes(warnings []contracts.Warning) []string { + scopes := make([]string, len(warnings)) + for index := range warnings { + scopes[index] = warnings[index].Scope + } + return scopes +} diff --git a/internal/framework/pipeline/runner_typed.go b/internal/framework/pipeline/runner_typed.go index 7eed857..ed27bf3 100644 --- a/internal/framework/pipeline/runner_typed.go +++ b/internal/framework/pipeline/runner_typed.go @@ -249,52 +249,59 @@ func (r *Runner) runMergeStage(ctx context.Context, input RunInput, checkpoints if err := checkpointMergeRunning(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps); err != nil { return stageResult, err } - ok, rejection, runErr := runWithRetry(ctx, lane.Merge.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { + retryResult, runErr := runWithRetry(ctx, lane.Merge.Retries, func(attempt int) (retryAttemptResult, error) { started := time.Now().UTC() attemptPath := path.Join("merge", fileio.EncodePathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt)) attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "merge", llmScope, debugTimedEnvelope{Stage: string(StageMerge), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Merge.Module, Attempt: attempt, StartedAt: started}) requestMetadata, metadataErr := cloneMetadata(input.Metadata) if metadataErr != nil { - return false, nil, terminal.record(nil, fmt.Errorf("clone merge request metadata: %w", metadataErr)) + return retryAttemptResult{}, terminal.record(nil, fmt.Errorf("clone merge request metadata: %w", metadataErr)) } result, callErr := typed.merge(attemptCtx, typed.merger, contracts.TypedMergeRequest[any]{Source: doc, LaneID: lane.ID, ExtractOutputs: mergeInputs, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(mergeReferences), LLMProfile: lane.Merge.LLMProfile, Metadata: requestMetadata}) if callErr != nil { attemptErr := fmt.Errorf("merge lane %q with merger %q: %w", lane.ID, lane.Merge.Module, callErr) - return false, nil, terminal.record(nil, attemptErr) + return retryAttemptResult{}, terminal.record(nil, attemptErr) } candidate := erasedMergeArtifact{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: result.Value} attemptWarnings := cloneWarnings(result.Warnings) serializedCandidate, encodeErr := serializeCandidateArtifact(typed.codec, candidate.LaneID, candidate.MergerKey, candidate.SourceID, candidate.Value) if encodeErr != nil { attemptErr := fmt.Errorf("serialize merge candidate for lane %q: %w", lane.ID, encodeErr) - return false, nil, terminal.record(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr) + return retryAttemptResult{}, terminal.record(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr) } warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageMerge, stepID: input.stepID, laneID: lane.ID, moduleKey: lane.Merge.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: mergeReferences, metadata: input.Metadata, value: result.Value, candidate: &serializedCandidate}, prepared.mergeValidators, attempt, input.Debug) attemptWarnings = append(attemptWarnings, warnings...) payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected)} - if validateErr != nil || rejected != nil { - return false, rejected, terminal.record(payload, validateErr) + if validateErr != nil { + return retryAttemptResult{}, terminal.record(payload, validateErr) + } + if rejected != nil { + if debugErr := terminal.record(payload, nil); debugErr != nil { + return retryAttemptResult{}, debugErr + } + return retryAttemptResult{rejection: rejected, warnings: attemptWarnings}, nil } stored, encodeErr := checkpointArtifact(typed.codec, candidate.LaneID, candidate.MergerKey, candidate.SourceID, candidate.Value) if encodeErr != nil { attemptErr := fmt.Errorf("serialize accepted merge output for lane %q: %w", lane.ID, encodeErr) - return false, nil, terminal.record(payload, attemptErr) + return retryAttemptResult{}, terminal.record(payload, attemptErr) } if debugErr := terminal.record(payload, nil); debugErr != nil { - return false, nil, debugErr + return retryAttemptResult{}, debugErr } merged, serializedMerge = candidate, stored mergeWarnings = attemptWarnings - return true, nil, nil + return retryAttemptResult{accepted: true}, nil }) if runErr != nil { _ = checkpointMergeFailed(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, 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 { + if !retryResult.accepted { + output.Warnings = append(output.Warnings, retryResult.warnings...) + output.Rejected = append(output.Rejected, *retryResult.rejection) + if err := checkpointMergeRejected(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, *retryResult.rejection); err != nil { return stageResult, err } stageResult.terminal = true @@ -344,30 +351,30 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi if err := checkpointNormalizeRunning(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps); err != nil { return stageResult, err } - ok, rejection, runErr := runWithRetry(ctx, lane.Normalize.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { + retryResult, runErr := runWithRetry(ctx, lane.Normalize.Retries, func(attempt int) (retryAttemptResult, error) { started := time.Now().UTC() attemptPath := path.Join("normalize", fileio.EncodePathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt)) attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "normalize", llmScope, debugTimedEnvelope{Stage: string(StageNormalize), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module, Attempt: attempt, StartedAt: started}) requestMetadata, metadataErr := cloneMetadata(input.Metadata) if metadataErr != nil { - return false, nil, terminal.record(nil, fmt.Errorf("clone normalize request metadata: %w", metadataErr)) + return retryAttemptResult{}, terminal.record(nil, fmt.Errorf("clone normalize request metadata: %w", metadataErr)) } result, callErr := typed.normalize(attemptCtx, typed.normalizer, contracts.TypedNormalizeRequest[any]{Source: doc, LaneID: lane.ID, MergeOutput: contracts.MergeArtifact[any]{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: merged.Value}, SourceInput: sourceInput.Clone(), SessionID: sessionID, References: CloneReferenceSet(normalizeReferences), LLMProfile: lane.Normalize.LLMProfile, Metadata: requestMetadata}) if callErr != nil { attemptErr := fmt.Errorf("normalize lane %q with normalizer %q: %w", lane.ID, lane.Normalize.Module, callErr) - return false, nil, terminal.record(nil, attemptErr) + return retryAttemptResult{}, terminal.record(nil, attemptErr) } attemptWarnings := cloneWarnings(result.Warnings) serializedCandidate, encodeErr := serializeCandidateArtifact(typed.codec, lane.ID, lane.Normalize.Module, doc.ID, result.Value) if encodeErr != nil { attemptErr := fmt.Errorf("serialize normalize candidate for lane %q: %w", lane.ID, encodeErr) - return false, nil, terminal.record(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr) + return retryAttemptResult{}, terminal.record(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr) } var retryPayload map[string]any if result.Retry != nil { if err := validateNormalizeRetry(result.Retry); err != nil { - return false, nil, terminal.record(map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings)}, fmt.Errorf("normalize lane %q returned invalid retry directive: %w", lane.ID, err)) + return retryAttemptResult{}, terminal.record(map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings)}, fmt.Errorf("normalize lane %q returned invalid retry directive: %w", lane.ID, err)) } retryRemaining := attempt <= lane.Normalize.Retries retryPayload = map[string]any{ @@ -378,7 +385,7 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi } if retryRemaining { payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings), "retry": retryPayload} - return false, nil, terminal.record(payload, nil) + return retryAttemptResult{}, terminal.record(payload, nil) } attemptWarnings = append(attemptWarnings, cloneWarnings(result.Retry.FallbackWarnings)...) } @@ -388,28 +395,35 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi if retryPayload != nil { payload["retry"] = retryPayload } - if validateErr != nil || rejected != nil { - return false, rejected, terminal.record(payload, validateErr) + if validateErr != nil { + return retryAttemptResult{}, terminal.record(payload, validateErr) + } + if rejected != nil { + if debugErr := terminal.record(payload, nil); debugErr != nil { + return retryAttemptResult{}, debugErr + } + return retryAttemptResult{rejection: rejected, warnings: attemptWarnings}, nil } stored, encodeErr := checkpointArtifact(typed.codec, lane.ID, lane.Normalize.Module, doc.ID, result.Value) if encodeErr != nil { attemptErr := fmt.Errorf("serialize accepted normalize output for lane %q: %w", lane.ID, encodeErr) - return false, nil, terminal.record(payload, attemptErr) + return retryAttemptResult{}, terminal.record(payload, attemptErr) } if debugErr := terminal.record(payload, nil); debugErr != nil { - return false, nil, debugErr + return retryAttemptResult{}, debugErr } serializedNormalize = stored normalizeWarnings = attemptWarnings - return true, nil, nil + return retryAttemptResult{accepted: true}, nil }) if runErr != nil { _ = checkpointNormalizeFailed(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, 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 { + if !retryResult.accepted { + output.Warnings = append(output.Warnings, retryResult.warnings...) + output.Rejected = append(output.Rejected, *retryResult.rejection) + if err := checkpointNormalizeRejected(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, *retryResult.rejection); err != nil { return stageResult, err } return stageResult, nil