From fc2e3efc35a0c6fd9a746b9ec067db0675d94283 Mon Sep 17 00:00:00 2001 From: hcastc00 Date: Tue, 6 Oct 2026 11:52:19 +0200 Subject: [PATCH] feat: implement SystemOne endpoint and related functionality --- .../adapters/http/handlers/models_handler.go | 6 + .../http/handlers/systemone_handler.go | 45 ++++++ .../adapters/http/mapper/systemone.go | 91 +++++++++++ apps/gateway/internal/adapters/http/router.go | 4 + .../internal/adapters/http/router_test.go | 7 + .../adapters/providers/openai/client.go | 7 +- .../adapters/providers/typesafe/adapter.go | 43 +++++ .../internal/application/services/pii.go | 41 ++++- .../internal/application/services/proxy.go | 26 ++- .../application/services/systemone.go | 35 ++++ .../application/services/systemone_test.go | 153 ++++++++++++++++++ .../enclave/network/network_nitro_test.go | 1 + .../internal/enclave/profile/manifest.go | 1 + apps/gateway/main.go | 4 +- pkg/domain/generation.go | 1 + pkg/domain/model.go | 8 + 16 files changed, 463 insertions(+), 10 deletions(-) create mode 100644 apps/gateway/internal/adapters/http/handlers/systemone_handler.go create mode 100644 apps/gateway/internal/adapters/http/mapper/systemone.go create mode 100644 apps/gateway/internal/adapters/providers/typesafe/adapter.go create mode 100644 apps/gateway/internal/application/services/systemone.go create mode 100644 apps/gateway/internal/application/services/systemone_test.go diff --git a/apps/gateway/internal/adapters/http/handlers/models_handler.go b/apps/gateway/internal/adapters/http/handlers/models_handler.go index d56f781..f3d44aa 100644 --- a/apps/gateway/internal/adapters/http/handlers/models_handler.go +++ b/apps/gateway/internal/adapters/http/handlers/models_handler.go @@ -95,6 +95,9 @@ func toRouterData(r domain.RouterEntry) dto.ModelData { } func modelFeatures(m domain.PublicModel) []string { + if m.SupportsEndpoint(domain.EndpointSystemOne) { + return []string{"structured-outputs"} + } features := make([]string, 0, 5) if m.SupportsChatCompletionsStream { features = append(features, "streaming") @@ -120,6 +123,9 @@ func modelFeatures(m domain.PublicModel) []string { func modelEndpoints(m domain.PublicModel) []string { endpoints := make([]string, 0, 1) + if m.SupportsEndpoint(domain.EndpointSystemOne) { + return []string{"systemone"} + } if m.SupportsChatCompletions { endpoints = append(endpoints, "chat/completions") } diff --git a/apps/gateway/internal/adapters/http/handlers/systemone_handler.go b/apps/gateway/internal/adapters/http/handlers/systemone_handler.go new file mode 100644 index 0000000..91792e4 --- /dev/null +++ b/apps/gateway/internal/adapters/http/handlers/systemone_handler.go @@ -0,0 +1,45 @@ +package handlers + +import ( + "io" + "net/http" + + "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/http/mapper" + "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/application/ports" + "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/application/services" + "github.com/dappnode/dappnode-nexus-gateway/pkg/domain" +) + +type SystemOneHandler struct { + service *services.GenerateService + logger ports.Logger +} + +func NewSystemOneHandler(service *services.GenerateService, logger ports.Logger) *SystemOneHandler { + return &SystemOneHandler{service: service, logger: logger} +} + +func (h *SystemOneHandler) Handle(w http.ResponseWriter, r *http.Request) { + token, err := ExtractBearerToken(r) + if err != nil { + WriteErrorWithLog(w, r, h.logger, err) + return + } + body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxRequestBodySize)) + if err != nil { + WriteErrorWithLog(w, r, h.logger, domain.ErrInvalidField("request body unreadable or too large")) + return + } + req, err := mapper.SystemOneRequestToDomain(body) + if err != nil { + WriteErrorWithLog(w, r, h.logger, err) + return + } + call, err := h.service.Proxy(r.Context(), domain.EndpointSystemOne, body, req, token) + if err != nil { + WriteErrorWithLog(w, r, h.logger, err) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write(call.Body) +} diff --git a/apps/gateway/internal/adapters/http/mapper/systemone.go b/apps/gateway/internal/adapters/http/mapper/systemone.go new file mode 100644 index 0000000..ce48fff --- /dev/null +++ b/apps/gateway/internal/adapters/http/mapper/systemone.go @@ -0,0 +1,91 @@ +package mapper + +import ( + "bytes" + "encoding/json" + "strings" + + "github.com/dappnode/dappnode-nexus-gateway/pkg/domain" +) + +// SystemOneRequestToDomain reads routing fields while preserving the native body. +func SystemOneRequestToDomain(raw []byte) (domain.GenerateRequest, error) { + var body struct { + Model string `json:"model"` + State json.RawMessage `json:"state"` + Questions map[string]struct { + Type string `json:"type"` + Instructions json.RawMessage `json:"instructions"` + Criteria json.RawMessage `json:"criteria"` + } `json:"questions"` + Stream bool `json:"stream"` + } + invalid := func(message string) (domain.GenerateRequest, error) { + err := domain.ErrInvalidField(message) + err.HTTPStatus = 422 + return domain.GenerateRequest{}, err + } + if json.Unmarshal(raw, &body) != nil { + return invalid("invalid systemone request") + } + if strings.TrimSpace(body.Model) == "" { + return invalid("model is required") + } + if body.Stream { + return invalid("systemone does not support streaming") + } + if !structuredValue(body.State) { + return invalid("state must be a string, object, or array") + } + if len(body.Questions) == 0 { + return invalid("questions must be a non-empty object") + } + for _, q := range body.Questions { + if !structuredValue(q.Instructions) { + return invalid("question instructions must be a string, object, or array") + } + switch q.Type { + case "noul": + if len(q.Criteria) == 0 { + continue + } + var criteria map[string]json.RawMessage + if json.Unmarshal(q.Criteria, &criteria) != nil || criteria == nil { + return invalid("noul criteria must be an object") + } + for key, value := range criteria { + if (key != "true" && key != "false") || !structuredValue(value) { + return invalid("noul criteria must describe true or false") + } + } + case "choice": + var criteria map[string]json.RawMessage + if json.Unmarshal(q.Criteria, &criteria) != nil || len(criteria) < 1 || len(criteria) > 255 { + return invalid("choice criteria must contain 1 to 255 options") + } + for _, value := range criteria { + if !bytes.Equal(bytes.TrimSpace(value), []byte("null")) && !structuredValue(value) { + return invalid("invalid choice criterion") + } + } + case "score": + var criteria []json.RawMessage + if json.Unmarshal(q.Criteria, &criteria) != nil || len(criteria) < 2 || len(criteria) > 10 { + return invalid("score criteria must contain 2 to 10 levels") + } + for _, value := range criteria { + if !structuredValue(value) { + return invalid("invalid score criterion") + } + } + default: + return invalid("question type must be noul, choice, or score") + } + } + return domain.GenerateRequest{PublicModelID: body.Model, RequestedModelID: body.Model}, nil +} + +func structuredValue(raw json.RawMessage) bool { + raw = bytes.TrimSpace(raw) + return len(raw) > 0 && (raw[0] == '"' || raw[0] == '{' || raw[0] == '[') +} diff --git a/apps/gateway/internal/adapters/http/router.go b/apps/gateway/internal/adapters/http/router.go index 25bd04a..29bb22e 100644 --- a/apps/gateway/internal/adapters/http/router.go +++ b/apps/gateway/internal/adapters/http/router.go @@ -13,6 +13,7 @@ func NewRouter( health *handlers.HealthHandler, models *handlers.ModelsHandler, chatCompletions *handlers.ChatCompletionsHandler, + systemOne *handlers.SystemOneHandler, confidentialChat *handlers.ConfidentialChatCompletionsHandler, tinfoil *handlers.TinfoilHandler, attestation *handlers.AttestationHandler, @@ -23,6 +24,9 @@ func NewRouter( mux.HandleFunc("GET /healthz", health.Handle) mux.HandleFunc("GET /v1/models", models.Handle) mux.HandleFunc("POST /v1/chat/completions", chatCompletions.Handle) + if systemOne != nil { + mux.HandleFunc("POST /v1/systemone", systemOne.Handle) + } if confidentialChat != nil { mux.HandleFunc("POST /v1/confidential/chat/completions", confidentialChat.Handle) } diff --git a/apps/gateway/internal/adapters/http/router_test.go b/apps/gateway/internal/adapters/http/router_test.go index 08eddc4..ff37283 100644 --- a/apps/gateway/internal/adapters/http/router_test.go +++ b/apps/gateway/internal/adapters/http/router_test.go @@ -22,6 +22,7 @@ func TestNewRouter_PreservesLegacyChatRouteWithoutConfidentialHandler(t *testing handlers.NewHealthHandler(), handlers.NewModelsHandler(nil), handlers.NewChatCompletionsHandler(nil, logger), + handlers.NewSystemOneHandler(nil, logger), nil, nil, nil, @@ -35,6 +36,12 @@ func TestNewRouter_PreservesLegacyChatRouteWithoutConfidentialHandler(t *testing t.Fatalf("legacy status = %d, want existing 401 auth behavior", legacyResponse.Code) } + evaluationResponse := httptest.NewRecorder() + router.ServeHTTP(evaluationResponse, httptest.NewRequest(http.MethodPost, "/v1/systemone", strings.NewReader(`{}`))) + if evaluationResponse.Code != http.StatusUnauthorized { + t.Fatalf("systemone status = %d", evaluationResponse.Code) + } + confidentialRequest := httptest.NewRequest(http.MethodPost, "/v1/confidential/chat/completions", strings.NewReader(`{"model":"test"}`)) confidentialResponse := httptest.NewRecorder() router.ServeHTTP(confidentialResponse, confidentialRequest) diff --git a/apps/gateway/internal/adapters/providers/openai/client.go b/apps/gateway/internal/adapters/providers/openai/client.go index 7e9cf85..32145ee 100644 --- a/apps/gateway/internal/adapters/providers/openai/client.go +++ b/apps/gateway/internal/adapters/providers/openai/client.go @@ -45,6 +45,11 @@ func NewClient(timeout time.Duration) *Client { } func (c *Client) Do(ctx context.Context, baseURL, apiKey string, body map[string]any) ([]byte, error) { + return c.DoEndpoint(ctx, baseURL, "/chat/completions", apiKey, body) +} + +// DoEndpoint uses the same bounded, enclave-aware transport for JSON APIs. +func (c *Client) DoEndpoint(ctx context.Context, baseURL, endpoint, apiKey string, body map[string]any) ([]byte, error) { ctx, cancel := context.WithTimeout(ctx, c.responseTimeout) defer cancel() @@ -53,7 +58,7 @@ func (c *Client) Do(ctx context.Context, baseURL, apiKey string, body map[string return nil, fmt.Errorf("failed to marshal request: %w", err) } - req, err := http.NewRequestWithContext(ctx, http.MethodPost, baseURL+"/chat/completions", bytes.NewReader(jsonBody)) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(baseURL, "/")+endpoint, bytes.NewReader(jsonBody)) if err != nil { return nil, fmt.Errorf("failed to create request: %w", err) } diff --git a/apps/gateway/internal/adapters/providers/typesafe/adapter.go b/apps/gateway/internal/adapters/providers/typesafe/adapter.go new file mode 100644 index 0000000..c6ca9ef --- /dev/null +++ b/apps/gateway/internal/adapters/providers/typesafe/adapter.go @@ -0,0 +1,43 @@ +package typesafe + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "os" + "time" + + "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/providers/openai" + "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/application/ports" + "github.com/dappnode/dappnode-nexus-gateway/pkg/domain" +) + +type Adapter struct{ client *openai.Client } + +func NewAdapter(timeout time.Duration) *Adapter { return &Adapter{client: openai.NewClient(timeout)} } + +func (a *Adapter) Complete(ctx context.Context, raw []byte, model domain.PublicModel) ([]byte, *domain.TinfoilTransportProof, error) { + apiKey := os.Getenv(model.ProviderConfig.APIKeySecretRef) + if apiKey == "" { + return nil, nil, domain.ErrInternal("provider API key is not configured") + } + var body map[string]any + dec := json.NewDecoder(bytes.NewReader(raw)) + dec.UseNumber() + if err := dec.Decode(&body); err != nil || body == nil { + return nil, nil, domain.ErrInvalidField("invalid JSON body") + } + body["model"] = model.UpstreamModelName + delete(body, "provider_options") + out, err := a.client.DoEndpoint(ctx, model.ProviderConfig.BaseURL, "/systemone", apiKey, body) + var upstream *openai.ProviderHTTPError + if errors.As(err, &upstream) && (upstream.StatusCode == 422 || upstream.StatusCode == 429 || upstream.StatusCode == 529) { + return nil, nil, domain.ErrProviderError(upstream.StatusCode, "TypeSafe rejected the request").WithMeta("upstream_status", upstream.StatusCode) + } + return out, nil, openai.MapProviderErrorWithCompatibilityContext(err, model, body) +} + +func (a *Adapter) Stream(context.Context, []byte, domain.PublicModel) (ports.ProviderStream, error) { + return ports.ProviderStream{}, domain.ErrUnsupportedFeature("streaming for systemone") +} diff --git a/apps/gateway/internal/application/services/pii.go b/apps/gateway/internal/application/services/pii.go index 8f51eb8..48069dd 100644 --- a/apps/gateway/internal/application/services/pii.go +++ b/apps/gateway/internal/application/services/pii.go @@ -25,7 +25,7 @@ import ( // other static fields are not scanned. It returns the body unchanged and no // mapping when the key doesn't mask, the filter is off, or nothing was found. // On filter failure it fails closed, unless configured to fail open. -func (s *GenerateService) maskBody(ctx context.Context, raw []byte, piiMode string) ([]byte, *domain.PIIMapping, error) { +func (s *GenerateService) maskBody(ctx context.Context, raw []byte, piiMode, endpoint string) ([]byte, *domain.PIIMapping, error) { mode, ok := domain.NormalizeAPIKeyPIIMode(piiMode) if !ok || mode == domain.APIKeyPIIModeOff || s.pii == nil || !s.pii.Enabled() { return raw, nil, nil @@ -43,7 +43,7 @@ func (s *GenerateService) maskBody(ctx context.Context, raw []byte, piiMode stri entityCounts: make(map[string]int), surfaceCounts: make(map[string]int), } - if err := masker.maskBody(body); err != nil { + if err := masker.maskBody(body, endpoint); err != nil { s.logger.Warn("pii filter error", "error", err, "fail_open", s.piiFailOpen) if s.piiFailOpen { return raw, nil, nil @@ -66,7 +66,30 @@ func (s *GenerateService) maskBody(ctx context.Context, raw []byte, piiMode stri return masked, masker.mapping, nil } -func (m *requestPIIMasker) maskBody(body map[string]any) error { +func (m *requestPIIMasker) maskBody(body map[string]any, endpoint string) error { + if endpoint == domain.EndpointSystemOne { + masked, _, err := m.maskAnyWithChanged("state", body["state"]) + if err != nil { + return err + } + body["state"] = masked + questions, _ := body["questions"].(map[string]any) + for _, item := range questions { + question, _ := item.(map[string]any) + for _, field := range []string{"instructions", "criteria"} { + value, exists := question[field] + if !exists { + continue + } + masked, _, err := m.maskAnyWithChanged(field, value) + if err != nil { + return err + } + question[field] = masked + } + } + return nil + } if user, ok := body["user"].(string); ok { masked, err := m.maskText("user", user) if err != nil { @@ -346,6 +369,18 @@ func (u *unmasker) body(data []byte) []byte { if err != nil { return data } + if answers, ok := obj["answers"].(map[string]any); ok { + for _, item := range answers { + answer, _ := item.(map[string]any) + legend, _ := answer["legend"].(map[string]any) + for key, value := range legend { + if text, ok := value.(string); ok { + legend[key] = domain.Unmask(text, u.mapping) + u.track("legend", legend[key].(string)) + } + } + } + } choices, _ := obj["choices"].([]any) for _, item := range choices { choice, _ := item.(map[string]any) diff --git a/apps/gateway/internal/application/services/proxy.go b/apps/gateway/internal/application/services/proxy.go index eaea5f4..9551486 100644 --- a/apps/gateway/internal/application/services/proxy.go +++ b/apps/gateway/internal/application/services/proxy.go @@ -32,7 +32,7 @@ type ProxyCall struct { Body []byte } -// Proxy runs a chat-completions request. The gateway authenticates, routes, +// Proxy runs an inference request. The gateway authenticates, routes, // validates, masks PII for keys that ask for it, reserves credit, and meters // the usage it reads from the response; it never rebuilds the request or the // response. @@ -45,7 +45,13 @@ func (s *GenerateService) Proxy(ctx context.Context, endpoint string, raw []byte s.recordTerminalOutcome(metrics.OutcomeError, endpoint, req, nil, time.Since(start).Milliseconds()) return nil, err } - model, execReq, err := s.resolveModel(ctx, req) + var model domain.PublicModel + execReq := req + if endpoint == domain.EndpointSystemOne { + model, err = s.catalog.GetPublicModel(ctx, req.PublicModelID) + } else { + model, execReq, err = s.resolveModel(ctx, req) + } if err != nil { s.recordTerminalOutcome(metrics.OutcomeError, endpoint, req, nil, time.Since(start).Milliseconds()) return nil, err @@ -55,7 +61,7 @@ func (s *GenerateService) Proxy(ctx context.Context, endpoint string, raw []byte s.recordFailure(ctx, nil, &authCtx, endpoint, &execReq, &model, err, nil, time.Since(start).Milliseconds()) return nil, err } - masked, mapping, err := s.maskBody(ctx, raw, authCtx.APIKey.PIIMode) + masked, mapping, err := s.maskBody(ctx, raw, authCtx.APIKey.PIIMode, endpoint) if err != nil { s.recordTerminalOutcome(metrics.OutcomeError, endpoint, execReq, &model, time.Since(start).Milliseconds()) s.recordFailure(ctx, nil, &authCtx, endpoint, &execReq, &model, err, nil, time.Since(start).Milliseconds()) @@ -76,6 +82,9 @@ func (s *GenerateService) Proxy(ctx context.Context, endpoint string, raw []byte } attempt := func(target domain.PublicModel) (*ProxyCall, error) { + if err := s.validateRequest(endpoint, execReq, target); err != nil { + return nil, err + } provider, err := s.registry.GetProvider(target.ProviderConfig.ProviderName) if err != nil { return nil, domain.ErrProviderUnavailable(target.ProviderConfig.ProviderName) @@ -87,7 +96,13 @@ func (s *GenerateService) Proxy(ctx context.Context, endpoint string, raw []byte if err != nil { return nil, err } - rewritten, o, err := observeJSON(body, target.PublicModelID, unmask(target)) + var rewritten []byte + var o outcome + if endpoint == domain.EndpointSystemOne { + rewritten, o, err = observeSystemOne(body, target.PublicModelID, unmask(target)) + } else { + rewritten, o, err = observeJSON(body, target.PublicModelID, unmask(target)) + } if err != nil { return nil, err } @@ -116,7 +131,8 @@ func (s *GenerateService) Proxy(ctx context.Context, endpoint string, raw []byte call, err := attempt(model) executed := model - if err != nil && shouldTryFallback(ctx, err, model.Fallback) { + if err != nil && shouldTryFallback(ctx, err, model.Fallback) && + withProviderTarget(model, *model.Fallback).SupportsEndpoint(endpoint) { executed = withProviderTarget(model, *model.Fallback) s.logFallback(requestID, model, executed, sanitizeErrorWithPIIMapping(err, mapping)) call, err = attempt(executed) diff --git a/apps/gateway/internal/application/services/systemone.go b/apps/gateway/internal/application/services/systemone.go new file mode 100644 index 0000000..a96c422 --- /dev/null +++ b/apps/gateway/internal/application/services/systemone.go @@ -0,0 +1,35 @@ +package services + +import ( + "encoding/json" + "math" + "net/http" + + "github.com/dappnode/dappnode-nexus-gateway/pkg/domain" +) + +func observeSystemOne(body []byte, public string, unmask *unmasker) ([]byte, outcome, error) { + var response struct { + Model *string `json:"model"` + Answers map[string]json.RawMessage `json:"answers"` + Usage *struct { + Input *int64 `json:"input_tokens"` + Output *int64 `json:"output_tokens"` + } `json:"usage"` + } + if json.Unmarshal(body, &response) != nil || response.Model == nil || len(response.Answers) == 0 || + response.Usage == nil || response.Usage.Input == nil || response.Usage.Output == nil { + return nil, outcome{}, domain.ErrProviderError(http.StatusBadGateway, "invalid systemone response or missing usage") + } + input, output := *response.Usage.Input, *response.Usage.Output + if input < 0 || output < 0 || input > math.MaxInt64-output { + return nil, outcome{}, domain.ErrProviderError(http.StatusBadGateway, "invalid systemone usage") + } + if unmask != nil { + body = unmask.body(body) + unmask.report(false) + } + return rewriteModel(body, &chunk{Model: response.Model}, public), outcome{ + usage: &domain.Usage{PromptTokens: input, CompletionTokens: output, TotalTokens: input + output}, end: "complete", + }, nil +} diff --git a/apps/gateway/internal/application/services/systemone_test.go b/apps/gateway/internal/application/services/systemone_test.go new file mode 100644 index 0000000..c0a1879 --- /dev/null +++ b/apps/gateway/internal/application/services/systemone_test.go @@ -0,0 +1,153 @@ +package services + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/http/mapper" + "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/providers/typesafe" + "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/application/ports" + "github.com/dappnode/dappnode-nexus-gateway/pkg/domain" +) + +const evaluationRequest = `{"model":"public-jev","state":{"email":"jane@example.com","id":9007199254740993},"questions":{"urgent":{"type":"noul","instructions":"Urgent?","criteria":{"true":"Yes","false":"No"}},"team":{"type":"choice","instructions":{"question":"Which team?"},"criteria":{"sales":null,"support":["Help"]}},"rating":{"type":"score","instructions":["Rate it"],"criteria":["jane@example.com","Other"]}}}` +const evaluationResponse = `{"model":"jev-1.13.0","answers":{"urgent":{"type":"noul","noul":0.95},"team":{"type":"choice","choice":"support","probabilities":{"support":1,"sales":0},"confidence":1},"rating":{"type":"score","score":0,"legend":{"0":"[EMAIL_1]","1":"Other"},"probabilities":{"0":1,"1":0},"confidence":1}},"usage":{"input_tokens":296,"output_tokens":20}}` + +func TestSystemOne(t *testing.T) { + t.Setenv("TEST_TYPESAFE_KEY", "upstream-secret") + status := http.StatusOK + response := evaluationResponse + upstreamCalls := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + upstreamCalls++ + if r.URL.Path != "/v1/systemone" || r.Method != "POST" || r.Header.Get("Authorization") != "Bearer upstream-secret" { + t.Errorf("unexpected upstream request: %s %s", r.Method, r.URL.Path) + } + raw, _ := io.ReadAll(r.Body) + if strings.Contains(string(raw), "jane@example.com") || !strings.Contains(string(raw), "9007199254740993") || !strings.Contains(string(raw), `"model":"jev-latest"`) { + t.Errorf("request masking/model/precision broken: %s", raw) + } + w.WriteHeader(status) + _, _ = w.Write([]byte(response)) + })) + defer server.Close() + model := domain.PublicModel{PublicModelID: "public-jev", ProviderModelID: "pm", UpstreamModelName: "jev-latest", ProviderConfig: domain.ProviderConfig{ProviderName: "typesafe", BaseURL: server.URL + "/v1/", APIKeySecretRef: "TEST_TYPESAFE_KEY"}, MaxContextWindow: 4096, MaxOutputTokens: 256} + meter := &resultMeter{} + auth := &stubAuthService{authCtx: domain.AuthContext{Account: domain.Account{ID: "acc"}, APIKey: domain.APIKey{ID: "key", PIIMode: domain.APIKeyPIIModeBalanced}}} + filter := &fakePIIFilter{enabled: true, byText: map[string][]domain.PIIEntity{"jane@example.com": {entity("EMAIL", "jane@example.com", "jane@example.com")}}} + catalog := &stubModelCatalog{model: model} + svc := NewGenerateService(auth, catalog, nil, &stubProviderRegistry{providers: map[string]ports.Provider{"typesafe": typesafe.NewAdapter(time.Second)}}, meter, filter, stubLogger{}) + req, err := mapper.SystemOneRequestToDomain([]byte(evaluationRequest)) + if err != nil { + t.Fatal(err) + } + run := func() (*ProxyCall, error) { + return svc.Proxy(context.Background(), domain.EndpointSystemOne, []byte(evaluationRequest), req, "nexus-key") + } + call, err := run() + if err != nil { + t.Fatal(err) + } + var got, want any + _ = json.Unmarshal(call.Body, &got) + _ = json.Unmarshal([]byte(strings.ReplaceAll(strings.ReplaceAll(evaluationResponse, "jev-1.13.0", "public-jev"), "[EMAIL_1]", "jane@example.com")), &want) + gotJSON, _ := json.Marshal(got) + wantJSON, _ := json.Marshal(want) + if string(gotJSON) != string(wantJSON) { + t.Fatalf("answers changed: %s", call.Body) + } + if meter.reserveCalls != 1 || meter.successCalls != 1 || meter.result.Usage.PromptTokens != 296 || meter.result.Usage.CompletionTokens != 20 || meter.result.Usage.TotalTokens != 316 { + t.Fatalf("incorrect billing: %+v", meter) + } + + for _, code := range []int{422, 429, 529, 401} { + status = code + response = `{"detail":"upstream failure"}` + _, err = run() + expected := code + if code == 401 { + expected = 502 + } + ge, ok := err.(*domain.GatewayError) + if !ok || ge.HTTPStatus != expected { + t.Fatalf("upstream %d mapped to %v", code, err) + } + } + if meter.failureCalls != 4 || meter.successCalls != 1 { + t.Fatal("upstream failures were charged") + } + status = http.StatusOK + for _, malformed := range []string{`null`, `{"model":"jev","answers":{"a":{}},"usage":{}}`, `{"model":"jev","answers":{"a":{}},"usage":{"input_tokens":-1,"output_tokens":2}}`} { + response = malformed + if _, err := run(); err == nil { + t.Fatalf("accepted unbillable response: %s", malformed) + } + } + response = evaluationResponse + before := upstreamCalls + auth.err = domain.ErrInvalidAPIKey("invalid") + if _, err := run(); err == nil { + t.Fatal("accepted invalid API key") + } + auth.err = nil + meter.reserveErr = domain.ErrInternal("no credit") + if _, err := run(); err == nil { + t.Fatal("called provider without reservation") + } + meter.reserveErr = nil + filter.err = fmt.Errorf("PII unavailable") + if _, err := run(); err == nil { + t.Fatal("PII failed open") + } + filter.err = nil + catalog.model.ProviderConfig.ProviderName = "openai" + if _, err := run(); err == nil { + t.Fatal("accepted chat model") + } + catalog.model = model + if _, err := svc.Proxy(context.Background(), domain.EndpointChatCompletions, []byte(evaluationRequest), req, "key"); err == nil { + t.Fatal("accepted typesafe on chat endpoint") + } + if upstreamCalls != before { + t.Fatal("rejected requests reached provider") + } + // A configured chat fallback must not receive structured evaluation requests. + fallback := &stubProvider{} + svc.registry = &stubProviderRegistry{providers: map[string]ports.Provider{"typesafe": typesafe.NewAdapter(time.Second), "openai": fallback}} + catalog.model.Fallback = &domain.ProviderTarget{ProviderConfig: domain.ProviderConfig{ProviderName: "openai"}} + status = 529 + if _, err := run(); err == nil || fallback.calls != 0 { + t.Fatal("incompatible fallback used") + } else if ge, ok := err.(*domain.GatewayError); !ok || ge.HTTPStatus != 529 { + t.Fatalf("incompatible fallback hid upstream status: %v", err) + } +} + +func TestSystemOneValidation(t *testing.T) { + for _, raw := range []string{ + `null`, `{}`, evaluationRequest + `{}`, strings.Replace(evaluationRequest, `"model":"public-jev"`, `"model":""`, 1), + strings.Replace(evaluationRequest, `"model":"public-jev"`, `"model":"public-jev","stream":true`, 1), + `{"model":"m","state":null,"questions":{}}`, `{"model":"m","state":42,"questions":{}}`, + `{"model":"m","state":"s","questions":{"q":{"type":"other","instructions":"i"}}}`, + `{"model":"m","state":[],"questions":{"q":{"type":"noul","instructions":true}}}`, + `{"model":"m","state":{},"questions":{"q":{"type":"score","instructions":"i","criteria":["one"]}}}`, + `{"model":"m","state":"s","questions":{"q":{"type":"choice","instructions":"i","criteria":{}}}}`, + } { + if _, err := mapper.SystemOneRequestToDomain([]byte(raw)); err == nil { + t.Fatalf("accepted %s", raw) + } + } + for _, state := range []string{`"text"`, `{}`, `[]`} { + raw := `{"model":"m","state":` + state + `,"questions":{"q":{"type":"noul","instructions":"i"}}}` + if _, err := mapper.SystemOneRequestToDomain([]byte(raw)); err != nil { + t.Fatal(err) + } + } +} diff --git a/apps/gateway/internal/enclave/network/network_nitro_test.go b/apps/gateway/internal/enclave/network/network_nitro_test.go index 282a1f5..46393a7 100644 --- a/apps/gateway/internal/enclave/network/network_nitro_test.go +++ b/apps/gateway/internal/enclave/network/network_nitro_test.go @@ -46,6 +46,7 @@ func TestProviderVsockPortUsesMeasuredAllowlist(t *testing.T) { {address: "tuf-repo-cdn.sigstore.dev:443", want: 8449}, {address: "kds-proxy.tinfoil.sh:443", want: 8450}, {address: "tdx-proxy.tinfoil.sh:443", want: 8451}, + {address: "api.typesafe.ai:443", want: 8452}, {address: "API.DEEPSEEK.COM.:443", want: 8443}, {address: "api.deepseek.com:80", wantErr: "not measured or allowed"}, {address: "example.com:443", wantErr: "not measured or allowed"}, diff --git a/apps/gateway/internal/enclave/profile/manifest.go b/apps/gateway/internal/enclave/profile/manifest.go index 034da51..dfc0be4 100644 --- a/apps/gateway/internal/enclave/profile/manifest.go +++ b/apps/gateway/internal/enclave/profile/manifest.go @@ -33,6 +33,7 @@ type ProviderEgressRoute struct { // production provider set. Keep this list in sync with the parent proxy units. func ProviderEgressRoutes() []ProviderEgressRoute { return []ProviderEgressRoute{ + {Name: "typesafe", Host: "api.typesafe.ai", Port: 443, VsockPort: 8452}, {Name: "deepseek", Host: "api.deepseek.com", Port: 443, VsockPort: 8443}, {Name: "novita", Host: "api.novita.ai", Port: 443, VsockPort: 8446}, {Name: "tinfoil-inference", Host: "inference.tinfoil.sh", Port: 443, VsockPort: 8447}, diff --git a/apps/gateway/main.go b/apps/gateway/main.go index f5fc99d..1e1bda4 100644 --- a/apps/gateway/main.go +++ b/apps/gateway/main.go @@ -21,6 +21,7 @@ import ( "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/providers/openai" "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/providers/registry" tinfoilprovider "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/providers/tinfoil" + "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/providers/typesafe" routerclient "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/router" "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/application/ports" "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/application/services" @@ -70,6 +71,7 @@ func main() { providerRegistry.Register("tinfoil", tinfoilprovider.NewAdapter(providerTimeout, zapLogger)) // Every other provider speaks the OpenAI wire format, so new ones (e.g. // novita, mistral) are added in the database only. + providerRegistry.Register("typesafe", typesafe.NewAdapter(providerTimeout)) providerRegistry.SetDefault(openai.NewAdapter(providerTimeout, zapLogger)) meteringClient := meteringadapter.NewClient(meteringURL, meteringToken, 5*time.Second) @@ -127,7 +129,7 @@ func main() { } } - router := gwhttp.NewRouter(healthHandler, modelsHandler, chatHandler, confidentialChatHandler, tinfoilHandler, attestationHandler, zapLogger) + router := gwhttp.NewRouter(healthHandler, modelsHandler, chatHandler, handlers.NewSystemOneHandler(generateSvc, zapLogger), confidentialChatHandler, tinfoilHandler, attestationHandler, zapLogger) metricsMux := http.NewServeMux() metricsMux.Handle("/metrics", metrics.Handler()) diff --git a/pkg/domain/generation.go b/pkg/domain/generation.go index c9146e0..9daabf6 100644 --- a/pkg/domain/generation.go +++ b/pkg/domain/generation.go @@ -3,6 +3,7 @@ package domain // Endpoint constants. const ( EndpointChatCompletions = "chat_completions" + EndpointSystemOne = "systemone" ) // GenerateRequest is what the gateway reads from a client's request: enough diff --git a/pkg/domain/model.go b/pkg/domain/model.go index a5c83b6..2b079b1 100644 --- a/pkg/domain/model.go +++ b/pkg/domain/model.go @@ -66,6 +66,11 @@ func ProofModeEnabled(mode string) bool { // SupportsEndpoint checks if the model supports the given endpoint. func (m PublicModel) SupportsEndpoint(endpoint string) bool { + // ponytail: the provider selects the wire protocol; add capability storage + // when one provider needs to serve multiple protocols. + if m.ProviderConfig.ProviderName == "typesafe" { + return endpoint == EndpointSystemOne + } switch endpoint { case EndpointChatCompletions: return m.SupportsChatCompletions @@ -76,6 +81,9 @@ func (m PublicModel) SupportsEndpoint(endpoint string) bool { // SupportsStreamForEndpoint checks if the model supports streaming for the given endpoint. func (m PublicModel) SupportsStreamForEndpoint(endpoint string) bool { + if m.ProviderConfig.ProviderName == "typesafe" { + return false + } switch endpoint { case EndpointChatCompletions: return m.SupportsChatCompletionsStream