Prepare v0.23.0 release notes and retire completed roadmaps

README.md

@@ -45,6 +45,10 @@ boundary and constraints that framework work must preserve.
 
 ## Release Guidance
 
+The [v0.23.0 release notes](docs/releases/v0.23.0.md) cover per-operation
+timeouts and opt-in prepared-request notifications for ordinary and coordinated
+execution.
+
 The [v0.22.0 release notes](docs/releases/v0.22.0.md) cover captured conversation
 exchanges, application-directed turn coordination, and explicit execution groups.
 

convert.go

@@ -1,6 +1,7 @@
 package promptkit
 
 import (
+	"context"
 	"fmt"
 	"unicode/utf8"
 
@@ -30,7 +31,15 @@ func toDomainRunRequest(req RunRequest) (domain.RunRequest, error) {
 		callback := req.Observer
 		observer = func(event domain.ExecutionEvent) { callback(fromDomainExecutionEvent(event)) }
 	}
+	var onPrepared func(context.Context, *domain.PreparedRun)
+	if callback := req.OnPrepared; callback != nil {
+		onPrepared = func(ctx context.Context, prepared *domain.PreparedRun) {
+			callback(ctx, *fromDomainPreparedRun(prepared))
+		}
+	}
 	return domain.RunRequest{
+		OperationTimeout: req.OperationTimeout,
+		OnPrepared:       onPrepared,
 		CaptureExchange:  req.CaptureExchange || req.ContinueFrom != nil,
 		PromptID:         req.PromptID,
 		ProfileID:        req.ProfileID,

coordinate_controls_contract_test.go

@@ -0,0 +1,86 @@
+package promptkit_test
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"testing/synctest"
+	"time"
+
+	"go.maximumdirect.net/promptkit"
+)
+
+func TestCoordinateUsesIndependentOperationBudgets(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		engine := newBackendCapacityEngine(t, observationClientFunc(func(_ context.Context, req promptkit.GenerateRequest) (*promptkit.GenerateResponse, error) {
+			if req.Prompt.SessionID == "continuation" {
+				time.Sleep(2 * time.Second)
+			}
+			return &promptkit.GenerateResponse{Content: "answer", Usage: promptkit.TokenUsage{PromptTokens: 2}}, nil
+		}), 1, capacityInt(0), nil)
+		parent := context.Background()
+		initial, err := engine.Run(parent, promptkit.RunRequest{PromptID: "prompt", CaptureExchange: true})
+		if err != nil {
+			t.Fatal(err)
+		}
+		var group promptkit.ExecutionGroup
+		var child context.Context
+		var details []promptkit.PreparedRun
+		var requestEvents, coordinatorEvents []promptkit.ExecutionEvent
+		start := time.Now()
+		result, err := engine.Coordinate(parent, promptkit.CoordinateRequest{Start: initial.Exchange, AdditionalTurns: 2, Observer: promptkit.ComposeObservers(group.Observe, func(e promptkit.ExecutionEvent) { coordinatorEvents = append(coordinatorEvents, e) }), Next: func(ctx context.Context, turn promptkit.ConversationTurn) (*promptkit.RunRequest, error) {
+			if turn.Result != nil {
+				if ctx.Err() != nil || child == nil || !errors.Is(child.Err(), context.Canceled) {
+					t.Fatal("child was not cleaned up before callback")
+				}
+				if _, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "prompt", OperationTimeout: time.Second}); err != nil {
+					t.Fatalf("assessment: %v", err)
+				}
+			}
+			if turn.RemainingTurns == 0 {
+				return nil, nil
+			}
+			return &promptkit.RunRequest{PromptID: "prompt", SessionID: "continuation", OperationTimeout: 3 * time.Second, Observer: func(e promptkit.ExecutionEvent) { requestEvents = append(requestEvents, e) }, OnPrepared: func(ctx context.Context, prepared promptkit.PreparedRun) {
+				child = ctx
+				details = append(details, prepared)
+			}}, nil
+		}})
+		if err != nil || result.Outcome != promptkit.CoordinationStopped || time.Since(start) != 4*time.Second || len(details) != 2 || len(details[0].Messages) != 3 || len(details[1].Messages) != 5 {
+			t.Fatalf("result=%+v error=%v details=%v elapsed=%v", result, err, len(details), time.Since(start))
+		}
+		summary := group.Snapshot()
+		if len(requestEvents) != 6 || len(coordinatorEvents) != 6 || len(summary.Members) != 2 || summary.Attempts != 2 || summary.Usage.PromptTokens.Value != 4 || !summary.TerminalCoverageComplete {
+			t.Fatalf("summary=%+v request=%d coordinator=%d", summary, len(requestEvents), len(coordinatorEvents))
+		}
+	})
+}
+
+func TestCoordinateTimeoutLeavesParentAndSiblingUsable(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		engine := newBackendCapacityEngine(t, observationClientFunc(func(ctx context.Context, req promptkit.GenerateRequest) (*promptkit.GenerateResponse, error) {
+			if req.Prompt.SessionID == "slow" {
+				<-ctx.Done()
+				return nil, ctx.Err()
+			}
+			return &promptkit.GenerateResponse{Content: "answer"}, nil
+		}), 1, capacityInt(0), nil)
+		parent := context.Background()
+		initial, err := engine.Run(parent, promptkit.RunRequest{PromptID: "prompt", CaptureExchange: true})
+		if err != nil {
+			t.Fatal(err)
+		}
+		var details promptkit.PreparedRun
+		calls := 0
+		result, err := engine.Coordinate(parent, promptkit.CoordinateRequest{Start: initial.Exchange, AdditionalTurns: 1, Next: func(context.Context, promptkit.ConversationTurn) (*promptkit.RunRequest, error) {
+			calls++
+			return &promptkit.RunRequest{PromptID: "prompt", SessionID: "slow", OperationTimeout: time.Second, OnPrepared: func(_ context.Context, p promptkit.PreparedRun) { details = p }}, nil
+		}})
+		if !errors.Is(err, promptkit.ErrOperationTimeout) || !errors.Is(err, context.DeadlineExceeded) || calls != 1 || result.Outcome != "" || result.Undelivered != nil || details.ParentRunID != initial.RunID || parent.Err() != nil {
+			t.Fatalf("result=%+v error=%v callbacks=%d", result, err, calls)
+		}
+		sibling, err := engine.Run(parent, promptkit.RunRequest{PromptID: "prompt", ContinueFrom: initial.Exchange})
+		if err != nil || sibling.Exchange == nil {
+			t.Fatalf("sibling=%v error=%v", sibling, err)
+		}
+	})
+}

docs/consumers/pkg-promptkit.md

@@ -581,6 +581,59 @@ 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.
 
+## Bound Execution And Retain Prepared Details
+
+Select the optional controls on [RunRequest](../../types.go). The operation
+budget is separate from the provider timeout and cancellation is cooperative.
+Notification is conditional on reaching the prepared point, synchronous, and
+runs while admission is held; its caller-owned details can be retained even if
+generation later fails. Do not rely on notification for preparation or admission
+failures. The application owns any subsequent diagnostic handling.
+
+```go
+var details *promptkit.PreparedRun
+result, err := engine.Run(ctx, promptkit.RunRequest{
+	PromptID: "answer", OperationTimeout: 30 * time.Second,
+	OnPrepared: func(_ context.Context, prepared promptkit.PreparedRun) { details = &prepared },
+})
+if errors.Is(err, promptkit.ErrOperationTimeout) && errors.Is(err, context.DeadlineExceeded) {
+	// details is available only if the notification was reached.
+}
+```
+
+For coordinated turns, put the controls on each returned request. Assessment
+in the next-turn callback runs after the previous admission and child scope
+are released, so it can select its own budget:
+
+```go
+next := func(ctx context.Context, turn promptkit.ConversationTurn) (*promptkit.RunRequest, error) {
+	if turn.Result != nil {
+		_, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "assessment", OperationTimeout: 10 * time.Second})
+		return nil, err
+	}
+	return &promptkit.RunRequest{PromptID: "answer", OperationTimeout: 30 * time.Second,
+		OnPrepared: func(_ context.Context, prepared promptkit.PreparedRun) { details = &prepared }}, nil
+}
+```
+
+For split execution, [PrepareExecution](../../engine.go) retains the controls
+without starting their timer; idle time does not count. Use the same caller
+deadline for both calls if preparation and execution need one shared bound.
+
+```go
+handle, err := engine.PrepareExecution(ctx, promptkit.RunRequest{
+	PromptID: "answer", OperationTimeout: 30 * time.Second,
+	OnPrepared: func(_ context.Context, prepared promptkit.PreparedRun) { details = &prepared },
+})
+if err != nil { return err }
+defer handle.Discard()
+result, err := engine.RunPrepared(ctx, handle)
+```
+
+See the [observation example](../../examples/go-library/observe/main.go) for an
+offline coordinated workflow and [ErrOperationTimeout](../../engine.go) for
+timeout attribution.
+
 ## Execute And Validate
 
 [`Engine.Run`](../../engine.go) performs the same preparation, invokes the

docs/internal/observation.md

@@ -14,6 +14,9 @@ The public projection retains private operation-owner metadata separately from
 inspectable child causes. Terminal classification uses that owner, preserving
 context precedence and generation-specific transport/provider details rather
 than mistaking a collaborator's nested lookup error for the parent's operation.
