diff --git a/docs/internal/generation.md b/docs/internal/generation.md index 41491c1..a10953a 100644 --- a/docs/internal/generation.md +++ b/docs/internal/generation.md @@ -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 diff --git a/docs/internal/observation.md b/docs/internal/observation.md index d533b7c..12330ee 100644 --- a/docs/internal/observation.md +++ b/docs/internal/observation.md @@ -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 diff --git a/docs/roadmap/remediation.md b/docs/roadmap/remediation.md index bb8118c..e7f41b6 100644 --- a/docs/roadmap/remediation.md +++ b/docs/roadmap/remediation.md @@ -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. diff --git a/engine.go b/engine.go index b4307a7..5c6e7c6 100644 --- a/engine.go +++ b/engine.go @@ -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) } diff --git a/internal/observation/recorder.go b/internal/observation/recorder.go index 223f3bb..c66ef70 100644 --- a/internal/observation/recorder.go +++ b/internal/observation/recorder.go @@ -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) diff --git a/internal/observation/recorder_test.go b/internal/observation/recorder_test.go index 1b7de5d..ac34777 100644 --- a/internal/observation/recorder_test.go +++ b/internal/observation/recorder_test.go @@ -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 diff --git a/llm_adapter.go b/llm_adapter.go index d4a2c46..454fca3 100644 --- a/llm_adapter.go +++ b/llm_adapter.go @@ -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 } diff --git a/observation_boundary_contract_test.go b/observation_boundary_contract_test.go new file mode 100644 index 0000000..f598d81 --- /dev/null +++ b/observation_boundary_contract_test.go @@ -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) + } + }) + } +}