Add concurrent comparison profile execution

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"
+}