package agent

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/desktop/internal/agent/drivercontract"
	"github.com/tidbcloud/AgentTeam/desktop/internal/consts"
)

const (
	codexThreadModelHydrationTimeout = 10 * time.Second
	codexThreadModelHydrationRetries = 3
	codexThreadModelHydrationBackoff = 50 * time.Millisecond
	maxPendingCodexThreadUsageEvents = 64
)

var errCodexThreadModelHydrationNotLoaded = errors.New("Codex child thread is not loaded")
var errCodexTurnDispatchBoundary = errors.New("Codex turn dispatch crossed a lifecycle boundary")

type pendingAppServerThreadTokenUsage struct {
	threadID         string
	turnID           string
	model            string
	usage            appServerThreadTokenUsage
	contextCompacted bool
}

func (d *CodexAppServerDriver) startThread(ctx context.Context, cfg drivercontract.LaunchConfig) (string, error) {
	sandboxMode, err := codexAppServerSandboxMode(cfg)
	if err != nil {
		return "", err
	}
	approvalPolicy, err := appServerApprovalPolicy(cfg)
	if err != nil {
		return "", err
	}
	params := map[string]any{
		"cwd":            cfg.WorkDir,
		"approvalPolicy": approvalPolicy,
		"sandbox":        sandboxMode,
	}
	if strings.TrimSpace(cfg.Model) != "" {
		params["model"] = cfg.Model
	}
	carrier, err := cfg.RequireCompiledPrompt()
	if err != nil {
		return "", err
	}
	baseInstructions := carrier.LaunchSurface("baseInstructions")
	developerInstructions := carrier.LaunchSurface("developerInstructions")
	if strings.TrimSpace(baseInstructions) != "" {
		params["baseInstructions"] = baseInstructions
	}
	if strings.TrimSpace(developerInstructions) != "" {
		params["developerInstructions"] = developerInstructions
	}
	dynamicTools, err := d.dynamicToolsForMCPServers(ctx)
	if err != nil {
		return "", err
	}
	if len(dynamicTools) > 0 {
		params["dynamicTools"] = dynamicTools
	}

	result, err := d.request(ctx, "thread/start", params)
	if err != nil {
		return "", err
	}

	var started appServerThreadResult
	if err := json.Unmarshal(result, &started); err != nil {
		return "", fmt.Errorf("decode thread/start result: %w", err)
	}
	threadID := strings.TrimSpace(started.Thread.ID)
	if threadID == "" {
		return "", fmt.Errorf("codex returned an empty thread id")
	}
	d.updateThreadSnapshot(started.Thread)
	d.updatePrimaryThreadModel(threadID, started.Model)
	return threadID, nil
}

func (d *CodexAppServerDriver) resumeThread(ctx context.Context, cfg drivercontract.LaunchConfig, threadID string) error {
	params, err := d.resumeThreadParams(cfg, threadID)
	if err != nil {
		return err
	}

	result, err := d.request(ctx, "thread/resume", params)
	if err != nil {
		return err
	}

	var resumed appServerThreadResult
	if err := json.Unmarshal(result, &resumed); err != nil {
		return fmt.Errorf("decode thread/resume result: %w", err)
	}
	resumedThreadID := strings.TrimSpace(resumed.Thread.ID)
	if resumedThreadID == "" {
		return fmt.Errorf("codex returned an empty resumed thread id")
	}
	// Bind the primary thread before the metadata fallback is applied. The
	// resume response intentionally excludes turns; the following full read
	// restores history when it is available.
	d.updateThreadSnapshot(resumed.Thread)
	d.updatePrimaryThreadModel(resumedThreadID, resumed.Model)
	detailed, err := d.readThreadForLifecycle(ctx, resumedThreadID)
	if err != nil {
		return fmt.Errorf("read resumed thread: %w", err)
	}
	d.applyLifecycleThreadRead(detailed)
	return nil
}

// resumeThreadParams reapplies AgentTeam's immutable launch-time policy when a
// thread is no longer resident in app-server. An explicit launch model is a
// PinCurrent-style override, so its reasoning effort must travel in the same
// request. Reasoning summary is a config override in thread/resume (unlike the
// top-level turn/start field). Empty optional instruction values are omitted:
// app-server distinguishes an absent override from Some("").
func (d *CodexAppServerDriver) resumeThreadParams(cfg drivercontract.LaunchConfig, threadID string) (map[string]any, error) {
	sandboxMode, err := codexAppServerSandboxMode(cfg)
	if err != nil {
		return nil, err
	}
	approvalPolicy, err := appServerApprovalPolicy(cfg)
	if err != nil {
		return nil, err
	}
	params := map[string]any{
		"threadId":       threadID,
		"approvalPolicy": approvalPolicy,
		"sandbox":        sandboxMode,
		// The app-server replays restored token usage when resume includes turns.
		// AgentTeam records usage as live request metrics, so take the cheap resume
		// path and hydrate turns with thread/read, which does not replay usage.
		"excludeTurns": true,
	}
	if strings.TrimSpace(cfg.WorkDir) != "" {
		params["cwd"] = cfg.WorkDir
	}
	if strings.TrimSpace(cfg.Model) != "" {
		params["model"] = cfg.Model
	}
	carrier, err := cfg.RequireCompiledPrompt()
	if err != nil {
		return nil, err
	}
	baseInstructions := carrier.LaunchSurface("baseInstructions")
	developerInstructions := carrier.LaunchSurface("developerInstructions")
	if strings.TrimSpace(baseInstructions) != "" {
		params["baseInstructions"] = baseInstructions
	}
	if strings.TrimSpace(developerInstructions) != "" {
		params["developerInstructions"] = developerInstructions
	}

	config := make(map[string]any, 2)
	if strings.TrimSpace(cfg.Model) != "" {
		if effort := strings.TrimSpace(cfg.ReasoningEffort); effort != "" {
			config["model_reasoning_effort"] = effort
		}
	}
	d.mu.Lock()
	summary := strings.TrimSpace(d.reasoningSummary)
	d.mu.Unlock()
	if summary != "" {
		config["model_reasoning_summary"] = summary
	}
	if len(config) > 0 {
		params["config"] = config
	}
	return params, nil
}

func (d *CodexAppServerDriver) recordSessionResumeFallback(reason, operation string) {
	d.mu.Lock()
	defer d.mu.Unlock()
	if d.sessionResumeFallback != nil {
		return
	}
	d.sessionResumeFallback = newSessionResumeFallback(reason, operation)
}

func (d *CodexAppServerDriver) takeSessionResumeFallback() *drivercontract.SessionResumeFallback {
	d.mu.Lock()
	defer d.mu.Unlock()
	fallback := d.sessionResumeFallback
	d.sessionResumeFallback = nil
	return fallback
}

const codexSessionInvalidRPCCode = -32000

func isAuthoritativeCodexSessionInvalidError(err error) bool {
	var requestErr *appServerRequestError
	if err == nil || !errors.As(err, &requestErr) {
		return false
	}
	switch strings.TrimSpace(requestErr.Method) {
	case "thread/resume", "turn/start", "turn/steer":
	default:
		return false
	}
	// The app-server's typed provider/session rejection boundary is the custom
	// server-error code. Transport ambiguity and JSON-RPC request/parameter
	// errors may carry the same words in data, but cannot prove that the saved
	// thread was rejected before a fresh request can safely be sent.
	if requestErr.Code != codexSessionInvalidRPCCode {
		return false
	}
	if codexSessionInvalidData(requestErr.Data) {
		return true
	}
	switch normalizeCodexSessionErrorToken(requestErr.Message) {
	case "thread not found", "thread does not exist", "invalid thread", "invalid thread id", "unknown thread":
		return true
	default:
		return false
	}
}

func codexSessionInvalidData(data json.RawMessage) bool {
	if len(data) == 0 || strings.TrimSpace(string(data)) == "null" {
		return false
	}
	// Only direct structured discriminator fields are authority. In particular,
	// provider/model diagnostics often use reason/status/error/message fields and
	// must never become a session cutover merely because they contain "invalid"
	// or "not found".
	var decoded struct {
		Code string `json:"code"`
		Kind string `json:"kind"`
		Type string `json:"type"`
	}
	if json.Unmarshal(data, &decoded) != nil {
		return false
	}
	for _, value := range []string{decoded.Code, decoded.Kind, decoded.Type} {
		switch normalizeCodexSessionErrorToken(value) {
		case "thread not found", "invalid thread id":
			return true
		}
	}
	return false
}

func normalizeCodexSessionErrorToken(value string) string {
	value = strings.ToLower(strings.TrimSpace(value))
	value = strings.NewReplacer("_", " ", "-", " ").Replace(value)
	return strings.Join(strings.Fields(value), " ")
}

