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/types"
)

// ---------------------------------------------------------------------------
// drivercontract.BusyStdinDeliveryDriver / drivercontract.BusyStdinMessageDriver
// ---------------------------------------------------------------------------

// SupportsBusyStdinDelivery reports whether the running thread can accept real
// routed-message input as an immediate freshness observation. Codex active-turn
// steer is ack-aware instead: transport acceptance only creates pending runtime
// input, and AgentTeam waits for the app-server userMessage.clientId echo before
// surfacing the body.
func (d *CodexAppServerDriver) SupportsBusyStdinDelivery() bool {
	return false
}

// SupportsAckedBusyRuntimeInputDelivery reports the structural app-server
// capability, not whether the driver's cached thread snapshot happens to be
// steerable at this instant. DeliverAckedBusyRuntimeInput reconciles an
// unsteerable cache with thread/read before choosing turn/steer or turn/start.
func (d *CodexAppServerDriver) SupportsAckedBusyRuntimeInputDelivery() bool {
	return true
}

// ReconcileSuccessfulAckedBusyRuntimeInput mirrors the Codex TUI's steer lifecycle:
// once turn/steer returns success, the runtime owns exactly one pending input
// until its committed user-message event arrives. The TUI does not poll
// thread/read or resubmit during that interval; it retries only explicit
// turn/steer rejection races. AgentTeam still reconciles ambiguous transport
// outcomes and ownership restored after a process restart.
func (d *CodexAppServerDriver) ReconcileSuccessfulAckedBusyRuntimeInput() bool {
	return false
}

// DeliverRuntimeInput is the correlated idle/direct counterpart of active-turn
// acked delivery. Both paths perform the same authoritative lifecycle read:
// fresh Codex threads use the bounded metadata fallback implemented by
// readThreadForLifecycle, while materialized threads retain full turn-aware
// routing. This prevents a stale cached idle state from speculatively starting
// a turn if app-server has already moved to an active lifecycle.
func (d *CodexAppServerDriver) DeliverRuntimeInput(ctx context.Context, input drivercontract.RuntimeInput, sessionID string) error {
	return d.DeliverAckedBusyRuntimeInput(ctx, input, sessionID)
}

// ReconcileRuntimeInput performs one read-only, authoritative history lookup
// for clientUserMessageId. It never resumes the thread and never polls: Process
// owns bounded retry timing for uncertain transport outcomes.
func (d *CodexAppServerDriver) ReconcileRuntimeInput(ctx context.Context, inputID, sessionID string) (RuntimeInputReconciliation, error) {
	inputID = strings.TrimSpace(inputID)
	threadID := strings.TrimSpace(sessionID)
	if inputID == "" || threadID == "" {
		return RuntimeInputReconciliationUnknown, fmt.Errorf("codex runtime input reconciliation requires input and thread ids")
	}

	read, err := d.readThreadForLifecycle(ctx, threadID)
	if err != nil {
		return RuntimeInputReconciliationUnknown, fmt.Errorf("read codex thread %s for runtime input reconciliation: %w", threadID, err)
	}
	if read.TurnsState != codexThreadTurnsAvailable {
		statusType := normalizeThreadStatus(read.Result.Thread.Status).Type
		switch read.TurnsState {
		case codexThreadTurnsBeforeFirstUserMessage:
			// The rollout has not materialized, but turn/start accepts its
			// submission through an asynchronous channel before the rollout is
			// necessarily visible. An inactive metadata snapshot therefore cannot
			// prove that an ambiguous mutation was never accepted. Stay fail-closed
			// rather than replaying the same client id speculatively.
			switch statusType {
			case "active":
				return RuntimeInputReconciliationPending, nil
			default:
				return RuntimeInputReconciliationUnknown, nil
			}
		case codexThreadTurnsUnavailable:
			// Ephemeral threads never expose turns, so an inactive status cannot
			// prove that this correlated item is absent from prior history.
			if statusType == "active" {
				return RuntimeInputReconciliationPending, nil
			}
			return RuntimeInputReconciliationUnknown, nil
		default:
			return RuntimeInputReconciliationUnknown, nil
		}
	}
	readResult := read.Result

	incompleteItems := false
	runtimeDeliveryMarker := ""
	if types.IsRuntimeDeliveryReceipt(inputID) {
		runtimeDeliveryMarker = types.RuntimeDeliveryReceiptMarker(inputID)
	}
	for _, turn := range readResult.Thread.Turns {
		for _, item := range turn.Items {
			if runtimeDeliveryMarker != "" && appServerItemContainsRuntimeDeliveryMarker(item, runtimeDeliveryMarker) {
				return RuntimeInputReconciliationAcked, nil
			}
			if runtimeDeliveryMarker == "" && item.Type == "userMessage" && strings.TrimSpace(item.ClientID) == inputID {
				return RuntimeInputReconciliationAcked, nil
			}
		}
		switch strings.TrimSpace(turn.ItemsView) {
		case "", "full":
			// Empty is accepted for compatibility with app-server versions that
			// predate itemsView; includeTurns still returned their full item list.
		case "notLoaded", "summary":
			incompleteItems = true
		default:
			return RuntimeInputReconciliationUnknown, nil
		}
	}
	if incompleteItems {
		if normalizeThreadStatus(readResult.Thread.Status).Type == "active" {
			return RuntimeInputReconciliationPending, nil
		}
		return RuntimeInputReconciliationUnknown, nil
	}
	switch normalizeThreadStatus(readResult.Thread.Status).Type {
	case "active":
		// A transport-accepted turn/start or turn/steer can precede the
		// userMessage item becoming visible in thread/read. Full history without
		// the client id is therefore not authoritative absence while active.
		return RuntimeInputReconciliationPending, nil
	case "idle", "systemError", "notLoaded":
		return RuntimeInputReconciliationAbsent, nil
	default:
		return RuntimeInputReconciliationUnknown, nil
	}
}

