eric/promptkit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Isolate execution observations and preserve injected error categories

Browse Source
This commit is contained in:
Eric Rakestraw
2026-09-23 17:56:15 +00:00
parent 48378a3269
commit 0c08038b4f
8 changed files with 291 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
docs/internal/generation.md
View File

@@ -35,6 +35,12 @@ unknown usage. Transport, timeout, malformed, and noneligible status failures
stop the sequence without retry exhaustion. Context is checked before each
wait, acquisition, and dispatch.
The public client adapter translates explicitly supplied public transport and
malformed-response identities into internal categories while retaining the
original error chain. This keeps injected attempt classifications consistent
with terminal observations without parsing diagnostic text or retrying injected
clients.
Focused coordinator tests use controlled transport and wait functions. They
cover eligibility, exhaustion, cancellation, backoff, request freezing, permit
release, and the injected-client path. The capacity manager tests own permit

6
docs/internal/observation.md
View File

@@ -11,6 +11,12 @@ 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.
The facade masks inherited recorder and prepared-claim state when entering
preparation or execution. Delegated calls from an injected collaborator retain
the caller's cancellation, deadline, and unrelated context values, but do not
reuse the enclosing invocation's accounting. Only the new request's observer
(or the claimed handle's retained observer) enables its observation lifecycle.
A recorder keeps only counters, resolved identity fields, and the selected
callback. It has no global history, queue, goroutine, or mutex held across a
callback. The caller supplies normalized failure metadata; the recorder never

43
docs/roadmap/remediation.md
View File

@@ -8,15 +8,15 @@ The intended architecture is substantially implemented: the public facade
adapts contracts, the runner owns execution and repair, one generation
coordinator owns retries and active permits, the HTTP adapter freezes requests
and decodes responses, and an invocation-local recorder produces observations.
The following two correctness gaps need remediation before release. This is a
review record, not an implementation change or a new public contract.
The two correctness gaps below have been remediated. This document preserves
the review evidence and resolution, not a new public contract.
## 1. Isolate Observation State Across Public Invocations (P2)
**Status:** Open.
**Status:** Resolved.
**Location:** [Engine.Run](../../engine.go), lines 535–547;
[Runner.Run](../../internal/usecase/runner.go), lines 138–145;
**Location:** [Engine.Run](../../engine.go);
[Runner.Run](../../internal/usecase/runner.go);
[recorder context propagation](../../internal/observation/recorder.go).
The observer is request-scoped, but its recorder is attached to the context
@@ -48,13 +48,21 @@ child with its own observer must have its own lifecycle. Include focused
prepared/preparation coverage where the same context plumbing is used, and
race-enabled coverage for concurrent delegation without sharing recorder state.
**Resolution:** Preparation and execution facade entry points mask inherited
recorder/claim state before installing any invocation-owned state. The context
boundary preserves caller values, deadlines, cancellation, and cancellation
causes. Public regression tests cover observed and unobserved delegated ordinary
and prepared runs, preparation isolation, independent accounting/identities,
and concurrent delegation. A focused recorder test checks context preservation
and claim-sink isolation.
## 2. Preserve Injected Failure Categories In Attempt Events (P2)
**Status:** Open.
**Status:** Resolved.
**Location:** [public client adapter](../../llm_adapter.go), lines 14–20;
[attempt classification](../../internal/generation/coordinator.go), lines
170–172 and 235–247; [terminal classification](../../observations.go).
**Location:** [public client adapter](../../llm_adapter.go);
[attempt classification](../../internal/generation/coordinator.go);
[terminal classification](../../observations.go).
The public adapter returns public error sentinels, including
`ErrMalformedResponse` for a successful nil response. The coordinator recognizes
@@ -81,15 +89,22 @@ error should assert both attempt and terminal classifications, preserved error
identity, and single dispatch. Retain context-first classification for wrapped
cancellation/deadline causes.
**Resolution:** The facade adapter adds internal classification identities for
explicit public transport/malformed errors, retaining the original diagnostic
and error chain. Nil success uses the same normalization path. Public tests
check attempt/terminal agreement, `errors.Is`/`errors.As`, single dispatch even
with a positive retry budget, arbitrary errors, and context-first classification.
## Validation And Review Outcome
- `./scripts/release-check.sh` passed on the committed implementation, including
- `./scripts/release-check.sh` passed on the review baseline and remediated tree, including
race-enabled tests, vet, the three offline examples, formatting, documentation
links, and repository hygiene checks.
- Temporary public-API probes reproduced both findings. They were removed after
review; no implementation or permanent test changes are included here.
- Temporary public-API probes reproduced both findings during review. They were
removed and replaced by focused permanent regression coverage during remediation.
- The existing tests provide useful coverage of retries, repair interaction,
capacity, prepared claims, usage presence, and redaction, but do not protect
these two observation-boundary cases.
capacity, prepared claims, usage presence, and redaction, but did not protect
these two observation-boundary cases before remediation. The new
[boundary tests](../../observation_boundary_contract_test.go) protect them.
- No other must-fix issue was identified in this review. This does not establish
live-provider behavior; all verification was offline.

4
engine.go
View File

@@ -441,6 +441,7 @@ func (e *Engine) InspectProfile(ctx context.Context, profileID string) (*Profile
// relationship to ctx.Err is promised. Prepare returns no partial result on
// error. It validates RunRequest.TransientRetries but makes no model attempts.
func (e *Engine) Prepare(ctx context.Context, req RunRequest) (*PreparedRun, error) {
ctx = observation.WithoutExecutionState(ctx)
if e == nil || e.internal == nil {
return nil, fmt.Errorf("%w: engine is nil", ErrInvalidConfig)
}
@@ -476,6 +477,7 @@ func (e *Engine) Prepare(ctx context.Context, req RunRequest) (*PreparedRun, err
// 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) {
ctx = observation.WithoutExecutionState(ctx)
if e == nil || e.internal == nil {
return nil, fmt.Errorf("%w: engine is nil", ErrInvalidConfig)
}
@@ -533,6 +535,7 @@ func (e *Engine) PrepareExecution(ctx context.Context, req RunRequest) (*Prepare
// behavior. A nil Engine returns ErrInvalidConfig. Run returns no partial
// result on error.
func (e *Engine) Run(ctx context.Context, req RunRequest) (*RunResult, error) {
ctx = observation.WithoutExecutionState(ctx)
if req.Observer == nil {
return e.run(ctx, req)
}
@@ -603,6 +606,7 @@ func (e *Engine) run(ctx context.Context, req RunRequest) (*RunResult, error) {
// 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) {
ctx = observation.WithoutExecutionState(ctx)
if e == nil || e.internal == nil {
return nil, fmt.Errorf("%w: engine is nil", ErrInvalidConfig)
}

18
internal/observation/recorder.go
View File

@@ -191,6 +191,24 @@ func (r *Recorder) deliver(event domain.ExecutionEvent) {
type contextKey struct{}
// WithoutExecutionState starts a public operation without inheriting another
// invocation's recorder or prepared-claim sink. Other context values and all
// cancellation/deadline behavior remain with the caller's context.
func WithoutExecutionState(ctx context.Context) context.Context {
return executionBoundary{Context: ctx}
}
type executionBoundary struct{ context.Context }
func (c executionBoundary) Value(key any) any {
switch key.(type) {
case contextKey, claimSinkKey:
return nil
default:
return c.Context.Value(key)
}
}
// WithRecorder carries one invocation's recorder through internal client calls.
func WithRecorder(ctx context.Context, recorder *Recorder) context.Context {
return context.WithValue(ctx, contextKey{}, recorder)

30
internal/observation/recorder_test.go
View File

@@ -2,6 +2,7 @@ package observation
import (
"context"
"errors"
"reflect"
"testing"
"time"
@@ -9,6 +10,35 @@ import (
"gitea.maximumdirect.net/eric/promptkit/internal/domain"
)
func TestExecutionBoundaryPreservesCallerContext(t *testing.T) {
type callerKey struct{}
parent, stop := context.WithTimeout(context.WithValue(context.Background(), callerKey{}, "caller"), time.Minute)
defer stop()
parent, cancel := context.WithCancelCause(parent)
defer cancel(nil)
recorder := newRecorder("outer", time.Now, nil)
sink := recorder
parent = WithClaimSink(WithRecorder(parent, recorder), &sink)
child := WithoutExecutionState(parent)
if FromContext(child) != nil || FromContext(parent) != recorder {
t.Fatal("execution boundary did not isolate recorder state")
}
RegisterClaim(child, nil)
if sink != recorder {
t.Fatal("execution boundary retained the parent claim sink")
}
wantDeadline, _ := parent.Deadline()
gotDeadline, ok := child.Deadline()
if !ok || gotDeadline != wantDeadline || child.Value(callerKey{}) != "caller" || child.Done() != parent.Done() {
t.Fatal("execution boundary changed caller context")
}
cause := errors.New("caller canceled")
cancel(cause)
if !errors.Is(child.Err(), context.Canceled) || context.Cause(child) != cause {
t.Fatal("execution boundary lost cancellation or its cause")
}
}
func TestRecorderAccountsDispatchedAttemptsAndReportedUsage(t *testing.T) {
start := time.Date(2026, 9, 23, 12, 0, 0, 0, time.UTC)
tick := 0

27
llm_adapter.go
View File

@@ -2,9 +2,11 @@ package promptkit
import (
"context"
"errors"
"fmt"
"gitea.maximumdirect.net/eric/promptkit/internal/domain"
"gitea.maximumdirect.net/eric/promptkit/internal/llm"
)
type publicLLMClientAdapter struct {
@@ -13,11 +15,28 @@ type publicLLMClientAdapter struct {
func (a publicLLMClientAdapter) Generate(ctx context.Context, req domain.GenerateRequest) (*domain.GenerateResponse, error) {
resp, err := a.client.Generate(ctx, fromDomainGenerateRequest(req))
if err != nil {
return nil, err
if err == nil && resp == nil {
err = fmt.Errorf("%w: %w: llm client returned nil response", ErrLLMGenerate, ErrMalformedResponse)
}
if resp == nil {
return nil, fmt.Errorf("%w: %w: llm client returned nil response", ErrLLMGenerate, ErrMalformedResponse)
if err != nil {
switch {
case errors.Is(err, ErrTransport):
return nil, classifiedClientError{cause: err, category: llm.ErrTransport}
case errors.Is(err, ErrMalformedResponse):
return nil, classifiedClientError{cause: err, category: llm.ErrMalformedResponse}
}
return nil, err
}
return toDomainGenerateResponse(resp), nil
}
// classifiedClientError makes an explicit public category available internally
// without replacing the client's diagnostic or original error chain.
type classifiedClientError struct {
cause error
category error
}
func (e classifiedClientError) Error() string { return e.cause.Error() }
func (e classifiedClientError) Unwrap() error { return e.cause }
func (e classifiedClientError) Is(target error) bool { return target == e.category }

175
observation_boundary_contract_test.go Normal file
View File

@@ -0,0 +1,175 @@
package promptkit_test
import (
"context"
"errors"
"fmt"
"testing"
"time"
"gitea.maximumdirect.net/eric/promptkit"
)
type observationClientFunc func(context.Context, promptkit.GenerateRequest) (*promptkit.GenerateResponse, error)
func (f observationClientFunc) Generate(ctx context.Context, req promptkit.GenerateRequest) (*promptkit.GenerateResponse, error) {
return f(ctx, req)
}
func TestDelegatedOperationsIsolateObservations(t *testing.T) {
for _, mode := range []string{"run", "observed run", "prepared", "observed prepared", "prepare", "prepare execution"} {
t.Run(mode, func(t *testing.T) {
type contextKey struct{}
ctx, cancel := context.WithTimeout(context.WithValue(context.Background(), contextKey{}, "caller-value"), time.Minute)
defer cancel()
deadline, _ := ctx.Deadline()
child, err := promptkit.NewEngine(promptkit.Config{},
promptkit.WithPromptFS(contractPromptFS("child", "profile", "child message"), "."),
promptkit.WithProfiles(promptkit.Profile{ID: "profile", Endpoint: "https://example.invalid/v1", Model: "child-model"}),
promptkit.WithLLMClient(observationClientFunc(func(callCtx context.Context, _ promptkit.GenerateRequest) (*promptkit.GenerateResponse, error) {
gotDeadline, ok := callCtx.Deadline()
if !ok || gotDeadline != deadline || callCtx.Value(contextKey{}) != "caller-value" || callCtx.Done() != ctx.Done() {
t.Error("delegation lost caller context behavior")
}
return &promptkit.GenerateResponse{Content: "child", Usage: promptkit.TokenUsage{PromptTokens: 7}}, nil
})))
if err != nil {
t.Fatal(err)
}
var childEvents []promptkit.ExecutionEvent
var childResult *promptkit.RunResult
outer := observationEngine(t, observationClientFunc(func(callCtx context.Context, _ promptkit.GenerateRequest) (*promptkit.GenerateResponse, error) {
req := promptkit.RunRequest{PromptID: "child"}
if mode == "observed run" || mode == "observed prepared" {
req.Observer = func(event promptkit.ExecutionEvent) { childEvents = append(childEvents, event) }
}
var callErr error
switch mode {
case "run", "observed run":
childResult, callErr = child.Run(callCtx, req)
case "prepared", "observed prepared", "prepare execution":
var handle *promptkit.PreparedExecution
handle, callErr = child.PrepareExecution(callCtx, req)
if callErr == nil {
defer handle.Discard()
if mode != "prepare execution" {
childResult, callErr = child.RunPrepared(callCtx, handle)
}
}
case "prepare":
_, callErr = child.Prepare(callCtx, req)
}
return &promptkit.GenerateResponse{Content: "outer", Usage: promptkit.TokenUsage{PromptTokens: 1}}, callErr
}))
var outerEvents []promptkit.ExecutionEvent
result, err := outer.Run(ctx, promptkit.RunRequest{PromptID: "prompt", Observer: func(event promptkit.ExecutionEvent) { outerEvents = append(outerEvents, event) }})
if err != nil {
t.Fatal(err)
}
if len(outerEvents) != 3 {
t.Fatalf("outer events=%+v", outerEvents)
}
for _, event := range outerEvents[1:] {
if event.RunID != result.RunID || event.PromptID != "prompt" || event.Model != "model" || event.Usage.PromptTokens.Value != 1 {
t.Fatalf("child contaminated outer event: %+v", event)
}
}
if outerEvents[2].Attempts != 1 {
t.Fatalf("outer attempts=%d", outerEvents[2].Attempts)
}
if childResult != nil && childResult.RunID == result.RunID {
t.Fatal("delegated run reused outer run ID")
}
if mode == "observed run" || mode == "observed prepared" {
if len(childEvents) != 3 || childEvents[0].Kind != promptkit.ExecutionStarted || childEvents[2].Kind != promptkit.ExecutionFinished || childEvents[2].Attempts != 1 || childEvents[2].Usage.PromptTokens.Value != 7 || childEvents[2].RunID != childResult.RunID {
t.Fatalf("child lifecycle=%+v", childEvents)
}
} else if len(childEvents) != 0 {
t.Fatalf("unobserved child events=%+v", childEvents)
}
})
}
}
func TestConcurrentDelegationDoesNotShareRecorder(t *testing.T) {
child := observationEngine(t, observationClientFunc(func(context.Context, promptkit.GenerateRequest) (*promptkit.GenerateResponse, error) {
return &promptkit.GenerateResponse{Content: "child"}, nil
}))
ids := make(chan string, 2)
outer := observationEngine(t, observationClientFunc(func(ctx context.Context, _ promptkit.GenerateRequest) (*promptkit.GenerateResponse, error) {
errs := make(chan error, 2)
for range 2 {
go func() {
r, err := child.Run(ctx, promptkit.RunRequest{PromptID: "prompt"})
if r != nil {
ids <- r.RunID
}
errs <- err
}()
}
return &promptkit.GenerateResponse{Content: "outer"}, errors.Join(<-errs, <-errs)
}))
var events []promptkit.ExecutionEvent
r, err := outer.Run(context.Background(), promptkit.RunRequest{PromptID: "prompt", Observer: func(e promptkit.ExecutionEvent) { events = append(events, e) }})
if err != nil {
t.Fatal(err)
}
if len(events) != 3 || events[2].Attempts != 1 {
t.Fatalf("outer events=%+v", events)
}
first, second := <-ids, <-ids
if first == second || first == r.RunID || second == r.RunID {
t.Fatal("delegated identities are not independent")
}
}
type classifiedObservationTestError struct{ cause error }
func (e *classifiedObservationTestError) Error() string { return "injected failure" }
func (e *classifiedObservationTestError) Unwrap() error { return e.cause }
func TestInjectedAttemptAndTerminalFailureCategoriesAgree(t *testing.T) {
for _, tc := range []struct {
name string
cause error
want promptkit.ExecutionFailureKind
}{
{"nil success", nil, promptkit.ExecutionFailureMalformedResponse},
{"malformed", promptkit.ErrMalformedResponse, promptkit.ExecutionFailureMalformedResponse},
{"transport", promptkit.ErrTransport, promptkit.ExecutionFailureTransport},
{"arbitrary", errors.New("transport failed"), promptkit.ExecutionFailureGeneration},
{"canceled", errors.Join(promptkit.ErrTransport, context.Canceled), promptkit.ExecutionFailureCanceled},
{"deadline", errors.Join(promptkit.ErrMalformedResponse, context.DeadlineExceeded), promptkit.ExecutionFailureDeadline},
} {
t.Run(tc.name, func(t *testing.T) {
var injected error
var typed *classifiedObservationTestError
if tc.cause != nil {
typed = &classifiedObservationTestError{cause: tc.cause}
injected = fmt.Errorf("wrapped: %w", typed)
}
calls := 0
engine := observationEngine(t, observationClientFunc(func(context.Context, promptkit.GenerateRequest) (*promptkit.GenerateResponse, error) {
calls++
return nil, injected
}))
var events []promptkit.ExecutionEvent
retries := 3
result, err := engine.Run(context.Background(), promptkit.RunRequest{PromptID: "prompt", TransientRetries: &retries, Observer: func(e promptkit.ExecutionEvent) { events = append(events, e) }})
if result != nil || !errors.Is(err, promptkit.ErrLLMGenerate) || calls != 1 || len(events) != 3 {
t.Fatalf("result=%+v err=%v calls=%d events=%+v", result, err, calls, events)
}
if events[1].Failure != tc.want || events[2].Failure != tc.want {
t.Fatalf("attempt=%s terminal=%s want=%s", events[1].Failure, events[2].Failure, tc.want)
}
if injected != nil {
var got *classifiedObservationTestError
if !errors.Is(err, injected) || !errors.Is(err, tc.cause) || !errors.As(err, &got) || got != typed {
t.Fatalf("lost injected identity: %v", err)
}
} else if !errors.Is(err, promptkit.ErrMalformedResponse) {
t.Fatalf("nil success error=%v", err)
}
})
}
}