904 lines
33 KiB
Go
904 lines
33 KiB
Go
package runner
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"path/filepath"
|
|
"sort"
|
|
"sync"
|
|
"time"
|
|
|
|
"gitea.maximumdirect.net/eric/audita/internal/core/chunking"
|
|
"gitea.maximumdirect.net/eric/audita/internal/core/config"
|
|
"gitea.maximumdirect.net/eric/audita/internal/core/schema"
|
|
"gitea.maximumdirect.net/eric/audita/internal/framework/contracts"
|
|
"gitea.maximumdirect.net/eric/audita/internal/framework/llm"
|
|
"gitea.maximumdirect.net/eric/audita/internal/framework/outcomes"
|
|
"gitea.maximumdirect.net/eric/audita/internal/framework/proposals"
|
|
"gitea.maximumdirect.net/eric/audita/internal/framework/validators"
|
|
validatormetadata "gitea.maximumdirect.net/eric/audita/internal/validators/metadata"
|
|
)
|
|
|
|
const (
|
|
ModuleStatusSuccess = "success"
|
|
ModuleStatusFailed = "failed"
|
|
)
|
|
|
|
// ModuleFactory resolves one module instance for one run spec.
|
|
type ModuleFactory interface {
|
|
ModuleForSpec(spec contracts.ModuleRunSpec) (contracts.TranscriptModule, error)
|
|
}
|
|
|
|
// Runner executes module instances sequentially against a working transcript.
|
|
type Runner struct {
|
|
factory ModuleFactory
|
|
}
|
|
|
|
// ModuleResult captures deterministic per-module execution output.
|
|
type ModuleResult struct {
|
|
ModuleKey string `json:"module_key"`
|
|
ModuleInstance string `json:"module_instance"`
|
|
ReplacementPolicy proposals.ReplacementPolicy `json:"replacement_policy"`
|
|
Status string `json:"status"`
|
|
ProposalCount int `json:"proposal_count"`
|
|
Operations []outcomes.Operation `json:"operations,omitempty"`
|
|
ValidatorDecisions []ValidatorDecisionRecord `json:"validator_decisions,omitempty"`
|
|
ValidatorRejected []ValidatorRejectedChange `json:"validator_rejected,omitempty"`
|
|
AppliedChanges []proposals.AppliedChange `json:"applied_changes,omitempty"`
|
|
SkippedChanges []proposals.SkippedChange `json:"skipped_changes,omitempty"`
|
|
ErrorMessage string `json:"error_message,omitempty"`
|
|
StartedAt time.Time `json:"started_at"`
|
|
CompletedAt time.Time `json:"completed_at"`
|
|
}
|
|
|
|
// WorkUnitOutcome keeps initial section identity while text evolves between
|
|
// module instances. Detailed candidate and decision records are debug evidence.
|
|
type WorkUnitOutcome struct {
|
|
ModuleIndex int `json:"module_index"`
|
|
ModuleKey string `json:"module_key"`
|
|
ModuleInstance string `json:"module_instance"`
|
|
Section contracts.SectionMetadata `json:"section"`
|
|
State outcomes.WorkState `json:"state"`
|
|
CorrectionsApplied int `json:"corrections_applied"`
|
|
Operations []outcomes.Operation `json:"operations,omitempty"`
|
|
}
|
|
|
|
type ValidatorDecisionRecord struct {
|
|
ValidatorName string `json:"validator_name"`
|
|
ProposalIndex int `json:"proposal_index"`
|
|
Approved bool `json:"approved"`
|
|
Disposition validators.DecisionDisposition `json:"disposition"`
|
|
ReasonCode string `json:"reason_code"`
|
|
Message string `json:"message"`
|
|
DiagnosticArtifactPath string `json:"diagnostic_artifact_path,omitempty"`
|
|
}
|
|
|
|
type ValidatorRejectedChange struct {
|
|
ValidatorName string `json:"validator_name"`
|
|
ProposalIndex int `json:"proposal_index"`
|
|
ModuleKey string `json:"module_key"`
|
|
ModuleInstance string `json:"module_instance"`
|
|
TargetSegmentID int `json:"target_segment_id"`
|
|
OriginalText string `json:"original_text"`
|
|
CorrectedText string `json:"corrected_text"`
|
|
Disposition validators.DecisionDisposition `json:"disposition"`
|
|
ReasonCode string `json:"reason_code"`
|
|
Message string `json:"message"`
|
|
}
|
|
|
|
// RunInput is the deterministic runner input.
|
|
type RunInput struct {
|
|
// ReportDiagnosticError may be called concurrently; it must not affect processing.
|
|
ReportDiagnosticError func(error) `json:"-"`
|
|
Config *config.Config
|
|
Transcript *schema.Transcript
|
|
Glossary *schema.Glossary
|
|
ModuleSpecs []contracts.ModuleRunSpec
|
|
ProposalLLMClient contracts.StructuredLLMClient
|
|
ProposalDiagnosticsDir string
|
|
ValidationLLMClient contracts.StructuredLLMClient
|
|
ValidationDiagnosticsDir string
|
|
CaptureCandidates bool
|
|
timings *validatorTimingCollector
|
|
}
|
|
|
|
// RunOutput is the deterministic runner output.
|
|
type RunOutput struct {
|
|
FinalTranscript *schema.Transcript `json:"-"`
|
|
ModuleResults []ModuleResult `json:"module_results"`
|
|
Sections []contracts.SectionMetadata `json:"sections"`
|
|
InventoryAvailable bool `json:"-"`
|
|
WorkUnits []WorkUnitOutcome `json:"work_units"`
|
|
Candidates []proposals.EnrichedCorrectionProposal `json:"-"`
|
|
Observations llm.ObservationSnapshot `json:"-"`
|
|
ObservationsAvailable bool `json:"-"`
|
|
LLMSummary llm.ExecutionSummary `json:"-"`
|
|
RunWallTimeMS int64 `json:"-"`
|
|
ValidatorTimings []ValidatorTiming `json:"-"`
|
|
}
|
|
|
|
func New(factory ModuleFactory) *Runner {
|
|
return &Runner{factory: factory}
|
|
}
|
|
|
|
func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) {
|
|
started := time.Now()
|
|
input.timings = newValidatorTimingCollector()
|
|
output, err := r.run(ctx, input)
|
|
output.RunWallTimeMS = time.Since(started).Milliseconds()
|
|
output.ValidatorTimings = input.timings.snapshot()
|
|
return output, err
|
|
}
|
|
|
|
func (r *Runner) run(ctx context.Context, input RunInput) (RunOutput, error) {
|
|
if r == nil || r.factory == nil {
|
|
return RunOutput{}, fmt.Errorf("runner module factory is required")
|
|
}
|
|
|
|
working := cloneTranscript(input.Transcript)
|
|
results := make([]ModuleResult, 0, len(input.ModuleSpecs))
|
|
sections, err := chunkWorkingTranscript(input.Config, working)
|
|
if err != nil {
|
|
return RunOutput{FinalTranscript: working}, fmt.Errorf("chunk normalized transcript: %w", err)
|
|
}
|
|
inventory := make([]contracts.SectionMetadata, len(sections))
|
|
workUnits := make([]WorkUnitOutcome, 0, len(input.ModuleSpecs)*len(sections))
|
|
var candidates []proposals.EnrichedCorrectionProposal
|
|
for i, section := range sections {
|
|
inventory[i] = contracts.SectionMetadataFromSection(section)
|
|
}
|
|
for moduleIndex, spec := range input.ModuleSpecs {
|
|
for _, section := range inventory {
|
|
workUnits = append(workUnits, WorkUnitOutcome{
|
|
ModuleIndex: moduleIndex, ModuleKey: spec.ModuleKey,
|
|
ModuleInstance: spec.InstanceName, Section: section,
|
|
State: outcomes.WorkNotStarted,
|
|
})
|
|
}
|
|
}
|
|
snapshot := func() RunOutput {
|
|
return RunOutput{
|
|
FinalTranscript: transcriptFromSections(sections), ModuleResults: results,
|
|
Sections: inventory, InventoryAvailable: true, WorkUnits: workUnits, Candidates: candidates,
|
|
}
|
|
}
|
|
|
|
for moduleIndex, spec := range input.ModuleSpecs {
|
|
startedAt := time.Now().UTC()
|
|
module, err := r.factory.ModuleForSpec(spec)
|
|
if err != nil {
|
|
failed := ModuleResult{
|
|
ModuleKey: spec.ModuleKey,
|
|
ModuleInstance: spec.InstanceName,
|
|
Status: ModuleStatusFailed,
|
|
ErrorMessage: err.Error(),
|
|
StartedAt: startedAt,
|
|
CompletedAt: time.Now().UTC(),
|
|
}
|
|
results = append(results, failed)
|
|
return snapshot(), fmt.Errorf("module %q setup failed: %w", spec.InstanceName, err)
|
|
}
|
|
|
|
policy := module.ReplacementPolicy()
|
|
pipelineResult, pipelineErr := runModulePipeline(ctx, collectSectionProposalsInput{
|
|
Module: module,
|
|
Spec: spec,
|
|
Policy: policy,
|
|
Config: input.Config,
|
|
Glossary: input.Glossary,
|
|
Sections: sections,
|
|
ProposalClient: input.ProposalLLMClient,
|
|
DiagnosticsDir: input.ProposalDiagnosticsDir,
|
|
ValidationClient: input.ValidationLLMClient,
|
|
ValidationDiagnosticsDir: input.ValidationDiagnosticsDir,
|
|
Timings: input.timings,
|
|
ReportDiagnosticError: input.ReportDiagnosticError,
|
|
})
|
|
if input.CaptureCandidates {
|
|
candidates = append(candidates, pipelineResult.Candidates...)
|
|
}
|
|
if pipelineErr != nil {
|
|
failed := ModuleResult{
|
|
ModuleKey: spec.ModuleKey,
|
|
ModuleInstance: spec.InstanceName,
|
|
ReplacementPolicy: policy,
|
|
Status: ModuleStatusFailed,
|
|
ProposalCount: pipelineResult.ProposalCount,
|
|
Operations: pipelineResult.Operations,
|
|
ValidatorDecisions: pipelineResult.ValidatorDecisions,
|
|
ValidatorRejected: pipelineResult.ValidatorRejected,
|
|
ErrorMessage: pipelineErr.Error(),
|
|
StartedAt: startedAt,
|
|
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)
|
|
}
|
|
|
|
updated, applied, skipped, appliedBySection, applyErr := applyWithinSections(sections, pipelineResult.Approved, policy)
|
|
if applyErr != nil {
|
|
return snapshot(), fmt.Errorf("module %q application failed: %w", spec.InstanceName, applyErr)
|
|
}
|
|
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,
|
|
ModuleInstance: spec.InstanceName,
|
|
ReplacementPolicy: policy,
|
|
Status: ModuleStatusSuccess,
|
|
ProposalCount: pipelineResult.ProposalCount,
|
|
Operations: moduleOperations,
|
|
ValidatorDecisions: pipelineResult.ValidatorDecisions,
|
|
ValidatorRejected: pipelineResult.ValidatorRejected,
|
|
AppliedChanges: applied,
|
|
SkippedChanges: skipped,
|
|
StartedAt: startedAt,
|
|
CompletedAt: time.Now().UTC(),
|
|
})
|
|
}
|
|
|
|
output := snapshot()
|
|
return output, nil
|
|
}
|
|
|
|
func transcriptFromSections(sections []chunking.Section) *schema.Transcript {
|
|
segments := make([]schema.Segment, 0)
|
|
for _, section := range sections {
|
|
segments = append(segments, section.Segments...)
|
|
}
|
|
return &schema.Transcript{Segments: segments}
|
|
}
|
|
|
|
// applyWithinSections leaves section membership and initial metadata unchanged.
|
|
// Approved proposals are applied only to their identified section.
|
|
func applyWithinSections(sections []chunking.Section, approved []proposals.EnrichedCorrectionProposal, policy proposals.ReplacementPolicy) ([]chunking.Section, []proposals.AppliedChange, []proposals.SkippedChange, []int, error) {
|
|
bySection := make(map[int][]proposals.EnrichedCorrectionProposal, len(sections))
|
|
known := make(map[int]struct{}, len(sections))
|
|
for _, section := range sections {
|
|
known[section.Index] = struct{}{}
|
|
}
|
|
for _, candidate := range approved {
|
|
if candidate.SectionIndex == nil {
|
|
return nil, nil, nil, nil, fmt.Errorf("approved proposal %d has no section identity", candidate.ProposalIndex)
|
|
}
|
|
if _, ok := known[*candidate.SectionIndex]; !ok {
|
|
return nil, nil, nil, nil, fmt.Errorf("approved proposal %d has unknown section identity", candidate.ProposalIndex)
|
|
}
|
|
bySection[*candidate.SectionIndex] = append(bySection[*candidate.SectionIndex], candidate)
|
|
}
|
|
updated := make([]chunking.Section, len(sections))
|
|
copy(updated, sections)
|
|
applied := make([]proposals.AppliedChange, 0)
|
|
skipped := make([]proposals.SkippedChange, 0)
|
|
appliedBySection := make([]int, len(sections))
|
|
for i, section := range sections {
|
|
result := proposals.ApplyProposals(transcriptFromSection(section), bySection[section.Index], policy)
|
|
updated[i].Segments = result.Transcript.Segments
|
|
applied = append(applied, result.Applied...)
|
|
skipped = append(skipped, result.Skipped...)
|
|
appliedBySection[i] = len(result.Applied)
|
|
}
|
|
return updated, applied, skipped, appliedBySection, nil
|
|
}
|
|
|
|
type collectSectionProposalsInput struct {
|
|
// ReportDiagnosticError may be called concurrently; it must not affect processing.
|
|
ReportDiagnosticError func(error) `json:"-"`
|
|
Module contracts.TranscriptModule
|
|
Spec contracts.ModuleRunSpec
|
|
Policy proposals.ReplacementPolicy
|
|
Config *config.Config
|
|
Glossary *schema.Glossary
|
|
Sections []chunking.Section
|
|
ProposalClient contracts.StructuredLLMClient
|
|
DiagnosticsDir string
|
|
ValidationClient contracts.StructuredLLMClient
|
|
ValidationDiagnosticsDir string
|
|
Timings *validatorTimingCollector
|
|
}
|
|
|
|
type sectionProposalResult struct {
|
|
sectionPos int
|
|
section chunking.Section
|
|
corrected []proposals.CorrectionProposal
|
|
operations []outcomes.Operation
|
|
started bool
|
|
err error
|
|
}
|
|
|
|
type sectionValidationResult struct {
|
|
sectionPos int
|
|
enriched []proposals.EnrichedCorrectionProposal
|
|
approved []proposals.EnrichedCorrectionProposal
|
|
decisions []ValidatorDecisionRecord
|
|
rejected []ValidatorRejectedChange
|
|
operations []outcomes.Operation
|
|
filtered []outcomes.FilteredProposal
|
|
err error
|
|
}
|
|
|
|
type modulePipelineResult struct {
|
|
ProposalCount int
|
|
Candidates []proposals.EnrichedCorrectionProposal
|
|
Approved []proposals.EnrichedCorrectionProposal
|
|
Operations []outcomes.Operation
|
|
WorkUnits []WorkUnitOutcome
|
|
ValidatorDecisions []ValidatorDecisionRecord
|
|
ValidatorRejected []ValidatorRejectedChange
|
|
}
|
|
|
|
func collectSectionProposals(ctx context.Context, input collectSectionProposalsInput) (context.Context, <-chan sectionProposalResult, context.CancelFunc) {
|
|
results := make(chan sectionProposalResult, len(input.Sections))
|
|
runCtx, cancel := context.WithCancel(ctx)
|
|
|
|
go func() {
|
|
defer close(results)
|
|
var wg sync.WaitGroup
|
|
for sectionPos, section := range input.Sections {
|
|
sectionPos := sectionPos
|
|
section := section
|
|
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{
|
|
Config: input.Config,
|
|
WorkingTranscript: transcriptFromSection(section),
|
|
Glossary: input.Glossary,
|
|
Section: &meta,
|
|
DiagnosticsDir: input.DiagnosticsDir,
|
|
ReportDiagnosticError: input.ReportDiagnosticError,
|
|
},
|
|
RunSpec: contracts.ModuleRunSpec{
|
|
ModuleKey: input.Spec.ModuleKey,
|
|
InstanceName: input.Spec.InstanceName,
|
|
ReplacementPolicy: input.Policy,
|
|
},
|
|
LLMClient: input.ProposalClient,
|
|
})
|
|
results <- sectionProposalResult{
|
|
sectionPos: sectionPos,
|
|
section: section,
|
|
corrected: proposalResult.Proposals,
|
|
operations: proposalResult.Operations,
|
|
started: true,
|
|
err: err,
|
|
}
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
}()
|
|
|
|
return runCtx, results, cancel
|
|
}
|
|
|
|
func runModulePipeline(ctx context.Context, input collectSectionProposalsInput) (modulePipelineResult, error) {
|
|
out := modulePipelineResult{
|
|
Approved: make([]proposals.EnrichedCorrectionProposal, 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
|
|
}
|
|
|
|
runCtx, proposalResults, cancel := collectSectionProposals(ctx, input)
|
|
defer cancel()
|
|
|
|
pending := make(map[int]sectionProposalResult, len(input.Sections))
|
|
validationResults := make(chan sectionValidationResult, len(input.Sections))
|
|
validationBySection := make(map[int]sectionValidationResult, len(input.Sections))
|
|
validatorsOrdered, validatorOrder := reorderValidatorsForPipeline(input.Module.Validators())
|
|
|
|
nextSectionToProcess := 0
|
|
nextProposalIndex := 0
|
|
|
|
var (
|
|
firstErr error
|
|
errMu sync.Mutex
|
|
vwg sync.WaitGroup
|
|
)
|
|
|
|
setErr := func(err error) {
|
|
if err == nil {
|
|
return
|
|
}
|
|
errMu.Lock()
|
|
defer errMu.Unlock()
|
|
if firstErr == nil {
|
|
firstErr = err
|
|
cancel()
|
|
}
|
|
}
|
|
getErr := func() error {
|
|
errMu.Lock()
|
|
defer errMu.Unlock()
|
|
return firstErr
|
|
}
|
|
|
|
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
|
|
}
|
|
if getErr() != nil {
|
|
continue
|
|
}
|
|
if !result.started {
|
|
continue
|
|
}
|
|
pending[result.sectionPos] = result
|
|
|
|
for {
|
|
if getErr() != nil {
|
|
break
|
|
}
|
|
sectionResult, ok := pending[nextSectionToProcess]
|
|
if !ok {
|
|
break
|
|
}
|
|
delete(pending, nextSectionToProcess)
|
|
sectionMeta := contracts.SectionMetadataFromSection(sectionResult.section)
|
|
sectionEnriched := make([]proposals.EnrichedCorrectionProposal, 0, len(sectionResult.corrected))
|
|
for i, corrected := range sectionResult.corrected {
|
|
sectionIndex := sectionMeta.Index
|
|
sectionEnriched = append(sectionEnriched, proposals.EnrichedCorrectionProposal{
|
|
CorrectionProposal: corrected,
|
|
ProposalMetadata: proposals.ProposalMetadata{
|
|
ProposalIndex: nextProposalIndex + i,
|
|
ModuleKey: input.Spec.ModuleKey,
|
|
ModuleInstance: input.Spec.InstanceName,
|
|
SectionIndex: §ionIndex,
|
|
},
|
|
})
|
|
}
|
|
nextProposalIndex += len(sectionEnriched)
|
|
out.ProposalCount += len(sectionEnriched)
|
|
out.Candidates = append(out.Candidates, 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)
|
|
}
|
|
}
|
|
}
|
|
|
|
vwg.Add(1)
|
|
go func(sectionPos int, enriched []proposals.EnrichedCorrectionProposal, sectionMetadata contracts.SectionMetadata) {
|
|
defer vwg.Done()
|
|
validated, err := validateSectionCandidates(runCtx, validateSectionCandidatesInput{
|
|
Spec: input.Spec,
|
|
Policy: input.Policy,
|
|
Glossary: input.Glossary,
|
|
Config: input.Config,
|
|
WorkingTranscript: transcriptFromSection(sectionResult.section),
|
|
SectionIndex: §ionMetadata.Index,
|
|
Validators: validatorsOrdered,
|
|
SectionEnriched: enriched,
|
|
ValidationLLMClient: input.ValidationClient,
|
|
DiagnosticsDir: input.ValidationDiagnosticsDir,
|
|
Timings: input.Timings,
|
|
ReportDiagnosticError: input.ReportDiagnosticError,
|
|
})
|
|
setErr(err)
|
|
validationResults <- sectionValidationResult{
|
|
sectionPos: sectionPos,
|
|
enriched: enriched,
|
|
approved: validated.approved,
|
|
decisions: validated.decisions,
|
|
rejected: validated.rejected,
|
|
operations: validated.operations,
|
|
filtered: validated.filtered,
|
|
err: err,
|
|
}
|
|
}(nextSectionToProcess, sectionEnriched, sectionMeta)
|
|
nextSectionToProcess++
|
|
}
|
|
}
|
|
|
|
vwg.Wait()
|
|
close(validationResults)
|
|
|
|
for v := range validationResults {
|
|
validationBySection[v.sectionPos] = v
|
|
if v.err != nil {
|
|
setErr(v.err)
|
|
}
|
|
}
|
|
|
|
for i := range input.Sections {
|
|
res, ok := validationBySection[i]
|
|
if !ok {
|
|
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.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 {
|
|
return out.ValidatorDecisions[i].ProposalIndex < out.ValidatorDecisions[j].ProposalIndex
|
|
}
|
|
return validatorOrder[out.ValidatorDecisions[i].ValidatorName] < validatorOrder[out.ValidatorDecisions[j].ValidatorName]
|
|
})
|
|
sort.SliceStable(out.ValidatorRejected, func(i, j int) bool {
|
|
if out.ValidatorRejected[i].ProposalIndex != out.ValidatorRejected[j].ProposalIndex {
|
|
return out.ValidatorRejected[i].ProposalIndex < out.ValidatorRejected[j].ProposalIndex
|
|
}
|
|
return validatorOrder[out.ValidatorRejected[i].ValidatorName] < validatorOrder[out.ValidatorRejected[j].ValidatorName]
|
|
})
|
|
if ctx.Err() != nil {
|
|
setErr(ctx.Err())
|
|
}
|
|
|
|
if err := getErr(); err != nil {
|
|
return out, err
|
|
}
|
|
|
|
return out, nil
|
|
}
|
|
|
|
type validateSectionCandidatesInput struct {
|
|
// ReportDiagnosticError may be called concurrently; it must not affect processing.
|
|
ReportDiagnosticError func(error) `json:"-"`
|
|
Spec contracts.ModuleRunSpec
|
|
Policy proposals.ReplacementPolicy
|
|
Glossary *schema.Glossary
|
|
Config *config.Config
|
|
WorkingTranscript *schema.Transcript
|
|
SectionIndex *int
|
|
Validators []contracts.Validator
|
|
SectionEnriched []proposals.EnrichedCorrectionProposal
|
|
ValidationLLMClient contracts.StructuredLLMClient
|
|
DiagnosticsDir string
|
|
Timings *validatorTimingCollector
|
|
}
|
|
|
|
type validateSectionCandidatesResult struct {
|
|
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)
|
|
operations := make([]outcomes.Operation, 0)
|
|
filtered := make([]outcomes.FilteredProposal, 0)
|
|
eligible := append([]proposals.EnrichedCorrectionProposal(nil), input.SectionEnriched...)
|
|
|
|
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{
|
|
writer: llm.NewDiagnosticsWriter(
|
|
filepath.Join(input.DiagnosticsDir, input.Spec.InstanceName),
|
|
llm.ConfiguredSecrets(input.Config),
|
|
),
|
|
}
|
|
}
|
|
|
|
validatorStarted := time.Now()
|
|
vResult, err := validator.Validate(ctx, contracts.ValidationRequest{
|
|
WorkingTranscript: input.WorkingTranscript,
|
|
CandidateProposal: eligible,
|
|
ModuleKey: input.Spec.ModuleKey,
|
|
ModuleInstance: input.Spec.InstanceName,
|
|
SectionIndex: input.SectionIndex,
|
|
ValidatorOrder: &validatorOrder,
|
|
ReplacementPolicy: input.Policy,
|
|
Glossary: input.Glossary,
|
|
Config: input.Config,
|
|
LLMClient: input.ValidationLLMClient,
|
|
DiagnosticsWriter: diagnosticsWriter,
|
|
ReportDiagnosticError: input.ReportDiagnosticError,
|
|
})
|
|
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...)
|
|
|
|
var decisionErr error
|
|
if err != nil {
|
|
decisionErr = validators.EnforcePartialDecisionCardinality(eligible, vResult.Decisions)
|
|
} else {
|
|
decisionErr = validators.EnforceDecisionCardinality(eligible, vResult.Decisions)
|
|
}
|
|
if decisionErr != nil {
|
|
return validateSectionCandidatesResult{
|
|
decisions: decisions, rejected: rejected, operations: operations, filtered: filtered,
|
|
}, fmt.Errorf("validator %q cardinality failed: %w", validator.Name(), errors.Join(err, decisionErr))
|
|
}
|
|
|
|
nextEligible := make([]proposals.EnrichedCorrectionProposal, 0, len(eligible))
|
|
byIndex := make(map[int]proposals.EnrichedCorrectionProposal, len(eligible))
|
|
for _, p := range eligible {
|
|
byIndex[p.ProposalIndex] = p
|
|
}
|
|
|
|
for _, d := range vResult.Decisions {
|
|
decisions = append(decisions, ValidatorDecisionRecord{
|
|
ValidatorName: validator.Name(),
|
|
ProposalIndex: d.ProposalIndex,
|
|
Approved: d.Approved,
|
|
Disposition: d.Disposition,
|
|
ReasonCode: d.ReasonCode,
|
|
Message: d.Message,
|
|
DiagnosticArtifactPath: d.DiagnosticArtifactPath,
|
|
})
|
|
if d.Approved {
|
|
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(),
|
|
ProposalIndex: p.ProposalIndex,
|
|
ModuleKey: p.ModuleKey,
|
|
ModuleInstance: p.ModuleInstance,
|
|
TargetSegmentID: p.TargetSegmentID,
|
|
OriginalText: p.OriginalText,
|
|
CorrectedText: p.CorrectedText,
|
|
Disposition: d.Disposition,
|
|
ReasonCode: d.ReasonCode,
|
|
Message: d.Message,
|
|
})
|
|
}
|
|
|
|
if err != nil {
|
|
// Completed decisions are evidence only. A fatal return never grants
|
|
// approval to candidates that have not finished the full chain.
|
|
return validateSectionCandidatesResult{
|
|
decisions: decisions, rejected: rejected, operations: operations, filtered: filtered,
|
|
}, fmt.Errorf("validator %q failed: %w", validator.Name(), err)
|
|
}
|
|
eligible = nextEligible
|
|
}
|
|
|
|
return validateSectionCandidatesResult{
|
|
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))
|
|
for _, validator := range in {
|
|
if validatormetadata.ClassOf(validator) == validatormetadata.ExecutionClassLLMBacked {
|
|
llmBacked = append(llmBacked, validator)
|
|
continue
|
|
}
|
|
deterministic = append(deterministic, validator)
|
|
}
|
|
ordered := append(deterministic, llmBacked...)
|
|
order := make(map[string]int, len(ordered))
|
|
for idx, validator := range ordered {
|
|
order[validator.Name()] = idx
|
|
}
|
|
return ordered, order
|
|
}
|
|
|
|
func chunkWorkingTranscript(cfg *config.Config, transcript *schema.Transcript) ([]chunking.Section, error) {
|
|
if cfg == nil {
|
|
if transcript == nil || len(transcript.Segments) == 0 {
|
|
return []chunking.Section{}, nil
|
|
}
|
|
return []chunking.Section{
|
|
{
|
|
Index: 0,
|
|
EstimatedTokens: estimateTranscriptTokens(transcript),
|
|
Segments: append([]schema.Segment(nil), transcript.Segments...),
|
|
StartSegmentID: transcript.Segments[0].ID,
|
|
EndSegmentID: transcript.Segments[len(transcript.Segments)-1].ID,
|
|
},
|
|
}, nil
|
|
}
|
|
chunker := chunking.NewChunker(chunking.ChunkingConfig{
|
|
MaxSectionTokens: cfg.Chunking.MaxSectionTokens,
|
|
MinSectionTokens: cfg.Chunking.MinSectionTokens,
|
|
TargetSections: cfg.Chunking.TargetSections,
|
|
})
|
|
return chunker.ChunkTranscript(transcript)
|
|
}
|
|
|
|
func estimateTranscriptTokens(transcript *schema.Transcript) int {
|
|
if transcript == nil || len(transcript.Segments) == 0 {
|
|
return 0
|
|
}
|
|
estimator := chunking.NewSimpleTokenEstimator()
|
|
total := 0
|
|
for _, segment := range transcript.Segments {
|
|
total += estimator.EstimateTokens(segment.Text)
|
|
}
|
|
return total
|
|
}
|
|
|
|
func transcriptFromSection(section chunking.Section) *schema.Transcript {
|
|
segments := make([]schema.Segment, len(section.Segments))
|
|
copy(segments, section.Segments)
|
|
return &schema.Transcript{Segments: segments}
|
|
}
|
|
|
|
func cloneTranscript(t *schema.Transcript) *schema.Transcript {
|
|
if t == nil {
|
|
return &schema.Transcript{}
|
|
}
|
|
segments := make([]schema.Segment, len(t.Segments))
|
|
for i, s := range t.Segments {
|
|
var categories []string
|
|
if s.Categories != nil {
|
|
categories = append([]string(nil), s.Categories...)
|
|
}
|
|
segments[i] = schema.Segment{
|
|
ID: s.ID,
|
|
Speaker: s.Speaker,
|
|
Start: s.Start,
|
|
End: s.End,
|
|
Text: s.Text,
|
|
Categories: categories,
|
|
}
|
|
}
|
|
return &schema.Transcript{Segments: segments}
|
|
}
|
|
|
|
type llmDiagnosticsWriterAdapter struct {
|
|
writer *llm.DiagnosticsWriter
|
|
}
|
|
|
|
func (a *llmDiagnosticsWriterAdapter) WriteInteraction(stage string, requestMetadata any, requestPayload any, responsePayload any, errorPayload any) (validators.InteractionArtifacts, error) {
|
|
if a == nil || a.writer == nil {
|
|
return validators.InteractionArtifacts{}, fmt.Errorf("diagnostics writer is not configured")
|
|
}
|
|
art, err := a.writer.WriteInteraction(stage, requestMetadata, requestPayload, responsePayload, errorPayload)
|
|
if err != nil {
|
|
return validators.InteractionArtifacts{}, err
|
|
}
|
|
return validators.InteractionArtifacts{
|
|
RequestMetadataPath: art.RequestMetadataPath,
|
|
RequestPayloadPath: art.RequestPayloadPath,
|
|
ResponsePayloadPath: art.ResponsePayloadPath,
|
|
ErrorPayloadPath: art.ErrorPayloadPath,
|
|
}, nil
|
|
}
|