func appServerItemContainsRuntimeDeliveryMarker(item appServerItem, marker string) bool {
	marker = strings.TrimSpace(marker)
	if marker == "" {
		return false
	}
	if strings.Contains(item.Text, marker) {
		return true
	}
	for _, contentItem := range item.ContentItems {
		if strings.Contains(contentItem.Text, marker) {
			return true
		}
	}
	return rawJSONContainsRuntimeDeliveryMarker(item.Content, marker) ||
		rawJSONContainsRuntimeDeliveryMarker(item.Output, marker) ||
		rawJSONContainsRuntimeDeliveryMarker(item.Result, marker)
}

func rawJSONContainsRuntimeDeliveryMarker(raw json.RawMessage, marker string) bool {
	if len(raw) == 0 || strings.TrimSpace(marker) == "" {
		return false
	}
	var value any
	if err := json.Unmarshal(raw, &value); err != nil {
		return false
	}
	return jsonValueContainsRuntimeDeliveryMarker(value, marker)
}

func jsonValueContainsRuntimeDeliveryMarker(value any, marker string) bool {
	switch typed := value.(type) {
	case string:
		return strings.Contains(typed, marker)
	case []any:
		for _, item := range typed {
			if jsonValueContainsRuntimeDeliveryMarker(item, marker) {
				return true
			}
		}
	case map[string]any:
		for _, item := range typed {
			if jsonValueContainsRuntimeDeliveryMarker(item, marker) {
				return true
			}
		}
	}
	return false
}

// DeliverAckedBusyRuntimeInput delivers a routed message with its stable
// RuntimeInput.ID. Live delivery target selection follows app-server state:
// idle/systemError start, active regular turns steer, notLoaded threads resume,
// and an active snapshot missing its turn id gets one bounded authoritative
// read before delivery stops at the lifecycle boundary. Both request shapes carry
// clientUserMessageId, so transport acceptance remains pending until a
// userMessage item echoes the same client id.
func (d *CodexAppServerDriver) DeliverAckedBusyRuntimeInput(ctx context.Context, input drivercontract.RuntimeInput, sessionID string) error {
	threadID := strings.TrimSpace(sessionID)
	if threadID == "" {
		return fmt.Errorf("codex acked delivery requires a thread id")
	}
	inputID := strings.TrimSpace(input.ID)
	if inputID == "" {
		return fmt.Errorf("codex acked delivery requires a runtime input id")
	}
	if input.Empty() {
		return fmt.Errorf("codex acked delivery requires non-empty input")
	}
	params, err := d.startTurnParams(d.currentLaunchConfig(), input.PromptText(), threadID)
	if err != nil {
		return err
	}
	if params == nil {
		return fmt.Errorf("codex acked delivery produced empty turn params")
	}
	params["clientUserMessageId"] = inputID
	return d.deliverRoutedRuntimeInput(ctx, threadID, params)
}

type codexLiveDeliveryTarget struct {
	method string
	turnID string
}

type codexLiveDeliverySnapshot struct {
	known            bool
	status           appServerThreadStatus
	activeTurnID     string
	nonSteerableKind string
	compacting       bool
}

// resolveLiveDeliveryTarget is deliberately bounded. thread/read is an
// authoritative reconciliation request, never a listener: at most one read is
// issued for an unknown/active-without-turn snapshot. A notLoaded thread gets
// one explicit thread/resume (whose normal hydration performs one read) and is
// then routed from that resulting snapshot.
func (d *CodexAppServerDriver) resolveLiveDeliveryTarget(ctx context.Context, threadID string) (codexLiveDeliveryTarget, error) {
	return d.resolveLiveDeliveryTargetAfterRead(ctx, threadID, false)
}

// resolveAckedLiveDeliveryTarget performs an authoritative read whenever the
// cached state is not already a concrete steerable turn. The Process calls this
// API while it considers the runtime busy, so a cached idle/systemError value
// can be stale (including a server-side notLoaded transition) and must not be
// used to start blindly.
func (d *CodexAppServerDriver) resolveAckedLiveDeliveryTarget(ctx context.Context, threadID string) (codexLiveDeliveryTarget, error) {
	threadID = strings.TrimSpace(threadID)
	snapshot := d.liveDeliverySnapshot(threadID)
	if snapshot.known && snapshot.status.Type == "active" {
		if snapshot.nonSteerableKind != "" {
			return codexLiveDeliveryTarget{}, fmt.Errorf("codex thread %s cannot accept delivery during %s turn", threadID, snapshot.nonSteerableKind)
		}
		if snapshot.compacting {
			return codexLiveDeliveryTarget{}, fmt.Errorf("codex thread %s cannot accept delivery during compaction", threadID)
		}
		if snapshot.activeTurnID != "" {
			return codexLiveDeliveryTarget{method: "turn/steer", turnID: snapshot.activeTurnID}, nil
		}
	}
	if err := d.refreshLiveDeliverySnapshot(ctx, threadID); err != nil {
		return codexLiveDeliveryTarget{}, err
	}
	return d.resolveLiveDeliveryTargetAfterRead(ctx, threadID, true)
}

