feat: implement SystemOne endpoint and related functionality (#7)

Author: hcastc00

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

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

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] == '[')
+}

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

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)

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

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

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)

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)