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

Fix the five resilience review findings

Browse Source
This commit is contained in:
Eric Rakestraw
2026-09-24 21:48:18 +00:00
parent 9fc4de22b7
commit ec78b5a5db
19 changed files with 550 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
internal/cli/config_input.go
View File

@@ -212,7 +212,7 @@ func selectedReportDir(cfg config.Config, outputPath string) (string, error) {
if !cfg.Reporting.Enabled {
return "", nil
}
if strings.TrimSpace(outputPath) != "" {
if outputPath != "" {
return filepath.Dir(outputPath), nil
}
if strings.TrimSpace(cfg.Reporting.StdoutDir) == "" {

43
internal/cli/process.go
View File

@@ -10,6 +10,7 @@ import (
"io"
"path/filepath"
"strings"
"sync"
"time"
"gitea.maximumdirect.net/eric/audita/internal/core/chunking"
@@ -18,6 +19,7 @@ import (
coreio "gitea.maximumdirect.net/eric/audita/internal/core/io"
"gitea.maximumdirect.net/eric/audita/internal/core/normalization"
"gitea.maximumdirect.net/eric/audita/internal/core/outputschema"
"gitea.maximumdirect.net/eric/audita/internal/core/redaction"
"gitea.maximumdirect.net/eric/audita/internal/core/reporting"
"gitea.maximumdirect.net/eric/audita/internal/core/schema"
"gitea.maximumdirect.net/eric/audita/internal/framework/contracts"
@@ -131,6 +133,7 @@ type processState struct {
session *diagnostics.OutputSession
debugDir *diagnostics.RunDirectory
debugWarnings []string
debugMu sync.Mutex
runOutput *runner.RunOutput
}
@@ -183,7 +186,7 @@ func (s *processState) debugWrite(label string, write func(*diagnostics.RunDirec
return
}
if err := write(s.debugDir); err != nil {
s.debugWarnings = append(s.debugWarnings, "debug "+label+": "+llm.RedactError(err, s.cfg))
s.reportDiagnosticError(fmt.Errorf("%s: %w", label, err))
}
}
@@ -193,7 +196,7 @@ func (s *processState) startDebug() {
}
dir, err := diagnostics.NewRunDirectory(s.cfg.Diagnostics.WorkDir, s.runID, llm.ConfiguredSecrets(&s.cfg)...)
if err != nil {
s.debugWarnings = append(s.debugWarnings, "debug directory: "+llm.RedactError(err, s.cfg))
s.reportDiagnosticError(fmt.Errorf("directory: %w", err))
return
}
s.debugDir = dir
@@ -209,17 +212,21 @@ func (s *processState) startDebug() {
if err := s.selected.Result.WriteEffective(&effective); err == nil {
s.debugWrite("effective settings", func(d *diagnostics.RunDirectory) error { return d.WriteEffectiveConfig(effective.Bytes()) })
} else {
s.debugWarnings = append(s.debugWarnings, "debug effective settings: "+llm.RedactError(err, s.cfg))
s.reportDiagnosticError(fmt.Errorf("effective settings: %w", err))
}
if origins, err := llm.ExplanationYAML(s.selected.Result); err == nil {
s.debugWrite("configuration origins", func(d *diagnostics.RunDirectory) error {
return d.WriteYAMLArtifact(diagnostics.ArtifactConfigOrigins, origins)
})
} else {
s.reportDiagnosticError(fmt.Errorf("configuration origins: %w", err))
}
if profiles, err := s.runtime.SelectedProfilesYAML(); err == nil {
s.debugWrite("selected profiles", func(d *diagnostics.RunDirectory) error {
return d.WriteYAMLArtifact(diagnostics.ArtifactSelectedProfiles, profiles)
})
} else {
s.reportDiagnosticError(fmt.Errorf("selected profiles: %w", err))
}
}
@@ -272,7 +279,8 @@ func (s *processState) execute() int {
Config: &s.cfg, Transcript: normalized, Glossary: glossary, ModuleSpecs: s.specs,
ProposalLLMClient: proposalClient, ProposalDiagnosticsDir: s.debugPath(),
ValidationLLMClient: validationClient, ValidationDiagnosticsDir: s.debugPath(),
CaptureCandidates: s.debugDir != nil,
CaptureCandidates: s.debugDir != nil,
ReportDiagnosticError: s.reportDiagnosticError,
})
result.Observations = observations.Snapshot()
result.ObservationsAvailable = true
@@ -362,7 +370,7 @@ func (s *processState) fail(phase string, reason outcomes.ReasonCode, cause erro
s.printFailures(writeFailures)
s.printDebugWarnings()
if s.debugDir != nil {
fmt.Fprintf(s.stderr, "audita process: diagnostics: %s\n", s.debugDir.Path())
fmt.Fprintf(s.stderr, "audita process: diagnostics: %s\n", s.safePath(s.debugDir.Path()))
}
return 1
}
@@ -374,7 +382,7 @@ func (s *processState) finish(outputBytes []byte) int {
if err != nil {
return s.fail("report_write", outcomes.ReasonInternalFailure, err)
}
if s.outputPath == "" {
if s.session.TranscriptPath() == "" {
if _, err := s.stdout.Write(outputBytes); err != nil {
return s.fail("transcript_write", outcomes.ReasonInternalFailure, err)
}
@@ -391,16 +399,16 @@ func (s *processState) finish(outputBytes []byte) int {
s.printFailures(writeFailures)
s.printDebugWarnings()
if s.debugDir != nil {
fmt.Fprintf(s.stderr, "audita process: diagnostics: %s\n", s.debugDir.Path())
fmt.Fprintf(s.stderr, "audita process: diagnostics: %s\n", s.safePath(s.debugDir.Path()))
}
return 1
}
fmt.Fprintln(s.stderr, reporting.CompletionLine(snapshot))
for _, path := range s.session.ReportPaths() {
fmt.Fprintf(s.stderr, "Report: %s\n", path)
fmt.Fprintf(s.stderr, "Report: %s\n", s.safePath(path))
}
if s.debugDir != nil {
fmt.Fprintf(s.stderr, "Debug: %s\n", s.debugDir.Path())
fmt.Fprintf(s.stderr, "Debug: %s\n", s.safePath(s.debugDir.Path()))
}
s.printDebugWarnings()
return 0
@@ -425,7 +433,24 @@ func (s *processState) printFailures(failures []diagnostics.FileFailure) {
}
func (s *processState) printDebugWarnings() {
s.debugMu.Lock()
defer s.debugMu.Unlock()
for _, warning := range s.debugWarnings {
fmt.Fprintf(s.stderr, "audita process: %s\n", warning)
}
}
// safePath sanitizes presentation only; filesystem destinations remain literal.
func (s *processState) safePath(path string) string {
return redaction.New(llm.ConfiguredSecrets(&s.cfg)).String(path)
}
func (s *processState) reportDiagnosticError(err error) {
if err == nil {
return
}
message := "debug " + llm.RedactError(err, s.cfg)
s.debugMu.Lock()
defer s.debugMu.Unlock()
s.debugWarnings = append(s.debugWarnings, message)
}

28
internal/core/diagnostics/output_session.go
View File

@@ -50,7 +50,7 @@ type OutputSession struct {
func PrepareOutput(selection OutputSelection) (*OutputSession, error) {
session := &OutputSession{enabled: selection.ReportingEnabled, replace: atomicReplace}
active := make([]string, 0, 4)
if strings.TrimSpace(selection.TranscriptPath) != "" {
if selection.TranscriptPath != "" {
session.transcript = selection.TranscriptPath
active = append(active, session.transcript)
}
@@ -74,7 +74,7 @@ func PrepareOutput(selection OutputSelection) (*OutputSession, error) {
}
protected := make([]string, 0, len(selection.ProtectedPaths))
for _, path := range selection.ProtectedPaths {
if strings.TrimSpace(path) != "" {
if path != "" {
protected = append(protected, path)
}
}
@@ -251,18 +251,26 @@ func atomicReplace(path string, payload []byte) error {
if err != nil {
return err
}
parent := filepath.Dir(publishPath)
temp, err := os.CreateTemp(parent, ".audita-write-*")
// A private temporary directory lets ordinary file creation apply the umask
// without reading or changing process-global state. Rename stays on the same
// filesystem, and the incomplete payload is not accessible through the parent.
dir, err := os.MkdirTemp(filepath.Dir(publishPath), ".audita-write-*")
if err != nil {
return err
}
name := temp.Name()
defer os.Remove(name)
mode := os.FileMode(0644)
if info, err := os.Stat(publishPath); err == nil {
mode = info.Mode().Perm()
defer os.Remove(dir)
name := filepath.Join(dir, "payload")
temp, err := os.OpenFile(name, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0644)
if err != nil {
return err
}
if err := temp.Chmod(mode); err != nil {
defer os.Remove(name)
if info, err := os.Stat(publishPath); err == nil {
if err := temp.Chmod(info.Mode().Perm()); err != nil {
_ = temp.Close()
return err
}
} else if !errors.Is(err, os.ErrNotExist) {
_ = temp.Close()
return err
}

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

@@ -61,11 +61,13 @@ func SectionMetadataFromSection(section chunking.Section) SectionMetadata {
// ExecutionContext carries shared execution state for proposal generation and validation.
type ExecutionContext struct {
Config *config.Config `json:"-"`
WorkingTranscript *schema.Transcript `json:"-"`
Glossary *schema.Glossary `json:"-"`
Section *SectionMetadata `json:"section,omitempty"`
DiagnosticsDir string `json:"diagnostics_dir,omitempty"`
// ReportDiagnosticError may be called concurrently; it must not affect processing.
ReportDiagnosticError func(error) `json:"-"`
Config *config.Config `json:"-"`
WorkingTranscript *schema.Transcript `json:"-"`
Glossary *schema.Glossary `json:"-"`
Section *SectionMetadata `json:"section,omitempty"`
DiagnosticsDir string `json:"diagnostics_dir,omitempty"`
}
// ProposalRequest is the input to module proposal generation.

99
internal/framework/processreport/partial_validation_test.go Normal file
View File

@@ -0,0 +1,99 @@
package processreport
import (
"context"
"errors"
"strings"
"testing"
"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/outcomes"
"gitea.maximumdirect.net/eric/audita/internal/framework/proposals"
"gitea.maximumdirect.net/eric/audita/internal/framework/runner"
"gitea.maximumdirect.net/eric/audita/internal/framework/structuredoutput"
"gitea.maximumdirect.net/eric/audita/internal/framework/validators"
)
type partialModule struct{ validator contracts.Validator }
func (m partialModule) Key() string { return "grammar" }
func (m partialModule) ReplacementPolicy() proposals.ReplacementPolicy {
return proposals.ReplacementPolicyRequireUnique
}
func (m partialModule) Validators() []contracts.Validator { return []contracts.Validator{m.validator} }
func (m partialModule) ModuleForSpec(contracts.ModuleRunSpec) (contracts.TranscriptModule, error) {
return m, nil
}
func (m partialModule) Propose(context.Context, contracts.ProposalRequest) (contracts.ProposalResult, error) {
result := contracts.ProposalResult{}
for _, text := range []string{"Hello", "Greetings", "Hi"} {
result.Proposals = append(result.Proposals, proposals.CorrectionProposal{TargetSegmentID: 1, OriginalText: "hello", CorrectedText: text, Confidence: 1})
}
return result, nil
}
type oneItemEstimator struct{}
func (oneItemEstimator) EstimateTokens(payload string) int {
return 100 * strings.Count(payload, `"correction_index"`)
}
type interruptedValidationClient struct {
cause error
calls int
}
func (c *interruptedValidationClient) CompleteStructured(_ context.Context, req structuredoutput.Request, target any) (structuredoutput.Response, error) {
c.calls++
if *req.Context.BatchIndex == 2 {
return structuredoutput.Response{}, &structuredoutput.Failure{Class: structuredoutput.FailureFatal, Reason: outcomes.ReasonSetupFailure, Message: "validation interrupted", Cause: c.cause}
}
index := *req.Context.BatchIndex
*target.(*validators.LLMValidationResponse) = validators.LLMValidationResponse{Validations: []validators.LLMValidationDecision{{CorrectionIndex: index, Approved: index == 1, Confidence: 1, Reason: "completed assessment"}}}
return structuredoutput.Response{}, nil
}
func TestFatalBatchRetainsEarlierDecisionsInLedger(t *testing.T) {
validator, err := validators.NewLLMBackedValidator("editorial_review", validators.LLMValidatorTypeEditorialReview)
if err != nil {
t.Fatal(err)
}
validator.SetTokenEstimator(oneItemEstimator{})
origin := errors.New("later batch failed")
client := &interruptedValidationClient{cause: origin}
cfg := config.Config{Chunking: config.Chunking{MaxSectionTokens: 8192, MinSectionTokens: 1}, Validation: config.Validation{MaxPromptTokens: 150}}
result, err := runner.New(partialModule{validator}).Run(context.Background(), runner.RunInput{
Config: &cfg, Transcript: &schema.Transcript{Segments: []schema.Segment{{ID: 1, Text: "hello"}}},
ModuleSpecs: []contracts.ModuleRunSpec{{ModuleKey: "grammar", InstanceName: "grammar"}}, ValidationLLMClient: client, CaptureCandidates: true,
})
if !errors.Is(err, origin) || client.calls != 3 {
t.Fatalf("fatal cause/calls: %v %d", err, client.calls)
}
if result.FinalTranscript.Segments[0].Text != "hello" || len(result.ModuleResults[0].AppliedChanges) != 0 {
t.Fatal("fatal partial approvals changed transcript")
}
if len(result.WorkUnits[0].Operations) != 3 || result.WorkUnits[0].State != outcomes.WorkInterrupted {
t.Fatalf("partial facts: %+v", result.WorkUnits)
}
ledger := BuildCorrectionLedger(CorrectionLedgerInput{RunOutput: &result})
candidates := make(map[int]CorrectionLedgerEntry)
for _, entry := range ledger {
if entry.Disposition != correctionDispositionFailed {
candidates[entry.ProposalIndex] = entry
}
}
if len(candidates) != 3 {
t.Fatalf("candidate evidence: %+v", ledger)
}
for index := 0; index < 2; index++ {
entry := candidates[index]
if len(entry.LLMValidatorResults) != 1 || entry.LLMValidatorResults[0].Message != "completed assessment" || entry.LLMValidatorResults[0].Approved != (index == 1) {
t.Errorf("lost batch %d decision: %+v", index, entry)
}
}
if candidates[0].Disposition != correctionDispositionRejected || candidates[1].Disposition != correctionDispositionInterrupted || candidates[2].Disposition != correctionDispositionInterrupted {
t.Fatalf("wrong partial dispositions: %+v", ledger)
}
}

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

@@ -50,21 +50,23 @@ type StructuredCorrectionSet struct {
// Request captures reusable proposal-generation inputs for future modules.
type Request struct {
ModuleKey string `json:"module_key"`
ModuleInstance string `json:"module_instance"`
ReplacementPolicy proposals.ReplacementPolicy `json:"replacement_policy"`
WorkingTranscript *schema.Transcript `json:"-"`
Section *contracts.SectionMetadata `json:"section,omitempty"`
Glossary *schema.Glossary `json:"-"`
Config *config.Config `json:"-"`
PromptID string `json:"prompt_id"`
PromptVersion string `json:"prompt_version"`
Variables map[string]string `json:"variables"`
StageName string `json:"stage_name,omitempty"`
StartIndex int `json:"start_index"`
LLMClient contracts.StructuredLLMClient
DiagnosticsDir string
DiagnosticsWriter InteractionDiagnosticsWriter
// ReportDiagnosticError may be called concurrently; it must not affect processing.
ReportDiagnosticError func(error) `json:"-"`
ModuleKey string `json:"module_key"`
ModuleInstance string `json:"module_instance"`
ReplacementPolicy proposals.ReplacementPolicy `json:"replacement_policy"`
WorkingTranscript *schema.Transcript `json:"-"`
Section *contracts.SectionMetadata `json:"section,omitempty"`
Glossary *schema.Glossary `json:"-"`
Config *config.Config `json:"-"`
PromptID string `json:"prompt_id"`
PromptVersion string `json:"prompt_version"`
Variables map[string]string `json:"variables"`
StageName string `json:"stage_name,omitempty"`
StartIndex int `json:"start_index"`
LLMClient contracts.StructuredLLMClient
DiagnosticsDir string
DiagnosticsWriter InteractionDiagnosticsWriter
}
// Result contains generated candidate proposals and optional diagnostics paths.
@@ -145,13 +147,17 @@ func GenerateCandidates(ctx context.Context, req Request) (Result, error) {
requestMetadata["response_schema"] = responseschema.MustLookup(responseschema.CorrectionSetKey).DiagnosticsMap()
if writer != nil {
artifacts, _ = writer.WriteInteraction(
var diagnosticErr error
artifacts, diagnosticErr = writer.WriteInteraction(
stage,
requestMetadata,
completion.Prepared,
map[string]any{"decoded": response, "final_content": completion.FinalContent, "model": completion.Model},
errPayload(callErr),
)
if diagnosticErr != nil && req.ReportDiagnosticError != nil {
req.ReportDiagnosticError(fmt.Errorf("proposal interaction %s: %w", stage, diagnosticErr))
}
}
if callErr != nil {

16
internal/framework/proposal_generation/generate_test.go
View File

@@ -51,6 +51,7 @@ func (f *fakeStructuredClient) CompleteStructured(ctx context.Context, req contr
}
type captureDiagnosticsWriter struct {
err error
lastStage string
lastRequestMetadata any
lastRequestPayload any
@@ -62,7 +63,7 @@ func (w *captureDiagnosticsWriter) WriteInteraction(stage string, requestMetadat
w.lastRequestPayload = requestPayload
_ = responsePayload
_ = errorPayload
return InteractionArtifacts{}, nil
return InteractionArtifacts{}, w.err
}
func defaultRequest(t *testing.T) Request {
@@ -383,3 +384,16 @@ func TestGenerateCandidatesClientError(t *testing.T) {
t.Fatalf("expected error diagnostics artifact under %s", req.DiagnosticsDir)
}
}
func TestProposalDiagnosticFailureDoesNotChangeOperation(t *testing.T) {
cause := errors.New("diagnostic storage unavailable")
req := defaultRequest(t)
req.LLMClient = &fakeStructuredClient{responses: []StructuredCorrectionSet{{}}}
req.DiagnosticsWriter = &captureDiagnosticsWriter{err: cause}
var reported error
req.ReportDiagnosticError = func(err error) { reported = err }
result, err := GenerateCandidates(context.Background(), req)
if err != nil || !errors.Is(reported, cause) || len(result.Operations) != 1 || !result.Operations[0].Succeeded || len(result.Operations[0].RecoveryReasons) != 0 || result.Artifacts.ResponsePayloadPath != "" {
t.Fatalf("debug failure changed outcome: %+v %v reported=%v", result, err, reported)
}
}

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

@@ -55,20 +55,21 @@ func ExecuteModuleProposal(ctx context.Context, req ModuleProposalRequest) (cont
}
generated, err := GenerateCandidates(ctx, Request{
ModuleKey: req.ProposalRequest.RunSpec.ModuleKey,
ModuleInstance: req.ProposalRequest.RunSpec.InstanceName,
ReplacementPolicy: req.ProposalRequest.RunSpec.ReplacementPolicy,
WorkingTranscript: req.ProposalRequest.WorkingTranscript,
Section: req.ProposalRequest.Section,
Glossary: req.ProposalRequest.Glossary,
Config: req.ProposalRequest.Config,
PromptID: promptMetadata.PromptID,
PromptVersion: promptMetadata.PromptVersion,
Variables: variables,
StageName: stagename.ModuleProposal(req.ProposalRequest.RunSpec.InstanceName, sectionIndexPtr(req.ProposalRequest.Section)),
StartIndex: 0,
LLMClient: req.ProposalRequest.LLMClient,
DiagnosticsDir: req.ProposalRequest.DiagnosticsDir,
ModuleKey: req.ProposalRequest.RunSpec.ModuleKey,
ModuleInstance: req.ProposalRequest.RunSpec.InstanceName,
ReplacementPolicy: req.ProposalRequest.RunSpec.ReplacementPolicy,
WorkingTranscript: req.ProposalRequest.WorkingTranscript,
Section: req.ProposalRequest.Section,
Glossary: req.ProposalRequest.Glossary,
Config: req.ProposalRequest.Config,
PromptID: promptMetadata.PromptID,
PromptVersion: promptMetadata.PromptVersion,
Variables: variables,
StageName: stagename.ModuleProposal(req.ProposalRequest.RunSpec.InstanceName, sectionIndexPtr(req.ProposalRequest.Section)),
StartIndex: 0,
LLMClient: req.ProposalRequest.LLMClient,
DiagnosticsDir: req.ProposalRequest.DiagnosticsDir,
ReportDiagnosticError: req.ProposalRequest.ReportDiagnosticError,
})
if err != nil {
return contracts.ProposalResult{Operations: generated.Operations}, err

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

@@ -2,6 +2,7 @@ package runner
import (
"context"
"errors"
"fmt"
"path/filepath"
"sort"
@@ -86,6 +87,8 @@ type ValidatorRejectedChange struct {
// 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
@@ -188,6 +191,7 @@ func (r *Runner) run(ctx context.Context, input RunInput) (RunOutput, error) {
ValidationClient: input.ValidationLLMClient,
ValidationDiagnosticsDir: input.ValidationDiagnosticsDir,
Timings: input.timings,
ReportDiagnosticError: input.ReportDiagnosticError,
})
if input.CaptureCandidates {
candidates = append(candidates, pipelineResult.Candidates...)
@@ -293,6 +297,8 @@ func applyWithinSections(sections []chunking.Section, approved []proposals.Enric
}
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
@@ -356,11 +362,12 @@ func collectSectionProposals(ctx context.Context, input collectSectionProposalsI
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,
Config: input.Config,
WorkingTranscript: transcriptFromSection(section),
Glossary: input.Glossary,
Section: &meta,
DiagnosticsDir: input.DiagnosticsDir,
ReportDiagnosticError: input.ReportDiagnosticError,
},
RunSpec: contracts.ModuleRunSpec{
ModuleKey: input.Spec.ModuleKey,
@@ -495,17 +502,18 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput)
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: &sectionMetadata.Index,
Validators: validatorsOrdered,
SectionEnriched: enriched,
ValidationLLMClient: input.ValidationClient,
DiagnosticsDir: input.ValidationDiagnosticsDir,
Timings: input.Timings,
Spec: input.Spec,
Policy: input.Policy,
Glossary: input.Glossary,
Config: input.Config,
WorkingTranscript: transcriptFromSection(sectionResult.section),
SectionIndex: &sectionMetadata.Index,
Validators: validatorsOrdered,
SectionEnriched: enriched,
ValidationLLMClient: input.ValidationClient,
DiagnosticsDir: input.ValidationDiagnosticsDir,
Timings: input.Timings,
ReportDiagnosticError: input.ReportDiagnosticError,
})
setErr(err)
validationResults <- sectionValidationResult{
@@ -578,17 +586,19 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput)
}
type validateSectionCandidatesInput struct {
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
// 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 {
@@ -622,38 +632,34 @@ func validateSectionCandidates(ctx context.Context, input validateSectionCandida
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,
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 {
return validateSectionCandidatesResult{
decisions: decisions,
rejected: rejected,
operations: operations,
filtered: filtered,
}, fmt.Errorf("validator %q failed: %w", validator.Name(), err)
decisionErr = validators.EnforcePartialDecisionCardinality(eligible, vResult.Decisions)
} else {
decisionErr = validators.EnforceDecisionCardinality(eligible, vResult.Decisions)
}
if err := validators.EnforceDecisionCardinality(eligible, vResult.Decisions); err != nil {
if decisionErr != nil {
return validateSectionCandidatesResult{
decisions: decisions,
rejected: rejected,
operations: operations,
filtered: filtered,
}, fmt.Errorf("validator %q cardinality failed: %w", validator.Name(), err)
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))
@@ -694,6 +700,13 @@ func validateSectionCandidates(ctx context.Context, input validateSectionCandida
})
}
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
}

17
internal/framework/runner/runner_test.go
View File

@@ -1280,3 +1280,20 @@ func TestRunnerGlossaryModuleUsesConfidenceThreshold(t *testing.T) {
t.Fatalf("expected low confidence reason, got %+v", out.ModuleResults[0].ValidatorRejected[0])
}
}
func TestFatalPartialDecisionsRejectUnknownAndDuplicateIdentity(t *testing.T) {
origin := errors.New("original fatal failure")
candidate := proposals.EnrichedCorrectionProposal{ProposalMetadata: proposals.ProposalMetadata{ProposalIndex: 7}}
for _, decisions := range [][]validators.Decision{
{{ProposalIndex: 8, Approved: true}},
{{ProposalIndex: 7, Approved: true}, {ProposalIndex: 7, Approved: true}},
} {
v := fakeValidator{name: "broken", validateF: func(contracts.ValidationRequest) (validators.Result, error) {
return validators.Result{Decisions: decisions}, origin
}}
result, err := validateSectionCandidates(context.Background(), validateSectionCandidatesInput{Validators: []contracts.Validator{v}, SectionEnriched: []proposals.EnrichedCorrectionProposal{candidate}})
if !errors.Is(err, origin) || len(result.decisions) != 0 || len(result.approved) != 0 {
t.Fatalf("invalid partial evidence accepted or origin lost: %+v %v", result, err)
}
}
}

6
internal/framework/validators/llm_validators.go
View File

@@ -151,7 +151,8 @@ func (v *LLMBackedValidator) Validate(ctx context.Context, req Request) (Result,
artifacts := InteractionArtifacts{}
if req.DiagnosticsWriter != nil {
artifacts, _ = req.DiagnosticsWriter.WriteInteraction(
var diagnosticErr error
artifacts, diagnosticErr = req.DiagnosticsWriter.WriteInteraction(
stage,
map[string]any{
"validator_name": v.name,
@@ -164,6 +165,9 @@ func (v *LLMBackedValidator) Validate(ctx context.Context, req Request) (Result,
map[string]any{"decoded": response, "final_content": completion.FinalContent, "model": completion.Model},
errPayload(err),
)
if diagnosticErr != nil && req.ReportDiagnosticError != nil {
req.ReportDiagnosticError(fmt.Errorf("validation interaction %s: %w", stage, diagnosticErr))
}
}
if err != nil {
var failure *structuredoutput.Failure

17
internal/framework/validators/llm_validators_test.go
View File

@@ -23,6 +23,7 @@ type fakeStructuredLLMClient struct {
}
type captureValidationDiagnosticsWriter struct {
err error
lastRequestMetadata any
}
@@ -32,7 +33,7 @@ func (w *captureValidationDiagnosticsWriter) WriteInteraction(stage string, requ
_ = responsePayload
_ = errorPayload
w.lastRequestMetadata = requestMetadata
return InteractionArtifacts{}, nil
return InteractionArtifacts{}, w.err
}
func (f *fakeStructuredLLMClient) CompleteStructured(ctx context.Context, req StructuredCompletionRequest, out any) (StructuredCompletionResponse, error) {
@@ -321,3 +322,17 @@ func TestEveryValidationTypeSelectsEmbeddedPromptAndVariables(t *testing.T) {
})
}
}
func TestValidationDiagnosticFailureDoesNotChangeDecision(t *testing.T) {
cause := errors.New("diagnostic storage unavailable")
req := makeReq([]proposals.EnrichedCorrectionProposal{mk(0, "gestures", "Jesters")})
req.LLMClient = &fakeStructuredLLMClient{responses: []LLMValidationResponse{{Validations: []LLMValidationDecision{{CorrectionIndex: 0, Approved: true, Confidence: 1, Reason: "ok"}}}}}
req.DiagnosticsWriter = &captureValidationDiagnosticsWriter{err: cause}
var reported error
req.ReportDiagnosticError = func(err error) { reported = err }
validator, _ := NewLLMBackedValidator("editorial_review", LLMValidatorTypeEditorialReview)
result, err := validator.Validate(context.Background(), req)
if err != nil || !errors.Is(reported, cause) || len(result.Decisions) != 1 || !result.Decisions[0].Approved || result.Decisions[0].DiagnosticArtifactPath != "" || len(result.Operations) != 1 || !result.Operations[0].Succeeded || len(result.Operations[0].RecoveryReasons) != 0 {
t.Fatalf("debug failure changed outcome: %+v %v reported=%v", result, err, reported)
}
}

39
internal/framework/validators/models.go
View File

@@ -29,17 +29,19 @@ const (
// Request is the runtime input shared by deterministic validators.
type Request struct {
SectionIndex *int `json:"section_index,omitempty"`
WorkingTranscript *schema.Transcript `json:"-"`
CandidateProposal []proposals.EnrichedCorrectionProposal `json:"candidate_proposals"`
ModuleKey string `json:"module_key"`
ModuleInstance string `json:"module_instance"`
ValidatorOrder *int `json:"validator_order,omitempty"`
ReplacementPolicy proposals.ReplacementPolicy `json:"replacement_policy"`
Glossary *schema.Glossary `json:"-"`
Config *config.Config `json:"-"`
LLMClient StructuredLLMClient `json:"-"`
DiagnosticsWriter InteractionDiagnosticsWriter `json:"-"`
// ReportDiagnosticError may be called concurrently; it must not affect processing.
ReportDiagnosticError func(error) `json:"-"`
SectionIndex *int `json:"section_index,omitempty"`
WorkingTranscript *schema.Transcript `json:"-"`
CandidateProposal []proposals.EnrichedCorrectionProposal `json:"candidate_proposals"`
ModuleKey string `json:"module_key"`
ModuleInstance string `json:"module_instance"`
ValidatorOrder *int `json:"validator_order,omitempty"`
ReplacementPolicy proposals.ReplacementPolicy `json:"replacement_policy"`
Glossary *schema.Glossary `json:"-"`
Config *config.Config `json:"-"`
LLMClient StructuredLLMClient `json:"-"`
DiagnosticsWriter InteractionDiagnosticsWriter `json:"-"`
}
// Decision is one validator decision for one proposal index.
@@ -62,7 +64,8 @@ const (
DispositionUnavailable DecisionDisposition = "unavailable"
)
// Result is one validator output containing exactly one decision per proposal index.
// Result contains one decision per proposal index on success. Fatal returns may
// include completed partial decisions as evidence, never as application approval.
type Result struct {
ValidatorName string `json:"validator_name"`
Decisions []Decision `json:"decisions"`
@@ -88,6 +91,16 @@ type Validator interface {
// EnforceDecisionCardinality verifies every candidate proposal index receives exactly one decision.
func EnforceDecisionCardinality(candidate []proposals.EnrichedCorrectionProposal, decisions []Decision) error {
return enforceDecisionCardinality(candidate, decisions, true)
}
// EnforcePartialDecisionCardinality accepts missing decisions on fatal returns,
// but still rejects duplicate and unknown proposal identities.
func EnforcePartialDecisionCardinality(candidate []proposals.EnrichedCorrectionProposal, decisions []Decision) error {
return enforceDecisionCardinality(candidate, decisions, false)
}
func enforceDecisionCardinality(candidate []proposals.EnrichedCorrectionProposal, decisions []Decision, requireAll bool) error {
expected := make(map[int]struct{}, len(candidate))
for _, c := range candidate {
expected[c.ProposalIndex] = struct{}{}
@@ -105,7 +118,7 @@ func EnforceDecisionCardinality(candidate []proposals.EnrichedCorrectionProposal
}
for idx := range expected {
if seen[idx] == 0 {
if requireAll && seen[idx] == 0 {
return fmt.Errorf("missing decision proposal index %d", idx)
}
}