package openai

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

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

const hotPathOpenAIResponseIDMetadata = "openai_response_id"

// hotPathStageUsage is the normalized, protocol-neutral token usage a single
// provider stage reported at its terminal. A zero field means the provider did
// not report that token type; ResponseID lets the outer turn deduplicate a
// stage that reports usage more than once.
type hotPathStageUsage struct {
	ResponseID        string
	InputTokens       int
	OutputTokens      int
	ReasoningTokens   int
	CachedInputTokens int
	Reported          bool
}

// hotPathStageUsageProbe exposes the final usage a stage source observed. The
// stage release sink reads it exactly once when the stage terminal is committed,
// so the outer turn aggregates usage without full-buffering deltas.
type hotPathStageUsageProbe interface {
	stageUsage() (hotPathStageUsage, bool)
}

type hotPathStageIdentityProbe interface {
	stageIdentity() (string, bool)
}

type hotPathStageTerminalReasonProbe interface {
	stageTerminalReason() string
}

// hotPathStageTerminalCauseProbe exposes transport/runtime cause without
// choosing endpoint status or bytes. Stage and outer lifecycle code translate
// it into the closed disposition vocabulary.
type hotPathStageTerminalCauseProbe interface {
	stageTerminalCause() hotPathTerminalDisposition
}

type hotPathStageSignatureProbe interface {
	stageReasoningSignature() string
}

type hotPathProviderIdentity struct {
	mu    sync.Mutex
	value string
}

func (i *hotPathProviderIdentity) bind(value string) error {
	value = strings.TrimSpace(value)
	if value == "" {
		return nil
	}
	i.mu.Lock()
	defer i.mu.Unlock()
	if i.value == "" {
		i.value = value
		return nil
	}
	if i.value != value {
		return fmt.Errorf("hot path provider response identity changed during stage")
	}
	return nil
}

func (i *hotPathProviderIdentity) bindRequired(value string) (string, error) {
	value = strings.TrimSpace(value)
	if value == "" {
		return "", fmt.Errorf("hot path provider response identity is required on every visible and complete event")
	}
	if err := i.bind(value); err != nil {
		return "", err
	}
	return value, nil
}

func (i *hotPathProviderIdentity) get() (string, bool) {
	i.mu.Lock()
	defer i.mu.Unlock()
	return i.value, i.value != ""
}

func (i *hotPathProviderIdentity) require() (string, error) {
	if value, ok := i.get(); ok {
		return value, nil
	}
	return "", fmt.Errorf("hot path provider response identity is required before visible output")
}

// hotPathProviderStageDecoder incrementally turns provider response bytes into
// normalized Core events and accumulates the stage's reported usage. Concrete
// decoders exist per provider wire protocol (OpenAI Chat SSE, Anthropic Messages
// SSE); both reuse the shared SSE frame primitives and carry no caller endpoint
// policy. Decoders never emit response-start or terminal events: the stage
// source owns those transport boundaries.
type hotPathProviderStageDecoder interface {
	decodeBody(body []byte) ([]streamgate.NormalizedEvent, error)
	finish() ([]streamgate.NormalizedEvent, error)
	usageValue() (hotPathStageUsage, bool)
	responseIdentity() (string, bool)
	terminalReason() string
}

type stageToolIdentity struct {
	id   string
	name string
}

// --- Normalized RunEvent stage source ---------------------------------------

// hotPathNormalizedStageSource adapts an edgeservice.RunStream to a stage event
// source, reusing the existing normalized RunEvent adapter and recording the
// stage's terminal usage for the outer turn.
type hotPathNormalizedStageSource struct {
	inner     *openAIRunEventSource
	usageHold *openAIStreamGateUsageHolder
	identity  hotPathProviderIdentity

	mu             sync.Mutex
	pending        []streamgate.NormalizedEvent
	terminalReason string
	terminalCause  hotPathTerminalDisposition
}

func newHotPathNormalizedStageSource(stream edgeservice.RunStream, waitTimeout time.Duration) *hotPathNormalizedStageSource {
	hold := &openAIStreamGateUsageHolder{}
	attempt := &openAIAttemptUsage{}
	source := &hotPathNormalizedStageSource{usageHold: hold}
	source.inner = newOpenAIRunEventSource(stream, waitTimeout, hold, attempt).observeRunEvents(source.observeRunEvent)
	return source
}

