diff --git a/cmd/audita/main_integration_test.go b/cmd/audita/main_integration_test.go index c5e665f..cb3b84b 100644 --- a/cmd/audita/main_integration_test.go +++ b/cmd/audita/main_integration_test.go @@ -16,6 +16,7 @@ import ( "gitea.maximumdirect.net/eric/audita/internal/cli" "gitea.maximumdirect.net/eric/audita/internal/core/reporting" + "gopkg.in/yaml.v3" ) func TestProcessSubprocessHelper(t *testing.T) { @@ -30,26 +31,16 @@ func TestProcessSubprocessHelper(t *testing.T) { os.Exit(2) } -type outboundCall struct { - Model string - Validation bool +type fixtureProvider struct { + mu sync.Mutex + calls int + validationCalls int + mode string + secret string } -type controlledProvider struct { - mu sync.Mutex - calls []outboundCall - proposalCount int - mode string - secret string - unselectedSecret string -} - -func (p *controlledProvider) serve(w http.ResponseWriter, r *http.Request) { +func (p *fixtureProvider) serve(w http.ResponseWriter, r *http.Request) { var request struct { - Model string `json:"model"` - Messages []struct { - Content string `json:"content"` - } `json:"messages"` ResponseFormat struct { JSONSchema struct { Schema struct { @@ -59,142 +50,87 @@ func (p *controlledProvider) serve(w http.ResponseWriter, r *http.Request) { } `json:"response_format"` } if err := json.NewDecoder(r.Body).Decode(&request); err != nil { - http.Error(w, err.Error(), 400) + http.Error(w, "bad request", 400) return } _, validation := request.ResponseFormat.JSONSchema.Schema.Properties["validations"] p.mu.Lock() - p.calls = append(p.calls, outboundCall{request.Model, validation}) - if !validation { - p.proposalCount++ + p.calls++ + if validation { + p.validationCalls++ } - proposalCount := p.proposalCount + validationCall := p.validationCalls p.mu.Unlock() - if p.mode == "failure" || (p.mode == "secret-partial-failure" && !validation && proposalCount > 1) { + if p.mode == "all-model-failure" || (p.mode == "partial-validation" && validation && validationCall == 1) { w.WriteHeader(http.StatusInternalServerError) _, _ = fmt.Fprintf(w, `{"error":{"message":"provider echoed %s"}}`, p.secret) return } - response := `{"corrections":[{"id":1,"original_text":"hello world","corrected_text":"Hello world","confidence":0.99}]}` + content := `{"corrections":[{"id":1,"original_text":"hello world","corrected_text":"Hello world","confidence":0.99}]}` if validation { - response = `{"validations":[{"correction_index":0,"approved":true,"confidence":0.99,"reason":"ok"}]}` + content = `{"validations":[{"correction_index":0,"approved":true,"confidence":0.99,"reason":"ok"}]}` } - if p.mode == "repair" && !validation && proposalCount == 1 { - response = `{"wrong":true}` - } - if p.mode == "warning" && !validation { - response = `{"wrong":true}` - } - if p.mode == "noop" && !validation { - response = `{"corrections":[]}` - } - 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) + encoded, _ := json.Marshal(content) 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) + _, _ = fmt.Fprintf(w, `{"model":"synthetic","choices":[{"message":{"content":%s}}],"usage":{"prompt_tokens":2,"completion_tokens":3,"total_tokens":5}}`, encoded) } +func (p *fixtureProvider) callCount() int { p.mu.Lock(); defer p.mu.Unlock(); return p.calls } + type processFixture struct { - dir, config, transcript, glossary, report, output string - provider *controlledProvider - server *httptest.Server + dir string + config string + transcript string + glossary string + output string + provider *fixtureProvider + server *httptest.Server } func newProcessFixture(t *testing.T, mode, modules string) processFixture { t.Helper() dir := t.TempDir() - p := &controlledProvider{mode: mode, secret: "synthetic-provider-secret-2468"} + p := &fixtureProvider{mode: mode, secret: `selected-secret-"\-123`} server := httptest.NewServer(http.HandlerFunc(p.serve)) t.Cleanup(server.Close) - if err := os.MkdirAll(filepath.Join(dir, "profiles"), 0700); err != nil { + if err := os.Mkdir(filepath.Join(dir, "profiles"), 0700); err != nil { t.Fatal(err) } - if err := os.MkdirAll(filepath.Join(dir, "overlays"), 0700); err != nil { - t.Fatal(err) - } - for name, model := range map[string]string{"proposal": "base-proposal", "validation": "base-validation", "proposal-testing": "testing-proposal", "validation-testing": "testing-validation", "alternate": "alternate-validation"} { - profile := fmt.Sprintf("id: %s\nbackend: local\nmodel: %s\ntimeout_seconds: 5\n", name, model) - if err := os.WriteFile(filepath.Join(dir, "profiles", name+".yaml"), []byte(profile), 0600); err != nil { + for _, id := range []string{"proposal", "validation"} { + profile := fmt.Sprintf("id: %s\nbackend: local\nmodel: local\ntimeout_seconds: 5\n", id) + if err := os.WriteFile(filepath.Join(dir, "profiles", id+".yaml"), []byte(profile), 0600); err != nil { t.Fatal(err) } } - if err := os.WriteFile(filepath.Join(dir, "overlays", "testing.yml"), []byte("llm:\n proposal_profile: proposal-testing\n validation_profile: validation-testing\n"), 0600); err != nil { - t.Fatal(err) - } configPath := filepath.Join(dir, "config.yml") - configText := fmt.Sprintf("version: 2\nbackends:\n local:\n endpoint: %s/v1\n credential: AUDITA_TEST_PROVIDER_KEY\n concurrency: 2\nllm:\n proposal_profile: proposal\n validation_profile: validation\n transient_retries: 0\n repair_attempts: 1\npipeline:\n modules: %s\ndiagnostics:\n work_dir: %s\n retention: always\n", server.URL, modules, filepath.Join(dir, "runs")) + configText := fmt.Sprintf("version: 2\nbackends:\n local:\n endpoint: %s/v1\n credential: AUDITA_TEST_PROVIDER_KEY\nllm:\n proposal_profile: proposal\n validation_profile: validation\n transient_retries: 0\n repair_attempts: 0\npipeline:\n modules: %s\ndiagnostics:\n work_dir: %s\n", server.URL, modules, filepath.Join(dir, "runs")) if err := os.WriteFile(configPath, []byte(configText), 0600); err != nil { t.Fatal(err) } - transcriptPath := filepath.Join(dir, "transcript.json") - if err := os.WriteFile(transcriptPath, []byte(`[{"id":1,"speaker":"A","start":0,"end":1,"text":"hello world"}]`), 0600); err != nil { + transcript := filepath.Join(dir, "input.json") + if err := os.WriteFile(transcript, []byte(`[{"id":1,"speaker":"A","start":0,"end":1,"text":"hello world"}]`), 0600); err != nil { t.Fatal(err) } - glossaryPath := filepath.Join(dir, "glossary.yaml") - if err := os.WriteFile(glossaryPath, []byte("glossary:\n - name: Example\n category: product\n summary: A sample.\n"), 0600); err != nil { + glossary := filepath.Join(dir, "glossary.yaml") + if err := os.WriteFile(glossary, []byte("glossary:\n - name: Example\n category: product\n summary: A sample.\n"), 0600); err != nil { t.Fatal(err) } - return processFixture{dir: dir, config: configPath, transcript: transcriptPath, glossary: glossaryPath, report: filepath.Join(dir, "report.json"), output: filepath.Join(dir, "output.json"), provider: p, server: server} + return processFixture{dir: dir, config: configPath, transcript: transcript, glossary: glossary, + output: filepath.Join(dir, "output.json"), provider: p, server: server} } -func (f processFixture) run(t *testing.T, extra ...string) (int, string, string, reporting.ProcessReport) { +func (f processFixture) run(t *testing.T, includeOutput bool, extra ...string) (int, string, string) { t.Helper() - args := []string{"-test.run=^TestProcessSubprocessHelper$", "--", "process", f.transcript, "--config", f.config, "--glossary", f.glossary, "--report-json", f.report, "--output", f.output} + args := []string{"-test.run=^TestProcessSubprocessHelper$", "--", "process", f.transcript, "--config", f.config, "--glossary", f.glossary} + if includeOutput { + args = append(args, "--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, "AUDITA_TEST_UNUSED_KEY="+f.provider.unselectedSecret) + command := exec.Command(os.Args[0], args...) + command.Env = append(os.Environ(), "AUDITA_HELPER_PROCESS=1", "AUDITA_TEST_PROVIDER_KEY="+f.provider.secret) var stdout, stderr bytes.Buffer - cmd.Stdout, cmd.Stderr = &stdout, &stderr - err := cmd.Run() + command.Stdout, command.Stderr = &stdout, &stderr + err := command.Run() code := 0 if err != nil { if exit, ok := err.(*exec.ExitError); ok { @@ -203,328 +139,194 @@ func (f processFixture) run(t *testing.T, extra ...string) (int, string, string, t.Fatal(err) } } - raw, err := os.ReadFile(f.report) + return code, stdout.String(), stderr.String() +} + +func (f processFixture) errors(t *testing.T) reporting.ErrorsDocument { + t.Helper() + raw, err := os.ReadFile(filepath.Join(f.dir, "errors.yml")) if err != nil { - t.Fatalf("report missing: %v, stderr=%s", err, stderr.String()) - } - var report reporting.ProcessReport - if err := json.Unmarshal(raw, &report); err != nil { t.Fatal(err) } - return code, stdout.String(), stderr.String(), report -} - -func (f processFixture) calls() []outboundCall { - f.provider.mu.Lock() - defer f.provider.mu.Unlock() - return append([]outboundCall(nil), f.provider.calls...) -} - -func TestSubprocessSelectionAndRepair(t *testing.T) { - for _, tc := range []struct { - name string - extra []string - proposal, validation string - }{ - {"base", nil, "base-proposal", "base-validation"}, - {"overlay and explicit validation", []string{"--overlay", "testing", "--validation-profile", "alternate", "--grammar-confidence-threshold", "0.9"}, "testing-proposal", "alternate-validation"}, - } { - t.Run(tc.name, func(t *testing.T) { - f := newProcessFixture(t, "repair", "[grammar]") - code, stdout, stderr, report := f.run(t, tc.extra...) - if code != 0 || stdout != "" || stderr != "" || report.Status != "success" { - t.Fatalf("code=%d stdout=%q stderr=%q report=%+v", code, stdout, stderr, report) - } - raw, err := os.ReadFile(f.output) - if err != nil { - t.Fatal(err) - } - if !bytes.Contains(raw, []byte("Hello world")) { - t.Fatalf("correction missing: %s", raw) - } - calls := f.calls() - if len(calls) < 3 || calls[0].Model != tc.proposal || calls[1].Model != tc.proposal { - t.Fatalf("proposal models: %+v", calls) - } - for _, call := range calls[2:] { - if !call.Validation || call.Model != tc.validation { - t.Fatalf("validation model: %+v", calls) - } - } - if report.LLMSummary == nil || report.LLMSummary.StructuralRepairs != 1 || report.LLMSummary.ProviderAttempts != uint64(len(calls)) { - t.Fatalf("LLM summary: %+v", report.LLMSummary) - } - if report.Diagnostics == nil || report.Diagnostics.LLMExecutionPath == "" || report.Diagnostics.SelectedProfilesPath == "" { - t.Fatalf("paths: %+v", report.Diagnostics) - } - }) + var document reporting.ErrorsDocument + if err := yaml.Unmarshal(raw, &document); err != nil { + t.Fatal(err) } + return document } -func TestSubprocessWarningEmptyRepeatedAndFailure(t *testing.T) { - t.Run("domain rejection", func(t *testing.T) { - f := newProcessFixture(t, "low-confidence", "[grammar]") - code, _, stderr, report := f.run(t) - if code != 0 || stderr != "" || report.Status != "success" || len(report.ModuleResults) != 1 || len(report.ModuleResults[0].ValidatorRejected) == 0 { - t.Fatalf("domain rejection: %d %q %+v", code, stderr, report) - } - if len(f.calls()) != 1 { - t.Fatalf("semantic rejection retried: %+v", f.calls()) - } - }) - t.Run("warning", func(t *testing.T) { - f := newProcessFixture(t, "warning", "[grammar]") - code, _, stderr, report := f.run(t) - if code != 0 || stderr != "" || report.Status != "success" || len(report.ModuleResults) != 1 || len(report.ModuleResults[0].Warnings) == 0 { - t.Fatalf("warning result: %d %q %+v", code, stderr, report) - } - if len(f.calls()) != 2 || report.LLMSummary == nil || report.LLMSummary.StructuralRepairs != 1 { - t.Fatalf("warning observations: %+v %+v", f.calls(), report.LLMSummary) - } - }) - t.Run("empty", func(t *testing.T) { - f := newProcessFixture(t, "noop", "[]") - code, _, stderr, report := f.run(t) - if code != 0 || stderr != "" || len(f.calls()) != 0 || report.LLMSummary == nil || !report.LLMSummary.Usage.Total.Known || !report.LLMSummary.Usage.Total.Complete { - t.Fatalf("empty result: %d %q %+v", code, stderr, report.LLMSummary) - } - }) - t.Run("repeated", func(t *testing.T) { - f := newProcessFixture(t, "noop", "[grammar, grammar]") - code, _, stderr, report := f.run(t) - if code != 0 || stderr != "" || len(report.ModuleResults) != 2 || report.ModuleResults[0].ModuleInstance == report.ModuleResults[1].ModuleInstance || len(f.calls()) != 2 { - t.Fatalf("repeated result: %d %q %+v %+v", code, stderr, report.ModuleResults, f.calls()) - } - }) - t.Run("failure redaction", func(t *testing.T) { - f := newProcessFixture(t, "failure", "[grammar]") - code, stdout, stderr, report := f.run(t) - if code != 1 || stdout != "" || report.Status != "failed" || report.ErrorPhase != "runner_execution" || report.LLMSummary == nil || report.LLMSummary.LogicalCalls != 1 { - t.Fatalf("failure result: %d %q %q %+v", code, stdout, stderr, report) - } - if strings.Contains(stderr, f.provider.secret) { - t.Fatal("secret in stderr") - } - for _, path := range []string{f.report, report.Diagnostics.DirectoryPath} { - _ = filepath.Walk(path, func(name string, info os.FileInfo, err error) error { - if err != nil { - t.Fatal(err) - } - if info.IsDir() { - return nil - } - content, err := os.ReadFile(name) - if err != nil { - t.Fatal(err) - } - if bytes.Contains(content, []byte(f.provider.secret)) { - t.Fatalf("secret in %s", name) - } - return nil - }) - } - if _, err := os.Stat(report.Diagnostics.LLMExecutionPath); err != nil { +func TestSubprocessDefaultFileOutputAndReports(t *testing.T) { + f := newProcessFixture(t, "normal", "[grammar]") + code, stdout, stderr := f.run(t, true) + if code != 0 || stdout != "" || !strings.HasPrefix(stderr, "Completed: 1 corrections applied; 0 work units with errors; 0 recoveries.") { + t.Fatalf("code=%d stdout=%q stderr=%q", code, stdout, stderr) + } + output, err := os.ReadFile(f.output) + if err != nil || !bytes.Contains(output, []byte("Hello world")) { + t.Fatalf("output: %s %v", output, err) + } + document := f.errors(t) + if document.Status != reporting.StatusCompleted || document.Summary.WorkUnitsCompleted == nil || *document.Summary.WorkUnitsCompleted != 1 || len(document.Errors) != 0 { + t.Fatalf("errors document: %+v", document) + } + for _, name := range []string{"errors.yml", "recoveries.yml", "report.md"} { + if _, err := os.Stat(filepath.Join(f.dir, name)); err != nil { t.Fatal(err) } - }) + } + if _, err := os.Stat(filepath.Join(f.dir, "runs")); !os.IsNotExist(err) { + t.Fatalf("debug created by default: %v", err) + } } -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 { +func TestSubprocessStdoutAndSelectedReportDirectory(t *testing.T) { + f := newProcessFixture(t, "normal", "[grammar]") + reports := filepath.Join(f.dir, "reports") + if err := os.Mkdir(reports, 0700); err != nil { + t.Fatal(err) + } + code, stdout, stderr := f.run(t, false, "--report-dir", reports) + if code != 0 || !strings.Contains(stdout, "Hello world") || !strings.HasPrefix(stderr, "Completed:") { + t.Fatalf("code=%d stdout=%q stderr=%q", code, stdout, stderr) + } + if _, err := os.Stat(f.output); !os.IsNotExist(err) { + t.Fatalf("stdout run wrote transcript file: %v", err) + } + if _, err := os.Stat(filepath.Join(reports, "errors.yml")); err != nil { + t.Fatal(err) + } +} + +func TestSubprocessReportingAndDebugSelections(t *testing.T) { + for _, debug := range []bool{false, true} { + f := newProcessFixture(t, "normal", "[grammar]") + extra := []string{"--reporting=false", "--report-dir", filepath.Join(f.dir, "missing")} + if debug { + extra = append(extra, "--debug") + } + code, stdout, stderr := f.run(t, true, extra...) + if code != 0 || stdout != "" || !strings.HasPrefix(stderr, "Completed:") { + t.Fatalf("debug=%v code=%d stdout=%q stderr=%q", debug, code, stdout, stderr) + } + if _, err := os.Stat(filepath.Join(f.dir, "errors.yml")); !os.IsNotExist(err) { + t.Fatalf("disabled report created: %v", err) + } + dirs, err := filepath.Glob(filepath.Join(f.dir, "runs", "run-*")) + if err != nil || len(dirs) != boolInt(debug) { + t.Fatalf("debug=%v dirs=%v %v", debug, dirs, err) + } + if debug { + for _, name := range []string{"source-transcript.json", "normalized-transcript.json", "correction-ledger.json", "llm-execution.json"} { + if _, err := os.Stat(filepath.Join(dirs[0], name)); 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) +func boolInt(value bool) int { + if value { + return 1 } - 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) + return 0 +} + +func TestSubprocessAllModelFailureStillCompletes(t *testing.T) { + f := newProcessFixture(t, "all-model-failure", "[grammar]") + code, stdout, stderr := f.run(t, true) + if code != 0 || stdout != "" || !strings.Contains(stderr, "1 work units with errors") { + t.Fatalf("code=%d stdout=%q stderr=%q", code, stdout, stderr) } - 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) + output, err := os.ReadFile(f.output) + if err != nil || !bytes.Contains(output, []byte("hello world")) { + t.Fatalf("normalized output: %s %v", output, err) } - code, _, stderr, report := f.run(t) - if code != 0 { - t.Fatalf("process: %d %s", code, stderr) + document := f.errors(t) + if document.Status != reporting.StatusCompleted || len(document.Errors) != 1 || document.Errors[0].Scope != "work_unit" || *document.Summary.WorkUnitsWithErrors != 1 { + t.Fatalf("model failure report: %+v", document) } - 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) - } + if strings.Contains(stderr, f.provider.secret) { + t.Fatal("secret leaked in stderr") + } +} + +func TestSubprocessPartialValidationAndLaterStage(t *testing.T) { + f := newProcessFixture(t, "partial-validation", "[grammar, grammar]") + code, _, stderr := f.run(t, true) + if code != 0 || !strings.Contains(stderr, "1 corrections applied; 1 work units with errors") { + t.Fatalf("code=%d stderr=%q", code, stderr) + } + document := f.errors(t) + if len(document.Errors) != 1 || document.Errors[0].ModuleInstance != "grammar_1" || *document.Summary.WorkUnitsCompleted != 2 { + t.Fatalf("partial outcome: %+v", document) + } + output, err := os.ReadFile(f.output) + if err != nil || !bytes.Contains(output, []byte("Hello world")) { + t.Fatalf("later correction: %s %v", output, err) + } +} + +func TestSubprocessInvalidInputLeavesFailedReports(t *testing.T) { + f := newProcessFixture(t, "normal", "[grammar]") + if err := os.WriteFile(f.transcript, []byte("not JSON"), 0600); err != nil { + t.Fatal(err) + } + code, _, _ := f.run(t, true) + if code != 1 || f.provider.callCount() != 0 { + t.Fatalf("invalid input dispatched: code=%d calls=%d", code, f.provider.callCount()) + } + if document := f.errors(t); document.Status != reporting.StatusFailed || len(document.Errors) != 1 || document.Errors[0].Scope != "run" { + t.Fatalf("failed report: %+v", document) + } + if _, err := os.Stat(f.output); !os.IsNotExist(err) { + t.Fatalf("failed transcript published: %v", err) + } + var recoveries reporting.RecoveriesDocument + if raw, err := os.ReadFile(filepath.Join(f.dir, "recoveries.yml")); err != nil { + t.Fatal(err) + } else if err := yaml.Unmarshal(raw, &recoveries); err != nil { + t.Fatal(err) + } + if recoveries.Status != reporting.StatusFailed { + t.Fatalf("recoveries status: %+v", recoveries) + } + if raw, err := os.ReadFile(filepath.Join(f.dir, "report.md")); err != nil || !bytes.Contains(raw, []byte("Status: failed")) { + t.Fatalf("Markdown status: %s %v", raw, err) + } +} + +func TestSubprocessOutputFailureBeforeModelDispatch(t *testing.T) { + f := newProcessFixture(t, "normal", "[grammar]") + f.output = filepath.Join(f.dir, "missing", "output.json") + code, _, stderr := f.run(t, true) + if code == 0 || f.provider.callCount() != 0 || !strings.Contains(stderr, "report directory") { + t.Fatalf("preflight: code=%d calls=%d stderr=%q", code, f.provider.callCount(), stderr) + } + if _, err := os.Stat(filepath.Join(f.dir, "errors.yml")); !os.IsNotExist(err) { + t.Fatalf("report initialized before preflight: %v", err) + } +} + +func TestSubprocessOptionalDebugFailureDoesNotChangeOutcome(t *testing.T) { + f := newProcessFixture(t, "normal", "[grammar]") + code, _, stderr := f.run(t, true, "--debug", "--work-dir", f.transcript) + if code != 0 || !strings.HasPrefix(stderr, "Completed: 1 corrections applied") || !strings.Contains(stderr, "debug directory") { + t.Fatalf("optional debug failure changed processing: code=%d stderr=%q", code, stderr) + } + if _, err := os.Stat(f.output); err != nil { + t.Fatal(err) + } + if document := f.errors(t); document.Status != reporting.StatusCompleted { + t.Fatalf("report status: %+v", document) + } +} + +func TestSubprocessTranscriptPreflightWithoutReporting(t *testing.T) { + f := newProcessFixture(t, "normal", "[grammar]") + f.output = filepath.Join(f.dir, "missing", "output.json") + code, _, stderr := f.run(t, true, "--reporting=false") + if code == 0 || f.provider.callCount() != 0 || !strings.Contains(stderr, "transcript output parent") { + t.Fatalf("transcript preflight: code=%d calls=%d stderr=%q", code, f.provider.callCount(), stderr) } } diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index d35236f..7749709 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -743,6 +743,19 @@ reporting-on/debug-off both work without coupling their destinations. ## Stage 12 — Compose the Final CLI Run Lifecycle +**Completed 2026-09-24.** CLI processing now inspects settings and profiles, +preflights and initializes required outputs before input/model work, creates +debug evidence only when selected, runs the fixed-chunk runner with caller +context, and finalizes transcript/reports in order. Failed processing and output +writes produce failed-status reports from available partial state; successful +stderr begins with the shared correction/error/recovery counts. The duplicate +CLI chunking pass and old JSON report flow are removed. Subprocess tests cover +file/stdout destinations, disabled reports, debug-only runs and optional debug +failure, total model failure, partial validation and later work, invalid input +after initialization, and pre-dispatch output failures. Focused package tests +and `git diff --check` passed; `go test ./...` passed before the last small +integration assertions were added and will be rerun in Stage 14. + **Work:** Replace the current monolithic process/report/retention flow with the final composition: load/inspect settings, resolve and initialize required outputs, optionally initialize debug, load/normalize input, run fixed-chunk processing, diff --git a/internal/cli/process.go b/internal/cli/process.go new file mode 100644 index 0000000..a4f4ccb --- /dev/null +++ b/internal/cli/process.go @@ -0,0 +1,431 @@ +package cli + +import ( + "bytes" + "context" + "crypto/rand" + "encoding/hex" + "errors" + "fmt" + "io" + "path/filepath" + "strings" + "time" + + "gitea.maximumdirect.net/eric/audita/internal/core/chunking" + "gitea.maximumdirect.net/eric/audita/internal/core/config" + "gitea.maximumdirect.net/eric/audita/internal/core/diagnostics" + 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/reporting" + "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/modules" + "gitea.maximumdirect.net/eric/audita/internal/framework/outcomes" + "gitea.maximumdirect.net/eric/audita/internal/framework/processreport" + "gitea.maximumdirect.net/eric/audita/internal/framework/runner" + "gitea.maximumdirect.net/eric/audita/internal/framework/structuredoutput" +) + +func runProcess(ctx context.Context, args []string, stdout, stderr io.Writer) int { + input, help, err := parseConfigInput("process", args, stderr) + if err != nil { + fmt.Fprintf(stderr, "audita process: %s\n", configInputError(err)) + return 2 + } + if help { + writeProcessUsage(stdout) + return 0 + } + selected, err := loadConfigInput(input, "process") + if err != nil { + fmt.Fprintf(stderr, "audita process: %s\n", configInputError(err)) + return 2 + } + cfg := selected.Result.Config + reportDir, err := selectedReportDir(cfg, input.OutputPath) + if err != nil { + fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(err, cfg)) + return 2 + } + if cfg.Diagnostics.Debug && strings.TrimSpace(cfg.Diagnostics.WorkDir) == "" { + fmt.Fprintln(stderr, "audita process: debug work directory is required") + return 2 + } + if err := ctx.Err(); err != nil { + fmt.Fprintf(stderr, "audita process: %s\n", err) + return 1 + } + runtime, err := llm.NewRuntime(ctx, cfg, selected.Path, nil) + if err != nil { + fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(err, cfg)) + if ctx.Err() != nil || llm.IsAssetError(err) { + return 1 + } + return 2 + } + specs, err := contracts.ResolveModuleRunSpecs(cfg.Pipeline.Modules) + if err != nil { + fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(err, cfg)) + return 2 + } + runID, err := generateRunID() + if err != nil { + fmt.Fprintf(stderr, "audita process: run ID: %s\n", llm.RedactError(err, cfg)) + return 1 + } + started := time.Now().UTC() + state := &processState{cfg: cfg, runtime: runtime, specs: specs, runID: runID, started: started, + transcriptPath: input.TranscriptPath, glossaryPath: input.GlossaryPath, outputPath: input.OutputPath, + selected: selected, overlay: input.Overlay, stderr: stderr, stdout: stdout, ctx: ctx} + protected := []string{input.TranscriptPath, input.GlossaryPath} + for _, source := range selected.Result.Sources() { + if source.Path != "" { + protected = append(protected, source.Path) + } + } + protected = append(protected, selectedProfilePaths(cfg, selected.Path)...) + session, err := diagnostics.PrepareOutput(diagnostics.OutputSelection{TranscriptPath: input.OutputPath, + ReportDir: reportDir, ReportingEnabled: cfg.Reporting.Enabled, ProtectedPaths: protected}) + if err != nil { + fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(err, cfg)) + return 2 + } + state.session = session + initial := state.snapshot(reporting.StatusInProgress, nil, nil) + initialBundle, err := state.bundle(initial) + if err != nil { + fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(err, cfg)) + return 1 + } + initFailure := []processreport.RunFailure{{Phase: "initialization", Reason: outcomes.ReasonInternalFailure}} + failedInit, err := state.bundle(state.snapshot(reporting.StatusFailed, nil, initFailure)) + if err != nil { + fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(err, cfg)) + return 1 + } + if err := session.Initialize(initialBundle, failedInit); err != nil { + fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(err, cfg)) + return 1 + } + state.startDebug() + return state.execute() +} + +type processState struct { + cfg config.Config + runtime *llm.Runtime + specs []contracts.ModuleRunSpec + runID string + started time.Time + transcriptPath string + glossaryPath string + outputPath string + selected selectedConfig + overlay string + ctx context.Context + stdout io.Writer + stderr io.Writer + session *diagnostics.OutputSession + debugDir *diagnostics.RunDirectory + debugWarnings []string + runOutput *runner.RunOutput +} + +func generateRunID() (string, error) { + var random [12]byte + if _, err := rand.Read(random[:]); err != nil { + return "", err + } + return "run-" + hex.EncodeToString(random[:]), nil +} + +func selectedProfilePaths(cfg config.Config, configPath string) []string { + dir := cfg.ProfilesDir + if !filepath.IsAbs(dir) { + dir = filepath.Join(filepath.Dir(configPath), dir) + } + ids := []string{cfg.LLM.ProposalProfile, cfg.LLM.ValidationProfile} + paths := make([]string, 0, len(ids)*2) + for _, id := range ids { + paths = append(paths, filepath.Join(dir, id+".yaml"), filepath.Join(dir, id+".yml")) + } + return paths +} + +func (s *processState) snapshot(status string, completed *time.Time, failures []processreport.RunFailure) reporting.Snapshot { + return processreport.Build(processreport.BuildInput{RunID: s.runID, Status: status, StartedAt: s.started, + CompletedAt: completed, TranscriptPath: s.transcriptPath, GlossaryPath: s.glossaryPath, + OutputPath: s.outputPath, ModuleSpecs: s.specs, RunOutput: s.runOutput, RunFailures: failures}) +} + +func (s *processState) bundle(snapshot reporting.Snapshot) (diagnostics.ReportBundle, error) { + if !s.cfg.Reporting.Enabled { + return diagnostics.ReportBundle{}, nil + } + secrets := llm.ConfiguredSecrets(&s.cfg) + errorsYAML, err := reporting.YAML(snapshot.ErrorsDocument(), secrets...) + if err != nil { + return diagnostics.ReportBundle{}, err + } + recoveriesYAML, err := reporting.YAML(snapshot.RecoveriesDocument(), secrets...) + if err != nil { + return diagnostics.ReportBundle{}, err + } + return diagnostics.ReportBundle{Errors: errorsYAML, Recoveries: recoveriesYAML, + Markdown: reporting.Markdown(snapshot, secrets...)}, nil +} + +func (s *processState) debugWrite(label string, write func(*diagnostics.RunDirectory) error) { + if s.debugDir == nil { + return + } + if err := write(s.debugDir); err != nil { + s.debugWarnings = append(s.debugWarnings, "debug "+label+": "+llm.RedactError(err, s.cfg)) + } +} + +func (s *processState) startDebug() { + if !s.cfg.Diagnostics.Debug { + return + } + 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)) + return + } + s.debugDir = dir + s.debugWrite("invocation", func(d *diagnostics.RunDirectory) error { + return d.WriteInvocationMetadata(diagnostics.InvocationMetadata{ + Operation: "process", TranscriptPath: s.transcriptPath, GlossaryPath: s.glossaryPath, + OutputPath: s.outputPath, ConfigPath: s.selected.Path, ConfigSource: s.selected.Source, + Overlay: s.overlay, ConfigVersion: &s.cfg.Version, TranscriptDescription: s.cfg.Context.Description, + Modules: append([]string(nil), s.cfg.Pipeline.Modules...), RunID: s.runID, StartedAt: s.started, + }) + }) + var effective bytes.Buffer + 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)) + } + 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) + }) + } + if profiles, err := s.runtime.SelectedProfilesYAML(); err == nil { + s.debugWrite("selected profiles", func(d *diagnostics.RunDirectory) error { + return d.WriteYAMLArtifact(diagnostics.ArtifactSelectedProfiles, profiles) + }) + } +} + +func (s *processState) debugPath() string { + if s.debugDir == nil { + return "" + } + return s.debugDir.Path() +} + +func (s *processState) execute() int { + transcriptBytes, err := coreio.ReadRequiredFile(s.transcriptPath, "transcript") + if err != nil { + return s.fail("input", outcomes.ReasonInternalFailure, err) + } + glossaryBytes, err := coreio.ReadRequiredFile(s.glossaryPath, "glossary") + if err != nil { + return s.fail("input", outcomes.ReasonInternalFailure, err) + } + source, err := schema.ParseSourceTranscriptJSON(transcriptBytes) + if err != nil { + return s.fail("input", outcomes.ReasonInputInvalid, err) + } + glossary, err := schema.ParseGlossaryYAML(glossaryBytes) + if err != nil { + return s.fail("input", outcomes.ReasonInputInvalid, err) + } + s.debugWrite("source transcript", func(d *diagnostics.RunDirectory) error { return d.WriteSourceTranscript(source, transcriptBytes) }) + normalizer := normalization.NewNormalizer(normalization.NormalizationConfig{ + MaxSegmentGap: s.cfg.Normalization.MaxSegmentGap, EllipsisGap: s.cfg.Normalization.EllipsisGap, + MaxSegmentDuration: s.cfg.Normalization.MaxSegmentDuration, MaxSegmentTokens: s.cfg.Normalization.MaxSegmentTokens, + }) + normalized, normSummary := normalizer.Normalize(sourceToCanonicalTranscript(source)) + s.debugWrite("normalized transcript", func(d *diagnostics.RunDirectory) error { return d.WriteNormalizedTranscript(normalized) }) + s.debugWrite("normalization summary", func(d *diagnostics.RunDirectory) error { return d.WriteNormalizationSummary(normSummary) }) + if err := s.ctx.Err(); err != nil { + return s.fail("processing", outcomes.ReasonCanceled, err) + } + observations := llm.NewObservationCollector() + proposalClient, err := s.runtime.NewRoleClient("proposal", observations.Hooks()) + if err != nil { + return s.fail("processing", outcomes.ReasonSetupFailure, err) + } + validationClient, err := s.runtime.NewRoleClient("validation", observations.Hooks()) + if err != nil { + return s.fail("processing", outcomes.ReasonSetupFailure, err) + } + factory := modules.NewFactory(modules.Dependencies{Config: &s.cfg, Glossary: glossary, DiagnosticsDir: s.debugPath()}) + result, runErr := runner.New(factory).Run(s.ctx, runner.RunInput{ + Config: &s.cfg, Transcript: normalized, Glossary: glossary, ModuleSpecs: s.specs, + ProposalLLMClient: proposalClient, ProposalDiagnosticsDir: s.debugPath(), + ValidationLLMClient: validationClient, ValidationDiagnosticsDir: s.debugPath(), + CaptureCandidates: s.debugDir != nil, + }) + result.Observations = observations.Snapshot() + result.ObservationsAvailable = true + result.LLMSummary = llm.Summarize(result.Observations) + s.runOutput = &result + s.writeExecutionEvidence() + if len(result.Sections) > 0 || runErr == nil { + s.writeChunkSummary(result.Sections) + } + if runErr != nil { + return s.fail("processing", fatalReason(s.ctx, runErr), runErr) + } + encoder, err := outputschema.Resolve(s.cfg.Output.Schema) + if err != nil { + return s.fail("encoding", outcomes.ReasonSetupFailure, err) + } + outputBytes, err := encoder.Encoder(result.FinalTranscript) + if err != nil { + return s.fail("encoding", outcomes.ReasonInternalFailure, err) + } + return s.finish(outputBytes) +} + +func (s *processState) writeChunkSummary(inventory []contracts.SectionMetadata) { + if s.debugDir == nil { + return + } + sections := make([]chunking.Section, len(inventory)) + for i, metadata := range inventory { + sections[i] = chunking.Section{Index: metadata.Index, StartSegmentID: metadata.StartSegmentID, + EndSegmentID: metadata.EndSegmentID, EstimatedTokens: metadata.EstimatedTokens, + Segments: make([]schema.Segment, metadata.SegmentCount)} + } + config := chunking.ChunkingConfig{TargetSections: s.cfg.Chunking.TargetSections, + MaxSectionTokens: s.cfg.Chunking.MaxSectionTokens, MinSectionTokens: s.cfg.Chunking.MinSectionTokens} + summary := chunking.ComputeDetailedSummary(sections, config) + s.debugWrite("chunking summary", func(d *diagnostics.RunDirectory) error { return d.WriteChunkingSummary(&summary) }) +} + +func (s *processState) writeExecutionEvidence() { + if s.debugDir == nil || s.runOutput == nil { + return + } + output := s.runOutput + s.debugWrite("model execution", func(d *diagnostics.RunDirectory) error { + return d.WriteJSONArtifact(diagnostics.ArtifactLLMExecution, processreport.BuildLLMExecution(output.Observations)) + }) + s.debugWrite("utilization", func(d *diagnostics.RunDirectory) error { + return d.WriteJSONArtifact(diagnostics.ArtifactUtilizationSummary, processreport.BuildUtilization(output, s.cfg.Backends)) + }) + s.debugWrite("correction ledger", func(d *diagnostics.RunDirectory) error { + return d.WriteJSONArtifact(diagnostics.ArtifactCorrectionLedger, processreport.BuildCorrectionLedger(processreport.CorrectionLedgerInput{ + RunDirectoryPath: d.Path(), RunOutput: output})) + }) +} + +func fatalReason(ctx context.Context, err error) outcomes.ReasonCode { + if ctx.Err() != nil || errors.Is(err, context.Canceled) { + return outcomes.ReasonCanceled + } + var failure *structuredoutput.Failure + if errors.As(err, &failure) && failure.Reason != "" { + return failure.Reason + } + return outcomes.ReasonInternalFailure +} + +func (s *processState) fail(phase string, reason outcomes.ReasonCode, cause error) int { + s.debugWrite("error log", func(d *diagnostics.RunDirectory) error { + return d.WriteErrorLog(phase + ": " + llm.RedactError(cause, s.cfg)) + }) + failures := []processreport.RunFailure{{Phase: phase, Reason: reason}} + completed := time.Now().UTC() + base, err := s.bundle(s.snapshot(reporting.StatusFailed, &completed, failures)) + if err != nil { + fmt.Fprintf(s.stderr, "audita process: %s\n", llm.RedactError(err, s.cfg)) + } + writeFailures := s.session.Finalize(nil, base, func(extra []diagnostics.FileFailure) diagnostics.ReportBundle { + all := append(append([]processreport.RunFailure(nil), failures...), mapWriteFailures(extra)...) + bundle, renderErr := s.bundle(s.snapshot(reporting.StatusFailed, &completed, all)) + if renderErr != nil { + return base + } + return bundle + }) + fmt.Fprintf(s.stderr, "audita process: %s\n", llm.RedactError(cause, s.cfg)) + s.printFailures(writeFailures) + s.printDebugWarnings() + if s.debugDir != nil { + fmt.Fprintf(s.stderr, "audita process: diagnostics: %s\n", s.debugDir.Path()) + } + return 1 +} + +func (s *processState) finish(outputBytes []byte) int { + completed := time.Now().UTC() + snapshot := s.snapshot(reporting.StatusCompleted, &completed, nil) + complete, err := s.bundle(snapshot) + if err != nil { + return s.fail("report_write", outcomes.ReasonInternalFailure, err) + } + if s.outputPath == "" { + if _, err := s.stdout.Write(outputBytes); err != nil { + return s.fail("transcript_write", outcomes.ReasonInternalFailure, err) + } + outputBytes = nil + } + writeFailures := s.session.Finalize(outputBytes, complete, func(extra []diagnostics.FileFailure) diagnostics.ReportBundle { + failed, renderErr := s.bundle(s.snapshot(reporting.StatusFailed, &completed, mapWriteFailures(extra))) + if renderErr != nil { + return diagnostics.ReportBundle{} + } + return failed + }) + if len(writeFailures) > 0 { + s.printFailures(writeFailures) + s.printDebugWarnings() + if s.debugDir != nil { + fmt.Fprintf(s.stderr, "audita process: diagnostics: %s\n", 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) + } + if s.debugDir != nil { + fmt.Fprintf(s.stderr, "Debug: %s\n", s.debugDir.Path()) + } + s.printDebugWarnings() + return 0 +} + +func mapWriteFailures(failures []diagnostics.FileFailure) []processreport.RunFailure { + out := make([]processreport.RunFailure, len(failures)) + for i, failure := range failures { + phase := "report_write" + if failure.Kind == "transcript write" { + phase = "transcript_write" + } + out[i] = processreport.RunFailure{Phase: phase, Reason: outcomes.ReasonInternalFailure, OutputPath: failure.Path} + } + return out +} + +func (s *processState) printFailures(failures []diagnostics.FileFailure) { + for _, failure := range failures { + fmt.Fprintf(s.stderr, "audita process: %s\n", llm.RedactError(failure, s.cfg)) + } +} + +func (s *processState) printDebugWarnings() { + for _, warning := range s.debugWarnings { + fmt.Fprintf(s.stderr, "audita process: %s\n", warning) + } +} diff --git a/internal/cli/run.go b/internal/cli/run.go index 20b5fb5..a665fda 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -1,229 +1,14 @@ package cli import ( - "bytes" "context" "fmt" "io" - "strings" - "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/diagnostics" - 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/schema" - "gitea.maximumdirect.net/eric/audita/internal/framework/contracts" "gitea.maximumdirect.net/eric/audita/internal/framework/llm" - "gitea.maximumdirect.net/eric/audita/internal/framework/modules" - "gitea.maximumdirect.net/eric/audita/internal/framework/processreport" - "gitea.maximumdirect.net/eric/audita/internal/framework/runner" ) -type processInvocation struct { - TranscriptPath string - GlossaryPath string - OutputPath string - ReportDir string - RunID string - Config config.Config - ConfigResult *ck.Result[config.Config] - LLMRuntime *llm.Runtime - ConfigPath string - ConfigSource string - Overlay string - ConfigVersion *int -} - -var processRunner = func(ctx context.Context, inv processInvocation, stdout, stderr io.Writer) (*normalization.NormalizationSummary, *chunking.Summary, *runner.RunOutput, *diagnostics.RunDirectory, error) { - var runDir *diagnostics.RunDirectory - if inv.Config.Diagnostics.Debug { - var err error - runDir, err = diagnostics.NewRunDirectory(inv.Config.Diagnostics.WorkDir, inv.RunID, llm.ConfiguredSecrets(&inv.Config)...) - if err != nil { - fmt.Fprintf(stderr, "audita process: debug directory: %s\n", llm.RedactError(err, inv.Config)) - } - } - debugPath := "" - if runDir != nil { - debugPath = runDir.Path() - } - debugWrite := func(label string, write func(*diagnostics.RunDirectory) error) { - if runDir == nil { - return - } - if err := write(runDir); err != nil { - fmt.Fprintf(stderr, "audita process: debug %s: %s\n", label, llm.RedactError(err, inv.Config)) - } - } - - fail := func(phase string, err error, runOutput *runner.RunOutput) (*normalization.NormalizationSummary, *chunking.Summary, *runner.RunOutput, *diagnostics.RunDirectory, error) { - debugWrite("error log", func(d *diagnostics.RunDirectory) error { - return d.WriteErrorLog(fmt.Sprintf("%s: %s", phase, llm.RedactError(err, inv.Config))) - }) - return nil, nil, runOutput, runDir, fmt.Errorf("%s: %w", phase, err) - } - - debugWrite("invocation", func(d *diagnostics.RunDirectory) error { - return d.WriteInvocationMetadata(diagnostics.InvocationMetadata{ - Operation: "process", - TranscriptPath: inv.TranscriptPath, - GlossaryPath: inv.GlossaryPath, - OutputPath: inv.OutputPath, - ConfigPath: inv.ConfigPath, - ConfigSource: inv.ConfigSource, - Overlay: inv.Overlay, - ConfigVersion: inv.ConfigVersion, - TranscriptDescription: inv.Config.Context.Description, - Modules: append([]string(nil), inv.Config.Pipeline.Modules...), - }) - }) - if runDir != nil && inv.ConfigResult != nil { - var effective bytes.Buffer - if err := inv.ConfigResult.WriteEffective(&effective); err == nil { - debugWrite("effective settings", func(d *diagnostics.RunDirectory) error { return d.WriteEffectiveConfig(effective.Bytes()) }) - } - if origins, err := llm.ExplanationYAML(inv.ConfigResult); err == nil { - debugWrite("configuration origins", func(d *diagnostics.RunDirectory) error { - return d.WriteYAMLArtifact(diagnostics.ArtifactConfigOrigins, origins) - }) - } - } - if runDir != nil && inv.LLMRuntime != nil { - if profiles, err := inv.LLMRuntime.SelectedProfilesYAML(); err == nil { - debugWrite("selected profiles", func(d *diagnostics.RunDirectory) error { - return d.WriteYAMLArtifact(diagnostics.ArtifactSelectedProfiles, profiles) - }) - } - } - - transcriptBytes, err := coreio.ReadRequiredFile(inv.TranscriptPath, "transcript") - if err != nil { - return fail("transcript_read", err, nil) - } - - glossaryBytes, err := coreio.ReadRequiredFile(inv.GlossaryPath, "glossary") - if err != nil { - return fail("glossary_read", err, nil) - } - - sourceTranscript, err := schema.ParseSourceTranscriptJSON(transcriptBytes) - if err != nil { - return fail("transcript_schema", err, nil) - } - - glossary, err := schema.ParseGlossaryYAML(glossaryBytes) - if err != nil { - return fail("glossary_schema", err, nil) - } - - debugWrite("source transcript", func(d *diagnostics.RunDirectory) error { - return d.WriteSourceTranscript(sourceTranscript, transcriptBytes) - }) - - canonical := sourceToCanonicalTranscript(sourceTranscript) - normalizer := normalization.NewNormalizer(normalization.NormalizationConfig{ - MaxSegmentGap: inv.Config.Normalization.MaxSegmentGap, - EllipsisGap: inv.Config.Normalization.EllipsisGap, - MaxSegmentDuration: inv.Config.Normalization.MaxSegmentDuration, - MaxSegmentTokens: inv.Config.Normalization.MaxSegmentTokens, - }) - - normalizedTranscript, normSummary := normalizer.Normalize(canonical) - - debugWrite("normalized transcript", func(d *diagnostics.RunDirectory) error { return d.WriteNormalizedTranscript(normalizedTranscript) }) - debugWrite("normalization summary", func(d *diagnostics.RunDirectory) error { return d.WriteNormalizationSummary(normSummary) }) - - // Compute chunks after normalization - chunker := chunking.NewChunker(chunking.ChunkingConfig{ - MaxSectionTokens: inv.Config.Chunking.MaxSectionTokens, - MinSectionTokens: inv.Config.Chunking.MinSectionTokens, - TargetSections: inv.Config.Chunking.TargetSections, - }) - - sections, chunkErr := chunker.ChunkTranscript(normalizedTranscript) - if chunkErr != nil { - return fail("chunking", chunkErr, nil) - } - - chunkConfig := chunking.ChunkingConfig{ - MaxSectionTokens: inv.Config.Chunking.MaxSectionTokens, - MinSectionTokens: inv.Config.Chunking.MinSectionTokens, - TargetSections: inv.Config.Chunking.TargetSections, - } - chunkSummary := chunking.ComputeSummary(sections, chunkConfig) - chunkDetailedSummary := chunking.ComputeDetailedSummary(sections, chunkConfig) - - debugWrite("chunking summary", func(d *diagnostics.RunDirectory) error { return d.WriteChunkingSummary(&chunkDetailedSummary) }) - - workingTranscript := normalizedTranscript - var runOutput *runner.RunOutput - observations := llm.NewObservationCollector() - moduleFactory := modules.NewFactory(modules.Dependencies{Config: &inv.Config, Glossary: glossary, DiagnosticsDir: debugPath}) - if inv.LLMRuntime == nil { - return fail("runner_setup", fmt.Errorf("LLM runtime is required"), nil) - } - proposalLLMClient, err := inv.LLMRuntime.NewRoleClient("proposal", observations.Hooks()) - if err != nil { - return fail("runner_setup", err, nil) - } - validationLLMClient, err := inv.LLMRuntime.NewRoleClient("validation", observations.Hooks()) - if err != nil { - return fail("runner_setup", err, nil) - } - - moduleSpecs, err := contracts.ResolveModuleRunSpecs(inv.Config.Pipeline.Modules) - if err != nil { - return fail("runner_setup", err, nil) - } - - runnerResult, runErr := runner.New(moduleFactory).Run(ctx, runner.RunInput{ - Config: &inv.Config, - Transcript: normalizedTranscript, - Glossary: glossary, - ModuleSpecs: moduleSpecs, - - ProposalLLMClient: proposalLLMClient, - ProposalDiagnosticsDir: debugPath, - ValidationLLMClient: validationLLMClient, - ValidationDiagnosticsDir: debugPath, - CaptureCandidates: runDir != nil, - }) - runnerResult.Observations = observations.Snapshot() - runnerResult.ObservationsAvailable = true - runnerResult.LLMSummary = llm.Summarize(runnerResult.Observations) - runOutput = &runnerResult - if runErr != nil { - return fail("runner_execution", runErr, runOutput) - } - workingTranscript = runnerResult.FinalTranscript - - encoderDef, err := outputschema.Resolve(inv.Config.Output.Schema) - if err != nil { - return fail("output_schema", err, runOutput) - } - outputBytes, err := encoderDef.Encoder(workingTranscript) - if err != nil { - return fail("serialization", err, runOutput) - } - - if strings.TrimSpace(inv.OutputPath) != "" { - if err := coreio.WriteFile(inv.OutputPath, outputBytes); err != nil { - return fail("output_write", err, runOutput) - } - return normSummary, &chunkSummary, runOutput, runDir, nil - } - - if _, err := stdout.Write(outputBytes); err != nil { - return fail("stdout_write", err, runOutput) - } - - return normSummary, &chunkSummary, runOutput, runDir, nil -} - func sourceToCanonicalTranscript(source *schema.SourceTranscript) *schema.Transcript { segments := make([]schema.Segment, len(source.Segments)) for i, s := range source.Segments { @@ -277,108 +62,6 @@ func Run(ctx context.Context, args []string, stdout, stderr io.Writer) int { return 2 } -func runProcess(ctx context.Context, args []string, stdout, stderr io.Writer) int { - startedAt := time.Now().UTC() - input, help, err := parseConfigInput("process", args, stderr) - if err != nil { - fmt.Fprintf(stderr, "audita process: %s\n", configInputError(err)) - return 2 - } - if help { - writeProcessUsage(stdout) - return 0 - } - selected, err := loadConfigInput(input, "process") - if err != nil { - fmt.Fprintf(stderr, "audita process: %s\n", configInputError(err)) - return 2 - } - reportDir, err := selectedReportDir(selected.Result.Config, input.OutputPath) - if err != nil { - fmt.Fprintf(stderr, "audita process: %s\n", configInputError(err)) - return 2 - } - if selected.Result.Config.Diagnostics.Debug && strings.TrimSpace(selected.Result.Config.Diagnostics.WorkDir) == "" { - fmt.Fprintln(stderr, "audita process: debug work directory is required") - return 2 - } - llmRuntime, err := llm.NewRuntime(ctx, selected.Result.Config, selected.Path, nil) - if err != nil { - fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(err, selected.Result.Config)) - if llm.IsAssetError(err) { - return 1 - } - return 2 - } - version := selected.Result.Config.Version - - inv := processInvocation{ - TranscriptPath: input.TranscriptPath, - GlossaryPath: input.GlossaryPath, - OutputPath: input.OutputPath, - ReportDir: reportDir, - Config: selected.Result.Config, - ConfigResult: selected.Result, - LLMRuntime: llmRuntime, - ConfigPath: selected.Path, - ConfigSource: selected.Source, - Overlay: input.Overlay, - ConfigVersion: &version, - } - - normSummary, chunkSummary, runOutput, runDir, runErr := processRunner(ctx, inv, stdout, stderr) - completedAt := time.Now().UTC() - - if runErr != nil { - if runDir != nil && runOutput != nil { - writeExecutionArtifacts(runDir, runOutput, inv.Config, stderr) - - if debugErr := runDir.WriteJSONArtifact(diagnostics.ArtifactCorrectionLedger, processreport.BuildCorrectionLedger(processreport.CorrectionLedgerInput{ - RunDirectoryPath: runDir.Path(), - RunOutput: runOutput, - })); debugErr != nil { - fmt.Fprintf(stderr, "audita process: debug correction ledger: %s\n", llm.RedactError(debugErr, inv.Config)) - } - } - 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)) - - fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(runErr, inv.Config)) - if runDir != nil { - fmt.Fprintf(stderr, "audita process: diagnostics: %s\n", runDir.Path()) - } - return 1 - } - - if runDir != nil && runOutput != nil { - writeExecutionArtifacts(runDir, runOutput, inv.Config, stderr) - - if debugErr := runDir.WriteJSONArtifact(diagnostics.ArtifactCorrectionLedger, processreport.BuildCorrectionLedger(processreport.CorrectionLedgerInput{ - RunDirectoryPath: runDir.Path(), - RunOutput: runOutput, - })); debugErr != nil { - fmt.Fprintf(stderr, "audita process: debug correction ledger: %s\n", llm.RedactError(debugErr, inv.Config)) - } - } - - report := processreport.Build(processReportInput("success", inv, runDir, startedAt, completedAt, "", "", normSummary, chunkSummary, runOutput)) - - return 0 -} - -func writeExecutionArtifacts(runDir *diagnostics.RunDirectory, output *runner.RunOutput, cfg config.Config, stderr io.Writer) { - if !output.ObservationsAvailable { - return - } - if err := runDir.WriteJSONArtifact(diagnostics.ArtifactLLMExecution, processreport.BuildLLMExecution(output.Observations)); err != nil { - fmt.Fprintf(stderr, "audita process: debug model execution: %s\n", llm.RedactError(err, cfg)) - } - if err := runDir.WriteJSONArtifact(diagnostics.ArtifactUtilizationSummary, processreport.BuildUtilization(output, cfg.Backends)); err != nil { - fmt.Fprintf(stderr, "audita process: debug utilization: %s\n", llm.RedactError(err, cfg)) - } -} - func runConfig(ctx context.Context, args []string, stdout, stderr io.Writer) int { if len(args) == 0 || isHelpCommand(args) || args[0] == "--help" || args[0] == "-h" { writeConfigUsage(stdout) @@ -466,42 +149,6 @@ func runConfigPrintEffective(ctx context.Context, args []string, stdout, stderr return 0 } -func extractErrorPhase(err error) (phase string, message string) { - msg := err.Error() - if strings.Contains(msg, ": ") { - parts := strings.SplitN(msg, ": ", 2) - if len(parts) == 2 { - return parts[0], parts[1] - } - } - return "", msg -} - -func processReportInput(status string, inv processInvocation, runDir *diagnostics.RunDirectory, startedAt, completedAt time.Time, errorMessage string, errorPhase string, normalizationSummary *normalization.NormalizationSummary, chunkingSummary *chunking.Summary, runOutput *runner.RunOutput) processreport.BuildInput { - runDirectoryPath := "" - if runDir != nil { - runDirectoryPath = runDir.Path() - } - return processreport.BuildInput{ - Status: status, - TranscriptPath: inv.TranscriptPath, - GlossaryPath: inv.GlossaryPath, - OutputPath: inv.OutputPath, - Modules: inv.Config.Pipeline.Modules, - OutputSchema: inv.Config.Output.Schema, - ConfigVersion: inv.ConfigVersion, - StartedAt: startedAt, - CompletedAt: completedAt, - ErrorMessage: errorMessage, - ErrorPhase: errorPhase, - RunDirectoryPath: runDirectoryPath, - NormalizationSummary: normalizationSummary, - ChunkingSummary: chunkingSummary, - RunOutput: runOutput, - Backends: inv.Config.Backends, - } -} - func isHelpCommand(args []string) bool { if len(args) == 0 { return false diff --git a/internal/core/diagnostics/output_session.go b/internal/core/diagnostics/output_session.go index a2c2101..3385432 100644 --- a/internal/core/diagnostics/output_session.go +++ b/internal/core/diagnostics/output_session.go @@ -151,7 +151,7 @@ func (s *OutputSession) Initialize(initial, failed ReportBundle) error { // Finalize publishes the transcript first, then errors, recoveries, and Markdown. // An empty transcript payload means processing failed and no transcript is written. // Every enabled report is attempted even after an earlier write failure. -func (s *OutputSession) Finalize(transcript []byte, reports, failed ReportBundle) []FileFailure { +func (s *OutputSession) Finalize(transcript []byte, reports ReportBundle, failed func([]FileFailure) ReportBundle) []FileFailure { if s == nil { return []FileFailure{{Kind: "output finalization", Err: fmt.Errorf("output session is required")}} } @@ -168,7 +168,7 @@ func (s *OutputSession) Finalize(transcript []byte, reports, failed ReportBundle } } if len(failures) > 0 { - for i, payload := range bundleParts(failed) { + for i, payload := range bundleParts(failed(failures)) { _ = s.replace(s.reports[i], payload) } } diff --git a/internal/core/diagnostics/output_session_test.go b/internal/core/diagnostics/output_session_test.go index c88462e..546bec1 100644 --- a/internal/core/diagnostics/output_session_test.go +++ b/internal/core/diagnostics/output_session_test.go @@ -57,7 +57,7 @@ func TestPrepareOutputPreservesExistingTranscriptAndChecksParent(t *testing.T) { if got, err := os.ReadFile(output); err != nil || string(got) != "old content" { t.Fatalf("preflight truncated output: %s %v", got, err) } - if failures := session.Finalize([]byte("new content"), ReportBundle{}, ReportBundle{}); len(failures) != 0 { + if failures := session.Finalize([]byte("new content"), ReportBundle{}, func([]FileFailure) ReportBundle { return ReportBundle{} }); len(failures) != 0 { t.Fatal(failures) } if got, err := os.ReadFile(output); err != nil || string(got) != "new content" { @@ -127,7 +127,7 @@ func TestStdoutModeInitializesAndFinalizesReports(t *testing.T) { t.Fatalf("initial %s: %s %v", path, got, err) } } - if failures := session.Finalize(nil, finished, ReportBundle{}); len(failures) != 0 { + if failures := session.Finalize(nil, finished, func([]FileFailure) ReportBundle { return ReportBundle{} }); len(failures) != 0 { t.Fatal(failures) } for i, path := range session.ReportPaths() { @@ -186,7 +186,7 @@ func TestFinalizationCollectsFailuresAndRetainsPublishedTranscript(t *testing.T) } complete := ReportBundle{Errors: []byte("complete"), Recoveries: []byte("complete"), Markdown: []byte("complete")} failed := ReportBundle{Errors: []byte("failed"), Recoveries: []byte("failed"), Markdown: []byte("failed")} - failures := session.Finalize([]byte("transcript"), complete, failed) + failures := session.Finalize([]byte("transcript"), complete, func([]FileFailure) ReportBundle { return failed }) if len(failures) != 1 || failures[0].Path != filepath.Join(dir, ErrorsFile) { t.Fatalf("failures: %+v", failures) } diff --git a/internal/framework/contracts/contracts.go b/internal/framework/contracts/contracts.go index 50789d0..53c8cd4 100644 --- a/internal/framework/contracts/contracts.go +++ b/internal/framework/contracts/contracts.go @@ -45,6 +45,7 @@ type SectionMetadata struct { StartSegmentID int `json:"start_segment_id"` EndSegmentID int `json:"end_segment_id"` EstimatedTokens int `json:"estimated_tokens"` + SegmentCount int `json:"segment_count"` } // SectionMetadataFromSection converts a chunking section into framework metadata. @@ -54,6 +55,7 @@ func SectionMetadataFromSection(section chunking.Section) SectionMetadata { StartSegmentID: section.StartSegmentID, EndSegmentID: section.EndSegmentID, EstimatedTokens: section.EstimatedTokens, + SegmentCount: len(section.Segments), } } diff --git a/internal/framework/outcomes/outcomes.go b/internal/framework/outcomes/outcomes.go index f78af8d..a58ef09 100644 --- a/internal/framework/outcomes/outcomes.go +++ b/internal/framework/outcomes/outcomes.go @@ -50,6 +50,7 @@ const ( ReasonInvalidOutput ReasonCode = "invalid_output" ReasonValidationInputTooLarge ReasonCode = "validation_input_too_large" ReasonValidationDecisionsInvalid ReasonCode = "validation_decisions_invalid" + ReasonInputInvalid ReasonCode = "input_invalid" ReasonSetupFailure ReasonCode = "setup_failure" ReasonCanceled ReasonCode = "canceled" ReasonInternalFailure ReasonCode = "internal_failure" diff --git a/internal/framework/processreport/report.go b/internal/framework/processreport/report.go index f930225..50c4b85 100644 --- a/internal/framework/processreport/report.go +++ b/internal/framework/processreport/report.go @@ -43,7 +43,7 @@ func Build(input BuildInput) reporting.Snapshot { for i, spec := range input.ModuleSpecs { snapshot.StageNames[i] = spec.InstanceName } - if input.RunOutput != nil { + if input.RunOutput != nil && input.RunOutput.InventoryAvailable { units := append([]runner.WorkUnitOutcome(nil), input.RunOutput.WorkUnits...) sort.SliceStable(units, func(i, j int) bool { if units[i].ModuleIndex != units[j].ModuleIndex { @@ -200,6 +200,8 @@ func safeMessage(reason outcomes.ReasonCode) string { return "Processing was canceled." case outcomes.ReasonSetupFailure: return "Processing setup failed." + case outcomes.ReasonInputInvalid: + return "Input was invalid." default: return "Processing failed." } diff --git a/internal/framework/processreport/report_test.go b/internal/framework/processreport/report_test.go index c626c2a..0d5714d 100644 --- a/internal/framework/processreport/report_test.go +++ b/internal/framework/processreport/report_test.go @@ -23,7 +23,7 @@ func TestBuildPartialUnitsAndCombinedRecovery(t *testing.T) { batch := 1 validation := outcomes.Operation{Reference: outcomes.Reference{Kind: outcomes.OperationValidation, ValidatorName: "review", ValidatorOrder: intPtr(1), BatchIndex: &batch}, FailureReason: outcomes.ReasonTransportFailed, ProposalIndices: []int{9, 3}} inputFailure := outcomes.Operation{Reference: outcomes.Reference{Kind: outcomes.OperationValidationInput, ValidatorName: "review", ValidatorOrder: intPtr(1)}, FailureReason: outcomes.ReasonValidationInputTooLarge, ProposalIndices: []int{7}} - output := &runner.RunOutput{WorkUnits: []runner.WorkUnitOutcome{ + output := &runner.RunOutput{InventoryAvailable: true, WorkUnits: []runner.WorkUnitOutcome{ {ModuleIndex: 1, ModuleKey: "grammar", ModuleInstance: "grammar_2", Section: contracts.SectionMetadata{Index: 0, StartSegmentID: 1, EndSegmentID: 2}, State: outcomes.WorkNotStarted}, {ModuleIndex: 0, ModuleKey: "grammar", ModuleInstance: "grammar_1", Section: contracts.SectionMetadata{Index: 0, StartSegmentID: 1, EndSegmentID: 2}, State: outcomes.WorkCompleted, CorrectionsApplied: 1, Operations: []outcomes.Operation{validation, proposal, inputFailure}}, }} @@ -64,7 +64,7 @@ func TestBuildInitializationNoWorkAndRunFailure(t *testing.T) { if initialDoc.CompletedAt != nil || initialDoc.Summary.WorkUnitsTotal != nil || initialDoc.Errors == nil { t.Fatalf("initial: %+v", initialDoc) } - finished := Build(BuildInput{RunID: "r", Status: reporting.StatusCompleted, StartedAt: start, RunOutput: &runner.RunOutput{}}) + finished := Build(BuildInput{RunID: "r", Status: reporting.StatusCompleted, StartedAt: start, RunOutput: &runner.RunOutput{InventoryAvailable: true}}) if finished.Summary.WorkUnitsTotal == nil || *finished.Summary.WorkUnitsTotal != 0 || finished.RecoveriesDocument().Recoveries == nil { t.Fatalf("empty: %+v", finished) } diff --git a/internal/framework/runner/runner.go b/internal/framework/runner/runner.go index 21881e2..1df90e7 100644 --- a/internal/framework/runner/runner.go +++ b/internal/framework/runner/runner.go @@ -103,6 +103,7 @@ type RunOutput struct { FinalTranscript *schema.Transcript `json:"-"` ModuleResults []ModuleResult `json:"module_results"` Sections []contracts.SectionMetadata `json:"sections"` + InventoryAvailable bool `json:"-"` WorkUnits []WorkUnitOutcome `json:"work_units"` Candidates []proposals.EnrichedCorrectionProposal `json:"-"` Observations llm.ObservationSnapshot `json:"-"` @@ -154,7 +155,7 @@ func (r *Runner) run(ctx context.Context, input RunInput) (RunOutput, error) { snapshot := func() RunOutput { return RunOutput{ FinalTranscript: transcriptFromSections(sections), ModuleResults: results, - Sections: inventory, WorkUnits: workUnits, Candidates: candidates, + Sections: inventory, InventoryAvailable: true, WorkUnits: workUnits, Candidates: candidates, } }