From 29872b2e281134f58fddb9b717d95ba09a2834ca Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Thu, 27 Aug 2026 00:35:37 +0000 Subject: [PATCH] Integrate chunk validation retries --- docs/internal/pipeline.md | 7 + docs/roadmap/implementation.md | 2 +- internal/framework/pipeline/runner.go | 14 +- .../framework/pipeline/runner_chunk_plan.go | 156 ++++++++++++------ .../pipeline/runner_chunk_plan_test.go | 102 +++++++++++- .../runner_rejection_warnings_test.go | 10 +- .../pipeline/runner_terminal_debug_test.go | 5 + 7 files changed, 221 insertions(+), 75 deletions(-) diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index a92f1719..88dcd801 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -162,6 +162,13 @@ settles the semantic policy immediately. Structural-output errors alone use the structural policy, and validation failure without rejection settles the validator-failure policy without regenerating the producer. +Chunk planning uses this state machine for generated plans. A rejected +automatic cache hit is not model material and therefore falls through to a +fresh initial generation; it neither receives a correction nor overwrites the +stored record. Only a newly generated, completely validated plan is published +to the chunk-plan store. Rejected plans never advance, and validation-incomplete +plans remain unpublishable. + After terminal lane work, the runner assembles manifest provenance, normalized artifacts, rejections, warnings, and an optional accepted chunk map. When an output policy selected evidence lanes, it decodes accepted serialized normalize diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 2a5c59a0..2cdc6068 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -515,7 +515,7 @@ machine used by later stage integrations. One tested state machine owns attempt budgets and disposition, but no stage is partially migrated. This stage is one Terra prompt. -## Stage 11 — Integrate Chunk Validation And Correction +## Stage 11 — Integrate Chunk Validation And Correction ✅ ### Goal diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 289a9e5c..40342171 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -470,10 +470,10 @@ func runSimpleRetry(ctx context.Context, retries int, run func(attempt int) (ret 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) { +func (r *Runner) validateChunkReport(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) (validationReport, error) { content, err := json.Marshal(chunks) if err != nil { - return nil, nil, fmt.Errorf("encode canonical chunks for validation: %w", err) + return validationReport{}, fmt.Errorf("encode canonical chunks for validation: %w", err) } schema := contracts.ArtifactSchema{ID: "notarius.source.chunks", Name: "notarius_source_chunks", Version: "v1", JSONSchema: []byte(`{"$schema":"https://json-schema.org/draft/2020-12/schema","type":"array"}`)} report, err := executeValidationChain(ctx, prepared, func(validatorCtx context.Context, item preparedValidator, validatorAttempt int) (validationInvocation, error) { @@ -520,15 +520,9 @@ func (r *Runner) validateChunks(ctx context.Context, doc *source.SourceDocument, return validationInvocation{result: result}, nil }) if err != nil { - return report.Warnings(), nil, err + return report, err } - if failure := report.FirstFailure(); failure != nil { - return report.Warnings(), nil, validatorFailureError(*failure) - } - if rejection := report.FirstRejection(); rejection != nil { - return report.Warnings(), &contracts.RejectedOutput{Stage: string(StageChunk), ModuleKey: moduleKey, ValidatorName: rejection.validatorName, ReasonCode: rejection.reasonCode, Message: rejection.message, AttemptCount: attempt, DiagnosticArtifactPath: rejection.diagnosticPath}, nil - } - return report.Warnings(), nil, nil + return report, nil } func resolvedValidatorChain(stage ModuleStage, laneID string, moduleKey string, chains []ResolvedValidatorChain) ResolvedValidatorChain { diff --git a/internal/framework/pipeline/runner_chunk_plan.go b/internal/framework/pipeline/runner_chunk_plan.go index e9107a62..d7348898 100644 --- a/internal/framework/pipeline/runner_chunk_plan.go +++ b/internal/framework/pipeline/runner_chunk_plan.go @@ -23,6 +23,14 @@ type chunkPlanExecution struct { summary artifacts.ChunkPlanSummary } +type generatedChunkPlanCandidate struct { + plan source.ChunkPlan + chunks []source.Chunk + record ChunkPlanRecord + producerWarnings []contracts.Warning + terminal *attemptTerminalRecorder +} + func effectiveChunkCacheMode(mode ChunkCacheMode) ChunkCacheMode { if mode == "" { return ChunkCacheBypass @@ -58,17 +66,31 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S case ChunkPlanHit: plan, chunks, validationErr := validateAndMaterializeChunkPlan(doc, record.Plan) if validationErr == nil { - if err := result.setCandidate(record, "reused"); err != nil { - return result, fmt.Errorf("clone reused chunk plan record: %w", err) + report, err := r.validateChunkReport(ctx, doc, chunker.Key(), chunks, sourceInput, sessionID, input.pipeline.ChunkReferences.ReferenceSet, input.Metadata, input.Prepared.chunkValidators, 1, input.Debug) + if err != nil { + result.setValidation(report.Warnings(), nil, err) + return result, err } - 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 - result.setValidation(validationWarnings, rejection, err) - return result, err + if report.FirstRejection() == nil { + if incomplete := firstIncompleteValidation(report); incomplete != nil && input.pipeline.ChunkValidationPolicy.ValidatorFailure == ValidatorFailureFailRun { + return result, validatorFailureError(*incomplete) + } + if err := result.setCandidate(record, "reused"); err != nil { + return result, fmt.Errorf("clone reused chunk plan record: %w", err) + } + result.plan = &plan + result.chunks = chunks + result.warnings = append(cloneWarnings(record.Warnings), report.Warnings()...) + result.accepted = true + result.setValidation(report.Warnings(), nil, nil) + if firstIncompleteValidation(report) != nil { + result.summary.ValidationStatus = "incomplete" + } + return result, nil + } + // A cache hit is not model material. Its rejection is discarded and + // generation begins with the ordinary initial request below. + result.setValidation(report.Warnings(), chunkRejection(report, 1, chunker.Key()), nil) } result.lookup = ChunkPlanDecision{Status: ChunkPlanInvalid, Reason: chunkPlanLookupReason(ChunkPlanInvalid)} result.summary.LookupStatus = "invalid" @@ -83,37 +105,41 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S result.summary.PublicationStatus = "not_published" } - var producerWarnings []contracts.Warning - retryResult, err := runSimpleRetry(ctx, input.pipeline.Chunk.Retries, func(attempt int) (retryAttemptResult, error) { + terminal, err := runProducerAttempts(ctx, producerAttemptConfig{ + Retries: input.pipeline.Chunk.Retries, + Policy: input.pipeline.ChunkValidationPolicy, + AllowStructuralRetry: input.pipeline.ChunkExecutionClass == contracts.ExecutionClassLLMBacked, + }, func(attemptCtx context.Context, request producerAttemptRequest) (producerAttemptOutput, error) { + attempt := request.Number 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}) + attemptCtx, llmScope := withDebugLLMScope(attemptCtx, attemptPath) + attemptTerminal := 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 retryAttemptResult{}, terminal.record(nil, fmt.Errorf("clone chunk request metadata: %w", metadataErr)) + return producerAttemptOutput{}, attemptTerminal.record(nil, fmt.Errorf("clone chunk request metadata: %w", metadataErr)) } 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, StructuredOutputRepairAttempts: cloneStructuredOutputRepairAttempts(input.pipeline.Chunk.StructuredOutputRepairAttempts), Metadata: requestMetadata, + LLMProfile: input.pipeline.Chunk.LLMProfile, StructuredOutputRepairAttempts: cloneStructuredOutputRepairAttempts(input.pipeline.Chunk.StructuredOutputRepairAttempts), Correction: request.Correction, Metadata: requestMetadata, }) if callErr != nil { - return retryAttemptResult{}, terminal.record(nil, fmt.Errorf("chunk source with chunker %q: %w", chunker.Key(), callErr)) + return producerAttemptOutput{}, attemptTerminal.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 retryAttemptResult{}, terminal.record(payload, attemptErr) + return producerAttemptOutput{}, attemptTerminal.record(payload, fmt.Errorf("%w: %v", contracts.ErrInvalidStructuredOutput, attemptErr)) } planDigest, digestErr := source.DigestChunkPlan(plan) if digestErr != nil { - return retryAttemptResult{}, terminal.record(nil, fmt.Errorf("digest generated chunk plan: %w", digestErr)) + return producerAttemptOutput{}, attemptTerminal.record(nil, fmt.Errorf("digest generated chunk plan: %w", digestErr)) } producerMetadata, _, metadataErr := moduleManifestMetadata(chunker) if metadataErr != nil { - return retryAttemptResult{}, terminal.record(nil, fmt.Errorf("clone chunker manifest metadata: %w", metadataErr)) + return producerAttemptOutput{}, attemptTerminal.record(nil, fmt.Errorf("clone chunker manifest metadata: %w", metadataErr)) } profile := "" if input.pipeline.ChunkExecutionClass == contracts.ExecutionClassLLMBacked { @@ -129,42 +155,27 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S }, Warnings: cloneWarnings(chunkResult.Warnings), CreatedAt: time.Now().UTC(), } - action := "generated" - if mode == ChunkCacheRefresh { - action = "refreshed" + return producerAttemptOutput{Value: generatedChunkPlanCandidate{plan: plan, chunks: chunks, record: candidate, producerWarnings: cloneWarnings(chunkResult.Warnings), terminal: &attemptTerminal}, Candidate: chunkResult.ModelCandidate, Warnings: cloneWarnings(chunkResult.Warnings)}, nil + }, func(validationCtx context.Context, output producerAttemptOutput) (validationReport, error) { + candidate, ok := output.Value.(generatedChunkPlanCandidate) + if !ok { + return validationReport{}, fmt.Errorf("chunk attempt candidate has incompatible type") } - if mode == ChunkCacheBypass { - action = "bypassed" - } - if candidateErr := result.setCandidate(candidate, action); candidateErr != nil { - 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...) + report, validationErr := r.validateChunkReport(validationCtx, doc, chunker.Key(), candidate.chunks, sourceInput, sessionID, input.pipeline.ChunkReferences.ReferenceSet, input.Metadata, input.Prepared.chunkValidators, candidate.terminal.envelope.Attempt, input.Debug) + attemptWarnings := append(cloneWarnings(output.Warnings), report.Warnings()...) payload := map[string]any{ - "plan": debugChunkPlanEnvelope(plan), "materialized_chunks": debugSourceChunkEnvelopes(chunks), - "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected), + "plan": debugChunkPlanEnvelope(candidate.plan), "materialized_chunks": debugSourceChunkEnvelopes(candidate.chunks), + "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(chunkRejection(report, candidate.terminal.envelope.Attempt, chunker.Key())), } if validationErr != nil { - result.setValidation(validationWarnings, rejected, validationErr) - return retryAttemptResult{}, terminal.record(payload, validationErr) + return report, candidate.terminal.record(payload, validationErr) } - if rejected != nil { - result.setValidation(validationWarnings, rejected, nil) - if debugErr := terminal.record(payload, nil); debugErr != nil { - return retryAttemptResult{}, debugErr + if report.FirstRejection() == nil && input.pipeline.ChunkValidationPolicy.ValidatorFailure == ValidatorFailureFailRun { + if failure := firstIncompleteValidation(report); failure != nil { + return report, candidate.terminal.record(payload, validatorFailureError(*failure)) } - return retryAttemptResult{rejection: rejected, warnings: attemptWarnings}, nil } - result.chunks = chunks - result.plan = &plan - result.warnings = attemptWarnings - producerWarnings = cloneWarnings(chunkResult.Warnings) - result.setValidation(validationWarnings, nil, nil) - if debugErr := terminal.record(payload, nil); debugErr != nil { - return retryAttemptResult{}, debugErr - } - return retryAttemptResult{accepted: true}, nil + return report, candidate.terminal.record(payload, nil) }) if err != nil { if result.summary.ValidationStatus == "not_run" { @@ -172,19 +183,46 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S } return result, err } - result.accepted = retryResult.accepted - result.rejection = retryResult.rejection - if !retryResult.accepted { - result.warnings = cloneWarnings(retryResult.warnings) + if terminal.Action == producerTerminalRejected { + result.rejection = terminal.Rejection + if result.rejection != nil { + result.rejection.Stage = string(StageChunk) + result.rejection.ModuleKey = chunker.Key() + } + result.warnings = cloneWarnings(terminal.Warnings) + result.setValidation(terminal.Validation.Warnings(), result.rejection, nil) return result, nil } + candidate, ok := terminal.Value.(generatedChunkPlanCandidate) + if !ok { + return result, fmt.Errorf("chunk attempt terminal has incompatible value") + } + if err := result.setCandidate(candidate.record, "generated"); err != nil { + return result, fmt.Errorf("clone generated chunk plan record: %w", err) + } + if mode == ChunkCacheRefresh { + result.action = "refreshed" + result.summary.Action = "refreshed" + } + if mode == ChunkCacheBypass { + result.action = "bypassed" + result.summary.Action = "bypassed" + } + result.accepted = true + result.plan = &candidate.plan + result.chunks = candidate.chunks + result.warnings = cloneWarnings(terminal.Warnings) + result.setValidation(terminal.Validation.Warnings(), nil, nil) + if terminal.ValidationIncomplete { + result.summary.ValidationStatus = "incomplete" + } - if mode == ChunkCacheAuto || mode == ChunkCacheRefresh { + if (mode == ChunkCacheAuto || mode == ChunkCacheRefresh) && !terminal.ValidationIncomplete { record, cloneErr := cloneChunkPlanRecord(*result.record) if cloneErr != nil { return result, fmt.Errorf("clone chunk plan record for publication: %w", cloneErr) } - record.Warnings = cloneWarnings(producerWarnings) + record.Warnings = cloneWarnings(candidate.producerWarnings) if err := input.ChunkPlans.Save(record); err != nil { return result, fmt.Errorf("save chunk plan: %w", err) } @@ -193,6 +231,14 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S return result, nil } +func chunkRejection(report validationReport, attempt int, moduleKey string) *contracts.RejectedOutput { + rejection := report.FirstRejection() + if rejection == nil { + return nil + } + return &contracts.RejectedOutput{Stage: string(StageChunk), ModuleKey: moduleKey, ValidatorName: rejection.validatorName, ReasonCode: rejection.reasonCode, Message: rejection.message, AttemptCount: attempt, DiagnosticArtifactPath: rejection.diagnosticPath} +} + func chunkPlanLookupStatus(status ChunkPlanStatus) string { switch status { case ChunkPlanHit: diff --git a/internal/framework/pipeline/runner_chunk_plan_test.go b/internal/framework/pipeline/runner_chunk_plan_test.go index 008cc0d4..07666376 100644 --- a/internal/framework/pipeline/runner_chunk_plan_test.go +++ b/internal/framework/pipeline/runner_chunk_plan_test.go @@ -84,6 +84,35 @@ type retryingChunker struct { calls int } +type correctionAwareChunker struct { + key string + plan source.ChunkPlan + calls int + correction *contracts.SemanticCorrection +} + +func (c *correctionAwareChunker) Key() string { return c.key } +func (*correctionAwareChunker) ReferenceSlots() []contracts.ReferenceSlot { return nil } +func (c *correctionAwareChunker) Plan(_ context.Context, request contracts.ChunkRequest) (contracts.ChunkPlanResult, error) { + c.calls++ + if request.Correction != nil { + correction, err := contracts.CloneSemanticCorrection(request.Correction) + if err != nil { + return contracts.ChunkPlanResult{}, err + } + c.correction = correction + } + response := "initial chunk response" + if c.calls > 1 { + response = "corrected chunk response" + } + candidate, err := contracts.NewModelCandidate([]byte(response), contracts.CorrectionProtocolSingleResponseV1) + if err != nil { + return contracts.ChunkPlanResult{}, err + } + return contracts.ChunkPlanResult{Plan: source.CloneChunkPlan(c.plan), ModelCandidate: candidate}, nil +} + 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) { @@ -275,6 +304,53 @@ func TestRunnerOmitsChunkMapForRejectedPlan(t *testing.T) { } } +func TestRunnerCorrectsRejectedGeneratedChunkPlan(t *testing.T) { + prepared, plan := preparedTerminalDebugPipeline(t) + prepared.resolved.ChunkExecutionClass = contracts.ExecutionClassLLMBacked + prepared.resolved.Chunk.Retries = 1 + chunker := &correctionAwareChunker{key: prepared.resolved.Chunk.Module, plan: plan} + prepared.chunker = chunker + validatorCalls := 0 + prepared.chunkValidators.validators = []preparedValidator{{ + resolved: ResolvedValidator{Binding: Binding("reject-then-approve"), Target: ValidatorTargetChunk}, + chunk: chunkValidationFunc{name: "reject-then-approve", validate: func(context.Context, contracts.ChunkValidationRequest) (contracts.ValidationResult, error) { + validatorCalls++ + if validatorCalls == 1 { + return contracts.ValidationResult{ReasonCode: "chunk_scope", Message: "scope needs correction", CorrectionGuidance: "correct the chunk scope"}, nil + } + return contracts.ValidationResult{Approved: true}, nil + }}, + }} + output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input")}) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + if len(output.Rejected) != 0 || chunker.calls != 2 || validatorCalls != 2 { + t.Fatalf("output = %#v chunker calls = %d validator calls = %d", output.Rejected, chunker.calls, validatorCalls) + } + if got := string(chunker.correction.AssistantResponse); got != "initial chunk response" { + t.Fatalf("correction response = %q", got) + } +} + +func TestRunnerDoesNotPublishValidationIncompleteChunkPlan(t *testing.T) { + prepared, plan := preparedTerminalDebugPipeline(t) + prepared.resolved.ChunkValidationPolicy.ValidatorFailure = ValidatorFailureWarnContinue + prepared.chunker = terminalChunker{key: prepared.resolved.Chunk.Module, plan: plan} + prepared.chunkValidators.validators = []preparedValidator{{ + resolved: ResolvedValidator{Binding: Binding("unavailable"), Target: ValidatorTargetChunk}, + chunk: &countingChunkValidator{err: errors.New("validator unavailable")}, + }} + 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.Fatalf("Run() error = %v", err) + } + if output.ChunkPlan == nil || output.ChunkPlan.ValidationStatus != "incomplete" || output.ChunkPlan.PublicationStatus != "not_published" || store.saves != 0 { + t.Fatalf("chunk summary = %#v saves = %d", output.ChunkPlan, store.saves) + } +} + func TestRunnerRetainsChunkMapAfterLaneRejection(t *testing.T) { prepared, _ := preparedTerminalDebugPipeline(t) encoder := &capturingChunkMapOutput{} @@ -322,8 +398,8 @@ func TestRunnerChunkPlanManifestRetainsCandidateOnRejection(t *testing.T) { if err != nil { t.Fatal(err) } - if output.Manifest.ChunkPlan.Action != "refreshed" || output.Manifest.ChunkPlan.PlanDigest == "" || output.ChunkPlan.ValidationStatus != "rejected" || output.ChunkPlan.PublicationStatus != "not_published" { - t.Fatalf("manifest = %#v summary = %#v", output.Manifest.ChunkPlan, output.ChunkPlan) + if output.Manifest.ChunkPlan.Action != "" || output.Manifest.ChunkPlan.PlanDigest != "" || output.ChunkPlan.ValidationStatus != "rejected" || output.ChunkPlan.PublicationStatus != "not_published" { + t.Fatalf("manifest = %#v summary = %#v; rejected output must not retain a usable chunk candidate", output.Manifest.ChunkPlan, output.ChunkPlan) } } @@ -455,20 +531,32 @@ func TestRunnerAutoHitValidatesOnceWithoutRegenerationOrMutation(t *testing.T) { } else if err != nil { t.Fatalf("Run() error = %v", err) } - if validator.calls != 1 || calls != 0 || llmCalls != 0 || store.loads != 1 || store.saves != 0 { + wantProducerCalls := 0 + if tc.wantReject { + wantProducerCalls = 1 + } + if validator.calls != 1+wantProducerCalls || calls != wantProducerCalls || llmCalls != wantProducerCalls || 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.wantReject { + assertAttemptEnvelopeSequence(t, debug, "chunk", 1) + } else { + 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":`) { + if tc.wantReject { + if !strings.Contains(encoded, `"status":"invalid"`) || strings.Contains(encoded, `"plan":`) { + t.Fatalf("rejected cache debug = %s", encoded) + } + } else 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) { + if tc.wantError == "" && !tc.wantReject && len(output.Warnings) != 1+len(tc.result.Warnings) { t.Fatalf("warnings = %#v, want stored warning once plus current warnings", output.Warnings) } }) @@ -485,7 +573,7 @@ func TestRunnerPublishesOnlyAcceptedGeneratedPlans(t *testing.T) { 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: "validator rejection", validator: contracts.ValidationResult{Approved: false, ReasonCode: "rejected", Message: "rejected generated plan"}, wantCalls: 1, wantReject: true}, {name: "cancellation", cancel: true}, } { t.Run(tc.name, func(t *testing.T) { diff --git a/internal/framework/pipeline/runner_rejection_warnings_test.go b/internal/framework/pipeline/runner_rejection_warnings_test.go index becebe08..abf503ca 100644 --- a/internal/framework/pipeline/runner_rejection_warnings_test.go +++ b/internal/framework/pipeline/runner_rejection_warnings_test.go @@ -57,6 +57,7 @@ func TestRunnerPromotesOnlyTerminalRejectionWarnings(t *testing.T) { switch target { case StageChunk: + prepared.resolved.ChunkValidationPolicy.SemanticRejection = SemanticRejectionRejectOutput chunker := prepared.chunker.(*typedTestChunker) prepared.chunker = &warningChunker{key: prepared.resolved.Chunk.Module, plan: source.CloneChunkPlan(chunker.plan)} prepared.resolved.Chunk.Retries = 1 @@ -100,11 +101,16 @@ func TestRunnerPromotesOnlyTerminalRejectionWarnings(t *testing.T) { t.Fatalf("Run() error = %v, want nil", err) } wantScopes := []string{"operation-2", "validator-2"} + wantAttempts := 2 + if target == StageChunk { + wantScopes = []string{"operation-1", "validator-1"} + wantAttempts = 1 + } 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 len(output.Rejected) != 1 || output.Rejected[0].AttemptCount != wantAttempts { + t.Fatalf("rejections = %#v, want terminal rejection after %d attempt(s)", output.Rejected, wantAttempts) } if target == StageExtract && !reflect.DeepEqual(rejectionWarningScopes(recorder.checkpoint.Warnings), wantScopes) { t.Fatalf("extract checkpoint warnings = %#v, want %#v", recorder.checkpoint.Warnings, wantScopes) diff --git a/internal/framework/pipeline/runner_terminal_debug_test.go b/internal/framework/pipeline/runner_terminal_debug_test.go index 0d9c560f..e76e343d 100644 --- a/internal/framework/pipeline/runner_terminal_debug_test.go +++ b/internal/framework/pipeline/runner_terminal_debug_test.go @@ -123,6 +123,11 @@ func assertAttemptEnvelopeSequence(t *testing.T, debug *capturedDebugRecorder, p func preparedTerminalDebugPipeline(t *testing.T) (*PreparedPipeline, source.ChunkPlan) { t.Helper() prepared := preparedAttemptDebugPipeline(t) + prepared.resolved.ChunkValidationPolicy = ValidationPolicy{ + ProducerStructuralFailure: ProducerStructuralFailureFailRun, + SemanticRejection: SemanticRejectionRejectOutput, + ValidatorFailure: ValidatorFailureFailRun, + } chunker, ok := prepared.chunker.(*typedTestChunker) if !ok { t.Fatalf("prepared chunker = %T, want *typedTestChunker", prepared.chunker)