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

package openai

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

	"go.uber.org/zap"

	edgeservice "iop/apps/edge/internal/service"
	"iop/packages/go/streamgate"
	iop "iop/proto/gen/iop"
)

const anthropicChatBridgeRejectionLogMessage = "edge_anthropic_chat_bridge_rejection"

func (s *Server) observeAnthropicChatBridgeRejection(status int) {
	s.logger.Info(
		anthropicChatBridgeRejectionLogMessage,
		zap.String("surface", "messages"),
		zap.String("bridge", "chat"),
		zap.String("rejection_class", "provider_http"),
		zap.Int("http_status", status),
	)
}

type openAIChatStreamChunk struct {
	ID      string `json:"id"`
	Model   string `json:"model"`
	Choices []struct {
		Delta struct {
			Content          string `json:"content"`
			ReasoningContent string `json:"reasoning_content"`
			Reasoning        string `json:"reasoning"`
			ToolCalls        []struct {
				Index        int                        `json:"index"`
				ID           string                     `json:"id"`
				ExtraContent openAIChatToolExtraContent `json:"extra_content,omitempty"`
				Function     struct {
					Name      string `json:"name"`
					Arguments string `json:"arguments"`
				} `json:"function"`
			} `json:"tool_calls"`
		} `json:"delta"`
		FinishReason *string `json:"finish_reason"`
	} `json:"choices"`
	Usage *openAIChatBridgeUsage `json:"usage,omitempty"`
	Error *struct {
		Type    string `json:"type"`
		Message string `json:"message"`
	} `json:"error,omitempty"`
}

type anthropicBridgeToolState struct {
	id           string
	name         string
	extraContent openAIChatToolExtraContent
	arguments    strings.Builder
}

type anthropicBridgeStream struct {
	w          http.ResponseWriter
	model      string
	id         string
	started    bool
	stopped    bool
	nextBlock  int
	openBlock  bool
	openKind   string
	finish     *string
	usage      anthropicUsage
	tools      map[int]*anthropicBridgeToolState
	pendingSSE []byte
}

func newAnthropicBridgeStream(w http.ResponseWriter, model string) *anthropicBridgeStream {
	return &anthropicBridgeStream{w: w, model: model, tools: make(map[int]*anthropicBridgeToolState)}
}

func (s *anthropicBridgeStream) Feed(chunk []byte) error {
	if s.stopped {
		return nil
	}
	s.pendingSSE = append(s.pendingSSE, chunk...)
	s.pendingSSE = bytes.ReplaceAll(s.pendingSSE, []byte("\r\n"), []byte("\n"))
	for {
		index := bytes.Index(s.pendingSSE, []byte("\n\n"))
		if index < 0 {
			return nil
		}
		event := append([]byte(nil), s.pendingSSE[:index]...)
		s.pendingSSE = s.pendingSSE[index+2:]
		if err := s.consumeSSEEvent(event); err != nil {
			return err
		}
		if s.stopped {
			s.pendingSSE = nil
			return nil
		}
	}
}

func (s *anthropicBridgeStream) consumeSSEEvent(event []byte) error {
	var dataLines [][]byte
	for _, line := range bytes.Split(event, []byte("\n")) {
		line = bytes.TrimSpace(line)
		if len(line) == 0 || line[0] == ':' {
			continue
		}
		if bytes.HasPrefix(line, []byte("data:")) {
			dataLines = append(dataLines, bytes.TrimSpace(bytes.TrimPrefix(line, []byte("data:"))))
		}
	}
	if len(dataLines) == 0 {
		return nil
	}
	payload := bytes.Join(dataLines, []byte("\n"))
	if bytes.Equal(payload, []byte("[DONE]")) {
		return s.Finish()
	}
	var chunk openAIChatStreamChunk
	if err := json.Unmarshal(payload, &chunk); err != nil {
		return fmt.Errorf("decode Chat SSE event: %w", err)
	}
	if chunk.Error != nil {
		message := strings.TrimSpace(chunk.Error.Message)
		if message == "" {
			message = "upstream provider error"
		}
		return s.Error(chunk.Error.Type, message)
	}
	if chunk.ID != "" && s.id == "" {
		s.id = chunk.ID
	}
	if chunk.Usage != nil {
		s.usage.InputTokens = chunk.Usage.PromptTokens
		s.usage.OutputTokens = chunk.Usage.CompletionTokens
		s.usage.CacheReadInputTokens = chunk.Usage.PromptDetails.CachedTokens
	}
	if err := s.start(); err != nil {
		return err
	}
	for _, choice := range chunk.Choices {
		reasoning := choice.Delta.ReasoningContent
		if reasoning == "" {
			reasoning = choice.Delta.Reasoning
		}
		if reasoning != "" {
			if err := s.delta("thinking", reasoning); err != nil {
				return err
			}
		}
		if choice.Delta.Content != "" {
			if err := s.delta("text", choice.Delta.Content); err != nil {
				return err
			}
		}
		for _, delta := range choice.Delta.ToolCalls {
			state := s.tools[delta.Index]
			if state == nil {
				state = &anthropicBridgeToolState{}
				s.tools[delta.Index] = state
			}
			if delta.ID != "" {
				state.id = delta.ID
			}
			if delta.Function.Name != "" {
				state.name = delta.Function.Name
			}
			if delta.ExtraContent.Google != nil && delta.ExtraContent.Google.ThoughtSignature != "" {
				state.extraContent = delta.ExtraContent
			}
			state.arguments.WriteString(delta.Function.Arguments)
		}
		if choice.FinishReason != nil {
			s.finish = choice.FinishReason
		}
	}
	return nil
}

