package provider

import (
	"bytes"
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"runtime"
	"strings"
	"sync"
	"time"
)

// Codex (ChatGPT subscription) client. Uses OpenAI's Responses API via
// chatgpt.com/backend-api with the chatgpt-account-id handshake.
//
// Wire protocol notes:
//   - Endpoint: POST https://chatgpt.com/backend-api/codex/responses
//   - Headers: Authorization: Bearer <access_token>, chatgpt-account-id: <id>,
//     OpenAI-Beta: responses=experimental, originator: zot (required legacy value)
//   - Body: OpenAI Responses API shape (not chat/completions).
//     input: [{role, content: [{type: "input_text" | "input_image" | ... }]}]
//     instructions: <system prompt>
//     tools: [{type:"function", name, description, parameters, strict}]
//     stream: true
//   - SSE events: response.output_item.added, response.output_text.delta,
//     response.function_call_arguments.delta, response.output_item.done,
//     response.completed, response.failed, error
//
// The access_token comes from the OpenAI OAuth flow (auth.openai.com).
// The accountID is parsed from the id_token JWT's `chatgpt_account_id`
// claim; see auth/oauth.go.

const codexDefaultBaseURL = "https://chatgpt.com/backend-api/codex/responses"

type codexClient struct {
	token             string
	accountID         string
	baseURL           string
	errorLabel        string
	providerName      string
	modelName         func(string) string
	disableCLIRouting bool
	cliRoutingAll     bool
	capabilities      responsesCapabilities
	http              *http.Client
	modelOverridesMu  sync.RWMutex
	modelOverrides    map[string]Model
}

// responsesCapabilities records extensions verified for the exact provider
// client, rather than assuming every Responses-shaped endpoint accepts them.
type responsesCapabilities struct {
	StablePromptCacheKey          bool
	ExplicitPromptCacheBreakpoint bool
	SessionHeader                 bool
	ThreadHeader                  bool
	ClientRequestHeader           bool
}

// NewOpenAICodex creates a Responses client for ChatGPT's Codex endpoint
// using a subscription OAuth access token and the user's ChatGPT account ID.
// It prefers WebSocket mode and falls back to HTTP/SSE before visible output.
// baseURL may be empty to use the default.
func NewOpenAICodex(token, accountID, baseURL string) Client {
	return newResponsesWebSocketClient(newOpenAICodexClient(token, accountID, baseURL))
}

func newOpenAICodexClient(token, accountID, baseURL string) *codexClient {
	if baseURL == "" {
		baseURL = codexDefaultBaseURL
	}
	return &codexClient{
		token:        token,
		accountID:    accountID,
		baseURL:      strings.TrimRight(baseURL, "/"),
		errorLabel:   "codex",
		providerName: "openai-codex",
		// The ChatGPT Codex backend load-sheds requests from client
		// identities it does not recognize: unknown originator/user-agent
		// pairs receive "Our servers are currently overloaded" stream
		// errors near-deterministically while Codex CLI requests succeed.
		// Send the Codex CLI request shape for every model on this
		// backend.
		cliRoutingAll: true,
		capabilities: responsesCapabilities{
			StablePromptCacheKey: true,
			SessionHeader:        true,
			ThreadHeader:         true,
			ClientRequestHeader:  true,
		},
		http: &http.Client{Timeout: 0},
	}
}

func (c *codexClient) Name() string { return ProviderOpenAICodex }

func (c *codexClient) SetModelMetadata(model Model) {
	model = cloneModel(model)
	c.modelOverridesMu.Lock()
	if c.modelOverrides == nil {
		c.modelOverrides = make(map[string]Model)
	}
	c.modelOverrides[model.ID] = model
	c.modelOverridesMu.Unlock()
}

// ---- Responses API wire types (subset needed for zut's surface) ----

type codexInputText struct {
	Type                  string                      `json:"type"` // "input_text"
	Text                  string                      `json:"text"`
	PromptCacheBreakpoint *codexPromptCacheBreakpoint `json:"prompt_cache_breakpoint,omitempty"`
}

type codexInputImage struct {
	Type     string `json:"type"` // "input_image"
	Detail   string `json:"detail"`
	ImageURL string `json:"image_url"`
}

type codexOutputText struct {
	Type        string `json:"type"` // "output_text"
	Text        string `json:"text"`
	Annotations []any  `json:"annotations"`
	Phase       string `json:"phase,omitempty"`
}

type codexPromptCacheOptions struct {
	Mode string `json:"mode"`
}

type codexPromptCacheBreakpoint struct {
	Mode string `json:"mode"`
}

