package runner import ( "context" "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/proposals" "gitea.maximumdirect.net/eric/audita/internal/framework/validators" ) 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 } type ValidationScheduler = contracts.LLMScheduler // 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"` 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"` } type ValidatorDecisionRecord struct { ValidatorName string `json:"validator_name"` ProposalIndex int `json:"proposal_index"` Approved bool `json:"approved"` 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"` ReasonCode string `json:"reason_code"` Message string `json:"message"` } // RunInput is the deterministic runner input. type RunInput struct { Config *config.Config Transcript *schema.Transcript Glossary *schema.Glossary ModuleSpecs []contracts.ModuleRunSpec EffectiveConcurrency EffectiveConcurrencyLimits ProposalLLMClient contracts.StructuredLLMClient ProposalLLMScheduler contracts.LLMScheduler ProposalDiagnosticsDir string ValidationLLMClient contracts.StructuredLLMClient ValidationLLMScheduler ValidationScheduler ValidationDiagnosticsDir string } // RunOutput is the deterministic runner output. type RunOutput struct { FinalTranscript *schema.Transcript `json:"-"` ModuleResults []ModuleResult `json:"module_results"` Utilization *UtilizationDiagnostics } func New(factory ModuleFactory) *Runner { return &Runner{factory: factory} } 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") } startedAt := time.Now().UTC() instrumentation := newRunInstrumentation(startedAt) input.ProposalLLMClient = instrumentation.wrapProposalClient(input.ProposalLLMClient) input.ValidationLLMClient = instrumentation.wrapValidationClient(input.ValidationLLMClient) input.ProposalLLMScheduler = instrumentation.wrapScheduler(input.ProposalLLMScheduler, true) input.ValidationLLMScheduler = instrumentation.wrapScheduler(input.ValidationLLMScheduler, false) working := cloneTranscript(input.Transcript) results := make([]ModuleResult, 0, len(input.ModuleSpecs)) for _, 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 RunOutput{ FinalTranscript: working, ModuleResults: results, Utilization: instrumentation.finalize(&RunOutput{ModuleResults: results}, time.Now().UTC(), input.EffectiveConcurrency), }, fmt.Errorf("module %q setup failed: %w", spec.InstanceName, err) } policy := module.ReplacementPolicy() sections, err := chunkWorkingTranscript(input.Config, working) if err != nil { failed := ModuleResult{ ModuleKey: spec.ModuleKey, ModuleInstance: spec.InstanceName, ReplacementPolicy: policy, Status: ModuleStatusFailed, ErrorMessage: err.Error(), StartedAt: startedAt, CompletedAt: time.Now().UTC(), } results = append(results, failed) return RunOutput{ FinalTranscript: working, ModuleResults: results, Utilization: instrumentation.finalize(&RunOutput{ModuleResults: results}, time.Now().UTC(), input.EffectiveConcurrency), }, fmt.Errorf("module %q chunking failed: %w", spec.InstanceName, err) } pipelineResult, pipelineErr := runModulePipeline(ctx, collectSectionProposalsInput{ Module: module, Spec: spec, Policy: policy, Config: input.Config, Glossary: input.Glossary, Sections: sections, ProposalClient: input.ProposalLLMClient, ProposalScheduler: input.ProposalLLMScheduler, DiagnosticsDir: input.ProposalDiagnosticsDir, ValidationClient: input.ValidationLLMClient, ValidationScheduler: input.ValidationLLMScheduler, ValidationDiagnosticsDir: input.ValidationDiagnosticsDir, Instrumentation: instrumentation, }) if pipelineErr != nil { failed := ModuleResult{ ModuleKey: spec.ModuleKey, ModuleInstance: spec.InstanceName, ReplacementPolicy: policy, Status: ModuleStatusFailed, ProposalCount: pipelineResult.ProposalCount, ValidatorDecisions: pipelineResult.ValidatorDecisions, ValidatorRejected: pipelineResult.ValidatorRejected, ErrorMessage: pipelineErr.Error(), StartedAt: startedAt, CompletedAt: time.Now().UTC(), } results = append(results, failed) return RunOutput{ FinalTranscript: working, ModuleResults: results, Utilization: instrumentation.finalize(&RunOutput{ModuleResults: results}, time.Now().UTC(), input.EffectiveConcurrency), }, fmt.Errorf("module %q failed: %w", spec.InstanceName, pipelineErr) } applyResult := proposals.ApplyProposals(working, pipelineResult.Approved, policy) working = applyResult.Transcript results = append(results, ModuleResult{ ModuleKey: spec.ModuleKey, ModuleInstance: spec.InstanceName, ReplacementPolicy: policy, Status: ModuleStatusSuccess, ProposalCount: pipelineResult.ProposalCount, ValidatorDecisions: pipelineResult.ValidatorDecisions, ValidatorRejected: pipelineResult.ValidatorRejected, AppliedChanges: applyResult.Applied, SkippedChanges: applyResult.Skipped, StartedAt: startedAt, CompletedAt: time.Now().UTC(), }) } output := RunOutput{FinalTranscript: working, ModuleResults: results} output.Utilization = instrumentation.finalize(&output, time.Now().UTC(), input.EffectiveConcurrency) return output, nil } type collectSectionProposalsInput struct { Module contracts.TranscriptModule Spec contracts.ModuleRunSpec Policy proposals.ReplacementPolicy Config *config.Config Glossary *schema.Glossary Sections []chunking.Section ProposalClient contracts.StructuredLLMClient ProposalScheduler contracts.LLMScheduler DiagnosticsDir string ValidationClient contracts.StructuredLLMClient ValidationScheduler ValidationScheduler ValidationDiagnosticsDir string Instrumentation *runInstrumentation } type sectionProposals struct { meta contracts.SectionMetadata corrected []proposals.CorrectionProposal } type sectionProposalResult struct { sectionPos int section chunking.Section corrected []proposals.CorrectionProposal err error } type sectionValidationResult struct { sectionPos int enriched []proposals.EnrichedCorrectionProposal approved []proposals.EnrichedCorrectionProposal decisions []ValidatorDecisionRecord rejected []ValidatorRejectedChange err error } type modulePipelineResult struct { ProposalCount int Approved []proposals.EnrichedCorrectionProposal 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() meta := contracts.SectionMetadataFromSection(section) corrected, err := input.Module.Propose(withModuleInstanceContext(runCtx, input.Spec.InstanceName), contracts.ProposalRequest{ ExecutionContext: contracts.ExecutionContext{ Config: input.Config, WorkingTranscript: transcriptFromSection(section), Glossary: input.Glossary, Section: &meta, DiagnosticsDir: input.DiagnosticsDir, }, RunSpec: contracts.ModuleRunSpec{ ModuleKey: input.Spec.ModuleKey, InstanceName: input.Spec.InstanceName, ReplacementPolicy: input.Policy, }, LLMClient: input.ProposalClient, LLMScheduler: input.ProposalScheduler, }) select { case results <- sectionProposalResult{ sectionPos: sectionPos, section: section, corrected: corrected, err: err, }: case <-runCtx.Done(): return } }() } wg.Wait() }() return runCtx, results, cancel } func runModulePipeline(ctx context.Context, input collectSectionProposalsInput) (modulePipelineResult, error) { out := modulePipelineResult{ Approved: make([]proposals.EnrichedCorrectionProposal, 0), ValidatorDecisions: make([]ValidatorDecisionRecord, 0), ValidatorRejected: make([]ValidatorRejectedChange, 0), } 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 validationLaunches := 0 var ( firstErr error errOnce sync.Once vwg sync.WaitGroup ) setErr := func(err error) { if err == nil { return } errOnce.Do(func() { firstErr = err cancel() }) } for result := range proposalResults { if result.err != nil { setErr(result.err) continue } if firstErr != nil { continue } pending[result.sectionPos] = result for { 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) validationLaunches++ 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), ModuleInstanceForStages: fmt.Sprintf("%s:section-%04d", input.Spec.InstanceName, sectionMetadata.Index), Validators: validatorsOrdered, SectionEnriched: enriched, ValidationLLMClient: input.ValidationClient, ValidationScheduler: input.ValidationScheduler, DiagnosticsDir: input.ValidationDiagnosticsDir, Instrumentation: input.Instrumentation, }) validationResults <- sectionValidationResult{ sectionPos: sectionPos, enriched: enriched, approved: validated.approved, decisions: validated.decisions, rejected: validated.rejected, 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 := 0; i < validationLaunches; i++ { res, ok := validationBySection[i] if !ok { break } out.Approved = append(out.Approved, res.approved...) out.ValidatorDecisions = append(out.ValidatorDecisions, res.decisions...) out.ValidatorRejected = append(out.ValidatorRejected, res.rejected...) } 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 firstErr != nil { return out, firstErr } return out, nil } type validateSectionCandidatesInput struct { Spec contracts.ModuleRunSpec Policy proposals.ReplacementPolicy Glossary *schema.Glossary Config *config.Config WorkingTranscript *schema.Transcript ModuleInstanceForStages string Validators []contracts.Validator SectionEnriched []proposals.EnrichedCorrectionProposal ValidationLLMClient contracts.StructuredLLMClient ValidationScheduler ValidationScheduler DiagnosticsDir string Instrumentation *runInstrumentation } type validateSectionCandidatesResult struct { approved []proposals.EnrichedCorrectionProposal decisions []ValidatorDecisionRecord rejected []ValidatorRejectedChange } func validateSectionCandidates(ctx context.Context, input validateSectionCandidatesInput) (validateSectionCandidatesResult, error) { decisions := make([]ValidatorDecisionRecord, 0) rejected := make([]ValidatorRejectedChange, 0) eligible := append([]proposals.EnrichedCorrectionProposal(nil), input.SectionEnriched...) for _, validator := range input.Validators { var diagnosticsWriter validators.InteractionDiagnosticsWriter if input.DiagnosticsDir != "" { diagnosticsWriter = &llmDiagnosticsWriterAdapter{ writer: llm.NewDiagnosticsWriter( filepath.Join(input.DiagnosticsDir, input.Spec.InstanceName), validatorSecrets(input.Config), ), } } validatorStartedAt := time.Now().UTC() vResult, err := validator.Validate(withModuleInstanceContext(ctx, input.Spec.InstanceName), contracts.ValidationRequest{ WorkingTranscript: input.WorkingTranscript, CandidateProposal: eligible, ModuleKey: input.Spec.ModuleKey, ModuleInstance: input.ModuleInstanceForStages, ReplacementPolicy: input.Policy, Glossary: input.Glossary, Config: input.Config, LLMClient: validationLLMClientAdapter{client: input.ValidationLLMClient}, Scheduler: input.ValidationScheduler, DiagnosticsWriter: diagnosticsWriter, }) if input.Instrumentation != nil { input.Instrumentation.recordValidatorTiming( input.Spec.ModuleKey, input.Spec.InstanceName, validator.Name(), isLLMBackedValidator(validator), time.Since(validatorStartedAt), ) } if err != nil { return validateSectionCandidatesResult{ approved: eligible, decisions: decisions, rejected: rejected, }, 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, }, fmt.Errorf("validator %q cardinality failed: %w", validator.Name(), err) } 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, ReasonCode: d.ReasonCode, Message: d.Message, DiagnosticArtifactPath: d.DiagnosticArtifactPath, }) if d.Approved { nextEligible = append(nextEligible, byIndex[d.ProposalIndex]) continue } 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, ReasonCode: d.ReasonCode, Message: d.Message, }) } eligible = nextEligible } return validateSectionCandidatesResult{ approved: eligible, decisions: decisions, rejected: rejected, }, nil } 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 _, ok := validator.(*validators.LLMBackedValidator); ok { 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.MaxSectionTokens, MinSectionTokens: cfg.MinSectionTokens, TargetSections: cfg.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 validationLLMClientAdapter struct { client contracts.StructuredLLMClient } func (a validationLLMClientAdapter) CompleteStructured(ctx context.Context, req validators.StructuredCompletionRequest, out any) (validators.StructuredCompletionResponse, error) { if a.client == nil { return validators.StructuredCompletionResponse{}, fmt.Errorf("validation structured LLM client is not configured") } messages := make([]contracts.LLMMessage, len(req.Messages)) for i, m := range req.Messages { messages[i] = contracts.LLMMessage{Role: m.Role, Content: m.Content} } resp, err := a.client.CompleteStructured(ctx, contracts.StructuredCompletionRequest{ StageName: req.StageName, Messages: messages, Model: req.Model, ResponseSchema: req.ResponseSchema, }, out) if err != nil { return validators.StructuredCompletionResponse{}, err } return validators.StructuredCompletionResponse{ PromptTokens: resp.PromptTokens, CompletionTokens: resp.CompletionTokens, TotalTokens: resp.TotalTokens, }, nil } 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 } func validatorSecrets(cfg *config.Config) []string { if cfg == nil { return nil } effective := cfg.EffectiveValidationLLMConfig() return []string{ cfg.PrimaryLLM.APIKey, effective.APIKey, cfg.ValidationLLM.APIKey, } }