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

package openai

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

	"iop/packages/go/streamgate"
)

// errHotPathTurnTerminal is returned by the outer turn when a stage tries to
// open, release, or terminate after the single public terminal has committed.
var errHotPathTurnTerminal = errors.New("hot path outer turn already committed a terminal")

// hotPathDispositionKind is the closed terminal vocabulary shared by the
// stage runtime, the HTTP-turn sequencer, and Light cleanup/orphan handoff.
// Endpoint codecs translate these values later; no wire status or body shape
// is owned here.
type hotPathDispositionKind string

const (
	hotPathDispositionSuccess         hotPathDispositionKind = "success"
	hotPathDispositionToolTurn        hotPathDispositionKind = "tool_turn"
	hotPathDispositionLength          hotPathDispositionKind = "length"
	hotPathDispositionProviderError   hotPathDispositionKind = "provider_error"
	hotPathDispositionValidationError hotPathDispositionKind = "validation_error"
	hotPathDispositionTimeout         hotPathDispositionKind = "timeout"
	hotPathDispositionCallerCancel    hotPathDispositionKind = "caller_cancel"
)

type hotPathTerminalDisposition struct {
	Kind       hotPathDispositionKind
	Cause      string
	Source     string
	StageID    string
	Generation uint64
}

func (d hotPathTerminalDisposition) valid() bool {
	switch d.Kind {
	case hotPathDispositionSuccess,
		hotPathDispositionToolTurn,
		hotPathDispositionLength,
		hotPathDispositionProviderError,
		hotPathDispositionValidationError,
		hotPathDispositionTimeout,
		hotPathDispositionCallerCancel:
		return true
	default:
		return false
	}
}

func hotPathDispositionForSuccess(reason string, hasTools bool) hotPathDispositionKind {
	if hasTools || reason == "tool_calls" || reason == "tool_use" || reason == "function_call" {
		return hotPathDispositionToolTurn
	}
	if hotPathIsProviderLengthTerminal(reason) {
		return hotPathDispositionLength
	}
	return hotPathDispositionSuccess
}

func hotPathDispositionForError(err error) hotPathDispositionKind {
	if status, ok := providerHTTPStatus(err); ok && status == http.StatusBadRequest {
		return hotPathDispositionValidationError
	}
	switch {
	case errors.Is(err, context.Canceled):
		return hotPathDispositionCallerCancel
	case errors.Is(err, context.DeadlineExceeded), errors.Is(err, errRunTimedOut):
		return hotPathDispositionTimeout
	default:
		return hotPathDispositionProviderError
	}
}

type hotPathDispositionError struct {
	disposition hotPathTerminalDisposition
	err         error
}

func (e *hotPathDispositionError) Error() string {
	if e == nil || e.err == nil {
		return "hot path terminal disposition"
	}
	return e.err.Error()
}

func (e *hotPathDispositionError) Unwrap() error {
	if e == nil {
		return nil
	}
	return e.err
}

func hotPathDispositionFromError(err error) (hotPathTerminalDisposition, bool) {
	var dispositionErr *hotPathDispositionError
	if errors.As(err, &dispositionErr) && dispositionErr.disposition.valid() {
		return dispositionErr.disposition, true
	}
	return hotPathTerminalDisposition{}, false
}

func newHotPathDispositionError(kind hotPathDispositionKind, source, stageID string, err error) error {
	if err == nil {
		return nil
	}
	return &hotPathDispositionError{
		disposition: hotPathTerminalDisposition{
			Kind: kind, Cause: err.Error(), Source: source, StageID: strings.TrimSpace(stageID),
		},
		err: err,
	}
}

// hotPathStageMeta is the protocol-neutral correlation a provider stage
// contributes to one HTTP turn. It carries model/provider/path identity for the
// stage-scoped runtime but never credentials, provider targets, or caller
// endpoint wire state.
type hotPathStageMeta struct {
	StageID       string
	Protocol      string // "openai" | "anthropic"
	Model         string
	Provider      string
	ExecutionPath string
	ResponseID    string
	AttemptID     string
}

func (m hotPathStageMeta) token() string {
	return hotPathFirstNonEmpty(m.StageID, m.ResponseID, m.AttemptID, "stage")
}

// hotPathTurnUsage is the deduplicated, aggregated token usage across every
// internal stage of one HTTP turn.
type hotPathTurnUsage struct {
	InputTokens       int
	OutputTokens      int
	ReasoningTokens   int
	CachedInputTokens int
	Reported          bool
}

// hotPathStageTerminal is the typed transition evidence a stage runtime's
// terminal is converted into by the stage release sink. It is never rendered to
// the caller: the outer turn alone owns whether and when a single public
// terminal is committed.
type hotPathStageTerminal struct {
	Success     bool
	Reason      string
	ErrType     string
	ErrCode     string
	Usage       hotPathStageUsage
	HasUsage    bool
	Disposition hotPathTerminalDisposition
}

// hotPathReleasedDelta records one progressively released public delta in turn
// order. Tests use it as the ordering oracle; production renderers consume the
// compatibility accumulator instead.
type hotPathReleasedDelta struct {
	Kind     streamgate.EventKind
	Text     string
	PublicID string
	Name     string
	Args     string
}

type hotPathReleaseCallback func(hotPathReleasedDelta) error

// hotPathReleaseCallbackError marks a failure from the endpoint-owned release
// callback. Only this boundary means the caller can no longer receive output;
// release preparation failures must retain their normal stage-runtime meaning.
type hotPathReleaseCallbackError struct {
	err error
}

func (e *hotPathReleaseCallbackError) Error() string { return e.err.Error() }

func (e *hotPathReleaseCallbackError) Unwrap() error { return e.err }

type hotPathTurnTool struct {
	publicID   string
	providerID string
	name       string
	args       strings.Builder
}

// hotPathOutputBudget keeps the three caller-cap states distinct. Remaining
// zero is exhausted only when Limited is true; an unlimited turn never uses a
// sentinel provider value.
type hotPathOutputBudget struct {
	Limited      bool
	Remaining    int
	Exhausted    bool
	MissingUsage bool
}

type hotPathTurnError struct {
	errType string
	code    string
}

// hotPathOuterTurn is the single protocol-neutral sequencer that survives stage
// replacement inside one HTTP request. It owns the public block/tool id remap,
// deduplicated usage aggregation, the caller output-cap budget, response-start
// suppression, the single terminal guard, and a compatibility accumulator that
// later caller codecs render. It holds nothing on behalf of the stage runtimes:
// nonterminal deltas are appended as they are released.
type hotPathOuterTurn struct {
	releaseMu sync.Mutex
	mu        sync.Mutex

	publicResponseID string
	channel          string
	outputCapTokens  int // 0 => no caller token cap

	started           bool
	terminalCommitted bool
	terminalReason    string
	terminalError     *hotPathTurnError
	disposition       *hotPathTerminalDisposition
	activeStage       *hotPathActiveStageController
	activeGeneration  uint64

	stageSeq int

	toolPublic map[string]*hotPathTurnTool
	toolOrder  []*hotPathTurnTool
	toolSeq    int
	toolID     func() (string, error)

	usageSeen          map[string]struct{}
	usage              hotPathTurnUsage
	previewUsage       hotPathStageUsage
	reasoningSignature string
	missingUsage       bool
	capExhausted       bool

	content   strings.Builder
	reasoning strings.Builder

	released []hotPathReleasedDelta
	release  hotPathReleaseCallback
}

