746 lines
29 KiB
Go
746 lines
29 KiB
Go
package openai
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"go.uber.org/zap"
|
|
|
|
edgeservice "iop/apps/edge/internal/service"
|
|
)
|
|
|
|
func (s *Server) handleResponses(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost {
|
|
writeError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
|
|
return
|
|
}
|
|
defer r.Body.Close()
|
|
|
|
// The raw body is preserved for the provider tunnel passthrough path, which
|
|
// forwards the caller's own Responses payload (model rewritten to the served
|
|
// target) without strict normalization.
|
|
rawBody, err := io.ReadAll(r.Body)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", "failed to read request body")
|
|
return
|
|
}
|
|
|
|
// The route decision runs before strict normalization: only the
|
|
// routing-relevant envelope fields are decoded leniently so provider
|
|
// passthrough can preserve Codex/Responses unknown fields (max_output_tokens,
|
|
// tools, store, ...). Strict field validation stays on the normalized path.
|
|
env, err := decodeResponsesEnvelope(rawBody)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
|
|
return
|
|
}
|
|
dispatch, ok := s.resolveRouteDispatch(env.Model)
|
|
if !ok {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", "model is required")
|
|
return
|
|
}
|
|
|
|
if dispatch.ProviderPool {
|
|
// Provider-pool path: the one-shot SubmitProviderPool selects a candidate
|
|
// and dispatches either tunnel passthrough (OpenAI-compatible) or
|
|
// normalized RunEvent (Ollama/CLI/native) based on the candidate's
|
|
// executionPath. Raw body is preserved for the tunnel branch; the
|
|
// normalized branch performs strict decode + prompt build below.
|
|
s.handleResponsesProviderPool(r, rawBody, env, dispatch, w)
|
|
return
|
|
}
|
|
|
|
// Non-provider routes: tunnel or normalized. Tunnel routes relay the raw
|
|
// body verbatim; normalized routes require strict decode + prompt build.
|
|
if routeUsesProviderTunnel(dispatch) {
|
|
runMeta, workspace, err := parseOpenAIMetadata(env.Metadata)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
|
|
return
|
|
}
|
|
// Overwrite (not merge-if-absent): the authenticated caller identity must
|
|
// win over any caller-supplied metadata.iop_principal_* spoof attempt.
|
|
for k, v := range principalMetadata(r.Context()) {
|
|
runMeta[k] = v
|
|
}
|
|
// Provider routes always relay pure passthrough. Caller metadata is
|
|
// arbitrary context and never selects the route or response shape
|
|
// (SDD S01/S04); there is no caller-facing response mode selector.
|
|
if err := validateWorkspaceForRoute(dispatch, workspace); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
|
|
return
|
|
}
|
|
estimate := estimateInputTokens(string(rawBody), runMeta, nil, nil)
|
|
contextClass := classifyContext(estimate, s.longContextThreshold())
|
|
s.tunnelResponsesPassthrough(w, r, env, dispatch, runMeta, rawBody, estimate, contextClass)
|
|
return
|
|
}
|
|
|
|
// Normalized RunEvent path: strict decode, stream/background rejection,
|
|
// prompt build, and SubmitRun.
|
|
var req responsesRequest
|
|
if err := decodeResponsesRequest(json.NewDecoder(bytes.NewReader(rawBody)), &req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
|
|
return
|
|
}
|
|
|
|
if req.Stream {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", "streaming is not supported for /v1/responses")
|
|
return
|
|
}
|
|
if req.Background {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", "background is not supported for /v1/responses")
|
|
return
|
|
}
|
|
|
|
inputStr, err := parseResponsesInput(req.Input)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
|
|
return
|
|
}
|
|
|
|
prompt := buildResponsesPrompt(req.Instructions, inputStr)
|
|
outputPolicy := s.resolveOutputPolicy(prompt)
|
|
if instruction := strictOutputContractInstruction(outputPolicy); instruction != "" {
|
|
prompt = instruction + "\n" + prompt
|
|
}
|
|
|
|
runMeta, workspace, err := parseOpenAIMetadata(req.Metadata)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
|
|
return
|
|
}
|
|
// Overwrite (not merge-if-absent): the authenticated caller identity must
|
|
// win over any caller-supplied metadata.iop_principal_* spoof attempt.
|
|
for k, v := range principalMetadata(r.Context()) {
|
|
runMeta[k] = v
|
|
}
|
|
|
|
if err := validateWorkspaceForRoute(dispatch, workspace); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
|
|
return
|
|
}
|
|
var defaultThinkingTokenBudget int
|
|
if catalogEntry := s.findProviderPoolEntry(req.Model); catalogEntry != nil {
|
|
applyModelCatalogGenerationPolicyToResponses(&req, *catalogEntry)
|
|
if catalogEntry.DefaultThinkingTokenBudget > 0 {
|
|
defaultThinkingTokenBudget = catalogEntry.DefaultThinkingTokenBudget
|
|
}
|
|
}
|
|
|
|
runMeta["openai_model"] = req.Model
|
|
runMeta["openai_stream"] = fmt.Sprintf("%t", req.Stream)
|
|
runMeta["strict_output"] = fmt.Sprintf("%t", outputPolicy.Strict)
|
|
input := map[string]any{"prompt": prompt}
|
|
if defaultThinkingTokenBudget > 0 {
|
|
input["think"] = true
|
|
input["thinking_token_budget"] = defaultThinkingTokenBudget
|
|
} else if outputPolicy.Strict {
|
|
input["think"] = false
|
|
}
|
|
if options := req.providerOptions(); len(options) > 0 {
|
|
input["options"] = options
|
|
}
|
|
|
|
s.logger.Info("openai responses input",
|
|
zap.String("model", req.Model),
|
|
zap.String("target", dispatch.Target),
|
|
zap.String("adapter", dispatch.Adapter),
|
|
zap.Bool("strict_output", outputPolicy.Strict),
|
|
zap.String("xml_completion_tool", outputPolicy.XMLCompletionTool),
|
|
zap.Bool("contract_instruction", outputPolicy.ContractInstruction),
|
|
zap.Int("prompt_len", len(prompt)),
|
|
)
|
|
|
|
estimate := estimateInputTokens(prompt, runMeta, nil, nil)
|
|
contextClass := classifyContext(estimate, s.longContextThreshold())
|
|
runMeta["estimated_input_tokens"] = strconv.Itoa(estimate)
|
|
runMeta["context_class"] = contextClass
|
|
|
|
handle, err := s.service.SubmitRun(r.Context(), edgeservice.SubmitRunRequest{
|
|
NodeRef: dispatch.NodeRef,
|
|
ModelGroupKey: strings.TrimSpace(req.Model),
|
|
Adapter: dispatch.Adapter,
|
|
Target: dispatch.Target,
|
|
SessionID: dispatch.SessionID,
|
|
Workspace: workspace,
|
|
Prompt: prompt,
|
|
Input: input,
|
|
TimeoutSec: dispatch.TimeoutSec,
|
|
MaxQueue: dispatch.MaxQueue,
|
|
QueueTimeoutMS: dispatch.QueueTimeoutMS,
|
|
Metadata: runMeta,
|
|
EstimatedInputTokens: estimate,
|
|
ContextClass: contextClass,
|
|
ProviderPool: dispatch.ProviderPool,
|
|
})
|
|
if err != nil {
|
|
writeError(w, http.StatusBadGateway, "node_dispatch_error", err.Error())
|
|
return
|
|
}
|
|
defer handle.Close()
|
|
|
|
s.logger.Info("openai responses dispatch",
|
|
zap.String("run_id", handle.Dispatch().RunID),
|
|
zap.String("node_id", handle.Dispatch().NodeID),
|
|
zap.String("provider_id", handle.Dispatch().ProviderID),
|
|
zap.String("provider_type", handle.Dispatch().ProviderType),
|
|
zap.String("execution_path", handle.Dispatch().ExecutionPath),
|
|
zap.String("model_group", handle.Dispatch().ModelGroupKey),
|
|
zap.String("adapter", handle.Dispatch().Adapter),
|
|
zap.String("target", handle.Dispatch().Target),
|
|
zap.Int("estimated_input_tokens", handle.Dispatch().EstimatedInputTokens),
|
|
zap.String("context_class", handle.Dispatch().ContextClass),
|
|
zap.String("queue_reason", handle.Dispatch().QueueReason),
|
|
)
|
|
|
|
s.completeResponse(w, r, req, handle, outputPolicy)
|
|
}
|
|
|
|
// handleResponsesProviderPool dispatches a provider-pool /v1/responses request
|
|
// through the one-shot SubmitProviderPool surface. After a single queue
|
|
// admission, the selected candidate's executionPath determines whether the
|
|
// response is delivered via tunnel passthrough (OpenAI-compatible providers)
|
|
// or the normalized RunEvent path (Ollama/CLI/native). Provider auth is
|
|
// validated via providerTunnelAuthHeaders for tunnel-path only (SDD D01).
|
|
// Strict-output policy only applies to normalized dispatch, not to raw tunnel
|
|
// passthrough (SDD D02).
|
|
//
|
|
// Selection-first flow: the provider is selected by SubmitProviderPool before
|
|
// any path-specific decode, validation, or stream gating. The tunnel branch
|
|
// only rewrites the model field and relays the raw body (SDD S04). The
|
|
// normalized branch performs strict decode + stream/background rejection +
|
|
// prompt build, but only after the candidate is known. This prevents
|
|
// tunnel-preferring unknown or streaming requests from bypassing the selected
|
|
// provider's executionPath rule (SDD S02/S03).
|
|
func (s *Server) handleResponsesProviderPool(
|
|
r *http.Request,
|
|
rawBody []byte,
|
|
env responsesEnvelope,
|
|
dispatch routeDispatch,
|
|
w http.ResponseWriter,
|
|
) {
|
|
// Build a populated base Run for the normalized branch and a Tunnel for
|
|
// the tunnel branch. Both are sent to SubmitProviderPool so the one-shot
|
|
// admission selects the provider before any path-specific behavior.
|
|
runMeta, workspace, err := parseOpenAIMetadata(env.Metadata)
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
|
|
return
|
|
}
|
|
for k, v := range principalMetadata(r.Context()) {
|
|
runMeta[k] = v
|
|
}
|
|
if err := validateWorkspaceForRoute(dispatch, workspace); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
|
|
return
|
|
}
|
|
|
|
// Compute estimate and context class from the raw body so both tunnel and
|
|
// normalized branches carry the same observability / long-context
|
|
// information the direct tunnel path carries (SDD S02/S03).
|
|
estimate := estimateInputTokens(string(rawBody), runMeta, nil, nil)
|
|
contextClass := classifyContext(estimate, s.longContextThreshold())
|
|
runMeta["openai_model"] = env.Model
|
|
runMeta["openai_stream"] = strconv.FormatBool(env.Stream)
|
|
runMeta["estimated_input_tokens"] = strconv.Itoa(estimate)
|
|
runMeta["context_class"] = contextClass
|
|
|
|
// Base Run carries the model-group key that SubmitProviderPool requires
|
|
// for candidate resolution. Metadata, estimate, and context class are
|
|
// preserved by PrepareRun on the normalized branch; the tunnel branch
|
|
// inherits them from the base so tunnel dispatch keeps observability.
|
|
baseRun := edgeservice.SubmitRunRequest{
|
|
NodeRef: dispatch.NodeRef,
|
|
ModelGroupKey: strings.TrimSpace(env.Model),
|
|
SessionID: dispatch.SessionID,
|
|
Workspace: workspace,
|
|
Metadata: runMeta,
|
|
EstimatedInputTokens: estimate,
|
|
ContextClass: contextClass,
|
|
ProviderPool: true,
|
|
}
|
|
|
|
baseTunnel := edgeservice.SubmitProviderTunnelRequest{
|
|
Metadata: runMeta,
|
|
EstimatedInputTokens: estimate,
|
|
ContextClass: contextClass,
|
|
ModelGroupKey: strings.TrimSpace(env.Model),
|
|
SessionID: dispatch.SessionID,
|
|
Method: http.MethodPost,
|
|
Path: "/v1/responses",
|
|
Stream: env.Stream,
|
|
TimeoutSec: dispatch.TimeoutSec,
|
|
MaxQueue: dispatch.MaxQueue,
|
|
QueueTimeoutMS: dispatch.QueueTimeoutMS,
|
|
ProviderPool: true,
|
|
}
|
|
|
|
poolReq := edgeservice.ProviderPoolDispatchRequest{
|
|
Run: baseRun,
|
|
Tunnel: baseTunnel,
|
|
}
|
|
|
|
// Pre-dispatch provider auth header injection. Runs inside SubmitProviderPool
|
|
// BEFORE buildProviderTunnelRequest and the Node Send step, so the auth
|
|
// header lands on the actual wire request. On failure the slot is released
|
|
// and no tunnel request is sent (SDD S03).
|
|
poolReq.PrepareTunnel = func(tunnelReq edgeservice.SubmitProviderTunnelRequest) (edgeservice.SubmitProviderTunnelRequest, error) {
|
|
headers, err := s.providerTunnelAuthHeaders(r)
|
|
if err != nil {
|
|
return tunnelReq, err
|
|
}
|
|
tunnelReq.Headers = headers
|
|
return tunnelReq, nil
|
|
}
|
|
|
|
// Tunnel branch: rewrite only the model field, preserve all other fields
|
|
// (tools, max_output_tokens, custom fields). Stream/background gating is
|
|
// skipped: the provider itself enforces those constraints (SDD S04).
|
|
poolReq.Tunnel.BuildBody = func(target string) ([]byte, error) {
|
|
return rewriteResponsesModel(rawBody, target)
|
|
}
|
|
|
|
// Prepared values captured from the PrepareRun hook on the normalized path.
|
|
// PrepareRun runs strict decode + prompt build + output policy; the
|
|
// normalized branch must relay these exact values to completeResponse so
|
|
// strict-output XML wrapping survives the provider-pool path.
|
|
var (
|
|
preparedResponsesReq responsesRequest
|
|
preparedOutputPolicy strictOutputPolicy
|
|
prepareRunSucceeded bool
|
|
)
|
|
|
|
// Normalized branch: strict decode, stream/background rejection, prompt
|
|
// build, and generation policy. Only runs AFTER the candidate is selected
|
|
// by SubmitProviderPool (i.e. only when executionPath is normalized).
|
|
// Validation errors are wrapped with errProviderRequestValidation so
|
|
// SubmitProviderPool can relay them as HTTP 400.
|
|
poolReq.PrepareRun = func(runReq edgeservice.SubmitRunRequest) (edgeservice.SubmitRunRequest, error) {
|
|
req := responsesRequest{Model: env.Model}
|
|
if err := decodeResponsesRequest(json.NewDecoder(bytes.NewReader(rawBody)), &req); err != nil {
|
|
return edgeservice.SubmitRunRequest{}, fmt.Errorf("%w: invalid /v1/responses request format: %s", errProviderRequestValidation, err.Error())
|
|
}
|
|
if req.Stream {
|
|
return edgeservice.SubmitRunRequest{}, fmt.Errorf("%w: streaming is not supported for /v1/responses", errProviderRequestValidation)
|
|
}
|
|
if req.Background {
|
|
return edgeservice.SubmitRunRequest{}, fmt.Errorf("%w: background is not supported for /v1/responses", errProviderRequestValidation)
|
|
}
|
|
|
|
inputStr, parseInputErr := parseResponsesInput(req.Input)
|
|
if parseInputErr != nil {
|
|
return edgeservice.SubmitRunRequest{}, fmt.Errorf("%w: %s", errProviderRequestValidation, parseInputErr.Error())
|
|
}
|
|
|
|
prompt := buildResponsesPrompt(req.Instructions, inputStr)
|
|
outputPolicy := s.resolveOutputPolicy(prompt)
|
|
if instruction := strictOutputContractInstruction(outputPolicy); instruction != "" {
|
|
prompt = instruction + "\n" + prompt
|
|
}
|
|
|
|
preparedMeta, _, metaErr := parseOpenAIMetadata(req.Metadata)
|
|
if metaErr != nil {
|
|
return edgeservice.SubmitRunRequest{}, fmt.Errorf("%w: %s", errProviderRequestValidation, metaErr.Error())
|
|
}
|
|
for k, v := range principalMetadata(r.Context()) {
|
|
preparedMeta[k] = v
|
|
}
|
|
|
|
if workspaceErr := validateWorkspaceForRoute(dispatch, workspace); workspaceErr != nil {
|
|
return edgeservice.SubmitRunRequest{}, fmt.Errorf("%w: %s", errProviderRequestValidation, workspaceErr.Error())
|
|
}
|
|
|
|
var defaultThinkingTokenBudget int
|
|
if catalogEntry := s.findProviderPoolEntry(req.Model); catalogEntry != nil {
|
|
applyModelCatalogGenerationPolicyToResponses(&req, *catalogEntry)
|
|
if catalogEntry.DefaultThinkingTokenBudget > 0 {
|
|
defaultThinkingTokenBudget = catalogEntry.DefaultThinkingTokenBudget
|
|
}
|
|
}
|
|
|
|
preparedMeta["openai_model"] = req.Model
|
|
preparedMeta["openai_stream"] = fmt.Sprintf("%t", req.Stream)
|
|
preparedMeta["strict_output"] = fmt.Sprintf("%t", outputPolicy.Strict)
|
|
|
|
input := map[string]any{"prompt": prompt}
|
|
if defaultThinkingTokenBudget > 0 {
|
|
input["think"] = true
|
|
input["thinking_token_budget"] = defaultThinkingTokenBudget
|
|
} else if outputPolicy.Strict {
|
|
input["think"] = false
|
|
}
|
|
if options := req.providerOptions(); len(options) > 0 {
|
|
input["options"] = options
|
|
}
|
|
|
|
estimate := estimateInputTokens(prompt, preparedMeta, nil, nil)
|
|
contextClass := classifyContext(estimate, s.longContextThreshold())
|
|
preparedMeta["estimated_input_tokens"] = strconv.Itoa(estimate)
|
|
preparedMeta["context_class"] = contextClass
|
|
|
|
runReq.Prompt = prompt
|
|
runReq.Input = input
|
|
runReq.Metadata = preparedMeta
|
|
runReq.EstimatedInputTokens = estimate
|
|
runReq.ContextClass = contextClass
|
|
runReq.TimeoutSec = dispatch.TimeoutSec
|
|
runReq.MaxQueue = dispatch.MaxQueue
|
|
runReq.QueueTimeoutMS = dispatch.QueueTimeoutMS
|
|
|
|
preparedResponsesReq = req
|
|
preparedOutputPolicy = outputPolicy
|
|
prepareRunSucceeded = true
|
|
return runReq, nil
|
|
}
|
|
|
|
result, err := s.service.SubmitProviderPool(r.Context(), poolReq)
|
|
if err != nil {
|
|
metricLabels := s.usageLabelsFor(r.Context(), strings.TrimSpace(env.Model), usageEndpointResponses, responseModePassthrough)
|
|
emitUsageMetrics(metricLabels, usageStatusForError(err), usageObservation{})
|
|
// Provider auth failure is a client request error (400), not a
|
|
// backend dispatch error. The auth check runs inside
|
|
// SubmitProviderPool (PrepareTunnel) before any tunnel request is sent.
|
|
// Normalized validation failures (e.g. strict decode, stream) are
|
|
// also client errors: map to 400.
|
|
if errors.Is(err, errProviderAuthRequired) {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", "provider auth token is required")
|
|
return
|
|
}
|
|
if isValidationError(err) {
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
|
|
return
|
|
}
|
|
writeError(w, http.StatusBadGateway, "node_dispatch_error", err.Error())
|
|
return
|
|
}
|
|
|
|
s.logger.Info("openai responses provider-pool dispatch",
|
|
zap.String("run_id", result.DispatchInfo.RunID),
|
|
zap.String("node_id", result.DispatchInfo.NodeID),
|
|
zap.String("provider_id", result.DispatchInfo.ProviderID),
|
|
zap.String("provider_type", result.DispatchInfo.ProviderType),
|
|
zap.String("execution_path", result.DispatchInfo.ExecutionPath),
|
|
zap.String("model_group", result.DispatchInfo.ModelGroupKey),
|
|
zap.String("adapter", result.DispatchInfo.Adapter),
|
|
zap.String("target", result.DispatchInfo.Target),
|
|
zap.String("path", string(result.Path)),
|
|
zap.Int("estimated_input_tokens", result.DispatchInfo.EstimatedInputTokens),
|
|
zap.String("context_class", result.DispatchInfo.ContextClass),
|
|
zap.String("queue_reason", result.DispatchInfo.QueueReason),
|
|
)
|
|
|
|
switch result.Path {
|
|
case edgeservice.ProviderPoolPathTunnel:
|
|
// Tunnel path: provider auth was already validated and injected via
|
|
// PrepareTunnel before dispatch; on failure SubmitProviderPool returns
|
|
// an error and no tunnel handle exists. Provider bytes are relayed as
|
|
// pure passthrough; caller metadata never selects a sideband surface.
|
|
metricLabels := s.usageLabelsFor(r.Context(), strings.TrimSpace(env.Model), usageEndpointResponses, responseModePassthrough)
|
|
s.writeProviderTunnelResponse(w, r, result.Tunnel, env.Stream, env.Model, metricLabels)
|
|
|
|
case edgeservice.ProviderPoolPathNormalized:
|
|
// Normalized path: no auth required, collect from RunEvent stream.
|
|
handle := result.Run
|
|
if handle == nil {
|
|
writeError(w, http.StatusInternalServerError, "run_error", "provider-pool selection returned normalized path but no run result")
|
|
return
|
|
}
|
|
// PrepareRun is only invoked on the normalized path. A normalized
|
|
// result without a successful PrepareRun is an internal mismatch and
|
|
// must surface as run_error (SDD D02).
|
|
if !prepareRunSucceeded {
|
|
writeError(w, http.StatusInternalServerError, "run_error", "provider-pool normalized result received without a successful PrepareRun")
|
|
return
|
|
}
|
|
// Relay the prepared request and output policy from PrepareRun to
|
|
// completeResponse so strict-output XML wrapping survives the
|
|
// normalized provider-pool path.
|
|
s.completeResponse(w, r, preparedResponsesReq, handle, preparedOutputPolicy)
|
|
}
|
|
}
|
|
|
|
func decodeResponsesRequest(dec *json.Decoder, req *responsesRequest) error {
|
|
var raw map[string]json.RawMessage
|
|
if err := dec.Decode(&raw); err != nil {
|
|
return fmt.Errorf("invalid JSON request")
|
|
}
|
|
for key := range raw {
|
|
switch key {
|
|
case "model", "input", "instructions", "stream", "background", "metadata", "max_output_tokens", "temperature", "top_p":
|
|
default:
|
|
return fmt.Errorf("%s is not supported for /v1/responses", key)
|
|
}
|
|
}
|
|
normalized, err := json.Marshal(raw)
|
|
if err != nil {
|
|
return fmt.Errorf("invalid JSON request")
|
|
}
|
|
if err := json.Unmarshal(normalized, req); err != nil {
|
|
return fmt.Errorf("invalid /v1/responses request format")
|
|
}
|
|
if req.MaxOutputTokens != nil && *req.MaxOutputTokens <= 0 {
|
|
return fmt.Errorf("max_output_tokens must be greater than zero")
|
|
}
|
|
if req.Temperature != nil && (*req.Temperature < 0 || *req.Temperature > 2) {
|
|
return fmt.Errorf("temperature must be between 0 and 2")
|
|
}
|
|
if req.TopP != nil && (*req.TopP < 0 || *req.TopP > 1) {
|
|
return fmt.Errorf("top_p must be between 0 and 1")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// decodeResponsesEnvelope leniently extracts only the routing-relevant fields
|
|
// (model, metadata, stream, background) from a /v1/responses request body. It
|
|
// does not reject unknown fields so the provider tunnel passthrough can forward
|
|
// Codex/Responses payloads verbatim; strict field validation stays on the
|
|
// normalized non-provider path.
|
|
func decodeResponsesEnvelope(rawBody []byte) (responsesEnvelope, error) {
|
|
var env responsesEnvelope
|
|
if err := json.Unmarshal(rawBody, &env); err != nil {
|
|
return responsesEnvelope{}, fmt.Errorf("invalid JSON request")
|
|
}
|
|
return env, nil
|
|
}
|
|
|
|
// tunnelResponsesPassthrough serves a /v1/responses request over the raw
|
|
// provider tunnel (SDD S04): the caller's original body is forwarded with only
|
|
// the model field rewritten to the served target, provider auth is injected
|
|
// from the configured request header, and provider status/headers/body bytes
|
|
// are relayed to the caller unmodified. Streaming is honored when the caller
|
|
// requested it. No IOP sideband fields, model-echo rewrite, or output-token
|
|
// normalization are applied.
|
|
func (s *Server) tunnelResponsesPassthrough(w http.ResponseWriter, r *http.Request, env responsesEnvelope, dispatch routeDispatch, runMeta map[string]string, rawBody []byte, estimate int, contextClass string) {
|
|
providerAuthHeaders, err := s.providerTunnelAuthHeaders(r)
|
|
if err != nil {
|
|
// Missing required provider auth is rejected before dispatch; the raw
|
|
// token is never echoed into the error surface.
|
|
writeError(w, http.StatusBadRequest, "invalid_request_error", "provider auth token is required")
|
|
return
|
|
}
|
|
|
|
metadata := make(map[string]string, len(runMeta)+5)
|
|
for k, v := range runMeta {
|
|
metadata[k] = v
|
|
}
|
|
metadata["openai_model"] = env.Model
|
|
metadata["openai_stream"] = strconv.FormatBool(env.Stream)
|
|
metadata["estimated_input_tokens"] = strconv.Itoa(estimate)
|
|
metadata["context_class"] = contextClass
|
|
|
|
tunnelReq := edgeservice.SubmitProviderTunnelRequest{
|
|
NodeRef: dispatch.NodeRef,
|
|
ModelGroupKey: strings.TrimSpace(env.Model),
|
|
Adapter: dispatch.Adapter,
|
|
Target: dispatch.Target,
|
|
SessionID: dispatch.SessionID,
|
|
Method: http.MethodPost,
|
|
Path: "/v1/responses",
|
|
Headers: providerAuthHeaders,
|
|
BuildBody: func(target string) ([]byte, error) {
|
|
return rewriteResponsesModel(rawBody, target)
|
|
},
|
|
Stream: env.Stream,
|
|
TimeoutSec: dispatch.TimeoutSec,
|
|
MaxQueue: dispatch.MaxQueue,
|
|
QueueTimeoutMS: dispatch.QueueTimeoutMS,
|
|
Metadata: metadata,
|
|
EstimatedInputTokens: estimate,
|
|
ContextClass: contextClass,
|
|
ProviderPool: dispatch.ProviderPool,
|
|
}
|
|
|
|
metricLabels := s.usageLabelsFor(r.Context(), strings.TrimSpace(env.Model), usageEndpointResponses, responseModePassthrough)
|
|
handle, err := s.service.SubmitProviderTunnel(r.Context(), tunnelReq)
|
|
if err != nil {
|
|
emitUsageMetrics(metricLabels, usageStatusForError(err), usageObservation{})
|
|
writeError(w, http.StatusBadGateway, "node_dispatch_error", err.Error())
|
|
return
|
|
}
|
|
defer handle.Close()
|
|
|
|
s.logger.Info("openai responses passthrough dispatch",
|
|
zap.String("run_id", handle.Dispatch().RunID),
|
|
zap.String("node_id", handle.Dispatch().NodeID),
|
|
zap.String("provider_id", handle.Dispatch().ProviderID),
|
|
zap.String("provider_type", handle.Dispatch().ProviderType),
|
|
zap.String("execution_path", handle.Dispatch().ExecutionPath),
|
|
zap.String("model_group", handle.Dispatch().ModelGroupKey),
|
|
zap.String("adapter", handle.Dispatch().Adapter),
|
|
zap.String("target", handle.Dispatch().Target),
|
|
zap.Bool("stream", env.Stream),
|
|
zap.Int("estimated_input_tokens", handle.Dispatch().EstimatedInputTokens),
|
|
zap.String("context_class", handle.Dispatch().ContextClass),
|
|
zap.String("queue_reason", handle.Dispatch().QueueReason),
|
|
)
|
|
|
|
// requestModel is left empty so the shared tunnel writer relays provider
|
|
// bytes verbatim without rewriting the provider-echoed model back to a
|
|
// caller alias: Responses passthrough prefers provider-original bytes.
|
|
// metricLabels carries endpoint=responses, response_mode=passthrough, and
|
|
// the request model alias so success/error usage is attributed correctly.
|
|
s.writeProviderTunnelResponse(w, r, handle, env.Stream, "", metricLabels)
|
|
}
|
|
|
|
// rewriteResponsesModel replaces only the model field of the caller's original
|
|
// /v1/responses request JSON so the provider receives its served model name.
|
|
// Every other field (input, instructions, tools, max_output_tokens, and any
|
|
// Codex/Responses-specific field) is forwarded without IOP rewriting. An empty
|
|
// target leaves the body untouched; invalid JSON is rejected.
|
|
func rewriteResponsesModel(rawBody []byte, target string) ([]byte, error) {
|
|
if strings.TrimSpace(target) == "" {
|
|
return rawBody, nil
|
|
}
|
|
var raw map[string]json.RawMessage
|
|
if err := json.Unmarshal(rawBody, &raw); err != nil {
|
|
return nil, fmt.Errorf("invalid JSON request")
|
|
}
|
|
modelJSON, err := json.Marshal(target)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
raw["model"] = modelJSON
|
|
return json.Marshal(raw)
|
|
}
|
|
|
|
func (s *Server) completeResponse(w http.ResponseWriter, r *http.Request, req responsesRequest, handle edgeservice.RunResult, outputPolicy strictOutputPolicy) {
|
|
metricLabels := s.usageLabelsFor(r.Context(), strings.TrimSpace(req.Model), usageEndpointResponses, "normalized")
|
|
text, reasoning, _, _, usage, _, err := collectRunResult(r.Context(), handle.Stream(), handle.WaitTimeout())
|
|
if err != nil {
|
|
s.cancelRunOnHTTPGiveUp(handle.Dispatch(), err)
|
|
emitUsageMetrics(metricLabels, usageStatusForError(err), usageObservation{})
|
|
writeError(w, httpStatusForRunError(err), "run_error", err.Error())
|
|
return
|
|
}
|
|
text, reasoning, normalized := normalizeCompletionOutput(outputPolicy, text, reasoning, false)
|
|
s.logger.Info("openai responses output",
|
|
zap.String("run_id", handle.Dispatch().RunID),
|
|
zap.Bool("strict_output", outputPolicy.Strict),
|
|
zap.String("xml_completion_tool", outputPolicy.XMLCompletionTool),
|
|
zap.Bool("normalized", normalized),
|
|
zap.Int("content_len", len(text)),
|
|
zap.Int("reasoning_len", len(reasoning)),
|
|
)
|
|
|
|
var u openAIUsage
|
|
if usage != nil {
|
|
u = *usage
|
|
}
|
|
|
|
emitUsageMetrics(metricLabels, usageStatusSuccess, usageObservationFromOpenAIUsage(usage, len(reasoning)))
|
|
|
|
writeJSON(w, http.StatusOK, responsesResponse{
|
|
ID: "resp-" + handle.Dispatch().RunID,
|
|
Object: "response",
|
|
CreatedAt: time.Now().Unix(),
|
|
Model: responseModel(req.Model, handle.Dispatch().Target),
|
|
OutputText: text,
|
|
Output: []responsesOutputItem{{
|
|
Type: "message",
|
|
Role: "assistant",
|
|
Content: []responsesContentItem{{
|
|
Type: "output_text",
|
|
Text: text,
|
|
}},
|
|
}},
|
|
Usage: u,
|
|
})
|
|
}
|
|
|
|
func parseResponsesInput(raw json.RawMessage) (string, error) {
|
|
if len(raw) == 0 {
|
|
return "", fmt.Errorf("input is required")
|
|
}
|
|
var s string
|
|
if err := json.Unmarshal(raw, &s); err != nil {
|
|
return "", fmt.Errorf("input must be a string")
|
|
}
|
|
if strings.TrimSpace(s) == "" {
|
|
return "", fmt.Errorf("input is required")
|
|
}
|
|
return s, nil
|
|
}
|
|
|
|
func buildResponsesPrompt(instructions, input string) string {
|
|
instructions = strings.TrimSpace(instructions)
|
|
if instructions == "" {
|
|
return input
|
|
}
|
|
return instructions + "\n\n" + input
|
|
}
|
|
|
|
func parseOpenAIMetadata(raw json.RawMessage) (map[string]string, string, error) {
|
|
if len(raw) == 0 || string(raw) == "null" {
|
|
return make(map[string]string), "", nil
|
|
}
|
|
|
|
var rawMap map[string]json.RawMessage
|
|
if err := json.Unmarshal(raw, &rawMap); err != nil {
|
|
return nil, "", fmt.Errorf("metadata must be an object")
|
|
}
|
|
|
|
if len(rawMap) > 16 {
|
|
return nil, "", fmt.Errorf("metadata must contain at most 16 keys")
|
|
}
|
|
|
|
flat := make(map[string]string, len(rawMap))
|
|
var workspace string
|
|
for key, rawValue := range rawMap {
|
|
if len(key) > 64 {
|
|
return nil, "", fmt.Errorf("metadata key %q exceeds 64 characters", key)
|
|
}
|
|
if key == "source" {
|
|
return nil, "", fmt.Errorf("metadata.source is not supported")
|
|
}
|
|
if key == "workspace" {
|
|
workspaceValue, err := metadataStringValue(key, rawValue)
|
|
if err != nil {
|
|
return nil, "", err
|
|
}
|
|
workspace = strings.TrimSpace(workspaceValue)
|
|
continue
|
|
}
|
|
value, err := metadataStringValue(key, rawValue)
|
|
if err != nil {
|
|
return nil, "", err
|
|
}
|
|
flat[key] = value
|
|
}
|
|
|
|
return flat, workspace, nil
|
|
}
|
|
|
|
// errProviderRequestValidation is a sentinel for client request validation
|
|
// failures that occur inside the normalized provider-pool PrepareRun hook
|
|
// (strict decode, stream/background rejection, input parsing). These errors
|
|
// are reported as HTTP 400 invalid_request_error, distinct from backend
|
|
// dispatch errors that return 502.
|
|
var errProviderRequestValidation = errors.New("provider_request_validation")
|
|
|
|
// isValidationError reports whether err originated from a PrepareRun validation
|
|
// failure in the normalized provider-pool path.
|
|
func isValidationError(err error) bool {
|
|
return errors.Is(err, errProviderRequestValidation)
|
|
}
|
|
|
|
// metadataStringValue extracts a single string value from a json.RawMessage
|
|
// for the metadata field. It rejects non-string values and enforces the
|
|
// 512-character length cap.
|
|
func metadataStringValue(key string, raw json.RawMessage) (string, error) {
|
|
var value string
|
|
if err := json.Unmarshal(raw, &value); err != nil {
|
|
return "", fmt.Errorf("metadata.%s must be a string", key)
|
|
}
|
|
if len(value) > 512 {
|
|
return "", fmt.Errorf("metadata.%s exceeds 512 characters", key)
|
|
}
|
|
return value, nil
|
|
}
|