+Timeout attribution is finalized and the child scope released before terminal
+delivery, so observer latency cannot change the already selected outcome.
+Prepared-content notification is a separate opt-in callback, not an event.
 
 The facade masks inherited recorder and prepared-claim state when entering
 preparation or execution. Delegated calls from an injected collaborator retain

docs/internal/overview.md

@@ -14,7 +14,7 @@ contributor workflow and validation.
 | Root `promptkit` package | Provides the supported engine facade, ConfigKit source helpers, collection and resource mapping conveniences, injection options, public requests and results, resource and requirement inspection and identity inventories, concrete-request validation, explicit smoke diagnostics, observation values, opaque prepared-execution handles, value conversion, redacted formatting, and typed capacity and generation error mapping. | [Package GoDoc](../../doc.go), [observations](../../observations.go), [prepared execution](../../prepared_execution.go), [backend API](../../backends.go), [engine facade](../../engine.go), [smoke diagnostics](../../smoke_diagnostics.go), [identity listing](../../listing.go), [requirement facade](../../requirements.go) |
 | `examples/go-library/prepare` | Demonstrates offline preparation from two explicitly selected synthetic collections, source-relative imported content, a scoped view, filtered listings, and operator inventory. It is not a public library package. | [Example program](../../examples/go-library/prepare/main.go), [collection inputs](../../examples/go-library/prepare/collections/support/collection.yaml) |
 | `examples/go-library/run` | Demonstrates an offline downstream consumer using a prompt file, programmatic profile, inline input, an injected deterministic model client, and `Run`. It is not a public library package. | [Example program](../../examples/go-library/run/main.go) |
