package openai import ( "encoding/json" "fmt" "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() var req responsesRequest if err := decodeResponsesRequest(json.NewDecoder(r.Body), &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 } dispatch, ok := s.resolveRouteDispatch(req.Model) if !ok { writeError(w, http.StatusBadRequest, "invalid_request_error", "model is required") return } if err := validateWorkspaceForRoute(dispatch, workspace); err != nil { writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error()) return } if routeUsesProviderTunnel(dispatch) { writeError(w, http.StatusBadRequest, "invalid_request_error", "/v1/responses is not supported for OpenAI-compatible provider model groups until raw passthrough parity is implemented") 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("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) } 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 } 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 } 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 }