iop/apps/edge/internal/openai/hot_path_dispatch.go

package openai

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"sort"
	"strings"
	"time"

	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) {
	if result == nil {
		return normalizedStageOutput{}, 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)

	var (
		stage normalizedStageOutput
		err   error
	)
	switch result.Path {
	case edgeservice.ProviderPoolPathNormalized:
		stage, err = collectPresetNormalizedResult(ctx, result.Run, selected)
	case edgeservice.ProviderPoolPathTunnel:
		stage, err = collectPresetTunnelResult(ctx, result.Tunnel, selected, protocol)
	default:
		err = fmt.Errorf("preset selector returned unsupported execution path %q", result.Path)
	}
	return stage, gate, err
}

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 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")
	}
	defer handle.Close()
	if err := validateSelectedDispatch(selected, handle.Dispatch()); err != nil {
		return normalizedStageOutput{}, 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{ResponseID: selected.RunID}
	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.GetRunId() != "" {
				stage.ResponseID = event.GetRunId()
			}
			if event.GetTimestamp() != 0 {
				stage.Created = unixSeconds(event.GetTimestamp())
			}
			switch event.GetType() {
			case "delta":
				content.WriteString(event.GetDelta())
			case "reasoning_delta":
				reasoning.WriteString(event.GetDelta())
			case "complete":
				stage.Content = content.String()
				stage.Reasoning = reasoning.String()
				stage.TerminalReason = strings.TrimSpace(event.GetMetadata()["finish_reason"])
				if stage.TerminalReason == "" {
					stage.TerminalReason = "stop"
				}
				var err error
				stage.ToolCalls, err = normalizeRunEventToolCalls(event.GetMetadata())
				if err != nil {
					return normalizedStageOutput{}, err
				}
				if len(stage.ToolCalls) > 0 {
					stage.TerminalReason = "tool_calls"
				}
				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)
			}
		}
	}
}

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")
	}
	defer handle.Close()
	if err := validateSelectedDispatch(selected, handle.Dispatch()); err != nil {
		return normalizedStageOutput{}, 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),
	}
	stage.OpenAIUsage = decodeOpenAIUsage(response.Usage)
	return stage, nil
}

func decodeOpenAIPresetSSE(body []byte) (normalizedStageOutput, error) {
	stage := normalizedStageOutput{}
	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 chunk.ID != "" {
			stage.ResponseID = chunk.ID
		}
		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 {
			stage.Content += choice.Delta.Content
			reasoning := choice.Delta.ReasoningContent
			if reasoning == "" {
				reasoning = choice.Delta.Reasoning
			}
			stage.Reasoning += 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 choice.FinishReason != nil {
				stage.TerminalReason = *choice.FinishReason
			}
		}
	}
	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{}
	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)
			}
			stage.ResponseID = message.ID
			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
			}
			switch start.Block.Type {
			case "text":
				stage.Content += start.Block.Text
			case "thinking":
				stage.Reasoning += start.Block.Thinking
				stage.ReasoningSignature += start.Block.Signature
			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)
				}
				tools[start.Index] = state
			}
		case "content_block_delta":
			var delta struct {
				Index int `json:"index"`
				Delta struct {
					Type, Text, Thinking, Signature, PartialJSON string
				} `json:"delta"`
			}
			if err := json.Unmarshal(payload, &delta); err != nil {
				return normalizedStageOutput{}, err
			}
			switch delta.Delta.Type {
			case "text_delta":
				stage.Content += delta.Delta.Text
			case "thinking_delta":
				stage.Reasoning += 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)
				}
			}
		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")
		}
	}
	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
	case "thinking":
		stage.Reasoning += block.Thinking
		stage.ReasoningSignature += block.Signature
	case "tool_use":
		call, err := normalizedToolCallFromParts(block.ID, block.Name, string(block.Input))
		if err != nil {
			return err
		}
		stage.ToolCalls = append(stage.ToolCalls, 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"]
	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 {
		s.terminalPresetRequest(requestID, ownerEdgeID)
		if protocol == "anthropic" {
			writeAnthropicError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
		} else {
			writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
		}
		return err
	}
	if s.artifactFrontiers.pairRequired(requestID, ownerEdgeID) && decision.Mode != modeLight {
		s.terminalPresetRequest(requestID, ownerEdgeID)
		err := fmt.Errorf("artifact frontier requires the exact Plan/Review pair before local-stage handoff")
		if protocol == "anthropic" {
			writeAnthropicError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
		} else {
			writeError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
		}
		return err
	}

	switch decision.Mode {
	case modeDirect:
		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,
		}
		return s.runDirectTurn(r.Context(), turn, output)
	case modeLight:
		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,
		}
		return s.runArtifactPairTurn(turn, output, gate)
	default:
		s.terminalPresetRequest(requestID, ownerEdgeID)
		errMsg := fmt.Sprintf("unsupported mode %q", decision.Mode)
		if protocol == "anthropic" {
			writeAnthropicError(w, http.StatusBadRequest, "invalid_request_error", errMsg)
		} else {
			writeError(w, http.StatusBadRequest, "invalid_request_error", errMsg)
		}
		return fmt.Errorf("%s", errMsg)
	}
}

func (s *Server) submitHotPathStage(ctx context.Context, r *http.Request, snapshot hotPathDispatchSnapshot) (normalizedStageOutput, hotPathStageCorrelation, error) {
	if err := snapshot.Input.validate(); err != nil {
		return normalizedStageOutput{}, hotPathStageCorrelation{}, err
	}
	prompt, err := snapshot.Input.prompt(snapshot.Phase)
	if err != nil {
		return normalizedStageOutput{}, hotPathStageCorrelation{}, err
	}
	route, err := s.revalidateHotPathStageRoute(ctx, snapshot)
	if err != nil {
		return normalizedStageOutput{}, hotPathStageCorrelation{}, 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()
			output, collectErr := collectPresetTunnelResult(ctx, 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()
		output, collectErr := collectPresetNormalizedResult(ctx, 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")
	}
	if err := validateHotPathStageDispatch(snapshot, route, result.DispatchInfo); err != nil {
		if result.Run != nil {
			result.Run.Close()
		}
		if result.Tunnel != nil {
			result.Tunnel.Close()
		}
		return normalizedStageOutput{}, hotPathStageCorrelation{}, err
	}
	var output normalizedStageOutput
	switch result.Path {
	case edgeservice.ProviderPoolPathNormalized:
		output, err = collectPresetNormalizedResult(ctx, result.Run, result.DispatchInfo)
	case edgeservice.ProviderPoolPathTunnel:
		wireProtocol := "openai"
		if result.DispatchInfo.ProfileDriver == string(config.ProtocolDriverAnthropicMessages) {
			wireProtocol = "anthropic"
		}
		output, err = collectPresetTunnelResult(ctx, result.Tunnel, result.DispatchInfo, wireProtocol)
	default:
		err = 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 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,
	}
}

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"
	}
	if len(snapshot.Stage.Options) > 0 {
		input["options"] = cloneAnyMap(snapshot.Stage.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"
	}
	applyHotPathStageOptions(body, snapshot.Stage.Options, map[string]struct{}{"model": {}, "messages": {}, "tools": {}, "stream": {}})
	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"}
	}
	applyHotPathStageOptions(body, snapshot.Stage.Options, map[string]struct{}{"model": {}, "messages": {}, "tools": {}, "stream": {}})
	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)
	}
}