type codexInputMessage struct {
	Type    string `json:"type,omitempty"` // "message" (optional for input)
	Role    string `json:"role"`
	Content []any  `json:"content"`
}

type codexOutputMessage struct {
	Type    string            `json:"type"` // "message"
	Role    string            `json:"role"`
	Status  string            `json:"status,omitempty"`
	ID      string            `json:"id,omitempty"`
	Content []codexOutputText `json:"content"`
}

type codexFunctionCall struct {
	Type      string `json:"type"` // "function_call"
	ID        string `json:"id,omitempty"`
	CallID    string `json:"call_id"`
	Name      string `json:"name"`
	Arguments string `json:"arguments"` // JSON string
}

type codexFunctionCallOutput struct {
	Type   string `json:"type"` // "function_call_output"
	CallID string `json:"call_id"`
	Output string `json:"output"` // string (or ResponseFunctionCallOutputItemList for images; v1 only uses string)
}

// codexReasoningItem mirrors the Responses API "reasoning" output item.
// We capture it on incoming streams and replay it verbatim on follow-up
// requests: the API rejects assistant tool-call replays without it when
// thinking is enabled.
type codexReasoningItem struct {
	Type             string `json:"type"` // "reasoning"
	ID               string `json:"id,omitempty"`
	EncryptedContent string `json:"encrypted_content,omitempty"`
	// Summary is required by the Responses API even when no summary text
	// was streamed; encode an empty array rather than omitting the field.
	Summary []codexReasoningSummary `json:"summary"`
}

type codexReasoningSummary struct {
	Type string `json:"type"` // "summary_text"
	Text string `json:"text"`
}

type codexTool struct {
	Type        string          `json:"type"` // "function"
	Name        string          `json:"name"`
	Description string          `json:"description,omitempty"`
	Parameters  json.RawMessage `json:"parameters"`
	// Responses otherwise attempts to normalize schemas into strict mode,
	// which makes non-nullable optional properties required model output.
	Strict bool `json:"strict"`
}

type codexReasoningConfig struct {
	Effort string `json:"effort,omitempty"`
}

// supportsStrictToolSchema reports whether the top-level object already meets
// Responses strict-mode requirements. Keep this deliberately narrow: schemas
// with nested containers or composition remain non-strict rather than risking
// provider rejection or changing optional fields into required ones.
func supportsStrictToolSchema(raw json.RawMessage) bool {
	var schema struct {
		Type                 string                     `json:"type"`
		Properties           map[string]json.RawMessage `json:"properties"`
		Required             []string                   `json:"required"`
		AdditionalProperties *bool                      `json:"additionalProperties"`
	}
	if json.Unmarshal(raw, &schema) != nil || schema.Type != "object" || schema.AdditionalProperties == nil || *schema.AdditionalProperties {
		return false
	}
	required := make(map[string]struct{}, len(schema.Required))
	for _, name := range schema.Required {
		required[name] = struct{}{}
	}
	if len(required) != len(schema.Properties) {
		return false
	}
	for name, rawProperty := range schema.Properties {
		if _, ok := required[name]; !ok {
			return false
		}
		var property struct {
			Type string `json:"type"`
		}
		if json.Unmarshal(rawProperty, &property) != nil {
			return false
		}
		switch property.Type {
		case "string", "number", "integer", "boolean":
		default:
			return false
		}
	}
	return true
}

type codexRequest struct {
	Model              string                   `json:"model"`
	Store              bool                     `json:"store"`
	Stream             bool                     `json:"stream,omitempty"`
	Instructions       string                   `json:"instructions,omitempty"`
	Input              []any                    `json:"input"`
	Tools              []codexTool              `json:"tools,omitempty"`
	ToolChoice         string                   `json:"tool_choice,omitempty"`
	ParallelToolCalls  bool                     `json:"parallel_tool_calls"`
	Include            []string                 `json:"include,omitempty"`
	Reasoning          *codexReasoningConfig    `json:"reasoning,omitempty"`
	PromptCacheKey     string                   `json:"prompt_cache_key,omitempty"`
	PromptCacheOptions *codexPromptCacheOptions `json:"prompt_cache_options,omitempty"`
	ServiceTier        string                   `json:"service_tier,omitempty"`
	PreviousResponseID string                   `json:"previous_response_id,omitempty"`
}

// ---- Request building ----