func (d *CodexAppServerDriver) prepareInitialThread(ctx context.Context, cfg drivercontract.LaunchConfig) (threadID, method string, params map[string]any, resumed bool, err error) {
	threadID = strings.TrimSpace(cfg.SessionID)
	if threadID == "" {
		newThreadID, err := d.startThread(ctx, cfg)
		if err != nil {
			return "", "", nil, false, err
		}
		carrier, carrierErr := cfg.RequireCompiledPrompt()
		if carrierErr != nil {
			return "", "", nil, false, carrierErr
		}
		params, err := d.startTurnParams(cfg, carrier.LaunchSurface("freshPrompt"), newThreadID)
		if err != nil {
			return "", "", nil, false, err
		}
		addInitialDeliveryCorrelation(params, cfg.InitialDeliveryMessageID)
		if params == nil {
			return newThreadID, "", nil, false, nil
		}
		return newThreadID, "turn/start", params, false, nil
	}

	// A resumed app-server thread does not execute thread/start, so refresh the
	// trusted inventory explicitly. The child effect gate must not lose known
	// read-only annotations merely because this daemon resumed durable history.
	if _, err := d.dynamicToolsForMCPServers(ctx); err != nil {
		return "", "", nil, false, err
	}
	if err := d.resumeThread(ctx, cfg, threadID); err != nil {
		if !isAuthoritativeCodexSessionInvalidError(err) {
			return "", "", nil, false, fmt.Errorf("codex provider session resume: %w", err)
		}
		d.recordSessionResumeFallback("invalid_or_not_found_session", "thread/resume")
		return d.startFreshThreadForInitialTurn(ctx, cfg)
	}
	carrier, carrierErr := cfg.RequireCompiledPrompt()
	if carrierErr != nil {
		return "", "", nil, false, carrierErr
	}
	resumePrompt := carrier.LaunchSurface("resumePrompt")
	if strings.TrimSpace(resumePrompt) == "" {
		return threadID, "", nil, true, nil
	}

	method, params, err = d.resolveTurnRequestForDirectDelivery(ctx, resumePrompt, threadID)
	if err == nil {
		addInitialDeliveryCorrelation(params, cfg.InitialDeliveryMessageID)
		return threadID, method, params, true, nil
	}
	// A daemon restart can leave the prior thread resumable by id but unable to
	// accept the first new wake. Only an authoritative invalid/not-found result
	// permits a single fresh-thread cutover; transport and lifecycle ambiguity
	// remain fail-closed so an input cannot be replayed blindly.
	if !isAuthoritativeCodexSessionInvalidError(err) {
		return "", "", nil, false, fmt.Errorf("codex provider session initial delivery: %w", err)
	}
	d.recordSessionResumeFallback("invalid_or_not_found_session", "turn/start_or_steer")
	return d.startFreshThreadForInitialTurn(ctx, cfg)
}

func (d *CodexAppServerDriver) startFreshThreadForInitialTurn(ctx context.Context, cfg drivercontract.LaunchConfig) (string, string, map[string]any, bool, error) {
	threadID, err := d.startThread(ctx, cfg)
	if err != nil {
		return "", "", nil, false, err
	}
	carrier, carrierErr := cfg.RequireCompiledPrompt()
	if carrierErr != nil {
		return "", "", nil, false, carrierErr
	}
	params, err := d.startTurnParams(cfg, carrier.LaunchSurface("freshPrompt"), threadID)
	if err != nil {
		return "", "", nil, false, err
	}
	addInitialDeliveryCorrelation(params, cfg.InitialDeliveryMessageID)
	if params == nil {
		return threadID, "", nil, false, nil
	}
	return threadID, "turn/start", params, false, nil
}

// addInitialDeliveryCorrelation attaches the stable AgentTeam message id only to a
// Spawn-created initial request. startTurnParams is also used by later live
// turns, so teaching it to read LaunchConfig.InitialDeliveryMessageID directly
// would incorrectly reuse the first message id for unrelated future input.
func addInitialDeliveryCorrelation(params map[string]any, messageID string) {
	messageID = strings.TrimSpace(messageID)
	if params == nil || messageID == "" {
		return
	}
	params["clientUserMessageId"] = messageID
}

func (d *CodexAppServerDriver) readThread(ctx context.Context, threadID string, includeTurns bool) (appServerThreadResult, error) {
	result, err := d.request(ctx, "thread/read", map[string]any{
		"threadId":     threadID,
		"includeTurns": includeTurns,
	})
	if err != nil {
		return appServerThreadResult{}, err
	}

	var readResult appServerThreadResult
	if err := json.Unmarshal(result, &readResult); err != nil {
		return appServerThreadResult{}, fmt.Errorf("decode thread/read result: %w", err)
	}
	readThreadID := strings.TrimSpace(readResult.Thread.ID)
	if readThreadID == "" {
		readThreadID = strings.TrimSpace(threadID)
	}
	d.updateThreadModel(readThreadID, readResult.Model)
	return readResult, nil
}

type codexThreadTurnsState uint8

const (
	codexThreadTurnsAvailable codexThreadTurnsState = iota
	// Codex intentionally has no persisted rollout before the first user
	// message is materialized. Lifecycle metadata remains readable, but an
	// asynchronously accepted first submission may still be crossing that
	// boundary, so reconciliation must not infer authoritative absence.
	codexThreadTurnsBeforeFirstUserMessage
	// Ephemeral threads never expose turns. Metadata is usable for lifecycle
	// routing, but it cannot prove correlated history absence.
	codexThreadTurnsUnavailable
)

type codexLifecycleThreadRead struct {
	Result     appServerThreadResult
	TurnsState codexThreadTurnsState
}

// readThreadForLifecycle mirrors Codex TUI's bounded includeTurns fallback.
// Callers that need a lifecycle decision first request complete history; only
// the two app-server errors that explicitly say turns cannot be returned fall
// back to one metadata-only read. Transport, decode, not-loaded, and all other
// JSON-RPC errors fail closed.
func (d *CodexAppServerDriver) readThreadForLifecycle(ctx context.Context, threadID string) (codexLifecycleThreadRead, error) {
	threadID = strings.TrimSpace(threadID)
	if threadID == "" {
		return codexLifecycleThreadRead{}, fmt.Errorf("codex lifecycle read requires a thread id")
	}

	result, err := d.readThread(ctx, threadID, true)
	if err == nil {
		if err := validateCodexThreadReadResult(threadID, result); err != nil {
			return codexLifecycleThreadRead{}, err
		}
		return codexLifecycleThreadRead{Result: result, TurnsState: codexThreadTurnsAvailable}, nil
	}

	turnsState, ok := codexIncludeTurnsFallbackState(err)
	if !ok {
		return codexLifecycleThreadRead{}, err
	}
	metadata, metadataErr := d.readThread(ctx, threadID, false)
	if metadataErr != nil {
		return codexLifecycleThreadRead{}, fmt.Errorf("read codex thread metadata after includeTurns fallback: %w", metadataErr)
	}
	if err := validateCodexThreadReadResult(threadID, metadata); err != nil {
		return codexLifecycleThreadRead{}, err
	}
	return codexLifecycleThreadRead{Result: metadata, TurnsState: turnsState}, nil
}

func codexIncludeTurnsFallbackState(err error) (codexThreadTurnsState, bool) {
	var requestErr *appServerRequestError
	if !errors.As(err, &requestErr) || strings.TrimSpace(requestErr.Method) != "thread/read" {
		return codexThreadTurnsAvailable, false
	}
	message := strings.TrimSpace(requestErr.Message)
	switch {
	case strings.Contains(message, "includeTurns is unavailable before first user message"):
		return codexThreadTurnsBeforeFirstUserMessage, true
	case strings.Contains(message, "ephemeral threads do not support includeTurns"):
		return codexThreadTurnsUnavailable, true
	default:
		return codexThreadTurnsAvailable, false
	}
}

func validateCodexThreadReadResult(threadID string, result appServerThreadResult) error {
	expected := strings.TrimSpace(threadID)
	actual := strings.TrimSpace(result.Thread.ID)
	if actual != expected {
		return fmt.Errorf("codex thread/read for %s returned thread %q", expected, actual)
	}
	return nil
}

func (d *CodexAppServerDriver) applyLifecycleThreadRead(read codexLifecycleThreadRead) {
	if read.TurnsState == codexThreadTurnsAvailable {
		d.updateThreadSnapshot(read.Result.Thread)
		return
	}
	// A metadata-only response carries no turn list. Apply it through the same
	// notification fence used for asynchronous status changes: a late idle
	// response cannot erase a concrete active turn, and a response for a primary
	// thread that has since been replaced cannot bind the old thread again.
	d.updateThreadStatus(read.Result.Thread.ID, read.Result.Thread.Status)
}

func (d *CodexAppServerDriver) compactThread(ctx context.Context, threadID string) error {
	if strings.TrimSpace(threadID) == "" {
		return nil
	}
	params := map[string]any{
		"threadId": threadID,
	}
	_, err := d.request(ctx, "thread/compact/start", params)
	return err
}