-| `examples/go-library/observe` | Demonstrates offline built-in HTTP retry and synchronous metadata observation across structured proposal and assessment runs using embedded sources, two profiles on one finite backend, and an application decision. It is not a public library package. | [Example program](../../examples/go-library/observe/main.go) |
+| `examples/go-library/observe` | Demonstrates offline retry, observation, prepared notification, and separately budgeted coordinated proposal and assessment runs using two profiles on one finite backend. It is not a public library package. | [Example program](../../examples/go-library/observe/main.go) |
 | `internal/backend` | Holds a defensive, immutable lookup of final backend definitions, including diagnostic metadata, plus capacity policies validated by the configuration adapter. | [Backend registry](../../internal/backend/registry.go) |
 | `internal/configuration` | Loads root and namespaced resources through ConfigKit; validates final values, captures indexed resource views and source-relative prompt content, compiled templates, requirement metadata, identity inventories, backend and presence-aware execution profiles, and the read-only provenance inspection view. | [Typed loader](../../internal/configuration/load.go), [identity listing](../../internal/configuration/listing.go), [internal sources](sources.md#typed-configuration-loading) |
 | `internal/capacity` | Owns engine-local bounded execution admission and FIFO model-generation permits for limited backend IDs, including cancellation-safe waiter removal and client wrapping. | [Internal capacity management](capacity.md) |

docs/internal/runner.md

@@ -209,6 +209,16 @@ mapping and after the runner's deferred admission release. Preparation and
 prepared execution follows the same terminal mapping after a successful claim;
 invalid or discarded handles do not acquire an observer.
 
+The facade establishes an optional operation scope for ordinary runs before
+observation and preparation. A claimed prepared execution establishes its own
+scope from the retained duration; preparation and idle time do not consume it.
+The shared scope finalizer marks only a context failure caused by that scope's
+own deadline, preserving its underlying operation and cause. The scope is
+released before terminal delivery. At the shared prepared execution boundary,
+the runner conditionally invokes the adapter for the actual prepared snapshot
+while admission is held, with context checks before and after notification.
+The facade adapter projects a detached public value only when selected.
+
 Validation preparation and execution honor cancellation at every
 Promptkit-controlled boundary and do not publish a partial plan or result.
 Schema reads are bounded and context-checked between chunks; JSON decoding,

docs/releases/v0.23.0.md

@@ -0,0 +1,62 @@
+# Promptkit v0.23.0
+
+This minor release adds optional whole-operation timeouts and prepared-request
+notifications to ordinary and coordinated execution. Applications can bound
+each continuation independently and retain the request selected for execution
+without replacing Promptkit's turn loop or preparing the same request twice.
+
+## Bound Individual Operations
+
+`RunRequest.OperationTimeout` selects a fresh cooperative budget covering
+preparation and execution, including capacity waiting, retries, validation,
+structural repair, and prepared-request notification. It is independent of the
+provider generation timeout. Zero or negative values select no additional
+timeout; existing parent cancellation and earlier deadlines remain effective.
+
+For split execution, `PrepareExecution` retains the duration without starting
+a timer. `RunPrepared` starts that budget after claiming the handle, so prior
+preparation and idle time do not consume it. Callers can still supply one shared
+deadline context to bound both calls together.
+
+Local budget failures match both `ErrOperationTimeout` and
+`context.DeadlineExceeded`. Each coordinated turn uses its selected budget;
+expiry stops that coordination through the existing operational-error path
+without canceling its parent or sibling work. Completed outcomes are fixed
+before terminal observation and are not changed by a slow terminal observer.
+Cancellation does not forcibly interrupt callbacks or collaborators that ignore
+their context.
+
+See the [execution-controls guide](../consumers/pkg-promptkit.md#bound-execution-and-retain-prepared-details)
+and the [public request contract](../../types.go) and
+[timeout error contract](../../engine.go) for exact behavior.
+
+## Retain Prepared Details Before Generation
+
+`RunRequest.OnPrepared` optionally receives a detached, credential-redacted
+`PreparedRun` and the operation context before first generation. It uses the
+actual prepared snapshot, including continuation history, without rereading
+inputs or rerendering. Applications may retain these details even if generation
+later fails without returning a result.
+
+Notification is synchronous, at most once per logical execution, and runs while
+logical admission is held. It is not repeated for retries or structural repair.
+Preparation or admission failure can prevent delivery. The application owns
+diagnostic storage and disclosure decisions; ordinary observers remain
+metadata-only. See the [OnPrepared GoDoc](../../types.go) and updated
+[offline observation example](../../examples/go-library/observe/main.go).
+
+## Compatibility And Upgrade
+
+Both controls are opt-in. Existing keyed requests and configuration files need
+no migration from v0.22.0. `Prepare` and `ValidateRequest` continue using only
+their supplied contexts and do not invoke `OnPrepared`. `RunRequest` gained
+fields; consumers using positional struct literals must switch to keyed
+literals. Prepared/result JSON representations are unchanged.
+
+The Go consumer minimum remains 1.27.0, and dependency versions are unchanged.
+
+```sh
+go get go.maximumdirect.net/promptkit@v0.23.0
+```
+
+[COPYRIGHT](https://forge.maximumdirect.net/base-layer/promptkit/src/tag/v0.23.0/COPYRIGHT) | [LICENSE](https://forge.maximumdirect.net/base-layer/promptkit/src/tag/v0.23.0/LICENSE) | [ATTRIBUTION](https://forge.maximumdirect.net/base-layer/promptkit/src/tag/v0.23.0/docs/oss/attribution.md)

engine.go

@@ -60,6 +60,13 @@ var (
 	// A [CapacityError] reports the selected backend ID. It is not an invalid
 	// request, an LLM or provider rate-limit response, or ErrLLMGenerate.
 	ErrCapacityExceeded = errors.New("backend capacity exceeded")
+	// ErrOperationTimeout identifies expiration of the timeout selected by
+	// RunRequest.OperationTimeout. Such errors also match context.DeadlineExceeded
+	// and retain their underlying operation and collaborator identities.
+	// Earlier parent cancellation or deadlines and provider-only deadlines do
+	// not acquire this identity. A completed result remains completed even if
+	// a terminal observer outlasts the selected budget.
+	ErrOperationTimeout = errors.New("operation timeout exceeded")
 	// ErrLLMGenerate identifies a model-client failure or a nil successful
 	// response. A built-in OpenAI-compatible non-2xx response is available as a
 	// [GenerationError]. Errors returned by an injected LLMClient remain
@@ -386,6 +393,7 @@ func (e *Engine) InspectBackend(ctx context.Context, backendID string) (*Backend
 // reported in the applicable operation category; no general errors.Is
 // relationship to ctx.Err is promised. Prepare returns no partial result on
 // error. It validates RunRequest.TransientRetries but makes no model attempts.
+// RunRequest.OperationTimeout and RunRequest.OnPrepared have no effect here.
 func (e *Engine) Prepare(ctx context.Context, req RunRequest) (*PreparedRun, error) {
 	prepared, err := e.prepareRequest(ctx, req)
 	if err != nil {
@@ -407,6 +415,7 @@ func (e *Engine) Prepare(ctx context.Context, req RunRequest) (*PreparedRun, err
 // matching ErrInvalidConfig when the Engine is nil. Success does not guarantee
 // provider acceptance or later execution success. Use [Engine.PrepareExecution]
 // when execution must use the snapshot produced by preparation.
+// Execution-only controls do not start a timer or invoke a notification here.
 func (e *Engine) ValidateRequest(ctx context.Context, req RunRequest) error {
 	_, err := e.prepareRequest(ctx, req)
 	return err
@@ -449,6 +458,8 @@ func (e *Engine) prepareRequest(ctx context.Context, req RunRequest) (*domain.Pr
 // ErrInvalidConfig. The effective transient retry budget is frozen privately
 // for RunPrepared and is absent from Details. A supplied Observer is retained
 // privately for RunPrepared; preparation itself emits no events.
+// RunRequest.OperationTimeout and RunRequest.OnPrepared are retained without
+// starting a timer or invoking the notification during preparation.
 func (e *Engine) PrepareExecution(ctx context.Context, req RunRequest) (*PreparedExecution, error) {
 	ctx = observation.WithoutExecutionState(ctx)
 	if e == nil || e.internal == nil {
@@ -479,6 +490,9 @@ func (e *Engine) PrepareExecution(ctx context.Context, req RunRequest) (*Prepare
 // returns the final failed validation result with cumulative usage and actual
 // repair attempts. An inability to generate or validate returns an error and
 // no partial result.
+// RunRequest.OperationTimeout selects a cooperative whole-operation deadline;
+// RunRequest.OnPrepared selects conditional notification before generation.
+// The outcome and timeout attribution are fixed before terminal observation.
 //
 // The built-in client retries eligible provider statuses using the effective
 // Config or request budget. With R additional transient attempts and P repair
@@ -510,8 +524,14 @@ func (e *Engine) PrepareExecution(ctx context.Context, req RunRequest) (*Prepare
 // result on error.
 func (e *Engine) Run(ctx context.Context, req RunRequest) (*RunResult, error) {
 	ctx = observation.WithoutExecutionState(ctx)
+	scope := usecase.NewOperationScope(ctx, req.OperationTimeout)
+	defer scope.Close()
+	ctx = scope.Context
 	if req.Observer == nil {
-		return e.run(ctx, req)
+		result, err := e.run(ctx, req)
+		err = mapPublicError(scope.Finalize(err))
+		scope.Close()
+		return result, err
 	}
 	recorder, idErr := observation.New(func(event domain.ExecutionEvent) { req.Observer(fromDomainExecutionEvent(event)) })
 	if req.ContinueFrom != nil && req.ContinueFrom.internal != nil {
@@ -519,12 +539,15 @@ func (e *Engine) Run(ctx context.Context, req RunRequest) (*RunResult, error) {
 	}
 	recorder.Start()
 	if idErr != nil {
-		err := fmt.Errorf("failed to create run id: %w", idErr)
+		err := mapPublicError(scope.Finalize(fmt.Errorf("failed to create run id: %w", idErr)))
+		scope.Close()
 		terminal := observedFinish(nil, err)
 		recorder.Finish(observation.FinishData{Outcome: terminal.Outcome, Failure: terminal.Failure})
 		return nil, err
 	}
 	result, err := e.run(observation.WithRecorder(ctx, recorder), req)
+	err = mapPublicError(scope.Finalize(err))
+	scope.Close()
 	terminal := observedFinish(result, err)
 	recorder.Finish(observation.FinishData{Outcome: terminal.Outcome, Failure: terminal.Failure,
 		StatusCode: terminal.StatusCode, RetriesExhausted: terminal.RetriesExhausted})
@@ -543,7 +566,7 @@ func (e *Engine) run(ctx context.Context, req RunRequest) (*RunResult, error) {
 
 	result, err := e.internal.Run(ctx, domainReq)
 	if err != nil {
-		return nil, mapPublicError(err)
+		return nil, err
 	}
 	return fromDomainRunResult(result), nil
 }
@@ -587,6 +610,9 @@ func (e *Engine) run(ctx context.Context, req RunRequest) (*RunResult, 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.
+// The retained RunRequest.OperationTimeout starts after claim; the retained
+// RunRequest.OnPrepared is notified only if execution reaches its prepared
+// point. Final timeout attribution precedes terminal observation.
 func (e *Engine) RunPrepared(ctx context.Context, prepared *PreparedExecution) (*RunResult, error) {
 	ctx = observation.WithoutExecutionState(ctx)
 	if e == nil || e.internal == nil {

errors.go

@@ -96,6 +96,8 @@ func mapUnownedPublicError(err error) error {
 
 func publicErrorFor(err error) error {
 	switch {
+	case errors.Is(err, usecase.ErrOperationTimeout):
+		return ErrOperationTimeout
 	case errors.Is(err, promptdef.ErrPromptDefinitionNotFound):
 		return ErrPromptNotFound
 	case errors.Is(err, profile.ErrProfileNotFound):

examples/go-library/observe/main.go

@@ -9,6 +9,7 @@ import (
 	"net/http"
 	"os"
 	"strings"
+	"time"
 
 	"go.maximumdirect.net/configkit"
 	"go.maximumdirect.net/promptkit"
@@ -79,6 +80,8 @@ type summary struct {
 	ExhaustedOutcome           promptkit.CoordinationOutcome `json:"exhausted_outcome"`
 	GroupMembers               int                           `json:"group_members"`
 	GroupAttempts              uint64                        `json:"group_attempts"`
+	PreparedMessages           int                           `json:"prepared_messages"`
+	PreparedParentMatches      bool                          `json:"prepared_parent_matches"`
 }
 
 func run() (summary, error) {
@@ -123,17 +126,18 @@ func run() (summary, error) {
 		return summary{}, err
 	}
 	var assessmentLog eventLog
+	var preparedDetails promptkit.PreparedRun
 	decision := "rejected"
 	coordination, err := engine.Coordinate(context.Background(), promptkit.CoordinateRequest{
 		Start: proposal.Exchange, AdditionalTurns: 1, Observer: executions.Observe,
 		Next: func(ctx context.Context, turn promptkit.ConversationTurn) (*promptkit.RunRequest, error) {
 			if turn.Result == nil {
-				return &promptkit.RunRequest{PromptID: "example.proposal", Inputs: map[string]promptkit.ArtifactRef{"idea": promptkit.Inline("Refine the proposal.")}}, nil
+				return &promptkit.RunRequest{PromptID: "example.proposal", Inputs: map[string]promptkit.ArtifactRef{"idea": promptkit.Inline("Refine the proposal.")}, OperationTimeout: 10 * time.Second, OnPrepared: func(_ context.Context, details promptkit.PreparedRun) { preparedDetails = details }}, nil
 			}
 			if !turn.Result.Validation.IsValid {
 				return nil, fmt.Errorf("follow-up failed structural validation")
 			}
-			assessment, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "example.assessment", Inputs: map[string]promptkit.ArtifactRef{"proposal": promptkit.Inline(turn.Result.RawOutput)}, Observer: promptkit.ComposeObservers(assessmentLog.record, executions.Observe)})
+			assessment, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "example.assessment", Inputs: map[string]promptkit.ArtifactRef{"proposal": promptkit.Inline(turn.Result.RawOutput)}, OperationTimeout: 10 * time.Second, Observer: promptkit.ComposeObservers(assessmentLog.record, executions.Observe)})
 			if err != nil {
 				return nil, err
 			}
@@ -183,6 +187,8 @@ func run() (summary, error) {
 		ExhaustedOutcome:           exhausted.Outcome,
 		GroupMembers:               len(group.Members),
 		GroupAttempts:              group.Attempts,
+		PreparedMessages:           len(preparedDetails.Messages),
+		PreparedParentMatches:      preparedDetails.ParentRunID == proposal.RunID,
 	}, nil
 }
 

examples/go-library/observe/main_test.go

@@ -12,23 +12,25 @@ func TestObserveExample(t *testing.T) {
 		t.Fatal(err)
 	}
 	var got struct {
-		ProposalAttempts    *int   `json:"proposal_attempts"`
-		ProposalRetries     *int   `json:"proposal_transient_retries"`
-		UsageComplete       *bool  `json:"proposal_usage_complete"`
-		AssessmentAttempts  *int   `json:"assessment_attempts"`
-		AssessmentRetries   *int   `json:"assessment_transient_retries"`
-		Outcome             string `json:"assessment_outcome"`
-		CoordinationOutcome string `json:"coordination_outcome"`
-		ExhaustedOutcome    string `json:"exhausted_outcome"`
-		GroupMembers        int    `json:"group_members"`
-		GroupAttempts       int    `json:"group_attempts"`
+		ProposalAttempts      *int   `json:"proposal_attempts"`
+		ProposalRetries       *int   `json:"proposal_transient_retries"`
+		UsageComplete         *bool  `json:"proposal_usage_complete"`
+		AssessmentAttempts    *int   `json:"assessment_attempts"`
+		AssessmentRetries     *int   `json:"assessment_transient_retries"`
+		Outcome               string `json:"assessment_outcome"`
+		CoordinationOutcome   string `json:"coordination_outcome"`
+		ExhaustedOutcome      string `json:"exhausted_outcome"`
+		GroupMembers          int    `json:"group_members"`
+		GroupAttempts         int    `json:"group_attempts"`
+		PreparedMessages      int    `json:"prepared_messages"`
+		PreparedParentMatches bool   `json:"prepared_parent_matches"`
 	}
 	if err := json.Unmarshal(output.Bytes(), &got); err != nil {
 		t.Fatal(err)
 	}
 	if got.ProposalAttempts == nil || *got.ProposalAttempts != 2 || got.ProposalRetries == nil || *got.ProposalRetries != 1 ||
 		got.UsageComplete == nil || *got.UsageComplete || got.AssessmentAttempts == nil || *got.AssessmentAttempts != 1 ||
-		got.AssessmentRetries == nil || *got.AssessmentRetries != 0 || got.Outcome != "accepted" || got.CoordinationOutcome != "stopped" || got.ExhaustedOutcome != "exhausted" || got.GroupMembers != 3 || got.GroupAttempts != 4 {
+		got.AssessmentRetries == nil || *got.AssessmentRetries != 0 || got.Outcome != "accepted" || got.CoordinationOutcome != "stopped" || got.ExhaustedOutcome != "exhausted" || got.GroupMembers != 3 || got.GroupAttempts != 4 || got.PreparedMessages < 3 || !got.PreparedParentMatches {
 		t.Fatalf("unexpected observation summary: %+v", got)
 	}
 }

execution_budget_contract_test.go

@@ -0,0 +1,192 @@
+package promptkit_test
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"testing"
+	"testing/synctest"
+	"time"
+
+	"go.maximumdirect.net/promptkit"
+)
+
+func TestOperationBudgetCoversPreparationAndReleasesAdmission(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		release := make(chan struct{})
+		reader := &capacityArtifactReader{entered: make(chan struct{}, 2), release: release}
+		client := newCapacityGateClient(closedCapacityChannel(), 2)
+		engine := newBackendCapacityEngine(t, client, 1, capacityInt(0), reader)
+		parent := context.Background()
+		req := capacityInputRequest("http://example.invalid/v1")
+		req.OperationTimeout = time.Second
+		_, err := engine.Run(parent, req)
+		if !errors.Is(err, promptkit.ErrOperationTimeout) || !errors.Is(err, context.DeadlineExceeded) || !errors.Is(err, promptkit.ErrArtifactLoad) || parent.Err() != nil {
+			t.Fatalf("preparation error=%v parent=%v", err, parent.Err())
+		}
+		close(release)
+		if result, err := engine.Run(parent, req); err != nil || result == nil {
+			t.Fatalf("capacity not released: %v", err)
+		}
+	})
+}
+
+func TestOperationBudgetCoversGenerationCapacityWait(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		release := make(chan struct{})
+		client := newCapacityGateClient(release, 2)
+		engine := newBackendCapacityEngine(t, client, 1, capacityInt(1), nil)
+		results := make(chan capacityRunResult, 1)
+		go runCapacityRequest(engine, context.Background(), promptkit.RunRequest{PromptID: "prompt"}, results)
+		<-client.started
+		_, err := engine.Run(context.Background(), promptkit.RunRequest{PromptID: "prompt", OperationTimeout: time.Second})
+		if !errors.Is(err, promptkit.ErrOperationTimeout) || !errors.Is(err, promptkit.ErrLLMGenerate) || !errors.Is(err, context.DeadlineExceeded) {
+			t.Fatalf("capacity wait error=%v", err)
+		}
+		close(release)
+		if result := <-results; result.err != nil {
+			t.Fatal(result.err)
+		}
+		if _, err := engine.Run(context.Background(), promptkit.RunRequest{PromptID: "prompt"}); err != nil {
+			t.Fatal(err)
+		}
+	})
+}
+
+func TestOperationBudgetInterruptsRetryAndRepair(t *testing.T) {
+	for _, repair := range []bool{false, true} {
+		t.Run(map[bool]string{false: "retry", true: "repair"}[repair], func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				calls := 0
+				config := simpleConfig("message")
+				one := 1
+				config.TransientRetries = &one
+				config.HTTPClient = &http.Client{Transport: roundTripFunc(func(r *http.Request) (*http.Response, error) {
+					calls++
+					if repair && calls == 1 {
+						return retryHTTPResponse(200, "not json"), nil
+					}
+					if repair {
+						<-r.Context().Done()
+						return nil, r.Context().Err()
+					}
+					response := retryHTTPResponse(503, "")
+					response.Header = make(http.Header)
+					response.Header.Set("Retry-After", "5")
+					return response, nil
+				})}
+				engine, err := promptkit.NewEngine(config)
+				if err != nil {
+					t.Fatal(err)
+				}
+				req := simpleGenerationRequest()
+				req.OperationTimeout = time.Second
+				if repair {
+					req.Validation = &promptkit.OutputContract{Format: promptkit.FormatJSON, ValidationMode: promptkit.ValidationJSON, RepairAttempts: 1}
+				}
+				_, err = engine.Run(context.Background(), req)
+				if !errors.Is(err, promptkit.ErrOperationTimeout) || !errors.Is(err, promptkit.ErrLLMGenerate) || !errors.Is(err, context.DeadlineExceeded) {
+					t.Fatalf("error=%v calls=%d", err, calls)
+				}
+			})
+		})
+	}
+}
+
+func TestOperationOutcomeFrozenBeforeTerminalObserver(t *testing.T) {
+	for _, split := range []bool{false, true} {
+		t.Run(map[bool]string{false: "run", true: "prepared"}[split], func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				engine := observationEngine(t, observationClientFunc(func(context.Context, promptkit.GenerateRequest) (*promptkit.GenerateResponse, error) {
+					return &promptkit.GenerateResponse{Content: "ok"}, nil
+				}))
+				var terminal promptkit.ExecutionEvent
+				req := promptkit.RunRequest{PromptID: "prompt", OperationTimeout: time.Second, Observer: func(event promptkit.ExecutionEvent) {
+					if event.Kind == promptkit.ExecutionFinished {
+						time.Sleep(2 * time.Second)
+						terminal = event
+					}
+				}}
+				var result *promptkit.RunResult
+				var err error
+				if split {
+					var handle *promptkit.PreparedExecution
+					handle, err = engine.PrepareExecution(context.Background(), req)
+					if err == nil {
+						result, err = engine.RunPrepared(context.Background(), handle)
+					}
+				} else {
+					result, err = engine.Run(context.Background(), req)
+				}
+				if err != nil || result == nil || terminal.Outcome != promptkit.ExecutionOutcomeSuccess {
+					t.Fatalf("result=%v error=%v terminal=%+v", result, err, terminal)
+				}
+			})
+		})
+	}
+}
+
+func TestNotificationConsumesOperationBudget(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		engine := observationEngine(t, observationClientFunc(func(context.Context, promptkit.GenerateRequest) (*promptkit.GenerateResponse, error) {
+			t.Fatal("generation ran")
+			return nil, nil
+		}))
+		_, err := engine.Run(context.Background(), promptkit.RunRequest{PromptID: "prompt", OperationTimeout: time.Second, OnPrepared: func(ctx context.Context, _ promptkit.PreparedRun) { <-ctx.Done() }})
+		if !errors.Is(err, promptkit.ErrOperationTimeout) || !errors.Is(err, promptkit.ErrLLMGenerate) || !errors.Is(err, context.DeadlineExceeded) {
+			t.Fatal(err)
+		}
+	})
+}
+
+func TestOperationTimeoutDoesNotRelabelOtherFailures(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		failure := errors.New("independent failure")
+		engine := observationEngine(t, observationClientFunc(func(ctx context.Context, _ promptkit.GenerateRequest) (*promptkit.GenerateResponse, error) {
+			<-ctx.Done()
+			return nil, failure
+		}))
+		var terminal promptkit.ExecutionEvent
+		_, err := engine.Run(context.Background(), promptkit.RunRequest{PromptID: "prompt", OperationTimeout: time.Second, Observer: func(event promptkit.ExecutionEvent) {
+			if event.Kind == promptkit.ExecutionFinished {
+				terminal = event
+			}
+		}})
+		if !errors.Is(err, failure) || errors.Is(err, promptkit.ErrOperationTimeout) || terminal.Failure != promptkit.ExecutionFailureGeneration {
+			t.Fatalf("error=%v terminal=%+v", err, terminal)
+		}
+	})
+}
+
+func TestOperationTimeoutTerminalObserverSeesFinalFailure(t *testing.T) {
+	for _, split := range []bool{false, true} {
+		t.Run(map[bool]string{false: "run", true: "prepared"}[split], func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				engine := observationEngine(t, observationClientFunc(func(ctx context.Context, _ promptkit.GenerateRequest) (*promptkit.GenerateResponse, error) {
+					<-ctx.Done()
+					return nil, ctx.Err()
+				}))
+				var terminal promptkit.ExecutionEvent
+				req := promptkit.RunRequest{PromptID: "prompt", OperationTimeout: time.Second, Observer: func(event promptkit.ExecutionEvent) {
+					if event.Kind == promptkit.ExecutionFinished {
+						time.Sleep(time.Second)
+						terminal = event
+					}
+				}}
+				var err error
+				if split {
+					var handle *promptkit.PreparedExecution
+					handle, err = engine.PrepareExecution(context.Background(), req)
+					if err == nil {
+						_, err = engine.RunPrepared(context.Background(), handle)
+					}
+				} else {
+					_, err = engine.Run(context.Background(), req)
+				}
+				if !errors.Is(err, promptkit.ErrOperationTimeout) || !errors.Is(err, context.DeadlineExceeded) || terminal.Failure != promptkit.ExecutionFailureDeadline {
+					t.Fatalf("error=%v terminal=%+v", err, terminal)
+				}
+			})
+		})
+	}
+}

execution_controls_conversion_test.go

@@ -0,0 +1,38 @@
+package promptkit
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"go.maximumdirect.net/promptkit/internal/domain"
+)
+
+func TestExecutionControlsConversion(t *testing.T) {
+	ctx := context.WithValue(context.Background(), struct{}{}, "selected")
+	called := false
+	req := RunRequest{OperationTimeout: 3 * time.Second, OnPrepared: func(got context.Context, details PreparedRun) {
+		called = true
+		if got != ctx || details.PromptID != "prompt" {
+			t.Fatalf("unexpected notification: %v, %+v", got, details)
+		}
+		details.InputHashes["input"] = "changed"
+	}}
+	converted, err := toDomainRunRequest(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if converted.OperationTimeout != req.OperationTimeout || converted.OnPrepared == nil || called {
+		t.Fatal("execution controls were not retained without invocation")
+	}
+	req.OnPrepared = func(context.Context, PreparedRun) { t.Fatal("callback selection changed") }
+	prepared := &domain.PreparedRun{PromptID: "prompt", InputHashes: map[string]string{"input": "original"}}
+	converted.OnPrepared(ctx, prepared)
+	if !called || prepared.InputHashes["input"] != "original" {
+		t.Fatal("notification did not deliver detached details")
+	}
+	withoutCallback, err := toDomainRunRequest(RunRequest{})
+	if err != nil || withoutCallback.OnPrepared != nil {
+		t.Fatalf("nil callback was adapted: %v", err)
+	}
+}

execution_timeout_contract_test.go

@@ -0,0 +1,100 @@
+package promptkit_test
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"testing/synctest"
+	"time"
+
+	"go.maximumdirect.net/promptkit"
+)
+
+func TestRunOperationTimeoutAttribution(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		engine := observationEngine(t, observationClientFunc(func(ctx context.Context, _ promptkit.GenerateRequest) (*promptkit.GenerateResponse, error) {
+			<-ctx.Done()
+			return nil, ctx.Err()
+		}))
+		var terminal promptkit.ExecutionEvent
+		_, err := engine.Run(context.Background(), promptkit.RunRequest{PromptID: "prompt", OperationTimeout: time.Second,
+			Observer: func(event promptkit.ExecutionEvent) {
+				if event.Kind == promptkit.ExecutionFinished {
+					terminal = event
+				}
+			}})
+		if !errors.Is(err, promptkit.ErrOperationTimeout) || !errors.Is(err, context.DeadlineExceeded) || !errors.Is(err, promptkit.ErrLLMGenerate) || terminal.Failure != promptkit.ExecutionFailureDeadline {
+			t.Fatalf("error=%v terminal=%+v", err, terminal)
+		}
+	})
+}
+
+func TestRunProviderDeadlineIsNotOperationTimeout(t *testing.T) {
+	engine := observationEngine(t, observationClientFunc(func(context.Context, promptkit.GenerateRequest) (*promptkit.GenerateResponse, error) {
+		return nil, context.DeadlineExceeded
+	}))
+	_, err := engine.Run(context.Background(), promptkit.RunRequest{PromptID: "prompt", OperationTimeout: time.Minute})
+	if !errors.Is(err, context.DeadlineExceeded) || errors.Is(err, promptkit.ErrOperationTimeout) {
+		t.Fatalf("provider deadline attribution: %v", err)
+	}
+}
+
+func TestPreparedExecutionRetainsTimeoutWithoutIdleTimer(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		engine := observationEngine(t, observationClientFunc(func(ctx context.Context, _ promptkit.GenerateRequest) (*promptkit.GenerateResponse, error) {
+			<-ctx.Done()
+			return nil, ctx.Err()
+		}))
+		req := promptkit.RunRequest{PromptID: "prompt", OperationTimeout: time.Second}
+		handle, err := engine.PrepareExecution(context.Background(), req)
+		if err != nil {
+			t.Fatal(err)
+		}
+		req.OperationTimeout = 0
+		time.Sleep(2 * time.Second)
+		start := time.Now()
+		_, err = engine.RunPrepared(context.Background(), handle)
+		if time.Since(start) != time.Second || !errors.Is(err, promptkit.ErrOperationTimeout) || !errors.Is(err, context.DeadlineExceeded) || !errors.Is(err, promptkit.ErrLLMGenerate) {
+			t.Fatalf("elapsed=%v error=%v", time.Since(start), err)
+		}
+	})
+}
+
+func TestPreparedExecutionRespectsCallerDeadline(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		engine := observationEngine(t, observationClientFunc(func(ctx context.Context, _ promptkit.GenerateRequest) (*promptkit.GenerateResponse, error) {
+			<-ctx.Done()
+			return nil, ctx.Err()
+		}))
+		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+		defer cancel()
+		handle, err := engine.PrepareExecution(ctx, promptkit.RunRequest{PromptID: "prompt", OperationTimeout: 5 * time.Second})
+		if err != nil {
+			t.Fatal(err)
+		}
+		time.Sleep(time.Second)
+		_, err = engine.RunPrepared(ctx, handle)
+		if !errors.Is(err, context.DeadlineExceeded) || errors.Is(err, promptkit.ErrOperationTimeout) {
+			t.Fatalf("caller deadline attribution: %v", err)
+		}
+	})
+}
+
+func TestPreparedExecutionDoesNotTimePreparation(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		release := make(chan struct{})
+		reader := &capacityArtifactReader{entered: make(chan struct{}, 1), release: release}
+		engine := newBackendCapacityEngine(t, newCapacityGateClient(closedCapacityChannel(), 1), 1, capacityInt(0), reader)
+		req := capacityInputRequest("http://example.invalid/v1")
+		req.OperationTimeout = time.Second
+		go func() { time.Sleep(2 * time.Second); close(release) }()
+		start := time.Now()
+		handle, err := engine.PrepareExecution(context.Background(), req)
+		if err != nil || time.Since(start) != 2*time.Second {
+			t.Fatalf("prepare elapsed=%v error=%v", time.Since(start), err)
+		}
+		if _, err = engine.RunPrepared(context.Background(), handle); err != nil {
+			t.Fatalf("execution after preparation: %v", err)
+		}
+	})
+}

