From ec78b5a5db3e1041ddafe4add1d0eb0283c633b3 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Thu, 24 Sep 2026 21:48:18 +0000 Subject: [PATCH] Fix the five resilience review findings --- cmd/audita/main_integration_test.go | 152 ++++++++++++++++++ docs/cli.md | 2 + docs/internal/diagnostics-reporting.md | 5 +- docs/internal/pipeline.md | 7 +- docs/operations.md | 9 +- docs/roadmap/remediation.md | 48 +++++- internal/cli/config_input.go | 2 +- internal/cli/process.go | 43 +++-- internal/core/diagnostics/output_session.go | 28 ++-- internal/framework/contracts/contracts.go | 12 +- .../processreport/partial_validation_test.go | 99 ++++++++++++ .../framework/proposal_generation/generate.go | 38 +++-- .../proposal_generation/generate_test.go | 16 +- .../proposal_generation/module_proposal.go | 29 ++-- internal/framework/runner/runner.go | 113 +++++++------ internal/framework/runner/runner_test.go | 17 ++ .../framework/validators/llm_validators.go | 6 +- .../validators/llm_validators_test.go | 17 +- internal/framework/validators/models.go | 39 +++-- 19 files changed, 550 insertions(+), 132 deletions(-) create mode 100644 internal/framework/processreport/partial_validation_test.go diff --git a/cmd/audita/main_integration_test.go b/cmd/audita/main_integration_test.go index 55eb315..c5dc2ef 100644 --- a/cmd/audita/main_integration_test.go +++ b/cmd/audita/main_integration_test.go @@ -11,6 +11,7 @@ import ( "os/exec" "os/signal" "path/filepath" + "strconv" "strings" "sync" "syscall" @@ -28,6 +29,13 @@ func TestProcessSubprocessHelper(t *testing.T) { } for i, arg := range os.Args { if arg == "--" { + if value := os.Getenv("AUDITA_TEST_UMASK"); value != "" { + mask, err := strconv.ParseUint(value, 8, 32) + if err != nil { + os.Exit(2) + } + syscall.Umask(int(mask)) + } ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) code := cli.Run(ctx, os.Args[i+1:], os.Stdout, os.Stderr) stop() @@ -45,6 +53,7 @@ type fixtureProvider struct { mode string secret string entered chan struct{} + beforeResponse func() } func (p *fixtureProvider) serve(w http.ResponseWriter, r *http.Request) { @@ -71,7 +80,11 @@ func (p *fixtureProvider) serve(w http.ResponseWriter, r *http.Request) { } validationCall := p.validationCalls proposalCall := p.proposalCalls + beforeResponse := p.beforeResponse p.mu.Unlock() + if beforeResponse != nil { + beforeResponse() + } if p.mode == "wait-cancel" { select { case p.entered <- struct{}{}: @@ -166,6 +179,7 @@ func (f processFixture) run(t *testing.T, includeOutput bool, extra ...string) ( } args = append(args, extra...) command := exec.Command(os.Args[0], args...) + command.Dir = f.dir command.Env = append(os.Environ(), "AUDITA_HELPER_PROCESS=1", "AUDITA_TEST_PROVIDER_KEY="+f.provider.secret) var stdout, stderr bytes.Buffer command.Stdout, command.Stderr = &stdout, &stderr @@ -486,3 +500,141 @@ func TestSubprocessTranscriptPreflightWithoutReporting(t *testing.T) { t.Fatalf("transcript preflight: code=%d calls=%d stderr=%q", code, f.provider.callCount(), stderr) } } + +func TestSubprocessOutputPermissionsHonorUmask(t *testing.T) { + for _, mask := range []string{"0077", "0022"} { + t.Run(mask, func(t *testing.T) { + t.Setenv("AUDITA_TEST_UMASK", mask) + f := newProcessFixture(t, "normal", "[]") + paths := []string{f.output, filepath.Join(f.dir, "errors.yml"), filepath.Join(f.dir, "recoveries.yml"), filepath.Join(f.dir, "report.md")} + parsed, _ := strconv.ParseUint(mask, 8, 32) + want := os.FileMode(0644 &^ parsed) + for attempt := 0; attempt < 2; attempt++ { + code, _, stderr := f.run(t, true) + if code != 0 { + t.Fatalf("run: %d %s", code, stderr) + } + for _, path := range paths { + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + if info.Mode().Perm() != want { + t.Errorf("%s mode=%o want=%o", filepath.Base(path), info.Mode().Perm(), want) + } + if err := os.Chmod(path, 0600); err != nil { + t.Fatal(err) + } + } + want = 0600 // Existing restrictive modes survive replacement under either mask. + } + }) + } +} + +func TestSubprocessWhitespaceOutputIsALiteralFile(t *testing.T) { + f := newProcessFixture(t, "normal", "[]") + f.output = " " + code, stdout, stderr := f.run(t, true) + if code != 0 || stdout != "" { + t.Fatalf("code=%d stdout=%q stderr=%q", code, stdout, stderr) + } + output, err := os.ReadFile(filepath.Join(f.dir, " ")) + if err != nil || !bytes.Contains(output, []byte("hello world")) { + t.Fatalf("missing transcript: %s %v", output, err) + } + if f.errors(t).Status != reporting.StatusCompleted { + t.Fatal("reports did not follow file output") + } +} + +func TestSubprocessPathAnnouncementsAreRedacted(t *testing.T) { + for _, fail := range []bool{false, true} { + t.Run(fmt.Sprint(fail), func(t *testing.T) { + f := newProcessFixture(t, "normal", "[]") + selected, unused := "selected-path-secret", "unused-path-secret" + f.provider.secret = selected + t.Setenv("AUDITA_UNUSED_PATH_KEY", unused) + cfg, err := os.ReadFile(f.config) + if err != nil { + t.Fatal(err) + } + cfg = bytes.Replace(cfg, []byte("backends:\n"), []byte("backends:\n unused:\n endpoint: http://unused.invalid/v1\n credential: AUDITA_UNUSED_PATH_KEY\n"), 1) + if err := os.WriteFile(f.config, cfg, 0600); err != nil { + t.Fatal(err) + } + reportDir := filepath.Join(f.dir, unused) + if err := os.Mkdir(reportDir, 0700); err != nil { + t.Fatal(err) + } + f.output = filepath.Join(reportDir, "out.json") + debugParent := filepath.Join(f.dir, selected) + if fail { + if err := os.WriteFile(f.transcript, []byte("invalid"), 0600); err != nil { + t.Fatal(err) + } + } + code, _, stderr := f.run(t, true, "--debug", "--work-dir", debugParent) + if (!fail && code != 0) || (fail && code != 1) { + t.Fatalf("code=%d stderr=%q", code, stderr) + } + if strings.Contains(stderr, selected) || strings.Contains(stderr, unused) { + t.Fatalf("credential in stderr: %q", stderr) + } + if !strings.Contains(stderr, "[REDACTED]") { + t.Fatalf("missing redacted announcement: %q", stderr) + } + if _, err := os.Stat(filepath.Join(reportDir, "report.md")); err != nil { + t.Fatal(err) + } + dirs, err := os.ReadDir(debugParent) + if err != nil || len(dirs) != 1 { + t.Fatalf("debug location: %v %v", dirs, err) + } + if !fail { + if _, err := os.Stat(f.output); err != nil { + t.Fatal(err) + } + } + }) + } +} + +func TestSubprocessInteractionWriteFailureIsVisibleAndNonfatal(t *testing.T) { + f := newProcessFixture(t, "normal", "[grammar]") + code, _, baselineSummary := f.run(t, true) + if code != 0 { + t.Fatal(baselineSummary) + } + baseline, err := os.ReadFile(f.output) + if err != nil { + t.Fatal(err) + } + f.provider.mu.Lock() + f.provider.beforeResponse = func() { + dirs, err := filepath.Glob(filepath.Join(f.dir, "runs", "run-*")) + if err != nil || len(dirs) != 1 { + t.Errorf("debug directory: %v %v", dirs, err) + return + } + if err := os.WriteFile(filepath.Join(dirs[0], "grammar"), []byte("block interaction writes"), 0600); err != nil { + t.Error(err) + } + } + f.provider.mu.Unlock() + code, _, stderr := f.run(t, true, "--debug") + if code != 0 || !strings.HasPrefix(stderr, strings.SplitN(baselineSummary, "\n", 2)[0]) { + t.Fatalf("changed outcome: %d %q", code, stderr) + } + if !strings.Contains(stderr, "debug proposal interaction") || !strings.Contains(stderr, "debug validation interaction") { + t.Fatalf("missing debug failures: %q", stderr) + } + output, err := os.ReadFile(f.output) + if err != nil || !bytes.Equal(output, baseline) { + t.Fatalf("changed transcript: %s %v", output, err) + } + document := f.errors(t) + if len(document.Errors) != 0 || *document.Summary.Recoveries != 0 { + t.Fatalf("debug failure classified as model incident: %+v", document) + } +} diff --git a/docs/cli.md b/docs/cli.md index cc70d5c..12ab86e 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -45,6 +45,8 @@ validation. The shared setting flags are: | Debug | `--debug[=true\|false]`, `--work-dir DIR` | `process` alone accepts `--glossary` (required) and `--output`. +An omitted or empty output value selects stdout. Nonempty output filenames are +used literally, including whitespace. Reporting defaults to enabled. `--reporting=false` disables all three ordinary reports; an explicit false overrides a true file or overlay setting. With file output, reports go beside the transcript and `--report-dir` has no effect. With diff --git a/docs/internal/diagnostics-reporting.md b/docs/internal/diagnostics-reporting.md index 177452d..1d24aa7 100644 --- a/docs/internal/diagnostics-reporting.md +++ b/docs/internal/diagnostics-reporting.md @@ -29,7 +29,10 @@ when debug is off. With debug on, source/normalized snapshots, invocation, effective settings, origins, profiles, initial chunk summary, correction ledger, indexed execution/utilization, and model interactions are written. Runner candidate capture includes proposals later rejected or interrupted. Debug -write failures produce safe stderr messages and do not abort processing. +write failures reach the CLI through an explicit concurrency-safe diagnostic +callback and produce redacted stderr messages without aborting processing. +Completed validator decisions returned with a fatal batch error remain available +to the debug ledger; they never authorize applying an incompletely validated edit. Every debug writer uses all configured backend secrets for redaction, including unselected backend credentials and source snapshots. Debug can still contain sensitive transcript and prompt content. diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index 5db45db..099a871 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -14,9 +14,10 @@ formats; [configuration](../config.md) owns settings; the 4. Module instances execute in configured order. Sections may propose concurrently, but proposal indices follow section order and response order. 5. Deterministic validators precede LLM-backed validators. Every submitted - candidate receives one decision per validator. Recoverable failed batches - mark their candidates unavailable; successful survivors continue through - all required validators. + candidate receives one decision per validator on a normal return. Fatal returns + may preserve a checked subset of completed decisions as diagnostic evidence. + Recoverable failed batches mark their candidates unavailable; successful + survivors continue through all required validators. 6. Only fully approved proposals reach application. Corrections apply in proposal-index order within their owning section. Existing replacement policies and apply-time safety checks resolve stale conflicts. diff --git a/docs/operations.md b/docs/operations.md index a5ec4bd..b794a70 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -28,11 +28,12 @@ Audita checks active destinations against input, configuration, and selected profile files before model dispatch. It preflights file transcript output without truncating an existing file. With reporting enabled, all three reports are initialized as `in_progress` before input processing. Output is encoded -fully before publication. Each file is replaced through a temporary file in -its destination directory: transcript first, then `errors.yml`, +fully before publication. Each file is replaced through a temporary file on +its destination filesystem: transcript first, then `errors.yml`, `recoveries.yml`, and `report.md`. This is per-file replacement, not a -multi-file transaction. A late report error may leave the transcript published; -the process exits nonzero and attempts failed-status reports. On fatal processing +multi-file transaction. New files honor the process umask; replacing an existing +file preserves its permission bits. A late report error may leave the transcript +published; the process exits nonzero and attempts failed-status reports. On fatal processing or cancellation, no completed transcript is published, but partial reports are finalized where possible. An uncatchable termination can leave `in_progress`. diff --git a/docs/roadmap/remediation.md b/docs/roadmap/remediation.md index ea4b2b5..e916806 100644 --- a/docs/roadmap/remediation.md +++ b/docs/roadmap/remediation.md @@ -4,7 +4,8 @@ **Scope:** The fourteen implementation commits after `8ffbf41`, compared with [implementation.md](implementation.md), [resilience.md](resilience.md), and the current architecture and output contracts. -**Status:** Five open findings, R6–R10. This review implements no fixes. +**Status:** R6–R10 addressed and verified. The findings below describe the reviewed +source; their line references are historical. See the resolution record below. The earlier library-adoption findings R1–R5 were resolved before this work. Their findings and resolution evidence remain in the @@ -192,8 +193,49 @@ one assembled CLI case is sufficient. Debug-off must still create no artifacts. - A temporary runner probe reproduced R9; its expected evidence-preservation assertion failed. The probe and all temporary artifacts were removed. -Passing retained tests do not cover the five failures above. Add proportionate -regression coverage with their fixes. This review changes only this document; +At review time, the retained tests did not cover the five failures above. +The initial review changed only this document; application fixes, upstream changes, deferred feature work, and release publication remain outside scope. Target-platform release builds were not repeated for this review; the native temporary CLI build succeeded. + + +## Resolution Record + +All five findings are addressed without changing retry budgets, library ownership, +correction acceptance, or the independent reporting/debug selections. + +- **R6:** Per-file replacement uses ordinary file creation inside a private + temporary directory in the destination parent. Creation applies the umask; + existing permission bits are restored only for existing destinations. Temporary + files/directories are removed on success and failure. The subprocess permission + test covers both restrictive and ordinary masks and subsequent replacement of + restrictive files; existing output-session failure/cleanup tests still pass. +- **R7:** File selection uses the literal nonempty output path throughout report + placement and output preparation. Final CLI stream selection uses the prepared + output session. `TestSubprocessWhitespaceOutputIsALiteralFile` proves the output + and reports are produced; the existing stdout integration test remains passing. +- **R8:** Every CLI report/debug location announcement uses the shared configured- + secret string redactor. `TestSubprocessPathAnnouncementsAreRedacted` covers + success and failure with selected and unselected credentials in real paths, + while confirming that the real destinations are used. +- **R9:** Fatal validator returns use partial identity/cardinality validation, + retain completed decisions and rejection details, and never return application + approvals. `TestFatalBatchRetainsEarlierDecisionsInLedger` exercises the real + runner and batching validator with two completed batches and a third fatal + failure, retaining their decisions and all operation facts in the debug ledger. + A separate runner regression rejects unknown/duplicate partial identities while + retaining the originating failure. Existing cancellation/joining tests pass. +- **R10:** An explicit diagnostic-error callback connects proposal and validator + writers to the CLI's synchronized, redacted nonfatal diagnostics. Origin/profile + serialization failures use the same path. Narrow proposal/validator tests and + `TestSubprocessInteractionWriteFailureIsVisibleAndNonfatal` verify visible + failures, unchanged transcript/counts/exit status, and no fabricated artifact + paths. Existing reporting/debug-selection tests verify debug-off persistence + remains disabled. + +Validation completed: full `go test ./...`, `go vet ./...`, focused race tests for +runner, validators, proposal generation, report assembly, diagnostics/output, CLI, +and subprocess integration; required Linux amd64/arm64 builds in a temporary +output directory; and documentation link/whitespace checks. No live provider, +upstream change, tag, or publication was involved. diff --git a/internal/cli/config_input.go b/internal/cli/config_input.go index 2dfc128..65ea32a 100644 --- a/internal/cli/config_input.go +++ b/internal/cli/config_input.go @@ -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) == "" { diff --git a/internal/cli/process.go b/internal/cli/process.go index a4f4ccb..ce73bd6 100644 --- a/internal/cli/process.go +++ b/internal/cli/process.go @@ -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) +} diff --git a/internal/core/diagnostics/output_session.go b/internal/core/diagnostics/output_session.go index 3385432..edb1ef1 100644 --- a/internal/core/diagnostics/output_session.go +++ b/internal/core/diagnostics/output_session.go @@ -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 } diff --git a/internal/framework/contracts/contracts.go b/internal/framework/contracts/contracts.go index 53c8cd4..f6248bc 100644 --- a/internal/framework/contracts/contracts.go +++ b/internal/framework/contracts/contracts.go @@ -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. diff --git a/internal/framework/processreport/partial_validation_test.go b/internal/framework/processreport/partial_validation_test.go new file mode 100644 index 0000000..800e310 --- /dev/null +++ b/internal/framework/processreport/partial_validation_test.go @@ -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) + } +} diff --git a/internal/framework/proposal_generation/generate.go b/internal/framework/proposal_generation/generate.go index 07e70d2..21b84a5 100644 --- a/internal/framework/proposal_generation/generate.go +++ b/internal/framework/proposal_generation/generate.go @@ -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 { diff --git a/internal/framework/proposal_generation/generate_test.go b/internal/framework/proposal_generation/generate_test.go index be4f329..fc85e3a 100644 --- a/internal/framework/proposal_generation/generate_test.go +++ b/internal/framework/proposal_generation/generate_test.go @@ -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) + } +} diff --git a/internal/framework/proposal_generation/module_proposal.go b/internal/framework/proposal_generation/module_proposal.go index f7f77a0..d873187 100644 --- a/internal/framework/proposal_generation/module_proposal.go +++ b/internal/framework/proposal_generation/module_proposal.go @@ -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 diff --git a/internal/framework/runner/runner.go b/internal/framework/runner/runner.go index 1df90e7..51f3ada 100644 --- a/internal/framework/runner/runner.go +++ b/internal/framework/runner/runner.go @@ -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: §ionMetadata.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: §ionMetadata.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 } diff --git a/internal/framework/runner/runner_test.go b/internal/framework/runner/runner_test.go index 2c3d357..72071d3 100644 --- a/internal/framework/runner/runner_test.go +++ b/internal/framework/runner/runner_test.go @@ -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) + } + } +} diff --git a/internal/framework/validators/llm_validators.go b/internal/framework/validators/llm_validators.go index 1f672bf..4687d2f 100644 --- a/internal/framework/validators/llm_validators.go +++ b/internal/framework/validators/llm_validators.go @@ -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 diff --git a/internal/framework/validators/llm_validators_test.go b/internal/framework/validators/llm_validators_test.go index 2220a83..a7e2a90 100644 --- a/internal/framework/validators/llm_validators_test.go +++ b/internal/framework/validators/llm_validators_test.go @@ -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) + } +} diff --git a/internal/framework/validators/models.go b/internal/framework/validators/models.go index 59f47c7..1a4185e 100644 --- a/internal/framework/validators/models.go +++ b/internal/framework/validators/models.go @@ -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) } }