func (d *CodexAppServerDriver) resumeThreadForRecovery(ctx context.Context, cfg drivercontract.LaunchConfig, threadID string) (appServerThreadResult, error) {
	params, err := d.resumeThreadParams(cfg, threadID)
	if err != nil {
		return appServerThreadResult{}, err
	}
	result, err := d.request(ctx, "thread/resume", params)
	if err != nil {
		return appServerThreadResult{}, err
	}

	var resumed appServerThreadResult
	if err := json.Unmarshal(result, &resumed); err != nil {
		return appServerThreadResult{}, fmt.Errorf("decode thread/resume result: %w", err)
	}
	resumedThreadID := strings.TrimSpace(resumed.Thread.ID)
	if resumedThreadID == "" {
		return appServerThreadResult{}, fmt.Errorf("codex returned an empty resumed thread id")
	}
	d.updateThreadSnapshot(resumed.Thread)
	d.updatePrimaryThreadModel(resumedThreadID, resumed.Model)
	detailed, err := d.readThreadForLifecycle(ctx, resumedThreadID)
	if err != nil {
		return appServerThreadResult{}, fmt.Errorf("read resumed recovery thread: %w", err)
	}
	if detailed.TurnsState == codexThreadTurnsUnavailable {
		return appServerThreadResult{}, fmt.Errorf("read resumed recovery thread: codex thread %s does not expose turn history", resumedThreadID)
	}
	resumed.Thread = detailed.Result.Thread
	d.applyLifecycleThreadRead(detailed)
	return resumed, nil
}

func (d *CodexAppServerDriver) interruptTurn(ctx context.Context, threadID string, turnID string) error {
	if strings.TrimSpace(threadID) == "" || strings.TrimSpace(turnID) == "" {
		return nil
	}
	_, err := d.request(ctx, "turn/interrupt", map[string]any{
		"threadId": threadID,
		"turnId":   turnID,
	})
	return err
}

func (d *CodexAppServerDriver) interruptStartup(ctx context.Context, threadID string) error {
	if strings.TrimSpace(threadID) == "" {
		return nil
	}
	_, err := d.request(ctx, "turn/interrupt", map[string]any{
		"threadId": threadID,
		"turnId":   "",
	})
	return err
}

func (d *CodexAppServerDriver) rollbackThread(ctx context.Context, threadID string, numTurns int) error {
	result, err := d.request(ctx, "thread/rollback", map[string]any{
		"threadId": threadID,
		"numTurns": numTurns,
	})
	if err != nil {
		return err
	}

	var rolledBack appServerThreadResult
	if err := json.Unmarshal(result, &rolledBack); err != nil {
		return fmt.Errorf("decode thread/rollback result: %w", err)
	}
	d.updateThreadSnapshot(rolledBack.Thread)
	return nil
}

// ---------------------------------------------------------------------------
// Turn steering & management
// ---------------------------------------------------------------------------

func (d *CodexAppServerDriver) turnRequestForText(text string, sessionID string) (string, map[string]any, error) {
	cfg := d.currentLaunchConfig()
	params, err := d.startTurnParams(cfg, text, sessionID)
	if err != nil {
		return "", nil, err
	}
	if params == nil {
		return "", nil, nil
	}

	d.mu.Lock()
	defer d.mu.Unlock()
	if d.canSteerLocked(sessionID) {
		return "turn/steer", map[string]any{
			"threadId":       sessionID,
			"expectedTurnId": d.activeTurnID,
			"input":          params["input"],
		}, nil
	}
	if d.threadStatus.Type == "active" {
		return "", nil, nil
	}
	return "turn/start", params, nil
}

func (d *CodexAppServerDriver) startTurnParams(cfg drivercontract.LaunchConfig, text string, threadID string) (map[string]any, error) {
	if strings.TrimSpace(threadID) == "" {
		return nil, nil
	}
	if strings.TrimSpace(text) == "" {
		return nil, nil
	}
	sandboxPolicy, err := codexAppServerSandboxPolicy(cfg)
	if err != nil {
		return nil, err
	}

	// Codex routed-message delivery is ordinary turn input: turn/start when the
	// thread is idle, or turn/steer when an active regular turn is steerable.
	params := map[string]any{
		"threadId": threadID,
		"input": []map[string]any{
			{
				"type":          "text",
				"text":          text,
				"text_elements": []map[string]any{},
			},
		},
		"sandboxPolicy": sandboxPolicy,
	}

	d.mu.Lock()
	effort := d.reasoningEffort
	summary := d.reasoningSummary
	d.mu.Unlock()
	if effort != "" {
		params["effort"] = effort
	}
	if summary != "" {
		params["summary"] = summary
	}
	return params, nil
}

func (d *CodexAppServerDriver) sendTurnRequest(ctx context.Context, threadID string, method string, params map[string]any) error {
	// Revalidate the selected action at the physical dispatch boundary and keep
	// primary lifecycle notifications from changing that state until the frame
	// has been written. requestWithWriteBoundary releases this fence before it
	// waits for the response, because Codex may send turn/started first.
	d.primaryTurnDispatchMu.Lock()
	if err := d.validatePrimaryTurnDispatch(threadID, method, params); err != nil {
		d.primaryTurnDispatchMu.Unlock()
		return err
	}
	result, err := d.requestWithWriteBoundary(ctx, method, params, d.primaryTurnDispatchMu.Unlock)
	if err != nil {
		return err
	}

	switch method {
	case "turn/start":
		var started appServerTurnResult
		if err := json.Unmarshal(result, &started); err != nil {
			return codexAppServerAmbiguousOutcomeError(method, "decode accepted turn/start result", err)
		}
		if strings.TrimSpace(started.Turn.ID) == "" {
			return codexAppServerAmbiguousOutcomeError(method, "decode accepted turn/start result", errors.New("response is missing turn id"))
		}
		d.acknowledgeTurnStarted(threadID, started.Turn)
	case "turn/steer":
		var steered appServerTurnIDResult
		if err := json.Unmarshal(result, &steered); err != nil {
			return codexAppServerAmbiguousOutcomeError(method, "decode accepted turn/steer result", err)
		}
		if strings.TrimSpace(steered.TurnID) == "" {
			return codexAppServerAmbiguousOutcomeError(method, "decode accepted turn/steer result", errors.New("response is missing turn id"))
		}
		d.acknowledgeTurnStarted(threadID, appServerTurn{
			ID:     steered.TurnID,
			Status: appServerTurnStatus{Type: "inProgress"},
		})
	default:
		return fmt.Errorf("unsupported turn request method %q", method)
	}

	return nil
}

func (d *CodexAppServerDriver) validatePrimaryTurnDispatch(threadID string, method string, params map[string]any) error {
	threadID = strings.TrimSpace(threadID)
	d.mu.Lock()
	defer d.mu.Unlock()

	primaryThreadID := strings.TrimSpace(d.threadID)
	// Startup/unit embedders may dispatch before the first thread snapshot is
	// installed. Once a primary thread is bound, however, every action must be
	// valid for that exact cached lifecycle state.
	if primaryThreadID == "" {
		return nil
	}
	if threadID == "" || primaryThreadID != threadID {
		return fmt.Errorf("%w: selected %s for thread %q while primary thread is %q", errCodexTurnDispatchBoundary, method, threadID, primaryThreadID)
	}

	statusType := normalizeThreadStatus(d.threadStatus).Type
	switch method {
	case "turn/start":
		if (statusType == "idle" || statusType == "systemError") && !d.compactionInProgress {
			return nil
		}
		return fmt.Errorf("%w: turn/start selected for thread %s in %q state", errCodexTurnDispatchBoundary, threadID, statusType)

	case "turn/steer":
		expectedTurnID, _ := params["expectedTurnId"].(string)
		expectedTurnID = strings.TrimSpace(expectedTurnID)
		activeTurnID := strings.TrimSpace(d.activeTurnID)
		if statusType == "active" && expectedTurnID != "" && activeTurnID == expectedTurnID && strings.TrimSpace(d.activeTurnNonSteerableKind) == "" && !d.compactionInProgress {
			return nil
		}
		return fmt.Errorf("%w: turn/steer selected for thread %s expected %q while cached state=%q activeTurn=%q", errCodexTurnDispatchBoundary, threadID, expectedTurnID, statusType, activeTurnID)

	default:
		return fmt.Errorf("unsupported turn request method %q", method)
	}
}

// ---------------------------------------------------------------------------
// ParseLine / EncodeStdinMessage (drivercontract.Driver interface)
// ---------------------------------------------------------------------------

// ParseLine parses one translated JSON event line into normalized events.
func (d *CodexAppServerDriver) ParseLine(line string) []drivercontract.ParsedEvent {
	if !strings.HasPrefix(strings.TrimSpace(line), "{") {
		return nil
	}

	var event translatedEvent
	if err := json.Unmarshal([]byte(line), &event); err != nil {
		return nil
	}

	pe := drivercontract.ParsedEvent(event)
	if pe.Kind == "" {
		return nil
	}
	return []drivercontract.ParsedEvent{pe}
}

