package agent

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"strings"

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

// ContextRecoveryNeeded reports whether the process exited due to a context
// window exceeded condition and needs a continuation restart. If true, it
// also returns the exact wake targets that the continuation must preserve.
func (ap *Process) ContextRecoveryNeeded() (bool, []WakeInfo) {
	ap.mu.Lock()
	defer ap.mu.Unlock()
	return ap.contextWindowExceeded && ap.sleeping, append([]WakeInfo(nil), ap.contextRecoveryWakes...)
}

// SupportsLiveRecovery reports whether this agent process supports live
// recovery for the given strategy.
func (ap *Process) SupportsLiveRecovery(strategy string) bool {
	ap.mu.Lock()
	runtime := strings.TrimSpace(ap.driver.Name())
	_, ok := ap.driver.(drivercontract.RecoveryDriver)
	ap.mu.Unlock()
	if !ok {
		return false
	}
	capability, err := ValidateRecoveryCapability(runtime, strategy)
	if err != nil {
		return false
	}
	return capability.SupportsLiveRestore
}

// IsIdle reports whether the process is between turns and can be safely rewound.
func (ap *Process) IsIdle() bool {
	ap.mu.Lock()
	defer ap.mu.Unlock()
	return ap.isIdle
}

// RestoreRecoveryPoint performs a live recovery point restore on a running
// agent process. The process must be idle; otherwise ErrAgentBusy is returned.
func (ap *Process) RestoreRecoveryPoint(ctx context.Context, strategy string, recoveryRef string) error {
	return ap.restoreRecoveryPoint(ctx, strategy, recoveryRef, nil)
}

// RestoreRecoveryPointAndFinalize performs the same live restore as
// RestoreRecoveryPoint, then runs finalize before releasing the delivery
// admission gate. Dispatchers use this for Codex rollback, whose restored
// app-server thread must be durably detached and retired before a newer wake is
// allowed to enter the old Process.
func (ap *Process) RestoreRecoveryPointAndFinalize(
	ctx context.Context,
	strategy string,
	recoveryRef string,
	finalize func() error,
) error {
	return ap.restoreRecoveryPoint(ctx, strategy, recoveryRef, finalize)
}

func (ap *Process) restoreRecoveryPoint(
	ctx context.Context,
	strategy string,
	recoveryRef string,
	finalize func() error,
) error {
	if ctx == nil {
		ctx = context.Background()
	}
	admissionCtx, cancelAdmission := ap.contextWithDeliveryAdmissionLifetime(ctx)
	defer cancelAdmission()
	// A restore changes the same runtime thread that DeliverForHandoff targets.
	// Holding the outer gate through restore (and optional retirement) gives the
	// two operations one total order; ap.mu alone is insufficient because both
	// paths release it around bounded runtime/store calls.
	if err := ap.acquireDeliveryGate(admissionCtx); err != nil {
		return err
	}
	defer ap.releaseDeliveryGate()

	ap.mu.Lock()
	recoveryDriver, ok := ap.driver.(drivercontract.RecoveryDriver)
	if !ok {
		ap.mu.Unlock()
		return fmt.Errorf("runtime %q does not support live recovery", ap.driver.Name())
	}
	runtimeName := strings.TrimSpace(ap.driver.Name())
	ap.mu.Unlock()

	if _, err := ValidateRecoveryCapability(runtimeName, strategy); err != nil {
		return err
	}

	ap.mu.Lock()
	if ap.stopping || ap.state == processStateStopped {
		ap.mu.Unlock()
		return ErrProcessNotAcceptingDelivery
	}
	if !ap.isIdle {
		ap.mu.Unlock()
		return ErrAgentBusy
	}
	ap.mu.Unlock()
	if err := ap.requireNoLiveDeliveryOwnership(); err != nil {
		return err
	}
	ap.mu.Lock()
	if ap.stopping || ap.state == processStateStopped {
		ap.mu.Unlock()
		return ErrProcessNotAcceptingDelivery
	}
	if !ap.isIdle || ap.hasLocalLiveDeliveryOwnershipLocked() {
		ap.mu.Unlock()
		return ErrAgentBusy
	}
	ap.isIdle = false
	ap.mu.Unlock()

	if err := ap.markResumeSafetyUnsafe(); err != nil {
		ap.mu.Lock()
		ap.isIdle = true
		ap.mu.Unlock()
		return err
	}

	err := recoveryDriver.RestoreRecoveryPoint(admissionCtx, strategy, recoveryRef)

	ap.mu.Lock()
	if err != nil {
		ap.isIdle = true
		ap.mu.Unlock()
		return err
	}
	// requireNoLiveDeliveryOwnership proved the inbox and runtime-owned
	// sets empty while the admission gate was held. Do not assign inbox=nil here:
	// clearing an ownership container as cleanup is exactly the class of bug the
	// gate/proof is intended to make impossible.
	ap.pendingWake = nil
	ap.pendingWakeToken = ""
	ap.pendingDeliveryQueue = nil
	ap.isIdle = true
	ap.mu.Unlock()

	if finalize != nil {
		// Make the restored Process non-accepting before durable cleanup/session
		// retirement, but delay runtime termination until the finalizer has marked
		// the published session retired. Otherwise a fast onExit could race ahead of
		// dispatcher retirement and enter an obsolete continuation generation. The
		// finalizer must use a gate-held/no-public-Stop detach path: a public Stop is
		// a strong completion boundary and would wait for this transition itself.
		transition := ap.prepareStopTransition(false)
		if transition.owner {
			defer func() {
				transition.terminate()
				ap.completeStopTransition()
			}()
			ap.drainLifecycleMutations()
		}
		// Leave the just-restored Codex session fail-closed if durable cleanup or
		// retirement fails. A later launch must not silently resume a thread whose
		// rollback finalization did not complete.
		return finalize()
	}
	ap.markResumeSafetyIfQuiescent()
	return nil
}