// newHotPathOuterTurn builds one HTTP-turn sequencer. publicResponseID is the
// turn-scoped identity exposed to the caller regardless of internal stage
// response ids. Token budgeting is configured separately and never derives
// token counts from the caller-visible payload.
func newHotPathOuterTurn(publicResponseID string) *hotPathOuterTurn {
	publicResponseID = strings.TrimSpace(publicResponseID)
	return &hotPathOuterTurn{
		publicResponseID: publicResponseID,
		channel:          streamGateChannelDefault,
		toolPublic:       make(map[string]*hotPathTurnTool),
		usageSeen:        make(map[string]struct{}),
	}
}

// bindPublicResponseID fixes the first provider-owned response identity for
// the HTTP turn. Later stages may have different provider response identities,
// but they cannot replace the already-bound public outer identity.
func (t *hotPathOuterTurn) bindPublicResponseID(responseID string) error {
	if t == nil {
		return errors.New("hot path outer turn is unavailable")
	}
	responseID = strings.TrimSpace(responseID)
	if responseID == "" {
		return errors.New("hot path public response identity is empty")
	}
	t.mu.Lock()
	defer t.mu.Unlock()
	if t.publicResponseID == "" {
		t.publicResponseID = responseID
	}
	return nil
}

func (t *hotPathOuterTurn) publicResponseIdentity() (string, bool) {
	if t == nil {
		return "", false
	}
	t.mu.Lock()
	defer t.mu.Unlock()
	return t.publicResponseID, t.publicResponseID != ""
}

func (t *hotPathOuterTurn) setReleaseCallback(callback hotPathReleaseCallback) error {
	if t == nil {
		return errors.New("hot path outer turn is unavailable")
	}
	t.mu.Lock()
	defer t.mu.Unlock()
	if len(t.released) > 0 {
		return errors.New("hot path release callback was attached after visible output")
	}
	t.release = callback
	return nil
}

// setToolIDAllocator fixes the caller-owned tool identity allocator before a
// progressively released Light stage can expose its first tool fragment.
func (t *hotPathOuterTurn) setToolIDAllocator(allocate func() (string, error)) error {
	if t == nil || allocate == nil {
		return errors.New("hot path tool identity allocator is unavailable")
	}
	t.mu.Lock()
	defer t.mu.Unlock()
	if t.toolID != nil || len(t.toolOrder) > 0 {
		return errors.New("hot path tool identity allocator is already fixed")
	}
	t.toolID = allocate
	return nil
}

// newHotPathCallerCappedOuterTurn keeps the caller limit in provider-reported
// tokens. The outer turn never truncates content or fabricates token usage from
// characters or bytes.
func newHotPathCallerCappedOuterTurn(publicResponseID string, outputCapTokens int) *hotPathOuterTurn {
	outer := newHotPathOuterTurn(publicResponseID)
	if outputCapTokens > 0 {
		outer.outputCapTokens = outputCapTokens
	}
	return outer
}

// beginStage assigns the next stage-scope index. Tool ids are remapped per
// stage index so the same provider tool id emitted by two internal stages never
// collides in the public turn.
func (t *hotPathOuterTurn) beginStage() int {
	t.mu.Lock()
	defer t.mu.Unlock()
	t.stageSeq++
	return t.stageSeq
}

// openResponse records the single outer envelope open. The first internal stage
// opens it; every nested provider response-start is suppressed. It fails closed
// once the turn terminal is committed.
func (t *hotPathOuterTurn) openResponse(streamgate.ResponseStart) error {
	t.mu.Lock()
	defer t.mu.Unlock()
	if t.terminalCommitted {
		return errHotPathTurnTerminal
	}
	t.started = true
	return nil
}

// releaseDelta appends one progressively released nonterminal delta and remaps
// tool ids into the turn scope. Caller-visible payload is never locally
// truncated; output budgeting is based only on provider-reported token usage.
// It fails closed after the turn terminal is committed.
func (t *hotPathOuterTurn) releaseDelta(stageSeq int, ev streamgate.ReleaseEvent) error {
	_, err := t.releaseDeltaRecorded(stageSeq, ev)
	return err
}

func (t *hotPathOuterTurn) releaseDeltaRecorded(stageSeq int, ev streamgate.ReleaseEvent) (*hotPathReleasedDelta, error) {
	t.releaseMu.Lock()
	defer t.releaseMu.Unlock()
	t.mu.Lock()
	if t.terminalCommitted {
		t.mu.Unlock()
		return nil, errHotPathTurnTerminal
	}
	t.started = true
	var released hotPathReleasedDelta
	switch ev.Kind() {
	case streamgate.EventKindTextDelta:
		text, err := ev.AsTextDelta()
		if err != nil {
			t.mu.Unlock()
			return nil, err
		}
		t.content.WriteString(text)
		released = hotPathReleasedDelta{Kind: ev.Kind(), Text: text}
	case streamgate.EventKindReasoningDelta:
		text, err := ev.AsReasoningDelta()
		if err != nil {
			t.mu.Unlock()
			return nil, err
		}
		t.reasoning.WriteString(text)
		released = hotPathReleasedDelta{Kind: ev.Kind(), Text: text}
	case streamgate.EventKindToolCallFragment:
		call, err := ev.AsToolCallFragment()
		if err != nil {
			t.mu.Unlock()
			return nil, err
		}
		tool, err := t.remapToolLocked(stageSeq, call)
		if err != nil {
			t.mu.Unlock()
			return nil, err
		}
		tool.args.WriteString(call.Arguments)
		released = hotPathReleasedDelta{
			Kind: ev.Kind(), PublicID: tool.publicID, Name: tool.name, Args: call.Arguments,
		}
	default:
		t.mu.Unlock()
		return nil, fmt.Errorf("hot path outer turn cannot release event kind %q", ev.Kind())
	}
	t.released = append(t.released, released)
	callback := t.release
	t.mu.Unlock()
	if callback != nil {
		if err := callback(released); err != nil {
			return nil, &hotPathReleaseCallbackError{err: err}
		}
	}
	return &released, nil
}

func (t *hotPathOuterTurn) setToolNameLocked(tool *hotPathTurnTool, name string) {
	if tool == nil || name == "" {
		return
	}
	tool.name = name
}

