Bound ordinary runs with attributable operation timeouts

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.
 

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
 }

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):

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)
+	}
+}

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")
+		}
+	})
+}