func (c *codexClient) findModel(id string) (Model, error) {
	id = strings.TrimSpace(id)
	c.modelOverridesMu.RLock()
	model, ok := c.modelOverrides[id]
	if ok {
		model = cloneModel(model)
	}
	c.modelOverridesMu.RUnlock()
	if ok {
		return model, nil
	}
	if c.providerName != "" && c.providerName != ProviderOpenAICodex {
		if m, err := FindModel(c.providerName, id); err == nil {
			return m, nil
		}
	}
	if m, err := FindModel(ProviderOpenAICodex, id); err == nil {
		return m, nil
	}
	m, err := FindModel(ProviderOpenAI, id)
	if err == nil {
		return m, nil
	}
	if c.providerName == ProviderOpenCodeGo {
		if api := openCodeGoAPIForModel(id); api != "" {
			return Model{
				Provider:      ProviderOpenCodeGo,
				ID:            id,
				API:           api,
				ContextWindow: 128000,
				MaxOutput:     16384,
				Reasoning:     true,
			}, nil
		}
	}
	return m, err
}

func supportsOpenAIExplicitPromptCache(model string) bool {
	id := strings.ToLower(strings.TrimSpace(model))
	return strings.HasPrefix(id, "gpt-5.6-") || id == "gpt-6-astra"
}

func (c *codexClient) buildRequest(req Request) (*codexRequest, error) {
	req.Model = strings.TrimSpace(req.Model)
	providerName := c.providerName
	if providerName == "" {
		providerName = ProviderOpenAICodex
	}
	if err := ValidateFastMode(providerName, req.FastMode); err != nil {
		return nil, err
	}
	m, err := c.findModel(req.Model)
	if err != nil {
		return nil, err
	}
	reasoning := ClampReasoningForModel(m, req.Reasoning)

	body := &codexRequest{
		Model:             req.Model,
		Store:             false,
		Stream:            true,
		Instructions:      req.SystemPrompt(),
		ParallelToolCalls: true,
		Include:           []string{"reasoning.encrypted_content"},
	}
	if req.FastMode {
		body.ServiceTier = fastModeServiceTier
	}
	if c.capabilities.ExplicitPromptCacheBreakpoint && supportsOpenAIExplicitPromptCache(req.Model) {
		// Keep implicit mode so OpenAI can retain later conversation
		// checkpoints as well as this stable instruction/tool boundary.
		body.PromptCacheOptions = &codexPromptCacheOptions{Mode: "implicit"}
		body.Input = append(body.Input, codexInputMessage{
			Role: "developer",
			Content: []any{codexInputText{
				Type:                  "input_text",
				Text:                  "[Stable prompt boundary]",
				PromptCacheBreakpoint: &codexPromptCacheBreakpoint{Mode: "explicit"},
			}},
		})
	}
	if m.Reasoning {
		effort := OpenAICodexReasoningEffort(reasoning, req.Model)
		effort = reasoningEffortForModel(m, req.Reasoning, reasoning, effort)
		if effort != "" {
			body.Reasoning = &codexReasoningConfig{Effort: effort}
		}
	}
	activeTools := activeToolDefinitions(req.Tools, req.Messages)
	if len(activeTools) > 0 {
		body.ToolChoice = "auto"
		for _, t := range activeTools {
			params := t.Schema
			if len(params) == 0 {
				params = json.RawMessage(`{"type":"object","properties":{}}`)
			}
			body.Tools = append(body.Tools, codexTool{
				Type:        "function",
				Name:        t.Name,
				Description: t.Description,
				Parameters:  params,
				Strict:      providerName == ProviderOpenAICodex && supportsStrictToolSchema(params),
			})
		}
	}

	msgIdx := 0
	req.Messages = RepairOrphanedToolResults(req.Messages)
	for _, msg := range req.Messages {
		switch msg.Role {
		case RoleDeveloper, RoleUser:
			content := []any{}
			for _, c := range msg.Content {
				switch v := c.(type) {
				case TextBlock:
					if v.Text != "" {
						content = append(content, codexInputText{Type: "input_text", Text: v.Text})
					}
				case ImageBlock:
					url := "data:" + v.MimeType + ";base64," + base64.StdEncoding.EncodeToString(v.Data)
					content = append(content, codexInputImage{Type: "input_image", Detail: "auto", ImageURL: url})
				}
			}
			if len(content) == 0 {
				continue
			}
			role := "user"
			if msg.Role == RoleDeveloper {
				role = "developer"
			}
			body.Input = append(body.Input, codexInputMessage{Role: role, Content: content})
		case RoleAssistant:
			// Emit one output_message per text block, one function_call per
			// tool call, and one reasoning item per replayable ReasoningBlock,
			// preserving the order so the model sees the same interleaving
			// we captured. The reasoning replay is what keeps Responses
			// providers from rejecting follow-up tool calls when thinking
			// is enabled. Blocks without encrypted content are omitted:
			// an ID or summary alone is not a self-contained reasoning
			// item while store is false. See reasoningForReplay.
			for _, c := range replayContent(msg, providerName, req.Model) {
				switch v := c.(type) {
				case ReasoningBlock:
					item, ok := reasoningForReplay(v)
					if !ok {
						continue
					}
					body.Input = append(body.Input, item)
				case TextBlock:
					if v.Text == "" {
						continue
					}
					msgIdx++
					out := codexOutputText{Type: "output_text", Text: v.Text, Annotations: []any{}}
					if IsRecognizedTextPhase(v.Phase) {
						out.Phase = v.Phase
					}
					body.Input = append(body.Input, codexOutputMessage{
						Type:    "message",
						Role:    "assistant",
						Status:  "completed",
						ID:      fmt.Sprintf("msg_%d", msgIdx),
						Content: []codexOutputText{out},
					})
				case ToolCallBlock:
					args := string(v.Arguments)
					if args == "" || !json.Valid([]byte(args)) {
						args = "{}"
					}
					callID, _ := splitCallID(v.ID)
					body.Input = append(body.Input, codexFunctionCall{
						Type:      "function_call",
						CallID:    callID,
						Name:      v.Name,
						Arguments: args,
					})
				}
			}
		case RoleTool:
			for _, c := range msg.Content {
				if tr, ok := c.(ToolResultBlock); ok {
					var text strings.Builder
					imageCount := 0
					for _, inner := range tr.Content {
						switch v := inner.(type) {
						case TextBlock:
							if text.Len() > 0 {
								text.WriteString("\n")
							}
							text.WriteString(v.Text)
						case ImageBlock:
							imageCount++
						}
					}
					// The Responses API function_call_output only carries a
					// string, so image bytes cannot ride along here. The agent
					// loop mirrors any tool-result images into the following
					// user message (which this client does serialize as
					// input_image). Leave a short text note so an image-only
					// result is not an empty output the API may reject, and so
					// the model knows the image arrives next.
					out := text.String()
					if out == "" && imageCount > 0 {
						if imageCount == 1 {
							out = "[image returned; see the following message]"
						} else {
							out = fmt.Sprintf("[%d images returned; see the following message]", imageCount)
						}
					}
					callID, _ := splitCallID(tr.CallID)
					body.Input = append(body.Input, codexFunctionCallOutput{
						Type:   "function_call_output",
						CallID: callID,
						Output: out,
					})
				}
			}
		}
	}

	return body, nil
}

