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

Integrate partial processing outcomes and fatal cancellation

Browse Source
This commit is contained in:
Eric Rakestraw
2026-09-24 17:23:59 +00:00
parent bb21a81805
commit b98604e5ca
4 changed files with 294 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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 = &sections[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))