func (s *hotPathNormalizedStageSource) observeRunEvent(event *iop.RunEvent) error {
	if event == nil {
		return nil
	}
	identity := event.GetMetadata()[hotPathOpenAIResponseIDMetadata]
	switch event.GetType() {
	case "delta", "reasoning_delta", "complete":
		if _, err := s.identity.bindRequired(identity); err != nil {
			return err
		}
	default:
		if err := s.identity.bind(identity); err != nil {
			return err
		}
	}
	if event.GetType() == "error" || event.GetType() == "cancelled" {
		s.mu.Lock()
		s.terminalCause = hotPathTerminalDisposition{
			Kind:   hotPathDispositionProviderError,
			Cause:  hotPathFirstNonEmpty(event.GetError(), event.GetMessage(), event.GetType()),
			Source: "normalized_run_event",
		}
		s.mu.Unlock()
	}
	if event.GetType() != "complete" {
		return nil
	}
	s.mu.Lock()
	s.terminalReason = strings.TrimSpace(event.GetMetadata()["finish_reason"])
	s.mu.Unlock()
	calls, err := normalizeRunEventToolCalls(event.GetMetadata())
	if err != nil {
		return err
	}
	tools := make([]streamgate.NormalizedEvent, 0, len(calls))
	for _, call := range calls {
		providerID := hotPathFirstNonEmpty(call.ProviderCallID, call.ID)
		tool, err := streamgate.NewToolCallFragmentEvent(
			streamGateChannelDefault, providerID, call.Name, directToolArguments(call), time.Now(),
		)
		if err != nil {
			return err
		}
		tools = append(tools, tool)
	}
	if len(tools) > 0 {
		s.mu.Lock()
		s.pending = append(s.pending, tools...)
		s.mu.Unlock()
	}
	return nil
}

func (s *hotPathNormalizedStageSource) NextEvent(ctx context.Context) (streamgate.NormalizedEvent, error) {
	s.mu.Lock()
	if len(s.pending) > 0 {
		event := s.pending[0]
		s.pending = s.pending[1:]
		s.mu.Unlock()
		return event, nil
	}
	s.mu.Unlock()
	event, err := s.inner.NextEvent(ctx)
	if err != nil || event.Kind() != streamgate.EventKindTerminal {
		return event, err
	}
	s.mu.Lock()
	if len(s.pending) == 0 {
		s.mu.Unlock()
		return event, nil
	}
	s.pending = append(s.pending, event)
	first := s.pending[0]
	s.pending = s.pending[1:]
	s.mu.Unlock()
	return first, nil
}

func (s *hotPathNormalizedStageSource) stageUsage() (hotPathStageUsage, bool) {
	obs := s.usageHold.get()
	responseID, _ := s.identity.get()
	usage := hotPathStageUsage{
		ResponseID:        responseID,
		InputTokens:       obs.inputTokens,
		OutputTokens:      obs.outputTokens,
		ReasoningTokens:   obs.reasoningTokens,
		CachedInputTokens: obs.cachedInputTokens,
		Reported:          obs.providerReported,
	}
	return usage, obs.providerReported
}

func (s *hotPathNormalizedStageSource) stageIdentity() (string, bool) {
	return s.identity.get()
}

func (s *hotPathNormalizedStageSource) stageTerminalReason() string {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.terminalReason
}

func (s *hotPathNormalizedStageSource) stageTerminalCause() hotPathTerminalDisposition {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.terminalCause
}

var (
	_ streamgate.NormalizedEventSource = (*hotPathNormalizedStageSource)(nil)
	_ hotPathStageUsageProbe           = (*hotPathNormalizedStageSource)(nil)
	_ hotPathStageIdentityProbe        = (*hotPathNormalizedStageSource)(nil)
	_ hotPathStageTerminalReasonProbe  = (*hotPathNormalizedStageSource)(nil)
	_ hotPathStageTerminalCauseProbe   = (*hotPathNormalizedStageSource)(nil)
)

// --- Provider tunnel stage source -------------------------------------------

// hotPathTunnelStageSource adapts a provider tunnel frame stream to a stage
// event source. It requires the transport contract's explicit RESPONSE_START
// followed by exactly one END or ERROR; malformed ordering or a channel close
// before completion becomes one sanitized provider-error terminal. Usage frames
// override decoder usage. It never encodes caller-facing wire.
type hotPathTunnelStageSource struct {
	frames      <-chan *iop.ProviderTunnelFrame
	waitTimeout time.Duration
	decoder     hotPathProviderStageDecoder

	mu            sync.Mutex
	started       bool
	terminated    bool
	errorStatus   bool
	pending       []streamgate.NormalizedEvent
	usageProto    *hotPathStageUsage
	terminalCause hotPathTerminalDisposition
}

func newHotPathTunnelStageSource(stream edgeservice.ProviderTunnelStream, waitTimeout time.Duration, decoder hotPathProviderStageDecoder) *hotPathTunnelStageSource {
	return &hotPathTunnelStageSource{frames: stream.Frames, waitTimeout: waitTimeout, decoder: decoder}
}

// newHotPathStageDecoderForProtocol selects the provider stage decoder for a
// wire protocol: Anthropic Messages SSE or, by default, OpenAI Chat SSE.
func newHotPathStageDecoderForProtocol(protocol string) hotPathProviderStageDecoder {
	if protocol == "anthropic" {
		return newAnthropicMessagesStageDecoder()
	}
	return newOpenAIChatStageDecoder()
}

