From d642791c109db9dc9c5d51580c0dff0fb9e1bc05 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sun, 2 Aug 2026 05:39:51 +0000 Subject: [PATCH] Add concurrent comparison profile execution --- internal/app/comparison_execution.go | 169 ++++++++++++++ internal/app/comparison_execution_test.go | 271 ++++++++++++++++++++++ 2 files changed, 440 insertions(+) create mode 100644 internal/app/comparison_execution.go create mode 100644 internal/app/comparison_execution_test.go diff --git a/internal/app/comparison_execution.go b/internal/app/comparison_execution.go new file mode 100644 index 0000000..24fe7d0 --- /dev/null +++ b/internal/app/comparison_execution.go @@ -0,0 +1,169 @@ +package app + +import ( + "context" + "errors" + "fmt" + "sync" + + "gitea.maximumdirect.net/eric/weatherreporter/internal/comparison" + "gitea.maximumdirect.net/eric/weatherreporter/internal/promptdebug" + "gitea.maximumdirect.net/eric/weatherreporter/internal/promptexec" +) + +type comparisonExecutionRequest struct { + Prepared preparedReport + Inspection ComparisonInspectionResult + ComparisonID string + DebugWriter *promptdebug.PromptDebugWriter + Executor promptexec.Executor +} + +type comparisonExecutionResult struct { + Outcomes []comparisonProfileOutcome + Canceled bool +} + +type comparisonProfileOutcome struct { + Position int + ProfileID string + BackendID string + ModelName string + Status string + ValidationStatus promptexec.ValidationStatus + ReportPath string + Markdown []byte + LLMDebugPath string + Error *comparison.SafeError + err error +} + +func executeComparisonProfiles(ctx context.Context, req comparisonExecutionRequest) comparisonExecutionResult { + profiles := req.Inspection.Profiles + result := comparisonExecutionResult{Outcomes: make([]comparisonProfileOutcome, len(profiles))} + for index, profile := range profiles { + result.Outcomes[index] = comparisonProfileOutcome{ + Position: index + 1, + ProfileID: profile.ProfileID, + BackendID: profile.BackendID, + ModelName: profile.ModelName, + Status: comparison.StatusFailed, + } + } + + var waitGroup sync.WaitGroup + for index, profile := range profiles { + if err := ctx.Err(); err != nil { + result.Canceled = true + markUnstartedComparisonOutcomes(result.Outcomes[index:], err) + break + } + index, profile := index, profile + waitGroup.Add(1) + go func() { + defer waitGroup.Done() + result.Outcomes[index] = executeComparisonProfile(ctx, req, index, profile) + }() + } + waitGroup.Wait() + if err := ctx.Err(); err != nil { + result.Canceled = true + for index := range result.Outcomes { + if result.Outcomes[index].Status != comparison.StatusSucceeded { + markCanceledComparisonOutcome(&result.Outcomes[index], err) + } + } + } + return result +} + +func executeComparisonProfile(ctx context.Context, req comparisonExecutionRequest, index int, profile ComparisonProfileInspection) comparisonProfileOutcome { + position := index + 1 + outcome := comparisonProfileOutcome{ + Position: position, ProfileID: profile.ProfileID, BackendID: profile.BackendID, ModelName: profile.ModelName, + Status: comparison.StatusFailed, + } + debugRef := promptdebug.PromptDebugRef{ + ReportID: req.Prepared.resolved.Definition.ID, + ValidDate: req.Prepared.resolved.ValidPeriod.Start.Format("2006-01-02"), + RunID: comparisonDebugRunID(req.ComparisonID, position, len(req.Inspection.Profiles), profile.ProfileID), + } + execution, markdown, err := executePreparedProfile(ctx, profileExecutionRequest{ + Prepared: req.Prepared, + Prompt: PromptInspectionResult{ + PromptID: req.Inspection.PromptID, PromptVersion: req.Inspection.PromptVersion, PromptHash: req.Inspection.PromptHash, + }, + Profile: promptexec.ProfileInspection{ProfileID: profile.ProfileID, BackendID: profile.BackendID, ModelName: profile.ModelName}, + Executor: req.Executor, DebugWriter: req.DebugWriter, DebugRef: &debugRef, + }) + outcome.ProfileID, outcome.BackendID, outcome.ModelName = execution.ProfileID, execution.BackendID, execution.ModelName + outcome.ValidationStatus = execution.ValidationStatus + outcome.LLMDebugPath = execution.LLMDebugPath + if err != nil { + outcome.err = err + safe := comparisonSafeExecutionError(err) + outcome.Error = &safe + return outcome + } + reportPath, err := comparison.ReportFilename(position, len(req.Inspection.Profiles), profile.ProfileID) + if err != nil { + outcome.err = err + safe := comparison.NewSafeError("application", "derive comparison report filename failed") + outcome.Error = &safe + return outcome + } + outcome.Status = comparison.StatusSucceeded + outcome.ReportPath = reportPath + outcome.Markdown = append([]byte(nil), markdown...) + return outcome +} + +func comparisonDebugRunID(comparisonID string, position, profileCount int, profileID string) string { + return fmt.Sprintf("%s_%0*d-%s", comparisonID, comparison.OrdinalWidth(profileCount), position, comparison.ProfileSlug(profileID)) +} + +func markUnstartedComparisonOutcomes(outcomes []comparisonProfileOutcome, err error) { + for index := range outcomes { + markCanceledComparisonOutcome(&outcomes[index], err) + } +} + +func markCanceledComparisonOutcome(outcome *comparisonProfileOutcome, err error) { + outcome.Status = comparison.StatusFailed + outcome.ValidationStatus = promptexec.ValidationSkipped + outcome.ReportPath = "" + outcome.Markdown = nil + outcome.err = err + safe := comparisonSafeExecutionError(err) + outcome.Error = &safe +} + +func comparisonSafeExecutionError(err error) comparison.SafeError { + category := promptexec.CategoryOf(err) + if category == "" { + switch { + case errors.Is(err, context.Canceled): + category = promptexec.Canceled + case errors.Is(err, context.DeadlineExceeded): + category = promptexec.DeadlineExceeded + } + } + if category == "" { + return comparison.NewSafeError("application", comparisonExecutionMessage(err)) + } + return comparison.NewSafeError(string(category), comparisonExecutionMessage(err)) +} + +func comparisonExecutionMessage(err error) string { + if errors.Is(err, context.Canceled) { + return "profile execution canceled" + } + if errors.Is(err, context.DeadlineExceeded) { + return "profile execution deadline exceeded" + } + var execution *profileExecutionError + if errors.As(err, &execution) { + return comparison.TruncateErrorMessage(execution.operation + " failed") + } + return "profile execution failed" +} diff --git a/internal/app/comparison_execution_test.go b/internal/app/comparison_execution_test.go new file mode 100644 index 0000000..bde5154 --- /dev/null +++ b/internal/app/comparison_execution_test.go @@ -0,0 +1,271 @@ +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) + 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) + 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.err == nil || 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) + 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) + 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 + 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)), 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 { + 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) 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) { + t.Helper() + seen := map[string]struct{}{} + for range profiles { + profileID := <-executor.started + 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)