301 lines
11 KiB
Go
301 lines
11 KiB
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)
|