func (s *hotPathTunnelStageSource) NextEvent(ctx context.Context) (streamgate.NormalizedEvent, error) {
	s.mu.Lock()
	if len(s.pending) > 0 {
		ev := s.pending[0]
		s.pending = s.pending[1:]
		s.mu.Unlock()
		return ev, nil
	}
	terminated := s.terminated
	s.mu.Unlock()

	if terminated || s.frames == nil {
		return newOpenAIProviderErrorEvent(streamGateErrorTunnelClosed)
	}

	timer := time.NewTimer(s.waitTimeout)
	defer timer.Stop()
	for {
		select {
		case <-ctx.Done():
			return streamgate.NormalizedEvent{}, ctx.Err()
		case <-timer.C:
			return streamgate.NormalizedEvent{}, errRunTimedOut
		case frame, ok := <-s.frames:
			var events []streamgate.NormalizedEvent
			var err error
			if !ok {
				// The wire contract requires an explicit terminal frame. A close
				// before END must not be promoted into a successful stage.
				events, err = s.providerErrorEvents(streamGateErrorTunnelClosed)
			} else {
				events, err = s.translateFrame(frame)
			}
			if err != nil {
				return streamgate.NormalizedEvent{}, err
			}
			if len(events) == 0 {
				if !ok {
					return newOpenAIProviderErrorEvent(streamGateErrorTunnelClosed)
				}
				continue
			}
			first := events[0]
			if len(events) > 1 {
				s.mu.Lock()
				s.pending = append(s.pending, events[1:]...)
				s.mu.Unlock()
			}
			return first, nil
		}
	}
}

func (s *hotPathTunnelStageSource) markStarted() bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.started {
		return true
	}
	s.started = true
	return false
}

func (s *hotPathTunnelStageSource) setErrorStatus() {
	s.mu.Lock()
	s.errorStatus = true
	s.mu.Unlock()
}

func (s *hotPathTunnelStageSource) isErrorStatus() bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.errorStatus
}

func (s *hotPathTunnelStageSource) translateFrame(frame *iop.ProviderTunnelFrame) ([]streamgate.NormalizedEvent, error) {
	if frame == nil {
		return s.providerErrorEvents(streamGateErrorTunnelFailed)
	}
	switch frame.GetKind() {
	case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_RESPONSE_START:
		if s.markStarted() {
			return s.providerErrorEvents(streamGateErrorTunnelFailed)
		}
		status := int(frame.GetStatusCode())
		if status == 0 {
			status = http.StatusOK
		}
		if status >= http.StatusBadRequest {
			s.setErrorStatus()
		}
		ev, err := streamgate.NewResponseStartEvent(streamGateChannelDefault, status, sanitizedTunnelResponseHeaders(frame.GetHeaders()), time.Now())
		if err != nil {
			return nil, err
		}
		return []streamgate.NormalizedEvent{ev}, nil

	case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY:
		s.mu.Lock()
		started := s.started
		s.mu.Unlock()
		if !started {
			return s.providerErrorEvents(streamGateErrorTunnelFailed)
		}
		if s.isErrorStatus() {
			// A non-2xx body is opaque provider wire; the single terminal is a
			// provider error emitted when the transport closes.
			return nil, nil
		}
		decoded, err := s.decoder.decodeBody(frame.GetBody())
		if err != nil {
			return nil, err
		}
		return decoded, nil

	case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_USAGE:
		s.recordProtoUsage(frame.GetUsage())
		return nil, nil

	case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR:
		return s.providerErrorEvents(streamGateErrorTunnelFailed)

	case iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_END:
		return s.endEvents()

	default:
		return nil, nil
	}
}

// providerErrorEvents marks this stage terminal and returns exactly one
// sanitized provider-error event. It is shared by malformed frame ordering,
// incomplete channel closure, and provider ERROR frames.
func (s *hotPathTunnelStageSource) providerErrorEvents(code string) ([]streamgate.NormalizedEvent, error) {
	s.mu.Lock()
	if s.terminated {
		s.mu.Unlock()
		return nil, nil
	}
	s.terminated = true
	s.terminalCause = hotPathTerminalDisposition{
		Kind: hotPathDispositionProviderError, Cause: strings.TrimSpace(code), Source: "provider_tunnel",
	}
	s.mu.Unlock()
	ev, err := newOpenAIProviderErrorEvent(code)
	if err != nil {
		return nil, err
	}
	return []streamgate.NormalizedEvent{ev}, nil
}

