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

Verify the integrated resilient processing contract

Browse Source
This commit is contained in:
Eric Rakestraw
2026-09-24 18:03:25 +00:00
parent a770c77a23
commit 6523d7ac74
5 changed files with 242 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

188
cmd/audita/main_integration_test.go
View File

@@ -9,10 +9,13 @@ import (
"net/http/httptest"
"os"
"os/exec"
"os/signal"
"path/filepath"
"strings"
"sync"
"syscall"
"testing"
"time"
"gitea.maximumdirect.net/eric/audita/internal/cli"
"gitea.maximumdirect.net/eric/audita/internal/core/reporting"
@@ -25,7 +28,10 @@ func TestProcessSubprocessHelper(t *testing.T) {
}
for i, arg := range os.Args {
if arg == "--" {
os.Exit(cli.Run(context.Background(), os.Args[i+1:], os.Stdout, os.Stderr))
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
code := cli.Run(ctx, os.Args[i+1:], os.Stdout, os.Stderr)
stop()
os.Exit(code)
}
}
os.Exit(2)
@@ -34,9 +40,11 @@ func TestProcessSubprocessHelper(t *testing.T) {
type fixtureProvider struct {
mu sync.Mutex
calls int
proposalCalls int
validationCalls int
mode string
secret string
entered chan struct{}
}
func (p *fixtureProvider) serve(w http.ResponseWriter, r *http.Request) {
@@ -58,17 +66,44 @@ func (p *fixtureProvider) serve(w http.ResponseWriter, r *http.Request) {
p.calls++
if validation {
p.validationCalls++
} else {
p.proposalCalls++
}
validationCall := p.validationCalls
proposalCall := p.proposalCalls
p.mu.Unlock()
if p.mode == "all-model-failure" || (p.mode == "partial-validation" && validation && validationCall == 1) {
if p.mode == "wait-cancel" {
select {
case p.entered <- struct{}{}:
default:
}
<-r.Context().Done()
return
}
if p.mode == "all-model-failure" || (p.mode == "all-validation-failure" && validation) || (p.mode == "partial-validation" && validation && validationCall == 1) || (p.mode == "repair-filter-later-failure" && !validation && proposalCall == 3) {
w.WriteHeader(http.StatusInternalServerError)
_, _ = fmt.Fprintf(w, `{"error":{"message":"provider echoed %s"}}`, p.secret)
return
}
content := `{"corrections":[{"id":1,"original_text":"hello world","corrected_text":"Hello world","confidence":0.99}]}`
if p.mode == "repair-filter-later-failure" && !validation {
if proposalCall == 1 {
content = `{"wrong":true}`
} else {
content = `{"corrections":[{"id":1,"original_text":"absent text","corrected_text":"X","confidence":0.99},{"id":1,"original_text":"hello world","corrected_text":"Hello world","confidence":0.99}]}`
}
}
if validation {
content = `{"validations":[{"correction_index":0,"approved":true,"confidence":0.99,"reason":"ok"}]}`
if p.mode == "repair-filter-later-failure" {
content = `{"validations":[{"correction_index":1,"approved":true,"confidence":0.99,"reason":"ok"}]}`
} else if p.mode == "invalid-decision-after-repair" {
if validationCall == 1 {
content = `{"wrong":true}`
} else {
content = `{"validations":[{"correction_index":9,"approved":true,"confidence":0.99,"reason":"ok"}]}`
}
}
}
encoded, _ := json.Marshal(content)
w.Header().Set("Content-Type", "application/json")
@@ -90,7 +125,7 @@ type processFixture struct {
func newProcessFixture(t *testing.T, mode, modules string) processFixture {
t.Helper()
dir := t.TempDir()
p := &fixtureProvider{mode: mode, secret: `selected-secret-"\-123`}
p := &fixtureProvider{mode: mode, secret: `selected-secret-"\-123`, entered: make(chan struct{}, 1)}
server := httptest.NewServer(http.HandlerFunc(p.serve))
t.Cleanup(server.Close)
if err := os.Mkdir(filepath.Join(dir, "profiles"), 0700); err != nil {
@@ -103,7 +138,11 @@ func newProcessFixture(t *testing.T, mode, modules string) processFixture {
}
}
configPath := filepath.Join(dir, "config.yml")
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"))
repairs := 0
if mode == "repair-filter-later-failure" || mode == "invalid-decision-after-repair" {
repairs = 1
}
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: %d\npipeline:\n modules: %s\ndiagnostics:\n work_dir: %s\n", server.URL, repairs, modules, filepath.Join(dir, "runs"))
if err := os.WriteFile(configPath, []byte(configText), 0600); err != nil {
t.Fatal(err)
}
@@ -170,9 +209,13 @@ func TestSubprocessDefaultFileOutputAndReports(t *testing.T) {
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 {
report, err := os.ReadFile(filepath.Join(f.dir, name))
if err != nil {
t.Fatal(err)
}
if bytes.Contains(report, []byte("hello world")) || bytes.Contains(report, []byte("Hello world")) || bytes.Contains(report, []byte(f.provider.secret)) {
t.Fatalf("ordinary report contains payload or secret: %s", name)
}
}
if _, err := os.Stat(filepath.Join(f.dir, "runs")); !os.IsNotExist(err) {
t.Fatalf("debug created by default: %v", err)
@@ -198,24 +241,43 @@ func TestSubprocessStdoutAndSelectedReportDirectory(t *testing.T) {
}
func TestSubprocessReportingAndDebugSelections(t *testing.T) {
for _, debug := range []bool{false, true} {
var expectedOutput, expectedSummary string
for _, selection := range []struct{ reporting, debug bool }{
{false, false}, {false, true}, {true, false}, {true, true},
} {
f := newProcessFixture(t, "normal", "[grammar]")
extra := []string{"--reporting=false", "--report-dir", filepath.Join(f.dir, "missing")}
if debug {
if selection.reporting {
extra = []string{"--reporting=true", "--report-dir", filepath.Join(f.dir, "missing")}
}
if selection.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)
t.Fatalf("selection=%+v code=%d stdout=%q stderr=%q", selection, code, stdout, stderr)
}
if _, err := os.Stat(filepath.Join(f.dir, "errors.yml")); !os.IsNotExist(err) {
t.Fatalf("disabled report created: %v", err)
output, err := os.ReadFile(f.output)
if err != nil {
t.Fatal(err)
}
summary := strings.SplitN(stderr, "\n", 2)[0]
if expectedOutput == "" {
expectedOutput, expectedSummary = string(output), summary
} else if string(output) != expectedOutput || summary != expectedSummary {
t.Fatalf("selection changed processing: %+v summary=%q output=%s", selection, summary, output)
}
for _, name := range []string{"errors.yml", "recoveries.yml", "report.md"} {
_, err := os.Stat(filepath.Join(f.dir, name))
if selection.reporting && err != nil || !selection.reporting && !os.IsNotExist(err) {
t.Fatalf("selection=%+v report %s: %v", selection, name, 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 err != nil || len(dirs) != boolInt(selection.debug) {
t.Fatalf("selection=%+v dirs=%v %v", selection, dirs, err)
}
if debug {
if selection.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)
@@ -233,9 +295,9 @@ func boolInt(value bool) int {
}
func TestSubprocessAllModelFailureStillCompletes(t *testing.T) {
f := newProcessFixture(t, "all-model-failure", "[grammar]")
f := newProcessFixture(t, "all-model-failure", "[grammar, grammar]")
code, stdout, stderr := f.run(t, true)
if code != 0 || stdout != "" || !strings.Contains(stderr, "1 work units with errors") {
if code != 0 || stdout != "" || !strings.Contains(stderr, "2 work units with errors") {
t.Fatalf("code=%d stdout=%q stderr=%q", code, stdout, stderr)
}
output, err := os.ReadFile(f.output)
@@ -243,7 +305,7 @@ func TestSubprocessAllModelFailureStillCompletes(t *testing.T) {
t.Fatalf("normalized output: %s %v", output, err)
}
document := f.errors(t)
if document.Status != reporting.StatusCompleted || len(document.Errors) != 1 || document.Errors[0].Scope != "work_unit" || *document.Summary.WorkUnitsWithErrors != 1 {
if document.Status != reporting.StatusCompleted || len(document.Errors) != 2 || document.Errors[0].Scope != "work_unit" || *document.Summary.WorkUnitsWithErrors != 2 || document.Errors[0].ModuleInstance == document.Errors[1].ModuleInstance || f.provider.callCount() != 2 {
t.Fatalf("model failure report: %+v", document)
}
if strings.Contains(stderr, f.provider.secret) {
@@ -251,6 +313,100 @@ func TestSubprocessAllModelFailureStillCompletes(t *testing.T) {
}
}
func TestSubprocessAllValidationFailureStillCompletes(t *testing.T) {
f := newProcessFixture(t, "all-validation-failure", "[grammar, grammar]")
code, stdout, stderr := f.run(t, true)
if code != 0 || stdout != "" || !strings.Contains(stderr, "2 work units with errors") {
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")) || bytes.Contains(output, []byte("Hello world")) {
t.Fatalf("unchecked output: %s %v", output, err)
}
document := f.errors(t)
if document.Status != reporting.StatusCompleted || len(document.Errors) != 2 || *document.Summary.WorkUnitsWithErrors != 2 || document.Errors[0].ModuleInstance == document.Errors[1].ModuleInstance || f.provider.callCount() != 4 {
t.Fatalf("validation failure report: %+v; calls=%d", document, f.provider.callCount())
}
}
func TestSubprocessRecoveredFilteringSurvivesLaterFailure(t *testing.T) {
f := newProcessFixture(t, "repair-filter-later-failure", "[grammar, grammar]")
code, _, stderr := f.run(t, true)
if code != 0 || !strings.Contains(stderr, "1 corrections applied; 1 work units with errors; 1 recoveries") {
t.Fatalf("code=%d stderr=%q errors=%+v", code, stderr, f.errors(t))
}
output, err := os.ReadFile(f.output)
if err != nil || !bytes.Contains(output, []byte("Hello world")) {
t.Fatalf("output: %s %v", output, err)
}
errors := f.errors(t)
if len(errors.Errors) != 1 || errors.Errors[0].ModuleInstance != "grammar_2" {
t.Fatalf("later failure: %+v", errors)
}
raw, err := os.ReadFile(filepath.Join(f.dir, "recoveries.yml"))
if err != nil {
t.Fatal(err)
}
var recoveries reporting.RecoveriesDocument
if err := yaml.Unmarshal(raw, &recoveries); err != nil {
t.Fatal(err)
}
if len(recoveries.Recoveries) != 1 || recoveries.Recoveries[0].ModuleInstance != "grammar_1" || recoveries.Recoveries[0].Operation != "proposal" || len(recoveries.Recoveries[0].FilteredProposals) != 1 || *recoveries.Summary.Recoveries != 1 {
t.Fatalf("combined recovery: %+v", recoveries)
}
if !strings.Contains(strings.Join(recoveries.Recoveries[0].Reasons, ","), "output_repaired") || !strings.Contains(strings.Join(recoveries.Recoveries[0].Reasons, ","), "proposals_filtered") {
t.Fatalf("recovery reasons: %+v", recoveries.Recoveries[0])
}
}
func TestSubprocessRepairedInvalidDecisionsRemainAnError(t *testing.T) {
f := newProcessFixture(t, "invalid-decision-after-repair", "[grammar]")
code, _, stderr := f.run(t, true)
if code != 0 || !strings.Contains(stderr, "0 corrections applied; 1 work units with errors; 0 recoveries") {
t.Fatalf("code=%d stderr=%q", code, stderr)
}
document := f.errors(t)
if document.Status != reporting.StatusCompleted || len(document.Errors) != 1 || len(document.Errors[0].Failures) != 1 || document.Errors[0].Failures[0].Reason != "validation_decisions_invalid" {
t.Fatalf("invalid decisions report: %+v", document)
}
output, err := os.ReadFile(f.output)
if err != nil || !bytes.Contains(output, []byte("hello world")) || bytes.Contains(output, []byte("Hello world")) {
t.Fatalf("unchecked output: %s %v", output, err)
}
}
func TestSubprocessCancellationLeavesFailedReportsWithoutTranscript(t *testing.T) {
f := newProcessFixture(t, "wait-cancel", "[grammar, grammar]")
ctx, stop := context.WithTimeout(context.Background(), 5*time.Second)
defer stop()
args := []string{"-test.run=^TestProcessSubprocessHelper$", "--", "process", f.transcript, "--config", f.config, "--glossary", f.glossary, "--output", f.output}
command := exec.CommandContext(ctx, os.Args[0], args...)
command.Env = append(os.Environ(), "AUDITA_HELPER_PROCESS=1", "AUDITA_TEST_PROVIDER_KEY="+f.provider.secret)
var stderr bytes.Buffer
command.Stderr = &stderr
if err := command.Start(); err != nil {
t.Fatal(err)
}
select {
case <-f.provider.entered:
case <-ctx.Done():
t.Fatalf("provider was not reached: %v", ctx.Err())
}
if err := command.Process.Signal(os.Interrupt); err != nil {
t.Fatal(err)
}
err := command.Wait()
if ctx.Err() != nil || err == nil {
t.Fatalf("cancellation did not stop run: %v %v stderr=%q", err, ctx.Err(), stderr.String())
}
if document := f.errors(t); document.Status != reporting.StatusFailed || len(document.Errors) == 0 {
t.Fatalf("cancellation report: %+v", document)
}
if _, err := os.Stat(f.output); !os.IsNotExist(err) {
t.Fatalf("canceled run published transcript: %v", err)
}
}
func TestSubprocessPartialValidationAndLaterStage(t *testing.T) {
f := newProcessFixture(t, "partial-validation", "[grammar, grammar]")
code, _, stderr := f.run(t, true)