Validate correction repair through the processing command
This commit is contained in:
334
cmd/audita/correction_repair_integration_test.go
Normal file
334
cmd/audita/correction_repair_integration_test.go
Normal file
@@ -0,0 +1,334 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/core/reporting"
|
||||
)
|
||||
|
||||
type repairWireMessage struct {
|
||||
Role string `json:"role"`
|
||||
Content string `json:"content"`
|
||||
}
|
||||
type repairWireRequest struct {
|
||||
Messages []repairWireMessage `json:"messages"`
|
||||
ResponseFormat struct {
|
||||
JSONSchema struct {
|
||||
Schema struct {
|
||||
Properties map[string]any `json:"properties"`
|
||||
} `json:"schema"`
|
||||
} `json:"json_schema"`
|
||||
} `json:"response_format"`
|
||||
}
|
||||
type correctionProvider struct {
|
||||
mu sync.Mutex
|
||||
requests []repairWireRequest
|
||||
attempts map[int]int
|
||||
entered chan struct{}
|
||||
wait bool
|
||||
}
|
||||
|
||||
func (p *correctionProvider) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
var req repairWireRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
http.Error(w, "bad request", 400)
|
||||
return
|
||||
}
|
||||
_, repair := req.ResponseFormat.JSONSchema.Schema.Properties["proposal_ref"]
|
||||
_, validation := req.ResponseFormat.JSONSchema.Schema.Properties["validations"]
|
||||
p.mu.Lock()
|
||||
p.requests = append(p.requests, req)
|
||||
p.mu.Unlock()
|
||||
content := `{"corrections":[{"id":1,"original_text":"hello","corrected_text":"Hello","confidence":0.99},{"id":2,"original_text":"second","corrected_text":"Second","confidence":0.1},{"id":3,"original_text":"third","corrected_text":"Third","confidence":0.1}]}`
|
||||
if validation {
|
||||
var indices []int
|
||||
for _, message := range req.Messages {
|
||||
if pos := strings.LastIndex(message.Content, "Corrections to validate:"); pos >= 0 {
|
||||
var items []struct {
|
||||
CorrectionIndex int `json:"correction_index"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(strings.TrimSpace(message.Content[pos+len("Corrections to validate:"):])), &items); err != nil {
|
||||
http.Error(w, "bad validation", 400)
|
||||
return
|
||||
}
|
||||
for _, item := range items {
|
||||
indices = append(indices, item.CorrectionIndex)
|
||||
}
|
||||
}
|
||||
}
|
||||
var decisions []map[string]any
|
||||
for _, index := range indices {
|
||||
decisions = append(decisions, map[string]any{"correction_index": index, "approved": true, "confidence": 0.99, "reason": "controlled"})
|
||||
}
|
||||
raw, _ := json.Marshal(map[string]any{"validations": decisions})
|
||||
content = string(raw)
|
||||
}
|
||||
if repair {
|
||||
if p.wait {
|
||||
select {
|
||||
case p.entered <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
<-r.Context().Done()
|
||||
return
|
||||
}
|
||||
ref := 2
|
||||
if strings.Contains(req.Messages[len(req.Messages)-1].Content, `"proposal_ref":3`) {
|
||||
ref = 3
|
||||
}
|
||||
p.mu.Lock()
|
||||
p.attempts[ref]++
|
||||
attempt := p.attempts[ref]
|
||||
p.mu.Unlock()
|
||||
if ref == 3 {
|
||||
http.Error(w, `{"error":{"message":"context too long"}}`, http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
confidence := 0.1
|
||||
if attempt > 1 {
|
||||
confidence = 0.99
|
||||
}
|
||||
content = fmt.Sprintf(`{"proposal_ref":2,"action":"revise","correction":{"id":2,"original_text":"second","corrected_text":"Second","confidence":%v}}`, confidence)
|
||||
}
|
||||
raw, _ := json.Marshal(content)
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
fmt.Fprintf(w, `{"model":"fixture","choices":[{"message":{"content":%s}}]}`, raw)
|
||||
}
|
||||
func setupCorrectionFixture(t *testing.T, attempts int, replacement, wait bool) (processFixture, *correctionProvider) {
|
||||
t.Helper()
|
||||
f := newProcessFixture(t, "normal", "[grammar]")
|
||||
f.setSegments(t, "hello world", "second text", "third text")
|
||||
provider := &correctionProvider{attempts: map[int]int{}, entered: make(chan struct{}, 1), wait: wait}
|
||||
server := httptest.NewServer(provider)
|
||||
t.Cleanup(server.Close)
|
||||
resources := fmt.Sprintf("profiles:\n - id: proposal\n endpoint: %s/v1\n model: fixture\n - id: validation\n endpoint: %s/v1\n model: fixture\n", server.URL, server.URL)
|
||||
if replacement {
|
||||
resources += "collections:\n audita:\n prompts:\n - id: modules.grammar.repair\n messages:\n - role: user\n content_file: repair.txt\n output:\n format: json\n validation_mode: json_schema\n schema_path: correction_repair.v1.json\n"
|
||||
if err := os.WriteFile(filepath.Join(f.dir, "repair.txt"), []byte("OVERRIDE {{ .RepairRequestJSON }}"), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join(f.dir, "prompts.yml"), []byte(resources), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
data, err := os.ReadFile(f.config)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
text := strings.Replace(string(data), "correction_repair_attempts: 0", fmt.Sprintf("correction_repair_attempts: %d", attempts), 1)
|
||||
text += "chunking:\n target_sections: 1\n min_section_tokens: 1\nnormalization:\n max_segment_gap: 0\n"
|
||||
if err := os.WriteFile(f.config, []byte(text), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return f, provider
|
||||
}
|
||||
|
||||
func TestSubprocessCorrectionRepairConversation(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
debug, replacement bool
|
||||
}{{"default prompt without debug", false, false}, {"replacement with debug", true, true}} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
f, provider := setupCorrectionFixture(t, 2, tc.replacement, false)
|
||||
args := []string{"--verbose"}
|
||||
if tc.debug {
|
||||
args = append(args, "--debug")
|
||||
}
|
||||
code, stdout, stderr := f.run(t, true, args...)
|
||||
if code != 0 || stdout != "" {
|
||||
t.Fatalf("code=%d stdout=%s stderr=%s", code, stdout, stderr)
|
||||
}
|
||||
output := f.readTranscript(t, f.output)
|
||||
if len(output.Segments) != 3 || output.Segments[0].Text != "Hello world" || output.Segments[1].Text != "Second text" || output.Segments[2].Text != "third text" {
|
||||
t.Fatalf("transcript: %+v", output)
|
||||
}
|
||||
report, err := os.ReadFile(filepath.Join(f.dir, "report.md"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, fragment := range []string{"## Correction repair", "| 1 | 1 | 1 | 2 | applied |", "| 1 | 1 | 2 | 1 | unavailable |"} {
|
||||
if !strings.Contains(string(report), fragment) {
|
||||
t.Errorf("missing %q in %s", fragment, report)
|
||||
}
|
||||
}
|
||||
doc := f.errors(t)
|
||||
recoveries := f.recoveries(t)
|
||||
if doc.Summary.LLMOperations == nil || *doc.Summary.LLMOperations != 10 || *doc.Summary.CorrectionsApplied != 2 || *doc.Summary.Recoveries != 1 || !reflect.DeepEqual(doc.Summary, recoveries.Summary) || !strings.Contains(stderr, "2 corrections applied; 1 work units with errors; 1 recoveries") {
|
||||
t.Fatalf("counts: errors=%+v recoveries=%+v stderr=%s", doc.Summary, recoveries.Summary, stderr)
|
||||
}
|
||||
if len(doc.Errors) != 1 || len(doc.Errors[0].Failures) != 1 || doc.Errors[0].Failures[0].Operation != "correction_repair" || doc.Errors[0].Failures[0].Reason != "provider_failed" {
|
||||
t.Fatalf("errors: %+v", doc)
|
||||
}
|
||||
provider.mu.Lock()
|
||||
requests := append([]repairWireRequest(nil), provider.requests...)
|
||||
provider.mu.Unlock()
|
||||
var base []repairWireMessage
|
||||
var branches [][]repairWireMessage
|
||||
for _, req := range requests {
|
||||
if _, ok := req.ResponseFormat.JSONSchema.Schema.Properties["corrections"]; ok {
|
||||
base = req.Messages
|
||||
}
|
||||
if _, ok := req.ResponseFormat.JSONSchema.Schema.Properties["proposal_ref"]; ok {
|
||||
branches = append(branches, req.Messages)
|
||||
}
|
||||
}
|
||||
if len(base) != 2 || len(branches) != 3 || len(branches[0]) != 4 || len(branches[1]) != 6 || len(branches[2]) != 4 {
|
||||
t.Fatalf("request lengths base=%d branches=%v", len(base), branches)
|
||||
}
|
||||
for _, messages := range branches {
|
||||
for i := range base {
|
||||
if messages[i] != base[i] {
|
||||
t.Fatalf("changed prefix: %+v", messages)
|
||||
}
|
||||
}
|
||||
if messages[2].Role != "assistant" || !strings.Contains(messages[2].Content, `"corrections"`) {
|
||||
t.Fatalf("missing initial response: %+v", messages)
|
||||
}
|
||||
}
|
||||
if branches[1][3] != branches[0][3] || branches[1][4].Role != "assistant" || strings.Contains(branches[2][3].Content, `"proposal_ref":2`) {
|
||||
t.Fatalf("branch history: %+v", branches)
|
||||
}
|
||||
if tc.replacement && !strings.Contains(branches[0][3].Content, "OVERRIDE") {
|
||||
t.Fatal("replacement not used")
|
||||
}
|
||||
if tc.debug {
|
||||
_, ledger := f.debugArtifacts(t)
|
||||
if len(ledger) != 3 {
|
||||
t.Fatalf("ledger: %+v", ledger)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestSubprocessRepairDisabledAndOutputSelections(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
reporting, debug bool
|
||||
}{{"neither", false, false}, {"debug only", false, true}, {"report only", true, false}} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
f, provider := setupCorrectionFixture(t, 0, false, false)
|
||||
if !tc.reporting {
|
||||
file, err := os.OpenFile(f.config, os.O_APPEND|os.O_WRONLY, 0600)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
fmt.Fprintln(file, "reporting:\n enabled: false")
|
||||
file.Close()
|
||||
}
|
||||
var args []string
|
||||
if tc.debug {
|
||||
args = append(args, "--debug")
|
||||
}
|
||||
code, _, stderr := f.run(t, true, args...)
|
||||
if code != 0 {
|
||||
t.Fatalf("code=%d stderr=%s", code, stderr)
|
||||
}
|
||||
provider.mu.Lock()
|
||||
requests := append([]repairWireRequest(nil), provider.requests...)
|
||||
provider.mu.Unlock()
|
||||
for _, req := range requests {
|
||||
if _, ok := req.ResponseFormat.JSONSchema.Schema.Properties["proposal_ref"]; ok {
|
||||
t.Fatal("repair request with zero budget")
|
||||
}
|
||||
}
|
||||
_, err := os.Stat(filepath.Join(f.dir, "report.md"))
|
||||
if tc.reporting != (err == nil) {
|
||||
t.Fatalf("report selection: %v", err)
|
||||
}
|
||||
if tc.debug {
|
||||
_, ledger := f.debugArtifacts(t)
|
||||
if len(ledger) == 0 {
|
||||
t.Fatal("debug ledger missing")
|
||||
}
|
||||
} else if _, err := os.Stat(filepath.Join(f.dir, "runs")); !os.IsNotExist(err) {
|
||||
t.Fatalf("unexpected debug dir: %v", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestSubprocessCancellationDuringCorrectionRepair(t *testing.T) {
|
||||
f, provider := setupCorrectionFixture(t, 2, false, true)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
args := []string{"-test.run=^TestProcessSubprocessHelper$", "--", "process", f.transcript, "--config", f.config, "--output", f.output, "--debug"}
|
||||
command := exec.CommandContext(ctx, os.Args[0], args...)
|
||||
command.Env = append(os.Environ(), "AUDITA_HELPER_PROCESS=1")
|
||||
var stderr bytes.Buffer
|
||||
command.Stderr = &stderr
|
||||
if err := command.Start(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
select {
|
||||
case <-provider.entered:
|
||||
case <-ctx.Done():
|
||||
t.Fatalf("repair 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: %v ctx=%v stderr=%s", err, ctx.Err(), stderr.String())
|
||||
}
|
||||
if _, err := os.Stat(f.output); !os.IsNotExist(err) {
|
||||
t.Fatalf("transcript published: %v", err)
|
||||
}
|
||||
doc := f.errors(t)
|
||||
if doc.Status != reporting.StatusFailed || len(doc.Errors) == 0 {
|
||||
t.Fatalf("failed report: %+v", doc)
|
||||
}
|
||||
data, err := os.ReadFile(filepath.Join(f.dir, "report.md"))
|
||||
if err != nil || !strings.Contains(string(data), "| 1 | 1 | 1 | 1 | interrupted |") {
|
||||
t.Fatalf("partial report %s %v", data, err)
|
||||
}
|
||||
_, ledger := f.debugArtifacts(t)
|
||||
if len(ledger) == 0 {
|
||||
t.Fatal("partial ledger missing")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSubprocessRepairWithoutOrdinaryReporting(t *testing.T) {
|
||||
for _, debug := range []bool{false, true} {
|
||||
t.Run(fmt.Sprintf("debug=%t", debug), func(t *testing.T) {
|
||||
f, _ := setupCorrectionFixture(t, 2, false, false)
|
||||
file, err := os.OpenFile(f.config, os.O_APPEND|os.O_WRONLY, 0600)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
fmt.Fprintln(file, "reporting:\n enabled: false")
|
||||
file.Close()
|
||||
args := []string{"--verbose"}
|
||||
if debug {
|
||||
args = append(args, "--debug")
|
||||
}
|
||||
code, _, stderr := f.run(t, true, args...)
|
||||
if code != 0 || !strings.Contains(stderr, "2 corrections applied; 1 work units with errors; 1 recoveries") {
|
||||
t.Fatalf("code=%d stderr=%s", code, stderr)
|
||||
}
|
||||
if _, err := os.Stat(filepath.Join(f.dir, "report.md")); !os.IsNotExist(err) {
|
||||
t.Fatalf("unexpected report: %v", err)
|
||||
}
|
||||
if debug {
|
||||
_, ledger := f.debugArtifacts(t)
|
||||
if len(ledger) != 3 {
|
||||
t.Fatalf("ledger: %+v", ledger)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user