func (d *CodexAppServerDriver) resolveLiveDeliveryTargetAfterRead(ctx context.Context, threadID string, readAttempted bool) (codexLiveDeliveryTarget, error) {
	threadID = strings.TrimSpace(threadID)
	if threadID == "" {
		return codexLiveDeliveryTarget{}, fmt.Errorf("codex live delivery requires a thread id")
	}

	resumeAttempted := false
	for step := 0; step < 4; step++ {
		snapshot := d.liveDeliverySnapshot(threadID)
		if !snapshot.known {
			if readAttempted {
				return codexLiveDeliveryTarget{}, fmt.Errorf("codex thread %s state is unavailable after refresh", threadID)
			}
			if err := d.refreshLiveDeliverySnapshot(ctx, threadID); err != nil {
				return codexLiveDeliveryTarget{}, err
			}
			readAttempted = true
			continue
		}

		switch snapshot.status.Type {
		case "idle", "systemError":
			if snapshot.compacting {
				return codexLiveDeliveryTarget{}, fmt.Errorf("codex thread %s cannot accept delivery during compaction", threadID)
			}
			return codexLiveDeliveryTarget{method: "turn/start"}, nil

		case "active":
			if snapshot.nonSteerableKind != "" {
				return codexLiveDeliveryTarget{}, fmt.Errorf(
					"codex thread %s cannot accept delivery during %s turn",
					threadID,
					snapshot.nonSteerableKind,
				)
			}
			if snapshot.compacting {
				return codexLiveDeliveryTarget{}, fmt.Errorf("codex thread %s cannot accept delivery during compaction", threadID)
			}
			if snapshot.activeTurnID != "" {
				return codexLiveDeliveryTarget{method: "turn/steer", turnID: snapshot.activeTurnID}, nil
			}
			if !readAttempted {
				if err := d.refreshLiveDeliverySnapshot(ctx, threadID); err != nil {
					return codexLiveDeliveryTarget{}, err
				}
				readAttempted = true
				continue
			}
			// App-server derives Active from an in-progress turn, so includeTurns
			// should normally recover its id. Current Codex does not synchronously
			// reject turn/start while a turn is active: it can steer the input into
			// that turn while returning a new submission id. Starting without the
			// authoritative active id would therefore create a phantom local turn.
			return codexLiveDeliveryTarget{}, fmt.Errorf(
				"%w: codex thread %s is active but its active turn id is unavailable after authoritative read",
				errCodexTurnDispatchBoundary,
				threadID,
			)

		case "notLoaded":
			if resumeAttempted {
				return codexLiveDeliveryTarget{}, fmt.Errorf("codex thread %s remains notLoaded after resume", threadID)
			}
			if err := d.resumeThread(ctx, d.currentLaunchConfig(), threadID); err != nil {
				return codexLiveDeliveryTarget{}, fmt.Errorf("resume codex thread %s for live delivery: %w", threadID, err)
			}
			resumeAttempted = true
			// resumeThread hydrates the authoritative turn list with exactly one
			// thread/read, so do not issue another active-no-turn read afterward.
			readAttempted = true
			continue

		default:
			if !readAttempted {
				if err := d.refreshLiveDeliverySnapshot(ctx, threadID); err != nil {
					return codexLiveDeliveryTarget{}, err
				}
				readAttempted = true
				continue
			}
			return codexLiveDeliveryTarget{}, fmt.Errorf("codex thread %s cannot accept delivery in %q state", threadID, snapshot.status.Type)
		}
	}
	return codexLiveDeliveryTarget{}, fmt.Errorf("codex thread %s live delivery recovery did not converge", threadID)
}

func (d *CodexAppServerDriver) liveDeliverySnapshot(threadID string) codexLiveDeliverySnapshot {
	threadID = strings.TrimSpace(threadID)
	d.mu.Lock()
	defer d.mu.Unlock()
	if threadID == "" || strings.TrimSpace(d.threadID) != threadID {
		return codexLiveDeliverySnapshot{}
	}
	return codexLiveDeliverySnapshot{
		known:            true,
		status:           normalizeThreadStatus(d.threadStatus),
		activeTurnID:     strings.TrimSpace(d.activeTurnID),
		nonSteerableKind: strings.TrimSpace(d.activeTurnNonSteerableKind),
		compacting:       d.compactionInProgress,
	}
}

func (d *CodexAppServerDriver) refreshLiveDeliverySnapshot(ctx context.Context, threadID string) error {
	read, err := d.readThreadForLifecycle(ctx, threadID)
	if err != nil {
		return fmt.Errorf("refresh codex thread %s for live delivery: %w", threadID, err)
	}
	d.applyLifecycleThreadRead(read)
	return nil
}

const maxCodexDeliveryTurnMutations = 2