// remapToolLocked resolves the turn-scoped public tool identity for one provider
// fragment. Fragments that share a stage index and provider id assemble under
// one public id; a provider id reused by another stage gets a fresh public id.
func (t *hotPathOuterTurn) remapToolLocked(stageSeq int, call streamgate.ToolCall) (*hotPathTurnTool, error) {
	key := fmt.Sprintf("%d\x00%s", stageSeq, call.ID)
	if tool, ok := t.toolPublic[key]; ok {
		if tool.name == "" && call.Name != "" {
			t.setToolNameLocked(tool, call.Name)
		}
		return tool, nil
	}
	t.toolSeq++
	publicID := fmt.Sprintf("%s-tool-%d", t.publicResponseID, t.toolSeq)
	if t.toolID != nil {
		allocated, err := t.toolID()
		if err != nil {
			return nil, fmt.Errorf("allocate hot path public tool identity: %w", err)
		}
		if !validLogicalRequestID(allocated) {
			return nil, errors.New("allocated hot path public tool identity is invalid")
		}
		publicID = allocated
	}
	tool := &hotPathTurnTool{
		publicID:   publicID,
		providerID: call.ID,
	}
	t.setToolNameLocked(tool, call.Name)
	t.toolPublic[key] = tool
	t.toolOrder = append(t.toolOrder, tool)
	return tool, nil
}

// recordStageTerminal folds a held stage terminal's usage into the turn without
// committing any public terminal.
func (t *hotPathOuterTurn) recordStageTerminal(term hotPathStageTerminal) {
	t.mu.Lock()
	defer t.mu.Unlock()
	if term.Success && t.outputCapTokens > 0 && (!term.HasUsage || !term.Usage.Reported) {
		t.missingUsage = true
	}
	if term.HasUsage {
		t.aggregateUsageLocked(term.Usage)
	}
}

// selectDisposition elects the logical terminal intent once. Public HTTP-turn
// commitment remains separate so a provider/validation failure can first emit
// a caller-owned cleanup tool frontier while preserving the original terminal
// responsibility for the following continuation.
func (t *hotPathOuterTurn) selectDisposition(disposition hotPathTerminalDisposition) bool {
	if t == nil || !disposition.valid() {
		return false
	}
	t.mu.Lock()
	defer t.mu.Unlock()
	return t.selectDispositionLocked(disposition)
}

func (t *hotPathOuterTurn) selectDispositionLocked(disposition hotPathTerminalDisposition) bool {
	if t.disposition != nil {
		return false
	}
	selected := disposition
	t.disposition = &selected
	return true
}

func (t *hotPathOuterTurn) terminalDisposition() (hotPathTerminalDisposition, bool) {
	if t == nil {
		return hotPathTerminalDisposition{}, false
	}
	t.mu.Lock()
	defer t.mu.Unlock()
	if t.disposition == nil {
		return hotPathTerminalDisposition{}, false
	}
	return *t.disposition, true
}

func (t *hotPathOuterTurn) activeStageDisposition(kind hotPathDispositionKind, source, cause string) hotPathTerminalDisposition {
	disposition := hotPathTerminalDisposition{Kind: kind, Source: source, Cause: strings.TrimSpace(cause)}
	if t == nil {
		return disposition
	}
	t.mu.Lock()
	defer t.mu.Unlock()
	if t.activeStage != nil {
		disposition.StageID = t.activeStage.stageID
		disposition.Generation = t.activeStage.generation
	}
	return disposition
}

// cancelActiveStage elects timeout/caller-cancel ownership and aborts only the
// controller registered for the current generation. A caller cancel also
// closes the public release gate immediately, which keeps the wire silent even
// if a stale source callback arrives after context cancellation.
func (t *hotPathOuterTurn) cancelActiveStage(kind hotPathDispositionKind, source string, cause error) bool {
	if t == nil {
		return false
	}
	var active *hotPathActiveStageController
	disposition := hotPathTerminalDisposition{Kind: kind, Source: source}
	if cause != nil {
		disposition.Cause = cause.Error()
	}
	t.mu.Lock()
	if t.activeStage != nil {
		active = t.activeStage
		disposition.StageID = active.stageID
		disposition.Generation = active.generation
	}
	won := t.selectDispositionLocked(disposition)
	if won && kind == hotPathDispositionCallerCancel {
		t.terminalCommitted = true
		t.terminalReason = string(kind)
		t.terminalError = &hotPathTurnError{errType: string(kind), code: string(kind)}
	}
	t.mu.Unlock()
	if won && active != nil {
		_ = active.AbortAttempt(context.Background())
	}
	return won
}

// aggregateUsageLocked sums normalized stage usage, deduplicating by provider
// response id so a stage that reports usage twice (or a duplicate provider
// response id across stages) is only counted once.
func (t *hotPathOuterTurn) aggregateUsageLocked(u hotPathStageUsage) {
	if u.ResponseID != "" {
		if _, seen := t.usageSeen[u.ResponseID]; seen {
			return
		}
		t.usageSeen[u.ResponseID] = struct{}{}
	}
	t.usage.InputTokens += u.InputTokens
	t.usage.OutputTokens += u.OutputTokens
	t.usage.ReasoningTokens += u.ReasoningTokens
	t.usage.CachedInputTokens += u.CachedInputTokens
	if u.Reported {
		t.usage.Reported = true
	}
}

// commitTerminalSuccess commits the single public success terminal. It returns
// true only for the first terminal; every later success, error, or cancel is a
// guarded no-op so exactly one outer terminal ever wins. A visible tool owns
// the current HTTP terminal even at cap; exhaustion becomes length only when
// there is no caller continuation frontier.
func (t *hotPathOuterTurn) commitTerminalSuccess(reason string) bool {
	t.mu.Lock()
	defer t.mu.Unlock()
	if t.terminalCommitted {
		return false
	}
	t.selectDispositionLocked(hotPathTerminalDisposition{
		Kind:  hotPathDispositionForSuccess(reason, len(t.toolOrder) > 0),
		Cause: strings.TrimSpace(reason), Source: "outer_turn",
	})
	t.terminalCommitted = true
	reason = strings.TrimSpace(reason)
	switch {
	case len(t.toolOrder) > 0:
		if reason != "tool_calls" && reason != "tool_use" {
			reason = "tool_calls"
		}
	case reason == "":
		reason = "stop"
	}
	t.terminalReason = reason
	return true
}

// commitTerminalError commits the single public error/cancel terminal under the
// same exactly-once guard as commitTerminalSuccess.
func (t *hotPathOuterTurn) commitTerminalError(errType, code string) bool {
	t.mu.Lock()
	defer t.mu.Unlock()
	if t.terminalCommitted {
		return false
	}
	kind := hotPathDispositionProviderError
	if strings.Contains(strings.ToLower(errType), "invalid") || strings.Contains(strings.ToLower(code), "validation") {
		kind = hotPathDispositionValidationError
	}
	t.selectDispositionLocked(hotPathTerminalDisposition{
		Kind: kind, Cause: hotPathFirstNonEmpty(code, errType), Source: "outer_turn",
	})
	t.terminalCommitted = true
	t.terminalError = &hotPathTurnError{errType: strings.TrimSpace(errType), code: strings.TrimSpace(code)}
	t.terminalReason = strings.TrimSpace(errType)
	return true
}

