toki
e5869a53ae
caller별 예외 대신 요청 의미와 protocol profile capability로 operation을 선택해 tools와 effort 조합을 보존한다. 지원하지 않는 effort는 가장 가까운 하위 등급으로만 내리고 상향 매핑은 거부한다.
1824 lines
68 KiB
Go
1824 lines
68 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,
|
|
ProfileOperation: selected.ProfileOperation,
|
|
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.ProfileOperation, 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, operation, driver string, capabilities []string) bool {
|
|
required := "chat"
|
|
if protocol == "anthropic" {
|
|
switch operation {
|
|
case string(config.OperationMessages):
|
|
required = "messages"
|
|
case string(config.OperationResponses):
|
|
required = "responses"
|
|
case string(config.OperationChatCompletions):
|
|
required = "chat"
|
|
default:
|
|
if 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.ProfileOperation, 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.ProfileOperation == string(config.OperationMessages) {
|
|
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, operation, driver string) (normalizedStageOutput, error) {
|
|
streaming := strings.Contains(strings.ToLower(contentType), "text/event-stream") || bytes.Contains(body, []byte("data:"))
|
|
if protocol == "anthropic" && (operation == string(config.OperationMessages) || (operation == "" && driver == string(config.ProtocolDriverAnthropicMessages))) {
|
|
if streaming {
|
|
return decodeAnthropicPresetSSE(body)
|
|
}
|
|
return decodeAnthropicPresetJSON(body)
|
|
}
|
|
if protocol == "anthropic" && operation == string(config.OperationResponses) {
|
|
return decodeResponsesPresetBody(body, streaming)
|
|
}
|
|
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 decodeResponsesPresetBody(body []byte, streaming bool) (normalizedStageOutput, error) {
|
|
responseBody := body
|
|
if streaming {
|
|
for _, payload := range splitOpenAIResponsesSSE(body) {
|
|
var event struct {
|
|
Type string `json:"type"`
|
|
Response json.RawMessage `json:"response"`
|
|
}
|
|
if json.Unmarshal(payload, &event) == nil && event.Type == "response.completed" && len(event.Response) > 0 {
|
|
responseBody = event.Response
|
|
}
|
|
}
|
|
}
|
|
converted, err := convertResponsesResponseToAnthropic(responseBody, "")
|
|
if err != nil {
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
raw, err := json.Marshal(converted)
|
|
if err != nil {
|
|
return normalizedStageOutput{}, err
|
|
}
|
|
return decodeAnthropicPresetJSON(raw)
|
|
}
|
|
|
|
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.ProfileOperation == string(config.OperationMessages) ||
|
|
(dispatch.ProfileOperation == "" && 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)
|
|
}
|
|
}
|