package llm import ( "context" "encoding/json" "errors" "fmt" "net/http" "net/http/httptest" "os" "path/filepath" "strings" "sync" "sync/atomic" "testing" "time" ck "gitea.maximumdirect.net/eric/ConfigKit" "gitea.maximumdirect.net/eric/audita/internal/core/config" "gitea.maximumdirect.net/eric/audita/internal/framework/outcomes" "gitea.maximumdirect.net/eric/audita/internal/framework/responseschema" "gitea.maximumdirect.net/eric/audita/internal/framework/structuredoutput" "gitea.maximumdirect.net/eric/audita/internal/prompts" pk "gitea.maximumdirect.net/eric/promptkit" ) type reply struct { status int content string } func adapterFixture(t *testing.T, replies []reply, retries, repairs int, hook func(*http.Request)) (*Runtime, *atomic.Int32, string) { t.Helper() var count atomic.Int32 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { call := int(count.Add(1)) if hook != nil { hook(r) } step := replies[len(replies)-1] if call <= len(replies) { step = replies[call-1] } if step.status == 0 { connection, _, hijackErr := w.(http.Hijacker).Hijack() if hijackErr != nil { t.Error(hijackErr) return } _ = connection.Close() return } if step.status != http.StatusOK { w.WriteHeader(step.status) _, _ = fmt.Fprint(w, `{"error":{"message":"provider error"}}`) return } if step.content == "MALFORMED_ENVELOPE" { _, _ = fmt.Fprint(w, `{"choices":[]}`) return } encoded, _ := json.Marshal(step.content) w.Header().Set("Content-Type", "application/json") _, _ = fmt.Fprintf(w, `{"model":"returned-model","choices":[{"message":{"content":%s}}],"usage":{"prompt_tokens":2,"completion_tokens":3,"total_tokens":5}}`, encoded) })) t.Cleanup(server.Close) t.Setenv("ADAPTER_KEY", "adapter-secret") base := fmt.Sprintf("version: 2\nbackends:\n offline:\n endpoint: %s/v1\n credential: ADAPTER_KEY\nllm:\n proposal_profile: base\n transient_retries: %d\n repair_attempts: %d\n", server.URL, retries, repairs) dir := profileLayout(t, base, "") main := filepath.Join(dir, "config.yml") profile := "id: base\nbackend: offline\nmodel: full-model\n" if err := os.WriteFile(filepath.Join(dir, "profiles", "base.yaml"), []byte(profile), 0600); err != nil { t.Fatal(err) } loaded, err := config.Load(ck.File(main), "") if err != nil { t.Fatal(err) } runtime, err := NewRuntime(context.Background(), loaded.Config, main, server.Client()) if err != nil { t.Fatal(err) } return runtime, &count, dir } func adapterRequest(schema responseschema.Key) structuredoutput.Request { id := prompts.PromptIDModuleGlossaryProposal vars := map[string]string{"TranscriptDescriptionBlock": "", "GlossaryJSON": "[]", "SectionJSON": "[]"} if schema == responseschema.ValidatorDecisionSetKey { id = prompts.PromptIDValidatorEditorialReview vars = map[string]string{"TranscriptDescriptionBlock": "", "PayloadJSON": "[]"} } return structuredoutput.Request{PromptID: id, PromptVersion: prompts.VersionV1, Variables: vars, Schema: schema, Context: structuredoutput.DiagnosticContext{ModuleKey: "glossary"}} } func TestPreparedAdapterRepairsAndClassifies(t *testing.T) { for _, schema := range []responseschema.Key{responseschema.CorrectionSetKey, responseschema.ValidatorDecisionSetKey} { valid := `{"corrections":[]}` if schema == responseschema.ValidatorDecisionSetKey { valid = `{"validations":[]}` } for _, tc := range []struct { name string replies []reply retries, repairs, attempts, repairCalls, transientRetries int malformed bool reason outcomes.ReasonCode cause error }{ {"success", []reply{{200, valid}}, 0, 0, 1, 0, 0, false, "", nil}, {"repair succeeds", []reply{{200, `{"wrong":true}`}, {200, valid}}, 0, 1, 2, 1, 0, false, "", nil}, {"repair exhausted", []reply{{200, `{"wrong":true}`}, {200, `{"wrong":true}`}}, 0, 1, 2, 1, 0, true, outcomes.ReasonInvalidOutputExhausted, nil}, {"429 recovery", []reply{{429, ""}, {200, valid}}, 1, 0, 2, 0, 1, false, "", nil}, {"5xx recovery", []reply{{503, ""}, {200, valid}}, 1, 0, 2, 0, 1, false, "", nil}, {"zero retry", []reply{{503, ""}}, 0, 0, 1, 0, 0, false, outcomes.ReasonProviderFailed, pk.ErrLLMGenerate}, {"positive retry exhausted", []reply{{503, ""}, {503, ""}}, 1, 0, 2, 0, 1, false, outcomes.ReasonProviderFailed, pk.ErrRetriesExhausted}, {"transport failure", []reply{{0, ""}}, 0, 0, 1, 0, 0, false, outcomes.ReasonTransportFailed, pk.ErrTransport}, {"malformed envelope", []reply{{200, "MALFORMED_ENVELOPE"}}, 0, 0, 1, 0, 0, false, outcomes.ReasonProviderResponseInvalid, pk.ErrMalformedResponse}, {"retry and repair", []reply{{503, ""}, {200, `{"wrong":true}`}, {503, ""}, {200, valid}}, 1, 1, 4, 1, 2, false, "", nil}, } { t.Run(string(schema)+"/"+tc.name, func(t *testing.T) { runtime, count, _ := adapterFixture(t, tc.replies, tc.retries, tc.repairs, nil) role := "proposal" if schema == responseschema.ValidatorDecisionSetKey { role = "validation" } var events []pk.ExecutionEvent var records []structuredoutput.CallRecord client, err := runtime.NewRoleClient(role, CallHooks{OnEvent: func(e pk.ExecutionEvent) { events = append(events, e) }, OnCall: func(r structuredoutput.CallRecord) { records = append(records, r) }}) if err != nil { t.Fatal(err) } var out map[string]any response, err := client.CompleteStructured(context.Background(), adapterRequest(schema), &out) if int(count.Load()) != tc.attempts { t.Fatalf("HTTP attempts=%d, want %d", count.Load(), tc.attempts) } if tc.malformed != structuredoutput.IsMalformedError(err) { t.Fatalf("malformed classification: %v", err) } if tc.cause != nil && !errors.Is(err, tc.cause) { t.Fatalf("cause %v missing from %v", tc.cause, err) } if tc.cause == nil && !tc.malformed && err != nil { t.Fatal(err) } var failure *structuredoutput.Failure if tc.reason != "" { if !errors.As(err, &failure) || failure.Class != structuredoutput.FailureRecoverable || failure.Reason != tc.reason || !structuredoutput.IsRecoverable(err) { t.Fatalf("failure classification: %#v, %v", failure, err) } if strings.Contains(err.Error(), "provider error") || strings.Contains(err.Error(), "adapter-secret") { t.Fatalf("unsafe model error: %v", err) } } if !response.Execution.Available || response.Execution.ProviderAttempts != tc.attempts || response.Execution.StructuralRepairs != tc.repairCalls || response.Execution.TransientRetries != tc.transientRetries { t.Fatalf("returned execution counts: %+v", response.Execution) } if response.Execution.RetriesExhausted != (tc.name == "positive retry exhausted") { t.Fatalf("retry exhaustion: %+v", response.Execution) } if tc.reason == outcomes.ReasonProviderFailed && (response.Execution.HTTPStatus == nil || *response.Execution.HTTPStatus != http.StatusServiceUnavailable) { t.Fatalf("provider status missing: %+v", response.Execution) } if tc.name == "zero retry" && errors.Is(err, pk.ErrRetriesExhausted) { t.Fatal("zero retry reported exhaustion") } if response.Prepared == nil || response.Prompt.DefinitionHash == "" || response.Prompt.RenderedHash == "" || response.Prompt.ProfileID != "base" || response.Prompt.BackendID != "offline" { t.Fatalf("prepared metadata missing: %+v", response) } if len(records) != 2 || records[0].Finished.IsZero() == false || records[1].Finished.IsZero() || records[1].Context.Role != role { t.Fatalf("logical records: %+v", records) } if tc.malformed && records[1].FinalContent == "" { t.Fatal("invalid final content lost") } if len(events) == 0 || events[len(events)-1].Kind != pk.ExecutionFinished || int(events[len(events)-1].RepairCalls) != tc.repairCalls || int(events[len(events)-1].TransientRetries) != tc.transientRetries { t.Fatalf("execution counts: %+v", events) } if err == nil && (out == nil || string(response.Content) != valid || response.TotalTokens != 5*(tc.repairCalls+1) || response.Model != "full-model") { t.Fatalf("decoded response: %+v, %v", response, out) } }) } } } func TestCorrectionSchemaReplacementValuesThroughAdapter(t *testing.T) { for _, tc := range []struct { name, original, replacement string valid bool repairs int }{ {name: "empty replacement", original: "uh ", replacement: `"corrected_text":""`, valid: true, repairs: 2}, {name: "spaces replacement", original: "hello", replacement: `"corrected_text":" "`, valid: true, repairs: 2}, {name: "ordinary replacement", original: " ", replacement: `"corrected_text":" "`, valid: true, repairs: 2}, {name: "missing replacement", original: "uh ", replacement: "", valid: false}, {name: "null replacement", original: "uh ", replacement: `"corrected_text":null`, valid: false}, {name: "numeric replacement", original: "uh ", replacement: `"corrected_text":1`, valid: false}, {name: "empty original", original: "", replacement: `"corrected_text":"x"`, valid: false}, } { t.Run(tc.name, func(t *testing.T) { fields := []string{`"id":1`, `"original_text":` + fmt.Sprintf("%q", tc.original), `"confidence":0.9`} if tc.replacement != "" { fields = append(fields, tc.replacement) } payload := `{"corrections":[{` + strings.Join(fields, ",") + `}]}` runtime, count, _ := adapterFixture(t, []reply{{200, payload}}, 0, tc.repairs, nil) client, err := runtime.NewRoleClient("proposal", CallHooks{}) if err != nil { t.Fatal(err) } var out map[string]any response, err := client.CompleteStructured(context.Background(), adapterRequest(responseschema.CorrectionSetKey), &out) if got := count.Load(); got != 1 { t.Fatalf("expected one provider generation, got %d", got) } if response.Execution.StructuralRepairs != 0 || response.Execution.ProviderAttempts != 1 { t.Fatalf("unexpected repair/generation counts: %+v", response.Execution) } if tc.valid { if err != nil { t.Fatal(err) } corrections := out["corrections"].([]any) item := corrections[0].(map[string]any) var expected string if err := json.Unmarshal([]byte(strings.TrimPrefix(tc.replacement, `"corrected_text":`)), &expected); err != nil { t.Fatal(err) } if item["corrected_text"] != expected || item["original_text"] != tc.original { t.Fatalf("span values changed: %+v", item) } } else if !structuredoutput.IsMalformedError(err) { t.Fatalf("expected structural rejection, got %v", err) } }) } } func TestPreparedAdapterMetadataAndTargetGuard(t *testing.T) { requestSeen := false var posted struct { Messages []structuredoutput.Message `json:"messages"` } runtime, count, _ := adapterFixture(t, []reply{{200, `{"corrections":[]}`}}, 0, 0, func(r *http.Request) { requestSeen = true if r.Header.Get("Authorization") != "Bearer adapter-secret" { t.Error("backend credential not used") } if err := json.NewDecoder(r.Body).Decode(&posted); err != nil { t.Error(err) } }) client, err := runtime.NewRoleClient("proposal", CallHooks{}) if err != nil { t.Fatal(err) } req := adapterRequest(responseschema.CorrectionSetKey) req.Variables["HardeningPolicy"] = "attacker-controlled" var out map[string]any response, err := client.CompleteStructured(context.Background(), req, &out) if err != nil { t.Fatal(err) } if !requestSeen || count.Load() != 1 || response.Prepared == nil || len(response.Prepared.Messages) == 0 { t.Fatal("prepared call missing") } if len(posted.Messages) != len(response.Prepared.Messages) { t.Fatal("prepared and dispatched message counts differ") } for i := range posted.Messages { if posted.Messages[i] != response.Prepared.Messages[i] { t.Fatal("prepared messages differ from dispatched request") } } found := false for _, msg := range response.Prepared.Messages { if strings.Contains(msg.Content, prompts.HardeningText()) { found = true } if strings.Contains(msg.Content, "attacker-controlled") { t.Fatal("reserved hardening variable was overridden") } } if !found { t.Fatal("hardening policy missing") } } func TestPreparedAdapterRejectsTargetDriftAndPreparationErrors(t *testing.T) { runtime, count, dir := adapterFixture(t, []reply{{200, `{"corrections":[]}`}}, 0, 0, nil) client, err := runtime.NewRoleClient("proposal", CallHooks{}) if err != nil { t.Fatal(err) } req := adapterRequest(responseschema.CorrectionSetKey) req.PromptID = "missing.prompt" var out map[string]any if _, err := client.CompleteStructured(context.Background(), req, &out); err == nil || count.Load() != 0 { t.Fatalf("preparation failure dispatched: %v, %d", err, count.Load()) } // PrepareExecution reloads the profile, so a changed target must be blocked. if err := os.WriteFile(filepath.Join(dir, "profiles", "base.yaml"), []byte("id: base\nbackend: offline\nendpoint: http://127.0.0.1:1/v1\nmodel: full-model\n"), 0600); err != nil { t.Fatal(err) } req = adapterRequest(responseschema.CorrectionSetKey) _, err = client.CompleteStructured(context.Background(), req, &out) if err == nil || !strings.Contains(err.Error(), "target changed") || count.Load() != 0 { t.Fatalf("target drift dispatched: %v, %d", err, count.Load()) } } func TestPreparedAdapterCancellationStopsRetryWait(t *testing.T) { runtime, count, _ := adapterFixture(t, []reply{{503, ""}}, 2, 0, nil) ctx, cancel := context.WithCancel(context.Background()) defer cancel() var once sync.Once client, err := runtime.NewRoleClient("proposal", CallHooks{OnEvent: func(e pk.ExecutionEvent) { if e.Kind == pk.ExecutionAttempt && e.Dispatched { once.Do(cancel) } }}) if err != nil { t.Fatal(err) } var out map[string]any _, err = client.CompleteStructured(ctx, adapterRequest(responseschema.CorrectionSetKey), &out) if !errors.Is(err, context.Canceled) || count.Load() != 1 { t.Fatalf("cancellation: %v, HTTP calls=%d", err, count.Load()) } } func TestPreparedAdapterDistinguishesOperationTimeoutFromParentCancellation(t *testing.T) { blocking := func(*http.Request) { time.Sleep(500 * time.Millisecond) } runtime, _, _ := adapterFixture(t, []reply{{200, `{"corrections":[]}`}}, 0, 0, blocking) runtime.Config.LLM.OperationTimeout = 200 * time.Millisecond client, err := runtime.NewRoleClient("proposal", CallHooks{}) if err != nil { t.Fatal(err) } var output map[string]any response, err := client.CompleteStructured(context.Background(), adapterRequest(responseschema.CorrectionSetKey), &output) var failure *structuredoutput.Failure if !errors.As(err, &failure) || failure.Class != structuredoutput.FailureRecoverable || failure.Reason != outcomes.ReasonOperationTimeout || !response.Execution.Available { t.Fatalf("local timeout: failure=%#v counts=%+v", failure, response.Execution) } parent, cancel := context.WithCancel(context.Background()) cancel() response, err = client.CompleteStructured(parent, adapterRequest(responseschema.CorrectionSetKey), &output) if !errors.As(err, &failure) || failure.Class != structuredoutput.FailureCanceled || failure.Reason != outcomes.ReasonCanceled || !errors.Is(err, context.Canceled) || response.Execution.Available { t.Fatalf("parent cancellation: failure=%#v counts=%+v", failure, response.Execution) } } func TestPreparedAdapterClassifiesSetupAndDomainDecodeFailures(t *testing.T) { runtime, _, _ := adapterFixture(t, []reply{{200, `{"validations":[{"correction_index":9999999999999999999999999,"approved":true,"confidence":0.8,"reason":"ok"}]}`}}, 0, 0, nil) client, err := runtime.NewRoleClient("validation", CallHooks{}) if err != nil { t.Fatal(err) } var output struct { Validations []struct { CorrectionIndex int `json:"correction_index"` } `json:"validations"` } response, err := client.CompleteStructured(context.Background(), adapterRequest(responseschema.ValidatorDecisionSetKey), &output) var failure *structuredoutput.Failure if !errors.As(err, &failure) || failure.Class != structuredoutput.FailureRecoverable || failure.Reason != outcomes.ReasonInvalidOutput || !response.Execution.Available { t.Fatalf("decode overflow: failure=%#v counts=%+v", failure, response.Execution) } _, err = client.CompleteStructured(context.Background(), adapterRequest(responseschema.CorrectionSetKey), nil) if !errors.As(err, &failure) || failure.Class != structuredoutput.FailureFatal || failure.Reason != outcomes.ReasonSetupFailure { t.Fatalf("invalid target: %#v", failure) } req := adapterRequest("missing_schema") _, err = client.CompleteStructured(context.Background(), req, &output) if !errors.As(err, &failure) || failure.Class != structuredoutput.FailureFatal || failure.Reason != outcomes.ReasonSetupFailure { t.Fatalf("unknown schema: %#v", failure) } req = adapterRequest(responseschema.ValidatorDecisionSetKey) req.PromptID = "missing.prompt" _, err = client.CompleteStructured(context.Background(), req, &output) if !errors.As(err, &failure) || failure.Class != structuredoutput.FailureFatal || failure.Reason != outcomes.ReasonSetupFailure { t.Fatalf("missing prompt: %#v", failure) } } func TestPreparedAdapterNonRetryableProviderStatus(t *testing.T) { runtime, count, _ := adapterFixture(t, []reply{{http.StatusBadRequest, ""}}, 2, 0, nil) client, err := runtime.NewRoleClient("proposal", CallHooks{}) if err != nil { t.Fatal(err) } var output map[string]any response, err := client.CompleteStructured(context.Background(), adapterRequest(responseschema.CorrectionSetKey), &output) var failure *structuredoutput.Failure if !errors.As(err, &failure) || failure.Class != structuredoutput.FailureRecoverable || failure.Reason != outcomes.ReasonProviderFailed || count.Load() != 1 || response.Execution.HTTPStatus == nil || *response.Execution.HTTPStatus != http.StatusBadRequest { t.Fatalf("non-retryable provider failure: failure=%#v calls=%d counts=%+v", failure, count.Load(), response.Execution) } } func TestPreparedAdapterUnknownValidationFailureRemainsFatal(t *testing.T) { parent, cancel := context.WithCancel(context.Background()) defer cancel() cause := fmt.Errorf("%w: embedded contract defect", pk.ErrValidation) failure := classifiedFailure(parent, parent, cause, nil) if failure.Class != structuredoutput.FailureFatal || failure.Reason != outcomes.ReasonSetupFailure || !errors.Is(failure, pk.ErrValidation) { t.Fatalf("unknown validation failure: %#v", failure) } }