weatherreporter/internal/app/comparison_execution_test.go

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)