audita/internal/cli/process_ordered_integration_test.go

package cli

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"os"
	"path/filepath"
	"reflect"
	"strings"
	"sync"
	"testing"

	"gitea.maximumdirect.net/eric/audita/v3/internal/core/config"
	"gitea.maximumdirect.net/eric/audita/v3/internal/core/diagnostics"
	"gitea.maximumdirect.net/eric/audita/v3/internal/core/reporting"
	"gitea.maximumdirect.net/eric/audita/v3/internal/core/schema"
	"gitea.maximumdirect.net/eric/audita/v3/internal/framework/llm"
	commandkit "go.maximumdirect.net/commandkit"
	ck "go.maximumdirect.net/configkit"
)

type controlledTransport func(*http.Request) (*http.Response, error)

func (f controlledTransport) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }

func TestProcessingUsesPerEntryProfilesInOrder(t *testing.T) {
	var mu sync.Mutex
	var models []string
	var prompts []string
	proposalCalls := 0
	transport := controlledTransport(func(r *http.Request) (*http.Response, error) {
		var request struct {
			Model    string `json:"model"`
			Messages []struct {
				Content string `json:"content"`
			} `json:"messages"`
		}
		if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
			return nil, err
		}
		var contents string
		for _, message := range request.Messages {
			contents += message.Content
		}
		mu.Lock()
		models = append(models, request.Model)
		prompts = append(prompts, contents)
		if request.Model == "model-one" {
			proposalCalls++
		}
		firstProposalCall := proposalCalls
		mu.Unlock()
		result := `{"validations":[{"correction_index":1,"approved":true,"confidence":1,"reason":"ok"}]}`
		switch request.Model {
		case "model-one":
			result = `{"corrections":[{"id":1,"original_text":"Alyce","corrected_text":"Alice","confidence":1}]}`
			if firstProposalCall == 1 {
				result = `{"invalid":true}`
			}
		case "model-two":
			result = `{"corrections":[{"id":1,"original_text":"Sheeld","corrected_text":"Shield","confidence":1}]}`
		case "model-three":
			result = `{"corrections":[{"id":1,"original_text":"casts","corrected_text":"cast","confidence":1}]}`
		}
		encoded, _ := json.Marshal(result)
		body := fmt.Sprintf(`{"model":"fixture","choices":[{"message":{"role":"assistant","content":%s}}]}`, encoded)
		return &http.Response{StatusCode: 200, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(body)), Request: r}, nil
	})
	dir := t.TempDir()
	files := map[string]string{
		"config.yml": `version: 3
reporting: {enabled: false}
glossary_collections:
  party:
    entries: [{id: alice, name: Alice, category: person}]
  spells:
    entries: [{id: shield, name: Shield, category: spell}]
protected_terms: [party, spells]
llm: {validation_profile: review}
pipeline:
  stages:
    - module: glossary
      id: "party/../|<tag>"
      glossaries: [party]
      llm: {proposal_profile: one}
    - module: glossary
      id: spell
      glossaries: [spells]
      llm: {proposal_profile: two}
    - module: grammar
      glossaries: [party, spells]
      llm: {proposal_profile: three}
`,
		"prompts.yml":     fmt.Sprintf("profiles:\n  - {id: one, endpoint: %q, model: model-one, api_key_required: false}\n  - {id: two, endpoint: %q, model: model-two, api_key_required: false}\n  - {id: three, endpoint: %q, model: model-three, api_key_required: false}\n  - {id: review, endpoint: %q, model: review, api_key_required: false}\n", "https://example.invalid/v1", "https://example.invalid/v1", "https://example.invalid/v1", "https://example.invalid/v1"),
		"transcript.json": `[{"speaker":"Alice","start":0,"end":1,"text":"Alyce casts Sheeld."}]`,
	}
	for name, value := range files {
		if err := os.WriteFile(filepath.Join(dir, name), []byte(value), 0600); err != nil {
			t.Fatal(err)
		}
	}
	loaded, err := config.Load(ck.Input{Source: ck.File(filepath.Join(dir, "config.yml"))}, "")
	if err != nil {
		t.Fatal(err)
	}
	runtime, err := llm.NewRuntime(context.Background(), loaded, &http.Client{Transport: transport})
	if err != nil {
		t.Fatal(err)
	}
	source, err := schema.ParseSourceTranscriptJSON([]byte(files["transcript.json"]))
	if err != nil {
		t.Fatal(err)
	}
	inputs := &processInputs{Configuration: loaded, Runtime: runtime, Source: source}
	loaded.Config.Diagnostics.WorkDir = dir
	prepared, err := prepareProcess(context.Background(), processInput{Transcript: filepath.Join(dir, "transcript.json"), Debug: commandkit.Supplied[bool]{Value: true}}, inputs, &bytes.Buffer{})
	if err != nil {
		t.Fatal(err)
	}
	defer prepared.Prepared.Output.Close()
	result, err := executePipeline(context.Background(), inputs, prepared, sourceToCanonicalTranscript(source), func(error) {})
	if err != nil {
		t.Fatal(err)
	}
	if result.Transcript.Segments[0].Text != "Alice cast Shield." {
		t.Fatalf("result=%q models=%v", result.Transcript.Segments[0].Text, models)
	}
	if len(prompts) > 4 && !strings.Contains(prompts[4], "Alice casts Sheeld.") {
		t.Fatalf("second proposal missed earlier edit: %s", prompts[4])
	}
	if len(result.Snapshot.WorkUnits) != 3 || result.Snapshot.WorkUnits[0].StageID == nil || *result.Snapshot.WorkUnits[0].StageID != "party/../|<tag>" || result.Snapshot.WorkUnits[1].StageIndex != 1 || result.Snapshot.WorkUnits[2].StageID != nil {
		t.Fatalf("work unit identities: %+v", result.Snapshot.WorkUnits)
	}
	if len(result.Snapshot.Recoveries) != 1 || result.Snapshot.Recoveries[0].StageIndex != 0 || result.Snapshot.Recoveries[0].StageID == nil || *result.Snapshot.Recoveries[0].StageID != "party/../|<tag>" {
		t.Fatalf("recovery attribution: %+v", result.Snapshot.Recoveries)
	}
	markdown := string(reporting.Markdown(result.Snapshot))
	if !strings.Contains(markdown, `1: glossary (party/../\|&lt;tag&gt;)`) || !strings.Contains(markdown, "2: glossary (spell)") || !strings.Contains(markdown, "3: grammar") {
		t.Fatalf("report labels: %s", markdown)
	}
	for _, artifact := range []string{diagnostics.ArtifactCorrectionLedger, diagnostics.ArtifactLLMExecution} {
		raw, readErr := os.ReadFile(filepath.Join(prepared.DebugPath(), artifact))
		if readErr != nil {
			t.Fatal(readErr)
		}
		if !strings.Contains(string(raw), `"stage_index": 0`) || !strings.Contains(string(raw), `"stage_id": "party/../|\u003ctag\u003e"`) || !strings.Contains(string(raw), `"stage_index": 1`) || !strings.Contains(string(raw), `"stage_id": "spell"`) || !strings.Contains(string(raw), `"stage_index": 2`) {
			t.Fatalf("%s missing attribution: %s", artifact, raw)
		}
	}
	if !reflect.DeepEqual(models, []string{"model-one", "model-one", "review", "review", "model-two", "review", "review", "model-three", "review", "review"}) {
		t.Fatalf("models=%v", models)
	}
}