// SleepForRecovery stops a Process for detached recovery only after acquiring
// the same gate as concrete delivery admission and proving that no accepted
// queued/pending/uncertain input is owned by this Process or its journal.
func (ap *Process) SleepForRecovery(ctx context.Context) error {
	if ctx == nil {
		ctx = context.Background()
	}
	if err := ap.acquireDeliveryGate(ctx); err != nil {
		return err
	}
	defer ap.releaseDeliveryGate()

	ap.mu.Lock()
	if ap.stopping || ap.state == processStateStopped {
		ap.mu.Unlock()
		return ErrProcessNotAcceptingDelivery
	}
	if !ap.isIdle {
		ap.mu.Unlock()
		return ErrAgentBusy
	}
	ap.mu.Unlock()
	if err := ap.requireNoLiveDeliveryOwnership(); err != nil {
		return err
	}
	ap.mu.Lock()
	if ap.stopping || ap.state == processStateStopped {
		ap.mu.Unlock()
		return ErrProcessNotAcceptingDelivery
	}
	if !ap.isIdle || ap.hasLocalLiveDeliveryOwnershipLocked() {
		ap.mu.Unlock()
		return ErrAgentBusy
	}
	ap.mu.Unlock()

	// The internal stop helper does not acquire deliveryGate. Calling it before releasing the gate
	// makes every later DeliverForHandoff observe the non-accepting state rather
	// than acknowledging a body into a Process that is already being retired.
	ap.stopWhileDeliveryGateHeld(true)
	return nil
}

// hasLocalLiveDeliveryOwnershipLocked reports Process-local owners that make a
// restore or recovery retirement unsafe. The caller holds ap.mu.
func (ap *Process) hasLocalLiveDeliveryOwnershipLocked() bool {
	if len(ap.inbox) > 0 ||
		ap.hasLivePendingRuntimeInputLocked() ||
		strings.TrimSpace(ap.deliveryOwnershipHandoffID) != "" ||
		ap.transientRuntimeRetryPending ||
		ap.transientRuntimeRetryAwaitingTurnEnd {
		return true
	}
	return false
}

// requireNoLiveDeliveryOwnership is the restore/retirement admission proof.
// Callers hold deliveryGate, but deliberately do not hold ap.mu while loading
// the durable store. SetDeliveryOwnershipStore loads the store before taking
// ap.mu, so an ap.mu -> store order here would create an avoidable inversion.
// The store epoch and second local check make a concurrent store attachment
// retry the proof rather than trusting a stale snapshot.
func (ap *Process) requireNoLiveDeliveryOwnership() error {
	for {
		ap.mu.Lock()
		if ap.hasLocalLiveDeliveryOwnershipLocked() {
			ap.mu.Unlock()
			return ErrAgentBusy
		}
		store := ap.deliveryOwnershipStore
		storeEpoch := ap.deliveryOwnershipStoreEpoch
		ap.mu.Unlock()
		if store == nil {
			return nil
		}

		records, err := store.LoadDeliveryOwnership()
		if err != nil {
			return fmt.Errorf("load runtime delivery ownership before recovery: %w", err)
		}
		live := false
		for _, record := range records {
			if IsLiveDeliveryOwnershipState(record.State) {
				live = true
			}
		}

		ap.mu.Lock()
		localLive := ap.hasLocalLiveDeliveryOwnershipLocked()
		storeChanged := ap.deliveryOwnershipStoreEpoch != storeEpoch
		ap.mu.Unlock()
		if localLive {
			return ErrAgentBusy
		}
		if storeChanged {
			continue
		}
		if live {
			return ErrAgentBusy
		}
		return nil
	}
}

// SetOnRecoveryPoint sets or replaces the recovery point callback.
func (ap *Process) SetOnRecoveryPoint(fn func(types.RecoveryPointPreparedEvent)) {
	ap.mu.Lock()
	ap.onRecoveryPoint = fn
	var pending []types.RecoveryPointPreparedEvent
	if fn != nil && len(ap.pendingPreparedRecoveryPoints) > 0 {
		pending = append(pending, ap.pendingPreparedRecoveryPoints...)
		ap.pendingPreparedRecoveryPoints = nil
	}
	ap.mu.Unlock()

	for _, ev := range pending {
		fn(ev)
	}
}

