diff --git a/cmd/audita/main_integration_test.go b/cmd/audita/main_integration_test.go index 1141592..e6320b4 100644 --- a/cmd/audita/main_integration_test.go +++ b/cmd/audita/main_integration_test.go @@ -35,16 +35,20 @@ type outboundCall struct { } type controlledProvider struct { - mu sync.Mutex - calls []outboundCall - proposalCount int - mode string - secret string + mu sync.Mutex + calls []outboundCall + proposalCount int + mode string + secret string + unselectedSecret string } func (p *controlledProvider) serve(w http.ResponseWriter, r *http.Request) { var request struct { - Model string `json:"model"` + Model string `json:"model"` + Messages []struct { + Content string `json:"content"` + } `json:"messages"` ResponseFormat struct { JSONSchema struct { Schema struct { @@ -65,7 +69,7 @@ func (p *controlledProvider) serve(w http.ResponseWriter, r *http.Request) { } proposalCount := p.proposalCount p.mu.Unlock() - if p.mode == "failure" { + if p.mode == "failure" || (p.mode == "secret-partial-failure" && !validation && proposalCount > 1) { w.WriteHeader(http.StatusInternalServerError) _, _ = fmt.Fprintf(w, `{"error":{"message":"provider echoed %s"}}`, p.secret) return @@ -86,6 +90,53 @@ func (p *controlledProvider) serve(w http.ResponseWriter, r *http.Request) { if p.mode == "low-confidence" && !validation { response = `{"corrections":[{"id":1,"original_text":"hello world","corrected_text":"Hello world","confidence":0.1}]}` } + if validation && strings.HasPrefix(p.mode, "secret-") { + payload := map[string]any{"validations": []any{map[string]any{"correction_index": 0, "approved": false, "confidence": 0.99, "reason": "echo:" + p.secret + " unused:" + p.unselectedSecret}}} + raw, _ := json.Marshal(payload) + response = string(raw) + } + if p.mode == "sections" { + for _, message := range request.Messages { + marker := "Transcript section:" + if validation { + marker = "Corrections to validate:" + } + _, payload, found := strings.Cut(message.Content, marker) + if !found { + continue + } + var items []map[string]any + var parseErr error + if validation { + parseErr = json.Unmarshal([]byte(strings.TrimSpace(payload)), &items) + } else { + var section struct { + Segments []map[string]any `json:"segments"` + } + parseErr = json.Unmarshal([]byte(strings.TrimSpace(payload)), §ion) + items = section.Segments + } + if parseErr != nil { + http.Error(w, parseErr.Error(), 500) + return + } + + results := []map[string]any{} + for _, item := range items { + if validation { + results = append(results, map[string]any{"correction_index": item["correction_index"], "approved": false, "confidence": 0.99, "reason": "test rejection"}) + } else { + results = append(results, map[string]any{"id": item["id"], "original_text": item["text"], "corrected_text": strings.ToUpper(item["text"].(string)), "confidence": 0.99}) + } + } + key := "corrections" + if validation { + key = "validations" + } + raw, _ := json.Marshal(map[string]any{key: results}) + response = string(raw) + } + } content, _ := json.Marshal(response) w.Header().Set("Content-Type", "application/json") _, _ = fmt.Fprintf(w, `{"model":"synthetic-result","choices":[{"message":{"content":%s}}],"usage":{"prompt_tokens":2,"completion_tokens":3,"total_tokens":5}}`, content) @@ -139,7 +190,7 @@ func (f processFixture) run(t *testing.T, extra ...string) (int, string, string, args := []string{"-test.run=^TestProcessSubprocessHelper$", "--", "process", f.transcript, "--config", f.config, "--glossary", f.glossary, "--report-json", f.report, "--output", f.output} args = append(args, extra...) cmd := exec.Command(os.Args[0], args...) - cmd.Env = append(os.Environ(), "AUDITA_HELPER_PROCESS=1", "AUDITA_TEST_PROVIDER_KEY="+f.provider.secret) + cmd.Env = append(os.Environ(), "AUDITA_HELPER_PROCESS=1", "AUDITA_TEST_PROVIDER_KEY="+f.provider.secret, "AUDITA_TEST_UNUSED_KEY="+f.provider.unselectedSecret) var stdout, stderr bytes.Buffer cmd.Stdout, cmd.Stderr = &stdout, &stderr err := cmd.Run() @@ -276,3 +327,203 @@ func TestSubprocessWarningEmptyRepeatedAndFailure(t *testing.T) { } }) } + +func TestSubprocessRedactsValidModelReasonsOnSuccessAndPartialFailure(t *testing.T) { + for _, mode := range []string{"secret-rejection", "secret-partial-failure"} { + t.Run(mode, func(t *testing.T) { + f := newProcessFixture(t, mode, "[grammar, grammar]") + f.provider.secret = `selected-credential-"\-123` + f.provider.unselectedSecret = `unused-credential-"\-456` + raw, err := os.ReadFile(f.config) + if err != nil { + t.Fatal(err) + } + raw = []byte(strings.Replace(string(raw), "backends:\n", "backends:\n unused:\n endpoint: "+f.server.URL+"/v1\n credential: AUDITA_TEST_UNUSED_KEY\n", 1)) + if err := os.WriteFile(f.config, raw, 0600); err != nil { + t.Fatal(err) + } + code, _, stderr, report := f.run(t) + if mode == "secret-rejection" && (code != 0 || report.Status != "success") { + t.Fatalf("code=%d %s", code, stderr) + } + if mode == "secret-partial-failure" { + if code == 0 || report.Status != "failed" { + t.Fatalf("expected failure: %d %+v", code, report) + } + if _, err := os.Stat(f.output); !os.IsNotExist(err) { + t.Fatalf("failure wrote complete output: %v", err) + } + } + if len(report.ModuleResults) == 0 || len(report.ModuleResults[0].ValidatorRejected) == 0 { + t.Fatalf("missing rejected domain result: %+v", report) + } + reason := report.ModuleResults[0].ValidatorRejected[0].Message + if !strings.Contains(reason, "[REDACTED]") { + t.Fatalf("reason not exercised: %s", reason) + } + sawLedger := false + sawInteraction := false + check := func(path string) { + t.Helper() + raw, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + var walk func(any) + walk = func(value any) { + switch v := value.(type) { + case string: + for _, secret := range []string{f.provider.secret, f.provider.unselectedSecret} { + escaped, _ := json.Marshal(secret) + if strings.Contains(v, secret) || strings.Contains(v, string(escaped[1:len(escaped)-1])) { + t.Fatalf("credential in %s", path) + } + } + case []any: + for _, item := range v { + walk(item) + } + case map[string]any: + for key, item := range v { + walk(key) + walk(item) + } + } + } + if strings.HasSuffix(path, ".json") { + var value any + if err := json.Unmarshal(raw, &value); err != nil { + t.Fatal(err) + } + walk(value) + } else { + walk(string(raw)) + } + if strings.HasSuffix(path, "correction-ledger.json") { + sawLedger = true + if !strings.Contains(string(raw), "[REDACTED]") { + t.Fatal("ledger reason not exercised") + } + } + if strings.HasSuffix(path, "response-payload.json") { + sawInteraction = true + } + } + check(f.report) + err = filepath.WalkDir(filepath.Join(f.dir, "runs"), func(path string, entry os.DirEntry, err error) error { + if err != nil { + return err + } + if !entry.IsDir() { + check(path) + } + return nil + }) + if err != nil { + t.Fatal(err) + } + if !sawLedger || !sawInteraction { + t.Fatal("missing diagnostic artifacts") + } + }) + } +} + +func TestSubprocessRepeatedModulesKeepSectionAttribution(t *testing.T) { + f := newProcessFixture(t, "sections", "[grammar, grammar]") + raw, err := os.ReadFile(f.config) + if err != nil { + t.Fatal(err) + } + raw = append(raw, []byte("chunking:\n max_section_tokens: 3\n min_section_tokens: 1\n")...) + if err := os.WriteFile(f.config, raw, 0600); err != nil { + t.Fatal(err) + } + input := `[{"id":1,"speaker":"A","start":0,"end":1,"text":"hello world"},{"id":2,"speaker":"B","start":2,"end":3,"text":"good morning"}]` + if err := os.WriteFile(f.transcript, []byte(input), 0600); err != nil { + t.Fatal(err) + } + code, _, stderr, report := f.run(t) + if code != 0 { + t.Fatalf("process: %d %s", code, stderr) + } + if len(report.ModuleResults) != 2 { + t.Fatalf("module results: %+v", report) + } + dirs, err := filepath.Glob(filepath.Join(f.dir, "runs", "*")) + if err != nil || len(dirs) != 1 { + t.Fatalf("run directories: %v %v", dirs, err) + } + raw, err = os.ReadFile(filepath.Join(dirs[0], "utilization-diagnostics.json")) + if err != nil { + t.Fatal(err) + } + var utilization reporting.UtilizationDiagnostics + if err := json.Unmarshal(raw, &utilization); err != nil { + t.Fatal(err) + } + if len(utilization.Modules) != 2 { + t.Fatalf("spurious modules: %+v", utilization.Modules) + } + for i, module := range utilization.Modules { + if module.ModuleInstance != report.ModuleResults[i].ModuleInstance || module.ProposalLLMCalls != 2 || module.ValidationLLMCalls != 2 { + t.Fatalf("module attribution: %+v", utilization.Modules) + } + } + raw, err = os.ReadFile(filepath.Join(dirs[0], "llm-execution.json")) + if err != nil { + t.Fatal(err) + } + var decoded struct { + Calls []struct { + Context struct { + Role string `json:"role"` + ModuleInstance string `json:"module_instance"` + SectionIndex *int `json:"section_index"` + BatchIndex *int `json:"batch_index"` + } `json:"context"` + } `json:"calls"` + } + if err := json.Unmarshal(raw, &decoded); err != nil { + t.Fatal(err) + } + seen := map[string]map[int]bool{} + for _, call := range decoded.Calls { + c := call.Context + if c.Role != "validation" { + continue + } + if c.SectionIndex == nil || c.BatchIndex == nil || *c.BatchIndex != 0 { + t.Fatalf("missing section/batch: %+v", c) + } + if seen[c.ModuleInstance] == nil { + seen[c.ModuleInstance] = map[int]bool{} + } + seen[c.ModuleInstance][*c.SectionIndex] = true + } + for _, module := range utilization.Modules { + if len(seen[module.ModuleInstance]) != 2 { + t.Fatalf("section attribution: %+v", seen) + } + } + responseCount := 0 + if err := filepath.WalkDir(dirs[0], func(path string, entry os.DirEntry, err error) error { + if err != nil { + return err + } + if !entry.IsDir() && strings.Contains(path, "section-") && strings.Contains(path, "batch-") && strings.HasSuffix(path, "response-payload.json") { + responseCount++ + } + return nil + }); err != nil { + t.Fatal(err) + } + if responseCount != 4 { + t.Fatalf("validation artifact collision: %d", responseCount) + } + for _, module := range report.ModuleResults { + if len(module.ValidatorRejected) != 2 || module.ValidatorRejected[0].ProposalIndex != 0 || module.ValidatorRejected[1].ProposalIndex != 1 { + t.Fatalf("decision order: %+v", module.ValidatorRejected) + } + } +} diff --git a/docs/internal/diagnostics-reporting.md b/docs/internal/diagnostics-reporting.md index ededbff..3c128c0 100644 --- a/docs/internal/diagnostics-reporting.md +++ b/docs/internal/diagnostics-reporting.md @@ -63,15 +63,28 @@ artifacts and process exit status must be assessed together. ## Redaction -Effective configuration uses ConfigKit redaction. LLM diagnostic and error paths -use `llm.ConfiguredSecrets` and the runtime's redaction helpers. The CLI also -redacts every resolved backend secret from both report copies. Keep these separate -mechanisms aligned with the [architecture requirement](../policy/architecture.md#errors-and-secret-handling). +Effective configuration uses ConfigKit redaction. The CLI supplies all resolved +backend credentials, including unselected backends, to report and diagnostic +writers. `core/redaction` sanitizes the serialized representation rather than +mutating processing results. JSON string values and keys are redacted after +parsing with number preservation; YAML redaction preserves scalar types. +Serialization failure returns an error without writing the original object. +Interaction artifacts, correction ledgers, execution/utilization diagnostics, +selected profiles, both report copies, and surfaced LLM errors share this +string-redaction implementation. Longer overlapping values take precedence, +and replacement text is not processed again within a redaction pass. + +Execution reports index events by call ID. Utilization groups calls directly by +backend and actual module instance, keeping module result order and retaining +usage knownness/completeness. Section and batch indices remain separate from +module identity; section-qualified interaction filenames avoid collisions. +These boundaries implement the [architecture requirement](../policy/architecture.md#errors-and-secret-handling). Operational handling of transcript-bearing artifacts belongs to [operations](../operations.md#artifact-handling). ## Key Tests +- `internal/core/redaction/redaction_test.go` - `internal/core/diagnostics/*_test.go` - `internal/framework/processreport/*_test.go` - `internal/core/reporting/report_test.go` diff --git a/docs/roadmap/remediation.md b/docs/roadmap/remediation.md index 7bea293..43fb160 100644 --- a/docs/roadmap/remediation.md +++ b/docs/roadmap/remediation.md @@ -4,7 +4,9 @@ **Scope:** The twelve implementation commits after `1ad2119`, compared with the [implementation plan](implementation.md), [combined target](library-adoption.md), and [configuration target](configkit.md). -**Status:** Findings confirmed; fixes have not been implemented. +**Status:** R1–R5 addressed and verified on 2026-09-24. The evidence in the +original findings below describes source commit `7cefb3f`; line references are +historical. See the resolution record at the end for the implemented fixes. ## Architecture Assessment @@ -190,10 +192,98 @@ wall-clock timing assertions to the default test suite. The existing root `go test ./...`, focused race tests for LLM/runner/subprocess packages, `go vet ./...`, and Linux amd64/arm64 builds passed. Temporary review probes reproduced R1–R4 and measured R5; those probe files were removed afterward. -The retained suite currently passes despite these defects. Add durable regression +At review time, the retained suite passed despite these defects. Add durable regression coverage with the fixes, using the narrowest useful production boundaries. -Only this remediation document is changed by the review. No application fixes, +The initial review changed only this remediation document. No application fixes, upstream changes, live provider calls, release publication, or compatibility work were performed. Release-distribution compliance remains the explicitly deferred release task and is not counted as an implementation defect here. + +## Resolution Record + +All five findings are resolved without compatibility layers, new scheduling, +or changes to library ownership. + +### R1 and R2 — Shared Serialization Boundary + +`core/redaction` now sanitizes decoded JSON strings and YAML string nodes, +retaining numeric/boolean types and avoiding mutation of domain objects. +Longest matching credentials take precedence in a single replacement pass; +JSON-escaped credential forms are handled for embedded response content. +Serialization or key-collision failures return no payload. The unredacted report +fallback has been removed. + +The CLI supplies all resolved backend credentials to the external report writer +and run directory. Report copies, ledgers, execution/utilization artifacts, +interaction diagnostics, selected profiles, and error text share the redactor. +ConfigKit still owns credential-reference redaction of effective configuration. + +Evidence: + +- `TestSubprocessRedactsValidModelReasonsOnSuccessAndPartialFailure` exercises + schema-valid negative assessments containing selected and unselected secrets, + including quotes/backslashes. Both reports and retained artifacts are checked; + partial failure preserves earlier decisions and writes no corrected output. +- `TestReportWritersRedactStringsAndFailClosed` exercises both actual report + writers with credential `2`, preserving numeric configuration version 2 while + redacting string content, and verifies failed serialization writes no artifact. +- `TestStringRedactionPreservesSerializationAndDomainData` covers overlapping + secrets, escaped nested content, large numbers, booleans, YAML scalar types, + unchanged domain data, and failures without an original-payload fallback. + +### R3 — Immediate Shared Cancellation + +Validation workers record their error and cancel shared work immediately. +All accesses to the originating-error state use a mutex. Proposal collection +stops launching validation after a recorded failure; active workers are joined +and available results retain deterministic collection order. + +`TestValidatorFailureCancelsActiveAndPendingProviderWork` uses the real PromptKit +runtime with backend concurrency one and queue capacity one. A failing validator +cancels two peer executions, one active in a controlled transport and one pending, +without manually releasing either. The originating error survives, only one +provider attempt occurs, and both logical calls retain finished observations. +The runner and runtime pass the race detector. + +### R4 — Independent Module and Section Identity + +Validation requests carry the configured module instance and an explicit section +index. The structured call forwards both; only interaction filenames combine +module, section, validator, and batch into a diagnostic name. The runner test +helper now uses section index rather than batch index to identify sections. + +`TestSubprocessRepeatedModulesKeepSectionAttribution` runs two grammar instances +across two sections through the actual CLI/runtime. It verifies exactly two +utilization entries with two proposal and two validation calls each, explicit +section/batch identity, four distinct validation response artifacts, and unchanged +proposal ordering in rejected decisions. + +### R5 — Indexed Report Aggregation + +Execution mapping indexes events by call ID once. Utilization directly groups +calls/events by backend and indexes modules by real instance, seeded in module +result order. Neither builder scans all events for each logical call. + +Existing summary tests and `TestIndexedReportsPreserveOrderAndPartialUsage` +verify pre-dispatch failures, original call/event order, backend grouping, and +known-but-incomplete usage. `BenchmarkExecutionReports` retains a reproducible +1,000/10,000/20,000-call benchmark without timing assertions. + +A local three-iteration sample measured 20,000 calls at approximately **43 ms** +for execution mapping and **84 ms** for utilization, compared with the review +probe's 8.08 s and 2.42 s respectively. Timings are illustrative; the direct +indexes establish removal of the quadratic scans independently of machine load. + +### Completion Validation + +- Full offline suite: `go test ./...`. +- Race checks: runner, LLM runtime, process reporting, redaction, and CLI subprocess + packages. +- Static analysis: `go vet ./...`. +- Both required Linux architectures: `scripts/build-release.sh` in a temporary + output directory, without tagging or publishing. +- Documentation/diff whitespace checks. + +Release publication and deferred release-distribution work remain outside this +remediation unit. diff --git a/internal/cli/run.go b/internal/cli/run.go index 42ea9b7..ee6872d 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -3,7 +3,6 @@ package cli import ( "bytes" "context" - "encoding/json" "fmt" "io" "strings" @@ -40,7 +39,7 @@ type processInvocation struct { } var processRunner = func(inv processInvocation, stdout io.Writer) (*normalization.NormalizationSummary, *chunking.Summary, *runner.RunOutput, *diagnostics.RunDirectory, error) { - runDir, err := diagnostics.NewRunDirectory(inv.Config.Diagnostics.WorkDir, string(inv.Config.Diagnostics.Retention)) + runDir, err := diagnostics.NewRunDirectory(inv.Config.Diagnostics.WorkDir, string(inv.Config.Diagnostics.Retention), llm.ConfiguredSecrets(&inv.Config)...) if err != nil { return nil, nil, nil, nil, fmt.Errorf("run_dir_creation: %w", err) } @@ -74,12 +73,12 @@ var processRunner = func(inv processInvocation, stdout io.Writer) (*normalizatio _ = runDir.WriteErrorLog(fmt.Sprintf("effective_config: %v", err)) } if origins, err := llm.ExplanationYAML(inv.ConfigResult); err == nil { - _ = runDir.WriteArtifact(diagnostics.ArtifactConfigOrigins, origins) + _ = runDir.WriteYAMLArtifact(diagnostics.ArtifactConfigOrigins, origins) } } if inv.LLMRuntime != nil { if profiles, err := inv.LLMRuntime.SelectedProfilesYAML(); err == nil { - _ = runDir.WriteArtifact(diagnostics.ArtifactSelectedProfiles, profiles) + _ = runDir.WriteYAMLArtifact(diagnostics.ArtifactSelectedProfiles, profiles) } } @@ -320,10 +319,9 @@ func runProcess(args []string, stdout, stderr io.Writer) int { errorPhase, errorMessage := extractErrorPhase(runErr) errorMessage = llm.RedactError(fmt.Errorf("%s", errorMessage), inv.Config) report := processreport.Build(processReportInput("failed", inv, runDir, startedAt, completedAt, errorMessage, errorPhase, nil, nil, runOutput)) - report = redactProcessReport(report, inv.Config) if strings.TrimSpace(inv.ReportJSONPath) != "" { - if err := reporting.WriteProcessReport(inv.ReportJSONPath, report); err != nil { + if err := reporting.WriteProcessReport(inv.ReportJSONPath, report, llm.ConfiguredSecrets(&inv.Config)...); err != nil { fmt.Fprintf(stderr, "audita process: %v\n", err) } } @@ -352,10 +350,9 @@ func runProcess(args []string, stdout, stderr io.Writer) int { } report := processreport.Build(processReportInput("success", inv, runDir, startedAt, completedAt, "", "", normSummary, chunkSummary, runOutput)) - report = redactProcessReport(report, inv.Config) if strings.TrimSpace(inv.ReportJSONPath) != "" { - if err := reporting.WriteProcessReport(inv.ReportJSONPath, report); err != nil { + if err := reporting.WriteProcessReport(inv.ReportJSONPath, report, llm.ConfiguredSecrets(&inv.Config)...); err != nil { if runDir != nil { _ = runDir.WriteErrorLog(fmt.Sprintf("report_write: %v", err)) _ = runDir.ApplyRetention(diagnostics.RetentionDecisionInput{ @@ -389,24 +386,6 @@ func writeExecutionArtifacts(runDir *diagnostics.RunDirectory, output *runner.Ru _ = runDir.WriteJSONArtifact(diagnostics.ArtifactUtilizationSummary, processreport.BuildUtilization(output, cfg.Backends)) } -func redactProcessReport(report reporting.ProcessReport, cfg config.Config) reporting.ProcessReport { - raw, err := json.Marshal(report) - if err != nil { - return report - } - content := string(raw) - for _, secret := range llm.ConfiguredSecrets(&cfg) { - content = strings.ReplaceAll(content, secret, "[REDACTED]") - encoded, _ := json.Marshal(secret) - content = strings.ReplaceAll(content, string(encoded[1:len(encoded)-1]), "[REDACTED]") - } - var safe reporting.ProcessReport - if json.Unmarshal([]byte(content), &safe) != nil { - return report - } - return safe -} - func runConfig(args []string, stdout, stderr io.Writer) int { if len(args) == 0 || isHelpCommand(args) || args[0] == "--help" || args[0] == "-h" { writeConfigUsage(stdout) diff --git a/internal/core/diagnostics/run_dir.go b/internal/core/diagnostics/run_dir.go index 73e6ff6..e9022d8 100644 --- a/internal/core/diagnostics/run_dir.go +++ b/internal/core/diagnostics/run_dir.go @@ -1,7 +1,6 @@ package diagnostics import ( - "encoding/json" "fmt" "os" "path/filepath" @@ -9,12 +8,14 @@ import ( "gitea.maximumdirect.net/eric/audita/internal/core/chunking" "gitea.maximumdirect.net/eric/audita/internal/core/normalization" + "gitea.maximumdirect.net/eric/audita/internal/core/redaction" "gitea.maximumdirect.net/eric/audita/internal/core/reporting" "gitea.maximumdirect.net/eric/audita/internal/core/schema" ) // RunDirectory represents a per-run diagnostics directory type RunDirectory struct { + redactor redaction.Redactor path string retention string createdAt time.Time @@ -63,7 +64,7 @@ type InvocationMetadata struct { } // NewRunDirectory creates a new run directory under the configured work dir -func NewRunDirectory(workDir, retention string) (*RunDirectory, error) { +func NewRunDirectory(workDir, retention string, secrets ...string) (*RunDirectory, error) { if workDir == "" { workDir = ".audita-runs" } @@ -83,6 +84,7 @@ func NewRunDirectory(workDir, retention string) (*RunDirectory, error) { return &RunDirectory{ path: runPath, + redactor: redaction.New(secrets), retention: retention, createdAt: time.Now().UTC(), }, nil @@ -107,7 +109,7 @@ func (r *RunDirectory) WriteInvocationMetadata(metadata InvocationMetadata) erro } path := filepath.Join(r.path, ArtifactInvocationMetadata) - bytes, err := json.MarshalIndent(metadata, "", " ") + bytes, err := r.redactor.JSON(metadata) if err != nil { return fmt.Errorf("failed to marshal invocation metadata: %w", err) } @@ -120,11 +122,7 @@ func (r *RunDirectory) WriteInvocationMetadata(metadata InvocationMetadata) erro // WriteEffectiveConfig writes redacted effective config metadata for this run. func (r *RunDirectory) WriteEffectiveConfig(effectiveYAML []byte) error { - path := filepath.Join(r.path, ArtifactEffectiveConfig) - if err := os.WriteFile(path, effectiveYAML, 0o644); err != nil { - return fmt.Errorf("failed to write effective config: %w", err) - } - return nil + return r.WriteYAMLArtifact(ArtifactEffectiveConfig, effectiveYAML) } // WriteSourceTranscript writes the source transcript artifact @@ -137,7 +135,7 @@ func (r *RunDirectory) WriteSourceTranscript(transcript *schema.SourceTranscript // Write parsed source for debugging parsedPath := filepath.Join(r.path, ArtifactParsedSourceTranscript) - parsedBytes, err := json.MarshalIndent(transcript, "", " ") + parsedBytes, err := r.redactor.JSON(transcript) if err != nil { return fmt.Errorf("failed to marshal parsed source transcript: %w", err) } @@ -165,7 +163,7 @@ func (r *RunDirectory) WriteNormalizedTranscript(transcript *schema.Transcript) // WriteNormalizationSummary writes the normalization summary artifact func (r *RunDirectory) WriteNormalizationSummary(summary *normalization.NormalizationSummary) error { summaryPath := filepath.Join(r.path, ArtifactNormalizationSummary) - bytes, err := json.MarshalIndent(summary, "", " ") + bytes, err := r.redactor.JSON(summary) if err != nil { return fmt.Errorf("failed to marshal normalization summary: %w", err) } @@ -179,7 +177,7 @@ func (r *RunDirectory) WriteNormalizationSummary(summary *normalization.Normaliz // WriteReport writes the authoritative report artifact func (r *RunDirectory) WriteReport(report reporting.ProcessReport) error { reportPath := filepath.Join(r.path, ArtifactReport) - bytes, err := json.MarshalIndent(report, "", " ") + bytes, err := r.redactor.JSON(report) if err != nil { return fmt.Errorf("failed to marshal report: %w", err) } @@ -193,13 +191,13 @@ func (r *RunDirectory) WriteReport(report reporting.ProcessReport) error { // WriteErrorLog writes an error log on failure func (r *RunDirectory) WriteErrorLog(errorMessage string) error { errorPath := filepath.Join(r.path, ArtifactErrorLog) - return os.WriteFile(errorPath, []byte(errorMessage+"\n"), 0o644) + return os.WriteFile(errorPath, []byte(r.redactor.String(errorMessage)+"\n"), 0o644) } // WriteChunkingSummary writes the chunking summary artifact func (r *RunDirectory) WriteChunkingSummary(summary *chunking.DetailedSummary) error { summaryPath := filepath.Join(r.path, ArtifactChunkingSummary) - bytes, err := json.MarshalIndent(summary, "", " ") + bytes, err := r.redactor.JSON(summary) if err != nil { return fmt.Errorf("failed to marshal chunking summary: %w", err) } @@ -212,7 +210,7 @@ func (r *RunDirectory) WriteChunkingSummary(summary *chunking.DetailedSummary) e func (r *RunDirectory) WriteJSONArtifact(name string, payload any) error { artifactPath := filepath.Join(r.path, name) - bytes, err := json.MarshalIndent(payload, "", " ") + bytes, err := r.redactor.JSON(payload) if err != nil { return fmt.Errorf("failed to marshal %s: %w", name, err) } @@ -223,7 +221,13 @@ func (r *RunDirectory) WriteJSONArtifact(name string, payload any) error { return nil } -func (r *RunDirectory) WriteArtifact(name string, payload []byte) error { +// WriteYAMLArtifact writes a sanitized YAML diagnostic artifact. +func (r *RunDirectory) WriteYAMLArtifact(name string, payload []byte) error { + var err error + payload, err = r.redactor.YAML(payload) + if err != nil { + return err + } if err := os.WriteFile(filepath.Join(r.path, name), payload, 0o644); err != nil { return fmt.Errorf("failed to write %s: %w", name, err) } diff --git a/internal/core/redaction/redaction.go b/internal/core/redaction/redaction.go new file mode 100644 index 0000000..378ee1f --- /dev/null +++ b/internal/core/redaction/redaction.go @@ -0,0 +1,123 @@ +// Package redaction sanitizes diagnostic representations without modifying domain data. +package redaction + +import ( + "bytes" + "encoding/json" + "fmt" + "sort" + "strings" + + "gopkg.in/yaml.v3" +) + +// Redactor replaces configured values in a single pass, longest matches first. +// Its zero value is usable and leaves strings unchanged. +type Redactor struct{ replacer *strings.Replacer } + +func New(secrets []string) Redactor { + values := map[string]bool{} + for _, secret := range secrets { + if secret == "" { + continue + } + values[secret] = true + // Provider content may itself be serialized JSON inside a string field. + encoded, _ := json.Marshal(secret) + values[string(encoded[1:len(encoded)-1])] = true + } + keys := make([]string, 0, len(values)) + for value := range values { + keys = append(keys, value) + } + sort.Slice(keys, func(i, j int) bool { + if len(keys[i]) == len(keys[j]) { + return keys[i] < keys[j] + } + return len(keys[i]) > len(keys[j]) + }) + pairs := make([]string, 0, 2*len(keys)) + for _, key := range keys { + pairs = append(pairs, key, "[REDACTED]") + } + return Redactor{replacer: strings.NewReplacer(pairs...)} +} + +func (r Redactor) String(value string) string { + if r.replacer == nil { + return value + } + return r.replacer.Replace(value) +} + +// JSON serializes first, then redacts only decoded string values and keys. +// Numbers retain their exact representation. Failure never returns original data. +func (r Redactor) JSON(value any) ([]byte, error) { + raw, err := json.Marshal(value) + if err != nil { + return nil, fmt.Errorf("serialize diagnostic: %s", r.String(err.Error())) + } + decoder := json.NewDecoder(bytes.NewReader(raw)) + decoder.UseNumber() + var decoded any + if err := decoder.Decode(&decoded); err != nil { + return nil, fmt.Errorf("decode diagnostic: %s", r.String(err.Error())) + } + safe, err := r.walk(decoded) + if err != nil { + return nil, err + } + return json.MarshalIndent(safe, "", " ") +} + +func (r Redactor) walk(value any) (any, error) { + switch v := value.(type) { + case string: + return r.String(v), nil + case []any: + for i := range v { + var err error + v[i], err = r.walk(v[i]) + if err != nil { + return nil, err + } + } + case map[string]any: + safe := make(map[string]any, len(v)) + for key, item := range v { + safeKey := r.String(key) + if _, exists := safe[safeKey]; exists { + return nil, fmt.Errorf("redaction would produce duplicate object keys") + } + safeItem, err := r.walk(item) + if err != nil { + return nil, err + } + safe[safeKey] = safeItem + } + return safe, nil + } + return value, nil +} + +// YAML preserves scalar types while sanitizing string nodes, including map keys. +func (r Redactor) YAML(raw []byte) ([]byte, error) { + var node yaml.Node + if err := yaml.Unmarshal(raw, &node); err != nil { + return nil, fmt.Errorf("decode diagnostic YAML: %s", r.String(err.Error())) + } + var visit func(*yaml.Node) + visit = func(n *yaml.Node) { + if n.Kind == yaml.ScalarNode && n.Tag == "!!str" { + n.Value = r.String(n.Value) + } + n.HeadComment = r.String(n.HeadComment) + n.LineComment = r.String(n.LineComment) + n.FootComment = r.String(n.FootComment) + for _, child := range n.Content { + visit(child) + } + } + visit(&node) + return yaml.Marshal(&node) +} diff --git a/internal/core/redaction/redaction_test.go b/internal/core/redaction/redaction_test.go new file mode 100644 index 0000000..c32e851 --- /dev/null +++ b/internal/core/redaction/redaction_test.go @@ -0,0 +1,92 @@ +package redaction_test + +import ( + "encoding/json" + "math" + "os" + "path/filepath" + "strings" + "testing" + + "gitea.maximumdirect.net/eric/audita/internal/core/diagnostics" + "gitea.maximumdirect.net/eric/audita/internal/core/redaction" + "gitea.maximumdirect.net/eric/audita/internal/core/reporting" +) + +func TestStringRedactionPreservesSerializationAndDomainData(t *testing.T) { + secrets := []string{"2", `quoted"\key`, "overlap", "overlap-long", "REDACTED"} + r := redaction.New(secrets) + original := map[string]any{"number": 2, "large": json.Number("9007199254740993"), "bool": true, "text": `echo:2 quoted"\key overlap-long overlap`} + raw, err := r.JSON(original) + if err != nil { + t.Fatal(err) + } + var decoded map[string]any + if err := json.Unmarshal(raw, &decoded); err != nil { + t.Fatal(err) + } + if decoded["number"] != float64(2) || decoded["bool"] != true || !strings.Contains(string(raw), "9007199254740993") { + t.Fatalf("changed scalar meanings: %s", raw) + } + if decoded["text"] != "echo:[REDACTED] [REDACTED] [REDACTED] [REDACTED]" { + t.Fatalf("redaction: %s", raw) + } + if original["text"] != `echo:2 quoted"\key overlap-long overlap` { + t.Fatal("mutated domain data") + } + if raw, err := r.JSON(math.NaN()); err == nil || raw != nil { + t.Fatalf("unsafe fallback: %s %v", raw, err) + } + if raw, err := r.JSON(map[string]string{"overlap": "first", "overlap-long": "second"}); err == nil || raw != nil { + t.Fatalf("colliding redacted keys: %s %v", raw, err) + } + nested, _ := json.Marshal(map[string]string{"reason": `quoted"\key`}) + raw, err = r.JSON(map[string]string{"final_content": string(nested)}) + if err != nil || strings.Contains(string(raw), "key") { + t.Fatalf("nested content: %s %v", raw, err) + } + yaml, err := r.YAML([]byte("version: 2\nenabled: true\nmessage: echo:2\n")) + if err != nil || !strings.Contains(string(yaml), "version: 2") || !strings.Contains(string(yaml), "echo:[REDACTED]") { + t.Fatalf("YAML: %s %v", yaml, err) + } +} + +func TestReportWritersRedactStringsAndFailClosed(t *testing.T) { + dir := t.TempDir() + version := 2 + report := reporting.ProcessReport{ErrorMessage: "echo:2"} + report.ReportMetadata.ConfigVersion = &version + run, err := diagnostics.NewRunDirectory(dir, "always", "2") + if err != nil { + t.Fatal(err) + } + external := filepath.Join(dir, "external.json") + if err := reporting.WriteProcessReport(external, report, "2"); err != nil { + t.Fatal(err) + } + if err := run.WriteReport(report); err != nil { + t.Fatal(err) + } + for _, path := range []string{external, filepath.Join(run.Path(), diagnostics.ArtifactReport)} { + raw, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + var got reporting.ProcessReport + if err := json.Unmarshal(raw, &got); err != nil { + t.Fatal(err) + } + if got.ErrorMessage != "echo:[REDACTED]" || got.ReportMetadata.ConfigVersion == nil || *got.ReportMetadata.ConfigVersion != 2 { + t.Fatalf("report: %+v", got) + } + } + if report.ErrorMessage != "echo:2" { + t.Fatal("mutated input report") + } + if err := run.WriteJSONArtifact("invalid.json", map[string]any{"secret": "echo:2", "invalid": math.NaN()}); err == nil { + t.Fatal("expected serialization failure") + } + if _, err := os.Stat(filepath.Join(run.Path(), "invalid.json")); !os.IsNotExist(err) { + t.Fatalf("failed artifact persisted: %v", err) + } +} diff --git a/internal/core/reporting/report.go b/internal/core/reporting/report.go index 70d4199..c4bba11 100644 --- a/internal/core/reporting/report.go +++ b/internal/core/reporting/report.go @@ -1,11 +1,11 @@ package reporting import ( - "encoding/json" "fmt" "os" "time" + "gitea.maximumdirect.net/eric/audita/internal/core/redaction" "gitea.maximumdirect.net/eric/audita/internal/framework/proposals" stagewarnings "gitea.maximumdirect.net/eric/audita/internal/framework/warnings" ) @@ -124,8 +124,8 @@ type ChunkingSummary struct { MinSectionTokens int `json:"min_section_tokens"` } -func WriteProcessReport(path string, report ProcessReport) error { - payload, err := json.MarshalIndent(report, "", " ") +func WriteProcessReport(path string, report ProcessReport, secrets ...string) error { + payload, err := redaction.New(secrets).JSON(report) if err != nil { return fmt.Errorf("failed to marshal process report JSON: %w", err) } diff --git a/internal/framework/llm/diagnostics.go b/internal/framework/llm/diagnostics.go index 7748e17..22a4cf3 100644 --- a/internal/framework/llm/diagnostics.go +++ b/internal/framework/llm/diagnostics.go @@ -1,15 +1,14 @@ package llm import ( - "encoding/json" "fmt" "os" "path/filepath" "regexp" "strings" -) -const redactedSecret = "[REDACTED]" + "gitea.maximumdirect.net/eric/audita/internal/core/redaction" +) var stageSanitizer = regexp.MustCompile(`[^a-zA-Z0-9_.-]+`) @@ -30,18 +29,7 @@ type DiagnosticsWriter struct { // NewDiagnosticsWriter creates a diagnostics writer rooted at dir. func NewDiagnosticsWriter(dir string, secrets []string) *DiagnosticsWriter { - filtered := make([]string, 0, len(secrets)) - for _, secret := range secrets { - secret = strings.TrimSpace(secret) - if secret != "" { - filtered = append(filtered, secret) - } - } - - return &DiagnosticsWriter{ - dir: dir, - secrets: filtered, - } + return &DiagnosticsWriter{dir: dir, secrets: append([]string(nil), secrets...)} } // WriteInteraction writes request/response/error artifacts for a generic LLM @@ -94,33 +82,13 @@ func sanitizeStageName(stage string) string { } func (w *DiagnosticsWriter) writeJSON(path string, payload any) error { - raw, err := json.MarshalIndent(payload, "", " ") + redacted, err := redaction.New(w.secrets).JSON(payload) if err != nil { return fmt.Errorf("marshal json artifact %q: %w", filepath.Base(path), err) } - redacted := redactJSONSecrets(raw, w.secrets) - if !json.Valid(redacted) { - return fmt.Errorf("redacted json artifact %q is invalid", filepath.Base(path)) - } redacted = append(redacted, '\n') if err := os.WriteFile(path, redacted, 0o644); err != nil { return fmt.Errorf("write json artifact %q: %w", filepath.Base(path), err) } return nil } - -func redactJSONSecrets(raw []byte, secrets []string) []byte { - if len(secrets) == 0 { - return raw - } - - result := string(raw) - for _, secret := range secrets { - if secret == "" { - continue - } - result = strings.ReplaceAll(result, "Bearer "+secret, "Bearer "+redactedSecret) - result = strings.ReplaceAll(result, secret, redactedSecret) - } - return []byte(result) -} diff --git a/internal/framework/llm/diagnostics_test.go b/internal/framework/llm/diagnostics_test.go index 8e99674..936a48f 100644 --- a/internal/framework/llm/diagnostics_test.go +++ b/internal/framework/llm/diagnostics_test.go @@ -61,7 +61,7 @@ func TestDiagnosticsWriterRedactsSecrets(t *testing.T) { if strings.Contains(content, secret) { t.Fatalf("secret leaked in %s: %s", filepath.Base(path), content) } - if !strings.Contains(content, redactedSecret) { + if !strings.Contains(content, "[REDACTED]") { t.Fatalf("redaction marker missing in %s: %s", filepath.Base(path), content) } if !json.Valid(raw) { diff --git a/internal/framework/llm/runtime.go b/internal/framework/llm/runtime.go index cb0e694..8fe06de 100644 --- a/internal/framework/llm/runtime.go +++ b/internal/framework/llm/runtime.go @@ -12,6 +12,7 @@ import ( "sync/atomic" "gitea.maximumdirect.net/eric/audita/internal/core/config" + "gitea.maximumdirect.net/eric/audita/internal/core/redaction" "gitea.maximumdirect.net/eric/audita/internal/framework/responseschema" "gitea.maximumdirect.net/eric/audita/internal/prompts" pk "gitea.maximumdirect.net/eric/promptkit" @@ -95,16 +96,5 @@ func RedactError(err error, c config.Config) string { if err == nil { return "" } - message := err.Error() - var secrets []string - for _, backend := range c.Backends { - if backend.Credential != nil && backend.Credential.Value() != "" { - secrets = append(secrets, backend.Credential.Value()) - } - } - sort.Slice(secrets, func(i, j int) bool { return len(secrets[i]) > len(secrets[j]) }) - for _, secret := range secrets { - message = strings.ReplaceAll(message, secret, "[REDACTED]") - } - return message + return redaction.New(ConfiguredSecrets(&c)).String(err.Error()) } diff --git a/internal/framework/llm/safe_view.go b/internal/framework/llm/safe_view.go index 30b5f4e..d55eaf9 100644 --- a/internal/framework/llm/safe_view.go +++ b/internal/framework/llm/safe_view.go @@ -6,6 +6,7 @@ import ( ck "gitea.maximumdirect.net/eric/ConfigKit" "gitea.maximumdirect.net/eric/audita/internal/core/config" + "gitea.maximumdirect.net/eric/audita/internal/core/redaction" "gopkg.in/yaml.v3" ) @@ -45,10 +46,14 @@ func (r *Runtime) SelectedProfilesYAML() ([]byte, error) { if err != nil { return nil, err } - return yaml.Marshal(struct { + raw, err := yaml.Marshal(struct { Proposal selectedProfileView `yaml:"proposal"` Validation selectedProfileView `yaml:"validation"` }{proposal, validation}) + if err != nil { + return nil, err + } + return redaction.New(ConfiguredSecrets(&r.Config)).YAML(raw) } type derivationView struct { diff --git a/internal/framework/processreport/llm.go b/internal/framework/processreport/llm.go index 746e1b8..4c857f4 100644 --- a/internal/framework/processreport/llm.go +++ b/internal/framework/processreport/llm.go @@ -29,86 +29,74 @@ func BuildUtilization(output *runner.RunOutput, backends map[string]config.Backe return result } result.RunWallTimeMS = output.RunWallTimeMS - for id, backend := range backends { - group := llm.ObservationSnapshot{} - for _, call := range output.Observations.Calls { - if call.Prompt.BackendID == id { - group.Calls = append(group.Calls, call) - addModuleCall(&result, call) + groups := make(map[string]*llm.ObservationSnapshot, len(backends)) + callBackends := make(map[uint64]string, len(output.Observations.Calls)) + moduleIndices := make(map[string]int) + moduleIndex := func(key, instance string) int { + if i, ok := moduleIndices[instance]; ok { + return i + } + i := len(result.Modules) + moduleIndices[instance] = i + result.Modules = append(result.Modules, reporting.ModuleUtilization{ModuleKey: key, ModuleInstance: instance}) + return i + } + // Seed configured order, including modules that made no LLM calls. + for _, module := range output.ModuleResults { + i := moduleIndex(module.ModuleKey, module.ModuleInstance) + result.Modules[i].ModuleWallTimeMS = module.CompletedAt.Sub(module.StartedAt).Milliseconds() + } + for _, call := range output.Observations.Calls { + id := call.Prompt.BackendID + if groups[id] == nil { + groups[id] = &llm.ObservationSnapshot{} + } + groups[id].Calls = append(groups[id].Calls, call) + callBackends[call.CallID] = id + if call.Context.ModuleInstance != "" { + i := moduleIndex(call.Context.ModuleKey, call.Context.ModuleInstance) + switch call.Context.Role { + case "proposal": + result.Modules[i].ProposalLLMCalls++ + case "validation": + result.Modules[i].ValidationLLMCalls++ } } + } + for _, event := range output.Observations.Events { + if id, ok := callBackends[event.CallID]; ok { + groups[id].Events = append(groups[id].Events, event) + } + } + for id, backend := range backends { + group := groups[id] + if group == nil { + group = &llm.ObservationSnapshot{} + } var wait, generation, retry, wall time.Duration for _, call := range group.Calls { wall += call.WallDuration } - for _, event := range output.Observations.Events { - if hasCall(group.Calls, event.CallID) { - group.Events = append(group.Events, event) - if event.Kind == string(pk.ExecutionFinished) { - wait += event.CapacityWait - generation += event.GenerationDuration - retry += event.RetryDelay - } + for _, event := range group.Events { + if event.Kind == string(pk.ExecutionFinished) { + wait += event.CapacityWait + generation += event.GenerationDuration + retry += event.RetryDelay } } result.Backends[id] = reporting.BackendUtilization{ Concurrency: backend.Concurrency, QueueCapacity: backend.QueueCapacity, - LLMSummary: mapLLMSummary(llm.Summarize(group)), + LLMSummary: mapLLMSummary(llm.Summarize(*group)), CapacityWaitMS: wait.Milliseconds(), GenerationMS: generation.Milliseconds(), RetryDelayMS: retry.Milliseconds(), LogicalWallMS: wall.Milliseconds(), } } - for _, module := range output.ModuleResults { - found := false - for i := range result.Modules { - if result.Modules[i].ModuleInstance == module.ModuleInstance { - result.Modules[i].ModuleWallTimeMS = module.CompletedAt.Sub(module.StartedAt).Milliseconds() - found = true - break - } - } - if !found { - result.Modules = append(result.Modules, reporting.ModuleUtilization{ModuleKey: module.ModuleKey, ModuleInstance: module.ModuleInstance, ModuleWallTimeMS: module.CompletedAt.Sub(module.StartedAt).Milliseconds()}) - } - } for _, timing := range output.ValidatorTimings { result.Validators = append(result.Validators, reporting.ValidatorUtilization{ModuleKey: timing.ModuleKey, ModuleInstance: timing.ModuleInstance, ValidatorKey: timing.ValidatorName, LLMBacked: timing.LLMBacked, Calls: timing.Calls, ElapsedMS: timing.ElapsedMS}) } return result } -func hasCall(calls []llm.LogicalCall, id uint64) bool { - for _, call := range calls { - if call.CallID == id { - return true - } - } - return false -} - -func addModuleCall(result *reporting.UtilizationDiagnostics, call llm.LogicalCall) { - if call.Context.ModuleInstance == "" { - return - } - for i := range result.Modules { - if result.Modules[i].ModuleInstance == call.Context.ModuleInstance { - if call.Context.Role == "proposal" { - result.Modules[i].ProposalLLMCalls++ - } else if call.Context.Role == "validation" { - result.Modules[i].ValidationLLMCalls++ - } - return - } - } - module := reporting.ModuleUtilization{ModuleKey: call.Context.ModuleKey, ModuleInstance: call.Context.ModuleInstance} - if call.Context.Role == "proposal" { - module.ProposalLLMCalls++ - } else if call.Context.Role == "validation" { - module.ValidationLLMCalls++ - } - result.Modules = append(result.Modules, module) -} - type executionCall struct { CallID uint64 `json:"call_id"` Context structuredoutput.DiagnosticContext `json:"context"` @@ -150,13 +138,12 @@ type executionEvent struct { func BuildLLMExecution(snapshot llm.ObservationSnapshot) any { calls := make([]executionCall, 0, len(snapshot.Calls)) events := make([]executionEvent, 0, len(snapshot.Events)) + eventsByCall := make(map[uint64][]llm.ExecutionObservation, len(snapshot.Calls)) + for _, event := range snapshot.Events { + eventsByCall[event.CallID] = append(eventsByCall[event.CallID], event) + } for _, call := range snapshot.Calls { - group := llm.ObservationSnapshot{Calls: []llm.LogicalCall{call}} - for _, event := range snapshot.Events { - if event.CallID == call.CallID { - group.Events = append(group.Events, event) - } - } + group := llm.ObservationSnapshot{Calls: []llm.LogicalCall{call}, Events: eventsByCall[call.CallID]} var finished *time.Time if !call.Finished.IsZero() { value := call.Finished diff --git a/internal/framework/processreport/llm_test.go b/internal/framework/processreport/llm_test.go index c710d74..2fc1a53 100644 --- a/internal/framework/processreport/llm_test.go +++ b/internal/framework/processreport/llm_test.go @@ -2,6 +2,7 @@ package processreport import ( "encoding/json" + "fmt" "strings" "testing" "time" @@ -54,3 +55,72 @@ func TestReportSummaryAvailability(t *testing.T) { t.Fatalf("initialized empty summary: %+v", got) } } + +func BenchmarkExecutionReports(b *testing.B) { + for _, n := range []int{1000, 10000, 20000} { + snapshot := llm.ObservationSnapshot{} + for i := 0; i < n; i++ { + snapshot.Calls = append(snapshot.Calls, llm.LogicalCall{CallID: uint64(i + 1), Prompt: structuredoutput.PromptMetadata{BackendID: "local"}, Context: structuredoutput.DiagnosticContext{ModuleInstance: "grammar", Role: "proposal"}}) + snapshot.Events = append(snapshot.Events, llm.ExecutionObservation{CallID: uint64(i + 1), Kind: string(pk.ExecutionFinished), Attempts: 1}) + } + output := &runner.RunOutput{Observations: snapshot} + b.Run(fmt.Sprintf("execution/%d", n), func(b *testing.B) { + for i := 0; i < b.N; i++ { + BuildLLMExecution(snapshot) + } + }) + b.Run(fmt.Sprintf("utilization/%d", n), func(b *testing.B) { + for i := 0; i < b.N; i++ { + BuildUtilization(output, map[string]config.Backend{"local": {Concurrency: 1}}) + } + }) + } +} + +func TestIndexedReportsPreserveOrderAndPartialUsage(t *testing.T) { + snapshot := llm.ObservationSnapshot{ + Calls: []llm.LogicalCall{ + {CallID: 3, Context: structuredoutput.DiagnosticContext{Role: "validation", ModuleInstance: "grammar_2"}, Prompt: structuredoutput.PromptMetadata{BackendID: "b"}}, + {CallID: 1, Context: structuredoutput.DiagnosticContext{Role: "proposal", ModuleInstance: "grammar_1"}, Prompt: structuredoutput.PromptMetadata{BackendID: "a"}}, + {CallID: 2, Context: structuredoutput.DiagnosticContext{Role: "validation", ModuleInstance: "grammar_1"}, Prompt: structuredoutput.PromptMetadata{BackendID: "a"}}, + }, + Events: []llm.ExecutionObservation{ + {CallID: 1, Kind: string(pk.ExecutionFinished), Attempts: 1, Usage: llm.Usage{Total: llm.TokenCount{Value: 7, Known: true, Complete: true}}}, + {CallID: 3, Kind: string(pk.ExecutionFinished), Attempts: 1}, + {CallID: 2, Kind: string(pk.ExecutionFinished), Attempts: 1}, + }, + } + raw, err := json.Marshal(BuildLLMExecution(snapshot)) + if err != nil { + t.Fatal(err) + } + var execution struct { + Calls []executionCall `json:"calls"` + Events []executionEvent `json:"events"` + } + if err := json.Unmarshal(raw, &execution); err != nil { + t.Fatal(err) + } + for i, call := range snapshot.Calls { + if execution.Calls[i].CallID != call.CallID { + t.Fatal("call order changed") + } + } + for i, event := range snapshot.Events { + if execution.Events[i].CallID != event.CallID { + t.Fatal("event order changed") + } + } + if execution.Calls[1].Summary.Usage.Total.Value != 7 || !execution.Calls[1].Summary.Usage.Total.Complete || execution.Calls[0].Summary.Usage.Total.Known { + t.Fatalf("call usage: %+v", execution.Calls) + } + output := &runner.RunOutput{Observations: snapshot} + utilization := BuildUtilization(output, map[string]config.Backend{"a": {}, "b": {}}) + total := utilization.Backends["a"].Usage.Total + if total.Value != 7 || !total.Known || total.Complete || utilization.Backends["a"].ProviderAttempts != 2 { + t.Fatalf("partial backend usage: %+v", utilization) + } + if len(utilization.Modules) != 2 || utilization.Modules[1].ProposalLLMCalls != 1 || utilization.Modules[1].ValidationLLMCalls != 1 { + t.Fatalf("module grouping: %+v", utilization.Modules) + } +} diff --git a/internal/framework/runner/cancellation_test.go b/internal/framework/runner/cancellation_test.go new file mode 100644 index 0000000..ddf01c2 --- /dev/null +++ b/internal/framework/runner/cancellation_test.go @@ -0,0 +1,145 @@ +package runner + +import ( + "context" + "errors" + "net/http" + "os" + "path/filepath" + "sync/atomic" + "testing" + "time" + + ck "gitea.maximumdirect.net/eric/ConfigKit" + "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/responseschema" + "gitea.maximumdirect.net/eric/audita/internal/framework/structuredoutput" + "gitea.maximumdirect.net/eric/audita/internal/framework/validators" + "gitea.maximumdirect.net/eric/audita/internal/prompts" + pk "gitea.maximumdirect.net/eric/promptkit" +) + +type contextValidator struct { + validate func(context.Context, validators.Request) (validators.Result, error) +} + +func (v contextValidator) Name() string { return "cancel-test" } +func (v contextValidator) Validate(ctx context.Context, req validators.Request) (validators.Result, error) { + return v.validate(ctx, req) +} + +type cancelTransport struct { + entered chan struct{} + calls atomic.Int32 +} + +func (t *cancelTransport) RoundTrip(req *http.Request) (*http.Response, error) { + t.calls.Add(1) + t.entered <- struct{}{} + <-req.Context().Done() + return nil, req.Context().Err() +} + +func TestValidatorFailureCancelsActiveAndPendingProviderWork(t *testing.T) { + dir := t.TempDir() + if err := os.Mkdir(filepath.Join(dir, "profiles"), 0700); err != nil { + t.Fatal(err) + } + main := filepath.Join(dir, "config.yml") + body := "version: 2\nbackends:\n shared:\n endpoint: http://provider.invalid/v1\n concurrency: 1\n queue_capacity: 1\nllm:\n proposal_profile: base\n transient_retries: 0\n repair_attempts: 0\n" + if err := os.WriteFile(main, []byte(body), 0600); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, "profiles/base.yaml"), []byte("id: base\nbackend: shared\nmodel: local\ntimeout_seconds: 30\n"), 0600); err != nil { + t.Fatal(err) + } + loaded, err := config.Load(ck.File(main), "") + if err != nil { + t.Fatal(err) + } + transport := &cancelTransport{entered: make(chan struct{}, 3)} + runtime, err := llm.NewRuntime(context.Background(), loaded.Config, main, &http.Client{Transport: transport}) + if err != nil { + t.Fatal(err) + } + collector := llm.NewObservationCollector() + hooks := collector.Hooks() + started := make(chan struct{}, 2) + hooks.OnEvent = func(event pk.ExecutionEvent) { + if event.Kind == pk.ExecutionStarted { + started <- struct{}{} + } + } + client, err := runtime.NewRoleClient("validation", hooks) + if err != nil { + t.Fatal(err) + } + origin := errors.New("originating validator failure") + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + validator := contextValidator{validate: func(ctx context.Context, req validators.Request) (validators.Result, error) { + if *req.SectionIndex == 0 { + // Hold the failure until one call is active and both peer executions started. + for i := 0; i < 2; i++ { + select { + case <-started: + case <-ctx.Done(): + return validators.Result{}, ctx.Err() + } + } + select { + case <-transport.entered: + case <-ctx.Done(): + return validators.Result{}, ctx.Err() + } + return validators.Result{}, origin + } + var out map[string]any + _, err := client.CompleteStructured(ctx, structuredoutput.Request{ + PromptID: prompts.PromptIDValidatorEditorialReview, PromptVersion: prompts.VersionV1, + Variables: map[string]string{"TranscriptDescriptionBlock": "", "PayloadJSON": "[]"}, Schema: responseschema.ValidatorDecisionSetKey, + Context: structuredoutput.DiagnosticContext{ModuleKey: req.ModuleKey, ModuleInstance: req.ModuleInstance, SectionIndex: req.SectionIndex}, + }, &out) + return validators.Result{}, err + }} + module := fakeModule{key: "grammar", policy: proposals.ReplacementPolicyRequireUnique, validators: []contracts.Validator{validator}, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { + return []proposals.CorrectionProposal{{TargetSegmentID: 1, OriginalText: "hello", CorrectedText: "Hello", Confidence: 1}}, nil + }} + sections := []chunking.Section{} + for i := 0; i < 3; i++ { + sections = append(sections, chunking.Section{Index: i, Segments: []schema.Segment{{ID: 1, Text: "hello"}}}) + } + done := make(chan error, 1) + go func() { + _, err := runModulePipeline(ctx, collectSectionProposalsInput{Module: module, Spec: contracts.ModuleRunSpec{ModuleKey: "grammar", InstanceName: "grammar"}, Policy: module.policy, Config: &loaded.Config, Sections: sections}) + done <- err + }() + select { + case err := <-done: + if !errors.Is(err, origin) { + t.Fatalf("lost originating failure: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("validator failure did not cancel and join peers") + } + if transport.calls.Load() != 1 { + t.Fatalf("pending call dispatched: %d", transport.calls.Load()) + } + snapshot := collector.Snapshot() + if len(snapshot.Calls) != 2 { + t.Fatalf("partial observations missing: %+v", snapshot) + } + for _, call := range snapshot.Calls { + if call.Finished.IsZero() || call.Failure != structuredoutput.FailureRuntime { + t.Fatalf("unfinished/cause lost: %+v", call) + } + } + if got := llm.Summarize(snapshot); got.ProviderAttempts != 1 { + t.Fatalf("attempt count: %+v", got) + } +} diff --git a/internal/framework/runner/runner.go b/internal/framework/runner/runner.go index 7308f17..ed19974 100644 --- a/internal/framework/runner/runner.go +++ b/internal/framework/runner/runner.go @@ -332,7 +332,7 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput) var ( firstErr error - errOnce sync.Once + errMu sync.Mutex vwg sync.WaitGroup ) @@ -340,10 +340,17 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput) if err == nil { return } - errOnce.Do(func() { + errMu.Lock() + defer errMu.Unlock() + if firstErr == nil { firstErr = err cancel() - }) + } + } + getErr := func() error { + errMu.Lock() + defer errMu.Unlock() + return firstErr } for result := range proposalResults { @@ -351,13 +358,16 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput) setErr(result.err) continue } - if firstErr != nil { + if getErr() != nil { continue } out.Warnings = append(out.Warnings, result.warnings...) pending[result.sectionPos] = result for { + if getErr() != nil { + break + } sectionResult, ok := pending[nextSectionToProcess] if !ok { break @@ -385,18 +395,19 @@ 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), - ModuleInstanceForStages: fmt.Sprintf("%s:section-%04d", input.Spec.InstanceName, 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: §ionMetadata.Index, + Validators: validatorsOrdered, + SectionEnriched: enriched, + ValidationLLMClient: input.ValidationClient, + DiagnosticsDir: input.ValidationDiagnosticsDir, + Timings: input.Timings, }) + setErr(err) validationResults <- sectionValidationResult{ sectionPos: sectionPos, enriched: enriched, @@ -477,25 +488,25 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput) return out.Warnings[i].Message < out.Warnings[j].Message }) - if firstErr != nil { - return out, firstErr + if err := getErr(); err != nil { + return out, err } 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 - DiagnosticsDir string - Timings *validatorTimingCollector + 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 { @@ -527,7 +538,8 @@ func validateSectionCandidates(ctx context.Context, input validateSectionCandida WorkingTranscript: input.WorkingTranscript, CandidateProposal: eligible, ModuleKey: input.Spec.ModuleKey, - ModuleInstance: input.ModuleInstanceForStages, + ModuleInstance: input.Spec.InstanceName, + SectionIndex: input.SectionIndex, ReplacementPolicy: input.Policy, Glossary: input.Glossary, Config: input.Config, diff --git a/internal/framework/runner/runner_test.go b/internal/framework/runner/runner_test.go index df99e1b..fac330c 100644 --- a/internal/framework/runner/runner_test.go +++ b/internal/framework/runner/runner_test.go @@ -438,8 +438,8 @@ func (c *captureContractStructuredClient) CompleteStructured(ctx context.Context func (c *stageAwareStructuredClient) CompleteStructured(ctx context.Context, req contracts.StructuredCompletionRequest, out any) (contracts.StructuredCompletionResponse, error) { section := 0 - if req.Context.BatchIndex != nil { - section = *req.Context.BatchIndex + if req.Context.SectionIndex != nil { + section = *req.Context.SectionIndex } if c.startedSection != nil { select { diff --git a/internal/framework/validators/llm_validators.go b/internal/framework/validators/llm_validators.go index 568a715..4923389 100644 --- a/internal/framework/validators/llm_validators.go +++ b/internal/framework/validators/llm_validators.go @@ -107,11 +107,15 @@ func (v *LLMBackedValidator) Validate(ctx context.Context, req Request) (Result, var completion StructuredCompletionResponse responseSchema := responseschema.MustLookup(responseschema.ValidatorDecisionSetKey) promptMetadata := validatorPromptMetadata(v.validatorType) - stage := stagename.ValidatorBatch(req.ModuleInstance, v.name, batch.BatchIndex) + stageInstance := req.ModuleInstance + if req.SectionIndex != nil { + stageInstance = fmt.Sprintf("%s:section-%04d", stageInstance, *req.SectionIndex) + } + stage := stagename.ValidatorBatch(stageInstance, v.name, batch.BatchIndex) completion, err = req.LLMClient.CompleteStructured(ctx, StructuredCompletionRequest{ PromptID: promptMetadata.PromptID, PromptVersion: promptMetadata.PromptVersion, Variables: variables, Schema: responseschema.ValidatorDecisionSetKey, - Context: structuredoutput.DiagnosticContext{Role: "validation", ModuleKey: req.ModuleKey, ModuleInstance: req.ModuleInstance, BatchIndex: &batch.BatchIndex, ValidatorName: v.name}, + Context: structuredoutput.DiagnosticContext{Role: "validation", ModuleKey: req.ModuleKey, ModuleInstance: req.ModuleInstance, SectionIndex: req.SectionIndex, BatchIndex: &batch.BatchIndex, ValidatorName: v.name}, }, &response) artifacts := InteractionArtifacts{} diff --git a/internal/framework/validators/models.go b/internal/framework/validators/models.go index 079c815..ef083a4 100644 --- a/internal/framework/validators/models.go +++ b/internal/framework/validators/models.go @@ -29,6 +29,7 @@ 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"`