257 lines
9.4 KiB
Go
257 lines
9.4 KiB
Go
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()
|
|
}
|