// SetOnRuntimeEvent sets or replaces the runtime event callback.
func (ap *Process) SetOnRuntimeEvent(fn func(RuntimeEvent)) {
	ap.mu.Lock()
	defer ap.mu.Unlock()
	ap.onRuntimeEvent = fn
}

func (ap *Process) emitRuntimeEvent(ev RuntimeEvent) {
	ap.mu.Lock()
	fn := ap.onRuntimeEvent
	ap.mu.Unlock()
	if fn != nil {
		fn(ev)
	}
}

// SetOnResumeSafety sets or replaces the resume safety callback. Version is
// monotonic within this Process generation and sessionID is captured at the
// transition boundary; durable callers combine it with their launch epoch so
// stale callbacks from either an older transition or an older Process cannot
// overwrite a newer unsafe baseline.
func (ap *Process) SetOnResumeSafety(fn func(version uint64, safe bool, sessionID string) error) error {
	ap.mu.Lock()
	ap.onResumeSafety = fn
	ap.resumeSafetyVersion++
	version := ap.resumeSafetyVersion
	known := ap.resumeSafetyKnown
	safe := ap.resumeSafety
	disabled := ap.disableRecovery
	sessionID := strings.TrimSpace(ap.SessionID)
	driver := ap.driver
	isCodex := driver != nil && strings.TrimSpace(driver.Name()) == "codex"
	ap.mu.Unlock()

	// NewProcess starts parsing runtime output before Dispatcher can attach its
	// persistence callbacks. Replay the latest snapshot so an extremely fast
	// SessionInit -> TurnEnd cannot leave durable state stuck at unsafe.
	if fn != nil && known && !disabled && isCodex && sessionID != "" {
		if err := ap.notifyResumeSafety(version, safe, sessionID, fn); err != nil {
			if safe {
				// A failed safe promotion leaves the durable baseline unsafe, which
				// is conservative. Keep the Process usable and retry on a later
				// quiescent transition instead of failing runtime work.
				slog.Warn("persist safe codex resume boundary", "agent_id", ap.AgentID, "error", err)
				return nil
			}
			return err
		}
	}
	return nil
}

// markResumeSafety signals that the codex-app-server session is in a safe
// (or unsafe) state for resume after daemon restart. No-op for other runtimes.
func (ap *Process) markResumeSafety(safe bool) error {
	ap.mu.Lock()
	driver := ap.driver
	if ap.disableRecovery || driver == nil || strings.TrimSpace(driver.Name()) != "codex" {
		ap.mu.Unlock()
		return nil
	}
	sessionID := strings.TrimSpace(ap.SessionID)
	if sessionID == "" {
		ap.mu.Unlock()
		return nil
	}
	ap.resumeSafetyKnown = true
	ap.resumeSafety = safe
	ap.resumeSafetyVersion++
	version := ap.resumeSafetyVersion
	fn := ap.onResumeSafety
	ap.mu.Unlock()

	if fn == nil {
		if safe {
			// The dispatcher installs the persister after Process construction
			// and replays this local snapshot. Safe promotion can wait for that
			// replay; an unsafe mutation may not proceed without it.
			return nil
		}
		return fmt.Errorf("codex resume-safety persister is unavailable")
	}
	if err := ap.notifyResumeSafety(version, safe, sessionID, fn); err != nil {
		if safe {
			// Persisting safe=true is an optimization. Failure leaves the last
			// durable unsafe boundary intact, forcing a fresh start after crash.
			slog.Warn("persist safe codex resume boundary", "agent_id", ap.AgentID, "error", err)
			return nil
		}
		return err
	}
	return nil
}

func (ap *Process) notifyResumeSafety(version uint64, safe bool, sessionID string, fn func(uint64, bool, string) error) error {
	if fn == nil {
		return fmt.Errorf("codex resume-safety persister is unavailable")
	}
	ap.resumeSafetyNotifyMu.Lock()
	defer ap.resumeSafetyNotifyMu.Unlock()

	ap.mu.Lock()
	current := ap.resumeSafetyVersion == version &&
		ap.resumeSafetyKnown &&
		ap.resumeSafety == safe &&
		strings.TrimSpace(ap.SessionID) == strings.TrimSpace(sessionID) &&
		ap.onResumeSafety != nil
	ap.mu.Unlock()
	if !current {
		if safe {
			return nil
		}
		return fmt.Errorf("codex resume-unsafe transition was superseded before persistence")
	}
	return fn(version, safe, sessionID)
}

