Expose backend identity and capacity errors
This commit is contained in:
@@ -57,6 +57,7 @@ type metadataDTO struct {
|
||||
PromptHash string `json:"prompt_hash"`
|
||||
RenderedPromptHash string `json:"rendered_prompt_hash"`
|
||||
SelectedProfileID string `json:"selected_profile_id"`
|
||||
SelectedBackendID string `json:"selected_backend_id,omitempty"`
|
||||
SessionID string `json:"session_id,omitempty"`
|
||||
ModelName string `json:"model_name"`
|
||||
Endpoint string `json:"endpoint"`
|
||||
@@ -73,6 +74,7 @@ type metadataDTO struct {
|
||||
|
||||
type modelParamsDTO struct {
|
||||
Endpoint string `json:"endpoint"`
|
||||
BackendID string `json:"backend_id,omitempty"`
|
||||
Model string `json:"model"`
|
||||
Temperature float64 `json:"temperature"`
|
||||
MaxTokens int `json:"max_tokens"`
|
||||
|
||||
@@ -125,6 +125,7 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
PromptHash: res.PromptHash,
|
||||
RenderedPromptHash: res.RenderedPromptHash,
|
||||
SelectedProfileID: res.SelectedProfileID,
|
||||
SelectedBackendID: res.SelectedBackendID,
|
||||
SessionID: res.SessionID,
|
||||
ModelName: res.ModelName,
|
||||
Endpoint: res.Endpoint,
|
||||
@@ -173,6 +174,7 @@ func executionTargetOverrideFromModelOverrideDTO(dto *modelOverrideRequestDTO) *
|
||||
func modelParamsDTOFromExecutionTarget(target promptkit.ExecutionTarget) modelParamsDTO {
|
||||
return modelParamsDTO{
|
||||
Endpoint: target.Endpoint,
|
||||
BackendID: target.BackendID,
|
||||
Model: target.Model,
|
||||
Temperature: target.Temperature,
|
||||
MaxTokens: target.MaxTokens,
|
||||
@@ -220,6 +222,8 @@ func mapRunError(err error) (int, string, string) {
|
||||
return http.StatusBadRequest, "artifact_read_failed", "failed to read input artifact"
|
||||
case errors.Is(err, promptkit.ErrPromptRender):
|
||||
return http.StatusBadRequest, "prompt_render_failed", "failed to render prompt"
|
||||
case errors.Is(err, promptkit.ErrCapacityExceeded):
|
||||
return http.StatusServiceUnavailable, "capacity_exceeded", "model backend capacity is exhausted"
|
||||
case errors.Is(err, promptkit.ErrLLMGenerate):
|
||||
return http.StatusBadGateway, "llm_failed", "model generation request failed"
|
||||
case errors.Is(err, promptkit.ErrValidation):
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -69,6 +70,34 @@ func (handlerLLMClient) Generate(ctx context.Context, req promptkit.GenerateRequ
|
||||
return &promptkit.GenerateResponse{Content: "ok"}, nil
|
||||
}
|
||||
|
||||
type blockingLLMClient struct {
|
||||
started chan struct{}
|
||||
release chan struct{}
|
||||
current int32
|
||||
peak int32
|
||||
}
|
||||
|
||||
func (c *blockingLLMClient) Generate(ctx context.Context, req promptkit.GenerateRequest) (*promptkit.GenerateResponse, error) {
|
||||
current := atomic.AddInt32(&c.current, 1)
|
||||
defer atomic.AddInt32(&c.current, -1)
|
||||
for {
|
||||
peak := atomic.LoadInt32(&c.peak)
|
||||
if current <= peak || atomic.CompareAndSwapInt32(&c.peak, peak, current) {
|
||||
break
|
||||
}
|
||||
}
|
||||
select {
|
||||
case c.started <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
select {
|
||||
case <-c.release:
|
||||
return &promptkit.GenerateResponse{Content: "ok"}, nil
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandlerPostRunsSuccessWithExplicitProfileID(t *testing.T) {
|
||||
start := time.Now().UTC()
|
||||
end := start.Add(2 * time.Second)
|
||||
@@ -90,9 +119,11 @@ func TestHandlerPostRunsSuccessWithExplicitProfileID(t *testing.T) {
|
||||
PromptHash: "phash",
|
||||
RenderedPromptHash: "rhash",
|
||||
SelectedProfileID: "exec-default",
|
||||
SelectedBackendID: "local",
|
||||
ModelName: "m1",
|
||||
Endpoint: "http://llm/v1",
|
||||
EffectiveModelParams: promptkit.ExecutionTarget{
|
||||
BackendID: "local",
|
||||
Endpoint: "http://llm/v1",
|
||||
Model: "m1",
|
||||
Temperature: 0.2,
|
||||
@@ -151,6 +182,9 @@ func TestHandlerPostRunsSuccessWithExplicitProfileID(t *testing.T) {
|
||||
if metadata["selected_profile_id"] != "exec-default" {
|
||||
t.Fatalf("unexpected metadata.selected_profile_id: %#v", metadata["selected_profile_id"])
|
||||
}
|
||||
if metadata["selected_backend_id"] != "local" {
|
||||
t.Fatalf("unexpected metadata.selected_backend_id: %#v", metadata["selected_backend_id"])
|
||||
}
|
||||
if metadata["model_name"] != "m1" || metadata["endpoint"] != "http://llm/v1" {
|
||||
t.Fatalf("unexpected model metadata: name=%#v endpoint=%#v", metadata["model_name"], metadata["endpoint"])
|
||||
}
|
||||
@@ -162,6 +196,9 @@ func TestHandlerPostRunsSuccessWithExplicitProfileID(t *testing.T) {
|
||||
t.Fatalf("unexpected cache usage metadata: %#v", usage)
|
||||
}
|
||||
modelParams := metadata["model_params"].(map[string]any)
|
||||
if modelParams["backend_id"] != "local" {
|
||||
t.Fatalf("expected model_params.backend_id=local, got %#v", modelParams["backend_id"])
|
||||
}
|
||||
if modelParams["api_key_env"] != envName {
|
||||
t.Fatalf("expected model_params.api_key_env=%q, got %#v", envName, modelParams["api_key_env"])
|
||||
}
|
||||
@@ -664,6 +701,79 @@ output:
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandlerSharesBackendCapacityAcrossConcurrentRequests(t *testing.T) {
|
||||
promptDir := t.TempDir()
|
||||
profileDir := t.TempDir()
|
||||
if err := os.WriteFile(filepath.Join(promptDir, "prompt.yaml"), []byte(`id: p
|
||||
version: "1"
|
||||
default_profile: limited
|
||||
messages:
|
||||
- role: user
|
||||
content: "hi"
|
||||
output:
|
||||
format: text
|
||||
validation_mode: none
|
||||
`), 0o644); err != nil {
|
||||
t.Fatalf("write prompt fixture: %v", err)
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join(profileDir, "profile.yaml"), []byte(`id: limited
|
||||
backend: limited
|
||||
model: test
|
||||
`), 0o644); err != nil {
|
||||
t.Fatalf("write profile fixture: %v", err)
|
||||
}
|
||||
queueCapacity := 0
|
||||
client := &blockingLLMClient{started: make(chan struct{}, 1), release: make(chan struct{})}
|
||||
engine, err := promptkit.NewEngine(
|
||||
promptkit.Config{PromptDir: promptDir, ProfileDir: profileDir},
|
||||
promptkit.WithBackend(promptkit.Backend{
|
||||
ID: "limited",
|
||||
Endpoint: "http://127.0.0.1:1/v1",
|
||||
ConcurrencyLimit: 1,
|
||||
QueueCapacity: &queueCapacity,
|
||||
}),
|
||||
promptkit.WithLLMClient(client),
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("new engine: %v", err)
|
||||
}
|
||||
h := NewHandler(engine)
|
||||
first := httptest.NewRecorder()
|
||||
firstDone := make(chan struct{})
|
||||
go func() {
|
||||
defer close(firstDone)
|
||||
h.ServeHTTP(first, httptest.NewRequest(http.MethodPost, "/v1/runs", bytes.NewBufferString(`{"prompt_id":"p"}`)))
|
||||
}()
|
||||
select {
|
||||
case <-client.started:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("first request did not reach generation")
|
||||
}
|
||||
|
||||
second := httptest.NewRecorder()
|
||||
h.ServeHTTP(second, httptest.NewRequest(http.MethodPost, "/v1/runs", bytes.NewBufferString(`{"prompt_id":"p"}`)))
|
||||
assertHTTPErrorCode(t, second, http.StatusServiceUnavailable, "capacity_exceeded")
|
||||
if second.Header().Get("Retry-After") != "" {
|
||||
t.Fatalf("expected no Retry-After header, got %q", second.Header().Get("Retry-After"))
|
||||
}
|
||||
if strings.Contains(second.Body.String(), "limited") {
|
||||
t.Fatalf("capacity response leaked backend details: %s", second.Body.String())
|
||||
}
|
||||
|
||||
close(client.release)
|
||||
select {
|
||||
case <-firstDone:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("first request did not complete")
|
||||
}
|
||||
if first.Code != http.StatusOK {
|
||||
t.Fatalf("expected first request to succeed, got %d body=%s", first.Code, first.Body.String())
|
||||
}
|
||||
if atomic.LoadInt32(&client.peak) != 1 {
|
||||
t.Fatalf("expected peak generation concurrency of one, got %d", atomic.LoadInt32(&client.peak))
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandlerRejectsOverlongSessionIDAndNonStringReasoningEffort(t *testing.T) {
|
||||
engine := newHandlerEngine(t)
|
||||
for _, body := range []string{
|
||||
@@ -1014,6 +1124,7 @@ func TestHandlerPublicErrorMapping(t *testing.T) {
|
||||
{name: "file too large", err: ErrFileTooLarge, status: http.StatusRequestEntityTooLarge, code: "artifact_too_large", message: "file input artifact is too large"},
|
||||
{name: "artifact", err: wrap(promptkit.ErrArtifactLoad, fmt.Errorf("read failed")), status: http.StatusBadRequest, code: "artifact_read_failed", message: "failed to read input artifact", avoidCause: "read failed"},
|
||||
{name: "prompt render", err: wrap(promptkit.ErrPromptRender, fmt.Errorf("render failed")), status: http.StatusBadRequest, code: "prompt_render_failed", message: "failed to render prompt", avoidCause: "render failed"},
|
||||
{name: "capacity", err: &promptkit.CapacityError{BackendID: "private-backend"}, status: http.StatusServiceUnavailable, code: "capacity_exceeded", message: "model backend capacity is exhausted", avoidCause: "private-backend"},
|
||||
{name: "llm", err: wrap(promptkit.ErrLLMGenerate, fmt.Errorf("llm failed")), status: http.StatusBadGateway, code: "llm_failed", message: "model generation request failed", avoidCause: "llm failed"},
|
||||
{name: "validation runtime", err: wrap(promptkit.ErrValidation, fmt.Errorf("validator broke")), status: http.StatusInternalServerError, code: "validation_runtime_failed", message: "validation runtime failed", avoidCause: "validator broke"},
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user