package app import ( "context" "errors" "fmt" "net/http" "os" "path/filepath" "reflect" "strings" "sync" "testing" "time" "gitea.maximumdirect.net/eric/weatherreporter/internal/comparison" "gitea.maximumdirect.net/eric/weatherreporter/internal/promptdebug" "gitea.maximumdirect.net/eric/weatherreporter/internal/promptexec" ) func TestExecuteComparisonProfilesRunsOrderedProfilesConcurrently(t *testing.T) { prepared, prompt := preparedDailyProfile(t) profiles := comparisonProfiles(10) executor := newBarrierExecutor(profiles) results := startComparisonExecution(t, context.Background(), comparisonExecutionRequest{ Prepared: prepared, Inspection: comparisonInspection(prompt, profiles), ComparisonID: "comparison_daily", Executor: executor, }, executor) waitForProfileStarts(t, executor, profiles, results) if executor.maximumInFlight() < 2 { t.Fatalf("maximum in-flight executions = %d, want overlap", executor.maximumInFlight()) } for index := len(profiles) - 1; index >= 0; index-- { executor.release(profiles[index].ProfileID) } result := <-results if result.Canceled || len(result.Outcomes) != len(profiles) { t.Fatalf("result = %#v", result) } for index, profile := range profiles { outcome := result.Outcomes[index] wantPath, err := comparison.ReportFilename(index+1, len(profiles), profile.ProfileID) if err != nil { t.Fatal(err) } if outcome.Position != index+1 || outcome.ProfileID != profile.ProfileID || outcome.Status != comparison.StatusSucceeded || outcome.ValidationStatus != promptexec.ValidationPassed || outcome.ReportPath != wantPath || len(outcome.Markdown) == 0 || outcome.Error != nil { t.Fatalf("outcome[%d] = %#v", index, outcome) } request, ok := executor.request(profile.ProfileID) if !ok || request.PromptVersion != prompt.PromptVersion || !bytesEqual(request.DataPackage, prepared.dataPackage) { t.Fatalf("request for %q = %#v, want prompt version %q and shared data package", profile.ProfileID, request, prompt.PromptVersion) } } } func TestExecuteComparisonProfilesContinuesAfterProfileFailure(t *testing.T) { prepared, prompt := preparedDailyProfile(t) profiles := comparisonProfiles(3) executor := newBarrierExecutor(profiles) executor.setError(profiles[1].ProfileID, errors.New("provider response body must not escape")) results := startComparisonExecution(t, context.Background(), comparisonExecutionRequest{ Prepared: prepared, Inspection: comparisonInspection(prompt, profiles), ComparisonID: "comparison_daily", Executor: executor, }, executor) waitForProfileStarts(t, executor, profiles, results) for _, profile := range profiles { executor.release(profile.ProfileID) } result := <-results if result.Canceled || result.Outcomes[0].Status != comparison.StatusSucceeded || result.Outcomes[1].Status != comparison.StatusFailed || result.Outcomes[2].Status != comparison.StatusSucceeded { t.Fatalf("outcomes = %#v", result.Outcomes) } failure := result.Outcomes[1] if failure.Error == nil || failure.Error.Category != string(promptexec.Generation) || failure.Error.Message != "execute prompt failed" || failure.ReportPath != "" || len(failure.Markdown) != 0 { t.Fatalf("failure outcome = %#v", failure) } } func TestExecuteComparisonProfilesPreservesIndependentRepairOutcomes(t *testing.T) { prepared, prompt := preparedDailyProfile(t) profiles := comparisonProfiles(4) executor := newBarrierExecutor(profiles) executor.setValidation(profiles[0].ProfileID, promptexec.ValidationPassed, 0) executor.setValidation(profiles[1].ProfileID, promptexec.ValidationPassed, 1) executor.setValidation(profiles[2].ProfileID, promptexec.ValidationFailed, 1) executor.setError(profiles[3].ProfileID, errors.New("provider failure")) results := startComparisonExecution(t, context.Background(), comparisonExecutionRequest{ Prepared: prepared, Inspection: comparisonInspection(prompt, profiles), ComparisonID: "comparison_daily", Executor: executor, }, executor) waitForProfileStarts(t, executor, profiles, results) executor.releaseAll() result := <-results wantStatuses := []string{comparison.StatusSucceeded, comparison.StatusSucceeded, comparison.StatusFailed, comparison.StatusFailed} wantValidations := []promptexec.ValidationStatus{promptexec.ValidationPassed, promptexec.ValidationPassed, promptexec.ValidationFailed, ""} wantRepairs := []*int{intPointer(0), intPointer(1), intPointer(1), nil} for index, outcome := range result.Outcomes { if outcome.Status != wantStatuses[index] || outcome.ValidationStatus != wantValidations[index] || !reflect.DeepEqual(outcome.RepairAttempts, wantRepairs[index]) { t.Fatalf("outcome[%d] = %#v, want status/validation/repairs %q/%q/%#v", index, outcome, wantStatuses[index], wantValidations[index], wantRepairs[index]) } } } func TestExecuteComparisonProfilesCapturesConcurrentProviderFailures(t *testing.T) { prepared, prompt := preparedDailyProfile(t) profiles := comparisonProfiles(2) debugWriter, err := promptdebug.NewPromptDebugWriter(t.TempDir()) if errors.Is(err, promptdebug.ErrSecureCaptureUnsupported) { t.Skipf("secure prompt debug capture is unavailable: %v", err) } if err != nil { t.Fatalf("NewPromptDebugWriter() error = %v", err) } markers := []string{"first-provider-private-marker", "second-provider-private-marker"} statuses := []int{http.StatusTooManyRequests, http.StatusServiceUnavailable} executor := newBarrierExecutor(profiles) for index, profile := range profiles { executor.setError(profile.ProfileID, promptexec.NewGenerationError(statuses[index], "provider_code", "provider_type", markers[index], nil)) } results := startComparisonExecution(t, context.Background(), comparisonExecutionRequest{ Prepared: prepared, Inspection: comparisonInspection(prompt, profiles), ComparisonID: "comparison_daily", DebugWriter: debugWriter, Executor: executor, }, executor) waitForProfileStarts(t, executor, profiles, results) executor.releaseAll() result := <-results for index, outcome := range result.Outcomes { if outcome.Status != comparison.StatusFailed || outcome.Error == nil || outcome.Error.Category != string(promptexec.Generation) || outcome.Error.Message != fmt.Sprintf("execute prompt failed (HTTP %d)", statuses[index]) || strings.Contains(outcome.Error.Message, markers[index]) || outcome.LLMDebugPath == "" { t.Fatalf("outcome[%d] = %#v", index, outcome) } failure, readErr := os.ReadFile(filepath.Join(outcome.LLMDebugPath, "failure.json")) if readErr != nil { t.Fatal(readErr) } if !strings.Contains(string(failure), markers[index]) || strings.Contains(string(failure), markers[1-index]) { t.Fatalf("failure[%d] = %s", index, failure) } } } func TestExecuteComparisonProfilesPropagatesCancellationAndJoins(t *testing.T) { prepared, prompt := preparedDailyProfile(t) profiles := comparisonProfiles(4) executor := newBarrierExecutor(profiles) ctx, cancel := context.WithCancel(context.Background()) defer cancel() results := startComparisonExecution(t, ctx, comparisonExecutionRequest{ Prepared: prepared, Inspection: comparisonInspection(prompt, profiles), ComparisonID: "comparison_daily", Executor: executor, }, executor) waitForProfileStarts(t, executor, profiles, results) cancel() result := <-results if !result.Canceled || executor.inFlightCount() != 0 { t.Fatalf("result/in-flight = %#v/%d", result, executor.inFlightCount()) } for _, outcome := range result.Outcomes { if outcome.Status != comparison.StatusFailed || outcome.Error == nil || outcome.Error.Category != string(promptexec.Canceled) || outcome.ValidationStatus != promptexec.ValidationSkipped || outcome.ReportPath != "" || len(outcome.Markdown) != 0 { t.Fatalf("canceled outcome = %#v", outcome) } } } func TestExecuteComparisonProfilesUsesDistinctDeterministicDebugReferences(t *testing.T) { prepared, prompt := preparedDailyProfile(t) profiles := []ComparisonProfileInspection{ {ProfileID: "light.one", BackendID: "local", ModelName: "light"}, {ProfileID: "deep/two", BackendID: "cloud", ModelName: "deep"}, } debugWriter, err := promptdebug.NewPromptDebugWriter(t.TempDir()) if errors.Is(err, promptdebug.ErrSecureCaptureUnsupported) { t.Skipf("secure prompt debug capture is unavailable: %v", err) } if err != nil { t.Fatalf("NewPromptDebugWriter() error = %v", err) } executor := newBarrierExecutor(profiles) results := startComparisonExecution(t, context.Background(), comparisonExecutionRequest{ Prepared: prepared, Inspection: comparisonInspection(prompt, profiles), ComparisonID: "comparison_daily", DebugWriter: debugWriter, Executor: executor, }, executor) waitForProfileStarts(t, executor, profiles, results) for _, profile := range profiles { executor.release(profile.ProfileID) } result := <-results paths := map[string]struct{}{} for index, outcome := range result.Outcomes { wantName := fmt.Sprintf("comparison_daily_%0*d-%s", comparison.OrdinalWidth(len(profiles)), index+1, comparison.ProfileSlug(outcome.ProfileID)) if filepath.Base(outcome.LLMDebugPath) != wantName { t.Fatalf("debug path = %q, want base %q", outcome.LLMDebugPath, wantName) } if _, err := os.Stat(filepath.Join(outcome.LLMDebugPath, "preparation.json")); err != nil { t.Fatalf("preparation artifact %q: %v", outcome.LLMDebugPath, err) } paths[outcome.LLMDebugPath] = struct{}{} } if len(paths) != len(profiles) { t.Fatalf("debug paths = %#v", paths) } } type barrierExecutor struct { mu sync.Mutex started chan string callbackFailures chan error releases map[string]chan struct{} requests map[string]promptexec.ExecuteRequest errors map[string]error validations map[string]promptexec.ValidationStatus repairAttempts map[string]int profiles map[string]ComparisonProfileInspection inFlight int maximum int } func newBarrierExecutor(profiles []ComparisonProfileInspection) *barrierExecutor { releases := make(map[string]chan struct{}, len(profiles)) identities := make(map[string]ComparisonProfileInspection, len(profiles)) for _, profile := range profiles { releases[profile.ProfileID] = make(chan struct{}) identities[profile.ProfileID] = profile } return &barrierExecutor{ started: make(chan string, len(profiles)), callbackFailures: make(chan error, len(profiles)), releases: releases, requests: make(map[string]promptexec.ExecuteRequest, len(profiles)), errors: map[string]error{}, validations: map[string]promptexec.ValidationStatus{}, repairAttempts: map[string]int{}, profiles: identities, } } func (e *barrierExecutor) InspectPrompt(context.Context, string, string) (promptexec.PromptInspection, error) { return promptexec.PromptInspection{}, errors.New("unexpected prompt inspection") } func (e *barrierExecutor) InspectProfile(context.Context, string) (promptexec.ProfileInspection, error) { return promptexec.ProfileInspection{}, errors.New("unexpected profile inspection") } func (e *barrierExecutor) Execute(ctx context.Context, req promptexec.ExecuteRequest, callback promptexec.PreparationCallback) (*promptexec.Execution, error) { stamp := time.Date(2026, 5, 29, 15, 0, 0, 0, time.UTC) e.mu.Lock() profile := e.profiles[req.ProfileID] e.mu.Unlock() definition := generationDefinitionForPrompt(req.PromptID) if err := callback(promptexec.Preparation{PromptID: req.PromptID, PromptVersion: req.PromptVersion, PromptHash: generationPromptHash, ProfileID: req.ProfileID, BackendID: profile.BackendID, ModelName: profile.ModelName, Output: promptexec.OutputContract{Format: "json", ValidationMode: "json_schema", SchemaPath: definition.GeneratedTextSchemaID + ".generated_text.schema.json", RepairAttempts: definition.GeneratedTextRepairAttempts}, StartedAt: stamp, EndedAt: stamp}, nil); err != nil { e.callbackFailures <- err return nil, err } e.mu.Lock() e.requests[req.ProfileID] = promptexec.ExecuteRequest{PromptID: req.PromptID, PromptVersion: req.PromptVersion, ProfileID: req.ProfileID, DataPackage: append([]byte(nil), req.DataPackage...), CaptureDebug: req.CaptureDebug} e.inFlight++ if e.inFlight > e.maximum { e.maximum = e.inFlight } release := e.releases[req.ProfileID] e.mu.Unlock() e.started <- req.ProfileID select { case <-release: case <-ctx.Done(): e.mu.Lock() e.inFlight-- e.mu.Unlock() return nil, ctx.Err() } e.mu.Lock() e.inFlight-- err := e.errors[req.ProfileID] validationStatus := e.validations[req.ProfileID] repairAttempts := e.repairAttempts[req.ProfileID] e.mu.Unlock() if err != nil { return nil, err } if validationStatus == "" { validationStatus = promptexec.ValidationPassed } return &promptexec.Execution{ PromptID: req.PromptID, PromptVersion: req.PromptVersion, PromptHash: generationPromptHash, ProfileID: req.ProfileID, BackendID: profile.BackendID, ModelName: profile.ModelName, StartedAt: stamp, EndedAt: stamp, RawOutput: comparisonRawOutput(), Validation: promptexec.NewValidation(validationStatus, "json_schema", generationDefinitionForPrompt(req.PromptID).GeneratedTextSchemaID+".generated_text.schema.json", repairAttempts, nil), }, nil } func (e *barrierExecutor) request(profileID string) (promptexec.ExecuteRequest, bool) { e.mu.Lock() defer e.mu.Unlock() request, ok := e.requests[profileID] return request, ok } func (e *barrierExecutor) setError(profileID string, err error) { e.mu.Lock() defer e.mu.Unlock() e.errors[profileID] = err } func (e *barrierExecutor) setValidation(profileID string, status promptexec.ValidationStatus, repairAttempts int) { e.mu.Lock() defer e.mu.Unlock() e.validations[profileID] = status e.repairAttempts[profileID] = repairAttempts } func (e *barrierExecutor) release(profileID string) { close(e.releases[profileID]) } func (e *barrierExecutor) releaseAll() { for _, release := range e.releases { select { case <-release: default: close(release) } } } func (e *barrierExecutor) maximumInFlight() int { e.mu.Lock() defer e.mu.Unlock() return e.maximum } func (e *barrierExecutor) inFlightCount() int { e.mu.Lock() defer e.mu.Unlock() return e.inFlight } const comparisonExecutionTestTimeout = 5 * time.Second func startComparisonExecution(t *testing.T, ctx context.Context, request comparisonExecutionRequest, executor *barrierExecutor) <-chan comparisonExecutionResult { t.Helper() results := make(chan comparisonExecutionResult, 1) finished := make(chan struct{}) t.Cleanup(func() { executor.releaseAll() timeout := time.NewTimer(comparisonExecutionTestTimeout) defer timeout.Stop() select { case <-finished: case <-timeout.C: t.Error("comparison execution workers did not finish after release") } }) go func() { defer close(finished) results <- executeComparisonProfiles(ctx, request) }() return results } func waitForProfileStarts(t *testing.T, executor *barrierExecutor, profiles []ComparisonProfileInspection, results <-chan comparisonExecutionResult) { t.Helper() timeout := time.NewTimer(comparisonExecutionTestTimeout) defer timeout.Stop() seen := map[string]struct{}{} for range profiles { var profileID string select { case profileID = <-executor.started: case err := <-executor.callbackFailures: executor.releaseAll() select { case result := <-results: t.Fatalf("comparison profile preparation failed before executor entry: %v; result: %#v", err, result) case <-timeout.C: t.Fatalf("comparison profile preparation failed before executor entry: %v; comparison did not finish", err) } case result := <-results: t.Fatalf("comparison completed before all profiles started: %#v", result) case <-timeout.C: t.Fatal("timed out waiting for comparison profile starts") } if _, duplicate := seen[profileID]; duplicate { t.Fatalf("duplicate execution start for %q", profileID) } seen[profileID] = struct{}{} } } func comparisonProfiles(count int) []ComparisonProfileInspection { profiles := make([]ComparisonProfileInspection, 0, count) for index := 1; index <= count; index++ { profiles = append(profiles, ComparisonProfileInspection{ProfileID: fmt.Sprintf("profile.%02d", index), BackendID: "backend", ModelName: "model"}) } return profiles } func comparisonInspection(prompt PromptInspectionResult, profiles []ComparisonProfileInspection) ComparisonInspectionResult { return ComparisonInspectionResult{PromptID: prompt.PromptID, PromptVersion: prompt.PromptVersion, PromptHash: prompt.PromptHash, Profiles: profiles} } func comparisonRawOutput() []byte { return []byte(`{"summary":"Showers are possible during the selected day.","forecast_discussion":["A front will keep rain chances in the forecast."],"precipitation_timing":"Rain is most likely during the afternoon."}`) } func bytesEqual(left, right []byte) bool { return reflect.DeepEqual(left, right) } func intPointer(value int) *int { return &value } var _ promptexec.Executor = (*barrierExecutor)(nil)