// markResumeSafetyUnsafe is the synchronous durability prerequisite for every
// Codex mutation that is not already guarded by an atomic delivery-ownership
// write. Returning nil means safe=false was durably accepted for this exact
// Process epoch/version, or that this is a runtime without Codex resume reuse.
func (ap *Process) markResumeSafetyUnsafe() error {
	if err := ap.markResumeSafety(false); err != nil {
		return fmt.Errorf("persist codex resume-unsafe boundary: %w", err)
	}
	return nil
}

// markResumeSafetyIfQuiescent promotes an initialized Codex thread to a safe
// restart boundary only when no active turn or locally/runtime-owned delivery
// remains. It is intentionally re-checking state rather than trusting an older
// SessionInit snapshot: callback delivery and ownership-store restore can race
// between app-server initialization and dispatcher callback installation.
func (ap *Process) markResumeSafetyIfQuiescent() bool {
	ap.mu.Lock()
	driver := ap.driver
	quiescent := !ap.disableRecovery &&
		!ap.stopping &&
		driver != nil &&
		strings.TrimSpace(driver.Name()) == "codex" &&
		strings.TrimSpace(ap.SessionID) != "" &&
		ap.isIdle &&
		len(ap.inbox) == 0 &&
		!ap.hasLivePendingRuntimeInputLocked() &&
		strings.TrimSpace(ap.deliveryOwnershipHandoffID) == ""
	ap.mu.Unlock()
	if quiescent {
		ap.markResumeSafety(true)
	}
	return quiescent
}

// startIdleCompactionAfterTurnEnd attempts to start runtime-owned context
// compaction after a turn_end, if the driver supports it. Returns true if
// compaction was started or its dispatch outcome is ambiguous (the caller must
// skip marking the Process idle in either case).
func (ap *Process) startIdleCompactionAfterTurnEnd() bool {
	ap.mu.Lock()
	disabled := ap.disableRecovery
	hasLiveDelivery := ap.hasLocalLiveDeliveryOwnershipLocked()
	ap.mu.Unlock()
	if disabled || hasLiveDelivery {
		// Compaction is a new mutating request. It must stay behind every older
		// accepted delivery, especially a pending/uncertain correlated input whose
		// history is still being reconciled. The driver's compactQueued flag remains
		// intact and a later quiescent TurnEnd can retry it.
		return false
	}

	driver, ok := ap.driver.(drivercontract.IdleCompactionDriver)
	if !ok {
		return false
	}

	ap.mu.Lock()
	sessionID := ap.SessionID
	stopping := ap.stopping
	ap.mu.Unlock()
	if stopping || strings.TrimSpace(sessionID) == "" {
		return false
	}

	admissionCtx, cancelAdmission := ap.contextWithDeliveryAdmissionLifetime(context.Background())
	defer cancelAdmission()
	ctx, cancel := context.WithTimeout(admissionCtx, idleCompactionTimeout)
	defer cancel()

	// TurnEnd has just promoted the idle thread to safe=true. Persist unsafe
	// synchronously before compactThread can mutate it; a callback/store failure
	// must leave the driver completely untouched.
	if err := ap.markResumeSafetyUnsafe(); err != nil {
		slog.Warn("idle compaction blocked by resume-safety persistence", "agent_id", ap.AgentID, "error", err)
		return false
	}
	started, err := driver.StartIdleCompaction(ctx, sessionID)
	if err != nil {
		slog.Warn("idle compaction start failed", "agent_id", ap.AgentID, "error", err)
		if errors.Is(err, errCodexAppServerTransportAmbiguous) {
			// The request frame may already have started compaction in Codex. Treat
			// that uncertainty as an active mutation: otherwise the TurnEnd caller
			// would make the Process idle and a later quiescence check could persist
			// resumeSafe=true before Codex reports an authoritative terminal/idle
			// boundary.
			ap.mu.Lock()
			ap.isIdle = false
			ap.mu.Unlock()
			return true
		}
		ap.markResumeSafetyIfQuiescent()
		return false
	}
	if !started {
		ap.markResumeSafetyIfQuiescent()
		return false
	}

	ap.mu.Lock()
	ap.isIdle = false
	ap.mu.Unlock()
	return true
}

