weatherreporter/internal/app/comparison_execution_test.go

package app

import (
	"context"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"reflect"
	"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 := make(chan comparisonExecutionResult, 1)
	go func() {
		results <- executeComparisonProfiles(context.Background(), comparisonExecutionRequest{
			Prepared: prepared, Inspection: comparisonInspection(prompt, profiles), ComparisonID: "comparison_daily", 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 := make(chan comparisonExecutionResult, 1)
	go func() {
		results <- executeComparisonProfiles(context.Background(), comparisonExecutionRequest{
			Prepared: prepared, Inspection: comparisonInspection(prompt, profiles), ComparisonID: "comparison_daily", 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 TestExecuteComparisonProfilesPropagatesCancellationAndJoins(t *testing.T) {
	prepared, prompt := preparedDailyProfile(t)
	profiles := comparisonProfiles(4)
	executor := newBarrierExecutor(profiles)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	results := make(chan comparisonExecutionResult, 1)
	go func() {
		results <- executeComparisonProfiles(ctx, comparisonExecutionRequest{
			Prepared: prepared, Inspection: comparisonInspection(prompt, profiles), ComparisonID: "comparison_daily", 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 err != nil {
		t.Fatalf("NewPromptDebugWriter() error = %v", err)
	}
	executor := newBarrierExecutor(profiles)
	results := make(chan comparisonExecutionResult, 1)
	go func() {
		results <- executeComparisonProfiles(context.Background(), comparisonExecutionRequest{
			Prepared: prepared, Inspection: comparisonInspection(prompt, profiles), ComparisonID: "comparison_daily", DebugWriter: debugWriter, 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
	inFlight         int
	maximum          int
}

func newBarrierExecutor(profiles []ComparisonProfileInspection) *barrierExecutor {
	releases := make(map[string]chan struct{}, len(profiles))
	for _, profile := range profiles {
		releases[profile.ProfileID] = make(chan struct{})
	}
	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{},
	}
}

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)
	if err := callback(promptexec.Preparation{PromptID: req.PromptID, PromptVersion: req.PromptVersion, PromptHash: "prompt-hash", ProfileID: req.ProfileID, BackendID: "backend-" + req.ProfileID, ModelName: "model-" + req.ProfileID, 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]
	e.mu.Unlock()
	if err != nil {
		return nil, err
	}
	return &promptexec.Execution{
		PromptID: req.PromptID, PromptVersion: req.PromptVersion, PromptHash: "prompt-hash",
		ProfileID: req.ProfileID, BackendID: "backend-" + req.ProfileID, ModelName: "model-" + req.ProfileID,
		StartedAt: stamp, EndedAt: stamp, RawOutput: comparisonRawOutput(),
		Validation: promptexec.NewValidation(promptexec.ValidationPassed, "json_schema", generationDefinitionForPrompt(req.PromptID).GeneratedTextSchemaID+".generated_text.schema.json", 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) 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
}

func waitForProfileStarts(t *testing.T, executor *barrierExecutor, profiles []ComparisonProfileInspection, results <-chan comparisonExecutionResult) {
	t.Helper()
	timeout := time.NewTimer(5 * time.Second)
	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)
}

var _ promptexec.Executor = (*barrierExecutor)(nil)