// deliverRoutedRuntimeInput is the single bounded routing engine for both
// correlated and legacy busy delivery. It performs at most two physical
// turn/start or turn/steer mutations. A local dispatch-boundary rejection is
// known not to have written a frame and may be reconciled once; every transport
// ambiguous outcome is returned without any second mutation.
func (d *CodexAppServerDriver) deliverRoutedRuntimeInput(ctx context.Context, threadID string, startParams map[string]any) error {
	target, err := d.resolveAckedLiveDeliveryTarget(ctx, threadID)
	if err != nil {
		return fmt.Errorf("%w: resolve codex delivery target: %w", errRuntimeInputMutationNotAttempted, err)
	}

	turnMutations := 0
	dispatchRefreshes := 0
	for turnMutations < maxCodexDeliveryTurnMutations {
		slog.Debug("codex delivery resolved live target",
			"thread_id", threadID,
			"delivery_method", target.method,
			"active_turn_id", target.turnID,
			"mutation_attempt", turnMutations+1,
		)
		requestParams, paramsErr := routedRuntimeInputParams(threadID, target, startParams)
		if paramsErr != nil {
			return paramsErr
		}
		err = d.sendTurnRequest(ctx, threadID, target.method, requestParams)
		if err == nil {
			return nil
		}

		// validatePrimaryTurnDispatch rejected this request before its frame was
		// written. Re-read once and make a newly fenced selection; this does not
		// consume either of the two physical mutation slots.
		if errors.Is(err, errCodexTurnDispatchBoundary) {
			if dispatchRefreshes >= 1 {
				return err
			}
			if refreshErr := d.refreshLiveDeliverySnapshot(ctx, threadID); refreshErr != nil {
				return fmt.Errorf(
					"%w: reconcile codex delivery dispatch boundary: %w",
					errRuntimeInputMutationNotAttempted,
					refreshErr,
				)
			}
			target, err = d.resolveLiveDeliveryTargetAfterRead(ctx, threadID, true)
			if err != nil {
				return fmt.Errorf(
					"%w: resolve codex delivery target after dispatch boundary: %w",
					errRuntimeInputMutationNotAttempted,
					err,
				)
			}
			dispatchRefreshes++
			continue
		}

		turnMutations++
		if errors.Is(err, errCodexAppServerTransportAmbiguous) {
			// The server may have accepted this exact mutation before the
			// response was lost. Only correlated ownership reconciliation may
			// decide whether it is safe to redeliver; never retry here, including
			// for the legacy uncorrelated API.
			return err
		}
		if turnKind, ok := ackedActiveTurnNotSteerableKind(err); ok {
			d.markActiveTurnNonSteerable(threadID, turnKind)
			return err
		}

		race, actualTurnID, raced := ackedTurnSteerRace(err)
		if !raced {
			return err
		}
		slog.Debug("codex delivery recovered turn/steer race",
			"thread_id", threadID,
			"race", race,
			"expected_turn_id", target.turnID,
			"actual_turn_id", actualTurnID,
			"mutation_attempt", turnMutations,
		)
		if race == ackedTurnSteerRaceMissing {
			d.clearAckedMissingActiveTurn(threadID)
			target = codexLiveDeliveryTarget{method: "turn/start"}
		} else {
			d.reconcileAckedActiveTurn(threadID, actualTurnID)
			target = codexLiveDeliveryTarget{method: "turn/steer", turnID: actualTurnID}
		}
		if turnMutations >= maxCodexDeliveryTurnMutations {
			// Reconcile the cache from the second explicit server response, but
			// never turn mismatch -> missing into a third mutation.
			return err
		}
	}
	return err
}

func routedRuntimeInputParams(threadID string, target codexLiveDeliveryTarget, startParams map[string]any) (map[string]any, error) {
	switch target.method {
	case "turn/start":
		return startParams, nil
	case "turn/steer":
		turnID := strings.TrimSpace(target.turnID)
		if turnID == "" {
			return nil, fmt.Errorf("codex delivery requires an active turn id")
		}
		params := map[string]any{
			"threadId":       strings.TrimSpace(threadID),
			"expectedTurnId": turnID,
			"input":          startParams["input"],
		}
		if clientID, ok := startParams["clientUserMessageId"].(string); ok && strings.TrimSpace(clientID) != "" {
			params["clientUserMessageId"] = strings.TrimSpace(clientID)
		}
		return params, nil
	default:
		return nil, fmt.Errorf("unsupported codex delivery method %q", target.method)
	}
}

func (d *CodexAppServerDriver) reconcileAckedActiveTurn(threadID string, activeTurnID string) {
	threadID = strings.TrimSpace(threadID)
	activeTurnID = strings.TrimSpace(activeTurnID)
	if threadID == "" || activeTurnID == "" {
		return
	}
	d.mu.Lock()
	defer d.mu.Unlock()
	if strings.TrimSpace(d.threadID) != threadID {
		return
	}
	d.threadStatus = appServerThreadStatus{Type: "active"}
	d.activeTurnID = activeTurnID
	d.activeTurnNonSteerableKind = ""
	d.clearCompactionMutationLocked()
}

func (d *CodexAppServerDriver) clearAckedMissingActiveTurn(threadID string) {
	threadID = strings.TrimSpace(threadID)
	if threadID == "" {
		return
	}
	d.mu.Lock()
	defer d.mu.Unlock()
	if strings.TrimSpace(d.threadID) != threadID {
		return
	}
	d.threadStatus = appServerThreadStatus{Type: "idle"}
	d.activeTurnID = ""
	d.activeTurnNonSteerableKind = ""
	d.clearCompactionMutationLocked()
}

type ackedTurnSteerRaceKind string

const (
	ackedTurnSteerRaceMissing  ackedTurnSteerRaceKind = "missing"
	ackedTurnSteerRaceMismatch ackedTurnSteerRaceKind = "mismatch"
)

func ackedTurnSteerRace(err error) (ackedTurnSteerRaceKind, string, bool) {
	var requestErr *appServerRequestError
	if !errors.As(err, &requestErr) || strings.TrimSpace(requestErr.Method) != "turn/steer" {
		return "", "", false
	}
	message := strings.TrimSpace(requestErr.Message)
	if message == "no active turn to steer" {
		return ackedTurnSteerRaceMissing, "", true
	}

	const mismatchPrefix = "expected active turn id `"
	const mismatchSeparator = "` but found `"
	remainder, ok := strings.CutPrefix(message, mismatchPrefix)
	if !ok {
		return "", "", false
	}
	_, actualWithSuffix, ok := strings.Cut(remainder, mismatchSeparator)
	if !ok {
		return "", "", false
	}
	actualTurnID, ok := strings.CutSuffix(actualWithSuffix, "`")
	actualTurnID = strings.TrimSpace(actualTurnID)
	if !ok || actualTurnID == "" {
		return "", "", false
	}
	return ackedTurnSteerRaceMismatch, actualTurnID, true
}