// startProactiveIdleCompactionAfterTurnEnd attempts to start runtime-owned
// context compaction proactively at an idle boundary when the runtime's own
// context-usage signal crosses a configured threshold. Unlike the overflow
// path, this runs between turns: no continuation retry is injected and the
// next user turn continues from the compacted context. Returns true if
// compaction started (the caller must skip marking the Process idle).
func (ap *Process) startProactiveIdleCompactionAfterTurnEnd() bool {
	proactiveDriver, ok := ap.driver.(drivercontract.ProactiveIdleCompactionDriver)
	if !ok {
		return false
	}
	ap.mu.Lock()
	disabled := ap.disableRecovery
	hasLiveDelivery := ap.hasLocalLiveDeliveryOwnershipLocked()
	sessionID := ap.SessionID
	stopping := ap.stopping
	ap.mu.Unlock()
	if disabled || hasLiveDelivery || stopping || strings.TrimSpace(sessionID) == "" {
		return false
	}
	if !ap.workflowCompaction.allowProactiveCompactionCheck() {
		return false
	}

	admissionCtx, cancelAdmission := ap.contextWithDeliveryAdmissionLifetime(context.Background())
	defer cancelAdmission()
	ctx, cancel := context.WithTimeout(admissionCtx, idleCompactionTimeout)
	defer cancel()

	// Proactive compaction is a new mutating request, so persist unsafe
	// synchronously before summarize rewrites the session history; a store
	// failure must leave the driver untouched.
	if err := ap.markResumeSafetyUnsafe(); err != nil {
		slog.Warn("proactive idle compaction blocked by resume-safety persistence", "agent_id", ap.AgentID, "error", err)
		return false
	}
	started, err := proactiveDriver.StartProactiveIdleCompaction(ctx, sessionID)
	if err != nil {
		slog.Warn("proactive idle compaction start failed", "agent_id", ap.AgentID, "error", err)
		if started {
			// The mutation outcome is ambiguous. Keep the Process non-idle and
			// resume-unsafe until the runtime publishes an authoritative
			// compaction terminal boundary.
			ap.mu.Lock()
			ap.isIdle = false
			ap.mu.Unlock()
			return true
		}
		ap.markResumeSafetyIfQuiescent()
		return false
	}
	if !started {
		ap.markResumeSafetyIfQuiescent()
		return false
	}

	ap.mu.Lock()
	ap.isIdle = false
	ap.mu.Unlock()
	return true
}

var (
	outputTruncationContinuationPrompt  = promptcompiler.OutputTruncationContinuationPrompt()
	contextCompactionContinuationPrompt = promptcompiler.ContextCompactionContinuationPrompt()
)

const (
	contextCompactionConfirmedMetadataKey = "context_compaction_confirmed"
	contextCompactionFailedMetadataKey    = "context_compaction_failed"
	// contextCompactionMaintenanceMetadataKey distinguishes a failed
	// between-turn compaction from overflow recovery of interrupted work.
	contextCompactionMaintenanceMetadataKey = "context_compaction_maintenance"
)

func (ap *Process) noteContextCompactionBoundary(ev drivercontract.ParsedEvent) {
	if !parsedBoolMetadata(ev.Metadata, contextCompactionConfirmedMetadataKey) {
		return
	}
	ap.mu.Lock()
	ap.contextCompactionContinuation = true
	if ev.Kind == consts.ParsedEventKindTurnEnd {
		ap.contextCompactionInProgress = false
	}
	ap.mu.Unlock()
}

func (ap *Process) noteContextCompactionFailure(ev drivercontract.ParsedEvent) {
	if !parsedBoolMetadata(ev.Metadata, contextCompactionFailedMetadataKey) ||
		parsedBoolMetadata(ev.Metadata, contextCompactionMaintenanceMetadataKey) {
		return
	}
	ap.mu.Lock()
	ap.contextWindowExceeded = true
	ap.contextCompactionContinuation = false
	ap.contextCompactionInProgress = false
	if wakes := ap.pendingDeliveryInfosLocked(); len(wakes) > 0 {
		ap.contextRecoveryWakes = wakes
	}
	ap.mu.Unlock()
}

func (ap *Process) noteUnconfirmedContextCompactionTurnEnd(turnEndEv drivercontract.ParsedEvent, recoveryWakes []WakeInfo) {
	if turnEndEv.Kind != consts.ParsedEventKindTurnEnd ||
		parsedBoolMetadata(turnEndEv.Metadata, contextCompactionConfirmedMetadataKey) ||
		parsedBoolMetadata(turnEndEv.Metadata, contextCompactionFailedMetadataKey) {
		return
	}
	ap.mu.Lock()
	if !ap.contextCompactionInProgress || ap.contextCompactionContinuation {
		ap.mu.Unlock()
		return
	}
	ap.contextWindowExceeded = true
	ap.contextCompactionContinuation = false
	ap.contextCompactionInProgress = false
	if len(recoveryWakes) > 0 {
		ap.contextRecoveryWakes = append([]WakeInfo(nil), recoveryWakes...)
	}
	ap.mu.Unlock()
}