internal/domain/domain.go

@@ -1,6 +1,7 @@
 package domain
 
 import (
+	"context"
 	"time"
 )
 
@@ -60,6 +61,9 @@ type CacheControl struct {
 
 // RunRequest represents a request to generate a single artifact.
 type RunRequest struct {
+	OperationTimeout time.Duration
+	// OnPrepared receives borrowed, read-only internal state.
+	OnPrepared       func(context.Context, *PreparedRun)
 	CaptureExchange  bool
 	ContinueFrom     *Exchange
 	PromptID         string

internal/usecase/operation_budget_test.go

@@ -0,0 +1,54 @@
+package usecase
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"testing/synctest"
+	"time"
+
+	"go.maximumdirect.net/promptkit/internal/domain"
+	"go.maximumdirect.net/promptkit/internal/validate"
+)
+
+type deadlineValidation struct{ deadline time.Time }
+
+func (v *deadlineValidation) PrepareValidation(ctx context.Context, _ domain.OutputContract) (validate.PreparedValidation, error) {
+	v.deadline, _ = ctx.Deadline()
+	return v, nil
+}
+func (v *deadlineValidation) Validate(ctx context.Context, _ *domain.Artifact) (domain.ValidationResult, error) {
+	deadline, _ := ctx.Deadline()
+	if deadline != v.deadline {
+		return domain.ValidationResult{}, errors.New("validation deadline changed")
+	}
+	<-ctx.Done()
+	return domain.ValidationResult{}, ctx.Err()
+}
+func (*deadlineValidation) SchemaDocument() any { return nil }
+
+type deadlineModel struct{ validation *deadlineValidation }
+
+func (m deadlineModel) Generate(ctx context.Context, _ domain.GenerateRequest) (*domain.GenerateResponse, error) {
+	deadline, _ := ctx.Deadline()
+	if deadline != m.validation.deadline {
+		return nil, errors.New("generation deadline changed")
+	}
+	return &domain.GenerateResponse{Content: "ok"}, nil
+}
+
+func TestOperationDeadlineReachesPreparationGenerationAndValidation(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		validation := &deadlineValidation{}
+		runner := NewRunner(&fakePromptRepo{def: promptDef(domain.FormatText, domain.ValidationBasic, 0)},
+			&fakeExecutionProfileRepo{profiles: map[string]*domain.ExecutionProfile{"exec": defaultExecutionProfile()}},
+			defaultArtifactReader(), defaultRenderer(), deadlineModel{validation}, validation, nil)
+		scope := NewOperationScope(context.Background(), time.Second)
+		defer scope.Close()
+		_, err := runner.Run(scope.Context, domain.RunRequest{PromptID: "p", ProfileID: "exec", Inputs: singleInputRef()})
+		err = scope.Finalize(err)
+		if !errors.Is(err, ErrOperationTimeout) || !errors.Is(err, ErrValidation) || !errors.Is(err, context.DeadlineExceeded) {
+			t.Fatalf("validation error=%v", err)
+		}
+	})
+}