// reasoningForReplay decides whether a persisted ReasoningBlock carries a
// self-contained Responses payload. Only blocks with encrypted content are
// replayed; an ID or a readable summary alone cannot stand in for the
// provider-issued payload when store is false. The boolean is false when
// the block must be omitted from outgoing Responses input. Stored messages
// are never mutated by this decision.
func reasoningForReplay(block ReasoningBlock) (codexReasoningItem, bool) {
	if block.Encrypted == "" {
		return codexReasoningItem{}, false
	}
	// Gateway composite IDs and other providers' signatures are not native
	// Responses items. Never rewrite an opaque ID to make it look valid.
	for _, ch := range block.ID {
		switch {
		case ch >= 'a' && ch <= 'z', ch >= 'A' && ch <= 'Z',
			ch >= '0' && ch <= '9', ch == '_', ch == '-':
			// Responses item IDs use ASCII letters, digits, underscores, or dashes.
		default:
			return codexReasoningItem{}, false
		}
	}
	item := codexReasoningItem{
		Type:             "reasoning",
		ID:               block.ID,
		EncryptedContent: block.Encrypted,
		Summary:          []codexReasoningSummary{},
	}
	if block.Summary != "" {
		item.Summary = []codexReasoningSummary{{Type: "summary_text", Text: block.Summary}}
	}
	return item, true
}

func splitCallID(id string) (string, string) {
	if i := strings.Index(id, "|"); i >= 0 {
		return id[:i], id[i+1:]
	}
	return id, ""
}

// messageItemPhase extracts the Responses message phase from either the
// item-level phase field or the first nonempty phase on its output_text
// content parts. The return is fill-only input for itemState.phase: empty
// means absent, never a reason to erase an observed value. Unrecognized
// values are preserved so storage stays faithful; IsRecognizedTextPhase
// gates interpretation and replay.
func messageItemPhase(itemPhase string, content []struct {
	Type  string `json:"type"`
	Text  string `json:"text"`
	Phase string `json:"phase"`
}) string {
	if itemPhase != "" {
		return itemPhase
	}
	for _, part := range content {
		if part.Phase != "" {
			return part.Phase
		}
	}
	return ""
}

