From b98604e5ca3bc4b344b18fc5142a4311f97a991e Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Thu, 24 Sep 2026 17:23:59 +0000 Subject: [PATCH] Integrate partial processing outcomes and fatal cancellation --- docs/roadmap/implementation.md | 10 + .../framework/runner/cancellation_test.go | 22 +- internal/framework/runner/runner.go | 256 ++++++++++++------ internal/framework/runner/runner_test.go | 93 ++++++- 4 files changed, 294 insertions(+), 87 deletions(-) diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 8b4adf8..a288888 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -552,6 +552,16 @@ not rely on report tables to maintain chunk state. ## Stage 6 — Integrate Partial Outcomes and Fatal Cancellation +**Completed 2026-09-24.** The runner now preserves per-section operation facts, +continues after recoverable model failures, applies only candidates approved by +every required validator, and retains completed observations on fatal cancellation. +It tracks explicit work-unit states and merges defective-proposal filtering with +an existing proposal recovery. Runner tests cover total proposal failure, mixed +validation outcomes, later modules, multiple failures in one unit, merged recovery, +and fatal cancellation. `go test -race ./internal/framework/runner +./internal/framework/llm` and `git diff --check` passed. Integrated reporting +checks remain deferred until the reporting stages. + **Work:** Update `collectSectionProposals`, `runModulePipeline`, and `validateSectionCandidates` to consume the outcomes from Stages 3–4. Recoverable failures never enter shared fatal cancellation. Preserve candidates through every diff --git a/internal/framework/runner/cancellation_test.go b/internal/framework/runner/cancellation_test.go index ddf01c2..db20cc3 100644 --- a/internal/framework/runner/cancellation_test.go +++ b/internal/framework/runner/cancellation_test.go @@ -114,15 +114,25 @@ func TestValidatorFailureCancelsActiveAndPendingProviderWork(t *testing.T) { for i := 0; i < 3; i++ { sections = append(sections, chunking.Section{Index: i, Segments: []schema.Segment{{ID: 1, Text: "hello"}}}) } - done := make(chan error, 1) + type pipelineOutcome struct { + result modulePipelineResult + err error + } + done := make(chan pipelineOutcome, 1) go func() { - _, err := runModulePipeline(ctx, collectSectionProposalsInput{Module: module, Spec: contracts.ModuleRunSpec{ModuleKey: "grammar", InstanceName: "grammar"}, Policy: module.policy, Config: &loaded.Config, Sections: sections}) - done <- err + result, err := runModulePipeline(ctx, collectSectionProposalsInput{Module: module, Spec: contracts.ModuleRunSpec{ModuleKey: "grammar", InstanceName: "grammar"}, Policy: module.policy, Config: &loaded.Config, Sections: sections}) + done <- pipelineOutcome{result: result, err: err} }() select { - case err := <-done: - if !errors.Is(err, origin) { - t.Fatalf("lost originating failure: %v", err) + case outcome := <-done: + if !errors.Is(outcome.err, origin) { + t.Fatalf("lost originating failure: %v", outcome.err) + } + if len(outcome.result.WorkUnits) != 3 { + t.Fatalf("lost work inventory: %+v", outcome.result.WorkUnits) + } + if len(outcome.result.Approved) != 0 { + t.Fatalf("fatal validation returned unchecked candidates: %+v", outcome.result.Approved) } case <-time.After(2 * time.Second): t.Fatal("validator failure did not cancel and join peers") diff --git a/internal/framework/runner/runner.go b/internal/framework/runner/runner.go index 76b00bf..2f8c7e2 100644 --- a/internal/framework/runner/runner.go +++ b/internal/framework/runner/runner.go @@ -16,7 +16,6 @@ import ( "gitea.maximumdirect.net/eric/audita/internal/framework/outcomes" "gitea.maximumdirect.net/eric/audita/internal/framework/proposals" "gitea.maximumdirect.net/eric/audita/internal/framework/validators" - stagewarnings "gitea.maximumdirect.net/eric/audita/internal/framework/warnings" validatormetadata "gitea.maximumdirect.net/eric/audita/internal/validators/metadata" ) @@ -193,7 +192,7 @@ func (r *Runner) run(ctx context.Context, input RunInput) (RunOutput, error) { ReplacementPolicy: policy, Status: ModuleStatusFailed, ProposalCount: pipelineResult.ProposalCount, - Warnings: pipelineResult.Warnings, + Operations: pipelineResult.Operations, ValidatorDecisions: pipelineResult.ValidatorDecisions, ValidatorRejected: pipelineResult.ValidatorRejected, ErrorMessage: pipelineErr.Error(), @@ -201,6 +200,10 @@ func (r *Runner) run(ctx context.Context, input RunInput) (RunOutput, error) { CompletedAt: time.Now().UTC(), } results = append(results, failed) + copy(workUnits[moduleIndex*len(sections):(moduleIndex+1)*len(sections)], pipelineResult.WorkUnits) + for sectionPos := range sections { + workUnits[moduleIndex*len(sections)+sectionPos].ModuleIndex = moduleIndex + } return snapshot(), fmt.Errorf("module %q failed: %w", spec.InstanceName, pipelineErr) } @@ -208,12 +211,19 @@ func (r *Runner) run(ctx context.Context, input RunInput) (RunOutput, error) { if applyErr != nil { return snapshot(), fmt.Errorf("module %q application failed: %w", spec.InstanceName, applyErr) } - sections = updated + copy(workUnits[moduleIndex*len(sections):(moduleIndex+1)*len(sections)], pipelineResult.WorkUnits) for sectionPos := range sections { unit := &workUnits[moduleIndex*len(sections)+sectionPos] + unit.ModuleIndex = moduleIndex unit.State = outcomes.WorkCompleted unit.CorrectionsApplied = appliedBySection[sectionPos] } + attributeApplyFiltering(sections, pipelineResult.Approved, skipped, policy, workUnits[moduleIndex*len(sections):(moduleIndex+1)*len(sections)]) + sections = updated + moduleOperations := make([]outcomes.Operation, 0) + for _, unit := range workUnits[moduleIndex*len(sections) : (moduleIndex+1)*len(sections)] { + moduleOperations = append(moduleOperations, unit.Operations...) + } results = append(results, ModuleResult{ ModuleKey: spec.ModuleKey, @@ -221,7 +231,7 @@ func (r *Runner) run(ctx context.Context, input RunInput) (RunOutput, error) { ReplacementPolicy: policy, Status: ModuleStatusSuccess, ProposalCount: pipelineResult.ProposalCount, - Warnings: pipelineResult.Warnings, + Operations: moduleOperations, ValidatorDecisions: pipelineResult.ValidatorDecisions, ValidatorRejected: pipelineResult.ValidatorRejected, AppliedChanges: applied, @@ -289,17 +299,12 @@ type collectSectionProposalsInput struct { Timings *validatorTimingCollector } -type sectionProposals struct { - meta contracts.SectionMetadata - corrected []proposals.CorrectionProposal - warnings []stagewarnings.StageWarning -} - type sectionProposalResult struct { sectionPos int section chunking.Section corrected []proposals.CorrectionProposal - warnings []stagewarnings.StageWarning + operations []outcomes.Operation + started bool err error } @@ -309,14 +314,16 @@ type sectionValidationResult struct { approved []proposals.EnrichedCorrectionProposal decisions []ValidatorDecisionRecord rejected []ValidatorRejectedChange - warnings []stagewarnings.StageWarning + operations []outcomes.Operation + filtered []outcomes.FilteredProposal err error } type modulePipelineResult struct { ProposalCount int Approved []proposals.EnrichedCorrectionProposal - Warnings []stagewarnings.StageWarning + Operations []outcomes.Operation + WorkUnits []WorkUnitOutcome ValidatorDecisions []ValidatorDecisionRecord ValidatorRejected []ValidatorRejectedChange } @@ -334,6 +341,10 @@ func collectSectionProposals(ctx context.Context, input collectSectionProposalsI wg.Add(1) go func() { defer wg.Done() + if runCtx.Err() != nil { + results <- sectionProposalResult{sectionPos: sectionPos, section: section} + return + } meta := contracts.SectionMetadataFromSection(section) proposalResult, err := input.Module.Propose(runCtx, contracts.ProposalRequest{ ExecutionContext: contracts.ExecutionContext{ @@ -350,16 +361,13 @@ func collectSectionProposals(ctx context.Context, input collectSectionProposalsI }, LLMClient: input.ProposalClient, }) - select { - case results <- sectionProposalResult{ + results <- sectionProposalResult{ sectionPos: sectionPos, section: section, corrected: proposalResult.Proposals, - warnings: proposalResult.Warnings, + operations: proposalResult.Operations, + started: true, err: err, - }: - case <-runCtx.Done(): - return } }() } @@ -372,10 +380,17 @@ func collectSectionProposals(ctx context.Context, input collectSectionProposalsI func runModulePipeline(ctx context.Context, input collectSectionProposalsInput) (modulePipelineResult, error) { out := modulePipelineResult{ Approved: make([]proposals.EnrichedCorrectionProposal, 0), - Warnings: make([]stagewarnings.StageWarning, 0), + Operations: make([]outcomes.Operation, 0), + WorkUnits: make([]WorkUnitOutcome, len(input.Sections)), ValidatorDecisions: make([]ValidatorDecisionRecord, 0), ValidatorRejected: make([]ValidatorRejectedChange, 0), } + for i, section := range input.Sections { + out.WorkUnits[i] = WorkUnitOutcome{ + ModuleKey: input.Spec.ModuleKey, ModuleInstance: input.Spec.InstanceName, + Section: contracts.SectionMetadataFromSection(section), State: outcomes.WorkNotStarted, + } + } if len(input.Sections) == 0 { return out, nil @@ -391,7 +406,6 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput) nextSectionToProcess := 0 nextProposalIndex := 0 - validationLaunches := 0 var ( firstErr error @@ -417,6 +431,11 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput) } for result := range proposalResults { + unit := &out.WorkUnits[result.sectionPos] + if result.started { + unit.State = outcomes.WorkInterrupted + } + unit.Operations = append(unit.Operations, result.operations...) if result.err != nil { setErr(result.err) continue @@ -424,7 +443,9 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput) if getErr() != nil { continue } - out.Warnings = append(out.Warnings, result.warnings...) + if !result.started { + continue + } pending[result.sectionPos] = result for { @@ -452,8 +473,15 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput) } nextProposalIndex += len(sectionEnriched) out.ProposalCount += len(sectionEnriched) + for operationIndex := range out.WorkUnits[nextSectionToProcess].Operations { + operation := &out.WorkUnits[nextSectionToProcess].Operations[operationIndex] + if operation.Reference.Kind == outcomes.OperationProposal && operation.Succeeded { + for _, proposal := range sectionEnriched { + operation.ProposalIndices = append(operation.ProposalIndices, proposal.ProposalIndex) + } + } + } - validationLaunches++ vwg.Add(1) go func(sectionPos int, enriched []proposals.EnrichedCorrectionProposal, sectionMetadata contracts.SectionMetadata) { defer vwg.Done() @@ -477,7 +505,8 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput) approved: validated.approved, decisions: validated.decisions, rejected: validated.rejected, - warnings: validated.warnings, + operations: validated.operations, + filtered: validated.filtered, err: err, } }(nextSectionToProcess, sectionEnriched, sectionMeta) @@ -495,16 +524,26 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput) } } - for i := 0; i < validationLaunches; i++ { + for i := range input.Sections { res, ok := validationBySection[i] if !ok { - break + continue + } + unit := &out.WorkUnits[i] + unit.Operations = append(unit.Operations, res.operations...) + mergeFilteredProposals(unit, res.filtered) + if res.err == nil { + unit.State = outcomes.WorkCompleted + } else { + unit.State = outcomes.WorkInterrupted } out.Approved = append(out.Approved, res.approved...) - out.Warnings = append(out.Warnings, res.warnings...) out.ValidatorDecisions = append(out.ValidatorDecisions, res.decisions...) out.ValidatorRejected = append(out.ValidatorRejected, res.rejected...) } + for _, unit := range out.WorkUnits { + out.Operations = append(out.Operations, unit.Operations...) + } sort.SliceStable(out.ValidatorDecisions, func(i, j int) bool { if out.ValidatorDecisions[i].ProposalIndex != out.ValidatorDecisions[j].ProposalIndex { @@ -518,38 +557,9 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput) } return validatorOrder[out.ValidatorRejected[i].ValidatorName] < validatorOrder[out.ValidatorRejected[j].ValidatorName] }) - sort.SliceStable(out.Warnings, func(i, j int) bool { - leftSection, rightSection := -1, -1 - if out.Warnings[i].SectionIndex != nil { - leftSection = *out.Warnings[i].SectionIndex - } - if out.Warnings[j].SectionIndex != nil { - rightSection = *out.Warnings[j].SectionIndex - } - if leftSection != rightSection { - return leftSection < rightSection - } - leftBatch, rightBatch := -1, -1 - if out.Warnings[i].BatchIndex != nil { - leftBatch = *out.Warnings[i].BatchIndex - } - if out.Warnings[j].BatchIndex != nil { - rightBatch = *out.Warnings[j].BatchIndex - } - if leftBatch != rightBatch { - return leftBatch < rightBatch - } - if out.Warnings[i].ValidatorName != out.Warnings[j].ValidatorName { - return validatorOrder[out.Warnings[i].ValidatorName] < validatorOrder[out.Warnings[j].ValidatorName] - } - if out.Warnings[i].Scope != out.Warnings[j].Scope { - return out.Warnings[i].Scope < out.Warnings[j].Scope - } - if out.Warnings[i].ReasonCode != out.Warnings[j].ReasonCode { - return out.Warnings[i].ReasonCode < out.Warnings[j].ReasonCode - } - return out.Warnings[i].Message < out.Warnings[j].Message - }) + if ctx.Err() != nil { + setErr(ctx.Err()) + } if err := getErr(); err != nil { return out, err @@ -573,19 +583,24 @@ type validateSectionCandidatesInput struct { } type validateSectionCandidatesResult struct { - approved []proposals.EnrichedCorrectionProposal - decisions []ValidatorDecisionRecord - rejected []ValidatorRejectedChange - warnings []stagewarnings.StageWarning + approved []proposals.EnrichedCorrectionProposal + decisions []ValidatorDecisionRecord + rejected []ValidatorRejectedChange + operations []outcomes.Operation + filtered []outcomes.FilteredProposal } func validateSectionCandidates(ctx context.Context, input validateSectionCandidatesInput) (validateSectionCandidatesResult, error) { decisions := make([]ValidatorDecisionRecord, 0) rejected := make([]ValidatorRejectedChange, 0) - warnings := make([]stagewarnings.StageWarning, 0) + operations := make([]outcomes.Operation, 0) + filtered := make([]outcomes.FilteredProposal, 0) eligible := append([]proposals.EnrichedCorrectionProposal(nil), input.SectionEnriched...) - for _, validator := range input.Validators { + for validatorOrder, validator := range input.Validators { + if err := ctx.Err(); err != nil { + return validateSectionCandidatesResult{decisions: decisions, rejected: rejected, operations: operations, filtered: filtered}, err + } var diagnosticsWriter validators.InteractionDiagnosticsWriter if input.DiagnosticsDir != "" { diagnosticsWriter = &llmDiagnosticsWriterAdapter{ @@ -603,6 +618,7 @@ func validateSectionCandidates(ctx context.Context, input validateSectionCandida ModuleKey: input.Spec.ModuleKey, ModuleInstance: input.Spec.InstanceName, SectionIndex: input.SectionIndex, + ValidatorOrder: &validatorOrder, ReplacementPolicy: input.Policy, Glossary: input.Glossary, Config: input.Config, @@ -612,24 +628,24 @@ func validateSectionCandidates(ctx context.Context, input validateSectionCandida if input.Timings != nil { input.Timings.record(input.Spec.ModuleKey, input.Spec.InstanceName, validator.Name(), validatormetadata.ClassOf(validator) == validatormetadata.ExecutionClassLLMBacked, time.Since(validatorStarted)) } + operations = append(operations, vResult.Operations...) if err != nil { return validateSectionCandidatesResult{ - approved: eligible, - decisions: decisions, - rejected: rejected, - warnings: warnings, + decisions: decisions, + rejected: rejected, + operations: operations, + filtered: filtered, }, fmt.Errorf("validator %q failed: %w", validator.Name(), err) } if err := validators.EnforceDecisionCardinality(eligible, vResult.Decisions); err != nil { return validateSectionCandidatesResult{ - approved: eligible, - decisions: decisions, - rejected: rejected, - warnings: warnings, + decisions: decisions, + rejected: rejected, + operations: operations, + filtered: filtered, }, fmt.Errorf("validator %q cardinality failed: %w", validator.Name(), err) } - warnings = append(warnings, vResult.Warnings...) nextEligible := make([]proposals.EnrichedCorrectionProposal, 0, len(eligible)) byIndex := make(map[int]proposals.EnrichedCorrectionProposal, len(eligible)) @@ -650,6 +666,11 @@ func validateSectionCandidates(ctx context.Context, input validateSectionCandida nextEligible = append(nextEligible, byIndex[d.ProposalIndex]) continue } + if d.Disposition != validators.DispositionUnavailable { + if reason, ok := validators.DefectiveProposalReason(d.ReasonCode); ok { + filtered = append(filtered, outcomes.FilteredProposal{ProposalIndex: d.ProposalIndex, Reason: reason, Count: 1}) + } + } p := byIndex[d.ProposalIndex] rejected = append(rejected, ValidatorRejectedChange{ ValidatorName: validator.Name(), @@ -668,13 +689,92 @@ func validateSectionCandidates(ctx context.Context, input validateSectionCandida } return validateSectionCandidatesResult{ - approved: eligible, - decisions: decisions, - rejected: rejected, - warnings: warnings, + approved: eligible, + decisions: decisions, + rejected: rejected, + operations: operations, + filtered: filtered, }, nil } +// mergeFilteredProposals attributes defective candidates to their originating +// proposal response. Each response has at most one filtering recovery reason. +func mergeFilteredProposals(unit *WorkUnitOutcome, filtered []outcomes.FilteredProposal) { + if unit == nil || len(filtered) == 0 { + return + } + for i := range unit.Operations { + operation := &unit.Operations[i] + if operation.Reference.Kind != outcomes.OperationProposal || !operation.Succeeded { + continue + } + seen := make(map[int]struct{}, len(operation.Filtered)) + for _, existing := range operation.Filtered { + seen[existing.ProposalIndex] = struct{}{} + } + for _, item := range filtered { + if _, ok := seen[item.ProposalIndex]; ok { + continue + } + seen[item.ProposalIndex] = struct{}{} + operation.Filtered = append(operation.Filtered, item) + } + if len(operation.Filtered) == 0 { + continue + } + sort.Slice(operation.Filtered, func(a, b int) bool { + return operation.Filtered[a].ProposalIndex < operation.Filtered[b].ProposalIndex + }) + found := false + for _, reason := range operation.RecoveryReasons { + if reason == outcomes.RecoveryProposalsFiltered { + found = true + break + } + } + if !found { + operation.RecoveryReasons = append(operation.RecoveryReasons, outcomes.RecoveryProposalsFiltered) + } + return + } +} + +func attributeApplyFiltering(sections []chunking.Section, approved []proposals.EnrichedCorrectionProposal, skipped []proposals.SkippedChange, policy proposals.ReplacementPolicy, units []WorkUnitOutcome) { + byIndex := make(map[int]proposals.EnrichedCorrectionProposal, len(approved)) + sectionPosition := make(map[int]int, len(sections)) + for pos, section := range sections { + sectionPosition[section.Index] = pos + } + for _, proposal := range approved { + byIndex[proposal.ProposalIndex] = proposal + } + for _, skip := range skipped { + proposal, ok := byIndex[skip.ProposalIndex] + if !ok || proposal.SectionIndex == nil { + continue + } + pos, ok := sectionPosition[*proposal.SectionIndex] + if !ok { + continue + } + var segment *schema.Segment + for i := range sections[pos].Segments { + if sections[pos].Segments[i].ID == proposal.TargetSegmentID { + segment = §ions[pos].Segments[i] + break + } + } + // A proposal applicable to the input was made stale by a previously + // accepted correction. That conflict is not a response defect. + if proposals.PreviewProposalForSegment(segment, proposal.CorrectionProposal, policy).Applicable { + continue + } + if reason, ok := validators.DefectiveProposalReason(string(skip.SkipReason)); ok { + mergeFilteredProposals(&units[pos], []outcomes.FilteredProposal{{ProposalIndex: skip.ProposalIndex, Reason: reason, Count: 1}}) + } + } +} + func reorderValidatorsForPipeline(in []contracts.Validator) ([]contracts.Validator, map[string]int) { deterministic := make([]contracts.Validator, 0, len(in)) llmBacked := make([]contracts.Validator, 0, len(in)) diff --git a/internal/framework/runner/runner_test.go b/internal/framework/runner/runner_test.go index 1d0ce2a..72e97c1 100644 --- a/internal/framework/runner/runner_test.go +++ b/internal/framework/runner/runner_test.go @@ -16,6 +16,7 @@ import ( "gitea.maximumdirect.net/eric/audita/internal/core/schema" "gitea.maximumdirect.net/eric/audita/internal/framework/contracts" "gitea.maximumdirect.net/eric/audita/internal/framework/modules" + "gitea.maximumdirect.net/eric/audita/internal/framework/outcomes" "gitea.maximumdirect.net/eric/audita/internal/framework/proposal_generation" "gitea.maximumdirect.net/eric/audita/internal/framework/proposals" "gitea.maximumdirect.net/eric/audita/internal/framework/validators" @@ -45,6 +46,92 @@ type fakeModule struct { proposeF func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) } +type outcomeModule struct { + fakeModule + proposeResult func(contracts.ProposalRequest) (contracts.ProposalResult, error) +} + +func TestRunnerRecoverableProposalFailureAllowsLaterModule(t *testing.T) { + failed := outcomeModule{fakeModule: fakeModule{key: "first", policy: proposals.ReplacementPolicyRequireUnique}, proposeResult: func(req contracts.ProposalRequest) (contracts.ProposalResult, error) { + return contracts.ProposalResult{Operations: []outcomes.Operation{{Reference: outcomes.Reference{ModuleKey: "first", ModuleInstance: "first", Kind: outcomes.OperationProposal}, FailureReason: outcomes.ReasonTransportFailed}}}, nil + }} + second := fakeModule{key: "second", policy: proposals.ReplacementPolicyRequireUnique, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { + return []proposals.CorrectionProposal{{TargetSegmentID: 1, OriginalText: "teh", CorrectedText: "the", Confidence: 1}}, nil + }} + r := New(fakeFactory{modules: map[string]contracts.TranscriptModule{"first": failed, "second": second}}) + out, err := r.Run(context.Background(), RunInput{Transcript: &schema.Transcript{Segments: []schema.Segment{{ID: 1, Text: "teh"}}}, ModuleSpecs: []contracts.ModuleRunSpec{{ModuleKey: "first", InstanceName: "first"}, {ModuleKey: "second", InstanceName: "second"}}}) + if err != nil { + t.Fatal(err) + } + if out.FinalTranscript.Segments[0].Text != "the" || len(out.WorkUnits) != 2 || out.WorkUnits[0].State != outcomes.WorkCompleted || out.WorkUnits[1].CorrectionsApplied != 1 { + t.Fatalf("recoverable failure lost later work: %+v", out) + } + if len(out.WorkUnits[0].Operations) != 1 || out.WorkUnits[0].Operations[0].FailureReason != outcomes.ReasonTransportFailed { + t.Fatalf("missing proposal failure: %+v", out.WorkUnits[0]) + } +} + +func TestRunnerOnlyFullyValidatedCandidatesApplyAndRecoveriesMerge(t *testing.T) { + secondSaw := make([]int, 0) + partial := fakeValidator{name: "partial", validateF: func(req contracts.ValidationRequest) (validators.Result, error) { + if len(req.CandidateProposal) != 2 { + t.Fatalf("expected two candidates after shape filtering, got %d", len(req.CandidateProposal)) + } + return validators.Result{ValidatorName: "partial", Decisions: []validators.Decision{ + {ProposalIndex: req.CandidateProposal[0].ProposalIndex, Approved: false, Disposition: validators.DispositionUnavailable, ReasonCode: string(outcomes.ReasonTransportFailed)}, + {ProposalIndex: req.CandidateProposal[1].ProposalIndex, Approved: true, Disposition: validators.DispositionApproved}, + }, Operations: []outcomes.Operation{ + {Reference: outcomes.Reference{Kind: outcomes.OperationValidation}, FailureReason: outcomes.ReasonTransportFailed}, + {Reference: outcomes.Reference{Kind: outcomes.OperationValidation}, FailureReason: outcomes.ReasonProviderFailed}, + }}, nil + }} + last := fakeValidator{name: "last", validateF: func(req contracts.ValidationRequest) (validators.Result, error) { + for _, candidate := range req.CandidateProposal { + secondSaw = append(secondSaw, candidate.ProposalIndex) + } + return validators.Result{ValidatorName: "last", Decisions: []validators.Decision{{ProposalIndex: req.CandidateProposal[0].ProposalIndex, Approved: true, Disposition: validators.DispositionApproved}}}, nil + }} + module := outcomeModule{fakeModule: fakeModule{key: "m", policy: proposals.ReplacementPolicyRequireUnique, validators: []contracts.Validator{validators.ProposalShapeValidator{}, partial, last}}, proposeResult: func(req contracts.ProposalRequest) (contracts.ProposalResult, error) { + return contracts.ProposalResult{Proposals: []proposals.CorrectionProposal{ + {TargetSegmentID: 0, OriginalText: "bad", CorrectedText: "x", Confidence: 1}, + {TargetSegmentID: 1, OriginalText: "teh", CorrectedText: "the", Confidence: 1}, + {TargetSegmentID: 1, OriginalText: "cat", CorrectedText: "dog", Confidence: 1}, + }, Operations: []outcomes.Operation{{Reference: outcomes.Reference{Kind: outcomes.OperationProposal}, Succeeded: true, RecoveryReasons: []outcomes.RecoveryReason{outcomes.RecoveryOutputRepaired}}}}, nil + }} + r := New(fakeFactory{modules: map[string]contracts.TranscriptModule{"m": module}}) + out, err := r.Run(context.Background(), RunInput{Transcript: &schema.Transcript{Segments: []schema.Segment{{ID: 1, Text: "teh cat"}}}, ModuleSpecs: []contracts.ModuleRunSpec{{ModuleKey: "m", InstanceName: "m"}}}) + if err != nil { + t.Fatal(err) + } + if out.FinalTranscript.Segments[0].Text != "teh dog" || len(secondSaw) != 1 || secondSaw[0] != 2 { + t.Fatalf("unchecked candidate applied or survivor lost: text=%q final=%v", out.FinalTranscript.Segments[0].Text, secondSaw) + } + unit := out.WorkUnits[0] + if unit.State != outcomes.WorkCompleted || unit.CorrectionsApplied != 1 || len(unit.Operations) != 3 { + t.Fatalf("expected one completed unit with two operation failures: %+v", unit) + } + proposal := unit.Operations[0] + if len(proposal.RecoveryReasons) != 2 || proposal.RecoveryReasons[0] != outcomes.RecoveryOutputRepaired || proposal.RecoveryReasons[1] != outcomes.RecoveryProposalsFiltered || len(proposal.Filtered) != 1 || proposal.Filtered[0].ProposalIndex != 0 { + t.Fatalf("repair and filtering must share one recovery: %+v", proposal) + } +} + +func TestFatalValidatorNeverReturnsUncheckedCandidates(t *testing.T) { + origin := errors.New("validator setup failed") + proposal := proposals.EnrichedCorrectionProposal{CorrectionProposal: proposals.CorrectionProposal{TargetSegmentID: 1, OriginalText: "teh", CorrectedText: "the", Confidence: 1}, ProposalMetadata: proposals.ProposalMetadata{ProposalIndex: 7}} + broken := fakeValidator{name: "broken", validateF: func(req contracts.ValidationRequest) (validators.Result, error) { + return validators.Result{Operations: []outcomes.Operation{{Reference: outcomes.Reference{Kind: outcomes.OperationValidation}, FailureReason: outcomes.ReasonSetupFailure}}}, origin + }} + result, err := validateSectionCandidates(context.Background(), validateSectionCandidatesInput{Spec: contracts.ModuleRunSpec{ModuleKey: "m", InstanceName: "m"}, Validators: []contracts.Validator{broken}, SectionEnriched: []proposals.EnrichedCorrectionProposal{proposal}}) + if !errors.Is(err, origin) || len(result.approved) != 0 || len(result.operations) != 1 { + t.Fatalf("fatal result must retain facts without approving candidates: result=%+v err=%v", result, err) + } +} + +func (m outcomeModule) Propose(_ context.Context, req contracts.ProposalRequest) (contracts.ProposalResult, error) { + return m.proposeResult(req) +} + func (m fakeModule) Key() string { return m.key } func (m fakeModule) ReplacementPolicy() proposals.ReplacementPolicy { return m.policy } func (m fakeModule) Validators() []contracts.Validator { return m.validators } @@ -854,8 +941,8 @@ func TestRunnerLLMValidatorMalformedResponseRejectsBatchAndKeepsPartialProgress( if len(out.ModuleResults) != 2 || len(out.ModuleResults[1].ValidatorRejected) != 1 { t.Fatalf("expected second module rejection, got %+v", out.ModuleResults) } - if len(out.ModuleResults[1].Warnings) != 1 || out.ModuleResults[1].Warnings[0].ReasonCode != validators.ReasonValidatorMalformed { - t.Fatalf("expected malformed warning, got %+v", out.ModuleResults[1].Warnings) + if len(out.ModuleResults[1].Operations) != 1 || out.ModuleResults[1].Operations[0].FailureReason != outcomes.ReasonValidationDecisionsInvalid { + t.Fatalf("expected invalid decision-set operation, got %+v", out.ModuleResults[1].Operations) } } @@ -1024,7 +1111,7 @@ func (m proposalGenerationModule) Propose(ctx context.Context, req contracts.Pro if err != nil { return contracts.ProposalResult{}, err } - return contracts.ProposalResult{Proposals: result.Corrections, Warnings: result.Warnings}, nil + return contracts.ProposalResult{Proposals: result.Corrections, Operations: result.Operations}, nil } type fakeProposalStructuredClient struct {