Observe claimed prepared executions

convert.go

@@ -18,6 +18,11 @@ func toDomainRunRequest(req RunRequest) (domain.RunRequest, error) {
 	if err != nil {
 		return domain.RunRequest{}, err
 	}
+	var observer domain.ExecutionObserver
+	if req.Observer != nil {
+		callback := req.Observer
+		observer = func(event domain.ExecutionEvent) { callback(fromDomainExecutionEvent(event)) }
+	}
 	return domain.RunRequest{
 		PromptID:         req.PromptID,
 		PromptVersion:    req.PromptVersion,
@@ -30,6 +35,7 @@ func toDomainRunRequest(req RunRequest) (domain.RunRequest, error) {
 		Validation:       toDomainOutputContractPtr(req.Validation),
 		AppendedMessages: appendedMessages,
 		TransientRetries: copyIntPtr(req.TransientRetries),
+		Observer:         observer,
 	}, nil
 }
 

docs/consumers/pkg-promptkit.md

@@ -151,6 +151,14 @@ handling policy. The
 [engine operation GoDoc](../../engine.go) own exact lifecycle, engine-binding,
 credential, cancellation, timing, and error semantics.
 
+If the request supplied `Observer`, successful preparation retains that
+function for `RunPrepared`; preparation itself emits nothing. Observation
+starts only when the owning engine successfully claims the handle. Reused,
+discarded, foreign, and invalid handles emit nothing. `Discard` clears the
+retained callback without invoking it. The function's closure is retained by
+identity, so synchronize any shared state it uses. Prepared start events can
+include identities already resolved in the frozen snapshot.
+
 ## Execute And Validate
 
 [`Engine.Run`](../../engine.go) performs the same preparation, invokes the

docs/internal/observation.md

@@ -6,7 +6,10 @@ Its event values and callback type live in `internal/domain` so internal
 requests can carry them without depending on the recorder package. The
 recorder currently supplies run IDs to the runner and receives attempt and
 retry-delay measurements from the generation coordinator. Run lifecycle
-instrumentation from ordinary `Run` at the facade and runner boundaries.
+instrumentation for ordinary `Run` runs at the facade and runner boundaries.
+Prepared execution starts only after its private payload is atomically claimed;
+the facade receives that invocation's recorder privately and emits the terminal
+event after public error mapping. Failed claims and preparation emit nothing.
 
 A recorder keeps only counters, resolved identity fields, and the selected
 callback. It has no global history, queue, goroutine, or mutex held across a

docs/internal/runner.md

@@ -140,7 +140,8 @@ through context. Prompt and profile identities are enriched only after their
 respective resolutions succeed, before admission. The facade starts observation
 before request conversion and emits its terminal event after public error
 mapping and after the runner's deferred admission release. Preparation and
-prepared execution have no observation delivery yet.
+prepared execution follows the same terminal mapping after a successful claim;
+invalid or discarded handles do not acquire an observer.
 
 Validation preparation and execution honor cancellation at every
 Promptkit-controlled boundary and do not publish a partial plan or result.

docs/roadmap/implementation.md

@@ -407,6 +407,8 @@ guidance for ordinary observations only; run root and internal race tests; commi
 
 ## Stage 9 — Observe Prepared Execution Without Weakening Its Lifecycle
 
+**Completed:** 2026-09-23.
+
 Retain the adapted observer with the effective retries in the private prepared
 payload. Claim atomically before starting observation, release the claim mutex,
 install payload cleanup, then emit started. Preparation emits nothing; successful

engine.go

@@ -473,7 +473,8 @@ func (e *Engine) Prepare(ctx context.Context, req RunRequest) (*PreparedRun, err
 // PrepareExecution returns the same error categories as [Engine.Prepare] and
 // returns no handle on error. A nil Engine returns an error matching
 // ErrInvalidConfig. The effective transient retry budget is frozen privately
-// for RunPrepared and is absent from Details.
+// for RunPrepared and is absent from Details. A supplied Observer is retained
+// privately for RunPrepared; preparation itself emits no events.
 func (e *Engine) PrepareExecution(ctx context.Context, req RunRequest) (*PreparedExecution, error) {
 	if e == nil || e.internal == nil {
 		return nil, fmt.Errorf("%w: engine is nil", ErrInvalidConfig)
@@ -590,7 +591,12 @@ func (e *Engine) run(ctx context.Context, req RunRequest) (*RunResult, error) {
 // content-validation rejection, including repair exhaustion, is returned in
 // RunResult, not as an operational
 // error. RunPrepared uses the retry choice frozen during PrepareExecution. An
-// operational error returns no partial RunResult.
+// operational error returns no partial RunResult. A retained Observer begins
+// only after a successful atomic claim, with identities from the frozen
+// snapshot, and receives a terminal event on normal return after public error
+// mapping. Invalid, foreign, discarded, or reused handles emit no events.
+// Callbacks run outside the handle and capacity locks; panics propagate and
+// do not promise a terminal event.
 func (e *Engine) RunPrepared(ctx context.Context, prepared *PreparedExecution) (*RunResult, error) {
 	if e == nil || e.internal == nil {
 		return nil, fmt.Errorf("%w: engine is nil", ErrInvalidConfig)
@@ -600,9 +606,21 @@ func (e *Engine) RunPrepared(ctx context.Context, prepared *PreparedExecution) (
 	if prepared != nil {
 		internal = prepared.internal
 	}
-	result, err := e.internal.RunPrepared(ctx, internal)
+	var recorder *observation.Recorder
+	result, err := e.internal.RunPrepared(observation.WithClaimSink(ctx, &recorder), internal)
 	if err != nil {
-		return nil, mapPublicError(err)
+		mapped := mapPublicError(err)
+		if recorder != nil {
+			terminal := observedFinish(nil, mapped)
+			recorder.Finish(observation.FinishData{Outcome: terminal.Outcome, Failure: terminal.Failure,
+				StatusCode: terminal.StatusCode, RetriesExhausted: terminal.RetriesExhausted})
+		}
+		return nil, mapped
 	}
-	return fromDomainRunResult(result), nil
+	publicResult := fromDomainRunResult(result)
+	if recorder != nil {
+		terminal := observedFinish(publicResult, nil)
+		recorder.Finish(observation.FinishData{Outcome: terminal.Outcome})
+	}
+	return publicResult, nil
 }

internal/domain/domain.go

@@ -71,6 +71,7 @@ type RunRequest struct {
 	Validation       *OutputContract
 	AppendedMessages []RenderedMessage
 	TransientRetries *int
+	Observer         ExecutionObserver `json:"-"`
 }
 
 // RunResult represents the complete result of a prompt execution run.

internal/observation/recorder.go

@@ -202,6 +202,21 @@ func FromContext(ctx context.Context) *Recorder {
 	return recorder
 }
 
+type claimSinkKey struct{}
+
+// WithClaimSink lets the facade learn which invocation actually claimed a
+// prepared handle, so it can emit the terminal event after public mapping.
+func WithClaimSink(ctx context.Context, sink **Recorder) context.Context {
+	return context.WithValue(ctx, claimSinkKey{}, sink)
+}
+
+// RegisterClaim records an invocation after its prepared handle is claimed.
+func RegisterClaim(ctx context.Context, recorder *Recorder) {
+	if sink, ok := ctx.Value(claimSinkKey{}).(**Recorder); ok && sink != nil {
+		*sink = recorder
+	}
+}
+
 type usageField struct {
 	value           int
 	known, complete bool

internal/usecase/prepared_execution.go

@@ -34,6 +34,7 @@ type preparedExecutionPayload struct {
 	validation       validate.PreparedValidation
 	directKey        string
 	transientRetries uint64
+	observer         domain.ExecutionObserver
 }
 
 // PrepareExecution completes preparation without generation or admission and
@@ -68,6 +69,7 @@ func (r *Runner) PrepareExecution(ctx context.Context, req domain.RunRequest) (*
 			validation:       operation.validation,
 			directKey:        state.effectiveModel.APIKey,
 			transientRetries: state.transientRetries,
+			observer:         req.Observer,
 		},
 	}, nil
 }
@@ -115,9 +117,28 @@ func (r *Runner) RunPrepared(ctx context.Context, prepared *PreparedExecution) (
 	}
 	defer payload.clear()
 
-	runID, err := observation.NewRunID()
-	if err != nil {
-		return nil, fmt.Errorf("failed to create run id: %w", err)
+	var recorder *observation.Recorder
+	var runID string
+	if payload.observer != nil {
+		var idErr error
+		recorder, idErr = observation.New(payload.observer)
+		observation.RegisterClaim(ctx, recorder)
+		ctx = observation.WithRecorder(ctx, recorder)
+		selected := payload.prepared
+		recorder.UpdateSelection(observation.Selection{PromptID: selected.PromptID,
+			PromptVersion: selected.PromptVersion, ProfileID: selected.SelectedProfileID,
+			BackendID: selected.SelectedBackendID, Model: selected.EffectiveModelParams.Model})
+		recorder.Start()
+		if idErr != nil {
+			return nil, fmt.Errorf("failed to create run id: %w", idErr)
+		}
+		runID = recorder.RunID()
+	} else {
+		var err error
+		runID, err = observation.NewRunID()
+		if err != nil {
+			return nil, fmt.Errorf("failed to create run id: %w", err)
+		}
 	}
 	start := time.Now().UTC()
 
@@ -173,6 +194,7 @@ func (p *preparedExecutionPayload) clear() {
 	p.validation = nil
 	p.directKey = ""
 	p.transientRetries = 0
+	p.observer = nil
 }
 
 type noOpPreparedValidation struct {

prepared_execution.go

@@ -11,6 +11,9 @@ const preparedExecutionString = "promptkit.PreparedExecution{opaque}"
 // PreparedExecution contains no supported serializable state and cannot be
 // used as a restartable job. Copying the value preserves the same shared
 // lifecycle; it does not create another execution attempt.
+// A successful PrepareExecution may retain an Observer function in this
+// handle's private payload. The callback and other private execution settings
+// are cleared on claim completion or Discard. Details never exposes them.
 type PreparedExecution struct {
 	internal *usecase.PreparedExecution
 }

prepared_observation_contract_test.go

@@ -0,0 +1,186 @@
+package promptkit_test
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+
+	"gitea.maximumdirect.net/eric/promptkit"
+)
+
+func TestPreparedObservationStartsOnlyAfterClaimAndKeepsObserver(t *testing.T) {
+	engine := newPreparedContractEngine(t, &preparedRecordingClient{response: &promptkit.GenerateResponse{Content: "ok"}}, "hello")
+	var events []promptkit.ExecutionEvent
+	request := promptkit.RunRequest{PromptID: "prepared", Observer: func(event promptkit.ExecutionEvent) { events = append(events, event) }}
+	prepared, err := engine.PrepareExecution(context.Background(), request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	request.Observer = func(promptkit.ExecutionEvent) { t.Error("later observer was invoked") }
+	if len(events) != 0 {
+		t.Fatalf("preparation emitted events: %+v", events)
+	}
+	result, err := engine.RunPrepared(context.Background(), prepared)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(events) != 3 || events[0].Kind != promptkit.ExecutionStarted || events[1].Kind != promptkit.ExecutionAttempt || events[2].Kind != promptkit.ExecutionFinished {
+		t.Fatalf("events=%+v", events)
+	}
+	if events[0].PromptID != "prepared" || events[0].ProfileID != "profile" || events[0].Model != "model" || events[0].RunID != result.RunID || events[2].RunID != result.RunID || events[2].Attempts != 1 {
+		t.Fatalf("result=%+v events=%+v", result, events)
+	}
+	if again, err := engine.RunPrepared(context.Background(), prepared); again != nil || !errors.Is(err, promptkit.ErrInvalidRequest) || len(events) != 3 {
+		t.Fatalf("reused result=%+v error=%v events=%+v", again, err, events)
+	}
+}
+
+func TestPreparedObservationInvalidForeignAndDiscardedHandlesAreSilent(t *testing.T) {
+	engine := newPreparedContractEngine(t, &preparedRecordingClient{response: &promptkit.GenerateResponse{Content: "ok"}}, "hello")
+	foreign := newPreparedContractEngine(t, &preparedRecordingClient{response: &promptkit.GenerateResponse{Content: "ok"}}, "hello")
+	var events []promptkit.ExecutionEvent
+	request := promptkit.RunRequest{PromptID: "prepared", Observer: func(event promptkit.ExecutionEvent) { events = append(events, event) }}
+	prepared, err := engine.PrepareExecution(context.Background(), request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if result, err := foreign.RunPrepared(context.Background(), prepared); result != nil || !errors.Is(err, promptkit.ErrInvalidRequest) || len(events) != 0 {
+		t.Fatalf("foreign result=%+v error=%v events=%+v", result, err, events)
+	}
+	prepared.Discard()
+	if result, err := engine.RunPrepared(context.Background(), prepared); result != nil || !errors.Is(err, promptkit.ErrInvalidRequest) || len(events) != 0 {
+		t.Fatalf("discarded result=%+v error=%v events=%+v", result, err, events)
+	}
+	if result, err := engine.RunPrepared(context.Background(), nil); result != nil || !errors.Is(err, promptkit.ErrInvalidRequest) || len(events) != 0 {
+		t.Fatalf("nil result=%+v error=%v events=%+v", result, err, events)
+	}
+}
+
+func TestPreparedObservationCountsCompletedUsageBeforeRepairFailure(t *testing.T) {
+	client := &preparedRecordingClient{responses: []*promptkit.GenerateResponse{{Content: "invalid json", Usage: promptkit.TokenUsage{PromptTokens: 4}}, nil}}
+	engine := newPreparedContractEngine(t, client, "hello")
+	var events []promptkit.ExecutionEvent
+	prepared, err := engine.PrepareExecution(context.Background(), promptkit.RunRequest{PromptID: "prepared", Validation: &promptkit.OutputContract{Format: promptkit.FormatJSON, ValidationMode: promptkit.ValidationJSON, RepairAttempts: 1}, Observer: func(event promptkit.ExecutionEvent) { events = append(events, event) }})
+	if err != nil {
+		t.Fatal(err)
+	}
+	result, err := engine.RunPrepared(context.Background(), prepared)
+	if result != nil || !errors.Is(err, promptkit.ErrLLMGenerate) || len(events) != 4 {
+		t.Fatalf("result=%+v error=%v events=%+v", result, err, events)
+	}
+	terminal := events[3]
+	if terminal.Outcome != promptkit.ExecutionOutcomeError || terminal.Failure != promptkit.ExecutionFailureMalformedResponse || terminal.Attempts != 2 || terminal.RepairCalls != 1 || terminal.Usage.PromptTokens.Value != 4 || terminal.Usage.PromptTokens.Complete {
+		t.Fatalf("terminal=%+v", terminal)
+	}
+}
+
+func TestPreparedObservationClaimRaceAndPanicCleanup(t *testing.T) {
+	engine := newPreparedContractEngine(t, &preparedRecordingClient{response: &promptkit.GenerateResponse{Content: "ok"}}, "hello")
+	var mu sync.Mutex
+	var events []promptkit.ExecutionEvent
+	prepared, err := engine.PrepareExecution(context.Background(), promptkit.RunRequest{PromptID: "prepared", Observer: func(event promptkit.ExecutionEvent) { mu.Lock(); events = append(events, event); mu.Unlock() }})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var wg sync.WaitGroup
+	var successes, invalid int
+	var countMu sync.Mutex
+	for range 8 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			_, err := engine.RunPrepared(context.Background(), prepared)
+			countMu.Lock()
+			defer countMu.Unlock()
+			if err == nil {
+				successes++
+			} else if errors.Is(err, promptkit.ErrInvalidRequest) {
+				invalid++
+			} else {
+				t.Errorf("unexpected error: %v", err)
+			}
+		}()
+	}
+	wg.Wait()
+	if successes != 1 || invalid != 7 || len(events) != 3 {
+		t.Fatalf("successes=%d invalid=%d events=%+v", successes, invalid, events)
+	}
+
+	panicking, err := engine.PrepareExecution(context.Background(), promptkit.RunRequest{PromptID: "prepared", Observer: func(event promptkit.ExecutionEvent) {
+		if event.Kind == promptkit.ExecutionStarted {
+			panic("callback")
+		}
+	}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	func() {
+		defer func() {
+			if recover() == nil {
+				t.Error("expected panic")
+			}
+		}()
+		_, _ = engine.RunPrepared(context.Background(), panicking)
+	}()
+	if result, err := engine.RunPrepared(context.Background(), panicking); result != nil || !errors.Is(err, promptkit.ErrInvalidRequest) {
+		t.Fatalf("reused after panic: result=%+v error=%v", result, err)
+	}
+}
+
+func TestPreparedObservationFinishesAfterCredentialAndCapacityFailures(t *testing.T) {
+	const envName = "PROMPTKIT_PREPARED_OBSERVATION_KEY"
+	t.Setenv(envName, "available")
+	engine, err := promptkit.NewEngine(promptkit.Config{},
+		promptkit.WithPromptFS(contractPromptFS("prepared", "profile", "hello"), "."),
+		promptkit.WithProfiles(promptkit.Profile{ID: "profile", Endpoint: "https://example.invalid/v1", Model: "model", APIKeyRequired: true}),
+		promptkit.WithLLMClient(&fakeLLMClient{response: &promptkit.GenerateResponse{Content: "ok"}}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	var events []promptkit.ExecutionEvent
+	prepared, err := engine.PrepareExecution(context.Background(), promptkit.RunRequest{PromptID: "prepared", Execution: &promptkit.ExecutionTargetOverride{APIKeyEnv: envName}, Observer: func(event promptkit.ExecutionEvent) { events = append(events, event) }})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Setenv(envName, "")
+	result, err := engine.RunPrepared(context.Background(), prepared)
+	if result != nil || !errors.Is(err, promptkit.ErrInvalidRequest) || len(events) != 2 || events[1].Failure != promptkit.ExecutionFailureInvalidRequest || events[1].Attempts != 0 {
+		t.Fatalf("credential result=%+v error=%v events=%+v", result, err, events)
+	}
+
+	release := make(chan struct{})
+	client := newCapacityGateClient(release, 2)
+	limited := newBackendCapacityEngine(t, client, 1, capacityInt(0), nil)
+	first := make(chan capacityRunResult, 1)
+	go runCapacityRequest(limited, context.Background(), promptkit.RunRequest{PromptID: "prompt"}, first)
+	awaitCapacityRequest(t, client.started)
+	events = nil
+	prepared, err = limited.PrepareExecution(context.Background(), promptkit.RunRequest{PromptID: "prompt", Observer: func(event promptkit.ExecutionEvent) { events = append(events, event) }})
+	if err != nil {
+		t.Fatal(err)
+	}
+	result, err = limited.RunPrepared(context.Background(), prepared)
+	close(release)
+	if firstResult := awaitCapacityRun(t, first); firstResult.err != nil {
+		t.Fatalf("first: %v", firstResult.err)
+	}
+	if result != nil || !errors.Is(err, promptkit.ErrCapacityExceeded) || len(events) != 2 || events[1].Failure != promptkit.ExecutionFailureCapacity || events[1].Attempts != 0 {
+		t.Fatalf("capacity result=%+v error=%v events=%+v", result, err, events)
+	}
+}
+
+func TestPreparedObservationCanceledClaimStillFinishes(t *testing.T) {
+	engine := newPreparedContractEngine(t, &preparedRecordingClient{response: &promptkit.GenerateResponse{Content: "ok"}}, "hello")
+	var events []promptkit.ExecutionEvent
+	prepared, err := engine.PrepareExecution(context.Background(), promptkit.RunRequest{PromptID: "prepared", Observer: func(event promptkit.ExecutionEvent) { events = append(events, event) }})
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	result, err := engine.RunPrepared(ctx, prepared)
+	if result != nil || !errors.Is(err, context.Canceled) || len(events) != 3 || events[1].Dispatched || events[2].Failure != promptkit.ExecutionFailureCanceled || events[2].Attempts != 0 {
+		t.Fatalf("result=%+v error=%v events=%+v", result, err, events)
+	}
+}

types.go

@@ -143,8 +143,11 @@ type RunRequest struct {
 	// Negative values are invalid for Prepare, PrepareExecution, and Run.
 	// Injected LLMClient implementations remain single-dispatch.
 	TransientRetries *int
-	// Observer receives metadata about this Run invocation synchronously. It is
+	// Observer receives metadata about an execution invocation synchronously. It is
 	// excluded from JSON and is not invoked by Prepare or PrepareExecution.
+	// Successful PrepareExecution retains the function for its RunPrepared call;
+	// Discard or a completed claim clears that reference. The closure itself is
+	// not copied, so callers must synchronize its shared state.
 	Observer ExecutionObserver `json:"-"`
 }