const minOpenAIPromptCacheTokens = 1024

func responsesCacheDiagnostics(wire *codexRequest, transport, continuation string) CacheDiagnostics {
	diagnostics := CacheDiagnostics{Mode: "none", Transport: transport, Continuation: continuation}
	if wire == nil || wire.PromptCacheOptions == nil {
		return diagnostics
	}
	diagnostics.Mode = wire.PromptCacheOptions.Mode
	diagnostics.Eligible = estimatedOpenAIPromptTokens(wire) >= minOpenAIPromptCacheTokens
	return diagnostics
}

// estimatedOpenAIPromptTokens supplies a deliberately conservative local
// eligibility signal. OpenAI makes the final tokenization and cache decision;
// this estimate never leaves the process and is not persisted in diagnostics.
func estimatedOpenAIPromptTokens(wire *codexRequest) int {
	encoded, err := json.Marshal(wire)
	if err != nil {
		return 0
	}
	return len(encoded) / 4
}

func usesCodexCLIRouting(model string) bool {
	switch model {
	case "gpt-5.6-luna", "gpt-5.6-luna-pro", "gpt-5.6-terra", "gpt-5.6-terra-pro":
		return true
	default:
		return false
	}
}

// ---- Streaming ----

func (c *codexClient) Stream(ctx context.Context, req Request) (<-chan Event, error) {
	req.Model = strings.TrimSpace(req.Model)
	wire, err := c.buildRequest(req)
	if err != nil {
		return nil, err
	}
	ReportCacheDiagnostics(req.Lifecycle, responsesCacheDiagnostics(wire, "http_sse", "full"))
	if c.modelName != nil {
		wire.Model = c.modelName(wire.Model)
	}
	useCodexCLIRouting := !c.disableCLIRouting && (c.cliRoutingAll || usesCodexCLIRouting(wire.Model))
	cacheSessionID := strings.TrimSpace(req.Context.CacheSessionID)
	if c.capabilities.StablePromptCacheKey && cacheSessionID != "" {
		wire.PromptCacheKey = cacheSessionID
	}
	body, err := json.Marshal(wire)
	if err != nil {
		return nil, err
	}

	newReq := func() (*http.Request, error) {
		httpReq, err := http.NewRequestWithContext(ctx, "POST", c.baseURL, bytes.NewReader(body))
		if err != nil {
			return nil, err
		}
		httpReq.Header = c.responsesHeaders(req, useCodexCLIRouting, false)
		httpReq.Header.Set("content-type", "application/json")
		httpReq.Header.Set("accept", "text/event-stream")
		return httpReq, nil
	}

	resp, err := doStreamWithRetry(ctx, c.http, newReq, req.Lifecycle)
	if err != nil {
		label := c.providerName
		if label == "" {
			label = ProviderOpenAICodex
		}
		return nil, fmt.Errorf("%s: %w", label, err)
	}
	if resp.StatusCode != http.StatusOK {
		b, _ := io.ReadAll(resp.Body)
		resp.Body.Close()
		label := c.providerName
		if label == "" {
			label = ProviderOpenAICodex
		}
		return nil, fmt.Errorf("%s: http %d: %s", label, resp.StatusCode, strings.TrimSpace(string(b)))
	}

	out := make(chan Event, 16)
	go c.runStream(ctx, resp, req, out)
	return out, nil
}

func (c *codexClient) responsesHeaders(req Request, useCodexCLIRouting, websocketMode bool) http.Header {
	headers := make(http.Header)
	headers.Set("authorization", "Bearer "+c.token)
	if !websocketMode || c.accountID != "" {
		headers.Set("openai-beta", "responses=experimental")
	}
	if c.accountID != "" {
		headers.Set("chatgpt-account-id", c.accountID)
	}

	cacheSessionID := strings.TrimSpace(req.Context.CacheSessionID)
	threadID := strings.TrimSpace(req.Context.ThreadID)
	if useCodexCLIRouting {
		// The ChatGPT Codex backend only admits (and reliably serves)
		// requests that follow Codex CLI routing metadata; other client
		// identities are load-shed even when capacity is available.
		headers.Set("originator", "codex_cli_rs")
		if c.capabilities.SessionHeader && cacheSessionID != "" {
			headers.Set("session-id", cacheSessionID)
		}
		if c.capabilities.ThreadHeader && threadID != "" {
			headers.Set("thread-id", threadID)
		}
		if c.capabilities.ClientRequestHeader && threadID != "" {
			headers.Set("x-client-request-id", threadID)
		}
		headers.Set("user-agent", "codex_cli_rs/0.0.0")
		setOpenCodeGoHeaders(headers, c.providerName, req.Context)
		return headers
	}

	// ChatGPT's backend recognizes these established values. They are
	// external provider metadata, not zut product identifiers.
	if c.accountID != "" {
		headers.Set("originator", "zot")
		headers.Set("user-agent", fmt.Sprintf("zot (%s %s)", runtime.GOOS, runtime.GOARCH))
	}
	setOpenCodeGoHeaders(headers, c.providerName, req.Context)
	return headers
}