// accumulator returns the compatibility view a later caller codec renders: the
// turn public id, remapped tool calls with assembled arguments, aggregated
// usage, and the resolved terminal reason.
func (t *hotPathOuterTurn) accumulator() normalizedStageOutput {
	t.mu.Lock()
	defer t.mu.Unlock()
	out := normalizedStageOutput{
		ResponseID: t.publicResponseID,
		Content:    t.content.String(),
		Reasoning:  t.reasoning.String(),
	}
	for _, tool := range t.toolOrder {
		out.ToolCalls = append(out.ToolCalls, normalizedToolCall{
			ID:             tool.publicID,
			ProviderCallID: hotPathFirstNonEmpty(tool.providerID, tool.publicID),
			Name:           tool.name,
			RawArgs:        tool.args.String(),
		})
	}
	if t.usage.Reported {
		usage := &openAIUsage{
			PromptTokens:      t.usage.InputTokens,
			CompletionTokens:  t.usage.OutputTokens,
			TotalTokens:       t.usage.InputTokens + t.usage.OutputTokens,
			ReasoningTokens:   t.usage.ReasoningTokens,
			CachedInputTokens: t.usage.CachedInputTokens,
		}
		out.OpenAIUsage = usage
		out.Usage, _ = json.Marshal(usage)
	}
	out.TerminalReason = t.terminalReasonLocked()
	return out
}

func (t *hotPathOuterTurn) terminalReasonLocked() string {
	if t.terminalReason != "" {
		return t.terminalReason
	}
	if len(t.toolOrder) > 0 {
		return "tool_calls"
	}
	if t.capExhausted {
		return "length"
	}
	return "stop"
}

// releasedDeltas returns a defensive copy of the ordered release log.
func (t *hotPathOuterTurn) releasedDeltas() []hotPathReleasedDelta {
	t.mu.Lock()
	defer t.mu.Unlock()
	return append([]hotPathReleasedDelta(nil), t.released...)
}

func (t *hotPathOuterTurn) turnUsage() hotPathTurnUsage {
	t.mu.Lock()
	defer t.mu.Unlock()
	return t.usage
}

func (t *hotPathOuterTurn) setPreviewUsage(usage hotPathStageUsage) {
	if t == nil || !usage.Reported {
		return
	}
	t.mu.Lock()
	t.previewUsage = usage
	t.mu.Unlock()
}

func (t *hotPathOuterTurn) currentPreviewUsage() (hotPathStageUsage, bool) {
	if t == nil {
		return hotPathStageUsage{}, false
	}
	t.mu.Lock()
	defer t.mu.Unlock()
	return t.previewUsage, t.previewUsage.Reported
}

func (t *hotPathOuterTurn) setReasoningSignature(signature string) {
	if t == nil || signature == "" {
		return
	}
	t.mu.Lock()
	t.reasoningSignature = signature
	t.mu.Unlock()
}

func (t *hotPathOuterTurn) currentReasoningSignature() string {
	if t == nil {
		return ""
	}
	t.mu.Lock()
	defer t.mu.Unlock()
	return t.reasoningSignature
}

// reportedOutputTokens returns only deduplicated provider-reported output
// usage. Caller-visible payload length is intentionally unrelated.
func (t *hotPathOuterTurn) reportedOutputTokens() int {
	if t == nil {
		return 0
	}
	t.mu.Lock()
	defer t.mu.Unlock()
	return t.usage.OutputTokens
}

// outputBudget reports whether another provider stage may be dispatched from
// this HTTP turn. It is deliberately independent from current-terminal tool
// ownership: an exhausted budget blocks a later provider stage, but does not
// discard a visible tool call that still requires a caller result frontier.
func (t *hotPathOuterTurn) outputBudget() hotPathOutputBudget {
	if t == nil {
		return hotPathOutputBudget{}
	}
	t.mu.Lock()
	defer t.mu.Unlock()
	if t.outputCapTokens <= 0 {
		return hotPathOutputBudget{}
	}
	remaining := t.outputCapTokens - t.usage.OutputTokens
	exhausted := remaining <= 0
	if remaining < 0 {
		remaining = 0
	}
	return hotPathOutputBudget{
		Limited: true, Remaining: remaining, Exhausted: exhausted,
		MissingUsage: t.missingUsage,
	}
}

// commitLengthTerminal marks provider-usage-driven exhaustion before the
// endpoint codec renders the one public length terminal.
func (t *hotPathOuterTurn) commitLengthTerminal() bool {
	if t == nil {
		return false
	}
	t.mu.Lock()
	t.capExhausted = true
	t.mu.Unlock()
	return t.commitTerminalSuccess("length")
}

// projectToolIdentities installs the public/provider mapping allocated by the
// logical-request or workspace frontier without changing the accumulator's
// capped arguments or ordering. It must run before that frontier is registered.
func (t *hotPathOuterTurn) projectToolIdentities(calls []normalizedToolCall) error {
	if t == nil {
		return nil
	}
	t.mu.Lock()
	defer t.mu.Unlock()
	if len(calls) != len(t.toolOrder) {
		return fmt.Errorf("hot path tool projection count %d does not match accumulated count %d", len(calls), len(t.toolOrder))
	}
	for index, call := range calls {
		publicID := strings.TrimSpace(call.ID)
		if publicID == "" {
			return fmt.Errorf("hot path tool projection %d is missing public identity", index)
		}
		tool := t.toolOrder[index]
		tool.publicID = publicID
		tool.providerID = hotPathFirstNonEmpty(call.ProviderCallID, tool.providerID, publicID)
		if call.Name != "" {
			t.setToolNameLocked(tool, call.Name)
		}
	}
	return nil
}

// recordCollectedStage is the compatibility bridge for existing collectors.
// It keeps the outer turn's accounting and terminal ownership authoritative
// while legacy endpoint renderers still consume normalizedStageOutput rather
// than ReleaseEvent values directly.  Stage output is recorded once per
// provider response identity, matching normal stage-runtime aggregation.
func (t *hotPathOuterTurn) recordCollectedStage(output normalizedStageOutput) {
	if t == nil || output.OpenAIUsage == nil {
		return
	}
	t.recordStageTerminal(hotPathStageTerminal{HasUsage: true, Usage: hotPathStageUsage{
		ResponseID: output.ResponseID, InputTokens: output.OpenAIUsage.PromptTokens,
		OutputTokens: output.OpenAIUsage.CompletionTokens, ReasoningTokens: output.OpenAIUsage.ReasoningTokens,
		CachedInputTokens: output.OpenAIUsage.CachedInputTokens, Reported: true,
	}})
}

// hotPathCollectedStageSource adapts the pre-existing compatibility collector
// to the stage-scoped runtime.  It is intentionally transitional: provider
// transport sources can replace it without changing outer-turn ownership or
// endpoint rendering, while every collected stage already follows the same
// response-start/delta/held-terminal lifecycle.
type hotPathCollectedStageSource struct {
	events []streamgate.NormalizedEvent
	index  int
	usage  hotPathStageUsage
}