func ackedActiveTurnNotSteerableKind(err error) (string, bool) {
	if turnKind, ok := activeTurnNotSteerableKind(err); ok {
		return turnKind, true
	}

	var requestErr *appServerRequestError
	if !errors.As(err, &requestErr) || len(requestErr.Data) == 0 {
		return "", false
	}
	var data struct {
		CodexErrorInfo *struct {
			ActiveTurnNotSteerable *struct {
				TurnKind string `json:"turnKind"`
			} `json:"activeTurnNotSteerable,omitempty"`
		} `json:"codexErrorInfo,omitempty"`
	}
	if err := json.Unmarshal(requestErr.Data, &data); err != nil || data.CodexErrorInfo == nil || data.CodexErrorInfo.ActiveTurnNotSteerable == nil {
		return "", false
	}
	turnKind := strings.TrimSpace(data.CodexErrorInfo.ActiveTurnNotSteerable.TurnKind)
	return turnKind, turnKind != ""
}

// DeliverBusyStdinInput delivers a compiler-produced message to a currently active turn.
func (d *CodexAppServerDriver) DeliverBusyStdinInput(ctx context.Context, input drivercontract.RuntimeInput, sessionID string) error {
	threadID := strings.TrimSpace(sessionID)
	if threadID == "" {
		return fmt.Errorf("codex busy delivery requires a thread id")
	}
	if input.Empty() {
		return fmt.Errorf("codex busy delivery requires non-empty input")
	}
	params, err := d.startTurnParams(d.currentLaunchConfig(), input.PromptText(), threadID)
	if err != nil {
		return err
	}
	if params == nil {
		return fmt.Errorf("codex busy delivery produced empty turn params")
	}
	return d.deliverRoutedRuntimeInput(ctx, threadID, params)
}

// DeliverInterruptingRuntimeInput explicitly interrupts the active Codex turn,
// waits until the thread is no longer active, then starts a new turn with the
// supplied authoritative input.
func (d *CodexAppServerDriver) DeliverInterruptingRuntimeInput(ctx context.Context, input drivercontract.RuntimeInput, sessionID string) error {
	threadID := strings.TrimSpace(sessionID)
	if threadID == "" {
		return fmt.Errorf("codex interrupt delivery requires a thread id")
	}
	if input.Empty() {
		return fmt.Errorf("codex interrupt delivery requires non-empty input")
	}
	inputID := strings.TrimSpace(input.ID)
	if inputID == "" {
		return fmt.Errorf("codex interrupt delivery requires a runtime input id")
	}

	threadStatus, activeTurnID, ok := d.currentThreadState(threadID)
	if !ok || (normalizeThreadStatus(threadStatus).Type == "active" && strings.TrimSpace(activeTurnID) == "") {
		readResult, err := d.readThread(ctx, threadID, true)
		if err != nil {
			return err
		}
		d.updateThreadSnapshot(readResult.Thread)
		threadStatus, activeTurnID, ok = d.currentThreadState(threadID)
		if !ok {
			return fmt.Errorf("codex thread %s state is unavailable after refresh", threadID)
		}
	}

	if normalizeThreadStatus(threadStatus).Type == "active" {
		if strings.TrimSpace(activeTurnID) == "" {
			return fmt.Errorf("codex thread %s is active but active turn id is unknown", threadID)
		}
		if err := d.interruptActiveThread(ctx, threadID, activeTurnID); err != nil {
			return err
		}
		if err := d.waitThreadNotActive(ctx, threadID); err != nil {
			return err
		}
	}

	params, err := d.startTurnParams(d.currentLaunchConfig(), input.PromptText(), threadID)
	if err != nil {
		return err
	}
	if params == nil {
		return fmt.Errorf("codex interrupt delivery produced empty turn/start params")
	}
	params["clientUserMessageId"] = inputID
	return d.sendTurnRequest(ctx, threadID, "turn/start", params)
}

// InterruptActiveRun interrupts the active Codex turn while preserving the
// app-server thread/session. It intentionally does not start a replacement turn.
func (d *CodexAppServerDriver) InterruptActiveRun(ctx context.Context, sessionID string) error {
	threadID := strings.TrimSpace(sessionID)
	if threadID == "" {
		return fmt.Errorf("codex interrupt requires a thread id")
	}

	threadStatus, activeTurnID, ok := d.currentThreadState(threadID)
	if !ok || (normalizeThreadStatus(threadStatus).Type == "active" && strings.TrimSpace(activeTurnID) == "") {
		readResult, err := d.readThread(ctx, threadID, true)
		if err != nil {
			return err
		}
		d.updateThreadSnapshot(readResult.Thread)
		threadStatus, activeTurnID, ok = d.currentThreadState(threadID)
		if !ok {
			return fmt.Errorf("codex thread %s state is unavailable after refresh", threadID)
		}
	}

	if normalizeThreadStatus(threadStatus).Type != "active" {
		return nil
	}
	beforeCompletedTurnCount := d.currentCompletedTurnCount(threadID)
	interruptedTurnID, err := d.interruptActiveThreadWithTarget(ctx, threadID, activeTurnID)
	if err != nil {
		return err
	}
	if err := d.waitThreadNotActive(ctx, threadID); err != nil {
		return err
	}
	d.rememberActiveInterrupt(threadID, beforeCompletedTurnCount, interruptedTurnID)
	return nil
}