// endEvents flushes any buffered decoder content and appends the single stage
// terminal (or provider error for a non-2xx transport), exactly once. END is
// valid only after the explicit RESPONSE_START frame.
func (s *hotPathTunnelStageSource) endEvents() ([]streamgate.NormalizedEvent, error) {
	s.mu.Lock()
	if s.terminated {
		s.mu.Unlock()
		return nil, nil
	}
	started := s.started
	errStatus := s.errorStatus
	s.mu.Unlock()
	if !started {
		return s.providerErrorEvents(streamGateErrorTunnelFailed)
	}

	s.mu.Lock()
	if s.terminated {
		s.mu.Unlock()
		return nil, nil
	}
	s.terminated = true
	s.mu.Unlock()

	var events []streamgate.NormalizedEvent
	flushed, err := s.decoder.finish()
	if err != nil {
		return nil, err
	}
	if _, ok := s.decoder.responseIdentity(); !ok && !errStatus {
		return nil, fmt.Errorf("hot path provider response identity is required before stage completion")
	}
	events = append(events, flushed...)
	if errStatus {
		ev, err := newOpenAIProviderErrorEvent(streamGateErrorTunnelFailed)
		if err != nil {
			return nil, err
		}
		return append(events, ev), nil
	}
	term, err := streamgate.NewTerminalEvent(streamGateChannelDefault, time.Now())
	if err != nil {
		return nil, err
	}
	return append(events, term), nil
}

func (s *hotPathTunnelStageSource) recordProtoUsage(u *iop.Usage) {
	if u == nil {
		return
	}
	usage := hotPathStageUsage{
		InputTokens:       int(u.GetInputTokens()),
		OutputTokens:      int(u.GetOutputTokens()),
		ReasoningTokens:   int(u.GetReasoningTokens()),
		CachedInputTokens: int(u.GetCachedInputTokens()),
		Reported:          true,
	}
	s.mu.Lock()
	if s.usageProto != nil {
		usage.ResponseID = s.usageProto.ResponseID
	}
	s.usageProto = &usage
	s.mu.Unlock()
}

func (s *hotPathTunnelStageSource) stageUsage() (hotPathStageUsage, bool) {
	s.mu.Lock()
	proto := s.usageProto
	s.mu.Unlock()
	decoded, ok := s.decoder.usageValue()
	if proto != nil {
		combined := *proto
		if combined.ResponseID == "" {
			combined.ResponseID = decoded.ResponseID
		}
		return combined, true
	}
	return decoded, ok
}

func (s *hotPathTunnelStageSource) stageIdentity() (string, bool) {
	return s.decoder.responseIdentity()
}

func (s *hotPathTunnelStageSource) stageTerminalReason() string {
	return s.decoder.terminalReason()
}

func (s *hotPathTunnelStageSource) stageTerminalCause() hotPathTerminalDisposition {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.terminalCause
}

func (s *hotPathTunnelStageSource) stageReasoningSignature() string {
	if probe, ok := s.decoder.(hotPathStageSignatureProbe); ok {
		return probe.stageReasoningSignature()
	}
	return ""
}

var (
	_ streamgate.NormalizedEventSource = (*hotPathTunnelStageSource)(nil)
	_ hotPathStageUsageProbe           = (*hotPathTunnelStageSource)(nil)
	_ hotPathStageIdentityProbe        = (*hotPathTunnelStageSource)(nil)
	_ hotPathStageTerminalReasonProbe  = (*hotPathTunnelStageSource)(nil)
	_ hotPathStageTerminalCauseProbe   = (*hotPathTunnelStageSource)(nil)
	_ hotPathStageSignatureProbe       = (*hotPathTunnelStageSource)(nil)
)

// hotPathStageTransportController gives one live stage runtime ownership of
// its service handle. Abort propagates cancellation before closing; graceful
// completion only closes the transport. Both paths claim ownership once.
type hotPathStageTransportController struct {
	mu       sync.Mutex
	claimed  bool
	service  runService
	dispatch edgeservice.RunDispatch
	close    func()
}

func newHotPathStageTransportController(service runService, dispatch edgeservice.RunDispatch, closeTransport func()) *hotPathStageTransportController {
	return &hotPathStageTransportController{service: service, dispatch: dispatch, close: closeTransport}
}

func (c *hotPathStageTransportController) claim() (func(), bool) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.claimed {
		return nil, false
	}
	c.claimed = true
	closeTransport := c.close
	c.close = nil
	return closeTransport, true
}

func (c *hotPathStageTransportController) AbortAttempt(ctx context.Context) error {
	closeTransport, claimed := c.claim()
	if !claimed {
		return nil
	}
	var cancelErr error
	if c.service != nil && strings.TrimSpace(c.dispatch.RunID) != "" {
		cancelCtx := context.Background()
		if ctx != nil {
			cancelCtx = context.WithoutCancel(ctx)
		}
		_, cancelErr = c.service.CancelRun(cancelCtx, edgeservice.CancelRunRequest{
			NodeRef: c.dispatch.NodeID, RunID: c.dispatch.RunID,
		})
	}
	if closeTransport != nil {
		closeTransport()
	}
	return cancelErr
}

