From c4d7cf8cd42099fecfd5816eef2c7233d5b012b9 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sat, 14 Nov 2026 01:24:13 +0000 Subject: [PATCH] Expose conversation coordination through the engine --- coordinate.go | 73 ++++++++++++++++++++++++++++++++++ coordinate_contract_test.go | 61 ++++++++++++++++++++++++++++ docs/roadmap/implementation.md | 2 + 3 files changed, 136 insertions(+) create mode 100644 coordinate.go create mode 100644 coordinate_contract_test.go diff --git a/coordinate.go b/coordinate.go new file mode 100644 index 0000000..34e793a --- /dev/null +++ b/coordinate.go @@ -0,0 +1,73 @@ +package promptkit + +import ( + "context" + "fmt" + + "go.maximumdirect.net/promptkit/internal/domain" + "go.maximumdirect.net/promptkit/internal/usecase" +) + +// ConversationTurn is delivered before each requested continuation. Result is +// nil on the initial callback; RemainingTurns counts additional runs allowed. +type ConversationTurn struct { + Exchange *Exchange + Result *RunResult + RemainingTurns int +} + +// CoordinateRequest selects a starting exchange, turn budget and callback. +// Observer selects continuation runs only, independently of callback calls. +type CoordinateRequest struct { + Start *Exchange + AdditionalTurns int + Next func(context.Context, ConversationTurn) (*RunRequest, error) + Observer ExecutionObserver +} + +// CoordinationOutcome distinguishes caller-directed stopping from exhaustion. +type CoordinationOutcome string + +const ( + // CoordinationStopped means the callback chose not to continue. + CoordinationStopped CoordinationOutcome = "stopped" + // CoordinationExhausted means another run was requested without budget. + CoordinationExhausted CoordinationOutcome = "exhausted" +) + +// CoordinationResult contains the normal outcome or a response withheld by +// cancellation before delivery to the callback. +type CoordinationResult struct { + Outcome CoordinationOutcome `json:"outcome,omitempty"` + Undelivered *RunResult `json:"undelivered,omitempty"` +} + +// Coordinate advances one branch from Start using ordinary runs. Next is called +// initially even for a zero budget and receives each completed result once. +// A nil request stops; requesting a run without remaining budget exhausts it. +// Callback and execution errors retain their identities. Cancellation observed +// before delivery returns the single completed response as Undelivered; +// cancellation after callback invocation does not replay that response. +func (e *Engine) Coordinate(ctx context.Context, req CoordinateRequest) (CoordinationResult, error) { + if e == nil || e.internal == nil { + return CoordinationResult{}, fmt.Errorf("%w: engine is nil", ErrInvalidConfig) + } + if req.Start == nil || req.Start.internal == nil || req.Next == nil || req.AdditionalTurns < 0 { + return CoordinationResult{}, fmt.Errorf("%w: invalid coordination request", ErrInvalidRequest) + } + outcome, err := usecase.Coordinate(ctx, req.Start.internal, req.AdditionalTurns, + func(ctx context.Context, turn usecase.ConversationTurn[RunResult]) (*RunRequest, error) { + return req.Next(ctx, ConversationTurn{Exchange: &Exchange{internal: turn.Exchange}, Result: turn.Result, RemainingTurns: turn.RemainingTurns}) + }, + func(ctx context.Context, current *domain.Exchange, selected *RunRequest) (*RunResult, *domain.Exchange, error) { + request := *selected + request.ContinueFrom = &Exchange{internal: current} + request.Observer = ComposeObservers(request.Observer, req.Observer) + result, runErr := e.Run(ctx, request) + if runErr != nil { + return nil, nil, runErr + } + return result, result.Exchange.internal, nil + }) + return CoordinationResult{Outcome: CoordinationOutcome(outcome.Outcome), Undelivered: outcome.Undelivered}, err +} diff --git a/coordinate_contract_test.go b/coordinate_contract_test.go new file mode 100644 index 0000000..61299a5 --- /dev/null +++ b/coordinate_contract_test.go @@ -0,0 +1,61 @@ +package promptkit_test + +import ( + "context" + "errors" + "testing" + + "go.maximumdirect.net/promptkit" +) + +func TestCoordinatePublicBoundary(t *testing.T) { + ctx := context.Background() + engine := newSimpleEngine(t, "question", promptkit.WithLLMClient(&fakeLLMClient{response: &promptkit.GenerateResponse{Content: "answer"}})) + initial, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "prompt", CaptureExchange: true}) + if err != nil { + t.Fatal(err) + } + for _, request := range []promptkit.CoordinateRequest{{}, {Start: &promptkit.Exchange{}, Next: func(context.Context, promptkit.ConversationTurn) (*promptkit.RunRequest, error) { return nil, nil }}, {Start: initial.Exchange, AdditionalTurns: -1, Next: func(context.Context, promptkit.ConversationTurn) (*promptkit.RunRequest, error) { return nil, nil }}} { + if _, err := engine.Coordinate(ctx, request); !errors.Is(err, promptkit.ErrInvalidRequest) { + t.Fatalf("invalid request error = %v", err) + } + } + if _, err := (*promptkit.Engine)(nil).Coordinate(ctx, promptkit.CoordinateRequest{}); !errors.Is(err, promptkit.ErrInvalidConfig) { + t.Fatalf("nil engine error = %v", err) + } + callbacks := 0 + selected := &promptkit.RunRequest{PromptID: "prompt", ContinueFrom: &promptkit.Exchange{}} + result, err := engine.Coordinate(ctx, promptkit.CoordinateRequest{Start: initial.Exchange, AdditionalTurns: 2, Next: func(_ context.Context, turn promptkit.ConversationTurn) (*promptkit.RunRequest, error) { + callbacks++ + if callbacks == 1 { + if turn.Result != nil || turn.RemainingTurns != 2 { + t.Fatalf("initial turn = %#v", turn) + } + return selected, nil + } + if turn.Result == nil || turn.Exchange.Details().ParentRunID == "" { + t.Fatalf("delivered turn = %#v", turn) + } + if callbacks == 2 { + turn.Result.Exchange = nil + *turn.Exchange = promptkit.Exchange{} + return selected, nil + } + if turn.Result.ParentRunID == initial.RunID { + t.Fatal("callback mutation redirected branch") + } + return nil, nil + }}) + if err != nil || result.Outcome != promptkit.CoordinationStopped || callbacks != 3 || selected.ContinueFrom == nil { + t.Fatalf("result=%#v error=%v callbacks=%d", result, err, callbacks) + } + stop, err := engine.Coordinate(ctx, promptkit.CoordinateRequest{Start: initial.Exchange, Next: func(context.Context, promptkit.ConversationTurn) (*promptkit.RunRequest, error) { return nil, nil }}) + if err != nil || stop.Outcome != promptkit.CoordinationStopped { + t.Fatalf("zero budget = %#v, %v", stop, err) + } + sentinel := errors.New("callback") + _, err = engine.Coordinate(ctx, promptkit.CoordinateRequest{Start: initial.Exchange, Next: func(context.Context, promptkit.ConversationTurn) (*promptkit.RunRequest, error) { return nil, sentinel }}) + if !errors.Is(err, sentinel) { + t.Fatalf("callback error = %v", err) + } +} diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index f0ce29a..93f66ed 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -462,6 +462,8 @@ remains covered by integration tests. ## Stage 7 — Expose Coordination Through The Facade +**Status:** Complete; public tests await the observer composition wiring in Stage 9. + **Scope:** public types, argument checks, and ordinary-execution adapters. 1. Add the selected coordinator types and `Engine.Coordinate`, with GoDoc for