func newHotPathCollectedStageSource(output normalizedStageOutput) (*hotPathCollectedStageSource, error) {
	now := time.Now()
	events := make([]streamgate.NormalizedEvent, 0, 4+len(output.Deltas)+len(output.ToolCalls))
	start, err := streamgate.NewResponseStartEvent(streamGateChannelDefault, 200, nil, now)
	if err != nil {
		return nil, err
	}
	events = append(events, start)
	if len(output.Deltas) > 0 {
		for _, delta := range output.Deltas {
			var event streamgate.NormalizedEvent
			switch delta.Kind {
			case normalizedStageDeltaReasoning:
				event, err = streamgate.NewReasoningDeltaEvent(streamGateChannelDefault, delta.Text, now)
			case normalizedStageDeltaText:
				event, err = streamgate.NewTextDeltaEvent(streamGateChannelDefault, delta.Text, now)
			case normalizedStageDeltaTool:
				event, err = streamgate.NewToolCallFragmentEvent(
					streamGateChannelDefault, delta.ToolID, delta.ToolName, delta.Arguments, now,
				)
			default:
				err = fmt.Errorf("unsupported normalized stage delta kind %q", delta.Kind)
			}
			if err != nil {
				return nil, err
			}
			events = append(events, event)
		}
	} else {
		if output.Reasoning != "" {
			event, err := streamgate.NewReasoningDeltaEvent(streamGateChannelDefault, output.Reasoning, now)
			if err != nil {
				return nil, err
			}
			events = append(events, event)
		}
		if output.Content != "" {
			event, err := streamgate.NewTextDeltaEvent(streamGateChannelDefault, output.Content, now)
			if err != nil {
				return nil, err
			}
			events = append(events, event)
		}
		for _, call := range output.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)
		}
	}
	terminal, err := streamgate.NewTerminalEvent(streamGateChannelDefault, now)
	if err != nil {
		return nil, err
	}
	events = append(events, terminal)
	source := &hotPathCollectedStageSource{events: events, usage: hotPathStageUsage{ResponseID: output.ResponseID}}
	if output.OpenAIUsage != nil {
		source.usage.InputTokens = output.OpenAIUsage.PromptTokens
		source.usage.OutputTokens = output.OpenAIUsage.CompletionTokens
		source.usage.ReasoningTokens = output.OpenAIUsage.ReasoningTokens
		source.usage.CachedInputTokens = output.OpenAIUsage.CachedInputTokens
		source.usage.Reported = true
	} else if len(output.Usage) > 0 {
		var usage anthropicUsage
		if err := json.Unmarshal(output.Usage, &usage); err == nil {
			source.usage.InputTokens = usage.InputTokens
			source.usage.OutputTokens = usage.OutputTokens
			source.usage.CachedInputTokens = usage.CacheReadInputTokens
			source.usage.Reported = true
		}
	}
	return source, nil
}

func (s *hotPathCollectedStageSource) NextEvent(context.Context) (streamgate.NormalizedEvent, error) {
	if s.index >= len(s.events) {
		return streamgate.NormalizedEvent{}, errors.New("hot path collected stage exhausted")
	}
	event := s.events[s.index]
	s.index++
	return event, nil
}

func (s *hotPathCollectedStageSource) stageUsage() (hotPathStageUsage, bool) {
	return s.usage, s.usage.Reported
}

type hotPathCollectedStageController struct{}

func (hotPathCollectedStageController) AbortAttempt(context.Context) error { return nil }
func (hotPathCollectedStageController) CloseAttempt(context.Context) error { return nil }

func runHotPathCollectedStage(ctx context.Context, outer *hotPathOuterTurn, stageID string, output normalizedStageOutput) error {
	if err := outer.bindPublicResponseID(output.ResponseID); err != nil {
		return err
	}
	source, err := newHotPathCollectedStageSource(output)
	if err != nil {
		return err
	}
	_, err = runHotPathStage(ctx, outer, hotPathStageMeta{
		StageID: stageID, Model: "hot-path-collected", Provider: "collector",
		ExecutionPath: "collected", ResponseID: output.ResponseID, AttemptID: output.ResponseID,
	}, source, source, hotPathCollectedStageController{})
	return err
}

// hotPathRemainingOutputTokens is retained as a narrow compatibility helper for
// tests and builders. Exhaustion is zero; callers that need to distinguish it
// from unlimited use hotPathOutputBudget directly.
func hotPathRemainingOutputTokens(cap int, outer *hotPathOuterTurn) int {
	if cap <= 0 {
		return cap
	}
	if outer == nil {
		return cap
	}
	outer.mu.Lock()
	defer outer.mu.Unlock()
	used := outer.usage.OutputTokens
	if used >= cap {
		return 0
	}
	return cap - used
}

// hotPathCompatibilityOutput preserves endpoint-required provider metadata
// from the final stage while projecting every caller-visible payload, public
// tool identity, aggregate usage, and terminal reason from the outer turn.
func hotPathCompatibilityOutput(outer *hotPathOuterTurn, final normalizedStageOutput, protocol string) normalizedStageOutput {
	if outer == nil || final.CallerStageOnly {
		return final
	}
	result := cloneNormalizedStageOutput(final)
	accumulated := outer.accumulator()
	result.Content = accumulated.Content
	result.Reasoning = accumulated.Reasoning
	result.ToolCalls = cloneNormalizedStageOutput(accumulated).ToolCalls
	if accumulated.OpenAIUsage != nil {
		result.OpenAIUsage = accumulated.OpenAIUsage
		usage := make(map[string]any)
		_ = json.Unmarshal(final.Usage, &usage)
		if protocol == "anthropic" {
			delete(usage, "prompt_tokens")
			delete(usage, "completion_tokens")
			delete(usage, "total_tokens")
			delete(usage, "reasoning_tokens")
			delete(usage, "cached_input_tokens")
			usage["input_tokens"] = accumulated.OpenAIUsage.PromptTokens
			usage["output_tokens"] = accumulated.OpenAIUsage.CompletionTokens
			if accumulated.OpenAIUsage.CachedInputTokens > 0 {
				usage["cache_read_input_tokens"] = accumulated.OpenAIUsage.CachedInputTokens
			}
		} else {
			usage["prompt_tokens"] = accumulated.OpenAIUsage.PromptTokens
			usage["completion_tokens"] = accumulated.OpenAIUsage.CompletionTokens
			usage["total_tokens"] = accumulated.OpenAIUsage.PromptTokens + accumulated.OpenAIUsage.CompletionTokens
		}
		result.Usage, _ = json.Marshal(usage)
	}
	result.TerminalReason = accumulated.TerminalReason
	return result
}

func (t *hotPathOuterTurn) capExhaustedFlag() bool {
	t.mu.Lock()
	defer t.mu.Unlock()
	return t.capExhausted
}

func (t *hotPathOuterTurn) isTerminalCommitted() bool {
	t.mu.Lock()
	defer t.mu.Unlock()
	return t.terminalCommitted
}

