diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index edb2529..bed5d00 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -462,7 +462,7 @@ remains covered by integration tests. ## Stage 7 — Expose Coordination Through The Facade -**Status:** Complete; public tests await the observer composition wiring in Stage 9. +**Status:** Complete; public coordinator tests passed with observer composition. **Scope:** public types, argument checks, and ordinary-execution adapters. @@ -485,7 +485,7 @@ 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. +**Status:** Complete; focused race validation passed with observer composition. **Scope:** the subtle assembled execution lifecycle. @@ -510,6 +510,8 @@ contexts as failure guards, not timing-dependent synchronization. ## Stage 9 — Add Observer Composition +**Status:** Complete. + **Scope:** fan-out helper and final coordinator-observer wiring. 1. Implement internal synchronous fan-out over a copied function slice and the diff --git a/internal/observation/compose.go b/internal/observation/compose.go new file mode 100644 index 0000000..a58adb9 --- /dev/null +++ b/internal/observation/compose.go @@ -0,0 +1,19 @@ +package observation + +// Compose copies and synchronously invokes non-nil observers in order. +func Compose[T any, F ~func(T)](observers ...F) F { + selected := make([]F, 0, len(observers)) + for _, observer := range observers { + if observer != nil { + selected = append(selected, observer) + } + } + if len(selected) == 0 { + return nil + } + return func(event T) { + for _, observer := range selected { + observer(event) + } + } +} diff --git a/observations.go b/observations.go index 3e28f8a..f999fe8 100644 --- a/observations.go +++ b/observations.go @@ -6,6 +6,7 @@ import ( "time" "go.maximumdirect.net/promptkit/internal/domain" + "go.maximumdirect.net/promptkit/internal/observation" ) // ExecutionObserver receives immutable metadata events synchronously. Callers @@ -16,6 +17,12 @@ import ( // event, and Promptkit keeps no asynchronous queue or durable history. type ExecutionObserver func(ExecutionEvent) +// ComposeObservers returns a synchronous ordered fan-out of non-nil observers. +// It copies the input slice; a panic propagates and prevents later callbacks. +func ComposeObservers(observers ...ExecutionObserver) ExecutionObserver { + return observation.Compose[ExecutionEvent](observers...) +} + // ExecutionEventKind identifies one point in an execution's lifecycle. type ExecutionEventKind string diff --git a/observer_composition_contract_test.go b/observer_composition_contract_test.go new file mode 100644 index 0000000..73fb597 --- /dev/null +++ b/observer_composition_contract_test.go @@ -0,0 +1,78 @@ +package promptkit_test + +import ( + "context" + "reflect" + "sync" + "testing" + + "go.maximumdirect.net/promptkit" +) + +func TestComposeObserversCopiesAndOrders(t *testing.T) { + if promptkit.ComposeObservers() != nil || promptkit.ComposeObservers(nil, nil) != nil { + t.Fatal("empty composition is not nil") + } + var calls []int + first := func(promptkit.ExecutionEvent) { calls = append(calls, 1) } + second := func(promptkit.ExecutionEvent) { calls = append(calls, 2) } + values := []promptkit.ExecutionObserver{first, nil, second, first} + observer := promptkit.ComposeObservers(values...) + values[0] = second + observer(promptkit.ExecutionEvent{}) + if !reflect.DeepEqual(calls, []int{1, 2, 1}) { + t.Fatalf("calls=%v", calls) + } + defer func() { + if recover() != "panic" || len(calls) != 3 { + t.Errorf("panic or calls=%v", calls) + } + }() + promptkit.ComposeObservers(func(promptkit.ExecutionEvent) { panic("panic") }, second)(promptkit.ExecutionEvent{}) +} + +func TestCoordinateObserverCompositionAndNestedSelection(t *testing.T) { + engine := newSimpleEngine(t, "question", promptkit.WithLLMClient(&fakeLLMClient{response: &promptkit.GenerateResponse{Content: "answer"}})) + ctx := context.Background() + initial, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "prompt", CaptureExchange: true}) + if err != nil { + t.Fatal(err) + } + var sequence []string + var nested int + request := &promptkit.RunRequest{PromptID: "prompt", Observer: func(promptkit.ExecutionEvent) { sequence = append(sequence, "request") }} + result, err := engine.Coordinate(ctx, promptkit.CoordinateRequest{Start: initial.Exchange, AdditionalTurns: 1, Observer: func(promptkit.ExecutionEvent) { sequence = append(sequence, "coordinator") }, Next: func(ctx context.Context, turn promptkit.ConversationTurn) (*promptkit.RunRequest, error) { + if turn.Result == nil { + return request, nil + } + _, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "prompt", Observer: func(promptkit.ExecutionEvent) { nested++ }}) + return nil, err + }}) + if err != nil || result.Outcome != promptkit.CoordinationStopped || nested != 3 || !reflect.DeepEqual(sequence, []string{"request", "coordinator", "request", "coordinator", "request", "coordinator"}) { + t.Fatalf("result=%#v error=%v events=%v nested=%d", result, err, sequence, nested) + } + if request.ContinueFrom != nil { + t.Fatal("coordinator changed caller request") + } +} + +func TestComposedObserversSupportConcurrentRuns(t *testing.T) { + var mu sync.Mutex + count := 0 + observer := promptkit.ComposeObservers(func(promptkit.ExecutionEvent) { mu.Lock(); count++; mu.Unlock() }) + engine := newSimpleEngine(t, "question", promptkit.WithLLMClient(&preparedRecordingClient{response: &promptkit.GenerateResponse{Content: "answer"}})) + var wg sync.WaitGroup + for i := 0; i < 8; i++ { + wg.Add(1) + go func() { + defer wg.Done() + if _, err := engine.Run(context.Background(), promptkit.RunRequest{PromptID: "prompt", Observer: observer}); err != nil { + t.Error(err) + } + }() + } + wg.Wait() + if count != 24 { + t.Fatalf("event count=%d", count) + } +}