func (ap *Process) checkAndTriggerContextCompaction(turnEndEv drivercontract.ParsedEvent, recoveryWakes []WakeInfo) bool {
	ap.mu.Lock()
	disabled := ap.disableRecovery
	stopping := ap.stopping
	exceeded := ap.contextWindowExceeded
	ap.mu.Unlock()
	if disabled || stopping {
		return false
	}

	compactionDriver, hasContextCompaction := ap.driver.(drivercontract.ContextCompactionDriver)
	if !hasContextCompaction {
		return false
	}
	if !exceeded {
		cwDriver, ok := ap.driver.(drivercontract.ContextWindowExceededDriver)
		if !ok || !cwDriver.IsContextWindowExceeded(turnEndEv) {
			return false
		}
		if !compactionDriver.QueueContextCompaction(turnEndEv) {
			return false
		}
	}
	if len(recoveryWakes) > 0 {
		ap.mu.Lock()
		ap.contextRecoveryWakes = append([]WakeInfo(nil), recoveryWakes...)
		ap.mu.Unlock()
	}
	if !ap.startIdleCompactionAfterTurnEnd() {
		ap.mu.Lock()
		ap.contextCompactionContinuation = false
		ap.mu.Unlock()
		return false
	}
	ap.mu.Lock()
	ap.contextWindowExceeded = false
	ap.contextCompactionContinuation = false
	ap.contextCompactionInProgress = true
	ap.mu.Unlock()
	ap.RecordCheckpoint("context_compaction", promptcompiler.RecoveryCheckpoint("context_compaction"))
	ap.emitParsedEvent(drivercontract.ParsedEvent{
		Kind:      consts.ParsedEventKindRuntimeCompacting,
		SessionID: ap.GetSessionID(),
	})
	ap.scheduleInboxRetry(true)
	return true
}

func (ap *Process) injectContextCompactionContinuation() bool {
	ap.mu.Lock()
	pending := ap.contextCompactionContinuation
	if !pending || ap.disableRecovery || ap.stopping {
		ap.mu.Unlock()
		return false
	}
	sessionID := strings.TrimSpace(ap.SessionID)
	stdinPipe := ap.stdinPipe
	structuredDriver, hasStructuredDriver := ap.driver.(drivercontract.RuntimeInputDeliveryDriver)
	ap.isIdle = false
	ap.runtimeErrorDetail = ""
	ap.runtimeErrorEventEmitted = false
	ap.runtimeErrorWakeFailed = false
	ap.mu.Unlock()

	if sessionID == "" || (!hasStructuredDriver && stdinPipe == nil) {
		ap.markContextCompactionContinuationFailed()
		return false
	}
	if err := ap.markResumeSafetyUnsafe(); err != nil {
		slog.Warn("context compaction continuation blocked by resume-safety persistence",
			"agent_id", ap.AgentID, "error", err)
		ap.markContextCompactionContinuationFailed()
		return false
	}
	continuationInput, compileErr := ap.compileGeneratedRuntimeInput(promptcompiler.GeneratedContextCompaction, promptcompiler.EventDeferredContinuation)
	if compileErr != nil {
		ap.markContextCompactionContinuationFailed()
		return false
	}
	if hasStructuredDriver {
		deliveryCtx, cancel := ap.newProtocolDeliveryContextFor(context.Background())
		err := structuredDriver.DeliverRuntimeInput(deliveryCtx, continuationInput, sessionID)
		cancel()
		if err != nil {
			slog.Warn("context compaction continuation protocol delivery failed",
				"agent_id", ap.AgentID, "error", err)
			ap.markContextCompactionContinuationFailed()
			return false
		}
	} else {
		data := ap.driver.EncodeStdinMessage(continuationInput.PromptText(), sessionID)
		if len(data) == 0 || strings.TrimSpace(string(data)) == "{}" {
			ap.markContextCompactionContinuationFailed()
			return false
		}
		if _, err := stdinPipe.Write(data); err != nil {
			slog.Warn("context compaction continuation stdin write failed",
				"agent_id", ap.AgentID, "error", err)
			ap.markContextCompactionContinuationFailed()
			return false
		}
	}

	ap.mu.Lock()
	ap.contextCompactionContinuation = false
	ap.contextWindowExceeded = false
	ap.contextRecoveryWakes = nil
	ap.contextCompactionInProgress = false
	ap.mu.Unlock()

	ap.markRuntimeHandoffSucceeded()
	ap.recordGeneratedPromptCompileMetrics(
		promptcompiler.EventDeferredContinuation,
		promptcompiler.GeneratedContextCompaction,
	)
	ap.RecordCheckpoint("context_compaction_continuation", promptcompiler.RecoveryCheckpoint("context_compaction_continuation"))
	ap.emitParsedEvent(newPromptInputParsedEvent(
		contextCompactionContinuationPrompt,
		consts.PromptKindDeferredContinuation,
		ap.driver.Name(),
		promptcompiler.EventDeferredContinuation,
		sessionID,
	))
	return true
}

func (ap *Process) markContextCompactionContinuationFailed() {
	ap.mu.Lock()
	ap.contextWindowExceeded = true
	ap.contextCompactionContinuation = false
	ap.contextCompactionInProgress = false
	ap.mu.Unlock()
}

