diff --git a/docs/api.md b/docs/api.md index 5bd0830..926baf9 100644 --- a/docs/api.md +++ b/docs/api.md @@ -51,6 +51,7 @@ copyable shape. At the HTTP adapter boundary, the smallest valid shape is: | `model` | no | Runtime model-override object. | | `include_raw_output` | no | Include `raw_model_output` when true. | | `transient_retries` | no | Nonnegative additional transient attempt count for this run. Omitted or `null` inherits the configured engine value and then Promptkit's default; `0` disables additional attempts. | +| `run_timeout_seconds` | no | Optional signed 64-bit whole-second limit for engine execution. Omitted, `null`, or `0` adds no deadline; positive values shorten the caller context deadline. | An input reference has a required `type` of `file` or `inline`. A `file` reference requires `uri`; an `inline` reference requires `body`. @@ -98,6 +99,12 @@ returns `400 invalid_json`, while a negative integer returns `400 invalid_request` before execution. A request override does not change the shared engine setting used by other requests. Promptkit owns eligible statuses, retry waits, and the separate transient budget for each repair generation. +`run_timeout_seconds` must be a JSON int64. A wrong type or int64 overflow +returns `400 invalid_json`; a negative value or one whose conversion to a Go +duration would overflow returns `400 invalid_request`. The deadline starts +immediately before engine execution and excludes body decoding and response +writing. An earlier caller deadline or cancellation still wins. Synchronous +work is not guaranteed to stop immediately. ### Success Response @@ -157,6 +164,7 @@ Messages are concise and do not expose wrapped internal causes. | `404` | `prompt_not_found` | Prompt ID or version does not exist. | | `404` | `profile_not_found` | Profile ID does not exist. | | `405` | `method_not_allowed` | The route does not accept the method. | +| `408` | `request_canceled` | The caller context was canceled during a failed run. | | `413` | `request_too_large` | Encoded request exceeds its limit. | | `413` | `artifact_too_large` | File input exceeds its limit. | | `413` | `response_too_large` | Encoded response exceeds its limit. | @@ -164,6 +172,12 @@ Messages are concise and do not expose wrapped internal causes. | `500` | `internal_error` | Unclassified server failure. | | `502` | `llm_failed` | Outbound model request failed. | | `503` | `capacity_exceeded` | The selected model backend has no admission capacity. No retry timing is supplied. | +| `504` | `deadline_exceeded` | The execution context's deadline expired during a failed run. | + +These context outcomes require the execution context to have ended. A provider +attempt timeout while it remains live retains `502 llm_failed`. A successful +run remains successful even if cancellation races with response formatting; +delivery after a client disconnect cannot be guaranteed. ## Retry And Idempotency diff --git a/docs/operations.md b/docs/operations.md index 3c3bf1c..0c04ec7 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -128,6 +128,10 @@ be bounded. The existing generation timeout applies afresh to each attempt and does not cap retry waits or the complete invocation. The whole-run interval covers the engine call, including capacity waiting and repairs as supported by context cancellation; it excludes parsing, engine setup, and output writes. +HTTP callers can use [`run_timeout_seconds`](api.md) to shorten their request +context for the same engine interval. The API distinguishes an expired +execution deadline from caller cancellation and from a provider-attempt +timeout. A canceled capacity waiter releases its Promptkit admission state. Before increasing a limit: diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 5162f48..215ded6 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -210,7 +210,7 @@ Existing per-attempt timeout tests and generated-output behavior remain intact. ## Stage 5: Add HTTP Whole-Run Deadlines And Context Outcomes -**Status: Pending.** +**Status: Complete.** Selected interface: top-level `run_timeout_seconds`, an optional signed 64-bit integer. Omission/null/zero adds no deadline; positive whole seconds diff --git a/internal/adapter/http/dto.go b/internal/adapter/http/dto.go index e9dee4e..ca3ff9b 100644 --- a/internal/adapter/http/dto.go +++ b/internal/adapter/http/dto.go @@ -5,15 +5,16 @@ import ( ) type runRequestDTO struct { - PromptID string `json:"prompt_id"` - PromptVersion string `json:"prompt_version,omitempty"` - ProfileID string `json:"profile_id,omitempty"` - SessionID string `json:"session_id,omitempty"` - Inputs map[string]inputRefDTO `json:"inputs"` - Vars map[string]string `json:"vars,omitempty"` - Model *modelOverrideRequestDTO `json:"model,omitempty"` - IncludeRawOutput bool `json:"include_raw_output,omitempty"` - TransientRetries *int `json:"transient_retries,omitempty"` + PromptID string `json:"prompt_id"` + PromptVersion string `json:"prompt_version,omitempty"` + ProfileID string `json:"profile_id,omitempty"` + SessionID string `json:"session_id,omitempty"` + Inputs map[string]inputRefDTO `json:"inputs"` + Vars map[string]string `json:"vars,omitempty"` + Model *modelOverrideRequestDTO `json:"model,omitempty"` + IncludeRawOutput bool `json:"include_raw_output,omitempty"` + TransientRetries *int `json:"transient_retries,omitempty"` + RunTimeoutSeconds *int64 `json:"run_timeout_seconds,omitempty"` } type inputRefDTO struct { diff --git a/internal/adapter/http/handler.go b/internal/adapter/http/handler.go index 72996ac..55008e3 100644 --- a/internal/adapter/http/handler.go +++ b/internal/adapter/http/handler.go @@ -5,8 +5,10 @@ import ( "encoding/json" "errors" "io" + "math" "net/http" "strings" + "time" "gitea.maximumdirect.net/eric/promptkit" "gitea.maximumdirect.net/eric/scriptorium/internal/defaults" @@ -80,6 +82,12 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "invalid_request", "transient_retries must be greater than or equal to 0") return } + if req.RunTimeoutSeconds != nil { + if *req.RunTimeoutSeconds < 0 || *req.RunTimeoutSeconds > math.MaxInt64/int64(time.Second) { + writeError(w, http.StatusBadRequest, "invalid_request", "run_timeout_seconds is out of range") + return + } + } var mappedInputs map[string]promptkit.ArtifactRef if len(req.Inputs) > 0 { mappedInputs = make(map[string]promptkit.ArtifactRef, len(req.Inputs)) @@ -97,7 +105,13 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { model = executionTargetOverrideFromModelOverrideDTO(req.Model) } - res, err := h.runner.Run(r.Context(), promptkit.RunRequest{ + runContext := r.Context() + if req.RunTimeoutSeconds != nil && *req.RunTimeoutSeconds > 0 { + var cancel context.CancelFunc + runContext, cancel = context.WithTimeout(runContext, time.Duration(*req.RunTimeoutSeconds)*time.Second) + defer cancel() + } + res, err := h.runner.Run(runContext, promptkit.RunRequest{ PromptID: req.PromptID, PromptVersion: req.PromptVersion, ProfileID: req.ProfileID, @@ -108,6 +122,14 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { TransientRetries: req.TransientRetries, }) if err != nil { + switch runContext.Err() { + case context.DeadlineExceeded: + writeError(w, http.StatusGatewayTimeout, "deadline_exceeded", "run execution deadline exceeded") + return + case context.Canceled: + writeError(w, http.StatusRequestTimeout, "request_canceled", "run request canceled") + return + } status, code, message := mapRunError(err) writeError(w, status, code, message) return diff --git a/internal/adapter/http/handler_test.go b/internal/adapter/http/handler_test.go index 29b10f1..6c2286d 100644 --- a/internal/adapter/http/handler_test.go +++ b/internal/adapter/http/handler_test.go @@ -24,6 +24,12 @@ type fakeRunner struct { last promptkit.RunRequest } +type runnerFunc func(context.Context, promptkit.RunRequest) (*promptkit.RunResult, error) + +func (run runnerFunc) Run(ctx context.Context, req promptkit.RunRequest) (*promptkit.RunResult, error) { + return run(ctx, req) +} + func (f *fakeRunner) Run(ctx context.Context, req promptkit.RunRequest) (*promptkit.RunResult, error) { f.last = req if f.err != nil { @@ -98,6 +104,86 @@ func TestHandlerTransientRetryOverridePresenceAndStrictness(t *testing.T) { } } +func TestHandlerRunTimeoutValidationAndContextMapping(t *testing.T) { + for _, tc := range []struct { + name, body, code string + status int + }{ + {name: "negative", body: `{"prompt_id":"p","run_timeout_seconds":-1}`, status: http.StatusBadRequest, code: "invalid_request"}, + {name: "duration overflow", body: `{"prompt_id":"p","run_timeout_seconds":9223372037}`, status: http.StatusBadRequest, code: "invalid_request"}, + {name: "string", body: `{"prompt_id":"p","run_timeout_seconds":"1"}`, status: http.StatusBadRequest, code: "invalid_json"}, + {name: "fractional", body: `{"prompt_id":"p","run_timeout_seconds":1.5}`, status: http.StatusBadRequest, code: "invalid_json"}, + {name: "int64 overflow", body: `{"prompt_id":"p","run_timeout_seconds":9223372036854775808}`, status: http.StatusBadRequest, code: "invalid_json"}, + } { + t.Run(tc.name, func(t *testing.T) { + runner := &fakeRunner{result: &promptkit.RunResult{}} + w := httptest.NewRecorder() + NewHandler(runner).ServeHTTP(w, httptest.NewRequest(http.MethodPost, "/v1/runs", strings.NewReader(tc.body))) + assertHTTPErrorCode(t, w, tc.status, tc.code) + if runner.last.PromptID != "" { + t.Fatalf("runner invoked for invalid timeout: %+v", runner.last) + } + }) + } + + parent, cancel := context.WithDeadline(context.Background(), time.Now().Add(5*time.Second)) + defer cancel() + parentDeadline, _ := parent.Deadline() + for _, tc := range []struct { + name, body string + child bool + }{ + {name: "omitted", body: `{"prompt_id":"p"}`}, + {name: "null", body: `{"prompt_id":"p","run_timeout_seconds":null}`}, + {name: "zero", body: `{"prompt_id":"p","run_timeout_seconds":0}`}, + {name: "positive", body: `{"prompt_id":"p","run_timeout_seconds":10}`, child: true}, + } { + t.Run(tc.name, func(t *testing.T) { + called := false + h := NewHandler(runnerFunc(func(ctx context.Context, req promptkit.RunRequest) (*promptkit.RunResult, error) { + called = true + deadline, ok := ctx.Deadline() + if !ok || !deadline.Equal(parentDeadline) || (ctx != parent) != tc.child { + t.Fatalf("unexpected execution context: deadline=%v child=%t", deadline, ctx != parent) + } + return &promptkit.RunResult{}, nil + })) + w := httptest.NewRecorder() + h.ServeHTTP(w, httptest.NewRequest(http.MethodPost, "/v1/runs", strings.NewReader(tc.body)).WithContext(parent)) + if !called || w.Code != http.StatusOK { + t.Fatalf("context mapping: called=%t status=%d body=%s", called, w.Code, w.Body.String()) + } + }) + } + + canceled, cancelParent := context.WithCancel(context.Background()) + cancelParent() + h := NewHandler(runnerFunc(func(ctx context.Context, req promptkit.RunRequest) (*promptkit.RunResult, error) { + return nil, ctx.Err() + })) + w := httptest.NewRecorder() + h.ServeHTTP(w, httptest.NewRequest(http.MethodPost, "/v1/runs", strings.NewReader(`{"prompt_id":"p","run_timeout_seconds":10}`)).WithContext(canceled)) + assertHTTPErrorCode(t, w, http.StatusRequestTimeout, "request_canceled") + + h = NewHandler(runnerFunc(func(ctx context.Context, req promptkit.RunRequest) (*promptkit.RunResult, error) { + return nil, fmt.Errorf("attempt: %w: %w", promptkit.ErrLLMGenerate, context.DeadlineExceeded) + })) + w = httptest.NewRecorder() + h.ServeHTTP(w, httptest.NewRequest(http.MethodPost, "/v1/runs", strings.NewReader(`{"prompt_id":"p","run_timeout_seconds":10}`))) + assertHTTPErrorCode(t, w, http.StatusBadGateway, "llm_failed") + + successContext, cancelSuccess := context.WithCancel(context.Background()) + h = NewHandler(runnerFunc(func(ctx context.Context, req promptkit.RunRequest) (*promptkit.RunResult, error) { + cancelSuccess() + return &promptkit.RunResult{}, nil + })) + w = httptest.NewRecorder() + h.ServeHTTP(w, httptest.NewRequest(http.MethodPost, "/v1/runs", strings.NewReader(`{"prompt_id":"p","run_timeout_seconds":10}`)).WithContext(successContext)) + if w.Code != http.StatusOK { + t.Fatalf("successful result was reclassified after cancellation: %d %s", w.Code, w.Body.String()) + } +} + func intPointer(value int) *int { return &value } type handlerLLMClient struct{} @@ -840,13 +926,6 @@ func TestHandlerBuiltInTransientRetriesInheritAndAllowExplicitZero(t *testing.T) func TestHandlerRetainsSharedBackendLimitDuringRetry(t *testing.T) { started := make(chan struct{}, 1) release := make(chan struct{}) - defer func() { - select { - case <-release: - default: - close(release) - } - }() var hits, active, peak int32 provider := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { current := atomic.AddInt32(&active, 1) @@ -867,6 +946,13 @@ func TestHandlerRetainsSharedBackendLimitDuringRetry(t *testing.T) { _, _ = w.Write([]byte(`{"choices":[{"message":{"role":"assistant","content":"ok"}}]}`)) })) defer provider.Close() + defer func() { + select { + case <-release: + default: + close(release) + } + }() queueCapacity := 0 h := newRetryHandler(t, provider.URL+"/v1", intPointer(1), 1, &queueCapacity) first := httptest.NewRecorder() @@ -894,6 +980,78 @@ func TestHandlerRetainsSharedBackendLimitDuringRetry(t *testing.T) { } } +func TestHandlerDeadlineCancelsBuiltInRetryWait(t *testing.T) { + var hits int32 + provider := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&hits, 1) + w.Header().Set("Retry-After", "3600") + http.Error(w, "temporary", http.StatusServiceUnavailable) + })) + defer provider.Close() + h := newRetryHandler(t, provider.URL+"/v1", intPointer(1), 0, nil) + w := httptest.NewRecorder() + h.ServeHTTP(w, httptest.NewRequest(http.MethodPost, "/v1/runs", strings.NewReader(`{"prompt_id":"p","run_timeout_seconds":1}`))) + assertHTTPErrorCode(t, w, http.StatusGatewayTimeout, "deadline_exceeded") + if atomic.LoadInt32(&hits) != 1 { + t.Fatalf("expected one provider call before canceled retry wait, got %d", atomic.LoadInt32(&hits)) + } +} + +func TestHandlerDeadlineCancelsCapacityWaitAndReleasesAdmission(t *testing.T) { + started := make(chan struct{}, 1) + release := make(chan struct{}) + var hits int32 + provider := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if atomic.AddInt32(&hits, 1) == 1 { + started <- struct{}{} + <-release + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"choices":[{"message":{"role":"assistant","content":"ok"}}]}`)) + })) + defer provider.Close() + defer func() { + select { + case <-release: + default: + close(release) + } + }() + queue := 1 + h := newRetryHandler(t, provider.URL+"/v1", intPointer(0), 1, &queue) + first := httptest.NewRecorder() + done := make(chan struct{}) + go func() { + defer close(done) + h.ServeHTTP(first, httptest.NewRequest(http.MethodPost, "/v1/runs", strings.NewReader(`{"prompt_id":"p"}`))) + }() + select { + case <-started: + case <-time.After(3 * time.Second): + t.Fatal("first provider call did not start") + } + waiting := httptest.NewRecorder() + h.ServeHTTP(waiting, httptest.NewRequest(http.MethodPost, "/v1/runs", strings.NewReader(`{"prompt_id":"p","run_timeout_seconds":1}`))) + assertHTTPErrorCode(t, waiting, http.StatusGatewayTimeout, "deadline_exceeded") + if atomic.LoadInt32(&hits) != 1 { + t.Fatalf("capacity waiter reached provider: hits=%d", atomic.LoadInt32(&hits)) + } + close(release) + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("first request did not finish") + } + if first.Code != http.StatusOK { + t.Fatalf("first request: status=%d body=%s", first.Code, first.Body.String()) + } + third := httptest.NewRecorder() + h.ServeHTTP(third, httptest.NewRequest(http.MethodPost, "/v1/runs", strings.NewReader(`{"prompt_id":"p"}`))) + if third.Code != http.StatusOK || atomic.LoadInt32(&hits) != 2 { + t.Fatalf("admission was not released: status=%d hits=%d body=%s", third.Code, atomic.LoadInt32(&hits), third.Body.String()) + } +} + func newRetryHandler(t *testing.T, endpoint string, retries *int, concurrency int, queue *int) *Handler { t.Helper() promptDir := t.TempDir()