1784 lines
66 KiB
Go
1784 lines
66 KiB
Go
package openai
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"go.uber.org/zap"
|
|
|
|
edgeservice "iop/apps/edge/internal/service"
|
|
"iop/packages/go/config"
|
|
iop "iop/proto/gen/iop"
|
|
)
|
|
|
|
func presetSelectorModelGroupKey(dispatch routeDispatch, fallback string) string {
|
|
if binding, ok := dispatch.PresetResolvedBindings[dispatch.Preset.Selector.Model]; ok {
|
|
if key := binding.effectiveModelGroupKey(dispatch.Preset.Selector.Model); key != "" {
|
|
return key
|
|
}
|
|
}
|
|
if model := strings.TrimSpace(dispatch.Preset.Selector.Model); model != "" {
|
|
return model
|
|
}
|
|
return dispatch.effectiveModelGroupKey(fallback)
|
|
}
|
|
|
|
func presetHotPathEnabled(dispatch routeDispatch) bool {
|
|
return dispatch.IsPreset && strings.TrimSpace(dispatch.Preset.Selector.Model) != ""
|
|
}
|
|
|
|
// collectPresetSelectorResult consumes the single selected attempt and returns
|
|
// both its canonical output and immutable admission evidence. The output is
|
|
// never relayed before structural classification.
|
|
func (s *Server) collectPresetSelectorResult(
|
|
ctx context.Context,
|
|
dispatch routeDispatch,
|
|
protocol string,
|
|
result *edgeservice.ProviderPoolDispatchResult,
|
|
) (normalizedStageOutput, hotPathSelectorGate, error) {
|
|
selected, gate, err := presetSelectorAdmission(dispatch, protocol, result)
|
|
if err != nil {
|
|
return normalizedStageOutput{}, hotPathSelectorGate{}, err
|
|
}
|
|
rejection := s.newHotPathRejectedDispatchOwner(result)
|
|
if result.Run != nil && result.Tunnel != nil {
|
|
s.abortHotPathRejectedDispatch(rejection)
|
|
return normalizedStageOutput{}, gate, fmt.Errorf("preset selector returned multiple execution results")
|
|
}
|
|
|
|
var stage normalizedStageOutput
|
|
bufferedOuter := newHotPathOuterTurn("")
|
|
snapshot := hotPathDispatchSnapshot{StageID: hotPathFirstNonEmpty(selected.RunID, "selector-stage")}
|
|
switch result.Path {
|
|
case edgeservice.ProviderPoolPathNormalized:
|
|
if result.Run == nil {
|
|
s.abortHotPathRejectedDispatch(rejection)
|
|
return normalizedStageOutput{}, gate, fmt.Errorf("preset selector selected normalized path without a run result")
|
|
}
|
|
if err := validateSelectedDispatch(selected, result.Run.Dispatch()); err != nil {
|
|
s.abortHotPathRejectedDispatch(rejection)
|
|
return normalizedStageOutput{}, gate, err
|
|
}
|
|
stage, err = collectHotPathOwnedStage(ctx, bufferedOuter, snapshot.StageID, rejection, func() (normalizedStageOutput, error) {
|
|
return collectPresetNormalizedResult(ctx, result.Run, selected)
|
|
})
|
|
case edgeservice.ProviderPoolPathTunnel:
|
|
if result.Tunnel == nil {
|
|
s.abortHotPathRejectedDispatch(rejection)
|
|
return normalizedStageOutput{}, gate, fmt.Errorf("preset selector selected tunnel path without a tunnel result")
|
|
}
|
|
if err := validateSelectedDispatch(selected, result.Tunnel.Dispatch()); err != nil {
|
|
s.abortHotPathRejectedDispatch(rejection)
|
|
return normalizedStageOutput{}, gate, err
|
|
}
|
|
stage, err = collectHotPathOwnedStage(ctx, bufferedOuter, snapshot.StageID, rejection, func() (normalizedStageOutput, error) {
|
|
return collectPresetTunnelResult(ctx, result.Tunnel, selected, protocol)
|
|
})
|
|
default:
|
|
s.abortHotPathRejectedDispatch(rejection)
|
|
err = fmt.Errorf("preset selector returned unsupported execution path %q", result.Path)
|
|
}
|
|
// Selector classification still occurs before caller release. The temporary
|
|
// outer turn above exists only to own the exact active transport and typed
|
|
// terminal race; the classified output is collected into the caller turn.
|
|
stage.ProgressivelyReleased = false
|
|
return stage, gate, err
|
|
}
|
|
|
|
func presetSelectorAdmission(
|
|
dispatch routeDispatch,
|
|
protocol string,
|
|
result *edgeservice.ProviderPoolDispatchResult,
|
|
) (edgeservice.RunDispatch, hotPathSelectorGate, error) {
|
|
if result == nil {
|
|
return edgeservice.RunDispatch{}, hotPathSelectorGate{}, fmt.Errorf("preset selector returned no provider result")
|
|
}
|
|
selected := result.DispatchInfo
|
|
gate := hotPathSelectorGate{
|
|
PresetID: dispatch.Preset.ID,
|
|
SelectorModel: dispatch.Preset.Selector.Model,
|
|
ModelGroupKey: selected.ModelGroupKey,
|
|
ProviderID: selected.ProviderID,
|
|
RunID: selected.RunID,
|
|
NodeID: selected.NodeID,
|
|
ExecutionPath: selected.ExecutionPath,
|
|
ProfileDriver: selected.ProfileDriver,
|
|
ProfileCapabilities: append([]string(nil), selected.ProfileCapabilities...),
|
|
}
|
|
expectedGroup := presetSelectorModelGroupKey(dispatch, dispatch.ExternalModelID)
|
|
gate.Healthy = strings.TrimSpace(selected.RunID) != "" &&
|
|
strings.TrimSpace(selected.NodeID) != "" &&
|
|
strings.TrimSpace(selected.ProviderID) != "" &&
|
|
strings.TrimSpace(selected.ModelGroupKey) == strings.TrimSpace(expectedGroup) &&
|
|
strings.TrimSpace(selected.ExecutionPath) == string(result.Path)
|
|
gate.CapabilitySatisfied = selectedPresetCapability(protocol, selected.ProfileDriver, selected.ProfileCapabilities)
|
|
return selected, gate, nil
|
|
}
|
|
|
|
func (s *Server) runLivePresetSelectorResult(
|
|
ctx context.Context,
|
|
dispatch routeDispatch,
|
|
protocol string,
|
|
stageID string,
|
|
result *edgeservice.ProviderPoolDispatchResult,
|
|
outer *hotPathOuterTurn,
|
|
) (normalizedStageOutput, hotPathSelectorGate, error) {
|
|
selected, gate, err := presetSelectorAdmission(dispatch, protocol, result)
|
|
if err != nil {
|
|
return normalizedStageOutput{}, hotPathSelectorGate{}, err
|
|
}
|
|
rejection := s.newHotPathRejectedDispatchOwner(result)
|
|
if result.Run != nil && result.Tunnel != nil {
|
|
s.abortHotPathRejectedDispatch(rejection)
|
|
return normalizedStageOutput{}, gate, fmt.Errorf("preset selector returned multiple execution results")
|
|
}
|
|
snapshot := hotPathDispatchSnapshot{StageID: stageID}
|
|
switch result.Path {
|
|
case edgeservice.ProviderPoolPathNormalized:
|
|
if result.Run == nil {
|
|
s.abortHotPathRejectedDispatch(rejection)
|
|
return normalizedStageOutput{}, gate, fmt.Errorf("preset selector selected normalized path without a run result")
|
|
}
|
|
if err := validateSelectedDispatch(selected, result.Run.Dispatch()); err != nil {
|
|
s.abortHotPathRejectedDispatch(rejection)
|
|
return normalizedStageOutput{}, gate, err
|
|
}
|
|
output, _, err := s.runHotPathLiveNormalizedStage(ctx, snapshot, outer, result.Run, selected)
|
|
return output, gate, err
|
|
case edgeservice.ProviderPoolPathTunnel:
|
|
if result.Tunnel == nil {
|
|
s.abortHotPathRejectedDispatch(rejection)
|
|
return normalizedStageOutput{}, gate, fmt.Errorf("preset selector selected tunnel path without a tunnel result")
|
|
}
|
|
if err := validateSelectedDispatch(selected, result.Tunnel.Dispatch()); err != nil {
|
|
s.abortHotPathRejectedDispatch(rejection)
|
|
return normalizedStageOutput{}, gate, err
|
|
}
|
|
output, _, err := s.runHotPathLiveTunnelStage(ctx, snapshot, outer, result.Tunnel, selected)
|
|
return output, gate, err
|
|
default:
|
|
s.abortHotPathRejectedDispatch(rejection)
|
|
return normalizedStageOutput{}, gate, fmt.Errorf("preset selector returned unsupported execution path %q", result.Path)
|
|
}
|
|
}
|
|
|
|
func selectedPresetCapability(protocol, driver string, capabilities []string) bool {
|
|
required := "chat"
|
|
if protocol == "anthropic" && driver == string(config.ProtocolDriverAnthropicMessages) {
|
|
required = "messages"
|
|
}
|
|
for _, capability := range capabilities {
|
|
if strings.TrimSpace(capability) == required {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (s *Server) collectHotPathOwnedNormalizedStage(
|
|
ctx context.Context,
|
|
stageID string,
|
|
outer *hotPathOuterTurn,
|
|
handle edgeservice.RunResult,
|
|
dispatch edgeservice.RunDispatch,
|
|
) (normalizedStageOutput, error) {
|
|
if handle == nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("hot path normalized stage returned no run result")
|
|
}
|
|
controller := newHotPathStageTransportController(s.service, dispatch, handle.Close)
|
|
return collectHotPathOwnedStage(ctx, outer, stageID, controller, func() (normalizedStageOutput, error) {
|
|
return collectPresetNormalizedResult(ctx, handle, dispatch)
|
|
})
|
|
}
|
|
|
|
func (s *Server) collectHotPathOwnedTunnelStage(
|
|
ctx context.Context,
|
|
stageID string,
|
|
outer *hotPathOuterTurn,
|
|
handle edgeservice.ProviderTunnelResult,
|
|
dispatch edgeservice.RunDispatch,
|
|
protocol string,
|
|
) (normalizedStageOutput, error) {
|
|
if handle == nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("hot path tunnel stage returned no provider result")
|
|
}
|
|
controller := newHotPathStageTransportController(s.service, dispatch, handle.Close)
|
|
return collectHotPathOwnedStage(ctx, outer, stageID, controller, func() (normalizedStageOutput, error) {
|
|
return collectPresetTunnelResult(ctx, handle, dispatch, protocol)
|
|
})
|
|
}
|
|
|
|
func collectHotPathOwnedStage(
|
|
ctx context.Context,
|
|
outer *hotPathOuterTurn,
|
|
stageID string,
|
|
controller hotPathStageAttemptController,
|
|
collect func() (normalizedStageOutput, error),
|
|
) (normalizedStageOutput, error) {
|
|
if outer == nil {
|
|
outer = newHotPathOuterTurn("")
|
|
}
|
|
active, err := outer.registerActiveStage(stageID, controller)
|
|
if err != nil {
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
watchStop := make(chan struct{})
|
|
watchDone := make(chan struct{})
|
|
go func() {
|
|
defer close(watchDone)
|
|
select {
|
|
case <-ctx.Done():
|
|
outer.cancelActiveStage(hotPathDispositionForError(ctx.Err()), "caller_context", ctx.Err())
|
|
case <-watchStop:
|
|
}
|
|
}()
|
|
output, collectErr := collect()
|
|
close(watchStop)
|
|
<-watchDone
|
|
if collectErr == nil {
|
|
_ = active.CloseAttempt(context.Background())
|
|
return output, nil
|
|
}
|
|
|
|
disposition, typed := hotPathDispositionFromError(collectErr)
|
|
kind := hotPathDispositionForError(collectErr)
|
|
if typed {
|
|
kind = disposition.Kind
|
|
}
|
|
if kind == hotPathDispositionCallerCancel || kind == hotPathDispositionTimeout {
|
|
outer.cancelActiveStage(kind, "stage_collector", collectErr)
|
|
} else {
|
|
if !typed {
|
|
disposition = outer.activeStageDisposition(kind, "stage_collector", collectErr.Error())
|
|
} else if disposition.Generation == 0 {
|
|
owned := outer.activeStageDisposition(disposition.Kind, disposition.Source, disposition.Cause)
|
|
disposition.Generation = owned.Generation
|
|
if disposition.StageID == "" {
|
|
disposition.StageID = owned.StageID
|
|
}
|
|
}
|
|
outer.selectDisposition(disposition)
|
|
_ = active.AbortAttempt(context.Background())
|
|
}
|
|
return normalizedStageOutput{}, wrapHotPathDispositionError(outer, stageID, collectErr)
|
|
}
|
|
|
|
func collectPresetNormalizedResult(ctx context.Context, handle edgeservice.RunResult, selected edgeservice.RunDispatch) (normalizedStageOutput, error) {
|
|
if handle == nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("preset selector selected normalized path without a run result")
|
|
}
|
|
if err := validateSelectedDispatch(selected, handle.Dispatch()); err != nil {
|
|
return normalizedStageOutput{}, newHotPathDispositionError(
|
|
hotPathDispositionValidationError, "stage_dispatch_validation", selected.RunID, err,
|
|
)
|
|
}
|
|
stream := handle.Stream()
|
|
if stream.Events == nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("preset selector run stream is unavailable")
|
|
}
|
|
timer := time.NewTimer(handle.WaitTimeout())
|
|
defer timer.Stop()
|
|
stage := normalizedStageOutput{}
|
|
var identity hotPathProviderIdentity
|
|
var content, reasoning strings.Builder
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return normalizedStageOutput{}, ctx.Err()
|
|
case <-timer.C:
|
|
return normalizedStageOutput{}, errRunTimedOut
|
|
case nodeEvent, ok := <-stream.NodeEvents:
|
|
if !ok {
|
|
stream.NodeEvents = nil
|
|
continue
|
|
}
|
|
if edgeservice.IsNodeDisconnected(nodeEvent) {
|
|
return normalizedStageOutput{}, fmt.Errorf("node disconnected")
|
|
}
|
|
case event, ok := <-stream.Events:
|
|
if !ok {
|
|
return normalizedStageOutput{}, fmt.Errorf("preset selector run stream closed before completion")
|
|
}
|
|
if event == nil {
|
|
continue
|
|
}
|
|
if event.GetTimestamp() != 0 {
|
|
stage.Created = unixSeconds(event.GetTimestamp())
|
|
}
|
|
switch event.GetType() {
|
|
case "delta":
|
|
if _, err := identity.bindRequired(event.GetMetadata()[hotPathOpenAIResponseIDMetadata]); err != nil {
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
content.WriteString(event.GetDelta())
|
|
if event.GetDelta() != "" {
|
|
stage.Deltas = append(stage.Deltas, normalizedStageDelta{Kind: normalizedStageDeltaText, Text: event.GetDelta()})
|
|
}
|
|
case "reasoning_delta":
|
|
if _, err := identity.bindRequired(event.GetMetadata()[hotPathOpenAIResponseIDMetadata]); err != nil {
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
reasoning.WriteString(event.GetDelta())
|
|
if event.GetDelta() != "" {
|
|
stage.Deltas = append(stage.Deltas, normalizedStageDelta{Kind: normalizedStageDeltaReasoning, Text: event.GetDelta()})
|
|
}
|
|
case "complete":
|
|
responseID, err := identity.bindRequired(event.GetMetadata()[hotPathOpenAIResponseIDMetadata])
|
|
if err != nil {
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
stage.ResponseID = responseID
|
|
stage.Content = content.String()
|
|
stage.Reasoning = reasoning.String()
|
|
stage.TerminalReason = strings.TrimSpace(event.GetMetadata()["finish_reason"])
|
|
if stage.TerminalReason == "" {
|
|
stage.TerminalReason = "stop"
|
|
}
|
|
stage.ToolCalls, err = normalizeRunEventToolCalls(event.GetMetadata())
|
|
if err != nil {
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
if len(stage.ToolCalls) > 0 {
|
|
stage.TerminalReason = "tool_calls"
|
|
for _, call := range stage.ToolCalls {
|
|
stage.Deltas = append(stage.Deltas, normalizedStageDelta{
|
|
Kind: normalizedStageDeltaTool, ToolID: call.ProviderCallID,
|
|
ToolName: call.Name, Arguments: directToolArguments(call),
|
|
})
|
|
}
|
|
}
|
|
if usage := event.GetUsage(); usage != nil {
|
|
stage.OpenAIUsage = &openAIUsage{
|
|
PromptTokens: int(usage.GetInputTokens()),
|
|
CompletionTokens: int(usage.GetOutputTokens()),
|
|
TotalTokens: int(usage.GetInputTokens() + usage.GetOutputTokens()),
|
|
ReasoningTokens: int(usage.GetReasoningTokens()),
|
|
CachedInputTokens: int(usage.GetCachedInputTokens()),
|
|
}
|
|
stage.Usage, _ = json.Marshal(stage.OpenAIUsage)
|
|
}
|
|
return stage, nil
|
|
case "error", "cancelled":
|
|
message := event.GetError()
|
|
if message == "" {
|
|
message = event.GetMessage()
|
|
}
|
|
if message == "" {
|
|
message = "preset selector run failed"
|
|
}
|
|
return normalizedStageOutput{}, fmt.Errorf("%s", message)
|
|
default:
|
|
if err := identity.bind(event.GetMetadata()[hotPathOpenAIResponseIDMetadata]); err != nil {
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func collectPresetTunnelResult(ctx context.Context, handle edgeservice.ProviderTunnelResult, selected edgeservice.RunDispatch, protocol string) (normalizedStageOutput, error) {
|
|
if handle == nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("preset selector selected tunnel path without a tunnel result")
|
|
}
|
|
if err := validateSelectedDispatch(selected, handle.Dispatch()); err != nil {
|
|
return normalizedStageOutput{}, newHotPathDispositionError(
|
|
hotPathDispositionValidationError, "stage_dispatch_validation", selected.RunID, err,
|
|
)
|
|
}
|
|
frames := handle.Stream().Frames
|
|
if frames == nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("preset selector tunnel stream is unavailable")
|
|
}
|
|
timer := time.NewTimer(handle.WaitTimeout())
|
|
defer timer.Stop()
|
|
var body bytes.Buffer
|
|
status := 0
|
|
contentType := ""
|
|
var sideUsage *iop.Usage
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return normalizedStageOutput{}, ctx.Err()
|
|
case <-timer.C:
|
|
return normalizedStageOutput{}, errRunTimedOut
|
|
case frame, ok := <-frames:
|
|
if !ok {
|
|
return normalizedStageOutput{}, fmt.Errorf("preset selector tunnel closed before completion")
|
|
}
|
|
if frame == nil {
|
|
continue
|
|
}
|
|
switch frame.GetKind() {
|
|
case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START:
|
|
status = int(frame.GetStatusCode())
|
|
if status == 0 {
|
|
status = http.StatusOK
|
|
}
|
|
for name, value := range frame.GetHeaders() {
|
|
if strings.EqualFold(name, "Content-Type") {
|
|
contentType = value
|
|
}
|
|
}
|
|
case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY:
|
|
_, _ = body.Write(frame.GetBody())
|
|
case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_USAGE:
|
|
sideUsage = frame.GetUsage()
|
|
case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR:
|
|
message := strings.TrimSpace(frame.GetError())
|
|
if message == "" {
|
|
message = "provider tunnel failed"
|
|
}
|
|
return normalizedStageOutput{}, fmt.Errorf("%s", message)
|
|
case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_END:
|
|
if status < http.StatusOK || status >= http.StatusMultipleChoices {
|
|
return normalizedStageOutput{}, fmt.Errorf("preset selector provider returned HTTP %d", status)
|
|
}
|
|
stage, err := decodePresetTunnelBody(body.Bytes(), contentType, protocol, selected.ProfileDriver)
|
|
if err != nil {
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
if err := validateProviderStageMetadata(protocol, stage); err != nil {
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
if len(stage.Usage) == 0 && sideUsage != nil {
|
|
stage.OpenAIUsage = &openAIUsage{
|
|
PromptTokens: int(sideUsage.GetInputTokens()), CompletionTokens: int(sideUsage.GetOutputTokens()),
|
|
TotalTokens: int(sideUsage.GetInputTokens() + sideUsage.GetOutputTokens()),
|
|
ReasoningTokens: int(sideUsage.GetReasoningTokens()), CachedInputTokens: int(sideUsage.GetCachedInputTokens()),
|
|
}
|
|
if protocol == "anthropic" && selected.ProfileDriver == string(config.ProtocolDriverAnthropicMessages) {
|
|
stage.Usage, _ = json.Marshal(anthropicUsage{
|
|
InputTokens: int(sideUsage.GetInputTokens()), OutputTokens: int(sideUsage.GetOutputTokens()),
|
|
CacheReadInputTokens: int(sideUsage.GetCachedInputTokens()),
|
|
})
|
|
} else if protocol == "anthropic" {
|
|
stage.Usage = openAIUsageToAnthropic(mustMarshalRaw(stage.OpenAIUsage))
|
|
} else {
|
|
stage.Usage, _ = json.Marshal(stage.OpenAIUsage)
|
|
}
|
|
}
|
|
return stage, nil
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func validateProviderStageMetadata(protocol string, stage normalizedStageOutput) error {
|
|
if strings.TrimSpace(stage.ResponseID) == "" {
|
|
return fmt.Errorf("provider response is missing required identity")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func validateSelectedDispatch(selected, handle edgeservice.RunDispatch) error {
|
|
if handle.ProviderID != "" && selected.ProviderID != handle.ProviderID {
|
|
return fmt.Errorf("preset selector dispatch evidence changed after admission")
|
|
}
|
|
if handle.ModelGroupKey != "" && selected.ModelGroupKey != handle.ModelGroupKey {
|
|
return fmt.Errorf("preset selector dispatch evidence changed after admission")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func unixSeconds(timestamp int64) int64 {
|
|
if timestamp > 1_000_000_000_000 {
|
|
return timestamp / int64(time.Second)
|
|
}
|
|
return timestamp
|
|
}
|
|
|
|
func decodePresetTunnelBody(body []byte, contentType, protocol, driver string) (normalizedStageOutput, error) {
|
|
streaming := strings.Contains(strings.ToLower(contentType), "text/event-stream") || bytes.Contains(body, []byte("data:"))
|
|
if protocol == "anthropic" && driver == string(config.ProtocolDriverAnthropicMessages) {
|
|
if streaming {
|
|
return decodeAnthropicPresetSSE(body)
|
|
}
|
|
return decodeAnthropicPresetJSON(body)
|
|
}
|
|
var stage normalizedStageOutput
|
|
var err error
|
|
if streaming {
|
|
stage, err = decodeOpenAIPresetSSE(body)
|
|
} else {
|
|
stage, err = decodeOpenAIPresetJSON(body)
|
|
}
|
|
if err != nil {
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
if protocol == "anthropic" {
|
|
stage.Usage = openAIUsageToAnthropic(stage.Usage)
|
|
stage.TerminalReason = openAIReasonToAnthropic(stage.TerminalReason)
|
|
}
|
|
return stage, nil
|
|
}
|
|
|
|
func decodeOpenAIPresetJSON(body []byte) (normalizedStageOutput, error) {
|
|
var response struct {
|
|
ID string `json:"id"`
|
|
Created int64 `json:"created"`
|
|
Usage json.RawMessage `json:"usage"`
|
|
Choices []struct {
|
|
Message struct {
|
|
Content any `json:"content"`
|
|
ReasoningContent string `json:"reasoning_content"`
|
|
Reasoning string `json:"reasoning"`
|
|
ToolCalls []any `json:"tool_calls"`
|
|
} `json:"message"`
|
|
FinishReason string `json:"finish_reason"`
|
|
} `json:"choices"`
|
|
}
|
|
if err := json.Unmarshal(body, &response); err != nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("decode preset Chat response: %w", err)
|
|
}
|
|
if len(response.Choices) != 1 {
|
|
return normalizedStageOutput{}, fmt.Errorf("preset Chat response must contain exactly one choice")
|
|
}
|
|
choice := response.Choices[0]
|
|
reasoning := choice.Message.ReasoningContent
|
|
if reasoning == "" {
|
|
reasoning = choice.Message.Reasoning
|
|
}
|
|
toolCalls, err := normalizeProviderToolCalls(choice.Message.ToolCalls)
|
|
if err != nil {
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
stage := normalizedStageOutput{
|
|
ResponseID: response.ID, Created: response.Created, Content: contentToString(choice.Message.Content),
|
|
Reasoning: reasoning, ToolCalls: toolCalls,
|
|
TerminalReason: choice.FinishReason, Usage: cloneRawJSON(response.Usage),
|
|
}
|
|
if reasoning != "" {
|
|
stage.Deltas = append(stage.Deltas, normalizedStageDelta{Kind: normalizedStageDeltaReasoning, Text: reasoning})
|
|
}
|
|
if stage.Content != "" {
|
|
stage.Deltas = append(stage.Deltas, normalizedStageDelta{Kind: normalizedStageDeltaText, Text: stage.Content})
|
|
}
|
|
for _, call := range toolCalls {
|
|
stage.Deltas = append(stage.Deltas, normalizedStageDelta{
|
|
Kind: normalizedStageDeltaTool, ToolID: call.ProviderCallID,
|
|
ToolName: call.Name, Arguments: directToolArguments(call),
|
|
})
|
|
}
|
|
stage.OpenAIUsage = decodeOpenAIUsage(response.Usage)
|
|
return stage, nil
|
|
}
|
|
|
|
func decodeOpenAIPresetSSE(body []byte) (normalizedStageOutput, error) {
|
|
stage := normalizedStageOutput{}
|
|
identity := &hotPathProviderIdentity{}
|
|
type toolState struct {
|
|
id, name string
|
|
args strings.Builder
|
|
}
|
|
tools := make(map[int]*toolState)
|
|
for _, payload := range sseDataPayloads(body) {
|
|
if bytes.Equal(bytes.TrimSpace(payload), []byte("[DONE]")) {
|
|
continue
|
|
}
|
|
var chunk openAIChatStreamChunk
|
|
if err := json.Unmarshal(payload, &chunk); err != nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("decode preset Chat stream: %w", err)
|
|
}
|
|
if chunk.Error != nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("preset Chat stream error: %s", chunk.Error.Message)
|
|
}
|
|
if err := identity.bind(chunk.ID); err != nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("decode preset Chat stream identity: %w", err)
|
|
}
|
|
var raw struct {
|
|
Created int64 `json:"created"`
|
|
Usage json.RawMessage `json:"usage"`
|
|
}
|
|
_ = json.Unmarshal(payload, &raw)
|
|
if raw.Created != 0 {
|
|
stage.Created = raw.Created
|
|
}
|
|
if len(raw.Usage) > 0 && string(raw.Usage) != "null" {
|
|
stage.Usage = cloneRawJSON(raw.Usage)
|
|
stage.OpenAIUsage = decodeOpenAIUsage(raw.Usage)
|
|
}
|
|
for _, choice := range chunk.Choices {
|
|
visible := choice.Delta.Content != "" || choice.Delta.ReasoningContent != "" ||
|
|
choice.Delta.Reasoning != "" || len(choice.Delta.ToolCalls) > 0
|
|
if visible {
|
|
if _, err := identity.require(); err != nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("decode preset Chat stream identity: %w", err)
|
|
}
|
|
}
|
|
stage.Content += choice.Delta.Content
|
|
if choice.Delta.Content != "" {
|
|
stage.Deltas = append(stage.Deltas, normalizedStageDelta{Kind: normalizedStageDeltaText, Text: choice.Delta.Content})
|
|
}
|
|
reasoning := choice.Delta.ReasoningContent
|
|
if reasoning == "" {
|
|
reasoning = choice.Delta.Reasoning
|
|
}
|
|
stage.Reasoning += reasoning
|
|
if reasoning != "" {
|
|
stage.Deltas = append(stage.Deltas, normalizedStageDelta{Kind: normalizedStageDeltaReasoning, Text: reasoning})
|
|
}
|
|
for _, delta := range choice.Delta.ToolCalls {
|
|
state := tools[delta.Index]
|
|
if state == nil {
|
|
state = &toolState{}
|
|
tools[delta.Index] = state
|
|
}
|
|
if delta.ID != "" {
|
|
state.id = delta.ID
|
|
}
|
|
if delta.Function.Name != "" {
|
|
state.name = delta.Function.Name
|
|
}
|
|
state.args.WriteString(delta.Function.Arguments)
|
|
if delta.Function.Arguments != "" {
|
|
stage.Deltas = append(stage.Deltas, normalizedStageDelta{
|
|
Kind: normalizedStageDeltaTool, ToolID: state.id,
|
|
ToolName: state.name, Arguments: delta.Function.Arguments,
|
|
})
|
|
}
|
|
}
|
|
if choice.FinishReason != nil {
|
|
stage.TerminalReason = *choice.FinishReason
|
|
}
|
|
}
|
|
}
|
|
responseID, err := identity.require()
|
|
if err != nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("decode preset Chat stream identity: %w", err)
|
|
}
|
|
stage.ResponseID = responseID
|
|
for index := 0; index < len(tools); index++ {
|
|
state, ok := tools[index]
|
|
if !ok {
|
|
return normalizedStageOutput{}, fmt.Errorf("preset Chat stream tool indices are not contiguous")
|
|
}
|
|
call, err := normalizedToolCallFromParts(state.id, state.name, state.args.String())
|
|
if err != nil {
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
stage.ToolCalls = append(stage.ToolCalls, call)
|
|
}
|
|
return stage, nil
|
|
}
|
|
|
|
func decodeAnthropicPresetJSON(body []byte) (normalizedStageOutput, error) {
|
|
var response struct {
|
|
ID string `json:"id"`
|
|
Content []json.RawMessage `json:"content"`
|
|
StopReason string `json:"stop_reason"`
|
|
Usage json.RawMessage `json:"usage"`
|
|
}
|
|
if err := json.Unmarshal(body, &response); err != nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("decode preset Messages response: %w", err)
|
|
}
|
|
stage := normalizedStageOutput{ResponseID: response.ID, TerminalReason: response.StopReason, Usage: cloneRawJSON(response.Usage)}
|
|
for _, raw := range response.Content {
|
|
if err := appendAnthropicBlock(&stage, raw); err != nil {
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
}
|
|
return stage, nil
|
|
}
|
|
|
|
func decodeAnthropicPresetSSE(body []byte) (normalizedStageOutput, error) {
|
|
stage := normalizedStageOutput{}
|
|
identity := &hotPathProviderIdentity{}
|
|
type toolState struct {
|
|
id, name string
|
|
args strings.Builder
|
|
}
|
|
tools := make(map[int]*toolState)
|
|
for _, payload := range sseDataPayloads(body) {
|
|
var event map[string]json.RawMessage
|
|
if err := json.Unmarshal(payload, &event); err != nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("decode preset Messages stream: %w", err)
|
|
}
|
|
var eventType string
|
|
_ = json.Unmarshal(event["type"], &eventType)
|
|
switch eventType {
|
|
case "message_start":
|
|
var message struct {
|
|
ID string `json:"id"`
|
|
Usage json.RawMessage `json:"usage"`
|
|
}
|
|
if err := json.Unmarshal(event["message"], &message); err != nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("decode preset Messages start: %w", err)
|
|
}
|
|
if err := identity.bind(message.ID); err != nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("decode preset Messages stream identity: %w", err)
|
|
}
|
|
stage.Usage = mergeJSONObjects(stage.Usage, message.Usage)
|
|
case "content_block_start":
|
|
var start struct {
|
|
Index int `json:"index"`
|
|
Block struct {
|
|
Type, ID, Name, Text, Thinking, Signature string
|
|
Input json.RawMessage `json:"input"`
|
|
} `json:"content_block"`
|
|
}
|
|
if err := json.Unmarshal(payload, &start); err != nil {
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
if _, err := identity.require(); err != nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("decode preset Messages stream identity: %w", err)
|
|
}
|
|
switch start.Block.Type {
|
|
case "text":
|
|
stage.Content += start.Block.Text
|
|
if start.Block.Text != "" {
|
|
stage.Deltas = append(stage.Deltas, normalizedStageDelta{Kind: normalizedStageDeltaText, Text: start.Block.Text})
|
|
}
|
|
case "thinking":
|
|
stage.Reasoning += start.Block.Thinking
|
|
stage.ReasoningSignature += start.Block.Signature
|
|
if start.Block.Thinking != "" {
|
|
stage.Deltas = append(stage.Deltas, normalizedStageDelta{Kind: normalizedStageDeltaReasoning, Text: start.Block.Thinking})
|
|
}
|
|
case "tool_use":
|
|
state := &toolState{id: start.Block.ID, name: start.Block.Name}
|
|
if len(start.Block.Input) > 0 && string(start.Block.Input) != "{}" {
|
|
state.args.Write(start.Block.Input)
|
|
stage.Deltas = append(stage.Deltas, normalizedStageDelta{
|
|
Kind: normalizedStageDeltaTool, ToolID: state.id,
|
|
ToolName: state.name, Arguments: string(start.Block.Input),
|
|
})
|
|
}
|
|
tools[start.Index] = state
|
|
}
|
|
case "content_block_delta":
|
|
var delta struct {
|
|
Index int `json:"index"`
|
|
Delta struct {
|
|
Type string `json:"type"`
|
|
Text string `json:"text"`
|
|
Thinking string `json:"thinking"`
|
|
Signature string `json:"signature"`
|
|
PartialJSON string `json:"partial_json"`
|
|
} `json:"delta"`
|
|
}
|
|
if err := json.Unmarshal(payload, &delta); err != nil {
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
if _, err := identity.require(); err != nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("decode preset Messages stream identity: %w", err)
|
|
}
|
|
switch delta.Delta.Type {
|
|
case "text_delta":
|
|
stage.Content += delta.Delta.Text
|
|
if delta.Delta.Text != "" {
|
|
stage.Deltas = append(stage.Deltas, normalizedStageDelta{Kind: normalizedStageDeltaText, Text: delta.Delta.Text})
|
|
}
|
|
case "thinking_delta":
|
|
stage.Reasoning += delta.Delta.Thinking
|
|
if delta.Delta.Thinking != "" {
|
|
stage.Deltas = append(stage.Deltas, normalizedStageDelta{Kind: normalizedStageDeltaReasoning, Text: delta.Delta.Thinking})
|
|
}
|
|
case "signature_delta":
|
|
stage.ReasoningSignature += delta.Delta.Signature
|
|
case "input_json_delta":
|
|
if state := tools[delta.Index]; state != nil {
|
|
state.args.WriteString(delta.Delta.PartialJSON)
|
|
if delta.Delta.PartialJSON != "" {
|
|
stage.Deltas = append(stage.Deltas, normalizedStageDelta{
|
|
Kind: normalizedStageDeltaTool, ToolID: state.id,
|
|
ToolName: state.name, Arguments: delta.Delta.PartialJSON,
|
|
})
|
|
}
|
|
}
|
|
}
|
|
case "content_block_stop":
|
|
var stop struct {
|
|
Index int `json:"index"`
|
|
}
|
|
if err := json.Unmarshal(payload, &stop); err == nil {
|
|
if state := tools[stop.Index]; state != nil {
|
|
if state.args.Len() == 0 {
|
|
state.args.WriteString("{}")
|
|
stage.Deltas = append(stage.Deltas, normalizedStageDelta{
|
|
Kind: normalizedStageDeltaTool, ToolID: state.id,
|
|
ToolName: state.name, Arguments: "{}",
|
|
})
|
|
}
|
|
}
|
|
}
|
|
case "message_delta":
|
|
var delta struct {
|
|
Delta struct {
|
|
StopReason string `json:"stop_reason"`
|
|
} `json:"delta"`
|
|
Usage json.RawMessage `json:"usage"`
|
|
}
|
|
if err := json.Unmarshal(payload, &delta); err != nil {
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
stage.TerminalReason = delta.Delta.StopReason
|
|
stage.Usage = mergeJSONObjects(stage.Usage, delta.Usage)
|
|
case "error":
|
|
return normalizedStageOutput{}, fmt.Errorf("preset Messages stream returned an error")
|
|
}
|
|
}
|
|
responseID, err := identity.require()
|
|
if err != nil {
|
|
return normalizedStageOutput{}, fmt.Errorf("decode preset Messages stream identity: %w", err)
|
|
}
|
|
stage.ResponseID = responseID
|
|
indices := make([]int, 0, len(tools))
|
|
for index := range tools {
|
|
indices = append(indices, index)
|
|
}
|
|
sort.Ints(indices)
|
|
for _, index := range indices {
|
|
state := tools[index]
|
|
args := state.args.String()
|
|
if args == "" {
|
|
args = "{}"
|
|
}
|
|
call, err := normalizedToolCallFromParts(state.id, state.name, args)
|
|
if err != nil {
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
stage.ToolCalls = append(stage.ToolCalls, call)
|
|
}
|
|
return stage, nil
|
|
}
|
|
|
|
func appendAnthropicBlock(stage *normalizedStageOutput, raw json.RawMessage) error {
|
|
var block struct {
|
|
Type, Text, Thinking, Signature, ID, Name string
|
|
Input json.RawMessage `json:"input"`
|
|
}
|
|
if err := json.Unmarshal(raw, &block); err != nil {
|
|
return fmt.Errorf("decode preset Messages content block: %w", err)
|
|
}
|
|
switch block.Type {
|
|
case "text":
|
|
stage.Content += block.Text
|
|
if block.Text != "" {
|
|
stage.Deltas = append(stage.Deltas, normalizedStageDelta{Kind: normalizedStageDeltaText, Text: block.Text})
|
|
}
|
|
case "thinking":
|
|
stage.Reasoning += block.Thinking
|
|
stage.ReasoningSignature += block.Signature
|
|
if block.Thinking != "" {
|
|
stage.Deltas = append(stage.Deltas, normalizedStageDelta{Kind: normalizedStageDeltaReasoning, Text: block.Thinking})
|
|
}
|
|
case "tool_use":
|
|
call, err := normalizedToolCallFromParts(block.ID, block.Name, string(block.Input))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
stage.ToolCalls = append(stage.ToolCalls, call)
|
|
stage.Deltas = append(stage.Deltas, normalizedStageDelta{
|
|
Kind: normalizedStageDeltaTool, ToolID: call.ProviderCallID,
|
|
ToolName: call.Name, Arguments: directToolArguments(call),
|
|
})
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func normalizeProviderToolCalls(toolCalls []any) ([]normalizedToolCall, error) {
|
|
out := make([]normalizedToolCall, 0, len(toolCalls))
|
|
for _, value := range toolCalls {
|
|
raw, err := json.Marshal(value)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("encode preset selector tool call: %w", err)
|
|
}
|
|
var call struct {
|
|
ID string `json:"id"`
|
|
Name string `json:"name"`
|
|
Input json.RawMessage `json:"input"`
|
|
Function struct {
|
|
Name string `json:"name"`
|
|
Arguments any `json:"arguments"`
|
|
} `json:"function"`
|
|
}
|
|
if err := json.Unmarshal(raw, &call); err != nil {
|
|
return nil, fmt.Errorf("decode preset selector tool call: %w", err)
|
|
}
|
|
name := call.Function.Name
|
|
if name == "" {
|
|
name = call.Name
|
|
}
|
|
arguments := call.Function.Arguments
|
|
if arguments == nil && len(call.Input) > 0 {
|
|
arguments = call.Input
|
|
}
|
|
var rawArgs []byte
|
|
switch typed := arguments.(type) {
|
|
case string:
|
|
rawArgs = []byte(typed)
|
|
case json.RawMessage:
|
|
rawArgs = typed
|
|
default:
|
|
rawArgs, _ = json.Marshal(typed)
|
|
}
|
|
normalized, err := normalizedToolCallFromParts(call.ID, name, string(rawArgs))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, normalized)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func normalizeRunEventToolCalls(metadata map[string]string) ([]normalizedToolCall, error) {
|
|
raw := strings.TrimSpace(metadata[runtimeMetadataOpenAIToolCalls])
|
|
if raw == "" {
|
|
return nil, nil
|
|
}
|
|
var calls []any
|
|
decoder := json.NewDecoder(strings.NewReader(raw))
|
|
decoder.UseNumber()
|
|
if err := decoder.Decode(&calls); err != nil {
|
|
return nil, fmt.Errorf("decode preset selector run tool calls: %w", err)
|
|
}
|
|
if err := requireJSONEOF(decoder); err != nil {
|
|
return nil, fmt.Errorf("decode preset selector run tool calls: %w", err)
|
|
}
|
|
return normalizeProviderToolCalls(calls)
|
|
}
|
|
|
|
func normalizedToolCallFromParts(id, name, rawArgs string) (normalizedToolCall, error) {
|
|
if strings.TrimSpace(id) == "" || strings.TrimSpace(name) == "" {
|
|
return normalizedToolCall{}, fmt.Errorf("preset selector tool call requires id and name")
|
|
}
|
|
if strings.TrimSpace(rawArgs) == "" {
|
|
rawArgs = "{}"
|
|
}
|
|
var arguments map[string]any
|
|
decoder := json.NewDecoder(strings.NewReader(rawArgs))
|
|
decoder.UseNumber()
|
|
if err := decoder.Decode(&arguments); err != nil || arguments == nil {
|
|
return normalizedToolCall{}, fmt.Errorf("preset selector tool call %q has invalid arguments", id)
|
|
}
|
|
if err := requireJSONEOF(decoder); err != nil {
|
|
return normalizedToolCall{}, fmt.Errorf("preset selector tool call %q has invalid arguments", id)
|
|
}
|
|
return normalizedToolCall{ID: id, ProviderCallID: id, Name: name, Arguments: arguments, RawArgs: rawArgs}, nil
|
|
}
|
|
|
|
func requireJSONEOF(decoder *json.Decoder) error {
|
|
var extra any
|
|
if err := decoder.Decode(&extra); err != io.EOF {
|
|
if err == nil {
|
|
return fmt.Errorf("multiple JSON values")
|
|
}
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func mustMarshalRaw(value any) json.RawMessage {
|
|
raw, _ := json.Marshal(value)
|
|
return raw
|
|
}
|
|
|
|
func sseDataPayloads(body []byte) [][]byte {
|
|
normalized := bytes.ReplaceAll(body, []byte("\r\n"), []byte("\n"))
|
|
events := bytes.Split(normalized, []byte("\n\n"))
|
|
var payloads [][]byte
|
|
for _, event := range events {
|
|
var lines [][]byte
|
|
for _, line := range bytes.Split(event, []byte("\n")) {
|
|
line = bytes.TrimSpace(line)
|
|
if bytes.HasPrefix(line, []byte("data:")) {
|
|
lines = append(lines, bytes.TrimSpace(bytes.TrimPrefix(line, []byte("data:"))))
|
|
}
|
|
}
|
|
if len(lines) > 0 {
|
|
payloads = append(payloads, bytes.Join(lines, []byte("\n")))
|
|
}
|
|
}
|
|
return payloads
|
|
}
|
|
|
|
func cloneRawJSON(raw json.RawMessage) json.RawMessage {
|
|
if len(raw) == 0 || string(raw) == "null" {
|
|
return nil
|
|
}
|
|
return append(json.RawMessage(nil), raw...)
|
|
}
|
|
|
|
func decodeOpenAIUsage(raw json.RawMessage) *openAIUsage {
|
|
if len(raw) == 0 || string(raw) == "null" {
|
|
return nil
|
|
}
|
|
var usage openAIUsage
|
|
if json.Unmarshal(raw, &usage) != nil {
|
|
return nil
|
|
}
|
|
return &usage
|
|
}
|
|
|
|
func openAIUsageToAnthropic(raw json.RawMessage) json.RawMessage {
|
|
if len(raw) == 0 {
|
|
return nil
|
|
}
|
|
var usage struct {
|
|
PromptTokens int `json:"prompt_tokens"`
|
|
CompletionTokens int `json:"completion_tokens"`
|
|
PromptDetails struct {
|
|
CachedTokens int `json:"cached_tokens"`
|
|
} `json:"prompt_tokens_details"`
|
|
}
|
|
if json.Unmarshal(raw, &usage) != nil {
|
|
return nil
|
|
}
|
|
converted, _ := json.Marshal(anthropicUsage{
|
|
InputTokens: usage.PromptTokens, OutputTokens: usage.CompletionTokens,
|
|
CacheReadInputTokens: usage.PromptDetails.CachedTokens,
|
|
})
|
|
return converted
|
|
}
|
|
|
|
func openAIReasonToAnthropic(reason string) string {
|
|
switch reason {
|
|
case "tool_calls", "function_call":
|
|
return "tool_use"
|
|
case "length":
|
|
return "max_tokens"
|
|
case "stop", "":
|
|
return "end_turn"
|
|
default:
|
|
return reason
|
|
}
|
|
}
|
|
|
|
func mergeJSONObjects(left, right json.RawMessage) json.RawMessage {
|
|
values := make(map[string]any)
|
|
if len(left) > 0 {
|
|
_ = json.Unmarshal(left, &values)
|
|
}
|
|
if len(right) > 0 {
|
|
var extra map[string]any
|
|
if json.Unmarshal(right, &extra) == nil {
|
|
for key, value := range extra {
|
|
values[key] = value
|
|
}
|
|
}
|
|
}
|
|
if len(values) == 0 {
|
|
return nil
|
|
}
|
|
merged, _ := json.Marshal(values)
|
|
return merged
|
|
}
|
|
|
|
func (s *Server) dispatchPresetTurn(
|
|
w http.ResponseWriter,
|
|
r *http.Request,
|
|
dispatch routeDispatch,
|
|
protocol string,
|
|
stream bool,
|
|
runMeta map[string]string,
|
|
output normalizedStageOutput,
|
|
gate hotPathSelectorGate,
|
|
) error {
|
|
requestID := runMeta["iop_logical_request_id"]
|
|
stageID := runMeta["iop_stage_id"]
|
|
callID := runMeta["iop_call_id"]
|
|
initialAdmission := isInitialHotPathAdmission(runMeta)
|
|
ownerEdgeID := s.edgeIDValue()
|
|
issued := newReservedPaths(requestID)
|
|
preset := dispatch.Preset
|
|
if preset.ID == "" {
|
|
if found, ok := s.ExecutionPreset(dispatch.PresetID); ok {
|
|
preset = found
|
|
}
|
|
}
|
|
decision, err := classifyHotPathOutput(preset, issued, output, gate)
|
|
if err != nil {
|
|
if initialAdmission {
|
|
s.emitHotPathDispatchRejection(r.Context(), hotPathNormalizeMode(string(decision.Mode)), decision.Reason, requestID, stageID, preset.ID)
|
|
}
|
|
s.terminalPresetRequest(requestID, ownerEdgeID)
|
|
writeHotPathPresetDispatchError(w, r, protocol, http.StatusBadRequest, "invalid_request_error", err.Error())
|
|
return err
|
|
}
|
|
if s.artifactFrontiers.pairRequired(requestID, ownerEdgeID) && decision.Mode != modeLight {
|
|
if initialAdmission {
|
|
s.emitHotPathDispatchRejection(r.Context(), hotPathNormalizeMode(string(decision.Mode)), reasonArtifactRequired, requestID, stageID, preset.ID)
|
|
}
|
|
s.terminalPresetRequest(requestID, ownerEdgeID)
|
|
err := fmt.Errorf("artifact frontier requires the exact Plan/Review pair before local-stage handoff")
|
|
writeHotPathPresetDispatchError(w, r, protocol, http.StatusBadRequest, "invalid_request_error", err.Error())
|
|
return err
|
|
}
|
|
|
|
// Only the ingress-created logical request owns admission. Direct tool
|
|
// continuations retain request/stage correlation but never re-admit.
|
|
if initialAdmission {
|
|
s.observeHotPathDispatch(r.Context(), hotPathNormalizeMode(string(decision.Mode)), "", requestID, stageID, preset.ID)
|
|
}
|
|
|
|
switch decision.Mode {
|
|
case modeDirect:
|
|
outer := hotPathCallerOuterTurn(r, protocol, output.ResponseID, hotPathOutputTokenCap(runMeta))
|
|
turn := &hotPathTurn{
|
|
RequestID: requestID, StageID: stageID, CallID: callID, OwnerEdgeID: ownerEdgeID,
|
|
PrincipalRef: runMeta[principalMetaRef], Preset: preset, Dispatch: dispatch,
|
|
Protocol: protocol, Stream: stream, PublicModelID: dispatch.ExternalModelID,
|
|
Writer: w, Request: r, OuterTurn: outer,
|
|
}
|
|
return s.runDirectTurn(r.Context(), turn, output)
|
|
case modeLight:
|
|
outer := hotPathCallerOuterTurn(r, protocol, output.ResponseID, hotPathOutputTokenCap(runMeta))
|
|
turn := &hotPathTurn{
|
|
RequestID: requestID, StageID: stageID, CallID: callID, OwnerEdgeID: ownerEdgeID,
|
|
PrincipalRef: runMeta[principalMetaRef], Preset: preset, Dispatch: dispatch,
|
|
Protocol: protocol, Stream: stream, PublicModelID: dispatch.ExternalModelID,
|
|
Writer: w, Request: r, OuterTurn: outer,
|
|
}
|
|
return s.runArtifactPairTurn(turn, output, gate)
|
|
default:
|
|
if initialAdmission {
|
|
s.emitHotPathDispatchRejection(r.Context(), hotPathNormalizeMode(string(decision.Mode)), reasonModeDisabled, requestID, stageID, preset.ID)
|
|
}
|
|
s.terminalPresetRequest(requestID, ownerEdgeID)
|
|
errMsg := fmt.Sprintf("unsupported mode %q", decision.Mode)
|
|
writeHotPathPresetDispatchError(w, r, protocol, http.StatusBadRequest, "invalid_request_error", errMsg)
|
|
return fmt.Errorf("%s", errMsg)
|
|
}
|
|
}
|
|
|
|
// emitHotPathDispatchRejection records the admission rejection observation for a
|
|
// failed selector/route admission. It maps the decision reason to the closed
|
|
// route reason so raw error text never reaches logs or metric labels.
|
|
func (s *Server) emitHotPathDispatchRejection(ctx context.Context, mode hotPathMode, decisionReason string, requestID, stageID, presetID string) {
|
|
s.observeHotPathDispatch(ctx, mode, hotPathRouteReasonForDecision(decisionReason), requestID, stageID, presetID)
|
|
}
|
|
|
|
func writeHotPathPresetDispatchError(w http.ResponseWriter, r *http.Request, protocol string, status int, errorType, message string) {
|
|
disposition := hotPathTerminalDisposition{
|
|
Kind: hotPathDispositionProviderError, Cause: message, Source: "selector_dispatch",
|
|
}
|
|
if strings.Contains(strings.ToLower(errorType), "invalid") {
|
|
disposition.Kind = hotPathDispositionValidationError
|
|
}
|
|
if protocol == "anthropic" {
|
|
if codec := hotPathAnthropicCodecFromRequest(r); codec != nil {
|
|
codec.w = w
|
|
_ = codec.writeDisposition(disposition, status, errorType, message)
|
|
return
|
|
}
|
|
policy := anthropicHotPathPolicy(disposition)
|
|
writeAnthropicError(w, policy.status, policy.errorType, message)
|
|
return
|
|
}
|
|
turn := &hotPathTurn{Writer: w, Request: r}
|
|
if writeHotPathChatOuterError(turn, status, errorType, message, disposition) {
|
|
return
|
|
}
|
|
policy := chatHotPathPolicy(disposition)
|
|
writeError(w, policy.status, policy.errorType, message)
|
|
}
|
|
|
|
func (s *Server) submitHotPathStage(ctx context.Context, r *http.Request, snapshot hotPathDispatchSnapshot, outer *hotPathOuterTurn) (normalizedStageOutput, hotPathStageCorrelation, error) {
|
|
if err := snapshot.Input.validate(); err != nil {
|
|
return normalizedStageOutput{}, hotPathStageCorrelation{}, newHotPathDispositionError(
|
|
hotPathDispositionValidationError, "stage_input_validation", snapshot.StageID, err,
|
|
)
|
|
}
|
|
prompt, err := snapshot.Input.prompt(snapshot.Phase)
|
|
if err != nil {
|
|
return normalizedStageOutput{}, hotPathStageCorrelation{}, newHotPathDispositionError(
|
|
hotPathDispositionValidationError, "stage_input_validation", snapshot.StageID, err,
|
|
)
|
|
}
|
|
route, err := s.revalidateHotPathStageRoute(ctx, snapshot)
|
|
if err != nil {
|
|
return normalizedStageOutput{}, hotPathStageCorrelation{}, newHotPathDispositionError(
|
|
hotPathDispositionValidationError, "stage_route_validation", snapshot.StageID, err,
|
|
)
|
|
}
|
|
modelGroupKey := route.effectiveModelGroupKey(snapshot.Stage.Model)
|
|
metadata := map[string]string{
|
|
"iop_logical_request_id": snapshot.RequestID,
|
|
"iop_stage_id": snapshot.StageID,
|
|
"iop_stage_role": snapshot.Input.Role,
|
|
}
|
|
if snapshot.PrincipalRef != "" {
|
|
metadata[principalMetaRef] = snapshot.PrincipalRef
|
|
}
|
|
applyTrustedManagedBindingMetadata(metadata, route)
|
|
estimate := estimateInputTokensBytes([]byte(prompt), metadata, snapshot.Tools, nil)
|
|
contextClass := classifyContext(estimate, s.longContextThreshold())
|
|
runInput := hotPathStageRunInput(snapshot, prompt)
|
|
runReq := edgeservice.SubmitRunRequest{
|
|
NodeRef: route.NodeRef, ModelGroupKey: modelGroupKey, ProviderID: route.ProviderID,
|
|
UsageAttribution: route.UsageAttribution, Adapter: route.Adapter, Target: route.Target,
|
|
SessionID: route.SessionID, Prompt: prompt, Input: runInput, TimeoutSec: route.TimeoutSec,
|
|
MaxQueue: route.MaxQueue, QueueTimeoutMS: route.QueueTimeoutMS, Metadata: metadata,
|
|
EstimatedInputTokens: estimate, ContextClass: contextClass, ProviderPool: route.ProviderPool,
|
|
}
|
|
|
|
if !route.ProviderPool {
|
|
if routeUsesProviderTunnel(route) {
|
|
tunnelReq := hotPathStageTunnelRequest(snapshot, route, modelGroupKey, metadata, estimate, contextClass)
|
|
tunnelReq.Operation = string(config.OperationChatCompletions)
|
|
tunnelReq.Path = "/v1/chat/completions"
|
|
tunnelReq.BuildBody = func(target string) ([]byte, error) {
|
|
return hotPathChatStageBody(snapshot, prompt, target)
|
|
}
|
|
headers, headerErr := s.providerTunnelAuthHeaders(r)
|
|
if headerErr != nil {
|
|
return normalizedStageOutput{}, hotPathStageCorrelation{}, headerErr
|
|
}
|
|
tunnelReq.Headers = headers
|
|
handle, submitErr := s.service.SubmitProviderTunnel(ctx, tunnelReq)
|
|
if submitErr != nil {
|
|
return normalizedStageOutput{}, hotPathStageCorrelation{}, submitErr
|
|
}
|
|
dispatch := handle.Dispatch()
|
|
if shouldProgressivelyReleaseHotPathStage(snapshot, outer) {
|
|
return s.runHotPathLiveTunnelStage(ctx, snapshot, outer, handle, dispatch)
|
|
}
|
|
output, collectErr := s.collectHotPathOwnedTunnelStage(ctx, snapshot.StageID, outer, handle, dispatch, "openai")
|
|
if collectErr != nil {
|
|
return normalizedStageOutput{}, hotPathStageCorrelation{}, collectErr
|
|
}
|
|
return output, stageCorrelation(snapshot.StageID, output, dispatch), nil
|
|
}
|
|
handle, submitErr := s.service.SubmitRun(ctx, runReq)
|
|
if submitErr != nil {
|
|
return normalizedStageOutput{}, hotPathStageCorrelation{}, submitErr
|
|
}
|
|
dispatch := handle.Dispatch()
|
|
if shouldProgressivelyReleaseHotPathStage(snapshot, outer) {
|
|
return s.runHotPathLiveNormalizedStage(ctx, snapshot, outer, handle, dispatch)
|
|
}
|
|
output, collectErr := s.collectHotPathOwnedNormalizedStage(ctx, snapshot.StageID, outer, handle, dispatch)
|
|
if collectErr != nil {
|
|
return normalizedStageOutput{}, hotPathStageCorrelation{}, collectErr
|
|
}
|
|
return output, stageCorrelation(snapshot.StageID, output, dispatch), nil
|
|
}
|
|
|
|
poolReq := edgeservice.ProviderPoolDispatchRequest{
|
|
Run: runReq,
|
|
Tunnel: hotPathStageTunnelRequest(snapshot, route, modelGroupKey, metadata, estimate, contextClass),
|
|
}
|
|
poolReq.AcceptCandidate = hotPathStageCandidatePredicate(snapshot)
|
|
if route.Managed {
|
|
poolReq.AcceptCandidate = composeCandidatePredicates(poolReq.AcceptCandidate, route.CandidatePredicate())
|
|
}
|
|
poolReq.PrepareProtocolTunnel = s.prepareHotPathStageTunnel(r, snapshot, prompt)
|
|
result, err := s.service.SubmitProviderPool(ctx, poolReq)
|
|
if err != nil {
|
|
return normalizedStageOutput{}, hotPathStageCorrelation{}, err
|
|
}
|
|
if result == nil {
|
|
return normalizedStageOutput{}, hotPathStageCorrelation{}, fmt.Errorf("hot path stage returned no provider result")
|
|
}
|
|
rejection := s.newHotPathRejectedDispatchOwner(result)
|
|
if err := validateHotPathStageResultShape(result); err != nil {
|
|
s.abortHotPathRejectedDispatch(rejection)
|
|
return normalizedStageOutput{}, hotPathStageCorrelation{}, newHotPathDispositionError(
|
|
hotPathDispositionValidationError, "stage_dispatch_path", snapshot.StageID,
|
|
err,
|
|
)
|
|
}
|
|
if err := validateHotPathStageDispatch(snapshot, route, result.DispatchInfo); err != nil {
|
|
s.abortHotPathRejectedDispatch(rejection)
|
|
return normalizedStageOutput{}, hotPathStageCorrelation{}, newHotPathDispositionError(
|
|
hotPathDispositionValidationError, "stage_dispatch_validation", snapshot.StageID, err,
|
|
)
|
|
}
|
|
var output normalizedStageOutput
|
|
if shouldProgressivelyReleaseHotPathStage(snapshot, outer) {
|
|
switch result.Path {
|
|
case edgeservice.ProviderPoolPathNormalized:
|
|
return s.runHotPathLiveNormalizedStage(ctx, snapshot, outer, result.Run, result.DispatchInfo)
|
|
case edgeservice.ProviderPoolPathTunnel:
|
|
return s.runHotPathLiveTunnelStage(ctx, snapshot, outer, result.Tunnel, result.DispatchInfo)
|
|
default:
|
|
s.abortHotPathRejectedDispatch(rejection)
|
|
return normalizedStageOutput{}, hotPathStageCorrelation{}, newHotPathDispositionError(
|
|
hotPathDispositionValidationError, "stage_dispatch_path", snapshot.StageID,
|
|
fmt.Errorf("hot path stage returned unsupported execution path %q", result.Path),
|
|
)
|
|
}
|
|
}
|
|
switch result.Path {
|
|
case edgeservice.ProviderPoolPathNormalized:
|
|
output, err = s.collectHotPathOwnedNormalizedStage(ctx, snapshot.StageID, outer, result.Run, result.DispatchInfo)
|
|
case edgeservice.ProviderPoolPathTunnel:
|
|
output, err = s.collectHotPathOwnedTunnelStage(
|
|
ctx, snapshot.StageID, outer, result.Tunnel, result.DispatchInfo, hotPathStageWireProtocol(result.DispatchInfo),
|
|
)
|
|
default:
|
|
s.abortHotPathRejectedDispatch(rejection)
|
|
return normalizedStageOutput{}, hotPathStageCorrelation{}, newHotPathDispositionError(
|
|
hotPathDispositionValidationError, "stage_dispatch_path", snapshot.StageID,
|
|
fmt.Errorf("hot path stage returned unsupported execution path %q", result.Path),
|
|
)
|
|
}
|
|
if err != nil {
|
|
return normalizedStageOutput{}, hotPathStageCorrelation{}, err
|
|
}
|
|
if strings.TrimSpace(output.ResponseID) == "" {
|
|
return normalizedStageOutput{}, hotPathStageCorrelation{}, fmt.Errorf("hot path stage completion is missing provider identity")
|
|
}
|
|
return output, stageCorrelation(snapshot.StageID, output, result.DispatchInfo), nil
|
|
}
|
|
|
|
func shouldProgressivelyReleaseHotPathStage(snapshot hotPathDispatchSnapshot, outer *hotPathOuterTurn) bool {
|
|
return snapshot.Stream && (snapshot.Protocol == "openai" || snapshot.Protocol == "anthropic") && outer != nil
|
|
}
|
|
|
|
// newHotPathRejectedDispatchOwner builds one result-scoped disposal owner for a
|
|
// provider-pool result whose ownership has already transferred to Edge but which
|
|
// a local selector/downstream rejection will not consume. It reuses the
|
|
// exact-once hotPathStageTransportController claim: cancellation targets the
|
|
// immutable DispatchInfo (independent of which handle variant produced the
|
|
// rejection) and the close callback closes every non-nil returned handle. A nil
|
|
// result yields a nil owner. Because the claim is taken once, observing the same
|
|
// rejection repeatedly still sends exactly one CANCEL_RUN and closes each
|
|
// returned handle exactly once.
|
|
func (s *Server) newHotPathRejectedDispatchOwner(result *edgeservice.ProviderPoolDispatchResult) *hotPathStageTransportController {
|
|
if result == nil {
|
|
return nil
|
|
}
|
|
return newHotPathStageTransportController(s.service, result.DispatchInfo, func() {
|
|
if result.Run != nil {
|
|
result.Run.Close()
|
|
}
|
|
if result.Tunnel != nil {
|
|
result.Tunnel.Close()
|
|
}
|
|
})
|
|
}
|
|
|
|
// abortHotPathRejectedDispatch disposes an owned provider-pool result through its
|
|
// result-scoped owner: one exact CancelRun(CANCEL_RUN) to Node followed by a
|
|
// close of every returned handle. A nil owner (nil result) is a no-op, and every
|
|
// selector/downstream rejection branch shares one owner instance so repeated
|
|
// aborts collapse to a single cancel and a single close per handle.
|
|
func (s *Server) abortHotPathRejectedDispatch(owner *hotPathStageTransportController) {
|
|
if owner == nil {
|
|
return
|
|
}
|
|
if err := owner.AbortAttempt(context.Background()); err != nil {
|
|
s.logger.Warn("hot path rejected dispatch cancellation failed", zap.Error(err))
|
|
}
|
|
}
|
|
|
|
// validateHotPathStageResultShape accepts only the provider-pool result shape
|
|
// that can be consumed by the selected execution path. This boundary runs
|
|
// before either buffered or progressive dispatch so every invalid owned result
|
|
// is cancelled and closed by the result-scoped rejection owner.
|
|
func validateHotPathStageResultShape(result *edgeservice.ProviderPoolDispatchResult) error {
|
|
if result == nil {
|
|
return fmt.Errorf("hot path stage returned no provider result")
|
|
}
|
|
switch result.Path {
|
|
case edgeservice.ProviderPoolPathNormalized:
|
|
if result.Run == nil {
|
|
return fmt.Errorf("hot path normalized result is missing run handle")
|
|
}
|
|
if result.Tunnel != nil {
|
|
return fmt.Errorf("hot path normalized result returned unexpected tunnel handle")
|
|
}
|
|
case edgeservice.ProviderPoolPathTunnel:
|
|
if result.Tunnel == nil {
|
|
return fmt.Errorf("hot path tunnel result is missing tunnel handle")
|
|
}
|
|
if result.Run != nil {
|
|
return fmt.Errorf("hot path tunnel result returned unexpected run handle")
|
|
}
|
|
default:
|
|
return fmt.Errorf("hot path stage returned unsupported execution path %q", result.Path)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Server) runHotPathLiveNormalizedStage(
|
|
ctx context.Context,
|
|
snapshot hotPathDispatchSnapshot,
|
|
outer *hotPathOuterTurn,
|
|
handle edgeservice.RunResult,
|
|
dispatch edgeservice.RunDispatch,
|
|
) (normalizedStageOutput, hotPathStageCorrelation, error) {
|
|
if handle == nil {
|
|
return normalizedStageOutput{}, hotPathStageCorrelation{}, fmt.Errorf("hot path normalized stage returned no run result")
|
|
}
|
|
source := newHotPathNormalizedStageSource(handle.Stream(), handle.WaitTimeout())
|
|
controller := newHotPathStageTransportController(s.service, dispatch, handle.Close)
|
|
output, terminal, err := runHotPathStreamingStage(
|
|
ctx, outer, hotPathStageMetaFromDispatch(snapshot.StageID, dispatch), source, source, controller,
|
|
)
|
|
if err != nil {
|
|
return normalizedStageOutput{}, hotPathStageCorrelation{}, err
|
|
}
|
|
if !terminal.Success {
|
|
return normalizedStageOutput{}, hotPathStageCorrelation{}, wrapHotPathDispositionError(
|
|
outer, snapshot.StageID, fmt.Errorf("hot path normalized stage failed"),
|
|
)
|
|
}
|
|
if strings.TrimSpace(output.ResponseID) == "" {
|
|
return normalizedStageOutput{}, hotPathStageCorrelation{}, fmt.Errorf("hot path normalized stage completion is missing provider identity")
|
|
}
|
|
return output, stageCorrelation(snapshot.StageID, output, dispatch), nil
|
|
}
|
|
|
|
func (s *Server) runHotPathLiveTunnelStage(
|
|
ctx context.Context,
|
|
snapshot hotPathDispatchSnapshot,
|
|
outer *hotPathOuterTurn,
|
|
handle edgeservice.ProviderTunnelResult,
|
|
dispatch edgeservice.RunDispatch,
|
|
) (normalizedStageOutput, hotPathStageCorrelation, error) {
|
|
if handle == nil {
|
|
return normalizedStageOutput{}, hotPathStageCorrelation{}, fmt.Errorf("hot path tunnel stage returned no provider result")
|
|
}
|
|
decoder := newHotPathStageDecoderForProtocol(hotPathStageWireProtocol(dispatch))
|
|
source := newHotPathTunnelStageSource(handle.Stream(), handle.WaitTimeout(), decoder)
|
|
controller := newHotPathStageTransportController(s.service, dispatch, handle.Close)
|
|
output, terminal, err := runHotPathStreamingStage(
|
|
ctx, outer, hotPathStageMetaFromDispatch(snapshot.StageID, dispatch), source, source, controller,
|
|
)
|
|
if err != nil {
|
|
return normalizedStageOutput{}, hotPathStageCorrelation{}, err
|
|
}
|
|
if !terminal.Success {
|
|
return normalizedStageOutput{}, hotPathStageCorrelation{}, wrapHotPathDispositionError(
|
|
outer, snapshot.StageID, fmt.Errorf("hot path tunnel stage failed"),
|
|
)
|
|
}
|
|
if strings.TrimSpace(output.ResponseID) == "" {
|
|
return normalizedStageOutput{}, hotPathStageCorrelation{}, fmt.Errorf("hot path tunnel stage completion is missing provider identity")
|
|
}
|
|
return output, stageCorrelation(snapshot.StageID, output, dispatch), nil
|
|
}
|
|
|
|
func hotPathStageTunnelRequest(snapshot hotPathDispatchSnapshot, route routeDispatch, modelGroupKey string, metadata map[string]string, estimate int, contextClass string) edgeservice.SubmitProviderTunnelRequest {
|
|
return edgeservice.SubmitProviderTunnelRequest{
|
|
CredentialBinding: route.credentialBinding(), ModelGroupKey: modelGroupKey,
|
|
ProviderID: route.ProviderID, UsageAttribution: route.UsageAttribution,
|
|
SessionID: route.SessionID, Method: http.MethodPost, Stream: snapshot.Stream,
|
|
TimeoutSec: route.TimeoutSec, MaxQueue: route.MaxQueue, QueueTimeoutMS: route.QueueTimeoutMS,
|
|
Metadata: metadata, EstimatedInputTokens: estimate, ContextClass: contextClass, ProviderPool: route.ProviderPool,
|
|
}
|
|
}
|
|
|
|
func (s *Server) prepareHotPathStageTunnel(r *http.Request, snapshot hotPathDispatchSnapshot, prompt string) func(edgeservice.SubmitProviderTunnelRequest, edgeservice.ProviderPoolCandidate) (edgeservice.SubmitProviderTunnelRequest, error) {
|
|
return func(tunnelReq edgeservice.SubmitProviderTunnelRequest, selected edgeservice.ProviderPoolCandidate) (edgeservice.SubmitProviderTunnelRequest, error) {
|
|
if selected.ProtocolProfile == nil {
|
|
headers, err := s.providerTunnelAuthHeaders(r)
|
|
if err != nil {
|
|
return tunnelReq, err
|
|
}
|
|
tunnelReq.Headers = headers
|
|
tunnelReq.Path = "/v1/chat/completions"
|
|
tunnelReq.Operation = string(config.OperationChatCompletions)
|
|
tunnelReq.BuildBody = func(target string) ([]byte, error) {
|
|
return hotPathChatStageBody(snapshot, prompt, target)
|
|
}
|
|
return tunnelReq, nil
|
|
}
|
|
profile := selected.ProtocolProfile.Clone()
|
|
switch profile.Driver {
|
|
case config.ProtocolDriverOpenAIChat:
|
|
prepared, err := s.protocolTunnelPreparer(r, config.OperationChatCompletions)(tunnelReq, selected)
|
|
if err != nil {
|
|
return tunnelReq, err
|
|
}
|
|
prepared.Path = "/v1/chat/completions"
|
|
prepared.BuildBody = func(target string) ([]byte, error) {
|
|
return hotPathChatStageBody(snapshot, prompt, target)
|
|
}
|
|
return prepared, nil
|
|
case config.ProtocolDriverAnthropicMessages:
|
|
request := r.Clone(r.Context())
|
|
if strings.TrimSpace(request.Header.Get(anthropicVersionHeader)) == "" {
|
|
request.Header.Set(anthropicVersionHeader, anthropicSupportedVersion)
|
|
}
|
|
headers, err := s.anthropicUpstreamHeaders(request, profile, true)
|
|
if err != nil {
|
|
return tunnelReq, err
|
|
}
|
|
tunnelReq.Headers = headers
|
|
tunnelReq.Path = "/v1/messages"
|
|
tunnelReq.Operation = string(config.OperationMessages)
|
|
tunnelReq.BuildBody = func(target string) ([]byte, error) {
|
|
return hotPathAnthropicStageBody(snapshot, prompt, target)
|
|
}
|
|
return tunnelReq, nil
|
|
default:
|
|
return tunnelReq, fmt.Errorf("hot path stage does not support protocol driver %q", profile.Driver)
|
|
}
|
|
}
|
|
}
|
|
|
|
func hotPathStageCandidatePredicate(snapshot hotPathDispatchSnapshot) edgeservice.ProviderPoolCandidatePredicate {
|
|
needsTools := len(snapshot.Tools) > 0
|
|
return func(candidate edgeservice.ProviderPoolCandidate) bool {
|
|
if candidate.ExecutionPath == string(edgeservice.ProviderPoolPathNormalized) {
|
|
return true
|
|
}
|
|
profile := candidate.ProtocolProfile
|
|
if profile == nil {
|
|
return true
|
|
}
|
|
if snapshot.Stream && !profile.HasCapability("streaming") {
|
|
return false
|
|
}
|
|
if needsTools && !profile.HasCapability("tool_calling") {
|
|
return false
|
|
}
|
|
switch profile.Driver {
|
|
case config.ProtocolDriverOpenAIChat:
|
|
return profile.HasCapability("chat") && profileHasOperation(*profile, config.OperationChatCompletions)
|
|
case config.ProtocolDriverAnthropicMessages:
|
|
return profile.HasCapability("messages") && profileHasOperation(*profile, config.OperationMessages)
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Server) revalidateHotPathStageRoute(ctx context.Context, snapshot hotPathDispatchSnapshot) (routeDispatch, error) {
|
|
pinned := snapshot.Route
|
|
if !pinned.Managed {
|
|
return pinned, nil
|
|
}
|
|
currentPreset, err := s.resolveRouteDispatchForPrincipal(ctx, snapshot.PresetRoute.ExternalModelID)
|
|
if err != nil {
|
|
return routeDispatch{}, fmt.Errorf("revalidate hot path stage route: %w", err)
|
|
}
|
|
current, ok := currentPreset.PresetResolvedBindings[snapshot.Stage.Model]
|
|
if !ok || !samePinnedHotPathRoute(pinned, current) {
|
|
return routeDispatch{}, fmt.Errorf("hot path stage route or credential revision changed")
|
|
}
|
|
return current, nil
|
|
}
|
|
|
|
func samePinnedHotPathRoute(left, right routeDispatch) bool {
|
|
return left.Managed == right.Managed && left.PrincipalRef == right.PrincipalRef &&
|
|
left.ModelGroupKey == right.ModelGroupKey && left.RouteID == right.RouteID &&
|
|
left.CredentialSlotRef == right.CredentialSlotRef && left.ProfileID == right.ProfileID &&
|
|
left.UpstreamModel == right.UpstreamModel && left.ResourceSelector == right.ResourceSelector &&
|
|
left.RouteRevision == right.RouteRevision && left.CredentialRevision == right.CredentialRevision &&
|
|
left.ProjectionGeneration == right.ProjectionGeneration
|
|
}
|
|
|
|
func validateHotPathStageDispatch(snapshot hotPathDispatchSnapshot, route routeDispatch, selected edgeservice.RunDispatch) error {
|
|
if strings.TrimSpace(selected.RunID) == "" || strings.TrimSpace(selected.NodeID) == "" || strings.TrimSpace(selected.ProviderID) == "" {
|
|
return fmt.Errorf("hot path stage dispatch correlation is incomplete")
|
|
}
|
|
if selected.ModelGroupKey != route.effectiveModelGroupKey(snapshot.Stage.Model) {
|
|
return fmt.Errorf("hot path stage model binding changed after admission")
|
|
}
|
|
if route.ProviderID != "" && selected.ProviderID != route.ProviderID {
|
|
return fmt.Errorf("hot path stage provider binding changed after admission")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func stageCorrelation(stageID string, output normalizedStageOutput, dispatch edgeservice.RunDispatch) hotPathStageCorrelation {
|
|
return hotPathStageCorrelation{
|
|
StageID: stageID, ResponseID: output.ResponseID, RunID: dispatch.RunID,
|
|
ProviderID: dispatch.ProviderID, Terminal: output.TerminalReason,
|
|
}
|
|
}
|
|
|
|
// hotPathStageWireProtocol maps a committed stage dispatch to its provider wire
|
|
// protocol. The tunnel decode and the HTTP-turn stage source both select their
|
|
// decoder from this single fact rather than the caller endpoint.
|
|
func hotPathStageWireProtocol(dispatch edgeservice.RunDispatch) string {
|
|
if dispatch.ProfileDriver == string(config.ProtocolDriverAnthropicMessages) {
|
|
return "anthropic"
|
|
}
|
|
return "openai"
|
|
}
|
|
|
|
// hotPathStageMetaFromDispatch exposes the protocol-neutral stage correlation
|
|
// the HTTP-turn core consumes as a stage-source input. It carries only committed
|
|
// model/provider/path identity and never performs caller endpoint encoding.
|
|
func hotPathStageMetaFromDispatch(stageID string, dispatch edgeservice.RunDispatch) hotPathStageMeta {
|
|
return hotPathStageMeta{
|
|
StageID: stageID,
|
|
Protocol: hotPathStageWireProtocol(dispatch),
|
|
Model: dispatch.ModelGroupKey,
|
|
Provider: dispatch.ProviderID,
|
|
ExecutionPath: dispatch.ExecutionPath,
|
|
AttemptID: dispatch.RunID,
|
|
}
|
|
}
|
|
|
|
func hotPathStageRunInput(snapshot hotPathDispatchSnapshot, prompt string) map[string]any {
|
|
messages := hotPathChatStageMessages(snapshot, prompt)
|
|
input := map[string]any{"prompt": prompt, "messages": messages}
|
|
if tools := hotPathChatTools(snapshot.Tools); len(tools) > 0 {
|
|
input["tools"] = tools
|
|
input["tool_choice"] = "auto"
|
|
}
|
|
options := cloneAnyMap(snapshot.Stage.Options)
|
|
if options == nil {
|
|
options = make(map[string]any)
|
|
}
|
|
if snapshot.OutputBudget.Limited {
|
|
options["max_tokens"] = snapshot.OutputBudget.Remaining
|
|
}
|
|
if len(options) > 0 {
|
|
input["options"] = options
|
|
}
|
|
return input
|
|
}
|
|
|
|
func hotPathChatStageBody(snapshot hotPathDispatchSnapshot, prompt, target string) ([]byte, error) {
|
|
body := map[string]any{
|
|
"model": target, "messages": hotPathChatStageMessages(snapshot, prompt), "stream": snapshot.Stream,
|
|
}
|
|
if tools := hotPathChatTools(snapshot.Tools); len(tools) > 0 {
|
|
body["tools"] = tools
|
|
body["tool_choice"] = "auto"
|
|
}
|
|
reserved := map[string]struct{}{"model": {}, "messages": {}, "tools": {}, "stream": {}}
|
|
if snapshot.OutputBudget.Limited {
|
|
body["max_tokens"] = snapshot.OutputBudget.Remaining
|
|
reserved["max_tokens"] = struct{}{}
|
|
}
|
|
applyHotPathStageOptions(body, snapshot.Stage.Options, reserved)
|
|
return json.Marshal(body)
|
|
}
|
|
|
|
func hotPathAnthropicStageBody(snapshot hotPathDispatchSnapshot, prompt, target string) ([]byte, error) {
|
|
body := map[string]any{
|
|
"model": target, "max_tokens": 4096, "messages": hotPathAnthropicStageMessages(snapshot, prompt), "stream": snapshot.Stream,
|
|
}
|
|
if tools := hotPathAnthropicTools(snapshot.Tools); len(tools) > 0 {
|
|
body["tools"] = tools
|
|
body["tool_choice"] = map[string]any{"type": "auto"}
|
|
}
|
|
reserved := map[string]struct{}{"model": {}, "messages": {}, "tools": {}, "stream": {}}
|
|
if snapshot.OutputBudget.Limited {
|
|
body["max_tokens"] = snapshot.OutputBudget.Remaining
|
|
reserved["max_tokens"] = struct{}{}
|
|
}
|
|
applyHotPathStageOptions(body, snapshot.Stage.Options, reserved)
|
|
return json.Marshal(body)
|
|
}
|
|
|
|
func applyHotPathStageOptions(body map[string]any, options map[string]any, reserved map[string]struct{}) {
|
|
for key, value := range options {
|
|
if _, blocked := reserved[key]; blocked {
|
|
continue
|
|
}
|
|
body[key] = cloneAnyValue(value)
|
|
}
|
|
}
|
|
|
|
func hotPathChatStageMessages(snapshot hotPathDispatchSnapshot, prompt string) []any {
|
|
messages := []any{map[string]any{"role": "user", "content": prompt}}
|
|
for _, exchange := range snapshot.Transcript {
|
|
assistant := map[string]any{"role": "assistant", "content": exchange.Output.Content}
|
|
if exchange.Output.Reasoning != "" {
|
|
assistant["reasoning_content"] = exchange.Output.Reasoning
|
|
}
|
|
if len(exchange.Output.ToolCalls) > 0 {
|
|
calls := make([]any, 0, len(exchange.Output.ToolCalls))
|
|
for _, call := range exchange.Output.ToolCalls {
|
|
providerID := call.ProviderCallID
|
|
if providerID == "" {
|
|
providerID = call.ID
|
|
}
|
|
calls = append(calls, map[string]any{
|
|
"id": providerID, "type": "function",
|
|
"function": map[string]any{"name": call.Name, "arguments": directToolArguments(call)},
|
|
})
|
|
}
|
|
assistant["tool_calls"] = calls
|
|
}
|
|
messages = append(messages, assistant)
|
|
for _, result := range exchange.Results {
|
|
messages = append(messages, map[string]any{
|
|
"role": "tool", "tool_call_id": result.ProviderCallID, "content": result.Body,
|
|
})
|
|
}
|
|
}
|
|
return messages
|
|
}
|
|
|
|
func hotPathAnthropicStageMessages(snapshot hotPathDispatchSnapshot, prompt string) []any {
|
|
messages := []any{map[string]any{"role": "user", "content": prompt}}
|
|
for _, exchange := range snapshot.Transcript {
|
|
blocks := anthropicDirectBlocks(exchange.Output)
|
|
for _, block := range blocks {
|
|
if block["type"] == "tool_use" {
|
|
for _, call := range exchange.Output.ToolCalls {
|
|
if block["id"] == call.ID && call.ProviderCallID != "" {
|
|
block["id"] = call.ProviderCallID
|
|
}
|
|
}
|
|
}
|
|
}
|
|
messages = append(messages, map[string]any{"role": "assistant", "content": blocks})
|
|
results := make([]any, 0, len(exchange.Results))
|
|
for _, result := range exchange.Results {
|
|
results = append(results, map[string]any{
|
|
"type": "tool_result", "tool_use_id": result.ProviderCallID,
|
|
"content": result.Body, "is_error": result.IsError,
|
|
})
|
|
}
|
|
messages = append(messages, map[string]any{"role": "user", "content": results})
|
|
}
|
|
return messages
|
|
}
|
|
|
|
func hotPathChatTools(tools []any) []any {
|
|
schemas, _ := normalizeToolSchemas(tools)
|
|
names := make([]string, 0, len(schemas))
|
|
for name := range schemas {
|
|
names = append(names, name)
|
|
}
|
|
sort.Strings(names)
|
|
out := make([]any, 0, len(names))
|
|
for _, name := range names {
|
|
schema := schemas[name]
|
|
function := map[string]any{"name": schema.name, "parameters": cloneAnyMap(schema.schema)}
|
|
if schema.description != "" {
|
|
function["description"] = schema.description
|
|
}
|
|
out = append(out, map[string]any{"type": "function", "function": function})
|
|
}
|
|
return out
|
|
}
|
|
|
|
func hotPathAnthropicTools(tools []any) []any {
|
|
schemas, _ := normalizeToolSchemas(tools)
|
|
names := make([]string, 0, len(schemas))
|
|
for name := range schemas {
|
|
names = append(names, name)
|
|
}
|
|
sort.Strings(names)
|
|
out := make([]any, 0, len(names))
|
|
for _, name := range names {
|
|
schema := schemas[name]
|
|
tool := map[string]any{"name": schema.name, "input_schema": cloneAnyMap(schema.schema)}
|
|
if schema.description != "" {
|
|
tool["description"] = schema.description
|
|
}
|
|
out = append(out, tool)
|
|
}
|
|
return out
|
|
}
|
|
|
|
func (s *Server) terminalPresetRequest(requestID, ownerEdgeID string) {
|
|
if requestID != "" {
|
|
if s.lightFlows != nil {
|
|
s.lightFlows.remove(requestID, ownerEdgeID)
|
|
}
|
|
if s.artifactFrontiers != nil {
|
|
s.artifactFrontiers.remove(requestID, ownerEdgeID)
|
|
}
|
|
_ = s.requestCoordinator.terminal(requestID, ownerEdgeID)
|
|
}
|
|
}
|