Add sequential conversation coordinator

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

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
+	}
+}

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)
+}