diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 92b6336..f0ce29a 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -439,6 +439,8 @@ reconstruction implemented by the test itself. ## Stage 6 — Implement The Internal Sequential Coordinator +**Status:** Complete. + **Scope:** the small generic loop and its deterministic state-machine tests. 1. Add the internal loop and turn/termination values described above, injecting diff --git a/internal/usecase/coordinate.go b/internal/usecase/coordinate.go new file mode 100644 index 0000000..5b05243 --- /dev/null +++ b/internal/usecase/coordinate.go @@ -0,0 +1,66 @@ +package usecase + +import ( + "context" + + "go.maximumdirect.net/promptkit/internal/domain" +) + +// ConversationTurn is one delivery to a coordinator callback. +type ConversationTurn[R any] struct { + Exchange *domain.Exchange + Result *R + RemainingTurns int +} + +// CoordinationOutcome describes a normal coordinator termination. +type CoordinationOutcome string + +const ( + CoordinationStopped CoordinationOutcome = "stopped" + CoordinationExhausted CoordinationOutcome = "exhausted" +) + +// CoordinationResult holds a normal outcome or a completed undelivered result. +type CoordinationResult[R any] struct { + Outcome CoordinationOutcome + Undelivered *R +} + +// Coordinate advances one branch using the supplied callback and ordinary run. +func Coordinate[Q, R any](ctx context.Context, start *domain.Exchange, turns int, + next func(context.Context, ConversationTurn[R]) (*Q, error), + run func(context.Context, *domain.Exchange, *Q) (*R, *domain.Exchange, error), +) (CoordinationResult[R], error) { + current := start + var pending *R + for { + if err := ctx.Err(); err != nil { + return CoordinationResult[R]{Undelivered: pending}, err + } + request, err := next(ctx, ConversationTurn[R]{Exchange: current, Result: pending, RemainingTurns: turns}) + pending = nil + if contextErr := ctx.Err(); contextErr != nil { + return CoordinationResult[R]{}, contextErr + } + if err != nil { + return CoordinationResult[R]{}, err + } + if request == nil { + return CoordinationResult[R]{Outcome: CoordinationStopped}, nil + } + if turns == 0 { + return CoordinationResult[R]{Outcome: CoordinationExhausted}, nil + } + if err := ctx.Err(); err != nil { + return CoordinationResult[R]{}, err + } + turns-- + var nextExchange *domain.Exchange + pending, nextExchange, err = run(ctx, current, request) + if err != nil { + return CoordinationResult[R]{}, err + } + current = nextExchange + } +} diff --git a/internal/usecase/coordinate_test.go b/internal/usecase/coordinate_test.go new file mode 100644 index 0000000..77b34d4 --- /dev/null +++ b/internal/usecase/coordinate_test.go @@ -0,0 +1,109 @@ +package usecase + +import ( + "context" + "errors" + "fmt" + "testing" + + "go.maximumdirect.net/promptkit/internal/domain" +) + +func TestCoordinateBudgetAndDelivery(t *testing.T) { + for _, tc := range []struct { + name string + budget int + stop bool + outcome CoordinationOutcome + runs, callbacks int + }{ + {"zero stop", 0, true, CoordinationStopped, 0, 1}, + {"zero exhausted", 0, false, CoordinationExhausted, 0, 1}, + {"stop after last", 2, true, CoordinationStopped, 2, 3}, + {"request after last", 2, false, CoordinationExhausted, 2, 3}, + } { + t.Run(tc.name, func(t *testing.T) { + start := domain.CaptureExchange(nil, "initial", nil, "start") + runs, callbacks := 0, 0 + result, err := Coordinate(context.Background(), start, tc.budget, func(_ context.Context, turn ConversationTurn[string]) (*string, error) { + callbacks++ + if turn.RemainingTurns != tc.budget-runs || (runs == 0 && turn.Result != nil) || (runs > 0 && (turn.Result == nil || *turn.Result != fmt.Sprint(runs))) { + t.Fatalf("unexpected turn: %#v", turn) + } + if tc.stop && turn.RemainingTurns == 0 { + return nil, nil + } + value := "next" + return &value, nil + }, func(_ context.Context, current *domain.Exchange, _ *string) (*string, *domain.Exchange, error) { + runs++ + value := fmt.Sprint(runs) + return &value, domain.CaptureExchange(current, value, nil, value), nil + }) + if err != nil || result.Outcome != tc.outcome || result.Undelivered != nil || runs != tc.runs || callbacks != tc.callbacks { + t.Fatalf("result=%#v error=%v runs=%d callbacks=%d", result, err, runs, callbacks) + } + }) + } +} + +func TestCoordinateCancellationAndErrors(t *testing.T) { + start := domain.CaptureExchange(nil, "initial", nil, "start") + request := "continue" + for _, mode := range []string{"before", "after run", "inside callback", "callback error", "run error"} { + t.Run(mode, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + sentinel := errors.New("sentinel") + calls := 0 + if mode == "before" { + cancel() + } + result, err := Coordinate(ctx, start, 1, func(_ context.Context, turn ConversationTurn[string]) (*string, error) { + calls++ + if mode == "inside callback" && turn.Result != nil { + cancel() + return nil, sentinel + } + if mode == "callback error" { + return nil, fmt.Errorf("callback: %w", sentinel) + } + return &request, nil + }, func(_ context.Context, current *domain.Exchange, _ *string) (*string, *domain.Exchange, error) { + if mode == "run error" { + return nil, nil, fmt.Errorf("run: %w", sentinel) + } + value := "response" + if mode == "after run" { + cancel() + } + return &value, domain.CaptureExchange(current, "next", nil, value), nil + }) + if mode == "before" || mode == "after run" || mode == "inside callback" { + if !errors.Is(err, context.Canceled) { + t.Fatalf("error = %v", err) + } + } else if !errors.Is(err, sentinel) { + t.Fatalf("error = %v", err) + } + if result.Outcome != "" || (mode == "after run") != (result.Undelivered != nil) { + t.Fatalf("result = %#v", result) + } + if mode == "before" && calls != 0 { + t.Fatal("callback invoked after cancellation") + } + if mode == "inside callback" && calls != 2 { + t.Fatalf("callbacks = %d", calls) + } + }) + } +} + +func TestCoordinatePropagatesPanic(t *testing.T) { + defer func() { + if recover() != "callback panic" { + t.Error("panic did not propagate") + } + }() + Coordinate[string, string](context.Background(), nil, 0, func(context.Context, ConversationTurn[string]) (*string, error) { panic("callback panic") }, nil) +}