func (c *hotPathStageTransportController) CloseAttempt(context.Context) error {
	closeTransport, claimed := c.claim()
	if !claimed {
		return nil
	}
	if closeTransport != nil {
		closeTransport()
	}
	return nil
}

var _ hotPathStageAttemptController = (*hotPathStageTransportController)(nil)

// --- OpenAI Chat SSE provider decoder ---------------------------------------

type openAIChatStageDecoder struct {
	pending             []byte
	tools               map[int]stageToolIdentity
	usage               hotPathStageUsage
	identity            hotPathProviderIdentity
	terminalReasonValue string
}

func newOpenAIChatStageDecoder() *openAIChatStageDecoder {
	return &openAIChatStageDecoder{tools: make(map[int]stageToolIdentity)}
}

func (d *openAIChatStageDecoder) decodeBody(body []byte) ([]streamgate.NormalizedEvent, error) {
	d.pending = append(d.pending, body...)
	var out []streamgate.NormalizedEvent
	for {
		frame, rest, ok := takeOpenAISSEFrame(d.pending)
		if !ok {
			break
		}
		d.pending = rest
		events, err := d.decodeFrame(frame)
		if err != nil {
			return nil, err
		}
		out = append(out, events...)
	}
	return out, nil
}

func (d *openAIChatStageDecoder) finish() ([]streamgate.NormalizedEvent, error) {
	if len(d.pending) == 0 {
		return nil, nil
	}
	frame := d.pending
	d.pending = nil
	if payload := bytes.TrimSpace(frame); len(payload) > 0 && json.Valid(payload) {
		stage, err := decodeOpenAIPresetJSON(payload)
		if err != nil {
			return nil, err
		}
		if err := d.identity.bind(stage.ResponseID); err != nil {
			return nil, err
		}
		if stage.OpenAIUsage != nil {
			d.usage = hotPathStageUsage{
				InputTokens: stage.OpenAIUsage.PromptTokens, OutputTokens: stage.OpenAIUsage.CompletionTokens,
				ReasoningTokens: stage.OpenAIUsage.ReasoningTokens, CachedInputTokens: stage.OpenAIUsage.CachedInputTokens,
				Reported: true,
			}
		}
		d.terminalReasonValue = stage.TerminalReason
		return hotPathNormalizedEvents(stage)
	}
	return d.decodeFrame(frame)
}

func (d *openAIChatStageDecoder) usageValue() (hotPathStageUsage, bool) {
	if responseID, ok := d.identity.get(); ok {
		d.usage.ResponseID = responseID
	}
	return d.usage, d.usage.Reported
}

func (d *openAIChatStageDecoder) responseIdentity() (string, bool) {
	return d.identity.get()
}

func (d *openAIChatStageDecoder) terminalReason() string {
	return strings.TrimSpace(d.terminalReasonValue)
}