// hotPathStageReleaseSink is the boundary between one stage-scoped Core runtime
// and the shared outer turn. Nonterminal deltas are forwarded immediately; the
// stage terminal is captured as typed transition evidence and folded into the
// turn without committing any public terminal.
type hotPathStageReleaseSink struct {
	outer          *hotPathOuterTurn
	active         *hotPathActiveStageController
	stageSeq       int
	usage          hotPathStageUsageProbe
	identity       hotPathStageIdentityProbe
	terminalReason hotPathStageTerminalReasonProbe
	terminalCause  hotPathStageTerminalCauseProbe
	signature      hotPathStageSignatureProbe

	mu          sync.Mutex
	terminal    *hotPathStageTerminal
	content     strings.Builder
	reasoning   strings.Builder
	tools       map[string]*hotPathProjectedTool
	toolOrder   []string
	deltas      []normalizedStageDelta
	progressive bool
}

type hotPathProjectedTool struct {
	name string
	args strings.Builder
}

func (s *hotPathStageReleaseSink) CommitResponseStart(_ context.Context, rs streamgate.ResponseStart) (streamgate.CommitState, error) {
	if s.active != nil && !s.active.isCurrent() {
		return streamgate.CommitStateStreamOpen, nil
	}
	if err := s.outer.openResponse(rs); err != nil {
		return "", err
	}
	return streamgate.CommitStateStreamOpen, nil
}

func (s *hotPathStageReleaseSink) Release(_ context.Context, ev streamgate.ReleaseEvent) (streamgate.CommitState, error) {
	if s.active != nil && !s.active.isCurrent() {
		return streamgate.CommitStateStreamOpen, nil
	}
	if s.identity != nil {
		responseID, ok := s.identity.stageIdentity()
		if !ok {
			return "", errors.New("hot path live stage is missing provider response identity")
		}
		if err := s.outer.bindPublicResponseID(responseID); err != nil {
			return "", err
		}
	} else if _, ok := s.outer.publicResponseIdentity(); !ok {
		return "", errors.New("hot path stage cannot release without a public response identity")
	}
	if s.usage != nil {
		if usage, ok := s.usage.stageUsage(); ok {
			s.outer.setPreviewUsage(usage)
		}
	}
	if s.signature != nil {
		s.outer.setReasoningSignature(s.signature.stageReasoningSignature())
	}
	released, err := s.outer.releaseDeltaRecorded(s.stageSeq, ev)
	if err != nil {
		var callbackErr *hotPathReleaseCallbackError
		if !errors.As(err, &callbackErr) {
			return "", err
		}
		stageID := ""
		if s.active != nil {
			stageID = s.active.stageID
		}
		return "", newHotPathDispositionError(hotPathDispositionCallerCancel, "caller_write", stageID, callbackErr)
	}
	if released != nil {
		s.mu.Lock()
		s.progressive = true
		s.mu.Unlock()
		if err := s.recordReleased(ev, *released); err != nil {
			return "", err
		}
	}
	return streamgate.CommitStateStreamOpen, nil
}

func (s *hotPathStageReleaseSink) recordReleased(ev streamgate.ReleaseEvent, released hotPathReleasedDelta) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	switch released.Kind {
	case streamgate.EventKindTextDelta:
		s.content.WriteString(released.Text)
		s.deltas = append(s.deltas, normalizedStageDelta{Kind: normalizedStageDeltaText, Text: released.Text})
	case streamgate.EventKindReasoningDelta:
		s.reasoning.WriteString(released.Text)
		s.deltas = append(s.deltas, normalizedStageDelta{Kind: normalizedStageDeltaReasoning, Text: released.Text})
	case streamgate.EventKindToolCallFragment:
		call, err := ev.AsToolCallFragment()
		if err != nil {
			return err
		}
		tool := s.tools[call.ID]
		if tool == nil {
			tool = &hotPathProjectedTool{}
			s.tools[call.ID] = tool
			s.toolOrder = append(s.toolOrder, call.ID)
		}
		if call.Name != "" {
			tool.name = call.Name
		}
		tool.args.WriteString(released.Args)
		s.deltas = append(s.deltas, normalizedStageDelta{
			Kind: normalizedStageDeltaTool, ToolID: call.ID, ToolName: tool.name, Arguments: released.Args,
		})
	}
	return nil
}

func (s *hotPathStageReleaseSink) CommitTerminal(_ context.Context, tr streamgate.TerminalResult) (streamgate.CommitState, error) {
	if s.active != nil && !s.active.isCurrent() {
		return streamgate.CommitStateTerminalCommitted, nil
	}
	term := hotPathStageTerminal{Success: tr.Success()}
	if tr.Error() {
		if desc := tr.ExternalDesc(); desc != nil {
			term.ErrType = desc.Type()
			term.ErrCode = desc.Code()
		}
	} else {
		term.Reason = hotPathTerminalReasonOrStop("")
		if s.terminalReason != nil {
			term.Reason = hotPathTerminalReasonOrStop(s.terminalReason.stageTerminalReason())
		}
	}
	if term.Success {
		term.Disposition = hotPathTerminalDisposition{
			Kind: hotPathDispositionForSuccess(term.Reason, false), Cause: term.Reason,
			Source: "stage_terminal", StageID: s.active.stageID, Generation: s.active.generation,
		}
	} else {
		kind := hotPathDispositionProviderError
		if s.terminalCause != nil && s.terminalCause.stageTerminalCause().valid() {
			kind = s.terminalCause.stageTerminalCause().Kind
		}
		term.Disposition = hotPathTerminalDisposition{
			Kind: kind, Cause: hotPathFirstNonEmpty(term.ErrCode, term.ErrType),
			Source: "stage_terminal", StageID: s.active.stageID, Generation: s.active.generation,
		}
	}
	if s.usage != nil {
		if u, ok := s.usage.stageUsage(); ok {
			term.Usage = u
			term.HasUsage = true
		}
	}
	if term.Success && s.identity != nil {
		if _, ok := s.identity.stageIdentity(); !ok {
			return "", errors.New("hot path live stage completed without provider response identity")
		}
	}
	s.mu.Lock()
	if s.terminal != nil {
		s.mu.Unlock()
		return streamgate.CommitStateTerminalCommitted, nil
	}
	s.terminal = &term
	s.mu.Unlock()
	s.outer.recordStageTerminal(term)
	return streamgate.CommitStateTerminalCommitted, nil
}

func (s *hotPathStageReleaseSink) stageOutput() (normalizedStageOutput, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	responseID := ""
	if s.identity != nil {
		responseID, _ = s.identity.stageIdentity()
	}
	output := normalizedStageOutput{
		ResponseID: responseID, Content: s.content.String(), Reasoning: s.reasoning.String(),
		Deltas: append([]normalizedStageDelta(nil), s.deltas...), ProgressivelyReleased: s.progressive,
	}
	if s.signature != nil {
		output.ReasoningSignature = s.signature.stageReasoningSignature()
	}
	for _, providerID := range s.toolOrder {
		tool := s.tools[providerID]
		call, err := normalizedToolCallFromParts(providerID, tool.name, tool.args.String())
		if err != nil {
			return normalizedStageOutput{}, err
		}
		output.ToolCalls = append(output.ToolCalls, call)
	}
	if s.terminal != nil {
		output.TerminalReason = hotPathTerminalReasonOrStop(s.terminal.Reason)
	} else {
		output.TerminalReason = hotPathTerminalReasonOrStop("")
	}
	if len(output.ToolCalls) > 0 {
		output.TerminalReason = "tool_calls"
	}
	if s.terminal != nil && s.terminal.HasUsage {
		u := s.terminal.Usage
		output.OpenAIUsage = &openAIUsage{
			PromptTokens: u.InputTokens, CompletionTokens: u.OutputTokens,
			TotalTokens: u.InputTokens + u.OutputTokens, ReasoningTokens: u.ReasoningTokens,
			CachedInputTokens: u.CachedInputTokens,
		}
		output.Usage, _ = json.Marshal(output.OpenAIUsage)
	}
	return output, nil
}

