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

Add HTTP whole-run deadlines and context outcomes

Browse Source
This commit is contained in:
Eric Rakestraw
2026-09-26 15:39:17 +00:00
parent 911ca5485a
commit 7de2b02891
6 changed files with 217 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
internal/adapter/http/dto.go
View File

@@ -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 {

24
internal/adapter/http/handler.go
View File

@@ -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

172
internal/adapter/http/handler_test.go
View File

@@ -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()