// EncodeStdinMessage encodes a user message for codex-app-server stdin.
func (d *CodexAppServerDriver) EncodeStdinMessage(text, sessionID string) []byte {
	method, params, err := d.turnRequestForText(text, sessionID)
	if err != nil {
		return nil
	}
	if method == "" {
		if d.threadKnownActive(sessionID) {
			return nil
		}
		method = "turn/start"
		params, err = d.startTurnParams(d.currentLaunchConfig(), text, sessionID)
		if err != nil {
			return nil
		}
	}
	if method == "" || params == nil {
		return nil
	}
	d.mu.Lock()
	requestID := d.nextRequestIDLocked()
	d.mu.Unlock()

	request := map[string]any{
		"jsonrpc": "2.0",
		"id":      requestID,
		"method":  method,
		"params":  params,
	}
	data, err := json.Marshal(request)
	if err != nil {
		return nil
	}
	return append(data, '\n')
}

// ---------------------------------------------------------------------------
// JSON-RPC request / notify
// ---------------------------------------------------------------------------

func (d *CodexAppServerDriver) updateThreadSnapshot(thread appServerThread) {
	completedTurnIDs := make(map[string]struct{}, len(thread.Turns))
	completedTurnCount := 0
	activeTurnID := ""
	for _, turn := range thread.Turns {
		trimmedTurnID := strings.TrimSpace(turn.ID)
		if trimmedTurnID == "" {
			continue
		}
		if strings.TrimSpace(turn.Status.Type) == "inProgress" {
			activeTurnID = trimmedTurnID
			continue
		}
		completedTurnIDs[trimmedTurnID] = struct{}{}
		completedTurnCount++
	}

	threadStatus := normalizeThreadStatus(thread.Status)
	if threadStatus.Type == "" {
		if activeTurnID != "" {
			threadStatus = appServerThreadStatus{Type: "active"}
		} else {
			threadStatus = appServerThreadStatus{Type: "idle"}
		}
	}

	d.mu.Lock()
	defer d.mu.Unlock()
	prevThreadID := strings.TrimSpace(d.threadID)
	prevActiveTurnID := strings.TrimSpace(d.activeTurnID)
	previousCompactionTurnID := strings.TrimSpace(d.compactionTurnID)
	previousContextCompactionTurnID := strings.TrimSpace(d.contextCompactionItemFence.turnID)
	d.threadID = strings.TrimSpace(thread.ID)
	preservesKnownCompaction := d.compactionInProgress &&
		prevThreadID != "" &&
		prevThreadID == d.threadID &&
		previousCompactionTurnID != "" &&
		threadStatus.Type == "active" &&
		activeTurnID == previousCompactionTurnID
	preservesContextCompactionItem := prevThreadID != "" &&
		prevThreadID == d.threadID &&
		previousContextCompactionTurnID != "" &&
		threadStatus.Type == "active" &&
		activeTurnID == previousContextCompactionTurnID
	if prevThreadID != "" && prevThreadID != d.threadID {
		d.compactQueued = false
		d.queuedCompactionCauseTurnID = ""
		d.clearCompactionMutationLocked()
		d.contextCompactionItemFence = codexContextCompactionItemFence{}
		d.contextCompactionBoundary = codexContextCompactionBoundary{}
		d.lastContextCompactionCompletion = ""
		d.lastContextCompactionCompletedAt = time.Time{}
		d.terminalErrorTurnEndIDs = make(map[string]struct{})
	}
	if d.lastTokenUsageByThread == nil {
		d.lastTokenUsageByThread = make(map[string]appServerThreadTokenUsage)
	}
	d.lastTokenUsage = d.lastTokenUsageByThread[d.threadID]
	d.threadStatus = threadStatus
	d.activeTurnID = activeTurnID
	if threadStatus.Type != "active" || activeTurnID == "" || prevThreadID != d.threadID || prevActiveTurnID != activeTurnID {
		d.activeTurnNonSteerableKind = ""
	}
	// thread/read is also used for delivery/recovery reconciliation while a
	// turn is running. If it reports the same exact in-progress turn already
	// classified as compaction, the snapshot confirms rather than replaces that
	// classification. Clearing it here would make the compaction steerable and
	// discard an already latched pre-completion idle status. A different primary
	// thread, terminal snapshot, or different concrete active turn is an actual
	// lifecycle replacement and still clears the compaction state.
	if prevThreadID != d.threadID || threadStatus.Type != "active" || (activeTurnID != "" && !preservesKnownCompaction) {
		d.clearCompactionMutationLocked()
		if !preservesContextCompactionItem {
			d.contextCompactionItemFence = codexContextCompactionItemFence{}
		}
	}
	d.completedTurnIDs = completedTurnIDs
	d.completedTurnCount = completedTurnCount
}

func (d *CodexAppServerDriver) completedTurnsFromThread(thread appServerThread) int {
	completedTurnCount := 0
	for _, turn := range thread.Turns {
		if strings.TrimSpace(turn.ID) == "" {
			continue
		}
		if strings.TrimSpace(turn.Status.Type) == "inProgress" {
			continue
		}
		completedTurnCount++
	}
	return completedTurnCount
}

func (d *CodexAppServerDriver) updateThreadStatus(threadID string, status appServerThreadStatus) bool {
	trimmedThreadID := strings.TrimSpace(threadID)
	normalized := normalizeThreadStatus(status)

	d.mu.Lock()
	defer d.mu.Unlock()
	if !d.isPrimaryThreadEventLocked(trimmedThreadID) {
		return false
	}
	compactionCompleted := false
	// A status notification does not identify the turn it describes. If a
	// concrete primary turn is still active, a late idle/not-loaded update from
	// the preceding turn must not clear it. Codex is different for compaction:
	// app-server synchronously publishes idle before it publishes the matching
	// turn/completed. Latch that pre-terminal status only after turn/started has
	// bound the concrete compaction id, then let the matching completion consume
	// it atomically.
	if normalized.Type != "active" && strings.TrimSpace(d.activeTurnID) != "" {
		activeTurnID := strings.TrimSpace(d.activeTurnID)
		if normalized.Type != "" &&
			d.compactionInProgress &&
			strings.TrimSpace(d.compactionTurnID) == activeTurnID {
			d.compactionTerminalStatus = normalized
		}
		return false
	}
	if d.compactionInProgress && normalized.Type != "active" {
		compactionCompleted = normalized.Type != ""
		d.clearCompactionMutationLocked()
		d.contextCompactionItemFence = codexContextCompactionItemFence{}
	}
	d.threadStatus = normalized
	if normalized.Type != "active" {
		d.activeTurnID = ""
		d.activeTurnNonSteerableKind = ""
		if normalized.Type != "idle" {
			d.compactionTurnID = ""
		}
	}
	return compactionCompleted
}

func (d *CodexAppServerDriver) updateThreadTokenUsage(
	threadID string,
	turnID string,
	usage appServerThreadTokenUsage,
) {
	d.tokenUsageMu.Lock()
	defer d.tokenUsageMu.Unlock()

	eventThreadID := strings.TrimSpace(threadID)
	trimmedThreadID := eventThreadID

	d.mu.Lock()
	defer d.mu.Unlock()
	if trimmedThreadID == "" {
		trimmedThreadID = strings.TrimSpace(d.threadID)
	}
	if trimmedThreadID == "" {
		return
	}
	d.observeContextCompactionUsageLocked(eventThreadID, strings.TrimSpace(turnID), usage)
	if d.lastTokenUsageByThread == nil {
		d.lastTokenUsageByThread = make(map[string]appServerThreadTokenUsage)
	}
	d.lastTokenUsageByThread[trimmedThreadID] = usage
	if strings.TrimSpace(d.threadID) == trimmedThreadID {
		d.lastTokenUsage = usage
	}
}

func (d *CodexAppServerDriver) updateEffectiveModel(model string) {
	d.updatePrimaryThreadModel("", model)
}

func (d *CodexAppServerDriver) updatePrimaryThreadModel(threadID string, model string) {
	model = strings.TrimSpace(model)
	if model == "" {
		return
	}
	threadID = strings.TrimSpace(threadID)
	d.tokenUsageMu.Lock()
	defer d.tokenUsageMu.Unlock()
	d.mu.Lock()
	d.effectiveModel = model
	if threadID == "" {
		threadID = strings.TrimSpace(d.threadID)
	}
	if threadID != "" {
		if d.modelByThread == nil {
			d.modelByThread = make(map[string]string)
		}
		d.modelByThread[threadID] = model
	}
	emissions := d.collectPendingThreadUsageLocked(threadID)
	d.mu.Unlock()
	d.enqueueThreadUsageEmissions(emissions)
}

func (d *CodexAppServerDriver) updateThreadModel(threadID string, model string) {
	d.updateThreadModelWithPendingPolicy(threadID, model, true)
}

func (d *CodexAppServerDriver) updateThreadModelFromSettings(threadID string, model string) {
	d.updateThreadModelWithPendingPolicy(threadID, model, false)
}