// stageTerminal returns the held stage terminal evidence, if the stage runtime
// committed one.
func (s *hotPathStageReleaseSink) stageTerminal() (hotPathStageTerminal, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.terminal == nil {
		return hotPathStageTerminal{}, false
	}
	return *s.terminal, true
}

var _ streamgate.ReleaseSink = (*hotPathStageReleaseSink)(nil)

func hotPathTerminalReasonOrStop(reason string) string {
	reason = strings.TrimSpace(reason)
	if reason == "" {
		return "stop"
	}
	return reason
}

// hotPathStageAttemptController owns one real stage transport. Implementations
// must make both operations idempotent: Core invokes AbortAttempt for errors or
// cancellation and CloseAttempt after a successful terminal.
type hotPathStageAttemptController interface {
	streamgate.AttemptController
	CloseAttempt(context.Context) error
}

// hotPathStageAttemptOwner preserves one stage transport's ownership when Core
// reaches the same cleanup path through both an error terminal and final
// resource cleanup. The wrapped transport observes at most one abort and one
// graceful close request.
type hotPathStageAttemptOwner struct {
	controller hotPathStageAttemptController

	abortOnce sync.Once
	abortErr  error
	closeOnce sync.Once
	closeErr  error
}

func newHotPathStageAttemptOwner(controller hotPathStageAttemptController) *hotPathStageAttemptOwner {
	return &hotPathStageAttemptOwner{controller: controller}
}

func (o *hotPathStageAttemptOwner) AbortAttempt(ctx context.Context) error {
	o.abortOnce.Do(func() {
		o.abortErr = o.controller.AbortAttempt(ctx)
	})
	return o.abortErr
}

func (o *hotPathStageAttemptOwner) CloseAttempt(ctx context.Context) error {
	o.closeOnce.Do(func() {
		o.closeErr = o.controller.CloseAttempt(ctx)
	})
	return o.closeErr
}

// hotPathActiveStageController is the generation-fenced registration stored by
// one outer turn. It shares the same idempotent owner with the Core attempt, so
// a context watcher, Core abort, stale callback, and final resource cleanup can
// never issue duplicate CancelRun calls.
type hotPathActiveStageController struct {
	outer      *hotPathOuterTurn
	stageID    string
	generation uint64
	owner      *hotPathStageAttemptOwner

	actionOnce sync.Once
	actionErr  error
	finishOnce sync.Once
}

func (t *hotPathOuterTurn) registerActiveStage(stageID string, controller hotPathStageAttemptController) (*hotPathActiveStageController, error) {
	if t == nil || controller == nil {
		return nil, errors.New("hot path active stage controller is unavailable")
	}
	stageID = strings.TrimSpace(stageID)
	if stageID == "" {
		return nil, errors.New("hot path active stage identity is empty")
	}
	t.mu.Lock()
	defer t.mu.Unlock()
	if t.terminalCommitted {
		return nil, errHotPathTurnTerminal
	}
	if t.activeStage != nil {
		return nil, fmt.Errorf("hot path stage %q is still active", t.activeStage.stageID)
	}
	t.activeGeneration++
	active := &hotPathActiveStageController{
		outer: t, stageID: stageID, generation: t.activeGeneration,
		owner: newHotPathStageAttemptOwner(controller),
	}
	t.activeStage = active
	return active, nil
}

func (c *hotPathActiveStageController) isCurrent() bool {
	if c == nil || c.outer == nil {
		return false
	}
	c.outer.mu.Lock()
	defer c.outer.mu.Unlock()
	return c.outer.activeStage == c && c.outer.activeGeneration == c.generation
}

func (c *hotPathActiveStageController) unregister() {
	if c == nil || c.outer == nil {
		return
	}
	c.finishOnce.Do(func() {
		c.outer.mu.Lock()
		if c.outer.activeStage == c && c.outer.activeGeneration == c.generation {
			c.outer.activeStage = nil
		}
		c.outer.mu.Unlock()
	})
}

func (c *hotPathActiveStageController) AbortAttempt(ctx context.Context) error {
	if c == nil || c.owner == nil {
		return nil
	}
	c.actionOnce.Do(func() {
		c.actionErr = c.owner.AbortAttempt(ctx)
		c.unregister()
	})
	return c.actionErr
}

func (c *hotPathActiveStageController) CloseAttempt(ctx context.Context) error {
	if c == nil || c.owner == nil {
		return nil
	}
	c.actionOnce.Do(func() {
		c.actionErr = c.owner.CloseAttempt(ctx)
		c.unregister()
	})
	return c.actionErr
}

var _ hotPathStageAttemptController = (*hotPathActiveStageController)(nil)

// hotPathStageNoRecoveryDispatcher / hotPathStageNoRecoveryRebuilder satisfy the
// required Core recovery seams for a stage runtime configured with zero fault
// recovery. They are never invoked and fail closed if they ever are.
type hotPathStageNoRecoveryDispatcher struct{}

func (hotPathStageNoRecoveryDispatcher) DispatchAttempt(context.Context, streamgate.RebuiltRequest) (streamgate.AttemptBinding, error) {
	return streamgate.AttemptBinding{}, errors.New("hot path stage runtime does not recover")
}

type hotPathStageNoRecoveryRebuilder struct{}

func (hotPathStageNoRecoveryRebuilder) RebuildRequest(context.Context, streamgate.RecoveryRequestSnapshotRef, streamgate.RecoveryPlan) (streamgate.RebuiltRequestDraft, error) {
	return streamgate.RebuiltRequestDraft{}, errors.New("hot path stage runtime does not rebuild")
}