internal/usecase/operation_scope.go

@@ -0,0 +1,36 @@
+package usecase
+
+import (
+	"context"
+	"errors"
+	"time"
+)
+
+// OperationScope owns an optional deadline and its invocation-specific cause.
+type OperationScope struct {
+	Context context.Context
+	cancel  context.CancelFunc
+	cause   error
+}
+
+// NewOperationScope creates a child only when the selected duration is positive.
+func NewOperationScope(parent context.Context, duration time.Duration) *OperationScope {
+	scope := &OperationScope{Context: parent, cancel: func() {}}
+	if duration > 0 {
+		scope.cause = errors.New("operation deadline")
+		scope.Context, scope.cancel = context.WithTimeoutCause(parent, duration, scope.cause)
+	}
+	return scope
+}
+
+// Finalize preserves the failure and marks it only when this scope expired.
+func (s *OperationScope) Finalize(err error) error {
+	if err != nil && s.cause != nil && s.Context.Err() == context.DeadlineExceeded &&
+		errors.Is(err, s.Context.Err()) && context.Cause(s.Context) == s.cause {
+		return &OperationError{Category: ErrOperationTimeout, Cause: err}
+	}
+	return err
+}
+
+// Close releases the timer and associated child context.
+func (s *OperationScope) Close() { s.cancel() }