func (d *CodexAppServerDriver) updateThreadModelWithPendingPolicy(threadID string, model string, flushWhileHydrating bool) {
	threadID = strings.TrimSpace(threadID)
	model = strings.TrimSpace(model)
	if threadID == "" || model == "" {
		return
	}
	d.tokenUsageMu.Lock()
	defer d.tokenUsageMu.Unlock()
	d.mu.Lock()
	if d.modelByThread == nil {
		d.modelByThread = make(map[string]string)
	}
	d.modelByThread[threadID] = model
	if strings.TrimSpace(d.threadID) == threadID {
		d.effectiveModel = model
	}
	var emissions []translatedEvent
	if _, hydrating := d.modelHydrationInFlight[threadID]; flushWhileHydrating || !hydrating {
		emissions = d.collectPendingThreadUsageLocked(threadID)
	}
	d.mu.Unlock()
	d.enqueueThreadUsageEmissions(emissions)
}

func (d *CodexAppServerDriver) threadModel(threadID string) string {
	d.mu.Lock()
	defer d.mu.Unlock()
	return d.modelForThreadLocked(threadID)
}

func (d *CodexAppServerDriver) updateThreadTurnModel(threadID string, turnID string, model string) {
	key := threadTurnModelKey(threadID, turnID)
	model = strings.TrimSpace(model)
	if key == "" || model == "" {
		return
	}
	d.tokenUsageMu.Lock()
	defer d.tokenUsageMu.Unlock()
	d.mu.Lock()
	if d.modelByThreadTurn == nil {
		d.modelByThreadTurn = make(map[string]string)
	}
	d.modelByThreadTurn[key] = model
	var emissions []translatedEvent
	trimmedThreadID := strings.TrimSpace(threadID)
	if _, hydrating := d.modelHydrationInFlight[trimmedThreadID]; !hydrating {
		emissions = d.collectPendingThreadUsageLocked(trimmedThreadID)
	}
	d.mu.Unlock()
	d.enqueueThreadUsageEmissions(emissions)
}

func (d *CodexAppServerDriver) clearThreadTurnModel(threadID string, turnID string) {
	key := threadTurnModelKey(threadID, turnID)
	if key == "" {
		return
	}
	d.mu.Lock()
	defer d.mu.Unlock()
	delete(d.modelByThreadTurn, key)
}

func threadTurnModelKey(threadID string, turnID string) string {
	threadID = strings.TrimSpace(threadID)
	turnID = strings.TrimSpace(turnID)
	if threadID == "" || turnID == "" {
		return ""
	}
	return threadID + "\x00" + turnID
}

func (d *CodexAppServerDriver) recordThreadTokenUsage(
	threadID string,
	turnID string,
	usage appServerThreadTokenUsage,
) (string, bool, bool, *codexContextCompactionBoundary) {
	eventThreadID := strings.TrimSpace(threadID)
	trimmedThreadID := eventThreadID
	trimmedTurnID := strings.TrimSpace(turnID)

	d.mu.Lock()
	defer d.mu.Unlock()
	if trimmedThreadID == "" {
		trimmedThreadID = strings.TrimSpace(d.threadID)
	}
	if trimmedThreadID == "" {
		return d.effectiveModel, false, false, nil
	}
	d.observeContextCompactionUsageLocked(eventThreadID, trimmedTurnID, usage)
	model := d.modelForThreadTurnLocked(trimmedThreadID, trimmedTurnID)
	if model == "" {
		if _, blocked := d.modelHydrationBlocked[trimmedThreadID]; blocked {
			return "", false, false, nil
		}
	}
	contextCompacted, postCompactionBoundary := d.consumeContextCompactionBoundaryLocked(eventThreadID, usage)
	pending := d.pendingTokenUsageByThread[trimmedThreadID]
	_, hydrating := d.modelHydrationInFlight[trimmedThreadID]
	if d.lastTokenUsageByThread == nil {
		d.lastTokenUsageByThread = make(map[string]appServerThreadTokenUsage)
	}
	lastUsage := d.lastTokenUsageByThread[trimmedThreadID]
	if sameAppServerThreadTokenUsage(lastUsage, usage) && !contextCompacted {
		d.lastTokenUsageByThread[trimmedThreadID] = usage
		if strings.TrimSpace(d.threadID) == trimmedThreadID {
			d.lastTokenUsage = usage
		}
		return model, false, false, nil
	}
	d.lastTokenUsageByThread[trimmedThreadID] = usage
	if strings.TrimSpace(d.threadID) == trimmedThreadID {
		d.lastTokenUsage = usage
	}
	if model == "" || (hydrating && len(pending) > 0) {
		if d.pendingTokenUsageByThread == nil {
			d.pendingTokenUsageByThread = make(map[string][]pendingAppServerThreadTokenUsage)
		}
		next := pendingAppServerThreadTokenUsage{
			threadID:         trimmedThreadID,
			turnID:           trimmedTurnID,
			model:            model,
			usage:            usage,
			contextCompacted: contextCompacted,
		}
		if len(pending) >= maxPendingCodexThreadUsageEvents {
			copy(pending, pending[len(pending)-maxPendingCodexThreadUsageEvents+1:])
			pending = pending[:maxPendingCodexThreadUsageEvents-1]
		}
		d.pendingTokenUsageByThread[trimmedThreadID] = append(pending, next)
		return "", false, contextCompacted, postCompactionBoundary
	}
	return model, true, contextCompacted, postCompactionBoundary
}

func (d *CodexAppServerDriver) recordAndEnqueueThreadTokenUsage(threadID string, turnID string, usage appServerThreadTokenUsage) bool {
	d.tokenUsageMu.Lock()
	model, emit, contextCompacted, postCompactionBoundary :=
		d.recordThreadTokenUsage(threadID, turnID, usage)
	if emit {
		d.enqueueTranslatedEvent(translatedThreadUsageEvent(threadID, model, usage, contextCompacted))
	}
	d.tokenUsageMu.Unlock()
	if postCompactionBoundary != nil {
		logCodexContextCompactionUsage(*postCompactionBoundary)
	}
	return model == ""
}

func (d *CodexAppServerDriver) collectPendingThreadUsageLocked(threadID string) []translatedEvent {
	return d.collectPendingThreadUsageWithHydratedModelLocked(threadID, "", true)
}

func (d *CodexAppServerDriver) collectPendingThreadUsageWithHydratedModelLocked(threadID string, hydratedModel string, allowCurrentFallback bool) []translatedEvent {
	threadID = strings.TrimSpace(threadID)
	hydratedModel = strings.TrimSpace(hydratedModel)
	if threadID == "" || len(d.pendingTokenUsageByThread[threadID]) == 0 {
		return nil
	}
	pending := d.pendingTokenUsageByThread[threadID]
	emissions := make([]translatedEvent, 0, len(pending))
	remaining := make([]pendingAppServerThreadTokenUsage, 0, len(pending))
	for _, item := range pending {
		model := strings.TrimSpace(item.model)
		if model == "" && hydratedModel != "" {
			model = hydratedModel
		}
		if model == "" && allowCurrentFallback {
			model = d.modelForThreadTurnLocked(item.threadID, item.turnID)
		}
		if model == "" {
			remaining = append(remaining, item)
			continue
		}
		emissions = append(emissions, translatedThreadUsageEvent(
			item.threadID,
			model,
			item.usage,
			item.contextCompacted,
		))
	}
	if len(remaining) == 0 {
		delete(d.pendingTokenUsageByThread, threadID)
	} else {
		d.pendingTokenUsageByThread[threadID] = remaining
	}
	return emissions
}

func translatedThreadUsageEvent(
	threadID string,
	model string,
	usage appServerThreadTokenUsage,
	contextCompacted bool,
) translatedEvent {
	last := usage.Last
	var contextWindowTokens int64
	if usage.ModelContextWindow != nil {
		contextWindowTokens = *usage.ModelContextWindow
	}
	return translatedEvent{
		Kind:                consts.ParsedEventKindUsage,
		Runtime:             "codex",
		SessionID:           strings.TrimSpace(threadID),
		Model:               strings.TrimSpace(model),
		InputTokens:         nonNegativeInt64TokenCount(last.InputTokens),
		CachedInputTokens:   nonNegativeInt64TokenCount(last.CachedInputTokens),
		OutputTokens:        nonNegativeInt64TokenCount(last.OutputTokens),
		ActiveContextTokens: nonNegativeInt64TokenCount(last.TotalTokens),
		ContextWindowTokens: nonNegativeInt64TokenCount(contextWindowTokens),
		ContextCompacted:    contextCompacted,
	}
}

func nonNegativeInt64TokenCount(value int64) uint64 {
	if value <= 0 {
		return 0
	}
	return uint64(value)
}

func (d *CodexAppServerDriver) enqueueThreadUsageEmissions(events []translatedEvent) {
	for _, event := range events {
		d.enqueueTranslatedEvent(event)
	}
}