func (d *CodexAppServerDriver) interruptActiveThread(ctx context.Context, threadID string, activeTurnID string) error {
	_, err := d.interruptActiveThreadWithTarget(ctx, threadID, activeTurnID)
	return err
}

// interruptActiveThreadWithTarget returns the concrete turn ID that the
// successful interrupt targeted. Startup cancellation deliberately returns an
// empty ID because Codex does not emit/persist a turn for that boundary.
func (d *CodexAppServerDriver) interruptActiveThreadWithTarget(ctx context.Context, threadID string, activeTurnID string) (string, error) {
	if strings.TrimSpace(threadID) == "" {
		return "", nil
	}

	interruptedTurnID := strings.TrimSpace(activeTurnID)
	var err error
	if interruptedTurnID == "" {
		err = d.interruptStartup(ctx, threadID)
	} else {
		err = d.interruptTurn(ctx, threadID, interruptedTurnID)
	}
	if err == nil {
		return interruptedTurnID, nil
	}
	if actualTurnID, mismatch := codexInterruptTurnMismatch(err); mismatch && strings.TrimSpace(activeTurnID) != "" {
		actualTurnID = strings.TrimSpace(actualTurnID)
		if actualTurnID == strings.TrimSpace(activeTurnID) {
			return interruptedTurnID, err
		}
		interruptedTurnID = actualTurnID
		d.reconcileAckedActiveTurn(threadID, actualTurnID)
		retryErr := d.interruptTurn(ctx, threadID, actualTurnID)
		if retryErr == nil {
			return interruptedTurnID, nil
		}
		if retryActualTurnID, retryMismatch := codexInterruptTurnMismatch(retryErr); retryMismatch {
			interruptedTurnID = strings.TrimSpace(retryActualTurnID)
			d.reconcileAckedActiveTurn(threadID, retryActualTurnID)
		}
		err = retryErr
	}
	if !codexInterruptErrorCanRaceWithTerminalTurn(err) {
		return interruptedTurnID, err
	}
	inactive, refreshErr := d.refreshThreadInactive(ctx, threadID)
	if refreshErr != nil || !inactive {
		return interruptedTurnID, err
	}
	return interruptedTurnID, nil
}

// codexInterruptTurnMismatch mirrors the current Codex TUI parser. Interrupt
// mismatch messages intentionally omit the backticks used by turn/steer.
func codexInterruptTurnMismatch(err error) (string, bool) {
	var requestErr *appServerRequestError
	if !errors.As(err, &requestErr) || strings.TrimSpace(requestErr.Method) != "turn/interrupt" {
		return "", false
	}
	const mismatchPrefix = "expected active turn id "
	const mismatchSeparator = " but found "
	remainder, ok := strings.CutPrefix(strings.TrimSpace(requestErr.Message), mismatchPrefix)
	if !ok {
		return "", false
	}
	_, actualTurnID, ok := strings.Cut(remainder, mismatchSeparator)
	actualTurnID = strings.TrimSpace(actualTurnID)
	if !ok || actualTurnID == "" {
		return "", false
	}
	return actualTurnID, true
}

func codexInterruptErrorCanRaceWithTerminalTurn(err error) bool {
	var requestErr *appServerRequestError
	if !errors.As(err, &requestErr) {
		return false
	}
	if strings.TrimSpace(requestErr.Method) != "turn/interrupt" {
		return false
	}
	message := strings.TrimSpace(requestErr.Message)
	return message == "endpoint_error" || strings.Contains(message, "no active turn to interrupt")
}

func (d *CodexAppServerDriver) refreshThreadInactive(ctx context.Context, threadID string) (bool, error) {
	readResult, err := d.readThread(ctx, threadID, true)
	if err != nil {
		return false, err
	}
	d.updateThreadSnapshot(readResult.Thread)
	return normalizeThreadStatus(readResult.Thread.Status).Type != "active", nil
}

func (d *CodexAppServerDriver) currentCompletedTurnCount(threadID string) int {
	trimmedThreadID := strings.TrimSpace(threadID)
	d.mu.Lock()
	defer d.mu.Unlock()
	if trimmedThreadID != "" && strings.TrimSpace(d.threadID) != trimmedThreadID {
		return 0
	}
	return d.completedTurnCount
}

func (d *CodexAppServerDriver) rememberActiveInterrupt(threadID string, beforeCompletedTurnCount int, interruptedTurnID string) {
	trimmedThreadID := strings.TrimSpace(threadID)
	if trimmedThreadID == "" {
		return
	}
	d.mu.Lock()
	d.lastActiveInterrupt = activeInterruptSnapshot{
		threadID:                 trimmedThreadID,
		beforeCompletedTurnCount: beforeCompletedTurnCount,
		interruptedTurnID:        strings.TrimSpace(interruptedTurnID),
		recordedAt:               time.Now(),
	}
	d.mu.Unlock()
}