internal/usecase/operation_scope_test.go

@@ -0,0 +1,52 @@
+package usecase
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"testing/synctest"
+	"time"
+)
+
+func TestOperationScopeAttribution(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		parent := context.Background()
+		disabled := NewOperationScope(parent, 0)
+		defer disabled.Close()
+		if disabled.Context != parent || disabled.Finalize(context.DeadlineExceeded) != context.DeadlineExceeded {
+			t.Fatal("disabled scope changed context or error")
+		}
+		outer := NewOperationScope(parent, time.Second)
+		defer outer.Close()
+		inner := NewOperationScope(outer.Context, 2*time.Second)
+		defer inner.Close()
+		time.Sleep(time.Second)
+		synctest.Wait()
+		if !errors.Is(outer.Finalize(context.DeadlineExceeded), ErrOperationTimeout) {
+			t.Fatal("own deadline not attributed")
+		}
+		if errors.Is(inner.Finalize(context.DeadlineExceeded), ErrOperationTimeout) {
+			t.Fatal("inherited scope misattributed")
+		}
+		other := errors.New("other")
+		if outer.Finalize(other) != other || outer.Finalize(nil) != nil {
+			t.Fatal("unrelated failure or completion changed")
+		}
+		canceled, cancel := context.WithCancel(parent)
+		child := NewOperationScope(canceled, time.Second)
+		defer child.Close()
+		cancel()
+		if errors.Is(child.Finalize(context.Canceled), ErrOperationTimeout) {
+			t.Fatal("parent cancellation misattributed")
+		}
+		deadline, stop := context.WithTimeout(parent, time.Second)
+		defer stop()
+		child = NewOperationScope(deadline, 2*time.Second)
+		defer child.Close()
+		time.Sleep(time.Second)
+		synctest.Wait()
+		if errors.Is(child.Finalize(context.DeadlineExceeded), ErrOperationTimeout) {
+			t.Fatal("parent deadline misattributed")
+		}
+	})
+}

internal/usecase/prepared_execution.go