func (d *CodexAppServerDriver) ensureThreadModelHydration(threadID string) {
	threadID = strings.TrimSpace(threadID)
	if threadID == "" {
		return
	}
	d.mu.Lock()
	if d.rawStdin == nil || (strings.TrimSpace(d.modelByThread[threadID]) != "" && !d.pendingThreadUsageNeedsHydrationLocked(threadID)) {
		d.mu.Unlock()
		return
	}
	if _, blocked := d.modelHydrationBlocked[threadID]; blocked {
		d.mu.Unlock()
		return
	}
	if d.modelHydrationInFlight == nil {
		d.modelHydrationInFlight = make(map[string]uint64)
	}
	if _, exists := d.modelHydrationInFlight[threadID]; exists {
		d.mu.Unlock()
		return
	}
	generation := d.launchGeneration
	d.modelHydrationInFlight[threadID] = generation
	d.mu.Unlock()

	go d.hydrateThreadModel(generation, threadID)
}

type codexThreadModelHydrationState int

const (
	codexThreadModelHydrationStale codexThreadModelHydrationState = iota
	codexThreadModelHydrationResolved
	codexThreadModelHydrationNeeded
	codexThreadModelHydrationTerminal
)

func (d *CodexAppServerDriver) hydrateThreadModel(generation uint64, threadID string) {
	var lastErr error
	for attempt := 1; attempt <= codexThreadModelHydrationRetries; attempt++ {
		state := d.threadModelHydrationState(generation, threadID)
		if state != codexThreadModelHydrationNeeded {
			d.finishThreadModelHydration(generation, threadID, "", state == codexThreadModelHydrationTerminal)
			return
		}
		if attempt > 1 {
			time.Sleep(time.Duration(attempt-1) * codexThreadModelHydrationBackoff)
			state = d.threadModelHydrationState(generation, threadID)
			if state != codexThreadModelHydrationNeeded {
				d.finishThreadModelHydration(generation, threadID, "", state == codexThreadModelHydrationTerminal)
				return
			}
		}

		ctx, cancel := context.WithTimeout(context.Background(), codexThreadModelHydrationTimeout)
		model, err := d.readRunningThreadModel(ctx, generation, threadID)
		cancel()
		if err == nil {
			if dropped := d.finishThreadModelHydration(generation, threadID, model, false); dropped > 0 {
				slog.Warn("dropping unknown Codex child token usage after child closed during model hydration",
					"driver", "codex-app-server",
					"thread_id", threadID,
					"pending_usage_events", dropped)
			}
			return
		}
		if errors.Is(err, errCodexThreadModelHydrationNotLoaded) {
			dropped := d.finishThreadModelHydration(generation, threadID, "", true)
			if dropped > 0 {
				slog.Warn("dropping unknown Codex child token usage because child is not loaded",
					"driver", "codex-app-server",
					"thread_id", threadID,
					"pending_usage_events", dropped)
			}
			return
		}
		if errors.Is(err, errCodexAppServerRequestDispatchRejected) {
			state = d.threadModelHydrationState(generation, threadID)
			d.finishThreadModelHydration(
				generation,
				threadID,
				"",
				state == codexThreadModelHydrationTerminal,
			)
			return
		}
		lastErr = err
		slog.Debug("failed hydrating Codex child thread model",
			"driver", "codex-app-server",
			"thread_id", threadID,
			"attempt", attempt,
			"error", err)
	}

	dropped := d.finishThreadModelHydration(generation, threadID, "", true)
	if dropped > 0 {
		slog.Warn("dropping unknown Codex child token usage after model hydration retries",
			"driver", "codex-app-server",
			"thread_id", threadID,
			"attempts", codexThreadModelHydrationRetries,
			"pending_usage_events", dropped,
			"error", lastErr)
	}
}

func (d *CodexAppServerDriver) threadModelHydrationState(generation uint64, threadID string) codexThreadModelHydrationState {
	d.mu.Lock()
	defer d.mu.Unlock()
	inFlightGeneration, inFlight := d.modelHydrationInFlight[threadID]
	if d.launchGeneration != generation || !inFlight || inFlightGeneration != generation {
		return codexThreadModelHydrationStale
	}
	if len(d.pendingTokenUsageByThread[threadID]) == 0 || !d.pendingThreadUsageNeedsHydrationLocked(threadID) {
		return codexThreadModelHydrationResolved
	}
	if d.rawStdin == nil {
		return codexThreadModelHydrationTerminal
	}
	if _, blocked := d.modelHydrationBlocked[threadID]; blocked {
		return codexThreadModelHydrationTerminal
	}
	return codexThreadModelHydrationNeeded
}

func (d *CodexAppServerDriver) pendingThreadUsageNeedsHydrationLocked(threadID string) bool {
	for _, pending := range d.pendingTokenUsageByThread[strings.TrimSpace(threadID)] {
		if strings.TrimSpace(pending.model) == "" {
			return true
		}
	}
	return false
}

// finishThreadModelHydration serializes model installation and pending usage
// emission with live token usage. This preserves app-server notification order
// and refuses to overwrite a newer settings/collaboration model with an older
// thread/resume snapshot.
func (d *CodexAppServerDriver) finishThreadModelHydration(generation uint64, threadID string, model string, terminal bool) int {
	d.tokenUsageMu.Lock()
	defer d.tokenUsageMu.Unlock()

	d.mu.Lock()
	inFlightGeneration, inFlight := d.modelHydrationInFlight[threadID]
	if d.launchGeneration != generation || !inFlight || inFlightGeneration != generation {
		d.mu.Unlock()
		return 0
	}
	delete(d.modelHydrationInFlight, threadID)

	_, blocked := d.modelHydrationBlocked[threadID]
	if blocked {
		model = ""
		terminal = true
	}
	model = strings.TrimSpace(model)
	if !blocked && model != "" && strings.TrimSpace(d.modelByThread[threadID]) == "" {
		d.modelByThread[threadID] = model
		if strings.TrimSpace(d.threadID) == threadID {
			d.effectiveModel = model
		}
	}
	latestPendingAssociated := false
	if pending := d.pendingTokenUsageByThread[threadID]; len(pending) > 0 {
		latestModel := strings.TrimSpace(pending[len(pending)-1].model)
		if latestModel == "" {
			latestModel = model
		}
		latestPendingAssociated = latestModel != ""
	}
	// Even after close/interruption, usage that captured an authoritative model
	// before the fence is safe to publish. Never apply the late hydration model
	// or the current-model fallback while blocked; unresolved historical usage
	// remains below and is dropped.
	emissions := d.collectPendingThreadUsageWithHydratedModelLocked(threadID, model, false)
	dropped := 0
	if terminal && len(d.pendingTokenUsageByThread[threadID]) > 0 {
		dropped = len(d.pendingTokenUsageByThread[threadID])
		delete(d.pendingTokenUsageByThread, threadID)
		if !latestPendingAssociated {
			delete(d.lastTokenUsageByThread, threadID)
		}
		if strings.TrimSpace(d.threadID) == threadID {
			d.lastTokenUsage = appServerThreadTokenUsage{}
		}
	}
	d.mu.Unlock()
	d.enqueueThreadUsageEmissions(emissions)
	return dropped
}

func (d *CodexAppServerDriver) setThreadModelHydrationBlocked(threadIDs []string, blocked bool) {
	d.tokenUsageMu.Lock()
	d.mu.Lock()
	if d.modelHydrationBlocked == nil {
		d.modelHydrationBlocked = make(map[string]struct{})
	}
	dropped := make(map[string]int)
	for _, threadID := range threadIDs {
		threadID = strings.TrimSpace(threadID)
		if threadID == "" {
			continue
		}
		if blocked {
			d.modelHydrationBlocked[threadID] = struct{}{}
			if strings.TrimSpace(d.modelForThreadLocked(threadID)) == "" {
				delete(d.lastTokenUsageByThread, threadID)
			}
			if _, inFlight := d.modelHydrationInFlight[threadID]; !inFlight {
				dropped[threadID] = len(d.pendingTokenUsageByThread[threadID])
				delete(d.pendingTokenUsageByThread, threadID)
			}
			continue
		}
		delete(d.modelHydrationBlocked, threadID)
	}
	d.mu.Unlock()
	d.tokenUsageMu.Unlock()
	for threadID, count := range dropped {
		if count > 0 {
			slog.Warn("dropping unknown Codex child token usage after child close",
				"driver", "codex-app-server",
				"thread_id", threadID,
				"pending_usage_events", count)
		}
	}
}

