eric/audita
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 6 Wiki Activity

Repair rejected corrections in independent branches

Browse Source
This commit is contained in:
Eric Rakestraw
2026-11-11 04:29:18 +00:00
parent a5003a1e52
commit d927c61199
9 changed files with 451 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
cmd/audita/main_integration_test.go
View File

@@ -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"
}

8
docs/internal/pipeline.md
View File

@@ -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.

2
docs/roadmap/implementation.md
View File

@@ -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

7
internal/framework/contracts/contracts.go
View File

@@ -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.

24
internal/framework/proposal_generation/generate.go
View File

@@ -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
}

7
internal/framework/proposal_generation/module_proposal.go
View File

@@ -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
}

179
internal/framework/runner/repair.go Normal file
View File

@@ -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
}

181
internal/framework/runner/repair_test.go Normal file
View File

@@ -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 = &copyReplay
}
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)
}
}

60
internal/framework/runner/runner.go
View File

@@ -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: &sectionMetadata.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
}