From a07e95fd178fb7f01aced5f51137a0ef04d43273 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sat, 14 Nov 2026 01:25:47 +0000 Subject: [PATCH] Cover coordinator lifecycle boundaries --- coordinate_lifecycle_contract_test.go | 170 ++++++++++++++++++++++++++ docs/roadmap/implementation.md | 2 + 2 files changed, 172 insertions(+) create mode 100644 coordinate_lifecycle_contract_test.go diff --git a/coordinate_lifecycle_contract_test.go b/coordinate_lifecycle_contract_test.go new file mode 100644 index 0000000..8604754 --- /dev/null +++ b/coordinate_lifecycle_contract_test.go @@ -0,0 +1,170 @@ +package promptkit_test + +import ( + "context" + "errors" + "net/http" + "testing" + "time" + + "go.maximumdirect.net/promptkit" +) + +func TestCoordinateReleasesCapacityBeforeCallback(t *testing.T) { + client := &preparedRecordingClient{responses: []*promptkit.GenerateResponse{{Content: "initial"}, {Content: ""}, {Content: " "}, {Content: "validator"}}} + engine := newBackendCapacityEngine(t, client, 1, capacityInt(0), nil) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + initial, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "prompt", CaptureExchange: true}) + if err != nil { + t.Fatal(err) + } + calls := 0 + result, err := engine.Coordinate(ctx, promptkit.CoordinateRequest{Start: initial.Exchange, AdditionalTurns: 1, Next: func(ctx context.Context, turn promptkit.ConversationTurn) (*promptkit.RunRequest, error) { + calls++ + if turn.Result == nil { + return &promptkit.RunRequest{PromptID: "prompt", Validation: &promptkit.OutputContract{Format: promptkit.FormatText, ValidationMode: promptkit.ValidationBasic, RepairAttempts: 1}}, nil + } + if turn.Result.Validation.Status != promptkit.ValidationFailed || turn.Result.Validation.RepairAttempts != 1 || turn.RemainingTurns != 0 { + t.Fatalf("delivered turn = %#v", turn) + } + _, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "prompt"}) + return nil, err + }}) + if err != nil || result.Outcome != promptkit.CoordinationStopped || calls != 2 || len(client.snapshot()) != 4 { + t.Fatalf("outcome=%#v error=%v callbacks=%d calls=%d", result, err, calls, len(client.snapshot())) + } +} + +func TestCoordinatePreservesCapacityAndProviderErrors(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + sourceEngine := newSimpleEngine(t, "source", promptkit.WithLLMClient(&fakeLLMClient{response: &promptkit.GenerateResponse{Content: "answer"}})) + initial, err := sourceEngine.Run(ctx, promptkit.RunRequest{PromptID: "prompt", CaptureExchange: true}) + if err != nil { + t.Fatal(err) + } + release := make(chan struct{}) + client := newCapacityGateClient(release, 1) + engine := newBackendCapacityEngine(t, client, 1, capacityInt(0), nil) + done := make(chan capacityRunResult, 1) + go runCapacityRequest(engine, ctx, promptkit.RunRequest{PromptID: "prompt"}, done) + select { + case <-client.started: + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } + request := promptkit.CoordinateRequest{Start: initial.Exchange, AdditionalTurns: 1, Next: func(context.Context, promptkit.ConversationTurn) (*promptkit.RunRequest, error) { + return &promptkit.RunRequest{PromptID: "prompt"}, nil + }} + _, err = engine.Coordinate(ctx, request) + var capacity *promptkit.CapacityError + if !errors.As(err, &capacity) || capacity.BackendID != "limited" { + t.Fatalf("capacity error = %v", err) + } + close(release) + if finished := awaitCapacityRun(t, done); finished.err != nil { + t.Fatal(finished.err) + } + providerEngine := newBuiltInGenerationErrorEngine(t, http.StatusUnprocessableEntity, `{"error":{"message":"synthetic"}}`) + _, err = providerEngine.Coordinate(ctx, request) + var provider *promptkit.GenerationError + if !errors.As(err, &provider) || provider.StatusCode() != http.StatusUnprocessableEntity { + t.Fatalf("provider error = %v", err) + } +} + +func TestCoordinateRetriesAndRepairsDeliverOnce(t *testing.T) { + initialEngine := newSimpleEngine(t, "source", promptkit.WithLLMClient(&fakeLLMClient{response: &promptkit.GenerateResponse{Content: "answer"}})) + initial, err := initialEngine.Run(context.Background(), promptkit.RunRequest{PromptID: "prompt", CaptureExchange: true}) + if err != nil { + t.Fatal(err) + } + responses := []*http.Response{observedHTTPResponse(503, `{}`), observedHTTPResponse(200, `{"choices":[{"message":{"content":"invalid"}}]}`), observedHTTPResponse(200, `{"choices":[{"message":{"content":"{\"ok\":true}"}}]}`)} + calls := 0 + one := 1 + config := simpleConfig("follow-up") + config.TransientRetries = &one + config.HTTPClient = &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) { + response := responses[calls] + calls++ + return response, nil + })} + engine, err := promptkit.NewEngine(config) + if err != nil { + t.Fatal(err) + } + callbacks := 0 + result, err := engine.Coordinate(context.Background(), promptkit.CoordinateRequest{Start: initial.Exchange, AdditionalTurns: 1, Next: func(_ context.Context, turn promptkit.ConversationTurn) (*promptkit.RunRequest, error) { + callbacks++ + if turn.Result == nil { + return &promptkit.RunRequest{PromptID: "prompt", Validation: &promptkit.OutputContract{Format: promptkit.FormatJSON, ValidationMode: promptkit.ValidationJSON, RepairAttempts: 1}}, nil + } + if !turn.Result.Validation.IsValid || turn.RemainingTurns != 0 { + t.Fatalf("turn = %#v", turn) + } + return nil, nil + }}) + if err != nil || result.Outcome != promptkit.CoordinationStopped || callbacks != 2 || calls != 3 { + t.Fatalf("result=%#v error=%v callbacks=%d attempts=%d", result, err, callbacks, calls) + } +} + +func TestCoordinateCancellationDelivery(t *testing.T) { + for _, cancelInside := range []bool{false, true} { + t.Run(map[bool]string{false: "before delivery", true: "inside callback"}[cancelInside], func(t *testing.T) { + client := &fakeLLMClient{response: &promptkit.GenerateResponse{Content: "answer"}} + engine := newSimpleEngine(t, "question", promptkit.WithLLMClient(client)) + initial, err := engine.Run(context.Background(), promptkit.RunRequest{PromptID: "prompt", CaptureExchange: true}) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + calls := 0 + result, err := engine.Coordinate(ctx, promptkit.CoordinateRequest{Start: initial.Exchange, AdditionalTurns: 1, Next: func(_ context.Context, turn promptkit.ConversationTurn) (*promptkit.RunRequest, error) { + calls++ + if turn.Result != nil { + cancel() + return nil, nil + } + return &promptkit.RunRequest{PromptID: "prompt", Observer: func(event promptkit.ExecutionEvent) { + if !cancelInside && event.Kind == promptkit.ExecutionFinished { + cancel() + } + }}, nil + }}) + if !errors.Is(err, context.Canceled) || result.Outcome != "" || (result.Undelivered != nil) == cancelInside { + t.Fatalf("result=%#v error=%v callbacks=%d", result, err, calls) + } + if !cancelInside && (calls != 1 || result.Undelivered.Exchange == nil) { + t.Fatalf("undelivered result=%#v calls=%d", result, calls) + } + if cancelInside && calls != 2 { + t.Fatalf("callbacks=%d", calls) + } + }) + } +} + +func TestCoordinateFailureLeavesSiblingUsable(t *testing.T) { + client := &fakeLLMClient{response: &promptkit.GenerateResponse{Content: "answer"}} + engine := newSimpleEngine(t, "question", promptkit.WithLLMClient(client)) + ctx := context.Background() + initial, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "prompt", CaptureExchange: true}) + if err != nil { + t.Fatal(err) + } + sentinel := errors.New("provider") + client.err = sentinel + result, err := engine.Coordinate(ctx, promptkit.CoordinateRequest{Start: initial.Exchange, AdditionalTurns: 1, Next: func(context.Context, promptkit.ConversationTurn) (*promptkit.RunRequest, error) { + return &promptkit.RunRequest{PromptID: "prompt"}, nil + }}) + if !errors.Is(err, sentinel) || !errors.Is(err, promptkit.ErrLLMGenerate) || result.Outcome != "" || result.Undelivered != nil { + t.Fatalf("failure=%#v, %v", result, err) + } + client.err = nil + if sibling, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "prompt", ContinueFrom: initial.Exchange}); err != nil || sibling.Exchange == nil { + t.Fatalf("sibling=%#v, %v", sibling, err) + } +} diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 93f66ed..edb2529 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -485,6 +485,8 @@ explicit observer-composition dependency forward, without an intermediate gate. ## Stage 8 — Verify Coordinator Capacity And Cancellation Boundaries +**Status:** Complete; race validation awaits the observer composition wiring in Stage 9. + **Scope:** the subtle assembled execution lifecycle. 1. Configure one backend with concurrency one and waiting capacity zero. Have the