func (s *anthropicBridgeStream) start() error {
	if s.started {
		return nil
	}
	s.started = true
	id := s.id
	if id == "" {
		id = "msg_iop"
	}
	return writeAnthropicSSEEvent(s.w, "message_start", map[string]any{
		"type": "message_start",
		"message": anthropicMessageResponse{
			ID: id, Type: "message", Role: "assistant", Model: s.model,
			Content: []map[string]any{}, StopReason: nil, StopSequence: nil,
			Usage: anthropicUsage{InputTokens: s.usage.InputTokens},
		},
	})
}

func (s *anthropicBridgeStream) delta(kind, value string) error {
	if !s.openBlock || s.openKind != kind {
		if err := s.closeBlock(); err != nil {
			return err
		}
		block := map[string]any{"type": kind}
		if kind == "thinking" {
			block["thinking"] = ""
			block["signature"] = ""
		} else {
			block["text"] = ""
		}
		if err := writeAnthropicSSEEvent(s.w, "content_block_start", map[string]any{
			"type": "content_block_start", "index": s.nextBlock, "content_block": block,
		}); err != nil {
			return err
		}
		s.openBlock, s.openKind = true, kind
	}
	deltaType, field := "text_delta", "text"
	if kind == "thinking" {
		deltaType, field = "thinking_delta", "thinking"
	}
	return writeAnthropicSSEEvent(s.w, "content_block_delta", map[string]any{
		"type": "content_block_delta", "index": s.nextBlock,
		"delta": map[string]any{"type": deltaType, field: value},
	})
}

func (s *anthropicBridgeStream) closeBlock() error {
	if !s.openBlock {
		return nil
	}
	if err := writeAnthropicSSEEvent(s.w, "content_block_stop", map[string]any{
		"type": "content_block_stop", "index": s.nextBlock,
	}); err != nil {
		return err
	}
	s.nextBlock++
	s.openBlock = false
	s.openKind = ""
	return nil
}

func (s *anthropicBridgeStream) emitTools() error {
	if len(s.tools) == 0 {
		return nil
	}
	indices := make([]int, 0, len(s.tools))
	for index := range s.tools {
		indices = append(indices, index)
	}
	sort.Ints(indices)
	for _, index := range indices {
		tool := s.tools[index]
		arguments := tool.arguments.String()
		if tool.id == "" || tool.name == "" || !json.Valid([]byte(arguments)) {
			return fmt.Errorf("Chat stream tool call has invalid id, name, or arguments")
		}
		if err := writeAnthropicSSEEvent(s.w, "content_block_start", map[string]any{
			"type": "content_block_start", "index": s.nextBlock,
			"content_block": map[string]any{
				"type": "tool_use", "id": encodeAnthropicBridgeToolID(tool.id, tool.extraContent),
				"name": tool.name, "input": map[string]any{},
			},
		}); err != nil {
			return err
		}
		if err := writeAnthropicSSEEvent(s.w, "content_block_delta", map[string]any{
			"type": "content_block_delta", "index": s.nextBlock,
			"delta": map[string]any{"type": "input_json_delta", "partial_json": arguments},
		}); err != nil {
			return err
		}
		if err := writeAnthropicSSEEvent(s.w, "content_block_stop", map[string]any{
			"type": "content_block_stop", "index": s.nextBlock,
		}); err != nil {
			return err
		}
		s.nextBlock++
	}
	return nil
}

func (s *anthropicBridgeStream) Finish() error {
	if s.stopped {
		return nil
	}
	if err := s.start(); err != nil {
		return err
	}
	if err := s.closeBlock(); err != nil {
		return err
	}
	if err := s.emitTools(); err != nil {
		return err
	}
	stopReason, err := anthropicStopReason(s.finish)
	if err != nil {
		return err
	}
	if stopReason == nil {
		fallback := "end_turn"
		stopReason = &fallback
	}
	if err := writeAnthropicSSEEvent(s.w, "message_delta", map[string]any{
		"type": "message_delta", "delta": map[string]any{"stop_reason": *stopReason, "stop_sequence": nil},
		"usage": s.usage,
	}); err != nil {
		return err
	}
	if err := writeAnthropicSSEEvent(s.w, "message_stop", map[string]any{"type": "message_stop"}); err != nil {
		return err
	}
	s.stopped = true
	return nil
}

func (s *anthropicBridgeStream) Error(errorType, message string) error {
	if s.stopped {
		return nil
	}
	if strings.TrimSpace(errorType) == "" {
		errorType = "api_error"
	}
	s.stopped = true
	return writeAnthropicSSEEvent(s.w, "error", anthropicErrorResponse{
		Type: "error", Error: errorBody{Type: errorType, Message: message},
	})
}

func writeAnthropicSSEEvent(w http.ResponseWriter, event string, payload any) error {
	encoded, err := json.Marshal(payload)
	if err != nil {
		return err
	}
	if _, err := fmt.Fprintf(w, "event: %s\ndata: %s\n\n", event, encoded); err != nil {
		return err
	}
	return nil
}

// anthropicHotPathCodec is the caller-facing Messages codec for one preset
// HTTP turn. It consumes only the normalized outer-turn accumulator and
// release log; selected-provider wire decoding remains in the shared stage
// decoders. The codec owns exactly one caller envelope and terminal.
type anthropicHotPathCodec struct {
	mu sync.Mutex

	w         http.ResponseWriter
	model     string
	stream    bool
	requestID string
	maxTokens int
	outer     *hotPathOuterTurn

	flusher          http.Flusher
	started          bool
	terminal         bool
	releaseAttached  bool
	progressiveTools bool
	nextBlock        int
	openBlock        bool
	openKind         string
	openToolID       string
	emittedTools     map[string]struct{}
}