// newHotPathStageRuntime builds a stage-scoped Core runtime for one provider
// stage. The runtime commits a terminal exactly once per stage, but its release
// sink converts that into held evidence, so the runtime lifecycle ends while the
// outer turn survives for the next stage on the same HTTP request.
func newHotPathStageRuntime(outer *hotPathOuterTurn, meta hotPathStageMeta, source streamgate.NormalizedEventSource, usage hotPathStageUsageProbe, controller hotPathStageAttemptController) (*streamgate.RequestRuntime, *hotPathStageReleaseSink, error) {
	if outer == nil {
		return nil, nil, errors.New("hot path stage runtime requires an outer turn")
	}
	if source == nil {
		return nil, nil, errors.New("hot path stage runtime requires an event source")
	}
	if controller == nil {
		return nil, nil, errors.New("hot path stage runtime requires an attempt controller")
	}

	stageSeq := outer.beginStage()
	identity, _ := source.(hotPathStageIdentityProbe)
	terminalReason, _ := source.(hotPathStageTerminalReasonProbe)
	terminalCause, _ := source.(hotPathStageTerminalCauseProbe)
	signature, _ := source.(hotPathStageSignatureProbe)

	opts, err := streamgate.NewRuntimeOptions(
		streamgate.DefaultMaxEvidenceRunes,
		streamgate.DefaultMaxBufferRunes,
		streamgate.DefaultMaxIngressSnapshotBytes,
		0,
		streamgate.GateCoordinatorOptions{},
		streamgate.RecoveryCoordinatorOptions{},
	)
	if err != nil {
		return nil, nil, err
	}
	registry, err := openAIStreamGateRegistrySnapshotWith()
	if err != nil {
		return nil, nil, err
	}
	snapRef, err := streamgate.NewRecoveryRequestSnapshotRef(
		openAIStreamGateSafeToken("stage", meta.token()),
		0, 0, uint64(streamgate.DefaultMaxIngressSnapshotBytes),
	)
	if err != nil {
		return nil, nil, err
	}

	model := hotPathFirstNonEmpty(meta.Model, "hot-path-stage")
	provider := hotPathFirstNonEmpty(meta.Provider, "hot-path-provider")
	execPath := hotPathFirstNonEmpty(meta.ExecutionPath, "normalized")
	active, err := outer.registerActiveStage(meta.StageID, controller)
	if err != nil {
		return nil, nil, err
	}
	sink := &hotPathStageReleaseSink{
		outer: outer, active: active, stageSeq: stageSeq, usage: usage, identity: identity,
		terminalReason: terminalReason, terminalCause: terminalCause, signature: signature,
		tools: make(map[string]*hotPathProjectedTool),
	}

	binding, err := streamgate.NewAttemptBinding(
		openAIStreamGateSafeToken("attempt", hotPathFirstNonEmpty(meta.AttemptID, meta.ResponseID, meta.StageID)),
		model, provider, execPath, source, active,
	)
	if err != nil {
		_ = active.AbortAttempt(context.Background())
		return nil, nil, err
	}

	snapshot, err := streamgate.NewRequestRuntimeSnapshot(
		openAIStreamGateSafeToken("stage-req", meta.token()),
		streamGateConfigGeneration, streamGateEnvironment, "hot-path-stage", "hot-path",
		opts, registry, nil, snapRef,
		hotPathStageNoRecoveryDispatcher{}, hotPathStageNoRecoveryRebuilder{},
		nil, nil, sink,
	)
	if err != nil {
		_ = active.AbortAttempt(context.Background())
		return nil, nil, err
	}

	rt, err := streamgate.NewRequestRuntime(snapshot, model, binding)
	if err != nil {
		_ = active.AbortAttempt(context.Background())
		return nil, nil, err
	}
	return rt, sink, nil
}

// runHotPathStage runs one stage runtime to its held stage terminal and returns
// the typed evidence. The outer turn is untouched by stage completion, so the
// caller can immediately build the next stage on the same turn.
func runHotPathStage(ctx context.Context, outer *hotPathOuterTurn, meta hotPathStageMeta, source streamgate.NormalizedEventSource, usage hotPathStageUsageProbe, controller hotPathStageAttemptController) (hotPathStageTerminal, error) {
	rt, sink, err := newHotPathStageRuntime(outer, meta, source, usage, controller)
	if err != nil {
		return hotPathStageTerminal{}, err
	}
	term, _, runErr := runHotPathRequestRuntime(ctx, outer, rt, sink)
	if runErr != nil {
		return hotPathStageTerminal{}, wrapHotPathDispositionError(outer, meta.StageID, runErr)
	}
	return term, nil
}

func runHotPathStreamingStage(ctx context.Context, outer *hotPathOuterTurn, meta hotPathStageMeta, source streamgate.NormalizedEventSource, usage hotPathStageUsageProbe, controller hotPathStageAttemptController) (normalizedStageOutput, hotPathStageTerminal, error) {
	rt, sink, err := newHotPathStageRuntime(outer, meta, source, usage, controller)
	if err != nil {
		return normalizedStageOutput{}, hotPathStageTerminal{}, err
	}
	term, _, runErr := runHotPathRequestRuntime(ctx, outer, rt, sink)
	if runErr != nil {
		return normalizedStageOutput{}, hotPathStageTerminal{}, wrapHotPathDispositionError(outer, meta.StageID, runErr)
	}
	output, err := sink.stageOutput()
	if err != nil {
		return normalizedStageOutput{}, hotPathStageTerminal{}, err
	}
	return output, term, nil
}

func runHotPathRequestRuntime(
	ctx context.Context,
	outer *hotPathOuterTurn,
	rt *streamgate.RequestRuntime,
	sink *hotPathStageReleaseSink,
) (hotPathStageTerminal, bool, error) {
	watchStop := make(chan struct{})
	watchDone := make(chan struct{})
	go func() {
		defer close(watchDone)
		select {
		case <-ctx.Done():
			kind := hotPathDispositionForError(ctx.Err())
			if kind == hotPathDispositionCallerCancel || kind == hotPathDispositionTimeout {
				outer.cancelActiveStage(kind, "caller_context", ctx.Err())
			}
		case <-watchStop:
		}
	}()

	runErr := rt.Run(ctx)
	close(watchStop)
	<-watchDone
	term, committed := sink.stageTerminal()
	if runErr != nil {
		kind := hotPathDispositionForError(runErr)
		source := "stage_runtime"
		if disposition, ok := hotPathDispositionFromError(runErr); ok {
			kind = disposition.Kind
			source = disposition.Source
		}
		if kind == hotPathDispositionCallerCancel || kind == hotPathDispositionTimeout {
			outer.cancelActiveStage(kind, source, runErr)
		} else {
			outer.selectDisposition(outer.activeStageDisposition(kind, source, runErr.Error()))
		}
	} else if committed && !term.Success {
		if !term.Disposition.valid() {
			term.Disposition = outer.activeStageDisposition(hotPathDispositionProviderError, "stage_terminal", term.ErrCode)
		}
		outer.selectDisposition(term.Disposition)
	}
	_ = rt.CloseRequestResources(context.Background(), runErr == nil && committed && term.Success)
	return term, committed, runErr
}

func wrapHotPathDispositionError(outer *hotPathOuterTurn, stageID string, err error) error {
	if err == nil {
		return nil
	}
	if _, ok := hotPathDispositionFromError(err); ok {
		return err
	}
	if disposition, ok := outer.terminalDisposition(); ok {
		return &hotPathDispositionError{disposition: disposition, err: err}
	}
	return &hotPathDispositionError{disposition: hotPathTerminalDisposition{
		Kind: hotPathDispositionForError(err), Cause: err.Error(), Source: "stage_runtime", StageID: stageID,
	}, err: err}
}

func hotPathFirstNonEmpty(values ...string) string {
	for _, value := range values {
		if strings.TrimSpace(value) != "" {
			return value
		}
	}
	return ""
}