// waitForActiveInterruptTurnForRecovery closes the persistence race between a
// successful turn/interrupt and Codex's thread history. Codex can report the
// thread idle before the terminal interrupted turn is visible in includeTurns;
// both current<target and current==target are therefore provisional while a
// matching recent snapshot exists. Errors/cancellation before materialization
// retain the snapshot for retry. Once the exact terminal boundary is visible,
// Restore compare-clears it before rollback so an ambiguous successful rollback
// cannot make a retry wait for the turn that rollback intentionally removed.
func (d *CodexAppServerDriver) waitForActiveInterruptTurnForRecovery(
	ctx context.Context,
	threadID string,
	targetTurnCount int,
	initial appServerThread,
) (appServerThread, activeInterruptSnapshot, bool, error) {
	trimmedThreadID := strings.TrimSpace(threadID)
	if trimmedThreadID == "" {
		return initial, activeInterruptSnapshot{}, false, nil
	}
	if ctx == nil {
		ctx = context.Background()
	}

	snapshot, relevant, expired := d.activeInterruptSnapshotForRecovery(trimmedThreadID, targetTurnCount, time.Now())
	if !relevant {
		return initial, activeInterruptSnapshot{}, false, nil
	}
	if expired {
		return initial, snapshot, true, fmt.Errorf(
			"codex interrupted turn for thread %s did not materialize before the recovery snapshot expired",
			trimmedThreadID,
		)
	}

	deadline := snapshot.recordedAt.Add(codexActiveInterruptSnapshotTTL)
	waitCtx, cancel := context.WithDeadline(ctx, deadline)
	defer cancel()
	current := initial
	for {
		if activeInterruptTurnMaterialized(d, snapshot, current) {
			return current, snapshot, true, nil
		}
		if !d.activeInterruptSnapshotIsCurrent(snapshot) {
			return current, snapshot, true, fmt.Errorf(
				"codex active interrupt snapshot for thread %s was superseded while waiting for recovery history",
				trimmedThreadID,
			)
		}

		timer := time.NewTimer(codexInterruptPollInterval)
		select {
		case <-waitCtx.Done():
			timer.Stop()
			if ctxErr := ctx.Err(); ctxErr != nil {
				// Caller cancellation is retryable: leave the matching snapshot in
				// place so a later Restore can finish the same interrupt boundary.
				return current, snapshot, true, ctxErr
			}
			// The snapshot's own bounded visibility window expired. Clear only
			// the exact token; a concurrent newer interrupt remains authoritative.
			d.clearActiveInterruptSnapshotIfCurrent(snapshot)
			return current, snapshot, true, fmt.Errorf(
				"timed out waiting for interrupted turn on codex thread %s to materialize",
				trimmedThreadID,
			)
		case <-timer.C:
		}

		readResult, err := d.readThread(waitCtx, trimmedThreadID, true)
		if err != nil {
			if ctxErr := ctx.Err(); ctxErr != nil {
				return current, snapshot, true, ctxErr
			}
			if errors.Is(waitCtx.Err(), context.DeadlineExceeded) {
				d.clearActiveInterruptSnapshotIfCurrent(snapshot)
				return current, snapshot, true, fmt.Errorf(
					"timed out waiting for interrupted turn on codex thread %s to materialize: %w",
					trimmedThreadID,
					err,
				)
			}
			// Transport/read failures retain the snapshot for a later retry.
			return current, snapshot, true, fmt.Errorf(
				"read codex thread %s while waiting for interrupted turn: %w",
				trimmedThreadID,
				err,
			)
		}
		current = readResult.Thread
		d.updateThreadSnapshot(current)
	}
}

func (d *CodexAppServerDriver) activeInterruptSnapshotForRecovery(
	threadID string,
	targetTurnCount int,
	now time.Time,
) (activeInterruptSnapshot, bool, bool) {
	threadID = strings.TrimSpace(threadID)
	d.mu.Lock()
	defer d.mu.Unlock()
	snapshot := d.lastActiveInterrupt
	if threadID == "" || strings.TrimSpace(snapshot.threadID) != threadID ||
		targetTurnCount > snapshot.beforeCompletedTurnCount {
		return activeInterruptSnapshot{}, false, false
	}
	if snapshot.recordedAt.IsZero() || !now.Before(snapshot.recordedAt.Add(codexActiveInterruptSnapshotTTL)) {
		d.lastActiveInterrupt = activeInterruptSnapshot{}
		return snapshot, true, true
	}
	return snapshot, true, false
}

func activeInterruptTurnMaterialized(d *CodexAppServerDriver, snapshot activeInterruptSnapshot, thread appServerThread) bool {
	if d == nil || strings.TrimSpace(thread.ID) != strings.TrimSpace(snapshot.threadID) ||
		normalizeThreadStatus(thread.Status).Type == "active" {
		return false
	}
	interruptedTurnID := strings.TrimSpace(snapshot.interruptedTurnID)
	requiredCompletedTurns := snapshot.beforeCompletedTurnCount
	if interruptedTurnID != "" {
		requiredCompletedTurns++
	}
	if d.completedTurnsFromThread(thread) < requiredCompletedTurns {
		return false
	}
	if interruptedTurnID == "" {
		// Startup cancellation has no turn event/history row. Once the thread is
		// inactive, preserving the pre-interrupt completed history is the complete
		// authoritative boundary.
		return true
	}
	for _, turn := range thread.Turns {
		if strings.TrimSpace(turn.ID) == interruptedTurnID {
			return strings.TrimSpace(turn.Status.Type) != "inProgress"
		}
	}
	return false
}

func sameActiveInterruptSnapshot(left, right activeInterruptSnapshot) bool {
	return strings.TrimSpace(left.threadID) == strings.TrimSpace(right.threadID) &&
		left.beforeCompletedTurnCount == right.beforeCompletedTurnCount &&
		strings.TrimSpace(left.interruptedTurnID) == strings.TrimSpace(right.interruptedTurnID) &&
		left.recordedAt.Equal(right.recordedAt)
}

func (d *CodexAppServerDriver) activeInterruptSnapshotIsCurrent(snapshot activeInterruptSnapshot) bool {
	d.mu.Lock()
	defer d.mu.Unlock()
	return sameActiveInterruptSnapshot(d.lastActiveInterrupt, snapshot)
}

func (d *CodexAppServerDriver) clearActiveInterruptSnapshotIfCurrent(snapshot activeInterruptSnapshot) bool {
	d.mu.Lock()
	defer d.mu.Unlock()
	if !sameActiveInterruptSnapshot(d.lastActiveInterrupt, snapshot) {
		return false
	}
	d.lastActiveInterrupt = activeInterruptSnapshot{}
	return true
}

