Compose execution observers in order

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

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

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
 

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