feat: add Responses API compatibility
This commit is contained in:
@@ -20,20 +20,43 @@ func (c OpenAIClient) Run(ctx context.Context, request Request) (Response, error
|
||||
if apiKey == "" {
|
||||
return Response{}, &ClientError{Code: "missing_credentials", Message: "openai api key is required", Retryable: false}
|
||||
}
|
||||
endpoint := openAIEndpoint(request.Kind)
|
||||
protocol := request.UpstreamProtocol
|
||||
if protocol == "" && request.Kind == "responses" {
|
||||
protocol = ProtocolOpenAIResponses
|
||||
}
|
||||
endpointKind := request.Kind
|
||||
if request.Kind == "responses" && protocol == ProtocolOpenAIChatCompletions {
|
||||
endpointKind = "chat.completions"
|
||||
}
|
||||
endpoint := openAIEndpoint(endpointKind)
|
||||
if endpoint == "" {
|
||||
return Response{}, &ClientError{Code: "unsupported_kind", Message: "unsupported openai request kind", Retryable: false}
|
||||
}
|
||||
body := cloneBody(request.Body)
|
||||
if request.Kind == "chat.completions" {
|
||||
if request.Kind == "responses" && protocol == ProtocolOpenAIChatCompletions {
|
||||
var convertErr error
|
||||
body, convertErr = ResponsesRequestToChat(request.Body, request.PreviousResponseTurns)
|
||||
if convertErr != nil {
|
||||
return Response{}, convertErr
|
||||
}
|
||||
}
|
||||
if endpointKind == "chat.completions" {
|
||||
body = NormalizeChatCompletionRequestBody(body)
|
||||
applyOpenAIChatReasoningParams(body, request.Candidate)
|
||||
} else if request.Kind == "responses" {
|
||||
delete(body, "messages")
|
||||
if request.UpstreamPreviousResponseID != "" {
|
||||
body["previous_response_id"] = request.UpstreamPreviousResponseID
|
||||
} else {
|
||||
delete(body, "previous_response_id")
|
||||
}
|
||||
}
|
||||
body["model"] = upstreamModelName(request.Candidate)
|
||||
stream := openAIEndpointSupportsStream(request.Kind) && (request.Stream || boolValue(body, "stream"))
|
||||
ensureOpenAIStreamUsage(body, request.Kind, stream)
|
||||
stream := openAIEndpointSupportsStream(endpointKind) && (request.Stream || boolValue(body, "stream"))
|
||||
ensureOpenAIStreamUsage(body, endpointKind, stream)
|
||||
raw, _ := json.Marshal(body)
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, joinURL(openAIBaseURL(request.Kind, request.Candidate), endpoint), bytes.NewReader(raw))
|
||||
upstreamEndpoint := joinURL(openAIBaseURL(endpointKind, request.Candidate), endpoint)
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, upstreamEndpoint, bytes.NewReader(raw))
|
||||
if err != nil {
|
||||
return Response{}, err
|
||||
}
|
||||
@@ -45,7 +68,38 @@ func (c OpenAIClient) Run(ctx context.Context, request Request) (Response, error
|
||||
return Response{}, &ClientError{Code: "network", Message: err.Error(), Retryable: true}
|
||||
}
|
||||
requestID := requestIDFromHTTPResponse(resp)
|
||||
result, err := decodeOpenAIResponse(resp, stream, request.StreamDelta)
|
||||
var result map[string]any
|
||||
upstreamResponseID := ""
|
||||
if request.Kind == "responses" && protocol == ProtocolOpenAIResponses && stream {
|
||||
result, upstreamResponseID, err = decodeNativeResponsesStream(resp, request.StreamDelta)
|
||||
} else {
|
||||
var streamDelta StreamDelta = request.StreamDelta
|
||||
var adapter *chatResponsesStreamAdapter
|
||||
if request.Kind == "responses" && protocol == ProtocolOpenAIChatCompletions && stream {
|
||||
adapter = newChatResponsesStreamAdapter(request.PublicResponseID, request.Model)
|
||||
streamDelta = func(event StreamDeltaEvent) error { return adapter.delta(event, request.StreamDelta) }
|
||||
}
|
||||
result, err = decodeOpenAIResponse(resp, stream, streamDelta)
|
||||
if err == nil && endpointKind == "chat.completions" {
|
||||
result = NormalizeChatCompletionResult(result)
|
||||
}
|
||||
if err == nil && request.Kind == "responses" && protocol == ProtocolOpenAIChatCompletions {
|
||||
chatResult := result
|
||||
upstreamResponseID = requestIDFromResult(chatResult)
|
||||
result = ChatResultToResponse(chatResult, request.PublicResponseID, request.Model, request.Body)
|
||||
if adapter != nil {
|
||||
err = adapter.done(result, request.StreamDelta)
|
||||
}
|
||||
if err == nil {
|
||||
return Response{
|
||||
Result: result, InternalResult: chatResult, RequestID: firstNonEmptyString(requestID, upstreamResponseID), Usage: usageFromOpenAI(chatResult),
|
||||
Progress: providerProgress(request), ResponseStartedAt: responseStartedAt, ResponseFinishedAt: time.Now(),
|
||||
UpstreamProtocol: protocol, UpstreamEndpoint: endpoint, UpstreamResponseID: upstreamResponseID,
|
||||
PublicResponseID: request.PublicResponseID, ResponseConverted: true,
|
||||
}, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
if err == nil && request.Kind == "chat.completions" {
|
||||
result = NormalizeChatCompletionResult(result)
|
||||
}
|
||||
@@ -56,6 +110,15 @@ func (c OpenAIClient) Run(ctx context.Context, request Request) (Response, error
|
||||
if requestID == "" {
|
||||
requestID = requestIDFromResult(result)
|
||||
}
|
||||
if request.Kind == "responses" && protocol == ProtocolOpenAIResponses {
|
||||
if upstreamResponseID == "" {
|
||||
upstreamResponseID = requestIDFromResult(result)
|
||||
}
|
||||
}
|
||||
publicResponseID := request.PublicResponseID
|
||||
if request.Kind == "responses" && protocol == ProtocolOpenAIResponses {
|
||||
publicResponseID = upstreamResponseID
|
||||
}
|
||||
return Response{
|
||||
Result: result,
|
||||
RequestID: requestID,
|
||||
@@ -64,6 +127,10 @@ func (c OpenAIClient) Run(ctx context.Context, request Request) (Response, error
|
||||
ResponseStartedAt: responseStartedAt,
|
||||
ResponseFinishedAt: responseFinishedAt,
|
||||
ResponseDurationMS: responseDurationMS(responseStartedAt, responseFinishedAt),
|
||||
UpstreamProtocol: protocol,
|
||||
UpstreamEndpoint: endpoint,
|
||||
UpstreamResponseID: upstreamResponseID,
|
||||
PublicResponseID: publicResponseID,
|
||||
}, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,720 @@
|
||||
package clients
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
ProtocolOpenAIChatCompletions = "openai_chat_completions"
|
||||
ProtocolOpenAIResponses = "openai_responses"
|
||||
ProtocolAnthropicMessages = "anthropic_messages"
|
||||
)
|
||||
|
||||
var supportedResponseFallbackParameters = map[string]struct{}{
|
||||
"model": {}, "input": {}, "messages": {}, "instructions": {}, "tools": {}, "tool_choice": {},
|
||||
"parallel_tool_calls": {}, "max_output_tokens": {}, "temperature": {}, "top_p": {},
|
||||
"presence_penalty": {}, "frequency_penalty": {}, "reasoning": {}, "text": {},
|
||||
"stream": {}, "store": {}, "previous_response_id": {}, "metadata": {}, "user": {},
|
||||
}
|
||||
|
||||
func ResponsesRequestToChat(body map[string]any, history []ResponseTurn) (map[string]any, error) {
|
||||
for key := range body {
|
||||
if _, ok := supportedResponseFallbackParameters[key]; !ok {
|
||||
return nil, unsupportedResponseParameter(key)
|
||||
}
|
||||
}
|
||||
messages := make([]any, 0)
|
||||
for _, turn := range history {
|
||||
priorInput := turn.Request["input"]
|
||||
if priorInput == nil {
|
||||
priorInput = turn.Request["messages"]
|
||||
}
|
||||
priorMessages, err := responseInputMessages(priorInput)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
messages = append(messages, priorMessages...)
|
||||
if assistant := chatAssistantMessage(turn.Internal, turn.Response); assistant != nil {
|
||||
messages = append(messages, assistant)
|
||||
}
|
||||
}
|
||||
if instructions := strings.TrimSpace(stringFromAny(body["instructions"])); instructions != "" {
|
||||
messages = append(messages, map[string]any{"role": "system", "content": instructions})
|
||||
}
|
||||
if rawMessages, ok := body["messages"].([]any); ok {
|
||||
messages = append(messages, rawMessages...)
|
||||
} else {
|
||||
currentMessages, err := responseInputMessages(body["input"])
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
messages = append(messages, currentMessages...)
|
||||
}
|
||||
if len(messages) == 0 {
|
||||
return nil, &ClientError{Code: "invalid_parameter", Message: "input is required", StatusCode: http.StatusBadRequest}
|
||||
}
|
||||
out := map[string]any{"messages": messages}
|
||||
for _, key := range []string{"temperature", "top_p", "presence_penalty", "frequency_penalty", "parallel_tool_calls", "stream", "user"} {
|
||||
if value, ok := body[key]; ok {
|
||||
out[key] = value
|
||||
}
|
||||
}
|
||||
if value, ok := body["max_output_tokens"]; ok {
|
||||
out["max_tokens"] = value
|
||||
}
|
||||
if rawReasoning, ok := body["reasoning"]; ok {
|
||||
reasoning, ok := rawReasoning.(map[string]any)
|
||||
if !ok {
|
||||
return nil, unsupportedResponseParameter("reasoning")
|
||||
}
|
||||
for key := range reasoning {
|
||||
if key != "effort" {
|
||||
return nil, unsupportedResponseParameter("reasoning." + key)
|
||||
}
|
||||
}
|
||||
if effort, ok := reasoning["effort"]; ok {
|
||||
out["reasoning_effort"] = effort
|
||||
}
|
||||
}
|
||||
if rawText, ok := body["text"]; ok {
|
||||
responseFormat, err := responseTextFormat(rawText)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if responseFormat != nil {
|
||||
out["response_format"] = responseFormat
|
||||
}
|
||||
}
|
||||
if rawTools, ok := body["tools"]; ok {
|
||||
tools, err := responseToolsToChat(rawTools)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out["tools"] = tools
|
||||
}
|
||||
if rawChoice, ok := body["tool_choice"]; ok {
|
||||
choice, err := responseToolChoiceToChat(rawChoice)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out["tool_choice"] = choice
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func responseInputMessages(value any) ([]any, error) {
|
||||
switch typed := value.(type) {
|
||||
case nil:
|
||||
return nil, nil
|
||||
case string:
|
||||
return []any{map[string]any{"role": "user", "content": typed}}, nil
|
||||
case []any:
|
||||
out := make([]any, 0, len(typed))
|
||||
for _, raw := range typed {
|
||||
switch item := raw.(type) {
|
||||
case string:
|
||||
out = append(out, map[string]any{"role": "user", "content": item})
|
||||
case map[string]any:
|
||||
switch stringFromAny(item["type"]) {
|
||||
case "function_call_output":
|
||||
callID := firstNonEmptyString(item["call_id"], item["id"])
|
||||
if callID == "" {
|
||||
return nil, &ClientError{Code: "invalid_parameter", Message: "function_call_output.call_id is required", StatusCode: http.StatusBadRequest}
|
||||
}
|
||||
out = append(out, map[string]any{"role": "tool", "tool_call_id": callID, "content": toolResultContent(item["output"])})
|
||||
case "message", "":
|
||||
role := firstNonEmptyString(item["role"], "user")
|
||||
out = append(out, map[string]any{"role": role, "content": responseContentToChat(item["content"])})
|
||||
default:
|
||||
return nil, unsupportedResponseParameter("input.type=" + stringFromAny(item["type"]))
|
||||
}
|
||||
default:
|
||||
return nil, unsupportedResponseParameter("input")
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
case map[string]any:
|
||||
return responseInputMessages([]any{typed})
|
||||
default:
|
||||
return nil, unsupportedResponseParameter("input")
|
||||
}
|
||||
}
|
||||
|
||||
func responseContentToChat(value any) any {
|
||||
items, ok := value.([]any)
|
||||
if !ok {
|
||||
return value
|
||||
}
|
||||
out := make([]any, 0, len(items))
|
||||
for _, raw := range items {
|
||||
item, ok := raw.(map[string]any)
|
||||
if !ok {
|
||||
out = append(out, raw)
|
||||
continue
|
||||
}
|
||||
switch stringFromAny(item["type"]) {
|
||||
case "input_text", "output_text", "text":
|
||||
out = append(out, map[string]any{"type": "text", "text": stringFromAny(item["text"])})
|
||||
case "input_image":
|
||||
url := firstNonEmptyString(item["image_url"], item["url"])
|
||||
out = append(out, map[string]any{"type": "image_url", "image_url": map[string]any{"url": url}})
|
||||
default:
|
||||
out = append(out, item)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func responseToolsToChat(value any) ([]any, error) {
|
||||
items, ok := value.([]any)
|
||||
if !ok {
|
||||
return nil, unsupportedResponseParameter("tools")
|
||||
}
|
||||
out := make([]any, 0, len(items))
|
||||
for _, raw := range items {
|
||||
tool, ok := raw.(map[string]any)
|
||||
if !ok || stringFromAny(tool["type"]) != "function" {
|
||||
return nil, &ClientError{Code: "unsupported_response_tool", Message: "Chat fallback only supports custom function tools", StatusCode: http.StatusBadRequest}
|
||||
}
|
||||
function := map[string]any{
|
||||
"name": tool["name"],
|
||||
"description": tool["description"],
|
||||
"parameters": tool["parameters"],
|
||||
}
|
||||
if strict, ok := tool["strict"]; ok {
|
||||
function["strict"] = strict
|
||||
}
|
||||
out = append(out, map[string]any{"type": "function", "function": function})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func responseToolChoiceToChat(value any) (any, error) {
|
||||
if text, ok := value.(string); ok {
|
||||
switch text {
|
||||
case "auto", "none", "required":
|
||||
return text, nil
|
||||
default:
|
||||
return nil, unsupportedResponseParameter("tool_choice")
|
||||
}
|
||||
}
|
||||
choice, ok := value.(map[string]any)
|
||||
if !ok || stringFromAny(choice["type"]) != "function" || stringFromAny(choice["name"]) == "" {
|
||||
return nil, unsupportedResponseParameter("tool_choice")
|
||||
}
|
||||
return map[string]any{"type": "function", "function": map[string]any{"name": choice["name"]}}, nil
|
||||
}
|
||||
|
||||
func responseTextFormat(value any) (map[string]any, error) {
|
||||
text, ok := value.(map[string]any)
|
||||
if !ok {
|
||||
return nil, unsupportedResponseParameter("text")
|
||||
}
|
||||
for key := range text {
|
||||
if key != "format" {
|
||||
return nil, unsupportedResponseParameter("text." + key)
|
||||
}
|
||||
}
|
||||
format, ok := text["format"].(map[string]any)
|
||||
if !ok || len(format) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
switch stringFromAny(format["type"]) {
|
||||
case "text":
|
||||
return map[string]any{"type": "text"}, nil
|
||||
case "json_object":
|
||||
return map[string]any{"type": "json_object"}, nil
|
||||
case "json_schema":
|
||||
return map[string]any{"type": "json_schema", "json_schema": map[string]any{
|
||||
"name": format["name"], "schema": format["schema"], "strict": format["strict"],
|
||||
}}, nil
|
||||
default:
|
||||
return nil, unsupportedResponseParameter("text.format.type")
|
||||
}
|
||||
}
|
||||
|
||||
func chatAssistantMessage(internal map[string]any, visible map[string]any) map[string]any {
|
||||
if choices, ok := internal["choices"].([]any); ok && len(choices) > 0 {
|
||||
choice, _ := choices[0].(map[string]any)
|
||||
if message, ok := choice["message"].(map[string]any); ok {
|
||||
return cloneMapAny(message)
|
||||
}
|
||||
}
|
||||
output, _ := visible["output"].([]any)
|
||||
message := map[string]any{"role": "assistant"}
|
||||
textParts := make([]string, 0)
|
||||
toolCalls := make([]any, 0)
|
||||
for _, raw := range output {
|
||||
item, _ := raw.(map[string]any)
|
||||
switch stringFromAny(item["type"]) {
|
||||
case "message":
|
||||
content, _ := item["content"].([]any)
|
||||
for _, rawContent := range content {
|
||||
part, _ := rawContent.(map[string]any)
|
||||
if stringFromAny(part["type"]) == "output_text" {
|
||||
textParts = append(textParts, stringFromAny(part["text"]))
|
||||
}
|
||||
}
|
||||
case "function_call":
|
||||
toolCalls = append(toolCalls, map[string]any{
|
||||
"id": item["call_id"], "type": "function",
|
||||
"function": map[string]any{"name": item["name"], "arguments": item["arguments"]},
|
||||
})
|
||||
}
|
||||
}
|
||||
message["content"] = strings.Join(textParts, "")
|
||||
if len(toolCalls) > 0 {
|
||||
message["tool_calls"] = toolCalls
|
||||
}
|
||||
if len(textParts) == 0 && len(toolCalls) == 0 {
|
||||
return nil
|
||||
}
|
||||
return message
|
||||
}
|
||||
|
||||
func ChatResultToResponse(chat map[string]any, publicID string, model string, requestBody map[string]any) map[string]any {
|
||||
createdAt := time.Now().Unix()
|
||||
if value := intFromAny(chat["created"]); value > 0 {
|
||||
createdAt = int64(value)
|
||||
}
|
||||
output := make([]any, 0)
|
||||
outputText := ""
|
||||
choices, _ := chat["choices"].([]any)
|
||||
if len(choices) > 0 {
|
||||
choice, _ := choices[0].(map[string]any)
|
||||
message, _ := choice["message"].(map[string]any)
|
||||
if content := visibleChatContent(message["content"]); content != "" {
|
||||
outputText = content
|
||||
output = append(output, map[string]any{
|
||||
"id": "msg_" + responseIDSuffix(publicID), "type": "message", "status": "completed", "role": "assistant",
|
||||
"content": []any{map[string]any{"type": "output_text", "text": content, "annotations": []any{}, "logprobs": []any{}}},
|
||||
})
|
||||
}
|
||||
toolCalls, _ := message["tool_calls"].([]any)
|
||||
for index, rawToolCall := range toolCalls {
|
||||
toolCall, _ := rawToolCall.(map[string]any)
|
||||
function, _ := toolCall["function"].(map[string]any)
|
||||
callID := firstNonEmptyString(toolCall["id"], fmt.Sprintf("call_%d", index))
|
||||
output = append(output, map[string]any{
|
||||
"id": "fc_" + responseIDSuffix(publicID) + fmt.Sprintf("_%d", index),
|
||||
"type": "function_call", "status": "completed", "call_id": callID,
|
||||
"name": stringFromAny(function["name"]), "arguments": stringFromAny(function["arguments"]),
|
||||
})
|
||||
}
|
||||
}
|
||||
usage := responseUsageFromChat(chat["usage"])
|
||||
out := map[string]any{
|
||||
"id": publicID, "object": "response", "created_at": createdAt, "status": "completed",
|
||||
"model": model, "output": output, "output_text": outputText, "error": nil, "incomplete_details": nil,
|
||||
"usage": usage,
|
||||
}
|
||||
if previousResponseID := strings.TrimSpace(stringFromAny(requestBody["previous_response_id"])); previousResponseID != "" {
|
||||
out["previous_response_id"] = previousResponseID
|
||||
} else {
|
||||
out["previous_response_id"] = nil
|
||||
}
|
||||
applyResponseFinishReason(out, choices)
|
||||
for _, key := range []string{"instructions", "parallel_tool_calls", "temperature", "top_p", "tools", "tool_choice", "metadata"} {
|
||||
if value, ok := requestBody[key]; ok {
|
||||
out[key] = value
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func applyResponseFinishReason(response map[string]any, choices []any) {
|
||||
if len(choices) == 0 {
|
||||
return
|
||||
}
|
||||
choice, _ := choices[0].(map[string]any)
|
||||
switch strings.TrimSpace(stringFromAny(choice["finish_reason"])) {
|
||||
case "length":
|
||||
response["status"] = "incomplete"
|
||||
response["incomplete_details"] = map[string]any{"reason": "max_output_tokens"}
|
||||
case "content_filter":
|
||||
response["status"] = "incomplete"
|
||||
response["incomplete_details"] = map[string]any{"reason": "content_filter"}
|
||||
}
|
||||
}
|
||||
|
||||
func responseUsageFromChat(value any) map[string]any {
|
||||
usage, _ := value.(map[string]any)
|
||||
normalized := usageFromOpenAIUsage(usage)
|
||||
return map[string]any{
|
||||
"input_tokens": normalized.InputTokens,
|
||||
"input_tokens_details": map[string]any{"cached_tokens": normalized.CachedInputTokens},
|
||||
"output_tokens": normalized.OutputTokens,
|
||||
"output_tokens_details": map[string]any{"reasoning_tokens": reasoningTokensFromUsage(usage)},
|
||||
"total_tokens": normalized.TotalTokens,
|
||||
}
|
||||
}
|
||||
|
||||
func reasoningTokensFromUsage(usage map[string]any) int {
|
||||
for _, key := range []string{"completion_tokens_details", "output_tokens_details"} {
|
||||
if details, ok := usage[key].(map[string]any); ok {
|
||||
if value := intFromAny(details["reasoning_tokens"]); value > 0 {
|
||||
return value
|
||||
}
|
||||
}
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func visibleChatContent(value any) string {
|
||||
if text, ok := value.(string); ok {
|
||||
return text
|
||||
}
|
||||
items, _ := value.([]any)
|
||||
parts := make([]string, 0, len(items))
|
||||
for _, raw := range items {
|
||||
item, _ := raw.(map[string]any)
|
||||
if text := stringFromAny(firstPresent(item["text"], item["content"])); text != "" && !isReasoningContentBlock(item) {
|
||||
parts = append(parts, text)
|
||||
}
|
||||
}
|
||||
return strings.Join(parts, "")
|
||||
}
|
||||
|
||||
func decodeNativeResponsesStream(resp *http.Response, onDelta StreamDelta) (map[string]any, string, error) {
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||
raw, _ := io.ReadAll(io.LimitReader(resp.Body, 16*1024*1024))
|
||||
return nil, "", &ClientError{Code: statusCodeName(resp.StatusCode), Message: errorMessage(raw, resp.Status), StatusCode: resp.StatusCode, RequestID: requestIDFromHTTPResponse(resp), Retryable: HTTPRetryable(resp.StatusCode)}
|
||||
}
|
||||
scanner := bufio.NewScanner(resp.Body)
|
||||
scanner.Buffer(make([]byte, 0, 64*1024), 16*1024*1024)
|
||||
var completed map[string]any
|
||||
upstreamID := ""
|
||||
rawLines := make([]string, 0)
|
||||
eventName := ""
|
||||
dataLines := make([]string, 0)
|
||||
processFrame := func() error {
|
||||
payload := strings.TrimSpace(strings.Join(dataLines, "\n"))
|
||||
frameEventName := eventName
|
||||
eventName = ""
|
||||
dataLines = dataLines[:0]
|
||||
if payload == "" || payload == "[DONE]" {
|
||||
return nil
|
||||
}
|
||||
var event map[string]any
|
||||
if err := json.Unmarshal([]byte(payload), &event); err != nil {
|
||||
return nil
|
||||
}
|
||||
if stringFromAny(event["type"]) == "" && frameEventName != "" {
|
||||
event["type"] = frameEventName
|
||||
}
|
||||
if failure := responseStreamError(event); failure != nil {
|
||||
return failure
|
||||
}
|
||||
if response, ok := event["response"].(map[string]any); ok && upstreamID == "" {
|
||||
upstreamID = stringFromAny(response["id"])
|
||||
}
|
||||
if stringFromAny(event["type"]) == "response.completed" {
|
||||
completed, _ = event["response"].(map[string]any)
|
||||
if onDelta != nil {
|
||||
return onDelta(StreamDeltaEvent{Event: event})
|
||||
}
|
||||
return nil
|
||||
}
|
||||
if stringFromAny(event["type"]) == "response.failed" {
|
||||
if onDelta != nil {
|
||||
if err := onDelta(StreamDeltaEvent{Event: event}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
failed, _ := event["response"].(map[string]any)
|
||||
failure, _ := failed["error"].(map[string]any)
|
||||
code := firstNonEmptyString(failure["code"], "response_failed")
|
||||
message := firstNonEmptyString(failure["message"], "upstream Responses request failed")
|
||||
return &ClientError{Code: code, Message: message, StatusCode: http.StatusBadRequest, Retryable: false}
|
||||
}
|
||||
if onDelta != nil {
|
||||
return onDelta(StreamDeltaEvent{Text: nativeResponseDeltaText(event), Event: event})
|
||||
}
|
||||
return nil
|
||||
}
|
||||
for scanner.Scan() {
|
||||
rawLine := scanner.Text()
|
||||
rawLines = append(rawLines, rawLine)
|
||||
line := strings.TrimSpace(rawLine)
|
||||
if line == "" {
|
||||
if err := processFrame(); err != nil {
|
||||
return nil, upstreamID, err
|
||||
}
|
||||
continue
|
||||
}
|
||||
if strings.HasPrefix(line, "event:") {
|
||||
eventName = strings.TrimSpace(strings.TrimPrefix(line, "event:"))
|
||||
continue
|
||||
}
|
||||
if strings.HasPrefix(line, "data:") {
|
||||
dataLines = append(dataLines, strings.TrimSpace(strings.TrimPrefix(line, "data:")))
|
||||
}
|
||||
}
|
||||
if err := processFrame(); err != nil {
|
||||
return nil, upstreamID, err
|
||||
}
|
||||
if err := scanner.Err(); err != nil {
|
||||
return nil, upstreamID, &ClientError{Code: "stream_read_error", Message: err.Error(), Retryable: true}
|
||||
}
|
||||
if completed == nil {
|
||||
raw := strings.TrimSpace(strings.Join(rawLines, "\n"))
|
||||
var response map[string]any
|
||||
if raw != "" && json.Unmarshal([]byte(raw), &response) == nil {
|
||||
if failure := responseStreamError(response); failure != nil {
|
||||
return nil, upstreamID, failure
|
||||
}
|
||||
if stringFromAny(response["object"]) == "response" {
|
||||
upstreamID = stringFromAny(response["id"])
|
||||
return response, upstreamID, nil
|
||||
}
|
||||
}
|
||||
return nil, upstreamID, &ClientError{Code: "invalid_response", Message: "Responses stream ended without response.completed", Retryable: false}
|
||||
}
|
||||
return completed, upstreamID, nil
|
||||
}
|
||||
|
||||
func responseStreamError(event map[string]any) error {
|
||||
if stringFromAny(event["type"]) != "error" && event["error"] == nil {
|
||||
return nil
|
||||
}
|
||||
failure, _ := event["error"].(map[string]any)
|
||||
if failure == nil {
|
||||
failure = event
|
||||
}
|
||||
code := firstNonEmptyString(failure["code"], "response_failed")
|
||||
message := firstNonEmptyString(failure["message"], "upstream Responses stream failed")
|
||||
return &ClientError{Code: code, Message: message, StatusCode: http.StatusBadRequest, Retryable: false}
|
||||
}
|
||||
|
||||
func nativeResponseDeltaText(event map[string]any) string {
|
||||
if stringFromAny(event["type"]) == "response.output_text.delta" {
|
||||
return stringFromAny(event["delta"])
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
type chatResponsesStreamAdapter struct {
|
||||
publicID string
|
||||
model string
|
||||
started bool
|
||||
sequence int
|
||||
nextOutput int
|
||||
textOutput int
|
||||
textAdded bool
|
||||
text strings.Builder
|
||||
tools map[int]*chatResponseStreamTool
|
||||
}
|
||||
|
||||
type chatResponseStreamTool struct {
|
||||
ItemID string
|
||||
CallID string
|
||||
Name string
|
||||
Arguments strings.Builder
|
||||
Added bool
|
||||
OutputIndex int
|
||||
}
|
||||
|
||||
func newChatResponsesStreamAdapter(publicID string, model string) *chatResponsesStreamAdapter {
|
||||
return &chatResponsesStreamAdapter{publicID: publicID, model: model, textOutput: -1, tools: map[int]*chatResponseStreamTool{}}
|
||||
}
|
||||
|
||||
func (a *chatResponsesStreamAdapter) emit(onDelta StreamDelta, event map[string]any) error {
|
||||
if onDelta == nil {
|
||||
return nil
|
||||
}
|
||||
event["sequence_number"] = a.sequence
|
||||
a.sequence++
|
||||
return onDelta(StreamDeltaEvent{Text: nativeResponseDeltaText(event), Event: event})
|
||||
}
|
||||
|
||||
func (a *chatResponsesStreamAdapter) start(onDelta StreamDelta) error {
|
||||
if a.started || onDelta == nil {
|
||||
return nil
|
||||
}
|
||||
a.started = true
|
||||
response := map[string]any{"id": a.publicID, "object": "response", "created_at": time.Now().Unix(), "status": "in_progress", "model": a.model, "output": []any{}}
|
||||
for _, eventType := range []string{"response.created", "response.in_progress"} {
|
||||
if err := a.emit(onDelta, map[string]any{"type": eventType, "response": response}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *chatResponsesStreamAdapter) delta(event StreamDeltaEvent, onDelta StreamDelta) error {
|
||||
if onDelta == nil || event.Event == nil {
|
||||
return nil
|
||||
}
|
||||
if err := a.start(onDelta); err != nil {
|
||||
return err
|
||||
}
|
||||
choices, _ := event.Event["choices"].([]any)
|
||||
for _, rawChoice := range choices {
|
||||
choice, _ := rawChoice.(map[string]any)
|
||||
delta, _ := choice["delta"].(map[string]any)
|
||||
if content := stringFromAny(delta["content"]); content != "" {
|
||||
if !a.textAdded {
|
||||
a.textAdded = true
|
||||
a.textOutput = a.nextOutput
|
||||
a.nextOutput++
|
||||
itemID := "msg_" + responseIDSuffix(a.publicID)
|
||||
if err := a.emit(onDelta, map[string]any{
|
||||
"type": "response.output_item.added", "response_id": a.publicID, "output_index": a.textOutput,
|
||||
"item": map[string]any{"id": itemID, "type": "message", "status": "in_progress", "role": "assistant", "content": []any{}},
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := a.emit(onDelta, map[string]any{
|
||||
"type": "response.content_part.added", "response_id": a.publicID, "item_id": itemID,
|
||||
"output_index": a.textOutput, "content_index": 0,
|
||||
"part": map[string]any{"type": "output_text", "text": "", "annotations": []any{}, "logprobs": []any{}},
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
a.text.WriteString(content)
|
||||
if err := a.emit(onDelta, map[string]any{
|
||||
"type": "response.output_text.delta", "response_id": a.publicID,
|
||||
"item_id": "msg_" + responseIDSuffix(a.publicID), "output_index": a.textOutput, "content_index": 0, "delta": content,
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
toolCalls, _ := delta["tool_calls"].([]any)
|
||||
for _, rawToolCall := range toolCalls {
|
||||
toolCall, _ := rawToolCall.(map[string]any)
|
||||
index := intFromAny(toolCall["index"])
|
||||
tool := a.tools[index]
|
||||
if tool == nil {
|
||||
tool = &chatResponseStreamTool{ItemID: fmt.Sprintf("fc_%s_%d", responseIDSuffix(a.publicID), index), OutputIndex: a.nextOutput}
|
||||
a.nextOutput++
|
||||
a.tools[index] = tool
|
||||
}
|
||||
if callID := stringFromAny(toolCall["id"]); callID != "" {
|
||||
tool.CallID = callID
|
||||
}
|
||||
function, _ := toolCall["function"].(map[string]any)
|
||||
if name := stringFromAny(function["name"]); name != "" {
|
||||
tool.Name += name
|
||||
}
|
||||
arguments := stringFromAny(function["arguments"])
|
||||
if !tool.Added && (tool.CallID != "" || tool.Name != "" || arguments != "") {
|
||||
tool.Added = true
|
||||
if err := a.emit(onDelta, map[string]any{
|
||||
"type": "response.output_item.added", "response_id": a.publicID, "output_index": tool.OutputIndex,
|
||||
"item": map[string]any{"id": tool.ItemID, "type": "function_call", "status": "in_progress", "call_id": tool.CallID, "name": tool.Name, "arguments": ""},
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if arguments != "" {
|
||||
tool.Arguments.WriteString(arguments)
|
||||
if err := a.emit(onDelta, map[string]any{
|
||||
"type": "response.function_call_arguments.delta", "response_id": a.publicID,
|
||||
"item_id": tool.ItemID, "output_index": tool.OutputIndex, "delta": arguments,
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *chatResponsesStreamAdapter) done(result map[string]any, onDelta StreamDelta) error {
|
||||
if onDelta == nil {
|
||||
return nil
|
||||
}
|
||||
a.alignOutput(result)
|
||||
if a.text.Len() > 0 {
|
||||
text := a.text.String()
|
||||
itemID := "msg_" + responseIDSuffix(a.publicID)
|
||||
if err := a.emit(onDelta, map[string]any{"type": "response.output_text.done", "response_id": a.publicID, "item_id": itemID, "output_index": a.textOutput, "content_index": 0, "text": text}); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := a.emit(onDelta, map[string]any{
|
||||
"type": "response.content_part.done", "response_id": a.publicID, "item_id": itemID,
|
||||
"output_index": a.textOutput, "content_index": 0,
|
||||
"part": map[string]any{"type": "output_text", "text": text, "annotations": []any{}, "logprobs": []any{}},
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := a.emit(onDelta, map[string]any{"type": "response.output_item.done", "response_id": a.publicID, "output_index": a.textOutput, "item": firstResponseOutputItem(result, "message")}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
tools := make([]*chatResponseStreamTool, 0, len(a.tools))
|
||||
for _, tool := range a.tools {
|
||||
tools = append(tools, tool)
|
||||
}
|
||||
sort.Slice(tools, func(i, j int) bool { return tools[i].OutputIndex < tools[j].OutputIndex })
|
||||
for _, tool := range tools {
|
||||
arguments := tool.Arguments.String()
|
||||
if err := a.emit(onDelta, map[string]any{"type": "response.function_call_arguments.done", "response_id": a.publicID, "item_id": tool.ItemID, "output_index": tool.OutputIndex, "arguments": arguments}); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := a.emit(onDelta, map[string]any{"type": "response.output_item.done", "response_id": a.publicID, "output_index": tool.OutputIndex, "item": responseOutputItemByID(result, tool.ItemID)}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *chatResponsesStreamAdapter) alignOutput(response map[string]any) {
|
||||
items, _ := response["output"].([]any)
|
||||
if len(items) == 0 || a.nextOutput == 0 {
|
||||
return
|
||||
}
|
||||
ordered := make([]any, a.nextOutput)
|
||||
if a.textOutput >= 0 {
|
||||
ordered[a.textOutput] = firstResponseOutputItem(response, "message")
|
||||
}
|
||||
for _, tool := range a.tools {
|
||||
ordered[tool.OutputIndex] = responseOutputItemByID(response, tool.ItemID)
|
||||
}
|
||||
result := make([]any, 0, len(ordered))
|
||||
for _, item := range ordered {
|
||||
if item != nil {
|
||||
result = append(result, item)
|
||||
}
|
||||
}
|
||||
response["output"] = result
|
||||
}
|
||||
|
||||
func firstResponseOutputItem(response map[string]any, itemType string) any {
|
||||
items, _ := response["output"].([]any)
|
||||
for _, raw := range items {
|
||||
item, _ := raw.(map[string]any)
|
||||
if stringFromAny(item["type"]) == itemType {
|
||||
return item
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func responseOutputItemByID(response map[string]any, id string) any {
|
||||
items, _ := response["output"].([]any)
|
||||
for _, raw := range items {
|
||||
item, _ := raw.(map[string]any)
|
||||
if stringFromAny(item["id"]) == id {
|
||||
return item
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func responseIDSuffix(publicID string) string {
|
||||
return strings.TrimPrefix(publicID, "resp_")
|
||||
}
|
||||
|
||||
func unsupportedResponseParameter(parameter string) error {
|
||||
return &ClientError{Code: "unsupported_response_parameter", Message: "Chat fallback does not support Responses parameter: " + parameter, StatusCode: http.StatusBadRequest}
|
||||
}
|
||||
@@ -0,0 +1,332 @@
|
||||
package clients
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
|
||||
)
|
||||
|
||||
func TestOpenAIResponsesNativeUsesResponsesEndpointAndPreservesVendorIDs(t *testing.T) {
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Path != "/responses" {
|
||||
t.Fatalf("expected /responses, got %s", r.URL.Path)
|
||||
}
|
||||
var body map[string]any
|
||||
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if body["messages"] != nil {
|
||||
t.Fatalf("native Responses request must not contain messages: %+v", body)
|
||||
}
|
||||
if body["previous_response_id"] != "resp_upstream_parent" {
|
||||
t.Fatalf("expected translated upstream previous id, got %+v", body["previous_response_id"])
|
||||
}
|
||||
_ = json.NewEncoder(w).Encode(map[string]any{
|
||||
"id": "resp_upstream_child", "object": "response", "status": "completed", "previous_response_id": "resp_upstream_parent",
|
||||
"output": []any{
|
||||
map[string]any{
|
||||
"type": "message",
|
||||
"content": []any{map[string]any{"type": "output_text", "text": "ok"}},
|
||||
},
|
||||
},
|
||||
"usage": map[string]any{"input_tokens": 2, "output_tokens": 1, "total_tokens": 3},
|
||||
})
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
response, err := (OpenAIClient{}).Run(context.Background(), Request{
|
||||
Kind: "responses", Model: "Demo",
|
||||
Body: map[string]any{"input": "hello", "messages": []any{map[string]any{"role": "user", "content": "illegal"}}},
|
||||
Candidate: store.RuntimeModelCandidate{BaseURL: server.URL, ProviderModelName: "demo", Credentials: map[string]any{"apiKey": "secret"}},
|
||||
UpstreamProtocol: ProtocolOpenAIResponses, PublicResponseID: "resp_12345678901234567890123456789012",
|
||||
PublicPreviousResponseID: "resp_abcdefghijklmnopqrstuvwxyz123456", UpstreamPreviousResponseID: "resp_upstream_parent",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if response.Result["id"] != "resp_upstream_child" || response.Result["previous_response_id"] != "resp_upstream_parent" {
|
||||
t.Fatalf("vendor response ids were not preserved: %+v", response.Result)
|
||||
}
|
||||
if response.PublicResponseID != "resp_upstream_child" || response.UpstreamResponseID != "resp_upstream_child" || response.UpstreamEndpoint != "/responses" || response.ResponseConverted {
|
||||
t.Fatalf("unexpected native response metadata: %+v", response)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpenAIResponsesChatFallbackPreservesHistoryToolsUsageAndReasoningInternally(t *testing.T) {
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Path != "/chat/completions" {
|
||||
t.Fatalf("expected /chat/completions, got %s", r.URL.Path)
|
||||
}
|
||||
var body map[string]any
|
||||
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
messages, _ := body["messages"].([]any)
|
||||
if len(messages) != 4 {
|
||||
t.Fatalf("expected prior user/assistant plus current system/user/tool messages, got %+v", messages)
|
||||
}
|
||||
if body["max_tokens"] != float64(128) || body["reasoning_effort"] != "high" {
|
||||
t.Fatalf("expected mapped max/reasoning fields: %+v", body)
|
||||
}
|
||||
tools, _ := body["tools"].([]any)
|
||||
if len(tools) != 1 {
|
||||
t.Fatalf("expected mapped function tool: %+v", body["tools"])
|
||||
}
|
||||
_ = json.NewEncoder(w).Encode(map[string]any{
|
||||
"id": "chatcmpl-upstream", "object": "chat.completion", "created": 1710000000, "model": "demo",
|
||||
"choices": []any{map[string]any{"index": 0, "finish_reason": "tool_calls", "message": map[string]any{
|
||||
"role": "assistant", "content": "visible", "reasoning_content": "hidden reasoning",
|
||||
"tool_calls": []any{map[string]any{"id": "call_stable", "type": "function", "function": map[string]any{"name": "lookup", "arguments": "{\"x\":1}"}}},
|
||||
}}},
|
||||
"usage": map[string]any{"prompt_tokens": 20, "completion_tokens": 5, "total_tokens": 25, "prompt_tokens_details": map[string]any{"cached_tokens": 7}},
|
||||
})
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
response, err := (OpenAIClient{}).Run(context.Background(), Request{
|
||||
Kind: "responses", Model: "Demo", Candidate: store.RuntimeModelCandidate{BaseURL: server.URL, ProviderModelName: "demo", Credentials: map[string]any{"apiKey": "secret"}},
|
||||
UpstreamProtocol: ProtocolOpenAIChatCompletions, PublicResponseID: "resp_12345678901234567890123456789012",
|
||||
PreviousResponseTurns: []ResponseTurn{{
|
||||
Request: map[string]any{"input": "prior user", "instructions": "must not be replayed"},
|
||||
Internal: map[string]any{
|
||||
"choices": []any{map[string]any{
|
||||
"message": map[string]any{"role": "assistant", "content": "prior assistant"},
|
||||
}},
|
||||
},
|
||||
}},
|
||||
Body: map[string]any{
|
||||
"instructions": "current instruction", "previous_response_id": "resp_parent", "input": []any{map[string]any{"type": "function_call_output", "call_id": "call_prior", "output": "done"}},
|
||||
"max_output_tokens": 128, "reasoning": map[string]any{"effort": "high"},
|
||||
"tools": []any{map[string]any{"type": "function", "name": "lookup", "description": "demo", "parameters": map[string]any{"type": "object"}}},
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !response.ResponseConverted || response.UpstreamProtocol != ProtocolOpenAIChatCompletions || response.UpstreamEndpoint != "/chat/completions" {
|
||||
t.Fatalf("unexpected conversion metadata: %+v", response)
|
||||
}
|
||||
if response.Result["previous_response_id"] != "resp_parent" {
|
||||
t.Fatalf("Chat fallback response lost previous_response_id: %+v", response.Result)
|
||||
}
|
||||
output, _ := response.Result["output"].([]any)
|
||||
if len(output) != 2 {
|
||||
t.Fatalf("expected message plus function call: %+v", response.Result)
|
||||
}
|
||||
call, _ := output[1].(map[string]any)
|
||||
if call["call_id"] != "call_stable" || call["arguments"] != "{\"x\":1}" {
|
||||
t.Fatalf("function call continuity lost: %+v", call)
|
||||
}
|
||||
if strings.Contains(string(mustJSON(t, response.Result)), "hidden reasoning") {
|
||||
t.Fatalf("visible Responses result leaked reasoning: %+v", response.Result)
|
||||
}
|
||||
if !strings.Contains(string(mustJSON(t, response.InternalResult)), "hidden reasoning") {
|
||||
t.Fatalf("internal snapshot must retain reasoning for continuation: %+v", response.InternalResult)
|
||||
}
|
||||
usage, _ := response.Result["usage"].(map[string]any)
|
||||
details, _ := usage["input_tokens_details"].(map[string]any)
|
||||
if details["cached_tokens"] != 7 {
|
||||
t.Fatalf("cached usage was not preserved: %+v", usage)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpenAIResponsesNativeStreamForwardsEventsWithoutChatAggregation(t *testing.T) {
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
_, _ = w.Write([]byte("event: response.created\ndata: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_upstream\",\"object\":\"response\",\"status\":\"in_progress\"}}\n\n"))
|
||||
_, _ = w.Write([]byte("event: response.output_text.delta\ndata: {\"type\":\"response.output_text.delta\",\"response_id\":\"resp_upstream\",\"delta\":\"hello\"}\n\n"))
|
||||
_, _ = w.Write([]byte("event: response.completed\ndata: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_upstream\",\"object\":\"response\",\"status\":\"completed\",\"output\":[],\"usage\":{\"input_tokens\":1,\"output_tokens\":1,\"total_tokens\":2}}}\n\n"))
|
||||
}))
|
||||
defer server.Close()
|
||||
events := make([]StreamDeltaEvent, 0)
|
||||
response, err := (OpenAIClient{}).Run(context.Background(), Request{
|
||||
Kind: "responses", Model: "Demo", Body: map[string]any{"input": "hello", "stream": true}, Stream: true,
|
||||
Candidate: store.RuntimeModelCandidate{BaseURL: server.URL, ProviderModelName: "demo", Credentials: map[string]any{"apiKey": "secret"}},
|
||||
UpstreamProtocol: ProtocolOpenAIResponses, PublicResponseID: "resp_12345678901234567890123456789012",
|
||||
StreamDelta: func(event StreamDeltaEvent) error { events = append(events, event); return nil },
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(events) != 3 || events[0].Event["type"] != "response.created" || events[1].Event["type"] != "response.output_text.delta" || events[2].Event["type"] != "response.completed" {
|
||||
t.Fatalf("unexpected forwarded events: %+v", events)
|
||||
}
|
||||
if response.Result["object"] != "response" || response.Result["choices"] != nil || response.UpstreamResponseID != "resp_upstream" {
|
||||
t.Fatalf("native stream was rewritten as Chat: %+v", response)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpenAIResponsesChatFallbackStreamsFunctionArgumentFragments(t *testing.T) {
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Path != "/chat/completions" {
|
||||
t.Fatalf("expected chat fallback endpoint, got %s", r.URL.Path)
|
||||
}
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
_, _ = w.Write([]byte("data: {\"id\":\"chatcmpl-tool\",\"object\":\"chat.completion.chunk\",\"model\":\"demo\",\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\",\"tool_calls\":[{\"index\":0,\"id\":\"call_stream_0\",\"type\":\"function\",\"function\":{\"name\":\"lookup_x\",\"arguments\":\"{\\\"x\\\":\"}},{\"index\":1,\"id\":\"call_stream_1\",\"type\":\"function\",\"function\":{\"name\":\"lookup_y\",\"arguments\":\"{\\\"y\\\":\"}}]},\"finish_reason\":null}]}\n\n"))
|
||||
_, _ = w.Write([]byte("data: {\"id\":\"chatcmpl-tool\",\"object\":\"chat.completion.chunk\",\"model\":\"demo\",\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"1}\"}},{\"index\":1,\"function\":{\"arguments\":\"2}\"}}]},\"finish_reason\":null}]}\n\n"))
|
||||
_, _ = w.Write([]byte("data: {\"id\":\"chatcmpl-tool\",\"object\":\"chat.completion.chunk\",\"model\":\"demo\",\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"tool_calls\"}],\"usage\":{\"prompt_tokens\":2,\"completion_tokens\":3,\"total_tokens\":5}}\n\n"))
|
||||
_, _ = w.Write([]byte("data: [DONE]\n\n"))
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
events := make([]StreamDeltaEvent, 0)
|
||||
response, err := (OpenAIClient{}).Run(context.Background(), Request{
|
||||
Kind: "responses", Model: "Demo", Body: map[string]any{
|
||||
"input": "call it", "stream": true,
|
||||
"tools": []any{map[string]any{"type": "function", "name": "lookup", "parameters": map[string]any{"type": "object"}}},
|
||||
}, Stream: true,
|
||||
Candidate: store.RuntimeModelCandidate{BaseURL: server.URL, ProviderModelName: "demo", Credentials: map[string]any{"apiKey": "secret"}},
|
||||
UpstreamProtocol: ProtocolOpenAIChatCompletions, PublicResponseID: "resp_12345678901234567890123456789012",
|
||||
StreamDelta: func(event StreamDeltaEvent) error { events = append(events, event); return nil },
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
types := make([]string, 0, len(events))
|
||||
fragments := map[string]string{}
|
||||
outputIndexes := map[int]bool{}
|
||||
for _, event := range events {
|
||||
eventType := stringFromAny(event.Event["type"])
|
||||
types = append(types, eventType)
|
||||
if sequence := intFromAny(event.Event["sequence_number"]); sequence != len(types)-1 {
|
||||
t.Fatalf("converted event sequence is not contiguous at %d: %+v", len(types)-1, event.Event)
|
||||
}
|
||||
if eventType == "response.function_call_arguments.delta" {
|
||||
itemID := stringFromAny(event.Event["item_id"])
|
||||
fragments[itemID] += stringFromAny(event.Event["delta"])
|
||||
outputIndexes[intFromAny(event.Event["output_index"])] = true
|
||||
}
|
||||
}
|
||||
if !containsTestString(types, "response.created") || !containsTestString(types, "response.in_progress") || !containsTestString(types, "response.output_item.added") || !containsTestString(types, "response.function_call_arguments.done") || len(fragments) != 2 || len(outputIndexes) != 2 {
|
||||
t.Fatalf("unexpected converted stream events types=%v fragments=%v indexes=%v", types, fragments, outputIndexes)
|
||||
}
|
||||
output, _ := response.Result["output"].([]any)
|
||||
if len(output) != 2 {
|
||||
t.Fatalf("expected two final function calls: %+v", response.Result)
|
||||
}
|
||||
first, _ := output[0].(map[string]any)
|
||||
second, _ := output[1].(map[string]any)
|
||||
if first["call_id"] != "call_stream_0" || first["arguments"] != "{\"x\":1}" || second["call_id"] != "call_stream_1" || second["arguments"] != "{\"y\":2}" {
|
||||
t.Fatalf("parallel streamed calls were not aggregated: %+v", output)
|
||||
}
|
||||
}
|
||||
|
||||
func TestChatResultToResponseMapsIncompleteFinishReason(t *testing.T) {
|
||||
response := ChatResultToResponse(map[string]any{
|
||||
"choices": []any{map[string]any{"finish_reason": "length", "message": map[string]any{"role": "assistant", "content": "partial"}}},
|
||||
}, "resp_12345678901234567890123456789012", "demo", map[string]any{})
|
||||
if response["status"] != "incomplete" {
|
||||
t.Fatalf("expected incomplete response, got %+v", response)
|
||||
}
|
||||
details, _ := response["incomplete_details"].(map[string]any)
|
||||
if details["reason"] != "max_output_tokens" {
|
||||
t.Fatalf("finish reason was not mapped: %+v", details)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNativeResponsesStreamSupportsMultilineDataFrames(t *testing.T) {
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
_, _ = w.Write([]byte("event: response.completed\n"))
|
||||
_, _ = w.Write([]byte("data: {\"type\":\"response.completed\",\n"))
|
||||
_, _ = w.Write([]byte("data: \"response\":{\"id\":\"resp_vendor-with-dash\",\"object\":\"response\",\"status\":\"completed\",\"output\":[]}}\n\n"))
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
response, err := (OpenAIClient{}).Run(context.Background(), Request{
|
||||
Kind: "responses", Model: "Demo", Body: map[string]any{"input": "hello", "stream": true}, Stream: true,
|
||||
Candidate: store.RuntimeModelCandidate{BaseURL: server.URL, ProviderModelName: "demo", Credentials: map[string]any{"apiKey": "secret"}},
|
||||
UpstreamProtocol: ProtocolOpenAIResponses,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if response.Result["id"] != "resp_vendor-with-dash" {
|
||||
t.Fatalf("multiline frame was not decoded: %+v", response.Result)
|
||||
}
|
||||
}
|
||||
|
||||
func TestChatResponsesStreamAdapterEmitsTextItemLifecycle(t *testing.T) {
|
||||
adapter := newChatResponsesStreamAdapter("resp_12345678901234567890123456789012", "demo")
|
||||
events := make([]string, 0)
|
||||
onDelta := func(event StreamDeltaEvent) error {
|
||||
events = append(events, stringFromAny(event.Event["type"]))
|
||||
return nil
|
||||
}
|
||||
err := adapter.delta(StreamDeltaEvent{Event: map[string]any{
|
||||
"choices": []any{map[string]any{"delta": map[string]any{"content": "hello"}}},
|
||||
}}, onDelta)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
result := ChatResultToResponse(map[string]any{
|
||||
"choices": []any{map[string]any{"message": map[string]any{"role": "assistant", "content": "hello"}}},
|
||||
}, adapter.publicID, "demo", map[string]any{})
|
||||
if err := adapter.done(result, onDelta); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
want := []string{
|
||||
"response.created", "response.in_progress", "response.output_item.added", "response.content_part.added",
|
||||
"response.output_text.delta", "response.output_text.done", "response.content_part.done", "response.output_item.done",
|
||||
}
|
||||
if strings.Join(events, ",") != strings.Join(want, ",") {
|
||||
t.Fatalf("unexpected text event lifecycle got=%v want=%v", events, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResponsesChatFallbackRejectsBuiltInToolsAndUnknownParameters(t *testing.T) {
|
||||
_, err := ResponsesRequestToChat(map[string]any{"input": "hello", "tools": []any{map[string]any{"type": "web_search_preview"}}}, nil)
|
||||
if ErrorCode(err) != "unsupported_response_tool" {
|
||||
t.Fatalf("expected unsupported_response_tool, got %v", err)
|
||||
}
|
||||
_, err = ResponsesRequestToChat(map[string]any{"input": "hello", "conversation": "conv_1"}, nil)
|
||||
if ErrorCode(err) != "unsupported_response_parameter" {
|
||||
t.Fatalf("expected unsupported_response_parameter, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResponsesChatFallbackKeepsClientManagedStateAuthoritative(t *testing.T) {
|
||||
input := []any{
|
||||
map[string]any{"type": "message", "role": "user", "content": "first"},
|
||||
map[string]any{"type": "message", "role": "assistant", "content": "second"},
|
||||
map[string]any{"type": "message", "role": "user", "content": "third"},
|
||||
}
|
||||
body, err := ResponsesRequestToChat(map[string]any{"input": input}, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
messages, _ := body["messages"].([]any)
|
||||
if len(messages) != len(input) {
|
||||
t.Fatalf("client-managed input was supplemented by Gateway history: %+v", messages)
|
||||
}
|
||||
for index, raw := range messages {
|
||||
message, _ := raw.(map[string]any)
|
||||
expected, _ := input[index].(map[string]any)
|
||||
if message["role"] != expected["role"] || message["content"] != expected["content"] {
|
||||
t.Fatalf("client-managed message %d changed: got=%+v want=%+v", index, message, expected)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func mustJSON(t *testing.T, value any) []byte {
|
||||
t.Helper()
|
||||
data, err := json.Marshal(value)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return data
|
||||
}
|
||||
|
||||
func containsTestString(values []string, expected string) bool {
|
||||
for _, value := range values {
|
||||
if value == expected {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -11,17 +11,28 @@ import (
|
||||
)
|
||||
|
||||
type Request struct {
|
||||
Kind string
|
||||
ModelType string
|
||||
Model string
|
||||
Body map[string]any
|
||||
Candidate store.RuntimeModelCandidate
|
||||
HTTPClient *http.Client
|
||||
RemoteTaskID string
|
||||
RemoteTaskPayload map[string]any
|
||||
OnRemoteTaskSubmitted func(remoteTaskID string, payload map[string]any) error
|
||||
Stream bool
|
||||
StreamDelta StreamDelta
|
||||
Kind string
|
||||
ModelType string
|
||||
Model string
|
||||
Body map[string]any
|
||||
Candidate store.RuntimeModelCandidate
|
||||
HTTPClient *http.Client
|
||||
RemoteTaskID string
|
||||
RemoteTaskPayload map[string]any
|
||||
OnRemoteTaskSubmitted func(remoteTaskID string, payload map[string]any) error
|
||||
Stream bool
|
||||
StreamDelta StreamDelta
|
||||
UpstreamProtocol string
|
||||
PublicResponseID string
|
||||
PublicPreviousResponseID string
|
||||
UpstreamPreviousResponseID string
|
||||
PreviousResponseTurns []ResponseTurn
|
||||
}
|
||||
|
||||
type ResponseTurn struct {
|
||||
Request map[string]any
|
||||
Response map[string]any
|
||||
Internal map[string]any
|
||||
}
|
||||
|
||||
type Response struct {
|
||||
@@ -33,6 +44,14 @@ type Response struct {
|
||||
ResponseStartedAt time.Time
|
||||
ResponseFinishedAt time.Time
|
||||
ResponseDurationMS int64
|
||||
UpstreamProtocol string
|
||||
UpstreamEndpoint string
|
||||
UpstreamResponseID string
|
||||
PublicResponseID string
|
||||
ParentResponseID string
|
||||
ResponseChainDepth int
|
||||
ResponseConverted bool
|
||||
InternalResult map[string]any
|
||||
}
|
||||
|
||||
type Usage struct {
|
||||
|
||||
@@ -965,8 +965,6 @@ func (s *Server) listModelRateLimitStatuses(w http.ResponseWriter, r *http.Reque
|
||||
// @Router /api/v1/voice_clone [post]
|
||||
// @Router /chat/completions [post]
|
||||
// @Router /v1/chat/completions [post]
|
||||
// @Router /responses [post]
|
||||
// @Router /v1/responses [post]
|
||||
// @Router /embeddings [post]
|
||||
// @Router /v1/embeddings [post]
|
||||
// @Router /reranks [post]
|
||||
@@ -1085,6 +1083,26 @@ func (s *Server) createAPIV1ChatCompletions() http.Handler {
|
||||
return s.createTask("chat.completions", false)
|
||||
}
|
||||
|
||||
// openAIResponsesDoc godoc
|
||||
// @Summary 创建 OpenAI Responses
|
||||
// @Description 公开 OpenAI-compatible Responses 入口。模型声明 openai_responses 时原生转发，否则使用 Chat Completions 转换；store 缺省为 true。previous_response_id 严格绑定首次成功的平台模型和上游协议，链路不可用时不跨平台续接。未提供 previous_response_id 时由调用方管理完整状态，Gateway 以本轮 input/messages 为准且不追加本地历史。
|
||||
// @Tags responses
|
||||
// @Accept json
|
||||
// @Produce json
|
||||
// @Produce text/event-stream
|
||||
// @Security BearerAuth
|
||||
// @Param input body ResponsesRequest true "Responses 请求；Chat 回退只支持自定义 function tools"
|
||||
// @Success 200 {object} ResponsesCompatibleResponse
|
||||
// @Header 200 {string} X-Gateway-Task-Id "网关审计任务 ID"
|
||||
// @Failure 400 {object} ErrorEnvelope "invalid_previous_response_id / unsupported_response_tool / unsupported_response_parameter"
|
||||
// @Failure 401 {object} ErrorEnvelope
|
||||
// @Failure 402 {object} ErrorEnvelope
|
||||
// @Failure 403 {object} ErrorEnvelope
|
||||
// @Failure 503 {object} ErrorEnvelope "response_chain_unavailable"
|
||||
// @Router /responses [post]
|
||||
// @Router /v1/responses [post]
|
||||
func openAIResponsesDoc() {}
|
||||
|
||||
func (s *Server) requestExecutionContext(r *http.Request) (context.Context, context.CancelFunc) {
|
||||
base := context.WithoutCancel(r.Context())
|
||||
if s.ctx == nil {
|
||||
@@ -1116,6 +1134,7 @@ type taskExecutor interface {
|
||||
}
|
||||
|
||||
func writeCompatibleTaskResponse(runCtx context.Context, w http.ResponseWriter, r *http.Request, executor taskExecutor, kind string, model string, task store.GatewayTask, user *auth.User, streamMode bool, includeUsage bool) {
|
||||
w.Header().Set("X-Gateway-Task-Id", task.ID)
|
||||
if streamMode {
|
||||
flusher := prepareCompatibleStream(w)
|
||||
streamWriter := newCompatibleStreamWriter(kind, model, includeUsage)
|
||||
@@ -1306,6 +1325,10 @@ func scopeForTaskKind(kind string) string {
|
||||
|
||||
func statusFromRunError(err error) int {
|
||||
switch {
|
||||
case clients.ErrorCode(err) == "invalid_previous_response_id" || clients.ErrorCode(err) == "response_chain_too_deep" || clients.ErrorCode(err) == "unsupported_model_protocol" || clients.ErrorCode(err) == "unsupported_response_tool" || clients.ErrorCode(err) == "unsupported_response_parameter":
|
||||
return http.StatusBadRequest
|
||||
case clients.ErrorCode(err) == "response_chain_unavailable":
|
||||
return http.StatusServiceUnavailable
|
||||
case clients.ErrorCode(err) == "bad_request" || clients.ErrorCode(err) == "invalid_parameter" || clients.ErrorCode(err) == "cloned_voice_expired" || clients.ErrorCode(err) == "cloned_voice_unavailable" || clients.ErrorCode(err) == "cloned_voice_platform_unavailable" || clients.ErrorCode(err) == "unsupported_operation" || clients.ErrorCode(err) == "invalid_proxy":
|
||||
return http.StatusBadRequest
|
||||
case clients.ErrorCode(err) == "cloned_voice_not_found":
|
||||
|
||||
@@ -228,10 +228,32 @@ type ChatMessage struct {
|
||||
}
|
||||
|
||||
type ResponsesRequest struct {
|
||||
Model string `json:"model" example:"gpt-4o-mini"`
|
||||
Input interface{} `json:"input" example:"Tell me a short story"`
|
||||
Stream bool `json:"stream,omitempty" example:"false"`
|
||||
RunMode string `json:"runMode,omitempty" example:"simulation"`
|
||||
Model string `json:"model" example:"Doubao Seed 2.0 Pro"`
|
||||
Input interface{} `json:"input"`
|
||||
Instructions string `json:"instructions,omitempty" example:"Answer concisely"`
|
||||
PreviousResponseID string `json:"previous_response_id,omitempty" example:"resp_0123456789abcdef0123456789abcdef"`
|
||||
Tools []map[string]interface{} `json:"tools,omitempty"`
|
||||
ToolChoice interface{} `json:"tool_choice,omitempty"`
|
||||
ParallelToolCalls bool `json:"parallel_tool_calls,omitempty" example:"true"`
|
||||
MaxOutputTokens int `json:"max_output_tokens,omitempty" example:"512"`
|
||||
Reasoning map[string]interface{} `json:"reasoning,omitempty"`
|
||||
Text map[string]interface{} `json:"text,omitempty"`
|
||||
Temperature float64 `json:"temperature,omitempty" example:"0.7"`
|
||||
TopP float64 `json:"top_p,omitempty" example:"1"`
|
||||
Store *bool `json:"store,omitempty"`
|
||||
Stream bool `json:"stream,omitempty" example:"false"`
|
||||
}
|
||||
|
||||
type ResponsesCompatibleResponse struct {
|
||||
ID string `json:"id" example:"resp_0123456789abcdef0123456789abcdef"`
|
||||
Object string `json:"object" example:"response"`
|
||||
CreatedAt int64 `json:"created_at" example:"1710000000"`
|
||||
Status string `json:"status" example:"completed"`
|
||||
Model string `json:"model" example:"Doubao Seed 2.0 Pro"`
|
||||
PreviousResponseID string `json:"previous_response_id,omitempty" example:"resp_abcdef0123456789abcdef0123456789"`
|
||||
Output []map[string]interface{} `json:"output"`
|
||||
OutputText string `json:"output_text,omitempty" example:"Hello"`
|
||||
Usage map[string]interface{} `json:"usage,omitempty"`
|
||||
}
|
||||
|
||||
type ImageGenerationRequest struct {
|
||||
|
||||
@@ -0,0 +1,36 @@
|
||||
package httpapi
|
||||
|
||||
import (
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/easyai/easyai-ai-gateway/apps/api/internal/clients"
|
||||
)
|
||||
|
||||
func TestResponsesStreamWriterForwardsStandardEventsAndNeverWritesDoneMarker(t *testing.T) {
|
||||
recorder := httptest.NewRecorder()
|
||||
writer := newCompatibleStreamWriter("responses", "demo", true)
|
||||
writer.writeDelta(recorder, clients.StreamDeltaEvent{Event: map[string]any{
|
||||
"type": "response.created", "response": map[string]any{"id": "resp_123", "status": "in_progress"},
|
||||
}})
|
||||
writer.writeDelta(recorder, clients.StreamDeltaEvent{Event: map[string]any{
|
||||
"type": "response.function_call_arguments.delta", "item_id": "fc_1", "delta": "{\"x\":",
|
||||
}})
|
||||
writer.writeDelta(recorder, clients.StreamDeltaEvent{Event: map[string]any{
|
||||
"type": "response.completed", "sequence_number": 2, "response": map[string]any{"id": "resp_123", "status": "completed", "output": []any{}},
|
||||
}})
|
||||
writer.writeDone(recorder, map[string]any{"id": "resp_123", "object": "response", "status": "completed", "output": []any{}})
|
||||
body := recorder.Body.String()
|
||||
for _, expected := range []string{"event: response.created", "event: response.function_call_arguments.delta", "event: response.completed"} {
|
||||
if !strings.Contains(body, expected) {
|
||||
t.Fatalf("missing %q in stream: %s", expected, body)
|
||||
}
|
||||
}
|
||||
if strings.Contains(body, "[DONE]") || strings.Contains(body, "chat.completion") {
|
||||
t.Fatalf("Responses stream leaked Chat framing: %s", body)
|
||||
}
|
||||
if strings.Count(body, "event: response.completed") != 1 {
|
||||
t.Fatalf("Responses stream duplicated response.completed: %s", body)
|
||||
}
|
||||
}
|
||||
@@ -28,6 +28,8 @@ type compatibleStreamWriter struct {
|
||||
sentRole bool
|
||||
sentFinish bool
|
||||
sentUsage bool
|
||||
responseSequence int
|
||||
sentResponseDone bool
|
||||
}
|
||||
|
||||
func newCompatibleStreamWriter(kind string, model string, includeUsage bool) *compatibleStreamWriter {
|
||||
@@ -42,6 +44,19 @@ func newCompatibleStreamWriter(kind string, model string, includeUsage bool) *co
|
||||
|
||||
func (s *compatibleStreamWriter) writeDelta(w http.ResponseWriter, event clients.StreamDeltaEvent) {
|
||||
if s.kind == "responses" {
|
||||
if event.Event != nil {
|
||||
eventType, _ := event.Event["type"].(string)
|
||||
if eventType != "" {
|
||||
if sequence := intFromStreamValue(event.Event["sequence_number"]); sequence >= s.responseSequence {
|
||||
s.responseSequence = sequence + 1
|
||||
}
|
||||
if eventType == "response.completed" {
|
||||
s.sentResponseDone = true
|
||||
}
|
||||
sendSSE(w, eventType, event.Event)
|
||||
return
|
||||
}
|
||||
}
|
||||
if event.Text != "" {
|
||||
sendSSE(w, "response.output_text.delta", map[string]any{"type": "response.output_text.delta", "delta": event.Text})
|
||||
}
|
||||
@@ -67,7 +82,11 @@ func (s *compatibleStreamWriter) writeDelta(w http.ResponseWriter, event clients
|
||||
|
||||
func (s *compatibleStreamWriter) writeDone(w http.ResponseWriter, output map[string]any) {
|
||||
if s.kind == "responses" {
|
||||
sendSSE(w, "response.completed", map[string]any{"type": "response.completed", "response": output})
|
||||
if s.sentResponseDone {
|
||||
return
|
||||
}
|
||||
sendSSE(w, "response.completed", map[string]any{"type": "response.completed", "sequence_number": s.responseSequence, "response": output})
|
||||
s.sentResponseDone = true
|
||||
return
|
||||
}
|
||||
s.captureOutputMetadata(output)
|
||||
@@ -88,6 +107,19 @@ func (s *compatibleStreamWriter) writeDone(w http.ResponseWriter, output map[str
|
||||
s.writeDoneMarker(w)
|
||||
}
|
||||
|
||||
func intFromStreamValue(value any) int {
|
||||
switch typed := value.(type) {
|
||||
case int:
|
||||
return typed
|
||||
case int64:
|
||||
return int(typed)
|
||||
case float64:
|
||||
return int(typed)
|
||||
default:
|
||||
return -1
|
||||
}
|
||||
}
|
||||
|
||||
func (s *compatibleStreamWriter) writeChatChunk(w http.ResponseWriter, chunk map[string]any) {
|
||||
chunk = clients.NormalizeChatCompletionStreamEvent(chunk)
|
||||
s.captureChunkMetadata(chunk)
|
||||
|
||||
@@ -70,6 +70,15 @@ func taskMetrics(task store.GatewayTask, user *auth.User, body map[string]any, c
|
||||
if response.ResponseDurationMS > 0 {
|
||||
metrics["responseDurationMs"] = response.ResponseDurationMS
|
||||
}
|
||||
if response.UpstreamProtocol != "" {
|
||||
metrics["upstreamProtocol"] = response.UpstreamProtocol
|
||||
metrics["upstreamEndpoint"] = response.UpstreamEndpoint
|
||||
metrics["responseConverted"] = response.ResponseConverted
|
||||
metrics["publicResponseId"] = response.PublicResponseID
|
||||
metrics["upstreamResponseId"] = response.UpstreamResponseID
|
||||
metrics["parentResponseId"] = response.ParentResponseID
|
||||
metrics["responseChainDepth"] = response.ResponseChainDepth
|
||||
}
|
||||
switch task.Kind {
|
||||
case "chat.completions", "responses":
|
||||
metrics["stream"] = boolFromMap(body, "stream")
|
||||
@@ -119,6 +128,16 @@ func attemptMetrics(candidate store.RuntimeModelCandidate, attemptNo int, simula
|
||||
"loadAvoided": candidate.LoadAvoided,
|
||||
"simulated": simulated,
|
||||
}
|
||||
if candidate.ResponseProtocol != "" {
|
||||
metrics["upstreamProtocol"] = candidate.ResponseProtocol
|
||||
if candidate.ResponseProtocol == clients.ProtocolOpenAIResponses {
|
||||
metrics["upstreamEndpoint"] = "/responses"
|
||||
metrics["responseConverted"] = false
|
||||
} else if candidate.ResponseProtocol == clients.ProtocolOpenAIChatCompletions {
|
||||
metrics["upstreamEndpoint"] = "/chat/completions"
|
||||
metrics["responseConverted"] = true
|
||||
}
|
||||
}
|
||||
if candidate.LoadLimited {
|
||||
metrics["loadMetrics"] = map[string]any{
|
||||
"rpm": map[string]any{
|
||||
|
||||
@@ -0,0 +1,256 @@
|
||||
package runner
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/easyai/easyai-ai-gateway/apps/api/internal/clients"
|
||||
"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
|
||||
)
|
||||
|
||||
const (
|
||||
responseChainMaxTTL = 30 * 24 * time.Hour
|
||||
responseChainMaxDepth = 100
|
||||
)
|
||||
|
||||
type responseExecutionContext struct {
|
||||
PublicResponseID string
|
||||
PublicPreviousResponseID string
|
||||
UpstreamPreviousResponseID string
|
||||
Protocol string
|
||||
Store bool
|
||||
PreviousChain *store.ResponseChain
|
||||
PreviousTurns []clients.ResponseTurn
|
||||
ChainDepth int
|
||||
}
|
||||
|
||||
func (s *Service) prepareResponseExecution(ctx context.Context, task store.GatewayTask, body map[string]any) (responseExecutionContext, error) {
|
||||
result := responseExecutionContext{
|
||||
PublicResponseID: gatewayResponseID(task.ID),
|
||||
Store: true,
|
||||
}
|
||||
if value, ok := body["store"].(bool); ok {
|
||||
result.Store = value
|
||||
}
|
||||
previousID := strings.TrimSpace(stringFromAny(body["previous_response_id"]))
|
||||
if previousID == "" {
|
||||
// No Gateway response chain was requested. The caller owns the complete
|
||||
// conversation state in input/messages, so it must be forwarded as-is.
|
||||
return result, nil
|
||||
}
|
||||
chain, err := s.store.GetResponseChain(ctx, previousID, task)
|
||||
if err != nil {
|
||||
return responseExecutionContext{}, &clients.ClientError{
|
||||
Code: "invalid_previous_response_id", Message: "previous_response_id is invalid or expired", StatusCode: http.StatusBadRequest,
|
||||
}
|
||||
}
|
||||
if !sameResponseModel(chain.RequestedModel, task.Model) {
|
||||
return responseExecutionContext{}, &clients.ClientError{
|
||||
Code: "invalid_previous_response_id", Message: "previous_response_id belongs to a different model", StatusCode: http.StatusBadRequest,
|
||||
}
|
||||
}
|
||||
history, historyErr := s.store.ListResponseChainHistory(ctx, chain)
|
||||
if historyErr != nil {
|
||||
if errors.Is(historyErr, store.ErrResponseChainTooDeep) {
|
||||
return responseExecutionContext{}, &clients.ClientError{Code: "response_chain_too_deep", Message: historyErr.Error(), StatusCode: http.StatusBadRequest}
|
||||
}
|
||||
return responseExecutionContext{}, historyErr
|
||||
}
|
||||
chainDepth, depthErr := responseHistoryDepth(history)
|
||||
if depthErr != nil {
|
||||
return responseExecutionContext{}, depthErr
|
||||
}
|
||||
result.PublicPreviousResponseID = chain.PublicResponseID
|
||||
result.UpstreamPreviousResponseID = chain.UpstreamResponseID
|
||||
result.Protocol = chain.UpstreamProtocol
|
||||
result.PreviousChain = &chain
|
||||
result.ChainDepth = chainDepth
|
||||
if chain.UpstreamProtocol == clients.ProtocolOpenAIChatCompletions {
|
||||
// Chat upstreams are stateless from the Responses API perspective. Only
|
||||
// this Gateway-managed mode replays stored turns; native Responses keeps
|
||||
// the provider's previous_response_id state authoritative.
|
||||
result.PreviousTurns = make([]clients.ResponseTurn, 0, len(history))
|
||||
for _, turn := range history {
|
||||
result.PreviousTurns = append(result.PreviousTurns, clients.ResponseTurn{
|
||||
Request: turn.RequestSnapshot, Response: turn.ResponseSnapshot, Internal: turn.InternalSnapshot,
|
||||
})
|
||||
}
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func prepareResponseCandidates(candidates []store.RuntimeModelCandidate, execution responseExecutionContext) ([]store.RuntimeModelCandidate, error) {
|
||||
if execution.PreviousChain != nil {
|
||||
for _, candidate := range candidates {
|
||||
if candidate.PlatformModelID != execution.PreviousChain.PlatformModelID {
|
||||
continue
|
||||
}
|
||||
if !candidateSupportsProtocol(candidate, execution.PreviousChain.UpstreamProtocol) {
|
||||
break
|
||||
}
|
||||
candidate.ResponseProtocol = execution.PreviousChain.UpstreamProtocol
|
||||
return []store.RuntimeModelCandidate{candidate}, nil
|
||||
}
|
||||
return nil, responseChainUnavailableError()
|
||||
}
|
||||
|
||||
type indexedCandidate struct {
|
||||
candidate store.RuntimeModelCandidate
|
||||
index int
|
||||
group int
|
||||
}
|
||||
items := make([]indexedCandidate, 0, len(candidates))
|
||||
hasAnthropicOnly := false
|
||||
hasDeclaredUnsupported := false
|
||||
for index, candidate := range candidates {
|
||||
protocols := candidateSupportedProtocols(candidate)
|
||||
group := 1
|
||||
protocol := clients.ProtocolOpenAIChatCompletions
|
||||
if containsString(protocols, clients.ProtocolOpenAIResponses) {
|
||||
group = 0
|
||||
protocol = clients.ProtocolOpenAIResponses
|
||||
} else if !containsString(protocols, clients.ProtocolOpenAIChatCompletions) {
|
||||
hasDeclaredUnsupported = true
|
||||
if containsString(protocols, clients.ProtocolAnthropicMessages) {
|
||||
hasAnthropicOnly = true
|
||||
}
|
||||
continue
|
||||
}
|
||||
candidate.ResponseProtocol = protocol
|
||||
items = append(items, indexedCandidate{candidate: candidate, index: index, group: group})
|
||||
}
|
||||
if len(items) == 0 {
|
||||
if hasAnthropicOnly {
|
||||
return nil, &clients.ClientError{Code: "unsupported_model_protocol", Message: "the selected model only supports Anthropic Messages; the public Anthropic adapter is not available", StatusCode: http.StatusBadRequest}
|
||||
}
|
||||
if hasDeclaredUnsupported {
|
||||
return nil, &clients.ClientError{Code: "unsupported_model_protocol", Message: "the selected model does not declare an OpenAI-compatible protocol", StatusCode: http.StatusBadRequest}
|
||||
}
|
||||
return nil, responseChainUnavailableError()
|
||||
}
|
||||
sort.SliceStable(items, func(i, j int) bool {
|
||||
if items[i].group != items[j].group {
|
||||
return items[i].group < items[j].group
|
||||
}
|
||||
return items[i].index < items[j].index
|
||||
})
|
||||
out := make([]store.RuntimeModelCandidate, 0, len(items))
|
||||
for _, item := range items {
|
||||
out = append(out, item.candidate)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func candidateSupportedProtocols(candidate store.RuntimeModelCandidate) []string {
|
||||
capability := effectiveModelCapability(candidate)
|
||||
textCapability, _ := capability["text_generate"].(map[string]any)
|
||||
value, present := textCapability["supportedApiProtocols"]
|
||||
if !present {
|
||||
value, present = capability["supportedApiProtocols"]
|
||||
}
|
||||
if !present {
|
||||
return []string{clients.ProtocolOpenAIChatCompletions}
|
||||
}
|
||||
protocols := stringListFromAny(value)
|
||||
out := make([]string, 0, len(protocols))
|
||||
for _, protocol := range protocols {
|
||||
switch protocol {
|
||||
case clients.ProtocolOpenAIChatCompletions, clients.ProtocolOpenAIResponses, clients.ProtocolAnthropicMessages:
|
||||
out = append(out, protocol)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func candidateSupportsProtocol(candidate store.RuntimeModelCandidate, protocol string) bool {
|
||||
return containsString(candidateSupportedProtocols(candidate), protocol)
|
||||
}
|
||||
|
||||
func responseChainUnavailableError() error {
|
||||
return &clients.ClientError{
|
||||
Code: "response_chain_unavailable", Message: "the platform model pinned to this response chain is unavailable", StatusCode: http.StatusServiceUnavailable,
|
||||
}
|
||||
}
|
||||
|
||||
func gatewayResponseID(taskID string) string {
|
||||
return "resp_" + strings.ReplaceAll(strings.TrimSpace(taskID), "-", "")
|
||||
}
|
||||
|
||||
func (s *Service) persistResponseChain(ctx context.Context, task store.GatewayTask, body map[string]any, candidate store.RuntimeModelCandidate, execution responseExecutionContext, response clients.Response) error {
|
||||
if task.Kind != "responses" || !execution.Store {
|
||||
return nil
|
||||
}
|
||||
expiresAt := responseChainExpiry(response.Result, candidate.ResponseProtocol)
|
||||
publicResponseID := strings.TrimSpace(response.PublicResponseID)
|
||||
if publicResponseID == "" {
|
||||
publicResponseID = strings.TrimSpace(stringFromAny(response.Result["id"]))
|
||||
}
|
||||
if publicResponseID == "" {
|
||||
return &clients.ClientError{Code: "invalid_response", Message: "Responses result is missing id", StatusCode: http.StatusBadGateway}
|
||||
}
|
||||
return s.store.CreateResponseChain(ctx, store.CreateResponseChainInput{ResponseChain: store.ResponseChain{
|
||||
PublicResponseID: publicResponseID,
|
||||
ParentResponseID: execution.PublicPreviousResponseID,
|
||||
GatewayTaskID: task.ID,
|
||||
GatewayUserID: task.GatewayUserID,
|
||||
UserID: task.UserID,
|
||||
UserSource: task.UserSource,
|
||||
GatewayTenantID: task.GatewayTenantID,
|
||||
TenantID: task.TenantID,
|
||||
TenantKey: task.TenantKey,
|
||||
APIKeyID: task.APIKeyID,
|
||||
PlatformID: candidate.PlatformID,
|
||||
PlatformModelID: candidate.PlatformModelID,
|
||||
ClientID: candidate.ClientID,
|
||||
UpstreamProtocol: candidate.ResponseProtocol,
|
||||
UpstreamEndpoint: response.UpstreamEndpoint,
|
||||
UpstreamResponseID: response.UpstreamResponseID,
|
||||
RequestSnapshot: cloneMap(body),
|
||||
ResponseSnapshot: cloneMap(response.Result),
|
||||
InternalSnapshot: cloneMap(response.InternalResult),
|
||||
ExpiresAt: expiresAt,
|
||||
}})
|
||||
}
|
||||
|
||||
func sameResponseModel(previous string, current string) bool {
|
||||
return strings.EqualFold(strings.TrimSpace(previous), strings.TrimSpace(current))
|
||||
}
|
||||
|
||||
func responseHistoryDepth(history []store.ResponseChain) (int, error) {
|
||||
if len(history) >= responseChainMaxDepth {
|
||||
return 0, &clients.ClientError{Code: "response_chain_too_deep", Message: "response chain exceeds the maximum depth", StatusCode: http.StatusBadRequest}
|
||||
}
|
||||
return len(history), nil
|
||||
}
|
||||
|
||||
func responseChainExpiry(result map[string]any, protocol string) time.Time {
|
||||
now := time.Now()
|
||||
maximum := now.Add(responseChainMaxTTL)
|
||||
if protocol != clients.ProtocolOpenAIResponses {
|
||||
return maximum
|
||||
}
|
||||
var parsed time.Time
|
||||
switch value := result["expire_at"].(type) {
|
||||
case float64:
|
||||
parsed = time.Unix(int64(value), 0)
|
||||
case int64:
|
||||
parsed = time.Unix(value, 0)
|
||||
case string:
|
||||
parsed, _ = time.Parse(time.RFC3339, value)
|
||||
}
|
||||
if parsed.IsZero() || parsed.After(maximum) {
|
||||
return maximum
|
||||
}
|
||||
return parsed
|
||||
}
|
||||
|
||||
func responseExecutionFailure(err error) (string, string) {
|
||||
if errors.Is(err, store.ErrInvalidPreviousResponseID) {
|
||||
return "invalid_previous_response_id", "previous_response_id is invalid or expired"
|
||||
}
|
||||
return clients.ErrorCode(err), err.Error()
|
||||
}
|
||||
@@ -0,0 +1,129 @@
|
||||
package runner
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/easyai/easyai-ai-gateway/apps/api/internal/clients"
|
||||
"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
|
||||
)
|
||||
|
||||
func TestCandidateSupportedProtocolsDefaultsToChat(t *testing.T) {
|
||||
candidate := store.RuntimeModelCandidate{Capabilities: map[string]any{"text_generate": map[string]any{"supportTool": true}}}
|
||||
protocols := candidateSupportedProtocols(candidate)
|
||||
if len(protocols) != 1 || protocols[0] != clients.ProtocolOpenAIChatCompletions {
|
||||
t.Fatalf("legacy capability must default to Chat: %+v", protocols)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPrepareResponseCandidatesPrioritizesNativeAndKeepsGroupOrder(t *testing.T) {
|
||||
candidates := []store.RuntimeModelCandidate{
|
||||
{PlatformModelID: "chat-1", Capabilities: responseProtocolCapability(clients.ProtocolOpenAIChatCompletions)},
|
||||
{PlatformModelID: "native-1", Capabilities: responseProtocolCapability(clients.ProtocolOpenAIChatCompletions, clients.ProtocolOpenAIResponses)},
|
||||
{PlatformModelID: "native-2", Capabilities: responseProtocolCapability(clients.ProtocolOpenAIResponses)},
|
||||
{PlatformModelID: "chat-2", Capabilities: responseProtocolCapability(clients.ProtocolOpenAIChatCompletions)},
|
||||
}
|
||||
prepared, err := prepareResponseCandidates(candidates, responseExecutionContext{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
got := []string{prepared[0].PlatformModelID, prepared[1].PlatformModelID, prepared[2].PlatformModelID, prepared[3].PlatformModelID}
|
||||
want := []string{"native-1", "native-2", "chat-1", "chat-2"}
|
||||
for index := range want {
|
||||
if got[index] != want[index] {
|
||||
t.Fatalf("unexpected protocol ordering got=%v want=%v", got, want)
|
||||
}
|
||||
}
|
||||
if prepared[0].ResponseProtocol != clients.ProtocolOpenAIResponses || prepared[2].ResponseProtocol != clients.ProtocolOpenAIChatCompletions {
|
||||
t.Fatalf("protocol assignment mismatch: %+v", prepared)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPrepareResponseCandidatesPinsPreviousPlatformModelAndProtocol(t *testing.T) {
|
||||
chain := store.ResponseChain{PlatformModelID: "pinned", UpstreamProtocol: clients.ProtocolOpenAIChatCompletions}
|
||||
candidates := []store.RuntimeModelCandidate{
|
||||
{PlatformModelID: "other", Capabilities: responseProtocolCapability(clients.ProtocolOpenAIChatCompletions)},
|
||||
{PlatformModelID: "pinned", Capabilities: responseProtocolCapability(clients.ProtocolOpenAIChatCompletions, clients.ProtocolOpenAIResponses)},
|
||||
}
|
||||
prepared, err := prepareResponseCandidates(candidates, responseExecutionContext{PreviousChain: &chain})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(prepared) != 1 || prepared[0].PlatformModelID != "pinned" || prepared[0].ResponseProtocol != clients.ProtocolOpenAIChatCompletions {
|
||||
t.Fatalf("previous chain was not pinned: %+v", prepared)
|
||||
}
|
||||
|
||||
_, err = prepareResponseCandidates(candidates[:1], responseExecutionContext{PreviousChain: &chain})
|
||||
if clients.ErrorCode(err) != "response_chain_unavailable" {
|
||||
t.Fatalf("expected response_chain_unavailable, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPrepareResponseCandidatesRejectsAnthropicOnly(t *testing.T) {
|
||||
_, err := prepareResponseCandidates([]store.RuntimeModelCandidate{{Capabilities: responseProtocolCapability(clients.ProtocolAnthropicMessages)}}, responseExecutionContext{})
|
||||
if clients.ErrorCode(err) != "unsupported_model_protocol" {
|
||||
t.Fatalf("expected unsupported_model_protocol, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGatewayResponseIDAndExpiryPolicy(t *testing.T) {
|
||||
id := gatewayResponseID("01234567-89ab-cdef-0123-456789abcdef")
|
||||
if id != "resp_0123456789abcdef0123456789abcdef" {
|
||||
t.Fatalf("unexpected public response id: %s", id)
|
||||
}
|
||||
now := time.Now()
|
||||
chatExpiry := responseChainExpiry(map[string]any{}, clients.ProtocolOpenAIChatCompletions)
|
||||
if chatExpiry.Before(now.Add(responseChainMaxTTL-time.Minute)) || chatExpiry.After(now.Add(responseChainMaxTTL+time.Minute)) {
|
||||
t.Fatalf("Chat chain must use the 30-day retention policy: %v", chatExpiry)
|
||||
}
|
||||
upstreamExpiry := now.Add(24 * time.Hour).Unix()
|
||||
nativeExpiry := responseChainExpiry(map[string]any{"expire_at": float64(upstreamExpiry)}, clients.ProtocolOpenAIResponses)
|
||||
if nativeExpiry.Unix() != upstreamExpiry {
|
||||
t.Fatalf("native chain must honor an earlier upstream expiry: %v", nativeExpiry)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResponseModelAndDepthValidation(t *testing.T) {
|
||||
if !sameResponseModel("DeepSeek-V4-Pro", "deepseek-v4-pro") || sameResponseModel("Qwen3.7-Plus", "DeepSeek-V4-Pro") {
|
||||
t.Fatal("response continuation model matching is not strict")
|
||||
}
|
||||
history := make([]store.ResponseChain, responseChainMaxDepth)
|
||||
if _, err := responseHistoryDepth(history); clients.ErrorCode(err) != "response_chain_too_deep" {
|
||||
t.Fatalf("expected response_chain_too_deep, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCandidateProtocolOverrideAndExplicitEmpty(t *testing.T) {
|
||||
candidate := store.RuntimeModelCandidate{
|
||||
Capabilities: responseProtocolCapability(clients.ProtocolOpenAIChatCompletions, clients.ProtocolOpenAIResponses),
|
||||
CapabilityOverride: responseProtocolCapability(clients.ProtocolAnthropicMessages),
|
||||
}
|
||||
protocols := candidateSupportedProtocols(candidate)
|
||||
if len(protocols) != 1 || protocols[0] != clients.ProtocolAnthropicMessages {
|
||||
t.Fatalf("platform capability override was not applied: %v", protocols)
|
||||
}
|
||||
empty := store.RuntimeModelCandidate{Capabilities: responseProtocolCapability()}
|
||||
if protocols := candidateSupportedProtocols(empty); len(protocols) != 0 {
|
||||
t.Fatalf("an explicitly empty protocol list must not become legacy Chat: %v", protocols)
|
||||
}
|
||||
if _, err := prepareResponseCandidates([]store.RuntimeModelCandidate{empty}, responseExecutionContext{}); clients.ErrorCode(err) != "unsupported_model_protocol" {
|
||||
t.Fatalf("expected unsupported_model_protocol, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPersistResponseChainSkipsStoreFalse(t *testing.T) {
|
||||
service := &Service{}
|
||||
err := service.persistResponseChain(context.Background(), store.GatewayTask{Kind: "responses"}, nil, store.RuntimeModelCandidate{}, responseExecutionContext{Store: false}, clients.Response{})
|
||||
if err != nil {
|
||||
t.Fatalf("store:false must not require a response-chain store: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func responseProtocolCapability(protocols ...string) map[string]any {
|
||||
items := make([]any, len(protocols))
|
||||
for index, protocol := range protocols {
|
||||
items[index] = protocol
|
||||
}
|
||||
return map[string]any{"text_generate": map[string]any{"supportedApiProtocols": items}}
|
||||
}
|
||||
@@ -93,6 +93,7 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
|
||||
return Result{}, err
|
||||
}
|
||||
body := normalizeRequest(task.Kind, restoredRequest)
|
||||
responseExecution := responseExecutionContext{}
|
||||
modelType := modelTypeFromKind(task.Kind, body)
|
||||
if err := s.store.MarkTaskRunning(ctx, task.ID, modelType, s.slimTaskRequestSnapshot(task, body)); err != nil {
|
||||
return Result{}, err
|
||||
@@ -145,6 +146,18 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
|
||||
return Result{}, err
|
||||
}
|
||||
}
|
||||
if task.Kind == "responses" {
|
||||
responseExecution, err = s.prepareResponseExecution(ctx, task, body)
|
||||
if err != nil {
|
||||
code, message := responseExecutionFailure(err)
|
||||
s.recordFailedAttempt(ctx, failedAttemptRecord{Task: task, Body: body, AttemptNo: task.AttemptCount + 1, Code: code, Cause: err, Simulated: task.RunMode == "simulation", Scope: "response_chain", Reason: code, ModelType: modelType})
|
||||
failed, finishErr := s.failTask(ctx, task.ID, code, message, task.RunMode == "simulation", err)
|
||||
if finishErr != nil {
|
||||
return Result{}, finishErr
|
||||
}
|
||||
return Result{Task: failed, Output: failed.Result}, err
|
||||
}
|
||||
}
|
||||
runnerPolicy, err := s.store.GetActiveRunnerPolicy(ctx)
|
||||
if err != nil {
|
||||
return Result{}, err
|
||||
@@ -156,6 +169,9 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
|
||||
CacheAffinityPolicy: runnerPolicy.CacheAffinityPolicy,
|
||||
})
|
||||
if err != nil {
|
||||
if task.Kind == "responses" && responseExecution.PreviousChain != nil {
|
||||
err = responseChainUnavailableError()
|
||||
}
|
||||
s.recordFailedAttempt(ctx, failedAttemptRecord{
|
||||
Task: task,
|
||||
Body: body,
|
||||
@@ -214,6 +230,18 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
|
||||
}
|
||||
return Result{Task: failed, Output: failed.Result}, err
|
||||
}
|
||||
if task.Kind == "responses" {
|
||||
candidates, err = prepareResponseCandidates(candidates, responseExecution)
|
||||
if err != nil {
|
||||
code, message := responseExecutionFailure(err)
|
||||
s.recordFailedAttempt(ctx, failedAttemptRecord{Task: task, Body: body, AttemptNo: task.AttemptCount + 1, Code: code, Cause: err, Simulated: task.RunMode == "simulation", Scope: "response_protocol", Reason: code, ModelType: modelType})
|
||||
failed, finishErr := s.failTask(ctx, task.ID, code, message, task.RunMode == "simulation", err)
|
||||
if finishErr != nil {
|
||||
return Result{}, finishErr
|
||||
}
|
||||
return Result{Task: failed, Output: failed.Result}, err
|
||||
}
|
||||
}
|
||||
firstCandidateBody := body
|
||||
normalizedModelType := modelType
|
||||
attemptNo := task.AttemptCount
|
||||
@@ -325,7 +353,7 @@ candidatesLoop:
|
||||
break candidatesLoop
|
||||
}
|
||||
candidateBody := preprocessing.Body
|
||||
response, err := s.runCandidate(ctx, task, user, candidateBody, preprocessing.Log, candidate, nextAttemptNo, onDelta, singleSourceProtected, runnerPolicy.CacheAffinityPolicy, cacheAffinityKeys.Record)
|
||||
response, err := s.runCandidate(ctx, task, user, candidateBody, preprocessing.Log, candidate, nextAttemptNo, onDelta, responseExecution, singleSourceProtected, runnerPolicy.CacheAffinityPolicy, cacheAffinityKeys.Record)
|
||||
if err == nil {
|
||||
attemptNo = nextAttemptNo
|
||||
billings := s.billings(ctx, user, task.Kind, candidateBody, candidate, response, isSimulation(task, candidate))
|
||||
@@ -536,7 +564,7 @@ candidatesLoop:
|
||||
return Result{Task: failed, Output: failed.Result}, lastErr
|
||||
}
|
||||
|
||||
func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user *auth.User, body map[string]any, preprocessing parameterPreprocessingLog, candidate store.RuntimeModelCandidate, attemptNo int, onDelta clients.StreamDelta, singleSourceProtected bool, cacheAffinityPolicy map[string]any, cacheAffinityRecordKeys []string) (clients.Response, error) {
|
||||
func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user *auth.User, body map[string]any, preprocessing parameterPreprocessingLog, candidate store.RuntimeModelCandidate, attemptNo int, onDelta clients.StreamDelta, responseExecution responseExecutionContext, singleSourceProtected bool, cacheAffinityPolicy map[string]any, cacheAffinityRecordKeys []string) (clients.Response, error) {
|
||||
simulated := isSimulation(task, candidate)
|
||||
baseAttemptMetrics := mergeMetrics(attemptMetrics(candidate, attemptNo, simulated), parameterPreprocessingMetrics(preprocessing))
|
||||
reservations := s.rateLimitReservations(ctx, user, candidate, body)
|
||||
@@ -628,6 +656,12 @@ func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user
|
||||
return clients.Response{}, err
|
||||
}
|
||||
callStartedAt := time.Now()
|
||||
publicResponseID := ""
|
||||
publicPreviousResponseID := ""
|
||||
if task.Kind == "responses" && candidate.ResponseProtocol == clients.ProtocolOpenAIChatCompletions {
|
||||
publicResponseID = responseExecution.PublicResponseID
|
||||
publicPreviousResponseID = responseExecution.PublicPreviousResponseID
|
||||
}
|
||||
response, err := client.Run(ctx, clients.Request{
|
||||
Kind: task.Kind,
|
||||
ModelType: candidate.ModelType,
|
||||
@@ -643,8 +677,13 @@ func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user
|
||||
}
|
||||
return s.store.SetTaskRemoteTask(context.WithoutCancel(ctx), task.ID, attemptID, remoteTaskID, payload)
|
||||
},
|
||||
Stream: boolFromMap(providerBody, "stream"),
|
||||
StreamDelta: onDelta,
|
||||
Stream: boolFromMap(providerBody, "stream"),
|
||||
StreamDelta: onDelta,
|
||||
UpstreamProtocol: candidate.ResponseProtocol,
|
||||
PublicResponseID: publicResponseID,
|
||||
PublicPreviousResponseID: publicPreviousResponseID,
|
||||
UpstreamPreviousResponseID: responseExecution.UpstreamPreviousResponseID,
|
||||
PreviousResponseTurns: responseExecution.PreviousTurns,
|
||||
})
|
||||
callFinishedAt := time.Now()
|
||||
if response.ResponseStartedAt.IsZero() {
|
||||
@@ -715,6 +754,20 @@ func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user
|
||||
return clients.Response{}, err
|
||||
}
|
||||
response.Result = uploadedResult
|
||||
if task.Kind == "responses" {
|
||||
response.UpstreamProtocol = candidate.ResponseProtocol
|
||||
response.ParentResponseID = responseExecution.PublicPreviousResponseID
|
||||
response.ResponseChainDepth = responseExecution.ChainDepth
|
||||
if candidate.ResponseProtocol == clients.ProtocolOpenAIChatCompletions {
|
||||
response.Result["id"] = responseExecution.PublicResponseID
|
||||
response.PublicResponseID = responseExecution.PublicResponseID
|
||||
} else {
|
||||
response.PublicResponseID = strings.TrimSpace(stringFromAny(response.Result["id"]))
|
||||
}
|
||||
if err := s.persistResponseChain(ctx, task, body, candidate, responseExecution, response); err != nil {
|
||||
return clients.Response{}, fmt.Errorf("persist response chain: %w", err)
|
||||
}
|
||||
}
|
||||
if task.Kind == "voice.clone" {
|
||||
voice, err := s.persistVoiceCloneResult(ctx, task, user, candidate, attemptID, body, response.Result)
|
||||
if err != nil {
|
||||
@@ -1283,9 +1336,6 @@ func loadAvoidanceFallbackDecision(err error) failoverDecision {
|
||||
|
||||
func normalizeRequest(kind string, body map[string]any) map[string]any {
|
||||
out := cloneMap(body)
|
||||
if kind == "responses" && out["messages"] == nil && out["input"] != nil {
|
||||
out["messages"] = []any{map[string]any{"role": "user", "content": out["input"]}}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
|
||||
@@ -31,7 +31,18 @@ func (s *Store) ListModelCandidates(ctx context.Context, model string, modelType
|
||||
COALESCE(p.dynamic_priority, p.priority) AS effective_priority,
|
||||
m.id::text, COALESCE(m.base_model_id::text, ''), COALESCE(b.canonical_model_key, ''),
|
||||
COALESCE(NULLIF(m.provider_model_name, ''), m.model_name), m.model_name, COALESCE(m.model_alias, ''),
|
||||
$2::text AS requested_model_type, m.display_name, m.capabilities, m.capability_override,
|
||||
$2::text AS requested_model_type, m.display_name,
|
||||
CASE
|
||||
WHEN b.capabilities #> '{text_generate,supportedApiProtocols}' IS NOT NULL
|
||||
THEN jsonb_set(
|
||||
COALESCE(m.capabilities, '{}'::jsonb),
|
||||
'{text_generate,supportedApiProtocols}',
|
||||
b.capabilities #> '{text_generate,supportedApiProtocols}',
|
||||
true
|
||||
)
|
||||
ELSE m.capabilities
|
||||
END AS effective_capabilities,
|
||||
m.capability_override,
|
||||
COALESCE(b.base_billing_config, '{}'::jsonb), m.billing_config, m.billing_config_override,
|
||||
m.pricing_mode, COALESCE(m.discount_factor, 0)::float8, COALESCE(m.pricing_rule_set_id::text, ''),
|
||||
COALESCE(b.pricing_rule_set_id::text, ''),
|
||||
|
||||
@@ -0,0 +1,183 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
var ErrInvalidPreviousResponseID = errors.New("invalid previous response id")
|
||||
var ErrResponseChainTooDeep = errors.New("response chain exceeds the maximum depth")
|
||||
|
||||
const maxResponseChainHistory = 100
|
||||
|
||||
type ResponseChain struct {
|
||||
PublicResponseID string
|
||||
ParentResponseID string
|
||||
GatewayTaskID string
|
||||
RequestedModel string
|
||||
GatewayUserID string
|
||||
UserID string
|
||||
UserSource string
|
||||
GatewayTenantID string
|
||||
TenantID string
|
||||
TenantKey string
|
||||
APIKeyID string
|
||||
PlatformID string
|
||||
PlatformModelID string
|
||||
ClientID string
|
||||
UpstreamProtocol string
|
||||
UpstreamEndpoint string
|
||||
UpstreamResponseID string
|
||||
RequestSnapshot map[string]any
|
||||
ResponseSnapshot map[string]any
|
||||
InternalSnapshot map[string]any
|
||||
CreatedAt time.Time
|
||||
ExpiresAt time.Time
|
||||
}
|
||||
|
||||
type CreateResponseChainInput struct {
|
||||
ResponseChain
|
||||
}
|
||||
|
||||
func (s *Store) CreateResponseChain(ctx context.Context, input CreateResponseChainInput) error {
|
||||
requestJSON, err := json.Marshal(input.RequestSnapshot)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
responseJSON, err := json.Marshal(input.ResponseSnapshot)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
internalJSON, err := json.Marshal(input.InternalSnapshot)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = s.pool.Exec(ctx, `
|
||||
INSERT INTO gateway_response_chains (
|
||||
public_response_id, parent_response_id, gateway_task_id,
|
||||
gateway_user_id, user_id, user_source, gateway_tenant_id, tenant_id, tenant_key, api_key_id,
|
||||
platform_id, platform_model_id, client_id, upstream_protocol, upstream_endpoint, upstream_response_id,
|
||||
request_snapshot, response_snapshot, internal_snapshot, expires_at
|
||||
) VALUES (
|
||||
$1, NULLIF($2, ''), $3::uuid,
|
||||
NULLIF($4, '')::uuid, $5, $6, NULLIF($7, '')::uuid, NULLIF($8, ''), NULLIF($9, ''), NULLIF($10, ''),
|
||||
$11::uuid, $12::uuid, $13, $14, $15, NULLIF($16, ''),
|
||||
$17::jsonb, $18::jsonb, $19::jsonb, $20
|
||||
)
|
||||
ON CONFLICT (public_response_id) DO NOTHING`,
|
||||
input.PublicResponseID, input.ParentResponseID, input.GatewayTaskID,
|
||||
input.GatewayUserID, input.UserID, input.UserSource, input.GatewayTenantID, input.TenantID, input.TenantKey, input.APIKeyID,
|
||||
input.PlatformID, input.PlatformModelID, input.ClientID, input.UpstreamProtocol, input.UpstreamEndpoint, input.UpstreamResponseID,
|
||||
requestJSON, responseJSON, internalJSON, input.ExpiresAt,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *Store) GetResponseChain(ctx context.Context, publicResponseID string, owner GatewayTask) (ResponseChain, error) {
|
||||
chain, err := s.scanResponseChain(s.pool.QueryRow(ctx, `
|
||||
SELECT chain.public_response_id, COALESCE(chain.parent_response_id, ''), chain.gateway_task_id::text,
|
||||
task.requested_model,
|
||||
COALESCE(chain.gateway_user_id::text, ''), chain.user_id, chain.user_source,
|
||||
COALESCE(chain.gateway_tenant_id::text, ''), COALESCE(chain.tenant_id, ''), COALESCE(chain.tenant_key, ''), COALESCE(chain.api_key_id, ''),
|
||||
chain.platform_id::text, chain.platform_model_id::text, chain.client_id, chain.upstream_protocol, chain.upstream_endpoint,
|
||||
COALESCE(chain.upstream_response_id, ''), chain.request_snapshot, chain.response_snapshot, chain.internal_snapshot,
|
||||
chain.created_at, chain.expires_at
|
||||
FROM gateway_response_chains chain
|
||||
JOIN gateway_tasks task ON task.id = chain.gateway_task_id
|
||||
WHERE chain.public_response_id = $1`, strings.TrimSpace(publicResponseID)))
|
||||
if err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return ResponseChain{}, ErrInvalidPreviousResponseID
|
||||
}
|
||||
return ResponseChain{}, err
|
||||
}
|
||||
if chain.ExpiresAt.Before(time.Now()) || !sameResponseChainOwner(chain, owner) {
|
||||
return ResponseChain{}, ErrInvalidPreviousResponseID
|
||||
}
|
||||
return chain, nil
|
||||
}
|
||||
|
||||
func (s *Store) ListResponseChainHistory(ctx context.Context, current ResponseChain) ([]ResponseChain, error) {
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
WITH RECURSIVE history AS (
|
||||
SELECT chain.*, 0 AS depth
|
||||
FROM gateway_response_chains chain
|
||||
WHERE chain.public_response_id = $1
|
||||
UNION ALL
|
||||
SELECT parent.*, history.depth + 1
|
||||
FROM gateway_response_chains parent
|
||||
JOIN history ON parent.public_response_id = history.parent_response_id
|
||||
WHERE history.depth < $2
|
||||
)
|
||||
SELECT history.public_response_id, COALESCE(history.parent_response_id, ''), history.gateway_task_id::text,
|
||||
task.requested_model,
|
||||
COALESCE(history.gateway_user_id::text, ''), history.user_id, history.user_source,
|
||||
COALESCE(history.gateway_tenant_id::text, ''), COALESCE(history.tenant_id, ''), COALESCE(history.tenant_key, ''), COALESCE(history.api_key_id, ''),
|
||||
history.platform_id::text, history.platform_model_id::text, history.client_id, history.upstream_protocol, history.upstream_endpoint,
|
||||
COALESCE(history.upstream_response_id, ''), history.request_snapshot, history.response_snapshot, history.internal_snapshot,
|
||||
history.created_at, history.expires_at
|
||||
FROM history
|
||||
JOIN gateway_tasks task ON task.id = history.gateway_task_id
|
||||
ORDER BY depth DESC`, current.PublicResponseID, maxResponseChainHistory)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
out := make([]ResponseChain, 0)
|
||||
for rows.Next() {
|
||||
item, scanErr := s.scanResponseChain(rows)
|
||||
if scanErr != nil {
|
||||
return nil, scanErr
|
||||
}
|
||||
out = append(out, item)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(out) > maxResponseChainHistory {
|
||||
return nil, ErrResponseChainTooDeep
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
type responseChainScanner interface {
|
||||
Scan(dest ...any) error
|
||||
}
|
||||
|
||||
func (s *Store) scanResponseChain(scanner responseChainScanner) (ResponseChain, error) {
|
||||
var item ResponseChain
|
||||
var requestJSON, responseJSON, internalJSON []byte
|
||||
err := scanner.Scan(
|
||||
&item.PublicResponseID, &item.ParentResponseID, &item.GatewayTaskID, &item.RequestedModel,
|
||||
&item.GatewayUserID, &item.UserID, &item.UserSource,
|
||||
&item.GatewayTenantID, &item.TenantID, &item.TenantKey, &item.APIKeyID,
|
||||
&item.PlatformID, &item.PlatformModelID, &item.ClientID,
|
||||
&item.UpstreamProtocol, &item.UpstreamEndpoint, &item.UpstreamResponseID,
|
||||
&requestJSON, &responseJSON, &internalJSON, &item.CreatedAt, &item.ExpiresAt,
|
||||
)
|
||||
if err != nil {
|
||||
return ResponseChain{}, err
|
||||
}
|
||||
item.RequestSnapshot = decodeObject(requestJSON)
|
||||
item.ResponseSnapshot = decodeObject(responseJSON)
|
||||
item.InternalSnapshot = decodeObject(internalJSON)
|
||||
return item, nil
|
||||
}
|
||||
|
||||
func sameResponseChainOwner(chain ResponseChain, task GatewayTask) bool {
|
||||
if chain.GatewayUserID != "" || task.GatewayUserID != "" {
|
||||
if chain.GatewayUserID == "" || chain.GatewayUserID != task.GatewayUserID {
|
||||
return false
|
||||
}
|
||||
} else if chain.UserID != task.UserID || chain.UserSource != task.UserSource {
|
||||
return false
|
||||
}
|
||||
return chain.GatewayTenantID == task.GatewayTenantID &&
|
||||
chain.TenantID == task.TenantID &&
|
||||
chain.TenantKey == task.TenantKey
|
||||
}
|
||||
@@ -0,0 +1,36 @@
|
||||
package store
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestSameResponseChainOwnerIsolatesUserAndTenant(t *testing.T) {
|
||||
chain := ResponseChain{
|
||||
GatewayUserID: "user-1", GatewayTenantID: "tenant-1", TenantID: "external-tenant", TenantKey: "tenant-key",
|
||||
}
|
||||
owner := GatewayTask{
|
||||
GatewayUserID: "user-1", GatewayTenantID: "tenant-1", TenantID: "external-tenant", TenantKey: "tenant-key",
|
||||
}
|
||||
if !sameResponseChainOwner(chain, owner) {
|
||||
t.Fatal("matching user and tenant should own the response chain")
|
||||
}
|
||||
owner.GatewayUserID = "user-2"
|
||||
if sameResponseChainOwner(chain, owner) {
|
||||
t.Fatal("another user must not access the response chain")
|
||||
}
|
||||
owner.GatewayUserID = "user-1"
|
||||
owner.GatewayTenantID = "tenant-2"
|
||||
if sameResponseChainOwner(chain, owner) {
|
||||
t.Fatal("another tenant must not access the response chain")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSameResponseChainOwnerSupportsExternalIdentityFallback(t *testing.T) {
|
||||
chain := ResponseChain{UserID: "external-user", UserSource: "server-main", TenantID: "tenant-a"}
|
||||
owner := GatewayTask{UserID: "external-user", UserSource: "server-main", TenantID: "tenant-a"}
|
||||
if !sameResponseChainOwner(chain, owner) {
|
||||
t.Fatal("matching external identity should own the response chain")
|
||||
}
|
||||
owner.UserSource = "gateway"
|
||||
if sameResponseChainOwner(chain, owner) {
|
||||
t.Fatal("identity source must participate in isolation")
|
||||
}
|
||||
}
|
||||
@@ -167,6 +167,7 @@ type RuntimeModelCandidate struct {
|
||||
RunningCount float64
|
||||
WaitingCount float64
|
||||
LastAssignedUnix float64
|
||||
ResponseProtocol string
|
||||
}
|
||||
|
||||
type RuntimeCandidateCacheAffinity struct {
|
||||
|
||||
Reference in New Issue
Block a user