func TestRecoverableModelFailureKeepsNormalizedTranscript(t *testing.T) {
	dir := t.TempDir()
	files := map[string]string{
		"config.yml":      "version: 3\nreporting: {enabled: false}\nllm: {proposal_profile: local, transient_retries: 0}\npipeline:\n  stages: [{module: grammar, id: failing}]\n",
		"prompts.yml":     "profiles:\n  - {id: local, endpoint: 'https://example.invalid/v1', model: fixture, api_key_required: false}\n",
		"transcript.json": `[{"speaker":"Alice","start":0,"end":1,"text":"hello"}]`,
	}
	for name, value := range files {
		if err := os.WriteFile(filepath.Join(dir, name), []byte(value), 0600); err != nil {
			t.Fatal(err)
		}
	}
	loaded, err := config.Load(ck.Input{Source: ck.File(filepath.Join(dir, "config.yml"))}, "")
	if err != nil {
		t.Fatal(err)
	}
	transport := controlledTransport(func(r *http.Request) (*http.Response, error) {
		return &http.Response{StatusCode: http.StatusServiceUnavailable, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(`{"error":{"message":"temporarily unavailable"}}`)), Request: r}, nil
	})
	runtime, err := llm.NewRuntime(context.Background(), loaded, &http.Client{Transport: transport})
	if err != nil {
		t.Fatal(err)
	}
	source, err := schema.ParseSourceTranscriptJSON([]byte(files["transcript.json"]))
	if err != nil {
		t.Fatal(err)
	}
	inputs := &processInputs{Configuration: loaded, Runtime: runtime, Source: source}
	prepared, err := prepareProcess(context.Background(), processInput{Transcript: filepath.Join(dir, "transcript.json")}, inputs, &bytes.Buffer{})
	if err != nil {
		t.Fatal(err)
	}
	defer prepared.Prepared.Output.Close()
	result, err := executePipeline(context.Background(), inputs, prepared, sourceToCanonicalTranscript(source), func(error) {})
	if err != nil {
		t.Fatal(err)
	}
	if result.Transcript.Segments[0].Text != "hello" || len(result.Snapshot.Errors) != 1 || result.Snapshot.Errors[0].Scope != "work_unit" || result.Snapshot.Errors[0].StageID == nil || *result.Snapshot.Errors[0].StageID != "failing" {
		t.Fatalf("result=%+v errors=%+v", result.Transcript, result.Snapshot.Errors)
	}
}