func (d *CodexAppServerDriver) readRunningThreadModel(ctx context.Context, generation uint64, threadID string) (string, error) {
	threadID = strings.TrimSpace(threadID)
	if threadID == "" {
		return "", fmt.Errorf("Codex child thread id is empty")
	}
	readRaw, err := d.requestWithDispatchGuard(ctx, "thread/read", map[string]any{
		"threadId":     threadID,
		"includeTurns": false,
	}, func(driver *CodexAppServerDriver) bool {
		return driver.threadModelHydrationDispatchAllowedLocked(generation, threadID)
	})
	if err != nil {
		return "", err
	}
	var read appServerThreadResult
	if err := json.Unmarshal(readRaw, &read); err != nil {
		return "", fmt.Errorf("decode child thread/read result: %w", err)
	}
	responseThreadID := strings.TrimSpace(read.Thread.ID)
	if responseThreadID != "" && responseThreadID != threadID {
		return "", fmt.Errorf("Codex child thread/read returned thread %q for %q", responseThreadID, threadID)
	}
	if normalizeThreadStatus(read.Thread.Status).Type == "notLoaded" {
		return "", errCodexThreadModelHydrationNotLoaded
	}

	result, err := d.requestWithDispatchGuard(ctx, "thread/resume", map[string]any{
		"threadId":     threadID,
		"excludeTurns": true,
	}, func(driver *CodexAppServerDriver) bool {
		return driver.threadModelHydrationDispatchAllowedLocked(generation, threadID)
	})
	if err != nil {
		return "", err
	}
	var resumed appServerThreadResult
	if err := json.Unmarshal(result, &resumed); err != nil {
		return "", fmt.Errorf("decode child thread/resume result: %w", err)
	}
	responseThreadID = strings.TrimSpace(resumed.Thread.ID)
	if responseThreadID != "" && responseThreadID != threadID {
		return "", fmt.Errorf("Codex child thread/resume returned thread %q for %q", responseThreadID, threadID)
	}
	model := strings.TrimSpace(resumed.Model)
	if model == "" {
		return "", fmt.Errorf("Codex child thread/resume returned an empty model")
	}
	return model, nil
}

func (d *CodexAppServerDriver) threadModelHydrationDispatchAllowedLocked(generation uint64, threadID string) bool {
	threadID = strings.TrimSpace(threadID)
	inFlightGeneration, inFlight := d.modelHydrationInFlight[threadID]
	if threadID == "" || d.launchGeneration != generation || !inFlight || inFlightGeneration != generation {
		return false
	}
	if d.rawStdin == nil || !d.pendingThreadUsageNeedsHydrationLocked(threadID) {
		return false
	}
	_, blocked := d.modelHydrationBlocked[threadID]
	return !blocked
}

func (d *CodexAppServerDriver) modelForThreadTurnLocked(threadID string, turnID string) string {
	if key := threadTurnModelKey(threadID, turnID); key != "" {
		if model := strings.TrimSpace(d.modelByThreadTurn[key]); model != "" {
			return model
		}
	}
	return d.modelForThreadLocked(threadID)
}

func (d *CodexAppServerDriver) modelForThreadLocked(threadID string) string {
	threadID = strings.TrimSpace(threadID)
	if model := strings.TrimSpace(d.modelByThread[threadID]); model != "" {
		return model
	}
	primaryThreadID := strings.TrimSpace(d.threadID)
	// effectiveModel is the primary thread's compatibility fallback. Never use
	// it for a known foreign thread: Codex spawnAgent may select a different
	// model, and a missing source association is safer than charging child usage to the
	// parent model.
	if primaryThreadID == "" || primaryThreadID == threadID {
		return strings.TrimSpace(d.effectiveModel)
	}
	return ""
}

func sameAppServerThreadTokenUsage(a appServerThreadTokenUsage, b appServerThreadTokenUsage) bool {
	return sameAppServerTokenUsageBreakdown(a.Last, b.Last) &&
		sameAppServerTokenUsageBreakdown(a.Total, b.Total) &&
		appServerOptionalInt64Value(a.ModelContextWindow) == appServerOptionalInt64Value(b.ModelContextWindow)
}

func sameAppServerTokenUsageBreakdown(a appServerTokenUsageBreakdown, b appServerTokenUsageBreakdown) bool {
	return a.InputTokens == b.InputTokens &&
		a.CachedInputTokens == b.CachedInputTokens &&
		a.OutputTokens == b.OutputTokens &&
		a.ReasoningOutputTokens == b.ReasoningOutputTokens &&
		a.TotalTokens == b.TotalTokens
}

func appServerOptionalInt64Value(value *int64) int64 {
	if value == nil {
		return 0
	}
	return *value
}

func (d *CodexAppServerDriver) currentThreadState(threadID string) (appServerThreadStatus, string, bool) {
	d.mu.Lock()
	defer d.mu.Unlock()

	trimmedThreadID := strings.TrimSpace(threadID)
	primaryThreadID := strings.TrimSpace(d.threadID)
	if trimmedThreadID == "" || (primaryThreadID != "" && primaryThreadID != trimmedThreadID) {
		return appServerThreadStatus{}, "", false
	}
	return d.threadStatus, d.activeTurnID, true
}

func (d *CodexAppServerDriver) currentActiveTurnID() string {
	d.mu.Lock()
	defer d.mu.Unlock()
	return d.activeTurnID
}

// isPrimaryThreadEvent reports whether a thread-scoped app-server event may
// affect the primary AgentTeam runtime session. The primary thread ID is bound only
// by an authoritative thread start/resume/read snapshot. Before that snapshot
// arrives there is no competing child thread in a fresh app-server session, so
// accepting a well-formed event preserves startup compatibility without using
// the event to bind the ID.
func (d *CodexAppServerDriver) isPrimaryThreadEvent(threadID string) bool {
	d.mu.Lock()
	defer d.mu.Unlock()
	return d.isPrimaryThreadEventLocked(threadID)
}

func (d *CodexAppServerDriver) isPrimaryThreadEventLocked(threadID string) bool {
	trimmedThreadID := strings.TrimSpace(threadID)
	if trimmedThreadID == "" {
		return false
	}
	primaryThreadID := strings.TrimSpace(d.threadID)
	return primaryThreadID == "" || primaryThreadID == trimmedThreadID
}

// isCurrentPrimaryTurnActivity fences notifications that can make the root
// Process busy. Before the primary thread is bound, startup compatibility keeps
// accepting well-formed activity. Once bound, only the concrete active turn may
// publish activity; completed, stale, idle, not-loaded, and system-error turns
// are metadata/cleanup-only until a new TurnStarted is observed.
//
// Callers that need ordering against terminal notifications must also hold
// turnEventMu (handleNotification does this for app-server notifications).
func (d *CodexAppServerDriver) isCurrentPrimaryTurnActivity(threadID string, turnID string) bool {
	trimmedThreadID := strings.TrimSpace(threadID)
	trimmedTurnID := strings.TrimSpace(turnID)
	if trimmedThreadID == "" {
		return false
	}

	d.mu.Lock()
	defer d.mu.Unlock()
	if !d.isPrimaryThreadEventLocked(trimmedThreadID) {
		return false
	}
	if strings.TrimSpace(d.threadID) == "" {
		return true
	}
	if trimmedTurnID == "" {
		return false
	}
	if _, completed := d.completedTurnIDs[trimmedTurnID]; completed {
		return false
	}
	statusType := normalizeThreadStatus(d.threadStatus).Type
	if statusType == "" && strings.TrimSpace(d.activeTurnID) == "" {
		// Compatibility for unit embedders that bind only a primary thread id.
		// Production snapshots always carry a concrete app-server status.
		return true
	}
	return statusType == "active" &&
		strings.TrimSpace(d.activeTurnID) == trimmedTurnID
}

// CurrentActiveTurnID returns the turn ID currently being served, if any.
func (d *CodexAppServerDriver) CurrentActiveTurnID() string {
	return d.currentActiveTurnID()
}

// ---------------------------------------------------------------------------
// Turn lifecycle
// ---------------------------------------------------------------------------

func (d *CodexAppServerDriver) markTurnStarted(threadID string, turn appServerTurn) bool {
	trimmedThreadID := strings.TrimSpace(threadID)
	trimmedTurnID := strings.TrimSpace(turn.ID)

	d.mu.Lock()
	defer d.mu.Unlock()
	if !d.isPrimaryThreadEventLocked(trimmedThreadID) {
		return false
	}
	if trimmedTurnID != "" {
		if _, completed := d.completedTurnIDs[trimmedTurnID]; completed {
			return true
		}
	}
	isCompaction := false
	switch {
	case trimmedTurnID != "" && strings.TrimSpace(d.compactionTurnID) == trimmedTurnID:
		isCompaction = true
	case trimmedTurnID != "" && d.compactionInProgress && strings.TrimSpace(d.compactionTurnID) == "" && strings.TrimSpace(d.activeTurnID) == "":
		d.compactionTurnID = trimmedTurnID
		isCompaction = true
	}
	d.threadStatus = appServerThreadStatus{Type: "active"}
	d.activeTurnNonSteerableKind = ""
	if trimmedTurnID != "" {
		d.activeTurnID = trimmedTurnID
	}
	if !isCompaction {
		d.clearCompactionMutationLocked()
		d.contextCompactionItemFence = codexContextCompactionItemFence{}
	}
	return isCompaction
}

