diff --git a/convert.go b/convert.go index 9a0413d..bbf1052 100644 --- a/convert.go +++ b/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 } diff --git a/docs/consumers/pkg-promptkit.md b/docs/consumers/pkg-promptkit.md index c06600b..c8cb089 100644 --- a/docs/consumers/pkg-promptkit.md +++ b/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 diff --git a/docs/internal/observation.md b/docs/internal/observation.md index 3d8c015..4d40568 100644 --- a/docs/internal/observation.md +++ b/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 diff --git a/docs/internal/runner.md b/docs/internal/runner.md index a53d145..c8e7aba 100644 --- a/docs/internal/runner.md +++ b/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. diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index c97ce2d..798cbc7 100644 --- a/docs/roadmap/implementation.md +++ b/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 diff --git a/engine.go b/engine.go index 901364c..c3954a3 100644 --- a/engine.go +++ b/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 } diff --git a/internal/domain/domain.go b/internal/domain/domain.go index 6dd847f..b2dede8 100644 --- a/internal/domain/domain.go +++ b/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. diff --git a/internal/observation/recorder.go b/internal/observation/recorder.go index 1432a9f..223f3bb 100644 --- a/internal/observation/recorder.go +++ b/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 diff --git a/internal/usecase/prepared_execution.go b/internal/usecase/prepared_execution.go index 6612b66..bdef990 100644 --- a/internal/usecase/prepared_execution.go +++ b/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 { diff --git a/prepared_execution.go b/prepared_execution.go index d7a9c84..8572cc5 100644 --- a/prepared_execution.go +++ b/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 } diff --git a/prepared_observation_contract_test.go b/prepared_observation_contract_test.go new file mode 100644 index 0000000..f006bb3 --- /dev/null +++ b/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) + } +} diff --git a/types.go b/types.go index 4e0814c..436ae29 100644 --- a/types.go +++ b/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:"-"` }