func (d *openAIChatStageDecoder) decodeFrame(frame []byte) ([]streamgate.NormalizedEvent, error) {
	data := openAISSEData(frame)
	if data == "" && json.Valid(bytes.TrimSpace(frame)) {
		data = string(bytes.TrimSpace(frame))
	}
	trimmed := strings.TrimSpace(data)
	if trimmed == "" || trimmed == "[DONE]" {
		return nil, nil
	}
	var chunk struct {
		ID    string          `json:"id"`
		Usage json.RawMessage `json:"usage"`
		Error *struct {
			Message string `json:"message"`
		} `json:"error"`
		Choices []struct {
			Delta struct {
				Content          string `json:"content"`
				Reasoning        string `json:"reasoning"`
				ReasoningContent string `json:"reasoning_content"`
				ToolCalls        []struct {
					Index    int    `json:"index"`
					ID       string `json:"id"`
					Function struct {
						Name      string `json:"name"`
						Arguments string `json:"arguments"`
					} `json:"function"`
				} `json:"tool_calls"`
			} `json:"delta"`
			FinishReason *string `json:"finish_reason"`
		} `json:"choices"`
	}
	if err := json.Unmarshal([]byte(data), &chunk); err != nil {
		// Tolerate provider keep-alive/metadata frames that are not chat chunks.
		return nil, nil
	}
	if chunk.Error != nil {
		ev, err := newOpenAIProviderErrorEvent(streamGateErrorRunFailed)
		if err != nil {
			return nil, err
		}
		return []streamgate.NormalizedEvent{ev}, nil
	}
	if err := d.identity.bind(chunk.ID); err != nil {
		return nil, err
	}
	if len(chunk.Usage) > 0 && string(chunk.Usage) != "null" {
		if usage := decodeOpenAIUsage(chunk.Usage); usage != nil {
			d.usage.InputTokens = usage.PromptTokens
			d.usage.OutputTokens = usage.CompletionTokens
			d.usage.ReasoningTokens = usage.ReasoningTokens
			d.usage.CachedInputTokens = usage.CachedInputTokens
			d.usage.Reported = true
		}
	}
	var events []streamgate.NormalizedEvent
	for _, choice := range chunk.Choices {
		if choice.Delta.Content != "" || choice.Delta.Reasoning != "" || choice.Delta.ReasoningContent != "" || len(choice.Delta.ToolCalls) > 0 {
			if _, err := d.identity.require(); err != nil {
				return nil, err
			}
		}
		if choice.Delta.Content != "" {
			ev, err := streamgate.NewTextDeltaEvent(streamGateChannelDefault, choice.Delta.Content, time.Now())
			if err != nil {
				return nil, err
			}
			events = append(events, ev)
		}
		reasoning := choice.Delta.ReasoningContent
		if reasoning == "" {
			reasoning = choice.Delta.Reasoning
		}
		if reasoning != "" {
			ev, err := streamgate.NewReasoningDeltaEvent(streamGateChannelDefault, reasoning, time.Now())
			if err != nil {
				return nil, err
			}
			events = append(events, ev)
		}
		for _, tool := range choice.Delta.ToolCalls {
			identity := d.tools[tool.Index]
			if tool.ID != "" {
				identity.id = tool.ID
			}
			if tool.Function.Name != "" {
				identity.name = tool.Function.Name
			}
			d.tools[tool.Index] = identity
			if tool.Function.Arguments == "" {
				continue
			}
			ev, err := streamgate.NewToolCallFragmentEvent(
				streamGateChannelDefault,
				stageToolID(identity.id, tool.Index),
				stageToolName(identity.name),
				tool.Function.Arguments, time.Now(),
			)
			if err != nil {
				return nil, err
			}
			events = append(events, ev)
		}
		if choice.FinishReason != nil && strings.TrimSpace(*choice.FinishReason) != "" {
			d.terminalReasonValue = strings.TrimSpace(*choice.FinishReason)
		}
	}
	return events, nil
}

var _ hotPathProviderStageDecoder = (*openAIChatStageDecoder)(nil)

// --- Anthropic Messages SSE provider decoder --------------------------------

type anthropicStageTool struct {
	identity     stageToolIdentity
	inputEmitted bool
}

type anthropicMessagesStageDecoder struct {
	pending             []byte
	tools               map[int]anthropicStageTool
	usage               hotPathStageUsage
	identity            hotPathProviderIdentity
	terminalReasonValue string
	reasoningSignature  string
}

func newAnthropicMessagesStageDecoder() *anthropicMessagesStageDecoder {
	return &anthropicMessagesStageDecoder{tools: make(map[int]anthropicStageTool)}
}

func (d *anthropicMessagesStageDecoder) decodeBody(body []byte) ([]streamgate.NormalizedEvent, error) {
	d.pending = append(d.pending, body...)
	var out []streamgate.NormalizedEvent
	for {
		frame, rest, ok := takeOpenAISSEFrame(d.pending)
		if !ok {
			break
		}
		d.pending = rest
		events, err := d.decodeFrame(frame)
		if err != nil {
			return nil, err
		}
		out = append(out, events...)
	}
	return out, nil
}

func (d *anthropicMessagesStageDecoder) finish() ([]streamgate.NormalizedEvent, error) {
	if len(d.pending) == 0 {
		return nil, nil
	}
	frame := d.pending
	d.pending = nil
	if payload := bytes.TrimSpace(frame); len(payload) > 0 && json.Valid(payload) {
		stage, err := decodeAnthropicPresetJSON(payload)
		if err != nil {
			return nil, err
		}
		if err := d.identity.bind(stage.ResponseID); err != nil {
			return nil, err
		}
		d.recordAnthropicUsage(stage.Usage)
		d.terminalReasonValue = stage.TerminalReason
		d.reasoningSignature = stage.ReasoningSignature
		return hotPathNormalizedEvents(stage)
	}
	return d.decodeFrame(frame)
}

func (d *anthropicMessagesStageDecoder) usageValue() (hotPathStageUsage, bool) {
	if responseID, ok := d.identity.get(); ok {
		d.usage.ResponseID = responseID
	}
	return d.usage, d.usage.Reported
}

func (d *anthropicMessagesStageDecoder) responseIdentity() (string, bool) {
	return d.identity.get()
}

func (d *anthropicMessagesStageDecoder) terminalReason() string {
	return strings.TrimSpace(d.terminalReasonValue)
}

func (d *anthropicMessagesStageDecoder) stageReasoningSignature() string {
	return d.reasoningSignature
}