type hotPathAnthropicCodecContextKey struct{}

type anthropicHotPathBlock struct {
	kind      string
	id        string
	name      string
	signature string
	fragments []string
	toolIndex int
}

func newAnthropicHotPathCodec(
	w http.ResponseWriter,
	model string,
	stream bool,
	requestID string,
	maxTokens int,
) *anthropicHotPathCodec {
	return &anthropicHotPathCodec{
		w: w, model: model, stream: stream, requestID: requestID, maxTokens: maxTokens,
	}
}

func withHotPathAnthropicCodec(r *http.Request, codec *anthropicHotPathCodec) *http.Request {
	if r == nil || codec == nil {
		return r
	}
	return r.WithContext(context.WithValue(r.Context(), hotPathAnthropicCodecContextKey{}, codec))
}

func hotPathAnthropicCodecFromRequest(r *http.Request) *anthropicHotPathCodec {
	if r == nil {
		return nil
	}
	codec, _ := r.Context().Value(hotPathAnthropicCodecContextKey{}).(*anthropicHotPathCodec)
	return codec
}

func (c *anthropicHotPathCodec) callerOuterTurn(responseID string, outputCapTokens int) *hotPathOuterTurn {
	if c == nil {
		return newHotPathCallerCappedOuterTurn(responseID, outputCapTokens)
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.outer == nil {
		capTokens := c.maxTokens
		if capTokens <= 0 {
			capTokens = outputCapTokens
		}
		c.outer = newHotPathCallerCappedOuterTurn(responseID, capTokens)
	}
	return c.outer
}

func (c *anthropicHotPathCodec) currentOuterTurn() *hotPathOuterTurn {
	if c == nil {
		return nil
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.outer
}

// prepareProgressiveWriter connects the normalized outer-turn release seam to
// the caller-facing Messages codec. Initial-selector tool fragments stay held
// until structural classification; later, already-classified Light stages may
// release tool fragments as well as text and reasoning.
func (c *anthropicHotPathCodec) prepareProgressiveWriter(w http.ResponseWriter, outer *hotPathOuterTurn, releaseTools bool) error {
	if c == nil || !c.stream || outer == nil {
		return nil
	}
	flusher, ok := w.(http.Flusher)
	if !ok {
		return fmt.Errorf("response writer does not support flushing")
	}
	c.mu.Lock()
	c.w = w
	c.flusher = flusher
	c.outer = outer
	c.progressiveTools = releaseTools
	if c.emittedTools == nil {
		c.emittedTools = make(map[string]struct{})
	}
	attached := c.releaseAttached
	if !attached {
		c.releaseAttached = true
	}
	c.mu.Unlock()
	if attached {
		return nil
	}
	if err := outer.setReleaseCallback(func(delta hotPathReleasedDelta) error {
		return c.writeProgressiveDelta(outer, delta)
	}); err != nil {
		c.mu.Lock()
		c.releaseAttached = false
		c.mu.Unlock()
		return err
	}
	return nil
}

func (c *anthropicHotPathCodec) writeProgressiveDelta(outer *hotPathOuterTurn, delta hotPathReleasedDelta) error {
	responseID, ok := outer.publicResponseIdentity()
	if !ok {
		return fmt.Errorf("Anthropic Hot Path response is missing provider identity")
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.terminal {
		return errHotPathTurnTerminal
	}
	if delta.Kind == streamgate.EventKindToolCallFragment && !c.progressiveTools {
		return nil
	}
	usage := c.previewUsageLocked(outer)
	if _, err := c.startStreamLocked(responseID, usage); err != nil {
		return err
	}
	switch delta.Kind {
	case streamgate.EventKindReasoningDelta:
		if err := c.ensureProgressiveBlockLocked(outer, "thinking", "", ""); err != nil {
			return err
		}
		return c.writeProgressiveBlockDeltaLocked(map[string]any{"type": "thinking_delta", "thinking": delta.Text})
	case streamgate.EventKindTextDelta:
		if err := c.ensureProgressiveBlockLocked(outer, "text", "", ""); err != nil {
			return err
		}
		return c.writeProgressiveBlockDeltaLocked(map[string]any{"type": "text_delta", "text": delta.Text})
	case streamgate.EventKindToolCallFragment:
		if strings.TrimSpace(delta.PublicID) == "" || strings.TrimSpace(delta.Name) == "" {
			return fmt.Errorf("Anthropic Hot Path tool block is missing id or name")
		}
		if err := c.ensureProgressiveBlockLocked(outer, "tool_use", delta.PublicID, delta.Name); err != nil {
			return err
		}
		c.emittedTools[delta.PublicID] = struct{}{}
		return c.writeProgressiveBlockDeltaLocked(map[string]any{"type": "input_json_delta", "partial_json": delta.Args})
	default:
		return fmt.Errorf("unsupported progressive Anthropic delta kind %q", delta.Kind)
	}
}

func (c *anthropicHotPathCodec) previewUsageLocked(outer *hotPathOuterTurn) json.RawMessage {
	usage, ok := outer.currentPreviewUsage()
	if !ok {
		return nil
	}
	raw, _ := json.Marshal(anthropicUsage{
		InputTokens: usage.InputTokens, OutputTokens: usage.OutputTokens,
		CacheReadInputTokens: usage.CachedInputTokens,
	})
	return raw
}

func (c *anthropicHotPathCodec) ensureProgressiveBlockLocked(outer *hotPathOuterTurn, kind, toolID, toolName string) error {
	if c.openBlock && c.openKind == kind && (kind != "tool_use" || c.openToolID == toolID) {
		return nil
	}
	if err := c.closeProgressiveBlockLocked(outer, ""); err != nil {
		return err
	}
	block := map[string]any{"type": kind}
	switch kind {
	case "thinking":
		block["thinking"], block["signature"] = "", ""
	case "text":
		block["text"] = ""
	case "tool_use":
		block["id"], block["name"], block["input"] = toolID, toolName, map[string]any{}
	default:
		return fmt.Errorf("unsupported Anthropic content block kind %q", kind)
	}
	if err := writeDirectAnthropicEvent(c.w, c.flusher, "content_block_start", map[string]any{
		"type": "content_block_start", "index": c.nextBlock, "content_block": block,
	}); err != nil {
		return err
	}
	c.openBlock = true
	c.openKind = kind
	c.openToolID = toolID
	return nil
}

func (c *anthropicHotPathCodec) writeProgressiveBlockDeltaLocked(delta map[string]any) error {
	return writeDirectAnthropicEvent(c.w, c.flusher, "content_block_delta", map[string]any{
		"type": "content_block_delta", "index": c.nextBlock, "delta": delta,
	})
}

func (c *anthropicHotPathCodec) closeProgressiveBlockLocked(outer *hotPathOuterTurn, finalSignature string) error {
	if !c.openBlock {
		return nil
	}
	if c.openKind == "thinking" {
		signature := finalSignature
		if signature == "" && outer != nil {
			signature = outer.currentReasoningSignature()
		}
		if signature != "" {
			if err := c.writeProgressiveBlockDeltaLocked(map[string]any{"type": "signature_delta", "signature": signature}); err != nil {
				return err
			}
		}
	}
	if err := writeDirectAnthropicEvent(c.w, c.flusher, "content_block_stop", map[string]any{
		"type": "content_block_stop", "index": c.nextBlock,
	}); err != nil {
		return err
	}
	c.nextBlock++
	c.openBlock = false
	c.openKind = ""
	c.openToolID = ""
	return nil
}

func (c *anthropicHotPathCodec) runInitialPresetTurn(
	s *Server,
	w http.ResponseWriter,
	r *http.Request,
	dispatch routeDispatch,
	runMeta map[string]string,
	result *edgeservice.ProviderPoolDispatchResult,
) (normalizedStageOutput, bool, error) {
	var (
		stage normalizedStageOutput
		gate  hotPathSelectorGate
		err   error
	)
	if c.stream {
		outer := c.callerOuterTurn("", hotPathOutputTokenCap(runMeta))
		if err := c.prepareProgressiveWriter(w, outer, false); err != nil {
			return stage, false, err
		}
		stage, gate, err = s.runLivePresetSelectorResult(
			r.Context(), dispatch, "anthropic", runMeta["iop_stage_id"], result, outer,
		)
	} else {
		stage, gate, err = s.collectPresetSelectorResult(r.Context(), dispatch, "anthropic", result)
	}
	if err != nil {
		if contextErr := r.Context().Err(); contextErr != nil {
			// Exact active-run cancellation is complete; caller cancellation is
			// intentionally wire-silent.
			return stage, true, contextErr
		}
		return stage, false, err
	}
	err = s.dispatchPresetTurn(w, r, dispatch, "anthropic", c.stream, runMeta, stage, gate)
	return stage, true, err
}

func writeHotPathAnthropicOuterResponse(turn *hotPathTurn, output normalizedStageOutput) (bool, error) {
	if turn == nil {
		return false, nil
	}
	codec := hotPathAnthropicCodecFromRequest(turn.Request)
	if codec == nil {
		return false, nil
	}
	codec.w = turn.Writer
	if codec.model == "" {
		codec.model = directPublicModel(turn)
	}
	return true, codec.write(output)
}

func writeHotPathAnthropicOuterError(turn *hotPathTurn, status int, errorType, message string) bool {
	if turn == nil {
		return false
	}
	codec := hotPathAnthropicCodecFromRequest(turn.Request)
	if codec == nil {
		return false
	}
	codec.w = turn.Writer
	disposition := hotPathTerminalDisposition{
		Kind: hotPathDispositionProviderError, Cause: message, Source: "anthropic_outer_error",
	}
	selected := false
	if turn.OuterTurn != nil {
		if terminalDisposition, ok := turn.OuterTurn.terminalDisposition(); ok {
			disposition = terminalDisposition
			selected = true
		}
	}
	if !selected && strings.Contains(strings.ToLower(errorType), "invalid") {
		disposition.Kind = hotPathDispositionValidationError
	}
	_ = codec.writeDisposition(disposition, status, errorType, message)
	return true
}

func (c *anthropicHotPathCodec) bindResponseID(responseID string) error {
	responseID = strings.TrimSpace(responseID)
	if responseID == "" {
		return fmt.Errorf("Anthropic Hot Path response is missing provider identity")
	}
	if c == nil {
		return nil
	}
	c.mu.Lock()
	outer := c.outer
	c.mu.Unlock()
	if outer != nil {
		return outer.bindPublicResponseID(responseID)
	}
	return nil
}

func (c *anthropicHotPathCodec) write(output normalizedStageOutput) error {
	if c == nil || c.w == nil {
		return fmt.Errorf("Anthropic Hot Path codec is unavailable")
	}
	if err := c.bindResponseID(output.ResponseID); err != nil {
		return err
	}
	responseID := strings.TrimSpace(output.ResponseID)
	outer := c.currentOuterTurn()
	if outer != nil {
		var ok bool
		responseID, ok = outer.publicResponseIdentity()
		if !ok {
			return fmt.Errorf("Anthropic Hot Path response is missing provider identity")
		}
	}
	blocks, err := c.blocks(output)
	if err != nil {
		return err
	}
	stopReason := anthropicDirectStopReason(output.TerminalReason)
	if outer != nil {
		if disposition, ok := outer.terminalDisposition(); ok {
			policy := anthropicHotPathPolicy(disposition)
			switch {
			case policy.silent && outer.isTerminalCommitted():
				return c.writeDisposition(disposition, 0, "", "")
			case policy.errorTerminal && outer.isTerminalCommitted():
				return c.writeDisposition(disposition, policy.status, policy.errorType, disposition.Cause)
			case policy.stopReason != "":
				stopReason = policy.stopReason
			}
		}
	}
	if stopReason == "" {
		if len(output.ToolCalls) > 0 {
			stopReason = "tool_use"
		} else {
			stopReason = "end_turn"
		}
	}
	usage := anthropicHotPathUsage(output)
	if c.stream {
		return c.writeStream(responseID, blocks, stopReason, usage)
	}
	return c.writeJSON(responseID, blocks, stopReason, usage)
}

func (c *anthropicHotPathCodec) blocks(output normalizedStageOutput) ([]anthropicHotPathBlock, error) {
	var released []hotPathReleasedDelta
	if c.outer != nil && !output.CallerStageOnly {
		released = c.outer.releasedDeltas()
	}
	if len(released) == 0 {
		if output.Reasoning != "" {
			released = append(released, hotPathReleasedDelta{Kind: streamgate.EventKindReasoningDelta, Text: output.Reasoning})
		}
		if output.Content != "" {
			released = append(released, hotPathReleasedDelta{Kind: streamgate.EventKindTextDelta, Text: output.Content})
		}
		for _, call := range output.ToolCalls {
			released = append(released, hotPathReleasedDelta{
				Kind: streamgate.EventKindToolCallFragment, PublicID: call.ID,
				Name: call.Name, Args: directToolArguments(call),
			})
		}
	}

	blocks := make([]anthropicHotPathBlock, 0, len(released))
	toolBlocks := make(map[string]int)
	toolOrdinal := 0
	for _, delta := range released {
		switch delta.Kind {
		case streamgate.EventKindReasoningDelta, streamgate.EventKindTextDelta:
			kind := "text"
			if delta.Kind == streamgate.EventKindReasoningDelta {
				kind = "thinking"
			}
			if len(blocks) == 0 || blocks[len(blocks)-1].kind != kind {
				blocks = append(blocks, anthropicHotPathBlock{kind: kind, toolIndex: -1})
			}
			blocks[len(blocks)-1].fragments = append(blocks[len(blocks)-1].fragments, delta.Text)
		case streamgate.EventKindToolCallFragment:
			key := delta.PublicID
			if key == "" {
				key = fmt.Sprintf("tool-%d", toolOrdinal)
			}
			blockIndex, ok := toolBlocks[key]
			if !ok {
				block := anthropicHotPathBlock{kind: "tool_use", id: delta.PublicID, name: delta.Name, toolIndex: toolOrdinal}
				if toolOrdinal < len(output.ToolCalls) {
					call := output.ToolCalls[toolOrdinal]
					block.id = call.ID
					block.name = call.Name
				}
				blocks = append(blocks, block)
				blockIndex = len(blocks) - 1
				toolBlocks[key] = blockIndex
				toolOrdinal++
			}
			blocks[blockIndex].fragments = append(blocks[blockIndex].fragments, delta.Args)
		}
	}
	for toolOrdinal < len(output.ToolCalls) {
		call := output.ToolCalls[toolOrdinal]
		blocks = append(blocks, anthropicHotPathBlock{
			kind: "tool_use", id: call.ID, name: call.Name,
			fragments: []string{directToolArguments(call)}, toolIndex: toolOrdinal,
		})
		toolOrdinal++
	}
	for index := range blocks {
		block := &blocks[index]
		if block.kind == "tool_use" {
			if strings.TrimSpace(block.id) == "" || strings.TrimSpace(block.name) == "" {
				return nil, fmt.Errorf("Anthropic Hot Path tool block is missing id or name")
			}
			arguments := strings.Join(block.fragments, "")
			if block.toolIndex >= 0 && block.toolIndex < len(output.ToolCalls) {
				expected := directToolArguments(output.ToolCalls[block.toolIndex])
				if arguments == "" {
					arguments = expected
					block.fragments = []string{expected}
				} else if expected != "" && arguments != expected {
					return nil, fmt.Errorf("Anthropic Hot Path tool fragments do not match the issued call")
				}
			}
			if arguments == "" {
				arguments = "{}"
				block.fragments = []string{arguments}
			}
			if !json.Valid([]byte(arguments)) {
				return nil, fmt.Errorf("Anthropic Hot Path tool input is not valid JSON")
			}
		}
	}
	for index := len(blocks) - 1; index >= 0; index-- {
		if blocks[index].kind == "thinking" {
			blocks[index].signature = output.ReasoningSignature
			break
		}
	}
	return blocks, nil
}

func anthropicHotPathUsage(output normalizedStageOutput) json.RawMessage {
	if len(output.Usage) > 0 {
		var fields map[string]json.RawMessage
		if json.Unmarshal(output.Usage, &fields) == nil {
			if _, ok := fields["input_tokens"]; ok {
				return cloneRawJSON(output.Usage)
			}
		}
	}
	if output.OpenAIUsage != nil {
		raw, _ := json.Marshal(output.OpenAIUsage)
		return openAIUsageToAnthropic(raw)
	}
	return openAIUsageToAnthropic(output.Usage)
}

func (c *anthropicHotPathCodec) writeJSON(responseID string, blocks []anthropicHotPathBlock, stopReason string, usage json.RawMessage) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.terminal {
		return errHotPathTurnTerminal
	}
	content := make([]map[string]any, 0, len(blocks))
	for _, block := range blocks {
		switch block.kind {
		case "thinking":
			content = append(content, map[string]any{
				"type": "thinking", "thinking": strings.Join(block.fragments, ""), "signature": block.signature,
			})
		case "text":
			content = append(content, map[string]any{"type": "text", "text": strings.Join(block.fragments, "")})
		case "tool_use":
			var input any
			if err := json.Unmarshal([]byte(strings.Join(block.fragments, "")), &input); err != nil {
				return err
			}
			content = append(content, map[string]any{
				"type": "tool_use", "id": block.id, "name": block.name, "input": input,
			})
		}
	}
	response := map[string]any{
		"id": responseID, "type": "message", "role": "assistant", "model": c.model,
		"content": content, "stop_reason": stopReason, "stop_sequence": nil,
	}
	if len(usage) > 0 {
		response["usage"] = usage
	}
	c.terminal = true
	return writeDirectJSON(c.w, http.StatusOK, response)
}

func (c *anthropicHotPathCodec) startStreamLocked(responseID string, usage json.RawMessage) (http.Flusher, error) {
	flusher := c.flusher
	if flusher == nil {
		var ok bool
		flusher, ok = c.w.(http.Flusher)
		if !ok {
			return nil, fmt.Errorf("response writer does not support flushing")
		}
		c.flusher = flusher
	}
	if c.started {
		return flusher, nil
	}
	c.w.Header().Set("Content-Type", "text/event-stream")
	c.w.Header().Set("Cache-Control", "no-cache")
	c.w.WriteHeader(http.StatusOK)
	message := map[string]any{
		"id": responseID, "type": "message", "role": "assistant", "model": c.model,
		"content": []any{}, "stop_reason": nil, "stop_sequence": nil,
	}
	if startUsage := anthropicStartUsage(usage); len(startUsage) > 0 {
		message["usage"] = startUsage
	}
	if err := writeDirectAnthropicEvent(c.w, flusher, "message_start", map[string]any{
		"type": "message_start", "message": message,
	}); err != nil {
		return nil, err
	}
	c.started = true
	return flusher, nil
}

func (c *anthropicHotPathCodec) writeStream(responseID string, blocks []anthropicHotPathBlock, stopReason string, usage json.RawMessage) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.terminal {
		return errHotPathTurnTerminal
	}
	flusher, err := c.startStreamLocked(responseID, usage)
	if err != nil {
		return err
	}
	if c.started && c.openBlock {
		if err := c.closeProgressiveBlockLocked(c.outer, outputReasoningSignature(blocks)); err != nil {
			return err
		}
	}
	for _, block := range blocks {
		if c.releaseAttached {
			if block.kind != "tool_use" {
				continue
			}
			if _, emitted := c.emittedTools[block.id]; emitted {
				continue
			}
		}
		if err := c.writeCompleteBlockLocked(block); err != nil {
			return err
		}
	}
	delta := map[string]any{
		"type": "message_delta", "delta": map[string]any{"stop_reason": stopReason, "stop_sequence": nil},
	}
	if len(usage) > 0 {
		delta["usage"] = usage
	}
	if err := writeDirectAnthropicEvent(c.w, flusher, "message_delta", delta); err != nil {
		return err
	}
	if err := writeDirectAnthropicEvent(c.w, flusher, "message_stop", map[string]any{"type": "message_stop"}); err != nil {
		return err
	}
	c.terminal = true
	return nil
}

func outputReasoningSignature(blocks []anthropicHotPathBlock) string {
	for index := len(blocks) - 1; index >= 0; index-- {
		if blocks[index].kind == "thinking" {
			return blocks[index].signature
		}
	}
	return ""
}

func (c *anthropicHotPathCodec) writeCompleteBlockLocked(block anthropicHotPathBlock) error {
	index := c.nextBlock
	start := map[string]any{"type": block.kind}
	switch block.kind {
	case "thinking":
		start["thinking"], start["signature"] = "", ""
	case "text":
		start["text"] = ""
	case "tool_use":
		start["id"], start["name"], start["input"] = block.id, block.name, map[string]any{}
	}
	if err := writeDirectAnthropicEvent(c.w, c.flusher, "content_block_start", map[string]any{
		"type": "content_block_start", "index": index, "content_block": start,
	}); err != nil {
		return err
	}
	for _, fragment := range block.fragments {
		delta := map[string]any{"type": "text_delta", "text": fragment}
		switch block.kind {
		case "thinking":
			delta = map[string]any{"type": "thinking_delta", "thinking": fragment}
		case "tool_use":
			delta = map[string]any{"type": "input_json_delta", "partial_json": fragment}
		}
		if err := writeDirectAnthropicEvent(c.w, c.flusher, "content_block_delta", map[string]any{
			"type": "content_block_delta", "index": index, "delta": delta,
		}); err != nil {
			return err
		}
	}
	if block.kind == "thinking" && block.signature != "" {
		if err := writeDirectAnthropicEvent(c.w, c.flusher, "content_block_delta", map[string]any{
			"type": "content_block_delta", "index": index,
			"delta": map[string]any{"type": "signature_delta", "signature": block.signature},
		}); err != nil {
			return err
		}
	}
	if err := writeDirectAnthropicEvent(c.w, c.flusher, "content_block_stop", map[string]any{
		"type": "content_block_stop", "index": index,
	}); err != nil {
		return err
	}
	c.nextBlock++
	return nil
}

func (c *anthropicHotPathCodec) writeError(status int, errorType, message string) error {
	disposition := hotPathTerminalDisposition{
		Kind: hotPathDispositionProviderError, Cause: message, Source: "anthropic_codec_error",
	}
	if strings.Contains(strings.ToLower(errorType), "invalid") {
		disposition.Kind = hotPathDispositionValidationError
	}
	if outer := c.currentOuterTurn(); outer != nil {
		if selected, ok := outer.terminalDisposition(); ok {
			disposition = selected
		}
	}
	return c.writeDisposition(disposition, status, errorType, message)
}

func (c *anthropicHotPathCodec) writeDisposition(
	disposition hotPathTerminalDisposition,
	status int,
	errorType, message string,
) error {
	if c == nil || c.w == nil {
		return fmt.Errorf("Anthropic Hot Path codec is unavailable")
	}
	policy := anthropicHotPathPolicy(disposition)
	if policy.status != 0 {
		status = policy.status
	}
	if policy.errorType != "" {
		errorType = policy.errorType
	}
	if strings.TrimSpace(message) == "" {
		message = hotPathFirstNonEmpty(disposition.Cause, "hot path stage failed")
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.terminal {
		return errHotPathTurnTerminal
	}
	if policy.silent {
		c.terminal = true
		return nil
	}
	if !policy.errorTerminal {
		return fmt.Errorf("Anthropic disposition %q is not an error terminal", disposition.Kind)
	}
	if c.stream && c.started {
		flusher := c.flusher
		if flusher == nil {
			var ok bool
			flusher, ok = c.w.(http.Flusher)
			if !ok {
				return fmt.Errorf("response writer does not support flushing")
			}
		}
		c.terminal = true
		return writeDirectAnthropicEvent(c.w, flusher, "error", anthropicErrorResponse{
			Type: "error", Error: errorBody{Type: errorType, Message: message},
		})
	}
	c.terminal = true
	writeAnthropicError(c.w, status, errorType, message)
	return nil
}

func (s *Server) writeAnthropicChatBridgeResponse(w http.ResponseWriter, r *http.Request, handle edgeservice.ProviderTunnelResult, envelope anthropicRequestEnvelope) {
	frames := handle.Stream().Frames
	if frames == nil {
		writeAnthropicError(w, http.StatusBadGateway, "api_error", "provider tunnel is unavailable")
		return
	}
	timer := time.NewTimer(handle.WaitTimeout())
	defer timer.Stop()
	status := http.StatusOK
	headers := make(map[string]string)
	started := false
	committed := false
	var body []byte
	var stream *anthropicBridgeStream
	if envelope.Stream {
		stream = newAnthropicBridgeStream(w, envelope.Model)
	}
	flush := func() {
		if flusher, ok := w.(http.Flusher); ok {
			flusher.Flush()
		}
	}
	commitStream := func() {
		if committed {
			return
		}
		copyAnthropicResponseHeaders(w.Header(), headers)
		w.Header().Del("Content-Length")
		w.Header().Set("Content-Type", "text/event-stream")
		w.Header().Set("Cache-Control", "no-cache")
		w.WriteHeader(status)
		committed = true
		flush()
	}

	for {
		select {
		case <-r.Context().Done():
			s.cancelRunOnHTTPGiveUp(handle.Dispatch(), r.Context().Err())
			return
		case <-timer.C:
			s.cancelRunOnHTTPGiveUp(handle.Dispatch(), errRunTimedOut)
			if !committed {
				writeAnthropicError(w, http.StatusBadGateway, "api_error", "provider response timed out")
			} else {
				_ = stream.Error("api_error", "provider response timed out")
				flush()
			}
			return
		case frame, ok := <-frames:
			if !ok {
				if !committed {
					writeAnthropicError(w, http.StatusBadGateway, "api_error", "provider tunnel closed before a response")
				} else {
					_ = stream.Error("api_error", "provider tunnel closed before a response")
					flush()
				}
				return
			}
			switch frame.GetKind() {
			case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START:
				if started {
					continue
				}
				started = true
				status = int(frame.GetStatusCode())
				if status == 0 {
					status = http.StatusOK
				}
				if status >= http.StatusBadRequest {
					s.observeAnthropicChatBridgeRejection(status)
				}
				for key, value := range frame.GetHeaders() {
					headers[key] = value
				}
				if envelope.Stream && status < http.StatusBadRequest {
					commitStream()
				}
			case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY:
				if envelope.Stream && status < http.StatusBadRequest {
					commitStream()
					if err := stream.Feed(frame.GetBody()); err != nil {
						_ = stream.Error("api_error", "upstream stream could not be translated")
						flush()
						return
					}
					flush()
				} else {
					body = append(body, frame.GetBody()...)
				}
			case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR:
				if committed {
					_ = stream.Error("api_error", "provider tunnel failed")
					flush()
				} else {
					writeAnthropicError(w, http.StatusBadGateway, "api_error", "provider tunnel failed")
				}
				return
			case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_END:
				if envelope.Stream && status < http.StatusBadRequest {
					commitStream()
					if err := stream.Finish(); err != nil {
						_ = stream.Error("api_error", "upstream stream could not be translated")
					}
					flush()
					return
				}
				copyAnthropicResponseHeaders(w.Header(), headers)
				w.Header().Del("Content-Length")
				if status >= http.StatusBadRequest {
					writeJSON(w, status, convertChatErrorToAnthropic(body))
					return
				}
				converted, err := convertChatResponseToAnthropic(body, envelope.Model)
				if err != nil {
					writeAnthropicError(w, http.StatusBadGateway, "api_error", "upstream response could not be translated")
					return
				}
				writeJSON(w, http.StatusOK, converted)
				return
			case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_USAGE:
				continue
			}
		}
	}
}

// writeAnthropicResponsesBridgeResponse translates an OpenAI Responses
// provider result back to the Messages surface. Streaming input is currently
// buffered until the Responses terminal so the caller still receives one
// valid Anthropic SSE lifecycle without exposing provider-specific events.
func (s *Server) writeAnthropicResponsesBridgeResponse(w http.ResponseWriter, r *http.Request, handle edgeservice.ProviderTunnelResult, envelope anthropicRequestEnvelope) {
	frames := handle.Stream().Frames
	if frames == nil {
		writeAnthropicError(w, http.StatusBadGateway, "api_error", "provider tunnel is unavailable")
		return
	}
	timer := time.NewTimer(handle.WaitTimeout())
	defer timer.Stop()
	status := http.StatusOK
	headers := make(map[string]string)
	var body []byte
	for {
		select {
		case <-r.Context().Done():
			s.cancelRunOnHTTPGiveUp(handle.Dispatch(), r.Context().Err())
			return
		case <-timer.C:
			s.cancelRunOnHTTPGiveUp(handle.Dispatch(), errRunTimedOut)
			writeAnthropicError(w, http.StatusBadGateway, "api_error", "provider response timed out")
			return
		case frame, ok := <-frames:
			if !ok {
				writeAnthropicError(w, http.StatusBadGateway, "api_error", "provider tunnel closed before a response")
				return
			}
			switch frame.GetKind() {
			case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START:
				status = int(frame.GetStatusCode())
				if status == 0 {
					status = http.StatusOK
				}
				for key, value := range frame.GetHeaders() {
					headers[key] = value
				}
			case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY:
				body = append(body, frame.GetBody()...)
			case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR:
				writeAnthropicError(w, http.StatusBadGateway, "api_error", "provider tunnel failed")
				return
			case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_END:
				copyAnthropicResponseHeaders(w.Header(), headers)
				w.Header().Del("Content-Length")
				if status >= http.StatusBadRequest {
					writeAnthropicError(w, status, "api_error", "upstream provider rejected the request")
					return
				}
				responseBody := body
				if envelope.Stream {
					var terminal struct {
						Type     string          `json:"type"`
						Response json.RawMessage `json:"response"`
					}
					for _, event := range splitOpenAIResponsesSSE(body) {
						if json.Unmarshal(event, &terminal) == nil && terminal.Type == "response.completed" && len(terminal.Response) > 0 {
							responseBody = terminal.Response
						}
					}
				}
				converted, err := convertResponsesResponseToAnthropic(responseBody, envelope.Model)
				if err != nil {
					writeAnthropicError(w, http.StatusBadGateway, "api_error", "upstream response could not be translated")
					return
				}
				if !envelope.Stream {
					writeJSON(w, http.StatusOK, converted)
					return
				}
				writeBufferedAnthropicMessageStream(w, converted)
				return
			}
		}
	}
}

func splitOpenAIResponsesSSE(body []byte) [][]byte {
	var payloads [][]byte
	normalized := bytes.ReplaceAll(body, []byte("\r\n"), []byte("\n"))
	for _, event := range bytes.Split(normalized, []byte("\n\n")) {
		var data [][]byte
		for _, line := range bytes.Split(event, []byte("\n")) {
			line = bytes.TrimSpace(line)
			if bytes.HasPrefix(line, []byte("data:")) {
				part := bytes.TrimSpace(bytes.TrimPrefix(line, []byte("data:")))
				if !bytes.Equal(part, []byte("[DONE]")) {
					data = append(data, part)
				}
			}
		}
		if len(data) > 0 {
			payloads = append(payloads, bytes.Join(data, []byte("\n")))
		}
	}
	return payloads
}

func writeBufferedAnthropicMessageStream(w http.ResponseWriter, message anthropicMessageResponse) {
	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")
	w.WriteHeader(http.StatusOK)
	start := message
	start.Content = []map[string]any{}
	start.StopReason = nil
	start.Usage.OutputTokens = 0
	_ = writeAnthropicSSEEvent(w, "message_start", map[string]any{"type": "message_start", "message": start})
	for index, block := range message.Content {
		blockType, _ := block["type"].(string)
		opening := map[string]any{"type": blockType}
		var delta map[string]any
		switch blockType {
		case "text":
			opening["text"] = ""
			delta = map[string]any{"type": "text_delta", "text": block["text"]}
		case "thinking":
			opening["thinking"], opening["signature"] = "", ""
			delta = map[string]any{"type": "thinking_delta", "thinking": block["thinking"]}
		case "tool_use":
			opening["id"], opening["name"], opening["input"] = block["id"], block["name"], map[string]any{}
			encodedInput, _ := json.Marshal(block["input"])
			delta = map[string]any{"type": "input_json_delta", "partial_json": string(encodedInput)}
		default:
			continue
		}
		_ = writeAnthropicSSEEvent(w, "content_block_start", map[string]any{"type": "content_block_start", "index": index, "content_block": opening})
		_ = writeAnthropicSSEEvent(w, "content_block_delta", map[string]any{"type": "content_block_delta", "index": index, "delta": delta})
		_ = writeAnthropicSSEEvent(w, "content_block_stop", map[string]any{"type": "content_block_stop", "index": index})
	}
	_ = writeAnthropicSSEEvent(w, "message_delta", map[string]any{
		"type": "message_delta", "delta": map[string]any{"stop_reason": message.StopReason, "stop_sequence": nil},
		"usage": message.Usage,
	})
	_ = writeAnthropicSSEEvent(w, "message_stop", map[string]any{"type": "message_stop"})
}