diff --git a/cmd/audita/main_integration_test.go b/cmd/audita/main_integration_test.go index b22c45b..1a70953 100644 --- a/cmd/audita/main_integration_test.go +++ b/cmd/audita/main_integration_test.go @@ -263,10 +263,7 @@ func newProcessFixture(t *testing.T, mode, modules string) processFixture { if len(stageList) != 0 { pipeline = " stages:\n" + strings.Join(stageList, "\n") + "\n" } - configText := fmt.Sprintf("version: 3\nllm:\n proposal_profile: proposal\n validation_profile: validation\n transient_retries: 0\n repair_attempts: %d\npipeline:\n%sdiagnostics:\n work_dir: %s\n", repairs, pipeline, filepath.Join(dir, "runs")) - if mode == "decision-mixed" || mode == "all-validation-failure" || mode == "invalid-decision-after-repair" { - configText = strings.Replace(configText, " transient_retries:", " correction_repair_attempts: 0\n transient_retries:", 1) - } + configText := fmt.Sprintf("version: 3\nllm:\n proposal_profile: proposal\n validation_profile: validation\n transient_retries: 0\n correction_repair_attempts: 0\n repair_attempts: %d\npipeline:\n%sdiagnostics:\n work_dir: %s\n", repairs, pipeline, filepath.Join(dir, "runs")) if strings.HasPrefix(mode, "span-") || strings.HasPrefix(mode, "decision-") { configText += "validation:\n max_prompt_tokens: 4096\nchunking:\n target_sections: 1\n max_section_tokens: 8192\n min_section_tokens: 1\nnormalization:\n max_segment_gap: 0\n" } diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index fedc00b..ff44960 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -19,10 +19,14 @@ formats; [configuration](../config.md) owns settings; the may preserve a checked subset of completed decisions as diagnostic evidence. Recoverable failed batches mark their candidates unavailable; successful survivors continue through all required validators. -6. Only fully approved proposals reach application. Corrections apply in +6. Actionable rejections can enter independent section-local repair branches. + Each revision is mapped to the original section and revalidated from the + start. Withdrawal, exhaustion, and recoverable failures finish the branch + without affecting independent approvals. +7. Only fully approved proposals reach application. Corrections apply in proposal-index order within their owning section. Existing replacement policies and apply-time safety checks resolve stale conflicts. -7. CLI encodes the final transcript and publishes active files through the +8. CLI encodes the final transcript and publishes active files through the output session. `framework/processreport` builds one snapshot for YAML, Markdown, and CLI totals. Debug evidence is captured separately when enabled. diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 13bb11f..c67fce7 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -221,6 +221,8 @@ validator responses remains [deferred](future.md#domain-aware-response-recovery) ## Stage 6 — Run independent repair branches and apply accepted revisions +**Completed.** + 1. Extend the section validation worker in `runModulePipeline` after its initial pass. Keep the approved candidates as they are. For each candidate with an actionable rejection, initialize an independent branch on the diff --git a/internal/framework/contracts/contracts.go b/internal/framework/contracts/contracts.go index 014accd..694d894 100644 --- a/internal/framework/contracts/contracts.go +++ b/internal/framework/contracts/contracts.go @@ -94,9 +94,10 @@ type ProposalRequest struct { } type ProposalResult struct { - Exchange *structuredoutput.Exchange `json:"-"` - Proposals []proposals.CorrectionProposal `json:"proposals,omitempty"` - Operations []outcomes.Operation `json:"operations,omitempty"` + SegmentReferences []int `json:"-"` + Exchange *structuredoutput.Exchange `json:"-"` + Proposals []proposals.CorrectionProposal `json:"proposals,omitempty"` + Operations []outcomes.Operation `json:"operations,omitempty"` } // ValidationRequest is the input to validator execution. diff --git a/internal/framework/proposal_generation/generate.go b/internal/framework/proposal_generation/generate.go index 422afe8..01ed594 100644 --- a/internal/framework/proposal_generation/generate.go +++ b/internal/framework/proposal_generation/generate.go @@ -70,11 +70,12 @@ type Request struct { // Result contains generated candidate proposals and optional diagnostics paths. type Result struct { - Exchange *structuredoutput.Exchange `json:"-"` - Corrections []proposals.CorrectionProposal `json:"corrections"` - Enriched []proposals.EnrichedCorrectionProposal `json:"enriched"` - Operations []outcomes.Operation `json:"operations,omitempty"` - Artifacts InteractionArtifacts `json:"artifacts,omitempty"` + SegmentReferences []int `json:"-"` + Exchange *structuredoutput.Exchange `json:"-"` + Corrections []proposals.CorrectionProposal `json:"corrections"` + Enriched []proposals.EnrichedCorrectionProposal `json:"enriched"` + Operations []outcomes.Operation `json:"operations,omitempty"` + Artifacts InteractionArtifacts `json:"artifacts,omitempty"` } // GenerateCandidates executes one structured LLM call and deterministically maps @@ -186,9 +187,11 @@ func GenerateCandidates(ctx context.Context, req Request) (Result, error) { operation.Succeeded = true operation.RecoveryReasons = outcomes.ExecutionRecoveries(completion.Execution) + references := make([]int, 0, len(response.Corrections)) corrections := make([]proposals.CorrectionProposal, 0, len(response.Corrections)) enriched := make([]proposals.EnrichedCorrectionProposal, 0, len(response.Corrections)) for i, raw := range response.Corrections { + references = append(references, raw.SegmentReference) // Unknown references remain invalid candidates for the existing domain validators. targetID := 0 if req.WorkingTranscript != nil && raw.SegmentReference > 0 && raw.SegmentReference <= len(req.WorkingTranscript.Segments) { @@ -218,11 +221,12 @@ func GenerateCandidates(ctx context.Context, req Request) (Result, error) { } return Result{ - Corrections: corrections, - Exchange: completion.Exchange, - Enriched: enriched, - Operations: []outcomes.Operation{operation}, - Artifacts: artifacts, + Corrections: corrections, + SegmentReferences: references, + Exchange: completion.Exchange, + Enriched: enriched, + Operations: []outcomes.Operation{operation}, + Artifacts: artifacts, }, nil } diff --git a/internal/framework/proposal_generation/module_proposal.go b/internal/framework/proposal_generation/module_proposal.go index 6638ce7..baa89aa 100644 --- a/internal/framework/proposal_generation/module_proposal.go +++ b/internal/framework/proposal_generation/module_proposal.go @@ -73,8 +73,9 @@ func ExecuteModuleProposal(ctx context.Context, req ModuleProposalRequest) (cont } return contracts.ProposalResult{ - Proposals: generated.Corrections, - Exchange: generated.Exchange, - Operations: generated.Operations, + Proposals: generated.Corrections, + SegmentReferences: generated.SegmentReferences, + Exchange: generated.Exchange, + Operations: generated.Operations, }, nil } diff --git a/internal/framework/runner/repair.go b/internal/framework/runner/repair.go new file mode 100644 index 0000000..f267797 --- /dev/null +++ b/internal/framework/runner/repair.go @@ -0,0 +1,179 @@ +package runner + +import ( + "context" + "errors" + "fmt" + + "gitea.maximumdirect.net/eric/audita/v3/internal/framework/outcomes" + "gitea.maximumdirect.net/eric/audita/v3/internal/framework/proposal_generation" + "gitea.maximumdirect.net/eric/audita/v3/internal/framework/proposals" + "gitea.maximumdirect.net/eric/audita/v3/internal/framework/responseschema" + "gitea.maximumdirect.net/eric/audita/v3/internal/framework/structuredoutput" + "gitea.maximumdirect.net/eric/audita/v3/internal/framework/validators" +) + +// RepairAttempt retains compact facts for a single semantic request. +type RepairAttempt struct { + Revision *proposals.EnrichedCorrectionProposal + Decisions []ValidatorDecisionRecord + Rejected []ValidatorRejectedChange + Counts outcomes.ExecutionCounts + FailureReason outcomes.ReasonCode + Succeeded bool + Operations []outcomes.Operation +} + +// RepairOutcome keeps the proposal identity through revision and application. +type RepairOutcome struct { + ProposalIndex int + Attempts []RepairAttempt + Outcome string + Original proposals.EnrichedCorrectionProposal + Final proposals.EnrichedCorrectionProposal +} + +func actionable(index int, rejected []ValidatorRejectedChange) bool { + found := false + for _, decision := range rejected { + if decision.ProposalIndex != index { + continue + } + if decision.ReasonCode == validators.ReasonNoEffect { + return false + } + if decision.Disposition != validators.DispositionUnavailable { + found = true + } + } + return found +} + +// repairCandidates branches each rejected proposal from the same section exchange. +func repairCandidates(ctx context.Context, input validateSectionCandidatesInput, references []int, initial validateSectionCandidatesResult, promptID string) (validateSectionCandidatesResult, []RepairOutcome, error) { + if input.Spec.Settings.LLM.CorrectionRepairAttempts <= 0 { + return initial, nil, nil + } + var branches []RepairOutcome + for position, original := range input.SectionEnriched { + if !actionable(original.ProposalIndex, initial.rejected) { + continue + } + branch := RepairOutcome{ProposalIndex: original.ProposalIndex, Original: original, Final: original, Outcome: "interrupted"} + if err := ctx.Err(); err != nil { + branches = append(branches, branch) + return initial, branches, err + } + if input.Spec.RepairClient == nil || input.Exchange == nil || promptID == "" { + branches = append(branches, branch) + return initial, branches, fmt.Errorf("correction repair setup is unavailable") + } + localRef := 0 + if position < len(references) { + localRef = references[position] + } else if input.WorkingTranscript != nil { + for i, s := range input.WorkingTranscript.Segments { + if s.ID == original.TargetSegmentID { + localRef = i + 1 + break + } + } + } + current := original + findings := repairFindings(original.ProposalIndex, initial.rejected) + replay := &structuredoutput.Replay{Exchange: input.Exchange} + for attempt := 0; attempt < input.Spec.Settings.LLM.CorrectionRepairAttempts; attempt++ { + payload, err := (proposal_generation.RepairRequest{ProposalRef: position + 1, Correction: proposal_generation.StructuredCorrectionProposal{SegmentReference: localRef, OriginalText: current.OriginalText, CorrectedText: current.CorrectedText, Confidence: current.Confidence}, Findings: findings}).JSON() + if err != nil { + branches = append(branches, branch) + return initial, branches, err + } + var response proposal_generation.RepairResponse + completion, callErr := input.Spec.RepairClient.CompleteStructured(ctx, structuredoutput.Request{PromptID: promptID, Variables: map[string]string{"RepairRequestJSON": payload}, Schema: responseschema.CorrectionRepairKey, Replay: replay, Context: structuredoutput.DiagnosticContext{StageIndex: &input.Spec.Index, StageID: input.Spec.ID, Role: "correction_repair", ModuleKey: input.Spec.ModuleKey, ModuleInstance: input.Spec.InstanceName, SectionIndex: input.SectionIndex}}, &response) + record := RepairAttempt{Counts: completion.Execution} + if callErr != nil { + var failure *structuredoutput.Failure + if errors.As(callErr, &failure) { + record.FailureReason = failure.Reason + } else { + record.FailureReason = outcomes.ReasonInternalFailure + } + branch.Attempts = append(branch.Attempts, record) + if structuredoutput.IsRecoverable(callErr) { + branch.Outcome = "unavailable" + break + } + branches = append(branches, branch) + return initial, branches, callErr + } + decoded, err := proposal_generation.DecodeRepairResponse(completion.Content, position+1) + if err != nil { + record.FailureReason = outcomes.ReasonInvalidOutput + branch.Attempts = append(branch.Attempts, record) + branch.Outcome = "unavailable" + break + } + record.Succeeded = true + replay.Turns = append(replay.Turns, completion.RepairInstruction...) + replay.Turns = append(replay.Turns, structuredoutput.Message{Role: "assistant", Content: completion.FinalContent}) + if decoded.Action == "withdraw" { + branch.Attempts = append(branch.Attempts, record) + branch.Outcome = "withdrawn" + break + } + revision := decoded.Correction + localRef = revision.SegmentReference + target := 0 + if input.WorkingTranscript != nil && localRef > 0 && localRef <= len(input.WorkingTranscript.Segments) { + target = input.WorkingTranscript.Segments[localRef-1].ID + } + current = original + current.CorrectionProposal = proposals.CorrectionProposal{TargetSegmentID: target, OriginalText: revision.OriginalText, CorrectedText: revision.CorrectedText, Confidence: revision.Confidence} + branch.Final = current + next := input + next.SectionEnriched = []proposals.EnrichedCorrectionProposal{current} + validated, validationErr := validateSectionCandidates(ctx, next) + revisionSnapshot := current + record.Revision = &revisionSnapshot + record.Decisions = validated.decisions + record.Rejected = validated.rejected + record.Operations = validated.operations + branch.Attempts = append(branch.Attempts, record) + initial.operations = append(initial.operations, validated.operations...) + initial.decisions = append(initial.decisions, validated.decisions...) + if validationErr != nil { + branches = append(branches, branch) + return initial, branches, validationErr + } + if len(validated.approved) == 1 { + branch.Outcome = "approved" + initial.approved = append(initial.approved, current) + retained := initial.filtered[:0] + for _, item := range initial.filtered { + if item.ProposalIndex != original.ProposalIndex { + retained = append(retained, item) + } + } + initial.filtered = retained + break + } + findings = repairFindings(original.ProposalIndex, validated.rejected) + if actionable(original.ProposalIndex, validated.rejected) { + if attempt+1 == input.Spec.Settings.LLM.CorrectionRepairAttempts { + branch.Outcome = "exhausted" + } + continue + } + branch.Outcome = "unavailable" + for _, d := range validated.rejected { + if d.Disposition != validators.DispositionUnavailable { + branch.Outcome = "rejected" + break + } + } + break + } + branches = append(branches, branch) + } + return initial, branches, nil +} diff --git a/internal/framework/runner/repair_test.go b/internal/framework/runner/repair_test.go new file mode 100644 index 0000000..5814d00 --- /dev/null +++ b/internal/framework/runner/repair_test.go @@ -0,0 +1,181 @@ +package runner + +import ( + "context" + "encoding/json" + "errors" + "testing" + + "gitea.maximumdirect.net/eric/audita/v3/internal/core/config" + "gitea.maximumdirect.net/eric/audita/v3/internal/core/schema" + "gitea.maximumdirect.net/eric/audita/v3/internal/framework/contracts" + "gitea.maximumdirect.net/eric/audita/v3/internal/framework/outcomes" + "gitea.maximumdirect.net/eric/audita/v3/internal/framework/proposal_generation" + "gitea.maximumdirect.net/eric/audita/v3/internal/framework/proposals" + "gitea.maximumdirect.net/eric/audita/v3/internal/framework/structuredoutput" + "gitea.maximumdirect.net/eric/audita/v3/internal/framework/validators" +) + +type repairClient struct { + outputs []string + calls []structuredoutput.Request + err error +} + +func (c *repairClient) CompleteStructured(_ context.Context, req structuredoutput.Request, out any) (structuredoutput.Response, error) { + if req.Replay != nil { + copyReplay := *req.Replay + copyReplay.Turns = append([]structuredoutput.Message(nil), req.Replay.Turns...) + req.Replay = ©Replay + } + c.calls = append(c.calls, req) + if c.err != nil { + return structuredoutput.Response{}, c.err + } + value := c.outputs[len(c.calls)-1] + if err := json.Unmarshal([]byte(value), out); err != nil { + return structuredoutput.Response{}, err + } + return structuredoutput.Response{Content: json.RawMessage(value), FinalContent: value, RepairInstruction: []structuredoutput.Message{{Role: "user", Content: "instruction"}}}, nil +} + +func TestRepairBranchesRetargetAndPreserveOtherApprovals(t *testing.T) { + client := &repairClient{outputs: []string{`{"proposal_ref":2,"action":"revise","correction":{"id":2,"original_text":"uh ","corrected_text":"","confidence":0.9}}`}} + candidates := []proposals.EnrichedCorrectionProposal{ + {CorrectionProposal: proposals.CorrectionProposal{TargetSegmentID: 1, OriginalText: "cat", CorrectedText: "dog", Confidence: .9}, ProposalMetadata: proposals.ProposalMetadata{ProposalIndex: 4}}, + {CorrectionProposal: proposals.CorrectionProposal{TargetSegmentID: 0, OriginalText: "x", CorrectedText: "y", Confidence: .9}, ProposalMetadata: proposals.ProposalMetadata{ProposalIndex: 5}}, + } + input := validateSectionCandidatesInput{Spec: contracts.ModuleRunSpec{ModuleKey: "grammar", RepairClient: client, Settings: config.Stage{ConfidenceThreshold: .8, LLM: config.LLM{CorrectionRepairAttempts: 1}}}, Policy: proposals.ReplacementPolicyRequireUnique, WorkingTranscript: &schema.Transcript{Segments: []schema.Segment{{ID: 1, Text: "cat"}, {ID: 2, Text: "uh hello"}}}, Validators: []contracts.Validator{validators.ProposalShapeValidator{}, validators.NoEffectValidator{}, validators.OriginalTextPresenceValidator{}, validators.ConfidenceThresholdValidator{}, validators.NonEmptyResultingSegmentValidator{}}, SectionEnriched: candidates, Exchange: &structuredoutput.Exchange{PromptID: "proposal", FinalContent: "original response"}} + initial, err := validateSectionCandidates(context.Background(), input) + if err != nil { + t.Fatal(err) + } + result, branches, err := repairCandidates(context.Background(), input, []int{1, 9}, initial, "repair") + if err != nil { + t.Fatal(err) + } + if len(result.approved) != 2 || result.approved[1].ProposalIndex != 5 || result.approved[1].TargetSegmentID != 2 || result.approved[1].CorrectedText != "" || len(branches) != 1 || branches[0].Outcome != "approved" || len(result.filtered) != 0 { + t.Fatalf("result=%+v branches=%+v", result, branches) + } + var payload proposal_generation.RepairRequest + if err := json.Unmarshal([]byte(client.calls[0].Variables["RepairRequestJSON"]), &payload); err != nil { + t.Fatal(err) + } + if payload.ProposalRef != 2 || payload.Correction.SegmentReference != 9 { + t.Fatalf("request: %+v", payload) + } +} + +func TestRepairBranchTerminalOutcomes(t *testing.T) { + for _, tc := range []struct { + name string + outputs []string + err error + budget int + outcome string + calls int + }{ + {"withdraw", []string{`{"proposal_ref":1,"action":"withdraw","correction":null}`}, nil, 1, "withdrawn", 1}, + {"exhausted", []string{`{"proposal_ref":1,"action":"revise","correction":{"id":1,"original_text":"cat","corrected_text":"dog","confidence":0.1}}`, `{"proposal_ref":1,"action":"revise","correction":{"id":1,"original_text":"cat","corrected_text":"dog","confidence":0.2}}`}, nil, 2, "exhausted", 2}, + {"wrong reference", []string{`{"proposal_ref":2,"action":"withdraw","correction":null}`}, nil, 1, "unavailable", 1}, + {"provider failure", nil, &structuredoutput.Failure{Class: structuredoutput.FailureRecoverable, Reason: outcomes.ReasonProviderFailed, Message: "provider failed"}, 1, "unavailable", 1}, + } { + t.Run(tc.name, func(t *testing.T) { + client := &repairClient{outputs: tc.outputs, err: tc.err} + candidate := proposals.EnrichedCorrectionProposal{CorrectionProposal: proposals.CorrectionProposal{TargetSegmentID: 1, OriginalText: "cat", CorrectedText: "dog", Confidence: .1}, ProposalMetadata: proposals.ProposalMetadata{ProposalIndex: 0}} + input := validateSectionCandidatesInput{Spec: contracts.ModuleRunSpec{RepairClient: client, Settings: config.Stage{ConfidenceThreshold: .8, LLM: config.LLM{CorrectionRepairAttempts: tc.budget}}}, Policy: proposals.ReplacementPolicyRequireUnique, WorkingTranscript: &schema.Transcript{Segments: []schema.Segment{{ID: 1, Text: "cat"}}}, Validators: []contracts.Validator{validators.ProposalShapeValidator{}, validators.NoEffectValidator{}, validators.OriginalTextPresenceValidator{}, validators.ConfidenceThresholdValidator{}}, SectionEnriched: []proposals.EnrichedCorrectionProposal{candidate}, Exchange: &structuredoutput.Exchange{PromptID: "proposal"}} + initial, err := validateSectionCandidates(context.Background(), input) + if err != nil { + t.Fatal(err) + } + _, branches, err := repairCandidates(context.Background(), input, []int{1}, initial, "repair") + if err != nil { + t.Fatal(err) + } + if len(branches) != 1 || branches[0].Outcome != tc.outcome || len(client.calls) != tc.calls { + t.Fatalf("branches=%+v calls=%d", branches, len(client.calls)) + } + if tc.calls == 2 && len(client.calls[1].Replay.Turns) != 2 { + t.Fatalf("history: %+v", client.calls[1].Replay.Turns) + } + }) + } +} + +type repairModule struct{ outcomeModule } + +func (m repairModule) RepairPromptID() string { return "repair" } + +func TestRunnerAppliesRevisionsInOriginalOrder(t *testing.T) { + sectionIndex := 0 + candidates := []proposals.CorrectionProposal{{TargetSegmentID: 1, OriginalText: "cat", CorrectedText: "dog", Confidence: .9}, {TargetSegmentID: 1, OriginalText: "cat", CorrectedText: "fox", Confidence: .1}} + client := &repairClient{outputs: []string{`{"proposal_ref":2,"action":"revise","correction":{"id":1,"original_text":"cat","corrected_text":"fox","confidence":0.9}}`}} + module := repairModule{outcomeModule{fakeModule: fakeModule{key: "grammar", policy: proposals.ReplacementPolicyRequireUnique, validators: []contracts.Validator{validators.ProposalShapeValidator{}, validators.NoEffectValidator{}, validators.OriginalTextPresenceValidator{}, validators.ConfidenceThresholdValidator{}}}, proposeResult: func(req contracts.ProposalRequest) (contracts.ProposalResult, error) { + return contracts.ProposalResult{Proposals: candidates, SegmentReferences: []int{1, 1}, Exchange: &structuredoutput.Exchange{PromptID: "proposal"}}, nil + }}} + runner := New(fakeFactory{modules: map[string]contracts.TranscriptModule{"grammar": module}}) + out, err := runner.Run(context.Background(), RunInput{Transcript: &schema.Transcript{Segments: []schema.Segment{{ID: 1, Text: "cat"}}}, ModuleSpecs: []contracts.ModuleRunSpec{{ModuleKey: "grammar", InstanceName: "grammar", RepairClient: client, Settings: config.Stage{ConfidenceThreshold: .8, LLM: config.LLM{CorrectionRepairAttempts: 1}}}}}) + if err != nil { + t.Fatal(err) + } + if out.FinalTranscript.Segments[0].Text != "dog" || len(out.WorkUnits) != 1 || len(out.WorkUnits[0].Repairs) != 1 || out.WorkUnits[0].Repairs[0].Outcome != "approved_skipped" || out.WorkUnits[0].CorrectionsApplied != 1 || len(out.ModuleResults[0].SkippedChanges) != 1 || out.ModuleResults[0].SkippedChanges[0].ProposalIndex != 1 { + t.Fatalf("output=%+v section=%d", out, sectionIndex) + } +} + +func TestRepairRevalidatesEarlierApprovalAndRetainsUnrelatedApprovalOnFailure(t *testing.T) { + candidates := []proposals.EnrichedCorrectionProposal{ + {CorrectionProposal: proposals.CorrectionProposal{TargetSegmentID: 1, OriginalText: "cat", CorrectedText: "dog", Confidence: .9}, ProposalMetadata: proposals.ProposalMetadata{ProposalIndex: 0}}, + {CorrectionProposal: proposals.CorrectionProposal{TargetSegmentID: 1, OriginalText: "bird", CorrectedText: "owl", Confidence: .1}, ProposalMetadata: proposals.ProposalMetadata{ProposalIndex: 1}}, + } + for _, tc := range []struct { + name string + failure bool + }{{"revision", false}, {"provider failure", true}} { + t.Run(tc.name, func(t *testing.T) { + calls := 0 + check := fakeValidator{name: "custom", validateF: func(req contracts.ValidationRequest) (validators.Result, error) { + var decisions []validators.Decision + for _, candidate := range req.CandidateProposal { + calls++ + decisions = append(decisions, validators.Decision{ProposalIndex: candidate.ProposalIndex, Approved: true, Disposition: validators.DispositionApproved}) + } + return validators.Result{Decisions: decisions}, nil + }} + client := &repairClient{outputs: []string{`{"proposal_ref":2,"action":"revise","correction":{"id":1,"original_text":"bird","corrected_text":"owl","confidence":0.9}}`}} + if tc.failure { + client.err = &structuredoutput.Failure{Class: structuredoutput.FailureRecoverable, Reason: outcomes.ReasonProviderFailed, Message: "failed"} + } + input := validateSectionCandidatesInput{Spec: contracts.ModuleRunSpec{RepairClient: client, Settings: config.Stage{ConfidenceThreshold: .8, LLM: config.LLM{CorrectionRepairAttempts: 1}}}, Policy: proposals.ReplacementPolicyRequireUnique, WorkingTranscript: &schema.Transcript{Segments: []schema.Segment{{ID: 1, Text: "cat bird"}}}, Validators: []contracts.Validator{check, validators.ConfidenceThresholdValidator{}}, SectionEnriched: candidates, Exchange: &structuredoutput.Exchange{PromptID: "proposal"}} + initial, err := validateSectionCandidates(context.Background(), input) + if err != nil { + t.Fatal(err) + } + got, branches, err := repairCandidates(context.Background(), input, []int{1, 1}, initial, "repair") + if err != nil { + t.Fatal(err) + } + if tc.failure { + if len(got.approved) != 1 || got.approved[0].ProposalIndex != 0 || branches[0].Outcome != "unavailable" || calls != 2 { + t.Fatalf("result=%+v calls=%d", got, calls) + } + } else if len(got.approved) != 2 || branches[0].Outcome != "approved" || calls != 3 { + t.Fatalf("result=%+v calls=%d", got, calls) + } + }) + } +} + +func TestRepairInterruptionRetainsAttempt(t *testing.T) { + client := &repairClient{err: &structuredoutput.Failure{Class: structuredoutput.FailureCanceled, Reason: outcomes.ReasonCanceled, Message: "canceled", Cause: context.Canceled}} + candidate := proposals.EnrichedCorrectionProposal{CorrectionProposal: proposals.CorrectionProposal{TargetSegmentID: 1, OriginalText: "cat", CorrectedText: "dog", Confidence: .1}, ProposalMetadata: proposals.ProposalMetadata{ProposalIndex: 0}} + input := validateSectionCandidatesInput{Spec: contracts.ModuleRunSpec{RepairClient: client, Settings: config.Stage{ConfidenceThreshold: .8, LLM: config.LLM{CorrectionRepairAttempts: 1}}}, Policy: proposals.ReplacementPolicyRequireUnique, WorkingTranscript: &schema.Transcript{Segments: []schema.Segment{{ID: 1, Text: "cat"}}}, Validators: []contracts.Validator{validators.ConfidenceThresholdValidator{}}, SectionEnriched: []proposals.EnrichedCorrectionProposal{candidate}, Exchange: &structuredoutput.Exchange{PromptID: "proposal"}} + initial, err := validateSectionCandidates(context.Background(), input) + if err != nil { + t.Fatal(err) + } + _, branches, err := repairCandidates(context.Background(), input, []int{1}, initial, "repair") + if !errors.Is(err, context.Canceled) || len(branches) != 1 || branches[0].Outcome != "interrupted" || len(branches[0].Attempts) != 1 { + t.Fatalf("branches=%+v error=%v", branches, err) + } +} diff --git a/internal/framework/runner/runner.go b/internal/framework/runner/runner.go index 5c537b2..dcba057 100644 --- a/internal/framework/runner/runner.go +++ b/internal/framework/runner/runner.go @@ -65,6 +65,7 @@ type WorkUnitOutcome struct { Section contracts.SectionMetadata `json:"section"` State outcomes.WorkState `json:"state"` CorrectionsApplied int `json:"corrections_applied"` + Repairs []RepairOutcome `json:"-"` Operations []outcomes.Operation `json:"operations,omitempty"` } @@ -186,6 +187,16 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) { if input.CaptureCandidates { candidates = append(candidates, pipelineResult.Candidates...) } + if pipelineErr == nil && ctx.Err() != nil { + pipelineErr = ctx.Err() + for i := range pipelineResult.WorkUnits { + for j := range pipelineResult.WorkUnits[i].Repairs { + if pipelineResult.WorkUnits[i].Repairs[j].Outcome == "approved" { + pipelineResult.WorkUnits[i].Repairs[j].Outcome = "interrupted" + } + } + } + } if pipelineErr != nil { failed := ModuleResult{ StageIndex: spec.Index, StageID: spec.ID, @@ -209,6 +220,18 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) { updated, applied, skipped, appliedBySection, applyErr := applyWithinSections(sections, pipelineResult.Approved, policy, glossaryView.Protection, spec.ModuleKey == modulecatalog.KeySpokenWord) if applyErr != nil { + for i := range pipelineResult.WorkUnits { + unit := &pipelineResult.WorkUnits[i] + unit.ModuleIndex = moduleIndex + unit.State = outcomes.WorkInterrupted + for j := range unit.Repairs { + if unit.Repairs[j].Outcome == "approved" { + unit.Repairs[j].Outcome = "interrupted" + } + } + } + copy(workUnits[moduleIndex*len(sections):(moduleIndex+1)*len(sections)], pipelineResult.WorkUnits) + results = append(results, ModuleResult{StageIndex: spec.Index, StageID: spec.ID, ModuleKey: spec.ModuleKey, ModuleInstance: spec.InstanceName, ReplacementPolicy: policy, Status: ModuleStatusFailed, ProposalCount: pipelineResult.ProposalCount, Operations: pipelineResult.Operations, ValidatorDecisions: pipelineResult.ValidatorDecisions, ValidatorRejected: pipelineResult.ValidatorRejected, ErrorMessage: applyErr.Error()}) return snapshot(), fmt.Errorf("module %q application failed: %w", spec.InstanceName, applyErr) } copy(workUnits[moduleIndex*len(sections):(moduleIndex+1)*len(sections)], pipelineResult.WorkUnits) @@ -218,6 +241,22 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) { unit.State = outcomes.WorkCompleted unit.CorrectionsApplied = appliedBySection[sectionPos] } + for i := range workUnits[moduleIndex*len(sections) : (moduleIndex+1)*len(sections)] { + unit := &workUnits[moduleIndex*len(sections)+i] + for j := range unit.Repairs { + branch := &unit.Repairs[j] + if branch.Outcome != "approved" { + continue + } + branch.Outcome = "approved_skipped" + for _, change := range applied { + if change.ProposalIndex == branch.ProposalIndex { + branch.Outcome = "applied" + break + } + } + } + } attributeApplyFiltering(sections, pipelineResult.Approved, skipped, policy, workUnits[moduleIndex*len(sections):(moduleIndex+1)*len(sections)]) sections = updated moduleOperations := make([]outcomes.Operation, 0) @@ -310,6 +349,7 @@ type sectionProposalResult struct { sectionPos int section chunking.Section corrected []proposals.CorrectionProposal + references []int exchange *structuredoutput.Exchange operations []outcomes.Operation started bool @@ -317,6 +357,7 @@ type sectionProposalResult struct { } type sectionValidationResult struct { + repairs []RepairOutcome sectionPos int enriched []proposals.EnrichedCorrectionProposal approved []proposals.EnrichedCorrectionProposal @@ -379,6 +420,7 @@ func collectSectionProposals(ctx context.Context, input collectSectionProposalsI sectionPos: sectionPos, section: section, corrected: proposalResult.Proposals, + references: proposalResult.SegmentReferences, exchange: proposalResult.Exchange, operations: proposalResult.Operations, started: true, @@ -514,8 +556,18 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput) GlossaryView: input.GlossaryView, ReportDiagnosticError: input.ReportDiagnosticError, }) + var repairs []RepairOutcome + if err == nil && input.Spec.Settings.LLM.CorrectionRepairAttempts > 0 { + promptID := "" + if provider, ok := input.Module.(interface{ RepairPromptID() string }); ok { + promptID = provider.RepairPromptID() + } + repairInput := validateSectionCandidatesInput{Spec: input.Spec, Policy: input.Policy, WorkingTranscript: transcriptFromSection(sectionResult.section), SectionIndex: §ionMetadata.Index, Validators: validatorsOrdered, SectionEnriched: enriched, Exchange: sectionResult.exchange, ValidationLLMClient: input.ValidationClient, DiagnosticsDir: input.ValidationDiagnosticsDir, GlossaryView: input.GlossaryView, ReportDiagnosticError: input.ReportDiagnosticError} + validated, repairs, err = repairCandidates(runCtx, repairInput, sectionResult.references, validated, promptID) + } setErr(err) validationResults <- sectionValidationResult{ + repairs: repairs, sectionPos: sectionPos, enriched: enriched, approved: validated.approved, @@ -547,6 +599,7 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput) } unit := &out.WorkUnits[i] unit.Operations = append(unit.Operations, res.operations...) + unit.Repairs = res.repairs mergeFilteredProposals(unit, res.filtered) if res.err == nil { unit.State = outcomes.WorkCompleted @@ -578,6 +631,13 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput) } if err := getErr(); err != nil { + for i := range out.WorkUnits { + for j := range out.WorkUnits[i].Repairs { + if out.WorkUnits[i].Repairs[j].Outcome == "approved" { + out.WorkUnits[i].Repairs[j].Outcome = "interrupted" + } + } + } return out, err }