From d42b131de2471b2a46045a9475d1f4224f3937bc Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sat, 14 Nov 2026 17:37:25 +0000 Subject: [PATCH] Bound ordinary runs with attributable operation timeouts --- docs/roadmap/implementation.md | 2 + engine.go | 15 +++++-- errors.go | 2 + execution_timeout_contract_test.go | 40 ++++++++++++++++++ internal/usecase/operation_scope.go | 36 ++++++++++++++++ internal/usecase/operation_scope_test.go | 52 ++++++++++++++++++++++++ 6 files changed, 144 insertions(+), 3 deletions(-) create mode 100644 execution_timeout_contract_test.go create mode 100644 internal/usecase/operation_scope.go create mode 100644 internal/usecase/operation_scope_test.go diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index cba6ccb..26430a1 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -174,6 +174,8 @@ until the execution wiring lands; do not claim the feature works at this stage. ## Stage 2: Bound Ordinary Execution And Preserve Its Error Outcome +Status: complete. + Outcome: `Engine.Run` owns one continuous optional budget and returns the same failure classification it publishes to observers. diff --git a/engine.go b/engine.go index e29b544..009441e 100644 --- a/engine.go +++ b/engine.go @@ -514,8 +514,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 { @@ -523,12 +529,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}) @@ -547,7 +556,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 } diff --git a/errors.go b/errors.go index 34b8a1e..3e45d33 100644 --- a/errors.go +++ b/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): diff --git a/execution_timeout_contract_test.go b/execution_timeout_contract_test.go new file mode 100644 index 0000000..8771163 --- /dev/null +++ b/execution_timeout_contract_test.go @@ -0,0 +1,40 @@ +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) + } +} diff --git a/internal/usecase/operation_scope.go b/internal/usecase/operation_scope.go new file mode 100644 index 0000000..c1aca0b --- /dev/null +++ b/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() } diff --git a/internal/usecase/operation_scope_test.go b/internal/usecase/operation_scope_test.go new file mode 100644 index 0000000..5224291 --- /dev/null +++ b/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") + } + }) +}