func (d *CodexAppServerDriver) acknowledgeTurnStarted(threadID string, turn appServerTurn) {
	trimmedThreadID := strings.TrimSpace(threadID)
	trimmedTurnID := strings.TrimSpace(turn.ID)
	if trimmedThreadID == "" || trimmedTurnID == "" {
		return
	}

	d.mu.Lock()
	defer d.mu.Unlock()
	if !d.isPrimaryThreadEventLocked(trimmedThreadID) {
		return
	}
	if _, completed := d.completedTurnIDs[trimmedTurnID]; completed {
		return
	}
	if d.threadStatus.Type == "active" && strings.TrimSpace(d.activeTurnID) != "" && strings.TrimSpace(d.activeTurnID) != trimmedTurnID {
		return
	}
	d.threadStatus = appServerThreadStatus{Type: "active"}
	d.activeTurnID = trimmedTurnID
	d.activeTurnNonSteerableKind = ""
	d.clearCompactionMutationLocked()
	d.contextCompactionItemFence = codexContextCompactionItemFence{}
}

type codexTurnCompletion struct {
	isCompaction        bool
	turnEnded           bool
	compactionCompleted bool
	compactionBoundary  codexContextCompactionBoundary
}

func (d *CodexAppServerDriver) markTurnCompleted(threadID string, turn appServerTurn) codexTurnCompletion {
	trimmedThreadID := strings.TrimSpace(threadID)
	trimmedTurnID := strings.TrimSpace(turn.ID)

	d.mu.Lock()
	defer d.mu.Unlock()
	if !d.isPrimaryThreadEventLocked(trimmedThreadID) {
		return codexTurnCompletion{}
	}
	isCompaction := trimmedTurnID != "" && strings.TrimSpace(d.compactionTurnID) == trimmedTurnID
	activeTurnID := strings.TrimSpace(d.activeTurnID)
	completedActiveTurn := trimmedTurnID != "" && activeTurnID == trimmedTurnID
	_, alreadyCompleted := d.completedTurnIDs[trimmedTurnID]
	turnEnded := !alreadyCompleted && !isCompaction && (activeTurnID == "" || completedActiveTurn)
	compactionCompleted := false
	if trimmedTurnID != "" {
		if d.completedTurnIDs == nil {
			d.completedTurnIDs = make(map[string]struct{})
		}
		if _, exists := d.completedTurnIDs[trimmedTurnID]; !exists {
			d.completedTurnIDs[trimmedTurnID] = struct{}{}
			d.completedTurnCount++
		}
		if completedActiveTurn {
			d.activeTurnID = ""
			d.activeTurnNonSteerableKind = ""
		}
	} else if !isCompaction && activeTurnID == "" {
		d.activeTurnID = ""
		d.activeTurnNonSteerableKind = ""
	}
	if !isCompaction && (completedActiveTurn || (trimmedTurnID == "" && activeTurnID == "")) {
		d.threadStatus = appServerThreadStatus{Type: "idle"}
	}
	if isCompaction && !alreadyCompleted && d.compactionTerminalStatus.Type != "" {
		// app-server publishes this terminal status before turn/completed. Apply it
		// only when the completion carries the exact compaction turn id; a stale or
		// duplicate completion must not consume the boundary.
		d.threadStatus = normalizeThreadStatus(d.compactionTerminalStatus)
		d.activeTurnID = ""
		d.activeTurnNonSteerableKind = ""
		d.clearCompactionMutationLocked()
		compactionCompleted = true
	}
	compactionBoundary := codexContextCompactionBoundary{}
	if compactionCompleted &&
		d.contextCompactionBoundary.threadID == trimmedThreadID &&
		d.contextCompactionBoundary.turnID == trimmedTurnID {
		compactionBoundary = d.contextCompactionBoundary
	}
	if !d.compactionInProgress {
		d.clearCompactionMutationLocked()
	}
	return codexTurnCompletion{
		isCompaction:        isCompaction,
		turnEnded:           turnEnded,
		compactionCompleted: compactionCompleted,
		compactionBoundary:  compactionBoundary,
	}
}

func (d *CodexAppServerDriver) isCompactionTurn(threadID string, turnID string) bool {
	trimmedTurnID := strings.TrimSpace(turnID)
	if trimmedTurnID == "" {
		return false
	}

	trimmedThreadID := strings.TrimSpace(threadID)
	d.mu.Lock()
	defer d.mu.Unlock()
	if trimmedThreadID != "" && strings.TrimSpace(d.threadID) != trimmedThreadID {
		return false
	}
	return strings.TrimSpace(d.compactionTurnID) == trimmedTurnID
}

// ---------------------------------------------------------------------------
// Steering helpers
// ---------------------------------------------------------------------------

func (d *CodexAppServerDriver) canSteerLocked(sessionID string) bool {
	trimmedSessionID := strings.TrimSpace(sessionID)
	if trimmedSessionID == "" {
		return false
	}
	if strings.TrimSpace(d.threadID) != trimmedSessionID {
		return false
	}
	if strings.TrimSpace(d.activeTurnID) == "" {
		return false
	}
	if d.threadStatus.Type != "active" {
		return false
	}
	if strings.TrimSpace(d.activeTurnNonSteerableKind) != "" {
		return false
	}
	if d.compactionInProgress {
		return false
	}
	return true
}

func (d *CodexAppServerDriver) threadKnownActive(sessionID string) bool {
	trimmedSessionID := strings.TrimSpace(sessionID)
	if trimmedSessionID == "" {
		return false
	}

	d.mu.Lock()
	defer d.mu.Unlock()
	return strings.TrimSpace(d.threadID) == trimmedSessionID && d.threadStatus.Type == "active"
}

func (d *CodexAppServerDriver) markActiveTurnNonSteerable(threadID string, turnKind string) {
	trimmedThreadID := strings.TrimSpace(threadID)
	trimmedTurnKind := strings.TrimSpace(turnKind)
	if trimmedThreadID == "" || trimmedTurnKind == "" {
		return
	}

	d.mu.Lock()
	defer d.mu.Unlock()
	if strings.TrimSpace(d.threadID) != trimmedThreadID {
		return
	}
	if d.threadStatus.Type != "active" || strings.TrimSpace(d.activeTurnID) == "" {
		return
	}
	d.activeTurnNonSteerableKind = trimmedTurnKind
}

// ---------------------------------------------------------------------------
// Item streaming dedup
// ---------------------------------------------------------------------------

func (d *CodexAppServerDriver) markItemStreamed(itemID string) {
	trimmedItemID := strings.TrimSpace(itemID)
	if trimmedItemID == "" {
		return
	}

	d.mu.Lock()
	defer d.mu.Unlock()
	if d.streamedItemIDs == nil {
		d.streamedItemIDs = make(map[string]struct{})
	}
	d.streamedItemIDs[trimmedItemID] = struct{}{}
}

func (d *CodexAppServerDriver) clearItemStreamed(itemID string) {
	trimmedItemID := strings.TrimSpace(itemID)
	if trimmedItemID == "" {
		return
	}

	d.mu.Lock()
	defer d.mu.Unlock()
	if d.streamedItemIDs == nil {
		return
	}
	delete(d.streamedItemIDs, trimmedItemID)
}

func (d *CodexAppServerDriver) wasItemStreamed(itemID string) bool {
	trimmedItemID := strings.TrimSpace(itemID)
	if trimmedItemID == "" {
		return false
	}

	d.mu.Lock()
	defer d.mu.Unlock()
	_, exists := d.streamedItemIDs[trimmedItemID]
	return exists
}

// ---------------------------------------------------------------------------
// Pure helper functions
// ---------------------------------------------------------------------------

func trimmedAnyString(value any) string {
	text, _ := value.(string)
	return strings.TrimSpace(text)
}

func containsString(values []string, want string) bool {
	want = strings.TrimSpace(want)
	if want == "" {
		return false
	}
	for _, value := range values {
		if strings.TrimSpace(value) == want {
			return true
		}
	}
	return false
}

func appendUniqueString(values *[]string, value string) {
	value = strings.TrimSpace(value)
	if value == "" || containsString(*values, value) {
		return
	}
	*values = append(*values, value)
}

func normalizeThreadStatus(status appServerThreadStatus) appServerThreadStatus {
	normalized := appServerThreadStatus{
		Type: strings.TrimSpace(status.Type),
	}
	for _, flag := range status.ActiveFlags {
		trimmed := strings.TrimSpace(flag)
		if trimmed == "" {
			continue
		}
		normalized.ActiveFlags = append(normalized.ActiveFlags, trimmed)
	}
	return normalized
}

func activeTurnNotSteerableKind(err error) (string, bool) {
	var requestErr *appServerRequestError
	if !errors.As(err, &requestErr) || len(requestErr.Data) == 0 {
		return "", false
	}

	var data struct {
		ActiveTurnNotSteerable *struct {
			TurnKind string `json:"turnKind"`
		} `json:"activeTurnNotSteerable,omitempty"`
	}
	if err := json.Unmarshal(requestErr.Data, &data); err != nil || data.ActiveTurnNotSteerable == nil {
		return "", false
	}

	turnKind := strings.TrimSpace(data.ActiveTurnNotSteerable.TurnKind)
	if turnKind == "" {
		return "", false
	}
	return turnKind, true
}