func (c *codexClient) runStream(ctx context.Context, resp *http.Response, req Request, out chan<- Event) {
	defer close(out)
	defer resp.Body.Close()

	raw := make(chan sseEvent, 16)
	go readSSE(resp.Body, raw)
	c.runResponseEvents(ctx, req, out, raw, nil)
}

// runResponseEvents translates the Responses streaming event protocol into
// provider events. Both HTTP/SSE and Responses WebSocket mode use the same
// event vocabulary, so keeping the state machine here prevents the transports
// from drifting in tool-call, reasoning, and usage handling.
func (c *codexClient) runResponseEvents(ctx context.Context, req Request, out chan<- Event, raw <-chan sseEvent, completed func(string)) {
	var observe func(string, Message)
	if completed != nil {
		observe = func(responseID string, _ Message) { completed(responseID) }
	}
	c.runResponseEventsWithFirst(ctx, req, out, raw, nil, observe)
}

func (c *codexClient) runResponseEventsWithFirst(ctx context.Context, req Request, out chan<- Event, raw <-chan sseEvent, first *sseEvent, completed func(string, Message)) {
	model, _ := c.findModel(req.Model)
	providerName := c.providerName
	if providerName == "" {
		providerName = ProviderOpenAICodex
	}
	out <- EventStart{Model: req.Model, Provider: providerName}

	// Accumulators. The Responses API emits output_items in order; each
	// item is either a "message" (text) or a "function_call". We track
	// the in-flight item by its index.
	type itemState struct {
		kind      string // "message" | "function_call" | "reasoning"
		callID    string
		name      string
		argsBuf   strings.Builder
		textBuf   strings.Builder
		summary   strings.Builder
		rawID     string
		encrypted string
		// phase is the Responses output phase observed for a message item.
		// It is fill-only: an absent later field must never erase an
		// observed value. Any observed value (including unrecognized ones)
		// is retained in storage; only recognized values are replayed.
		phase     string
		announced bool
	}
	var (
		items    = map[int]*itemState{}
		order    []int
		usage    Usage
		stop     = StopEnd
		finalErr error
	)

	assemble := func() Message {
		content := []Content{}
		for _, idx := range order {
			it := items[idx]
			switch it.kind {
			case "message":
				if it.textBuf.Len() > 0 {
					content = append(content, TextBlock{Text: it.textBuf.String(), Phase: it.phase})
				}
			case "function_call":
				args := it.argsBuf.String()
				if args == "" || !json.Valid([]byte(args)) {
					args = "{}"
				}
				content = append(content, ToolCallBlock{
					ID: it.callID, Name: it.name, Arguments: json.RawMessage(args),
				})
			case "reasoning":
				if it.encrypted == "" && it.summary.Len() == 0 && it.rawID == "" {
					continue
				}
				content = append(content, ReasoningBlock{
					ID:        it.rawID,
					Summary:   it.summary.String(),
					Encrypted: it.encrypted,
				})
			}
		}
		return Message{Role: RoleAssistant, Content: content, Time: time.Now()}
	}

	sendDone := func() {
		usage.CostUSD = ComputeCost(model, usage)
		out <- EventUsage{Usage: usage}
		out <- EventDone{Stop: stop, Err: finalErr, Message: assemble()}
	}

	for {
		var ev sseEvent
		if first != nil {
			ev = *first
			first = nil
		} else {
			select {
			case <-ctx.Done():
				stop = StopAborted
				finalErr = ctx.Err()
				sendDone()
				return
			case next, ok := <-raw:
				if !ok {
					sendDone()
					return
				}
				ev = next
			}
		}
		if ev.Err != nil {
			stop = StopError
			finalErr = fmt.Errorf("read SSE: %w", ev.Err)
			sendDone()
			return
		}
		var head struct {
			Type string `json:"type"`
		}
		if err := json.Unmarshal([]byte(ev.Data), &head); err != nil {
			continue
		}

		switch head.Type {
		case "response.output_item.added":
			var p struct {
				OutputIndex int `json:"output_index"`
				Item        struct {
					Type             string `json:"type"` // "message" | "function_call" | "reasoning"
					ID               string `json:"id"`
					CallID           string `json:"call_id"`
					Name             string `json:"name"`
					EncryptedContent string `json:"encrypted_content"`
					Phase            string `json:"phase"`
					Content          []struct {
						Type  string `json:"type"`
						Text  string `json:"text"`
						Phase string `json:"phase"`
					} `json:"content"`
				} `json:"item"`
			}
			_ = json.Unmarshal([]byte(ev.Data), &p)
			it := &itemState{}
			switch p.Item.Type {
			case "message":
				it.kind = "message"
				it.phase = messageItemPhase(p.Item.Phase, p.Item.Content)
			case "function_call":
				it.kind = "function_call"
				it.callID = p.Item.CallID
				it.name = p.Item.Name
				if !it.announced {
					it.announced = true
					out <- EventToolStart{ID: it.callID, Name: it.name}
				}
			case "reasoning":
				it.kind = "reasoning"
				it.rawID = p.Item.ID
				it.encrypted = p.Item.EncryptedContent
			default:
				continue
			}
			items[p.OutputIndex] = it
			order = append(order, p.OutputIndex)
		case "response.output_text.delta":
			var p struct {
				OutputIndex int    `json:"output_index"`
				Delta       string `json:"delta"`
			}
			_ = json.Unmarshal([]byte(ev.Data), &p)
			if it, ok := items[p.OutputIndex]; ok && it.kind == "message" {
				it.textBuf.WriteString(p.Delta)
				out <- EventTextDelta{Delta: p.Delta}
			}
		case "response.reasoning_summary_text.delta":
			var p struct {
				OutputIndex int    `json:"output_index"`
				Delta       string `json:"delta"`
			}
			_ = json.Unmarshal([]byte(ev.Data), &p)
			if it, ok := items[p.OutputIndex]; ok && it.kind == "reasoning" {
				it.summary.WriteString(p.Delta)
			}
		case "response.reasoning_summary_text.done":
			// summary text already accumulated via deltas
		case "response.function_call_arguments.delta":
			var p struct {
				OutputIndex int    `json:"output_index"`
				Delta       string `json:"delta"`
			}
			_ = json.Unmarshal([]byte(ev.Data), &p)
			if it, ok := items[p.OutputIndex]; ok && it.kind == "function_call" {
				it.argsBuf.WriteString(p.Delta)
				out <- EventToolArgs{ID: it.callID, Delta: p.Delta}
			}
		case "response.output_item.done":
			var p struct {
				OutputIndex int `json:"output_index"`
				Item        struct {
					Type             string `json:"type"`
					ID               string `json:"id"`
					EncryptedContent string `json:"encrypted_content"`
					Phase            string `json:"phase"`
					Content          []struct {
						Type  string `json:"type"`
						Text  string `json:"text"`
						Phase string `json:"phase"`
					} `json:"content"`
					Summary []struct {
						Type string `json:"type"`
						Text string `json:"text"`
					} `json:"summary"`
				} `json:"item"`
			}
			_ = json.Unmarshal([]byte(ev.Data), &p)
			if it, ok := items[p.OutputIndex]; ok {
				switch it.kind {
				case "function_call":
					out <- EventToolEnd{ID: it.callID}
				case "message":
					// Fill-only: an absent phase on item-done must not
					// erase a phase already observed on item-added.
					if it.phase == "" {
						it.phase = messageItemPhase(p.Item.Phase, p.Item.Content)
					}
				case "reasoning":
					// Fill-only: earlier streamed payloads and summaries win
					// over later duplicates from output_item.done.
					if it.encrypted == "" && p.Item.EncryptedContent != "" {
						it.encrypted = p.Item.EncryptedContent
					}
					if it.rawID == "" && p.Item.ID != "" {
						it.rawID = p.Item.ID
					}
					if it.summary.Len() == 0 {
						for _, s := range p.Item.Summary {
							if s.Text == "" {
								continue
							}
							if it.summary.Len() > 0 {
								it.summary.WriteString("\n")
							}
							it.summary.WriteString(s.Text)
						}
					}
				}
			}
		case "response.completed", "response.done":
			var p struct {
				Response struct {
					ID    string `json:"id"`
					Usage struct {
						InputTokens         int                       `json:"input_tokens"`
						OutputTokens        int                       `json:"output_tokens"`
						InputTokensDetails  *openAIInputTokensDetails `json:"input_tokens_details"`
						OutputTokensDetails *struct {
							ReasoningTokens int `json:"reasoning_tokens"`
						} `json:"output_tokens_details"`
					} `json:"usage"`
					Status  string `json:"status"`
					EndTurn *bool  `json:"end_turn"`
					Output  []struct {
						Type             string `json:"type"`
						ID               string `json:"id"`
						EncryptedContent string `json:"encrypted_content"`
						Phase            string `json:"phase"`
						Content          []struct {
							Type  string `json:"type"`
							Text  string `json:"text"`
							Phase string `json:"phase"`
						} `json:"content"`
						Summary []struct {
							Type string `json:"type"`
							Text string `json:"text"`
						} `json:"summary"`
					} `json:"output"`
				} `json:"response"`
			}
			_ = json.Unmarshal([]byte(ev.Data), &p)
			// Backfill encrypted reasoning supplied only in the terminal
			// response output. Match by nonempty provider-issued item ID,
			// never by output-array position. Fill a missing payload from
			// a matching terminal item without overwriting an earlier
			// payload or duplicating streamed summaries. Message phase is
			// reconciled the same fill-only way, but by output-array
			// position: terminal text content without a phase must not
			// erase a phase observed during streaming.
			if len(p.Response.Output) > 0 {
				terminal := make(map[string]struct {
					encrypted string
					summary   string
				}, len(p.Response.Output))
				terminalPhases := make([]string, 0, len(p.Response.Output))
				for _, out := range p.Response.Output {
					if out.Type == "message" {
						terminalPhases = append(terminalPhases, messageItemPhase(out.Phase, out.Content))
					}
					if out.Type != "reasoning" || out.ID == "" {
						continue
					}
					if _, seen := terminal[out.ID]; seen {
						continue
					}
					var summary strings.Builder
					for _, s := range out.Summary {
						if s.Text == "" {
							continue
						}
						if summary.Len() > 0 {
							summary.WriteString("\n")
						}
						summary.WriteString(s.Text)
					}
					terminal[out.ID] = struct {
						encrypted string
						summary   string
					}{encrypted: out.EncryptedContent, summary: summary.String()}
				}
				for _, it := range items {
					if it == nil || it.kind != "reasoning" || it.rawID == "" {
						continue
					}
					match, ok := terminal[it.rawID]
					if !ok {
						continue
					}
					if it.encrypted == "" && match.encrypted != "" {
						it.encrypted = match.encrypted
					}
					if it.summary.Len() == 0 && match.summary != "" {
						it.summary.WriteString(match.summary)
					}
				}
				// Reconcile message phases by output-array position: the
				// n-th terminal message corresponds to the n-th streamed
				// message item. Fill only; never erase streamed phases.
				pos := 0
				for _, idx := range order {
					it := items[idx]
					if it == nil || it.kind != "message" {
						continue
					}
					if pos >= len(terminalPhases) {
						break
					}
					if it.phase == "" {
						it.phase = terminalPhases[pos]
					}
					pos++
				}
			}
			usage = normalizeOpenAIUsage(p.Response.Usage.InputTokens, p.Response.Usage.OutputTokens, p.Response.Usage.InputTokensDetails)
			if details := p.Response.Usage.OutputTokensDetails; details != nil {
				usage.ReasoningTokens = details.ReasoningTokens
				usage.ReasoningTokensKnown = true
			}

			hadTool := false
			for _, it := range items {
				if it.kind == "function_call" {
					hadTool = true
					break
				}
			}
			if hadTool {
				stop = StopToolUse
			} else {
				stop = StopEnd
				if p.Response.EndTurn != nil && !*p.Response.EndTurn {
					stop = StopContinue
				}
			}
			if completed != nil && p.Response.ID != "" {
				completed(p.Response.ID, assemble())
			}
			sendDone()
			return
		case "response.failed":
			var p struct {
				Response struct {
					Error struct {
						Message string `json:"message"`
					} `json:"error"`
				} `json:"response"`
			}
			_ = json.Unmarshal([]byte(ev.Data), &p)
			stop = StopError
			finalErr = fmt.Errorf("codex: %s", p.Response.Error.Message)
			sendDone()
			return
		case "error":
			var p struct {
				Message string `json:"message"`
				Code    string `json:"code"`
				Error   struct {
					Message string `json:"message"`
					Code    string `json:"code"`
				} `json:"error"`
			}
			_ = json.Unmarshal([]byte(ev.Data), &p)
			msg := p.Message
			if msg == "" {
				msg = p.Error.Message
			}
			if msg == "" {
				msg = p.Code
			}
			if msg == "" {
				msg = p.Error.Code
			}
			if msg == "" {
				msg = strings.TrimSpace(ev.Data)
			}
			stop = StopError
			label := c.errorLabel
			if label == "" {
				label = "codex"
			}
			finalErr = fmt.Errorf("%s error: %s", label, msg)
			sendDone()
			return
		}
	}
}