Expose conversation coordination through the engine
This commit is contained in:
73
coordinate.go
Normal file
73
coordinate.go
Normal file
@@ -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
|
||||||
|
}
|
||||||
61
coordinate_contract_test.go
Normal file
61
coordinate_contract_test.go
Normal file
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -462,6 +462,8 @@ remains covered by integration tests.
|
|||||||
|
|
||||||
## Stage 7 — Expose Coordination Through The Facade
|
## 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.
|
**Scope:** public types, argument checks, and ordinary-execution adapters.
|
||||||
|
|
||||||
1. Add the selected coordinator types and `Engine.Coordinate`, with GoDoc for
|
1. Add the selected coordinator types and `Engine.Coordinate`, with GoDoc for
|
||||||
|
|||||||
Reference in New Issue
Block a user