func (ap *Process) checkAndTriggerOutputTruncationContinuation(turnEndEv drivercontract.ParsedEvent) bool {
	ap.mu.Lock()
	if ap.disableRecovery || ap.stopping {
		ap.outputTruncationDetected = false
		ap.mu.Unlock()
		return false
	}
	detected := ap.outputTruncationDetected
	sessionID := strings.TrimSpace(ap.SessionID)
	stdinPipe := ap.stdinPipe
	ap.mu.Unlock()

	if truncationDriver, ok := ap.driver.(drivercontract.OutputTruncationDriver); ok &&
		truncationDriver.IsOutputTruncated(turnEndEv) {
		detected = true
	}
	if !detected {
		return false
	}
	if sessionID == "" || stdinPipe == nil {
		return false
	}
	if err := ap.markResumeSafetyUnsafe(); err != nil {
		slog.Warn("output truncation continuation blocked by resume-safety persistence",
			"agent_id", ap.AgentID, "error", err)
		return false
	}
	continuationInput, compileErr := ap.compileGeneratedRuntimeInput(promptcompiler.GeneratedOutputTruncation, promptcompiler.EventDeferredContinuation)
	if compileErr != nil {
		return false
	}
	data := ap.driver.EncodeStdinMessage(continuationInput.PromptText(), sessionID)
	if len(data) == 0 || strings.TrimSpace(string(data)) == "{}" {
		return false
	}
	if _, err := stdinPipe.Write(data); err != nil {
		slog.Warn("output truncation continuation stdin write failed",
			"agent_id", ap.AgentID, "error", err)
		return false
	}

	ap.mu.Lock()
	ap.outputTruncationDetected = false
	ap.runtimeErrorDetail = ""
	ap.runtimeErrorEventEmitted = false
	ap.runtimeErrorWakeFailed = false
	ap.isIdle = false
	ap.mu.Unlock()

	ap.markRuntimeHandoffSucceeded()
	ap.recordGeneratedPromptCompileMetrics(
		promptcompiler.EventDeferredContinuation,
		promptcompiler.GeneratedOutputTruncation,
	)
	ap.RecordCheckpoint("output_continuation", promptcompiler.RecoveryCheckpoint("output_continuation"))
	ap.emitParsedEvent(newPromptInputParsedEvent(
		outputTruncationContinuationPrompt,
		consts.PromptKindDeferredContinuation,
		ap.driver.Name(),
		promptcompiler.EventDeferredContinuation,
		sessionID,
	))
	return true
}

// checkAndTriggerContextRecovery checks whether the context window was exceeded
// during the current turn (either via error events or the turn_end stop_reason).
// If so, and the driver does not have its own compaction mechanism, it triggers
// a graceful session restart via Sleep(). Returns true if recovery was triggered.
func (ap *Process) checkAndTriggerContextRecovery(turnEndEv drivercontract.ParsedEvent, recoveryWakes []WakeInfo) bool {
	ap.mu.Lock()
	disabled := ap.disableRecovery
	ap.mu.Unlock()
	if disabled {
		ap.resetContextWindowFlag()
		return false
	}

	// Skip if the driver owns its compaction lifecycle internally (e.g.
	// codex-app-server). ContextCompactionDriver runtimes attempted same-session
	// compaction first; if that did not start, fall through to fresh-session
	// recovery instead of dropping the interrupted work.
	if _, hasCompaction := ap.driver.(drivercontract.IdleCompactionDriver); hasCompaction {
		if _, hasContextCompaction := ap.driver.(drivercontract.ContextCompactionDriver); hasContextCompaction {
			// Continue below.
		} else {
			ap.resetContextWindowFlag()
			return false
		}
	}

	// Check if context window was exceeded via error events during the turn.
	ap.mu.Lock()
	exceeded := ap.contextWindowExceeded
	stopping := ap.stopping
	ap.mu.Unlock()

	if !exceeded {
		// Also check the stop_reason on the turn_end event itself.
		cwDriver, ok := ap.driver.(drivercontract.ContextWindowExceededDriver)
		if !ok {
			ap.resetContextWindowFlag()
			return false
		}
		if !cwDriver.IsContextWindowExceeded(turnEndEv) {
			ap.resetContextWindowFlag()
			return false
		}
	}

	if stopping {
		// Don't reset the flag — the process is already shutting down
		// and the next turn won't execute.
		return false
	}

	slog.Info("context window exceeded, triggering graceful restart",
		"agent_id", ap.AgentID,
		"stop_reason", turnEndEv.StopReason,
	)

	// Preserve the exact wake targets captured before TurnEnd terminalized the
	// original obligations. The synthetic recovery wake starts a new lifecycle.
	ap.mu.Lock()
	ap.contextWindowExceeded = true
	if len(recoveryWakes) > 0 {
		ap.contextRecoveryWakes = append([]WakeInfo(nil), recoveryWakes...)
	}
	ap.mu.Unlock()

	ap.RecordCheckpoint("context_recovery", promptcompiler.RecoveryCheckpoint("context_recovery"))
	ap.emitRuntimeEvent(RuntimeEvent{
		EventClass: consts.RuntimeEventClassContextWindowExceeded,
		Detail:     promptcompiler.ContextRecoveryRestartDetail(),
	})

	// TurnEnd handling already holds deliveryGate while it checks this boundary.
	// Use the non-reentrant internal stop so context recovery preserves the
	// session entry without deadlocking through public Sleep's admission barrier.
	// The next message delivery creates a fresh session (reading MEMORY.md +
	// state/*) instead of removing the session entirely.
	ap.stopWhileDeliveryGateHeld(true)
	return true
}