@@ -29,6 +29,8 @@ type PreparedExecution struct {
 }
 
 type preparedExecutionPayload struct {
+	operationTimeout time.Duration
+	onPrepared       func(context.Context, *domain.PreparedRun)
 	prepared         *domain.PreparedRun
 	capture          bool
 	parent           *domain.Exchange
@@ -56,6 +58,8 @@ func (r *Runner) PrepareExecution(ctx context.Context, req domain.RunRequest) (*
 		state:   preparedExecutionReady,
 		details: operation.run,
 		payload: &preparedExecutionPayload{
+			operationTimeout: req.OperationTimeout,
+			onPrepared:       req.OnPrepared,
 			prepared:         operation.run,
 			capture:          operation.capture,
 			parent:           operation.parent,
@@ -100,12 +104,16 @@ func (p *PreparedExecution) Discard() {
 }
 
 // RunPrepared claims and executes one prepared execution owned by this runner.
-func (r *Runner) RunPrepared(ctx context.Context, prepared *PreparedExecution) (*domain.RunResult, error) {
+func (r *Runner) RunPrepared(ctx context.Context, prepared *PreparedExecution) (result *domain.RunResult, err error) {
 	payload, err := prepared.claim(r)
 	if err != nil {
 		return nil, err
 	}
 	defer payload.clear()
+	scope := NewOperationScope(ctx, payload.operationTimeout)
+	defer scope.Close()
+	defer func() { err = scope.Finalize(err) }()
+	ctx = scope.Context
 
 	var recorder *observation.Recorder
 	var runID string
@@ -144,7 +152,7 @@ func (r *Runner) RunPrepared(ctx context.Context, prepared *PreparedExecution) (
 	}
 	defer release()
 
-	return r.executePreparedRun(ctx, payload.prepared, payload.capture, payload.parent, payload.directKey, payload.transientRetries, runID, start, func(
+	return r.executePreparedRun(ctx, payload.prepared, payload.onPrepared, payload.capture, payload.parent, payload.directKey, payload.transientRetries, runID, start, func(
 		ctx context.Context,
 		artifact *domain.Artifact,
 		attemptsUsed int,
@@ -185,6 +193,8 @@ func (p *preparedExecutionPayload) clear() {
 	p.directKey = ""
 	p.transientRetries = 0
 	p.observer = nil
+	p.operationTimeout = 0
+	p.onPrepared = nil
 }
 
 type noOpPreparedValidation struct {

internal/usecase/prepared_notification_test.go

@@ -0,0 +1,42 @@
+package usecase
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"go.maximumdirect.net/promptkit/internal/domain"
+)
+
+func TestPreparedNotificationChecksContextBeforeInvocation(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	called := false
+	_, err := (&Runner{}).executePreparedRun(ctx, &domain.PreparedRun{}, func(context.Context, *domain.PreparedRun) { called = true }, false, nil, "", 0, "", time.Time{}, nil)
+	if called || !errors.Is(err, context.Canceled) || !errors.Is(err, ErrLLMGenerate) {
+		t.Fatalf("called=%v error=%v", called, err)
+	}
+}
+
+func TestPreparedNotificationPanicClearsClaimedPayload(t *testing.T) {
+	runner := NewRunner(&fakePromptRepo{def: promptDef(domain.FormatText, domain.ValidationNone, 0)},
+		&fakeExecutionProfileRepo{profiles: map[string]*domain.ExecutionProfile{"exec": defaultExecutionProfile()}},
+		defaultArtifactReader(), defaultRenderer(), &fakeLLM{resp: &domain.GenerateResponse{Content: "ok"}}, nil, nil)
+	handle, err := runner.PrepareExecution(context.Background(), domain.RunRequest{PromptID: "p", ProfileID: "exec", Inputs: singleInputRef(), OnPrepared: func(context.Context, *domain.PreparedRun) { panic("notification") }})
+	if err != nil {
+		t.Fatal(err)
+	}
+	payload := handle.payload
+	func() {
+		defer func() {
+			if recover() == nil {
+				t.Error("expected panic")
+			}
+		}()
+		_, _ = runner.RunPrepared(context.Background(), handle)
+	}()
+	if payload.prepared != nil || payload.onPrepared != nil || payload.directKey != "" || handle.payload != nil {
+		t.Fatal("claimed payload retained private state")
+	}
+}

internal/usecase/runner.go

@@ -39,6 +39,7 @@ var (
 	ErrPromptRender     = errors.New("failed to render prompt")
 	ErrLLMGenerate      = errors.New("failed to generate output")
 	ErrValidation       = errors.New("failed to validate output")
+	ErrOperationTimeout = errors.New("operation timeout exceeded")
 )
 
 // Runner executes the Promptkit core use case.
@@ -165,7 +166,7 @@ func (r *Runner) Run(ctx context.Context, req domain.RunRequest) (*domain.RunRes
 	}
 	directAPIKey := state.effectiveModel.APIKey
 
-	return r.executePreparedRun(ctx, operation.run, operation.capture, operation.parent, directAPIKey, state.transientRetries, runID, start, func(
+	return r.executePreparedRun(ctx, operation.run, req.OnPrepared, operation.capture, operation.parent, directAPIKey, state.transientRetries, runID, start, func(
 		ctx context.Context,
 		artifact *domain.Artifact,
 		attemptsUsed int,
@@ -185,6 +186,7 @@ type preparedValidationFunc func(
 func (r *Runner) executePreparedRun(
 	ctx context.Context,
 	prepared *domain.PreparedRun,
+	onPrepared func(context.Context, *domain.PreparedRun),
 	capture bool,
 	parent *domain.Exchange,
 	directAPIKey string,
@@ -193,6 +195,15 @@ func (r *Runner) executePreparedRun(
 	start time.Time,
 	validateArtifact preparedValidationFunc,
 ) (*domain.RunResult, error) {
+	if onPrepared != nil {
+		if err := ctx.Err(); err != nil {
+			return nil, wrapGenerationError(err)
+		}
+		onPrepared(ctx, prepared)
+		if err := ctx.Err(); err != nil {
+			return nil, wrapGenerationError(err)
+		}
+	}
 	executionTarget := prepared.EffectiveModelParams
 	executionTarget.APIKey = directAPIKey
 	genResp, err := r.llm.Generate(ctx, newGenerationRequest(

prepared_execution.go

@@ -18,6 +18,9 @@ const preparedExecutionString = "promptkit.PreparedExecution{opaque}"
 // It also retains RunRequest.CaptureExchange and RunRequest.ContinueFrom for
 // execution of the frozen snapshot. Only RunPrepared can produce its completed
 // [Exchange]; preparation alone does not.
+// RunRequest.OperationTimeout and RunRequest.OnPrepared are retained as
+// selections without an active idle timer; their execution lifecycle is
+// described on those fields.
 type PreparedExecution struct {
 	internal *usecase.PreparedExecution
 }

prepared_execution_contract_test.go

@@ -47,11 +47,36 @@ func TestPreparedExecutionFreezesSourcesAndReturnsIndependentDetails(t *testing.
 	}
 
 	temperature := 0.25
+	initial, err := newSimpleEngine(t, "history", promptkit.WithLLMClient(&fakeLLMClient{response: &promptkit.GenerateResponse{Content: "prior"}})).Run(context.Background(), promptkit.RunRequest{PromptID: "prompt", CaptureExchange: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	initialDetails := initial.Exchange.Details()
 	extraParams := map[string]any{
 		"nested": map[string]any{"source": "original"},
 	}
+	var notified *promptkit.PreparedRun
 	request := promptkit.RunRequest{
-		PromptID: "prepared",
+		PromptID:         "prepared",
+		APIKey:           "synthetic-direct-key",
+		ContinueFrom:     initial.Exchange,
+		SessionID:        "continuation-session",
+		AppendedMessages: []promptkit.RenderedMessage{{Role: promptkit.RoleUser, Content: "appended", CacheControl: &promptkit.CacheControl{Type: promptkit.CacheControlEphemeral}}},
+		OnPrepared: func(_ context.Context, details promptkit.PreparedRun) {
+			notified = &details
+			if details.ParentRunID != initial.RunID || details.SessionID != "continuation-session" || details.Messages[0].Content != "history" || details.Messages[2].Content != "Input=original artifact Label=original variable" || details.InputHashes["input"] != "original-input-hash" || details.StructuredOutput.JSONSchema.Schema.(map[string]any)["title"] != "original root" {
+				t.Error("notification did not describe the frozen continuation")
+			}
+			if strings.Contains(fmt.Sprintf("%#v", details), "synthetic-direct-key") {
+				t.Error("notification exposed credential")
+			}
+			details.Messages[0].Content = "mutated history"
+			details.Messages[2].Content = "mutated notification"
+			details.Messages[3].CacheControl.Type = "changed"
+			details.InputHashes["input"] = "mutated notification"
+			details.EffectiveModelParams.ExtraParams["nested"].(map[string]any)["source"] = "mutated notification"
+			details.StructuredOutput.JSONSchema.Schema.(map[string]any)["title"] = "mutated notification"
+		},
 		Inputs: map[string]promptkit.ArtifactRef{
 			"input": promptkit.Inline("original request input"),
 		},
@@ -87,14 +112,14 @@ func TestPreparedExecutionFreezesSourcesAndReturnsIndependentDetails(t *testing.
 	reader.set("changed artifact", "changed-input-hash")
 
 	first := prepared.Details()
-	first.Messages[0].Content = "changed details"
+	first.Messages[2].Content = "changed details"
 	first.InputHashes["input"] = "changed-details-hash"
 	first.EffectiveModelParams.ExtraParams["nested"].(map[string]any)["source"] = "changed details"
 	first.StructuredOutput.JSONSchema.Schema.(map[string]any)["title"] = "changed details"
 
 	second := prepared.Details()
-	if second.Messages[0].Content != "Input=original artifact Label=original variable" {
-		t.Fatalf("details message changed: %q", second.Messages[0].Content)
+	if second.Messages[2].Content != "Input=original artifact Label=original variable" {
+		t.Fatalf("details message changed: %q", second.Messages[2].Content)
 	}
 	if second.InputHashes["input"] != "original-input-hash" {
 		t.Fatalf("details input hash changed: %q", second.InputHashes["input"])
@@ -119,13 +144,19 @@ func TestPreparedExecutionFreezesSourcesAndReturnsIndependentDetails(t *testing.
 	if reader.callCount() != 1 {
 		t.Fatalf("execution reopened artifact source: calls=%d", reader.callCount())
 	}
+	if notified == nil || prepared.Details().Messages[2].Content != second.Messages[2].Content || prepared.Details().Messages[3].CacheControl.Type != promptkit.CacheControlEphemeral || !reflect.DeepEqual(initial.Exchange.Details(), initialDetails) || result.Exchange.Details().Messages[0].Content != "history" || prepared.Details().StructuredOutput.JSONSchema.Schema.(map[string]any)["title"] != "original root" {
+		t.Fatal("notification changed prepared state")
+	}
 
 	requests := client.snapshot()
 	if len(requests) != 1 {
 		t.Fatalf("generation calls=%d, want 1", len(requests))
 	}
 	generated := requests[0]
-	if generated.Prompt.Messages[0].Content != second.Messages[0].Content ||
+	if generated.APIKey != "synthetic-direct-key" {
+		t.Fatal("direct credential did not reach client")
+	}
+	if generated.Prompt.Messages[0].Content != "history" || generated.Prompt.Messages[2].Content != second.Messages[2].Content || generated.Prompt.Messages[3].CacheControl.Type != promptkit.CacheControlEphemeral ||
 		generated.Target.Model != second.EffectiveModelParams.Model ||
 		!reflect.DeepEqual(generated.Target.ExtraParams, second.EffectiveModelParams.ExtraParams) ||
 		!reflect.DeepEqual(generated.StructuredOutput, second.StructuredOutput) {

prepared_notification_contract_test.go

@@ -0,0 +1,246 @@
+package promptkit_test
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"reflect"
+	"strings"
+	"testing"
+
+	"go.maximumdirect.net/promptkit"
+)
+
+func TestPreparedNotificationSelectionAndOrdering(t *testing.T) {
+	for _, split := range []bool{false, true} {
+		t.Run(map[bool]string{false: "run", true: "prepared"}[split], func(t *testing.T) {
+			calls := 0
+			engine := observationEngine(t, observationClientFunc(func(_ context.Context, _ promptkit.GenerateRequest) (*promptkit.GenerateResponse, error) {
+				if calls != 1 {
+					t.Fatalf("generation before notification: %d", calls)
+				}
+				return &promptkit.GenerateResponse{Content: "ok"}, nil
+			}))
+			req := promptkit.RunRequest{PromptID: "prompt", OnPrepared: func(_ context.Context, details promptkit.PreparedRun) {
+				calls++
+				if details.PromptID != "prompt" || len(details.Messages) == 0 {
+					t.Fatalf("details=%+v", details)
+				}
+			}}
+			if split {
+				handle, err := engine.PrepareExecution(context.Background(), req)
+				if err != nil {
+					t.Fatal(err)
+				}
+				if calls != 0 {
+					t.Fatal("preparation notified")
+				}
+				req.OnPrepared = func(context.Context, promptkit.PreparedRun) { t.Fatal("selection changed") }
+				_, err = engine.RunPrepared(context.Background(), handle)
+				if err != nil {
+					t.Fatal(err)
+				}
+			} else if _, err := engine.Run(context.Background(), req); err != nil {
+				t.Fatal(err)
+			}
+			if calls != 1 {
+				t.Fatalf("notifications=%d", calls)
+			}
+		})
+	}
+}
+
+func TestNotificationRedactsEnvironmentCredential(t *testing.T) {
+	t.Setenv("PROMPTKIT_NOTIFICATION_KEY", "synthetic-environment-key")
+	config := simpleConfig("message")
+	config.HTTPClient = &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
+		if req.Header.Get("Authorization") != "Bearer synthetic-environment-key" {
+			t.Error("environment credential did not reach provider")
+		}
+		return retryHTTPResponse(200, "ok"), nil
+	})}
+	engine, err := promptkit.NewEngine(config)
+	if err != nil {
+		t.Fatal(err)
+	}
+	req := simpleGenerationRequest()
+	req.Execution = &promptkit.ExecutionTargetOverride{APIKeyEnv: "PROMPTKIT_NOTIFICATION_KEY"}
+	req.OnPrepared = func(_ context.Context, details promptkit.PreparedRun) {
+		if strings.Contains(fmt.Sprintf("%#v", details), "synthetic-environment-key") {
+			t.Error("notification exposed credential")
+		}
+	}
+	if _, err := engine.Run(context.Background(), req); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestContinuationNotificationIsDetachedAndSurvivesFailure(t *testing.T) {
+	failure := errors.New("provider failure")
+	calls := 0
+	engine := newSimpleEngine(t, "question", promptkit.WithLLMClient(observationClientFunc(func(_ context.Context, req promptkit.GenerateRequest) (*promptkit.GenerateResponse, error) {
+		calls++
+		if calls == 1 {
+			return &promptkit.GenerateResponse{Content: "answer"}, nil
+		}
+		if req.Prompt.Messages[0].Content != "question" || req.Prompt.Messages[3].Content != "follow-up" || req.APIKey != "synthetic-key" {
+			t.Fatalf("generation request=%+v", req)
+		}
+		return nil, failure
+	})))
+	initial, err := engine.Run(context.Background(), promptkit.RunRequest{PromptID: "prompt", CaptureExchange: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	before := initial.Exchange.Details()
+	var retained promptkit.PreparedRun
+	req := promptkit.RunRequest{PromptID: "prompt", ContinueFrom: initial.Exchange, APIKey: "synthetic-key", SessionID: "session", AppendedMessages: []promptkit.RenderedMessage{{Role: promptkit.RoleUser, Content: "follow-up", CacheControl: &promptkit.CacheControl{Type: promptkit.CacheControlEphemeral}}}, OnPrepared: func(_ context.Context, details promptkit.PreparedRun) {
+		retained = details
+		details.Messages[0].Content = "changed"
+		details.Messages[3].Content = "changed"
+		details.Messages[3].CacheControl.Type = "changed"
+	}}
+	result, err := engine.Run(context.Background(), req)
+	if result != nil || !errors.Is(err, failure) || retained.ParentRunID != initial.RunID || retained.SessionID != "session" || !reflect.DeepEqual(before, initial.Exchange.Details()) {
+		t.Fatalf("result=%v error=%v retained=%+v", result, err, retained)
+	}
+}
+
+func TestPreparedNotificationInvalidHandlesAndPreparationOnlyCalls(t *testing.T) {
+	calls := 0
+	engine := newPreparedContractEngine(t, &preparedRecordingClient{response: &promptkit.GenerateResponse{Content: "ok"}}, "content")
+	foreign := newPreparedContractEngine(t, &preparedRecordingClient{response: &promptkit.GenerateResponse{Content: "ok"}}, "other")
+	req := promptkit.RunRequest{PromptID: "prepared", OnPrepared: func(context.Context, promptkit.PreparedRun) { calls++ }}
+	if _, err := engine.Prepare(context.Background(), req); err != nil {
+		t.Fatal(err)
+	}
+	if err := engine.ValidateRequest(context.Background(), req); err != nil {
+		t.Fatal(err)
+	}
+	handle, err := engine.PrepareExecution(context.Background(), req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	_ = handle.Details()
+	if _, err := foreign.RunPrepared(context.Background(), handle); !errors.Is(err, promptkit.ErrInvalidRequest) {
+		t.Fatal(err)
+	}
+	if _, err := engine.RunPrepared(context.Background(), nil); !errors.Is(err, promptkit.ErrInvalidRequest) {
+		t.Fatal(err)
+	}
+	handle.Discard()
+	if _, err := engine.RunPrepared(context.Background(), handle); !errors.Is(err, promptkit.ErrInvalidRequest) {
+		t.Fatal(err)
+	}
+	if _, err := engine.Run(context.Background(), promptkit.RunRequest{PromptID: "missing", OnPrepared: req.OnPrepared}); err == nil {
+		t.Fatal("expected preparation failure")
+	}
+	if calls != 0 {
+		t.Fatalf("unexpected notifications=%d", calls)
+	}
+	handle, err = engine.PrepareExecution(context.Background(), req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := engine.RunPrepared(context.Background(), handle); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := engine.RunPrepared(context.Background(), handle); !errors.Is(err, promptkit.ErrInvalidRequest) || calls != 1 {
+		t.Fatalf("reused handle error=%v notifications=%d", err, calls)
+	}
+}
+
+func TestNotificationHoldsAdmission(t *testing.T) {
+	engine := newBackendCapacityEngine(t, observationClientFunc(func(context.Context, promptkit.GenerateRequest) (*promptkit.GenerateResponse, error) {
+		return &promptkit.GenerateResponse{Content: "ok"}, nil
+	}), 1, capacityInt(0), nil)
+	_, err := engine.Run(context.Background(), promptkit.RunRequest{PromptID: "prompt", OnPrepared: func(ctx context.Context, _ promptkit.PreparedRun) {
+		if _, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "prompt"}); !errors.Is(err, promptkit.ErrCapacityExceeded) {
+			t.Errorf("nested admission error=%v", err)
+		}
+	}})
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestNotificationPanicReleasesCapacityAndConsumesHandle(t *testing.T) {
+	engine := newBackendCapacityEngine(t, observationClientFunc(func(context.Context, promptkit.GenerateRequest) (*promptkit.GenerateResponse, error) {
+		return &promptkit.GenerateResponse{Content: "ok"}, nil
+	}), 1, capacityInt(0), nil)
+	for _, split := range []bool{false, true} {
+		req := promptkit.RunRequest{PromptID: "prompt", OnPrepared: func(context.Context, promptkit.PreparedRun) { panic("notification") }}
+		var handle *promptkit.PreparedExecution
+		if split {
+			var err error
+			handle, err = engine.PrepareExecution(context.Background(), req)
+			if err != nil {
+				t.Fatal(err)
+			}
+		}
+		func() {
+			defer func() {
+				if recover() == nil {
+					t.Error("expected panic")
+				}
+			}()
+			if split {
+				_, _ = engine.RunPrepared(context.Background(), handle)
+			} else {
+				_, _ = engine.Run(context.Background(), req)
+			}
+		}()
+		if split {
+			if _, err := engine.RunPrepared(context.Background(), handle); !errors.Is(err, promptkit.ErrInvalidRequest) {
+				t.Fatal("handle reusable after panic")
+			}
+		}
+		if _, err := engine.Run(context.Background(), promptkit.RunRequest{PromptID: "prompt"}); err != nil {
+			t.Fatalf("capacity not released: %v", err)
+		}
+	}
+}
+
+func TestNotificationSkippedOnAdmissionAndPreExecutionCancellation(t *testing.T) {
+	release := make(chan struct{})
+	client := newCapacityGateClient(release, 1)
+	engine := newBackendCapacityEngine(t, client, 1, capacityInt(0), nil)
+	results := make(chan capacityRunResult, 1)
+	go runCapacityRequest(engine, context.Background(), promptkit.RunRequest{PromptID: "prompt"}, results)
+	<-client.started
+	calls := 0
+	req := promptkit.RunRequest{PromptID: "prompt", OnPrepared: func(context.Context, promptkit.PreparedRun) { calls++ }}
+	if _, err := engine.Run(context.Background(), req); !errors.Is(err, promptkit.ErrCapacityExceeded) {
+		t.Fatalf("admission error=%v", err)
+	}
+	close(release)
+	if result := <-results; result.err != nil {
+		t.Fatal(result.err)
+	}
+	handle, err := engine.PrepareExecution(context.Background(), req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	if _, err = engine.RunPrepared(ctx, handle); !errors.Is(err, context.Canceled) {
+		t.Fatalf("cancellation error=%v", err)
+	}
+	if calls != 0 {
+		t.Fatalf("notifications=%d", calls)
+	}
+}
+
+func TestCancellationInPreparedNotificationPreventsGeneration(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	engine := observationEngine(t, observationClientFunc(func(context.Context, promptkit.GenerateRequest) (*promptkit.GenerateResponse, error) {
+		t.Fatal("generation after cancellation")
+		return nil, nil
+	}))
+	_, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "prompt", OnPrepared: func(context.Context, promptkit.PreparedRun) { cancel() }})
+	if !errors.Is(err, context.Canceled) || !errors.Is(err, promptkit.ErrLLMGenerate) {
+		t.Fatalf("error=%v", err)
+	}
+}

reliable_execution_contract_test.go

@@ -46,10 +46,12 @@ func TestPublicRetryRepairSequenceKeepsUsageAndEventsConsistent(t *testing.T) {
 	}
 	var events []promptkit.ExecutionEvent
 	req := simpleGenerationRequest()
+	notifications := 0
+	req.OnPrepared = func(context.Context, promptkit.PreparedRun) { notifications++ }
 	req.Validation = &promptkit.OutputContract{Format: promptkit.FormatJSON, ValidationMode: promptkit.ValidationJSON, RepairAttempts: 1}
 	req.Observer = func(event promptkit.ExecutionEvent) { events = append(events, event) }
 	result, err := engine.Run(context.Background(), req)
-	if err != nil || result == nil || !result.Validation.IsValid || calls != 4 {
+	if err != nil || result == nil || !result.Validation.IsValid || calls != 4 || notifications != 1 {
 		t.Fatalf("result=%+v error=%v calls=%d", result, err, calls)
 	}
 	if len(events) != 8 {

types.go

@@ -85,6 +85,29 @@ const (
 // own private execution snapshot for RunPrepared. Excessively deep or large
 // JSON-shaped values are rejected for safety.
 type RunRequest struct {
+	// OperationTimeout selects a whole-operation budget when positive. Run
+	// applies it across preparation and execution; PrepareExecution retains it
+	// for RunPrepared, without counting preparation or idle time. Zero or a
+	// negative value selects no additional timeout. This is independent of the
+	// provider generation timeout and respects earlier parent cancellation or
+	// deadlines. Local timeout failures match ErrOperationTimeout and
+	// context.DeadlineExceeded. Cancellation remains cooperative; callbacks or
+	// collaborators that ignore context cannot be forcibly interrupted.
+	OperationTimeout time.Duration `json:"-" yaml:"-"`
+	// OnPrepared synchronously receives detached, credential-redacted details of
+	// the logical request after preparation and admission, before the first
+	// generation, at most once per logical execution if it reaches that point.
+	// Transient retries and structural repair do not repeat the notification.
+	// Its context is the operation-scoped context, subject to OperationTimeout
+	// and parent cancellation/deadlines. In [Engine.Coordinate], this differs
+	// from the parent context passed to the next-turn callback.
+	// Admission remains held, and callback time consumes OperationTimeout. The
+	// callback owns the delivered mutable values; its closure is shared and must
+	// synchronize any shared state. It has no error return, and a panic propagates
+	// after resource cleanup. Prepare and ValidateRequest do not invoke it;
+	// PrepareExecution retains the selected function for RunPrepared and Discard
+	// releases it.
+	OnPrepared func(context.Context, PreparedRun) `json:"-" yaml:"-"`
 	// CaptureExchange requests an immutable snapshot of a completed run, even
 	// when its structural validation fails. Operational errors return no result.
 	// The snapshot is returned in RunResult.Exchange; see [Exchange] for its
@@ -167,7 +190,8 @@ type RunRequest struct {
 
 // PreparedRun contains prepared prompt execution state returned by
 // [Engine.Prepare] or [PreparedExecution.Details]. It does not include resolved
-// API key values, model output, validation results, or internal target presence
+// API key values. RunRequest.OnPrepared also supplies this detached value.
+// It does not include model output, validation results, or internal target presence
 // metadata. PreparedRun has a stable JSON representation.
 //
 // All maps, slices, pointers, and schema values are caller-owned copies. JSON