func (d *CodexAppServerDriver) waitThreadNotActive(ctx context.Context, threadID string) error {
	for {
		readResult, err := d.readThread(ctx, threadID, true)
		if err != nil {
			return err
		}
		d.updateThreadSnapshot(readResult.Thread)
		status := normalizeThreadStatus(readResult.Thread.Status)
		if status.Type != "active" {
			return nil
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(codexInterruptPollInterval):
		}
	}
}

func (d *CodexAppServerDriver) resolveTurnRequestForDirectDelivery(ctx context.Context, text string, sessionID string) (string, map[string]any, error) {
	params, err := d.startTurnParams(d.currentLaunchConfig(), text, sessionID)
	if err != nil {
		return "", nil, err
	}
	if params == nil {
		return "", nil, fmt.Errorf("codex direct delivery produced empty turn params")
	}
	target, err := d.resolveLiveDeliveryTarget(ctx, sessionID)
	if err != nil {
		return "", nil, err
	}
	if target.method == "turn/steer" {
		return target.method, map[string]any{
			"threadId":       strings.TrimSpace(sessionID),
			"expectedTurnId": target.turnID,
			"input":          params["input"],
		}, nil
	}
	return target.method, params, nil
}

// ---------------------------------------------------------------------------
// drivercontract.IdleCompactionDriver
// ---------------------------------------------------------------------------

// StartIdleCompaction attempts to start context compaction if queued.
func (d *CodexAppServerDriver) StartIdleCompaction(ctx context.Context, sessionID string) (bool, error) {
	threadID, ok := d.prepareIdleCompaction(sessionID)
	if !ok {
		return false, nil
	}
	d.beginCompaction(threadID)
	d.mu.Lock()
	generation := d.launchGeneration
	causeTurnID := strings.TrimSpace(d.compactionCauseTurnID)
	d.mu.Unlock()
	slog.Info("codex emergency context compaction start requested",
		"driver", "codex-app-server",
		"launch_generation", generation,
		"thread_id", threadID,
		"cause_turn_id", causeTurnID)
	if err := d.compactThread(ctx, threadID); err != nil {
		// Once the request frame may have reached Codex, rolling the local state
		// back to idle would be unsafe: Codex can still be compacting even though
		// its response was lost. Keep the in-progress marker until an
		// authoritative turn/status notification reconciles the thread. Definite
		// pre-dispatch failures and explicit server rejections did not start the
		// mutation, so they can restore the queued idle attempt immediately.
		if !errors.Is(err, errCodexAppServerTransportAmbiguous) {
			d.restoreQueuedCompaction(threadID, causeTurnID)
		}
		outcome := "definite_failure"
		if errors.Is(err, errCodexAppServerTransportAmbiguous) {
			outcome = "ambiguous"
		}
		slog.Warn("codex emergency context compaction start result",
			"driver", "codex-app-server",
			"launch_generation", generation,
			"thread_id", threadID,
			"cause_turn_id", causeTurnID,
			"outcome", outcome)
		return false, err
	}
	slog.Info("codex emergency context compaction start result",
		"driver", "codex-app-server",
		"launch_generation", generation,
		"thread_id", threadID,
		"cause_turn_id", causeTurnID,
		"outcome", "accepted")
	return true, nil
}

// StartProactiveIdleCompaction starts same-thread manual compaction when Codex
// is authoritatively idle and its latest active-context usage reaches the
// shared proactive threshold. A runtime-native compaction completion suppresses
// this path during the shared cooldown.
func (d *CodexAppServerDriver) StartProactiveIdleCompaction(ctx context.Context, sessionID string) (bool, error) {
	threadID := strings.TrimSpace(sessionID)
	if threadID == "" {
		return false, nil
	}

	d.tokenUsageMu.Lock()
	d.mu.Lock()
	if !d.proactiveCompactionDueLocked(threadID) {
		d.mu.Unlock()
		d.tokenUsageMu.Unlock()
		return false, nil
	}
	activeContextTokens := nonNegativeInt64TokenCount(d.lastTokenUsageByThread[threadID].Last.TotalTokens)
	threshold := d.proactiveCompactionThreshold
	generation := d.launchGeneration
	d.beginCompactionLocked(threadID, codexCompactionSourceLoopProactive, "")
	d.mu.Unlock()
	d.tokenUsageMu.Unlock()

	slog.Info("codex proactive context compaction start requested",
		"driver", "codex-app-server",
		"launch_generation", generation,
		"thread_id", threadID,
		"active_context_tokens", activeContextTokens,
		"threshold_tokens", threshold)
	if err := d.compactThread(ctx, threadID); err != nil {
		ambiguous := errors.Is(err, errCodexAppServerTransportAmbiguous)
		if !ambiguous && !d.restoreRejectedProactiveCompaction(threadID) {
			// A lifecycle event already bound the request to a concrete
			// compaction turn. The RPC error can no longer prove that no
			// mutation occurred.
			ambiguous = true
		}
		outcome := "definite_failure"
		if ambiguous {
			outcome = "ambiguous"
		}
		slog.Warn("codex proactive context compaction start result",
			"driver", "codex-app-server",
			"launch_generation", generation,
			"thread_id", threadID,
			"outcome", outcome)
		return ambiguous, err
	}
	slog.Info("codex proactive context compaction start result",
		"driver", "codex-app-server",
		"launch_generation", generation,
		"thread_id", threadID,
		"outcome", "accepted")
	return true, nil
}