// resetContextWindowFlag clears the contextWindowExceeded flag. Called when
// context recovery is not triggered (e.g. driver has compaction, or the turn
// ended normally) to prevent stale flags from causing false recoveries in
// subsequent turns.
func (ap *Process) resetContextWindowFlag() {
	ap.mu.Lock()
	ap.contextWindowExceeded = false
	ap.contextRecoveryWakes = nil
	ap.contextCompactionContinuation = false
	ap.contextCompactionInProgress = false
	ap.mu.Unlock()
}
func (ap *Process) recordSpawnPreparedRecoveryPoint(point *drivercontract.PreparedRecoveryPoint) {
	if point == nil {
		return
	}
	ev := types.RecoveryPointPreparedEvent{
		AgentID:     ap.AgentID,
		MessageID:   strings.TrimSpace(point.MessageID),
		ThreadID:    strings.TrimSpace(point.ThreadID),
		Strategy:    strings.TrimSpace(point.Strategy),
		RecoveryRef: strings.TrimSpace(point.RecoveryRef),
	}
	if ev.MessageID == "" || ev.ThreadID == "" || ev.Strategy == "" || ev.RecoveryRef == "" {
		return
	}

	ap.mu.Lock()
	onRecoveryPoint := ap.onRecoveryPoint
	if onRecoveryPoint == nil {
		ap.pendingPreparedRecoveryPoints = append(ap.pendingPreparedRecoveryPoints, ev)
		ap.mu.Unlock()
		return
	}
	ap.mu.Unlock()

	onRecoveryPoint(ev)
}

func prepareSpawnRecoveryPoint(driver drivercontract.Driver, messageID, threadID string) *drivercontract.PreparedRecoveryPoint {
	trimmedMessageID := strings.TrimSpace(messageID)
	trimmedThreadID := strings.TrimSpace(threadID)
	if trimmedMessageID == "" || trimmedThreadID == "" {
		return nil
	}

	strategy, recoveryRef, supported, err := prepareDriverRecoveryPoint(driver, trimmedMessageID)
	if err != nil || !supported {
		return nil
	}
	if strings.TrimSpace(strategy) == "" || strings.TrimSpace(recoveryRef) == "" {
		return nil
	}

	return &drivercontract.PreparedRecoveryPoint{
		MessageID:   trimmedMessageID,
		ThreadID:    trimmedThreadID,
		Strategy:    strings.TrimSpace(strategy),
		RecoveryRef: strings.TrimSpace(recoveryRef),
	}
}

func prepareDriverRecoveryPoint(driver drivercontract.Driver, messageID string) (strategy string, recoveryRef string, supported bool, err error) {
	trimmedMessageID := strings.TrimSpace(messageID)
	if trimmedMessageID == "" {
		return "", "", false, nil
	}
	if messageBound, ok := driver.(drivercontract.MessageBoundRecoveryDriver); ok {
		return messageBound.PrepareRecoveryPointForMessage(trimmedMessageID)
	}
	recoveryDriver, ok := driver.(drivercontract.RecoveryDriver)
	if !ok {
		return "", "", false, nil
	}
	return recoveryDriver.PrepareRecoveryPoint()
}

func (ap *Process) captureRecoveryPoint(messageID string) {
	trimmedMessageID := strings.TrimSpace(messageID)
	if trimmedMessageID == "" {
		return
	}

	ap.mu.Lock()
	recoveryDriver, hasRecoveryDriver := ap.driver.(drivercontract.RecoveryDriver)
	messageBoundDriver, hasMessageBoundDriver := ap.driver.(drivercontract.MessageBoundRecoveryDriver)
	threadID := strings.TrimSpace(ap.SessionID)
	agentID := ap.AgentID
	onRecoveryPoint := ap.onRecoveryPoint
	ap.mu.Unlock()

	if (!hasRecoveryDriver && !hasMessageBoundDriver) || threadID == "" || onRecoveryPoint == nil {
		return
	}

	var strategy, recoveryRef string
	var supported bool
	var err error
	if hasMessageBoundDriver {
		strategy, recoveryRef, supported, err = messageBoundDriver.PrepareRecoveryPointForMessage(trimmedMessageID)
	} else {
		strategy, recoveryRef, supported, err = recoveryDriver.PrepareRecoveryPoint()
	}
	if err != nil || !supported {
		return
	}

	if strings.TrimSpace(strategy) == "" || strings.TrimSpace(recoveryRef) == "" {
		return
	}

	ap.recordTransientRuntimeRetryPoint(trimmedMessageID, strategy, recoveryRef)

	onRecoveryPoint(types.RecoveryPointPreparedEvent{
		AgentID:     agentID,
		MessageID:   trimmedMessageID,
		ThreadID:    threadID,
		Strategy:    strategy,
		RecoveryRef: recoveryRef,
	})
}