func (d *anthropicMessagesStageDecoder) decodeFrame(frame []byte) ([]streamgate.NormalizedEvent, error) {
	data := openAISSEData(frame)
	if strings.TrimSpace(data) == "" {
		return nil, nil
	}
	var envelope struct {
		Type string `json:"type"`
	}
	if err := json.Unmarshal([]byte(data), &envelope); err != nil {
		return nil, nil
	}
	switch envelope.Type {
	case "message_start":
		var payload struct {
			Message struct {
				ID    string          `json:"id"`
				Usage json.RawMessage `json:"usage"`
			} `json:"message"`
		}
		if err := json.Unmarshal([]byte(data), &payload); err != nil {
			return nil, nil
		}
		if err := d.identity.bind(payload.Message.ID); err != nil {
			return nil, err
		}
		d.recordAnthropicUsage(payload.Message.Usage)
		return nil, nil
	case "content_block_start":
		if _, err := d.identity.require(); err != nil {
			return nil, err
		}
		return d.decodeBlockStart(data)
	case "content_block_delta":
		if _, err := d.identity.require(); err != nil {
			return nil, err
		}
		return d.decodeBlockDelta(data)
	case "content_block_stop":
		if _, err := d.identity.require(); err != nil {
			return nil, err
		}
		return d.decodeBlockStop(data)
	case "message_delta":
		var payload struct {
			Delta struct {
				StopReason string `json:"stop_reason"`
			} `json:"delta"`
			Usage json.RawMessage `json:"usage"`
		}
		if err := json.Unmarshal([]byte(data), &payload); err != nil {
			return nil, nil
		}
		if reason := strings.TrimSpace(payload.Delta.StopReason); reason != "" {
			d.terminalReasonValue = reason
		}
		d.recordAnthropicUsage(payload.Usage)
		return nil, nil
	case "error":
		ev, err := newOpenAIProviderErrorEvent(streamGateErrorTunnelFailed)
		if err != nil {
			return nil, err
		}
		return []streamgate.NormalizedEvent{ev}, nil
	default:
		return nil, nil
	}
}

func (d *anthropicMessagesStageDecoder) decodeBlockStart(data string) ([]streamgate.NormalizedEvent, error) {
	var payload struct {
		Index int `json:"index"`
		Block struct {
			Type     string          `json:"type"`
			Text     string          `json:"text"`
			Thinking string          `json:"thinking"`
			ID       string          `json:"id"`
			Name     string          `json:"name"`
			Input    json.RawMessage `json:"input"`
		} `json:"content_block"`
	}
	if err := json.Unmarshal([]byte(data), &payload); err != nil {
		return nil, nil
	}
	switch payload.Block.Type {
	case "text":
		if payload.Block.Text == "" {
			return nil, nil
		}
		ev, err := streamgate.NewTextDeltaEvent(streamGateChannelDefault, payload.Block.Text, time.Now())
		if err != nil {
			return nil, err
		}
		return []streamgate.NormalizedEvent{ev}, nil
	case "thinking":
		if payload.Block.Thinking == "" {
			return nil, nil
		}
		ev, err := streamgate.NewReasoningDeltaEvent(streamGateChannelDefault, payload.Block.Thinking, time.Now())
		if err != nil {
			return nil, err
		}
		return []streamgate.NormalizedEvent{ev}, nil
	case "tool_use":
		hasInput := len(payload.Block.Input) > 0 && string(payload.Block.Input) != "{}" && string(payload.Block.Input) != "null"
		d.tools[payload.Index] = anthropicStageTool{
			identity:     stageToolIdentity{id: payload.Block.ID, name: payload.Block.Name},
			inputEmitted: hasInput,
		}
		if !hasInput {
			return nil, nil
		}
		ev, err := streamgate.NewToolCallFragmentEvent(
			streamGateChannelDefault,
			stageToolID(payload.Block.ID, payload.Index),
			stageToolName(payload.Block.Name),
			string(payload.Block.Input), time.Now(),
		)
		if err != nil {
			return nil, err
		}
		return []streamgate.NormalizedEvent{ev}, nil
	default:
		return nil, nil
	}
}

func (d *anthropicMessagesStageDecoder) decodeBlockDelta(data string) ([]streamgate.NormalizedEvent, error) {
	var payload 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([]byte(data), &payload); err != nil {
		return nil, nil
	}
	switch payload.Delta.Type {
	case "text_delta":
		if payload.Delta.Text == "" {
			return nil, nil
		}
		ev, err := streamgate.NewTextDeltaEvent(streamGateChannelDefault, payload.Delta.Text, time.Now())
		if err != nil {
			return nil, err
		}
		return []streamgate.NormalizedEvent{ev}, nil
	case "thinking_delta":
		if payload.Delta.Thinking == "" {
			return nil, nil
		}
		ev, err := streamgate.NewReasoningDeltaEvent(streamGateChannelDefault, payload.Delta.Thinking, time.Now())
		if err != nil {
			return nil, err
		}
		return []streamgate.NormalizedEvent{ev}, nil
	case "signature_delta":
		d.reasoningSignature += payload.Delta.Signature
		return nil, nil
	case "input_json_delta":
		if payload.Delta.PartialJSON == "" {
			return nil, nil
		}
		tool, ok := d.tools[payload.Index]
		if !ok {
			return nil, nil
		}
		tool.inputEmitted = true
		d.tools[payload.Index] = tool
		ev, err := streamgate.NewToolCallFragmentEvent(
			streamGateChannelDefault,
			stageToolID(tool.identity.id, payload.Index),
			stageToolName(tool.identity.name),
			payload.Delta.PartialJSON, time.Now(),
		)
		if err != nil {
			return nil, err
		}
		return []streamgate.NormalizedEvent{ev}, nil
	default:
		return nil, nil
	}
}

func (d *anthropicMessagesStageDecoder) decodeBlockStop(data string) ([]streamgate.NormalizedEvent, error) {
	var payload struct {
		Index int `json:"index"`
	}
	if err := json.Unmarshal([]byte(data), &payload); err != nil {
		return nil, nil
	}
	tool, ok := d.tools[payload.Index]
	if !ok {
		return nil, nil
	}
	delete(d.tools, payload.Index)
	if tool.inputEmitted {
		return nil, nil
	}
	ev, err := streamgate.NewToolCallFragmentEvent(
		streamGateChannelDefault,
		stageToolID(tool.identity.id, payload.Index),
		stageToolName(tool.identity.name),
		"{}", time.Now(),
	)
	if err != nil {
		return nil, err
	}
	return []streamgate.NormalizedEvent{ev}, nil
}

func (d *anthropicMessagesStageDecoder) recordAnthropicUsage(raw json.RawMessage) {
	if len(raw) == 0 || string(raw) == "null" {
		return
	}
	var usage struct {
		InputTokens          int `json:"input_tokens"`
		OutputTokens         int `json:"output_tokens"`
		CacheReadInputTokens int `json:"cache_read_input_tokens"`
	}
	if json.Unmarshal(raw, &usage) != nil {
		return
	}
	if usage.InputTokens > 0 {
		d.usage.InputTokens = usage.InputTokens
	}
	if usage.OutputTokens > 0 {
		d.usage.OutputTokens = usage.OutputTokens
	}
	if usage.CacheReadInputTokens > 0 {
		d.usage.CachedInputTokens = usage.CacheReadInputTokens
	}
	d.usage.Reported = true
}

var _ hotPathProviderStageDecoder = (*anthropicMessagesStageDecoder)(nil)
var _ hotPathStageSignatureProbe = (*anthropicMessagesStageDecoder)(nil)

func hotPathNormalizedEvents(stage normalizedStageOutput) ([]streamgate.NormalizedEvent, error) {
	events := make([]streamgate.NormalizedEvent, 0, len(stage.Deltas)+len(stage.ToolCalls)+2)
	now := time.Now()
	if len(stage.Deltas) > 0 {
		for _, delta := range stage.Deltas {
			var (
				event streamgate.NormalizedEvent
				err   error
			)
			switch delta.Kind {
			case normalizedStageDeltaText:
				event, err = streamgate.NewTextDeltaEvent(streamGateChannelDefault, delta.Text, now)
			case normalizedStageDeltaReasoning:
				event, err = streamgate.NewReasoningDeltaEvent(streamGateChannelDefault, delta.Text, now)
			case normalizedStageDeltaTool:
				event, err = streamgate.NewToolCallFragmentEvent(
					streamGateChannelDefault, delta.ToolID, delta.ToolName, delta.Arguments, now,
				)
			default:
				continue
			}
			if err != nil {
				return nil, err
			}
			events = append(events, event)
		}
		return events, nil
	}
	if stage.Reasoning != "" {
		event, err := streamgate.NewReasoningDeltaEvent(streamGateChannelDefault, stage.Reasoning, now)
		if err != nil {
			return nil, err
		}
		events = append(events, event)
	}
	if stage.Content != "" {
		event, err := streamgate.NewTextDeltaEvent(streamGateChannelDefault, stage.Content, now)
		if err != nil {
			return nil, err
		}
		events = append(events, event)
	}
	for _, call := range stage.ToolCalls {
		providerID := hotPathFirstNonEmpty(call.ProviderCallID, call.ID)
		event, err := streamgate.NewToolCallFragmentEvent(
			streamGateChannelDefault, providerID, call.Name, directToolArguments(call), now,
		)
		if err != nil {
			return nil, err
		}
		events = append(events, event)
	}
	return events, nil
}

// --- Shared stage tool identity helpers -------------------------------------

func stageToolID(id string, index int) string {
	if strings.TrimSpace(id) != "" {
		return id
	}
	return fmt.Sprintf("stage-tool-%d", index)
}

func stageToolName(name string) string {
	if strings.TrimSpace(name) != "" {
		return name
	}
	return "function"
}
