package dispatcher

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"sort"
	"strings"
	"sync"
	"time"

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

const maxDeferredReplayAttempts = 3

type deferredContinuationMessage struct {
	MessageID                string                    `json:"message_id"`
	TargetKey                string                    `json:"target_key,omitempty"`
	TargetSeq                int64                     `json:"target_seq,omitempty"`
	ChannelType              string                    `json:"channel_type"`
	ChannelName              string                    `json:"channel_name"`
	ChannelDisplayName       string                    `json:"channel_display_name,omitempty"`
	ParentChannelType        string                    `json:"parent_channel_type"`
	ParentChannelName        string                    `json:"parent_channel_name"`
	ParentChannelDisplayName string                    `json:"parent_channel_display_name,omitempty"`
	AuthorProfile            types.AuthorProfile       `json:"author_profile,omitempty"`
	Content                  string                    `json:"content"`
	Timestamp                string                    `json:"timestamp"`
	MessageType              string                    `json:"message_type,omitempty"`
	InterruptPolicy          string                    `json:"interruptPolicy,omitempty"`
	Source                   types.ContributionSource  `json:"source,omitempty,omitzero"`
	Metadata                 map[string]any            `json:"metadata,omitempty"`
	ActiveTaskTarget         types.ActiveTaskTarget    `json:"activeTaskTarget,omitempty,omitzero"`
	TeamContexts             []types.TeamContext       `json:"teamContexts,omitempty"`
	AttachmentIDs            string                    `json:"attachment_ids,omitempty"`
	AttachmentNames          string                    `json:"attachment_names,omitempty"`
	Attachments              []types.MessageAttachment `json:"attachments,omitempty"`
}

// oldDaemonDeferredPayloadCompat is the retained reader boundary for deferred
// delivery records written by an older daemon before author_profile existed.
// New deferred payloads are authored through deferredContinuationMessage and do
// not write these fields.
type oldDaemonDeferredPayloadCompat struct {
	LegacyActorType string `json:"sender_type"`
	LegacyAuthor    string `json:"sender_name"`
}

func bindingWithDeferredAgentTeamMessage(binding conversationBinding, msg types.InboxMessage) conversationBinding {
	target := strings.TrimSpace(msg.Target)
	if target == "" {
		return binding
	}

	var metadata map[string]any
	if len(msg.Metadata) > 0 {
		metadata = make(map[string]any, len(msg.Metadata))
		for key, value := range msg.Metadata {
			metadata[key] = value
		}
	}

	msg, workflowContext := normalizeInboxWorkflowContext(msg)
	binding.targetID = target
	binding.meta = &types.AgentTeamMeta{
		ID:                       strings.TrimSpace(msg.MessageID),
		TargetKey:                strings.TrimSpace(msg.TargetKey),
		TargetSeq:                msg.TargetSeq,
		Target:                   target,
		TargetChannelType:        strings.TrimSpace(msg.ChannelType),
		TargetChannelName:        strings.TrimSpace(msg.ChannelName),
		ChannelDisplayName:       strings.TrimSpace(msg.ChannelDisplayName),
		ParentTargetChannelType:  strings.TrimSpace(msg.ParentChannelType),
		ParentTargetChannelName:  strings.TrimSpace(msg.ParentChannelName),
		ParentChannelDisplayName: strings.TrimSpace(msg.ParentChannelDisplayName),
		AgentID:                  strings.TrimSpace(binding.serverAgentID),
		AuthorProfile:            msg.AuthorProfile,
		Timestamp:                strings.TrimSpace(msg.Timestamp),
		MessageType:              strings.TrimSpace(msg.MessageType),
		InterruptPolicy:          strings.TrimSpace(msg.InterruptPolicy),
		Source:                   msg.Source,
		Metadata:                 types.MetadataWithoutWorkflowContext(metadata),
		WorkflowContext:          workflowContext,
		ActiveTaskTarget:         msg.ActiveTaskTarget.Normalize(),
		TeamContexts:             append([]types.TeamContext(nil), msg.TeamContexts...),
		AttachmentIDs:            strings.TrimSpace(msg.AttachmentIDs),
		AttachmentNames:          strings.TrimSpace(msg.AttachmentNames),
		Attachments:              msg.Attachments(),
	}
	return binding
}

// bindInitialDeferredLaunchMessage preloads the exact message context into
// both prompt construction and session state before Spawn. Some runtimes can
// execute the initial prompt before NewProcess returns, so the later acknowledgement
// observer is necessarily too late for this one boundary. Callers must use
// this helper only when msg is truly the Spawn initial prompt; no-turn durable
// handoff/reconciliation launches intentionally keep their context empty.
func bindInitialDeferredLaunchMessage(sess *interactiveSession, binding conversationBinding, msg types.InboxMessage) conversationBinding {
	msg, _ = normalizeInboxWorkflowContext(msg)
	binding = bindingWithDeferredAgentTeamMessage(binding, msg)
	if sess == nil {
		return binding
	}
	sess.mu.Lock()
	sess.binding = binding
	sess.mu.Unlock()
	return binding
}

func runtimeSupportsDeferredStdin(runtimeName string) bool {
	rt := runtimeconfig.RuntimeType(strings.TrimSpace(runtimeName))
	var (
		driver drivercontract.Driver
		err    error
	)
	if hook, ok := overrideNewAgentDriver(); ok {
		driver, err = hook(rt)
	} else {
		driver, err = agent.NewDriver(rt)
	}
	if err != nil {
		return false
	}
	return driver.SupportsStdinNotification()
}

func normalizeDeferredReplyThreadShortID(value string) string {
	trimmed := strings.TrimSpace(value)
	if strings.HasPrefix(strings.ToLower(trimmed), "thread-") {
		return trimmed[len("thread-"):]
	}
	return trimmed
}

func deferredPromptTarget(channelType, channelName, parentChannelType, parentChannelName string) string {
	if channelType == "thread" && parentChannelName != "" {
		shortID := normalizeDeferredReplyThreadShortID(channelName)
		if parentChannelType == "dm" {
			return fmt.Sprintf("dm:@%s:%s", parentChannelName, shortID)
		}
		return fmt.Sprintf("#%s:%s", parentChannelName, shortID)
	}
	if channelType == "dm" {
		return "dm:@" + channelName
	}
	return "#" + channelName
}

func deferredInboxMessage(raw json.RawMessage) (types.InboxMessage, bool) {
	var payload deferredContinuationMessage
	if err := json.Unmarshal(raw, &payload); err != nil {
		return types.InboxMessage{}, false
	}
	if strings.TrimSpace(payload.ChannelType) == "" || strings.TrimSpace(payload.ChannelName) == "" {
		return types.InboxMessage{}, false
	}
	authorProfile := payload.AuthorProfile
	if strings.TrimSpace(authorProfile.ActorType) == "" && strings.TrimSpace(authorProfile.DisplayName) == "" {
		var compat oldDaemonDeferredPayloadCompat
		if err := json.Unmarshal(raw, &compat); err == nil {
			authorProfile = types.AuthorProfile{
				ActorType:   strings.TrimSpace(compat.LegacyActorType),
				DisplayName: strings.TrimSpace(compat.LegacyAuthor),
				TargetID:    strings.TrimSpace(compat.LegacyAuthor),
			}
		}
	}
	msg := types.InboxMessage{
		MessageID:                strings.TrimSpace(payload.MessageID),
		TargetKey:                strings.TrimSpace(payload.TargetKey),
		TargetSeq:                payload.TargetSeq,
		Target:                   deferredPromptTarget(payload.ChannelType, payload.ChannelName, payload.ParentChannelType, payload.ParentChannelName),
		ChannelType:              strings.TrimSpace(payload.ChannelType),
		ChannelName:              strings.TrimSpace(payload.ChannelName),
		ChannelDisplayName:       strings.TrimSpace(payload.ChannelDisplayName),
		ParentChannelType:        strings.TrimSpace(payload.ParentChannelType),
		ParentChannelName:        strings.TrimSpace(payload.ParentChannelName),
		ParentChannelDisplayName: strings.TrimSpace(payload.ParentChannelDisplayName),
		AuthorProfile:            authorProfile,
		Content:                  payload.Content,
		Timestamp:                strings.TrimSpace(payload.Timestamp),
		MessageType:              strings.TrimSpace(payload.MessageType),
		InterruptPolicy:          strings.TrimSpace(payload.InterruptPolicy),
		Source:                   payload.Source,
		Metadata:                 payload.Metadata,
		ActiveTaskTarget:         payload.ActiveTaskTarget.Normalize(),
		TeamContexts:             append([]types.TeamContext(nil), payload.TeamContexts...),
		AttachmentIDs:            strings.TrimSpace(payload.AttachmentIDs),
		AttachmentNames:          strings.TrimSpace(payload.AttachmentNames),
		AttachmentList:           payload.Attachments,
	}
	msg, _ = normalizeInboxWorkflowContext(msg)
	return msg, true
}

func firstReplayPendingIndex(entries []deferredDeliveryEntry) int {
	for i := range entries {
		if entries[i].ReplayPending {
			return i
		}
	}
	return -1
}

func firstReplayAvailableIndex(entries []deferredDeliveryEntry) int {
	for i := range entries {
		if !entries[i].ReplayPending {
			return i
		}
	}
	return -1
}

func contextRecoveryReplayIndex(entries []deferredDeliveryEntry) int {
	for i := range entries {
		if entries[i].ReplayPending {
			continue
		}
		msg, ok := deferredInboxMessage(entries[i].Message)
		if ok && isContextRecoveryMessage(msg) {
			return i
		}
	}
	return -1
}

func removeDeferredDeliveryAt(entries []deferredDeliveryEntry, index int) []deferredDeliveryEntry {
	next := make([]deferredDeliveryEntry, 0, len(entries)-1)
	next = append(next, entries[:index]...)
	next = append(next, entries[index+1:]...)
	return next
}

func removeDeferredDeliveryByMessageID(entries []deferredDeliveryEntry, messageID string) []deferredDeliveryEntry {
	messageID = strings.TrimSpace(messageID)
	if messageID == "" {
		return entries
	}
	next := make([]deferredDeliveryEntry, 0, len(entries))
	for _, entry := range entries {
		if strings.TrimSpace(entry.MessageID) == messageID || deferredMessageID(entry.Message) == messageID {
			continue
		}
		next = append(next, entry)
	}
	return next
}

func setDeferredDeliveryReplayPending(entries []deferredDeliveryEntry, messageID string, pending bool) bool {
	messageID = strings.TrimSpace(messageID)
	if messageID == "" {
		return false
	}
	for index := range entries {
		if strings.TrimSpace(entries[index].MessageID) == messageID || deferredMessageID(entries[index].Message) == messageID {
			entries[index].ReplayPending = pending
			return true
		}
	}
	return false
}

func (d *Dispatcher) removeContextRecoveryMarker(workDir, messageID string) error {
	if d != nil && d.contextRecoveryMarkerRemoval != nil {
		return d.contextRecoveryMarkerRemoval(workDir, messageID)
	}
	return updateSessionResumeState(workDir, func(latest *sessionResumeState) error {
		latest.DeferredDeliveries = removeDeferredDeliveryByMessageID(latest.DeferredDeliveries, messageID)
		return nil
	})
}

func isContextRecoveryMessage(msg types.InboxMessage) bool {
	return strings.TrimSpace(msg.MessageType) == "context_recovery"
}

func (d *Dispatcher) attachRecoveryReporter(sess *interactiveSession, process *agent.Process) {
	if ch, ok := d.serverConnections.ServerConnection(sess.binding.channelID); ok {
		if reporter, ok := ch.(types.EventReporter); ok {
			process.SetOnRecoveryPoint(func(ev types.RecoveryPointPreparedEvent) {
				reporter.Emit(ev)
			})
		}
	}
}

func (d *Dispatcher) configureResumeSafety(sess *interactiveSession, process *agent.Process, epoch uint64) error {
	serverURL := d.serverURL
	return process.SetOnResumeSafety(func(version uint64, safe bool, sessionID string) error {
		applied, err := PersistResumeSafety(sess.workDir, serverURL, sessionID, epoch, version, safe)
		if err != nil {
			return fmt.Errorf("persist resume safety: %w", err)
		}
		if !applied {
			return fmt.Errorf("persist resume safety rejected stale or non-quiescent process transition (epoch=%d version=%d safe=%t)", epoch, version, safe)
		}
		return nil
	})
}

func (d *Dispatcher) persistStartState(launch *processLaunchContext) (uint64, error) {
	sess := launch.sess
	startCfg := launch.startCfg
	fingerprint := strings.TrimSpace(launch.fingerprint)
	if fingerprint == "" && startCfg.CompiledPrompt != nil {
		fingerprint = instructionsFingerprint(startCfg.CompiledPrompt.LaunchSurface("baseInstructions"), startCfg.CompiledPrompt.LaunchSurface("developerInstructions"))
	}
	if sess.runtime.Type == runtimeconfig.TypeCodex && fingerprint == "" {
		state, err := loadSessionResumeState(sess.workDir)
		if err == nil && state.LastStartConfig != nil {
			fingerprint = strings.TrimSpace(state.LastStartConfig.InstructionsFingerprint)
		}
	}
	identityName, identityDisplayName := normalizeAgentIdentityNames(
		launch.binding.serverAgentName,
		launch.binding.serverAgentDisplayName,
	)
	if identityName == "" {
		sess.mu.Lock()
		identityName, identityDisplayName = normalizeAgentIdentityNames(sess.agent.Name, "")
		sess.mu.Unlock()
	}
	startConfig := &types.AgentStartConfig{
		Name:                    identityName,
		DisplayName:             identityDisplayName,
		Runtime:                 string(sess.runtime.Type),
		PermissionMode:          startCfg.PermissionMode,
		RuntimeDebug:            runtimeDebugConfigFromEnabled(startCfg.RuntimeDebugEnabled),
		Model:                   startCfg.Model,
		ReasoningEffort:         startCfg.ReasoningEffort,
		InstructionsFingerprint: fingerprint,
		SessionID:               startCfg.SessionID,
	}
	return persistStartConfigStateForLaunch(
		sess.workDir,
		string(sess.runtime.Type),
		d.serverURL,
		startCfg.SessionID,
		startConfig,
		!launch.authoritativeStartIdentity,
		launch.runtimePolicy,
	)
}

func (d *Dispatcher) handleProcessExit(sess *interactiveSession, ap *agent.Process) {
	// The Exec authorization is exact to this Process generation. Revoke it before
	// registry persistence or session lifecycle locks: either can block, while
	// detached descendants may still know the departed Process's credential.
	// The session MCP token remains session-shared and is revoked only after the
	// current-Process fence below.
	d.revokeExecProcessAuthorization(ap)

	d.unregisterRuntimeProcess(ap)
	processSleeping := ap.IsSleeping()
	sess.mu.Lock()
	wasCurrentProcess := sess.process == ap
	// Do NOT clear sess.process here. The terminal stopped/idle turn-state
	// publication below must still observe the current generation
	// (sess.process == ap) so the synchronous AgentTeamReporterSink resolves idle
	// instead of an empty snapshot (process == nil). The pointer is cleared
	// after publication under the identity fence, so a replacement committed
	// during the report is never cleared.
	sessionKey := sess.sessionKey
	if strings.TrimSpace(sessionKey) == "" {
		sessionKey = sess.agentID
	}
	resetting := sess.resetting
	ephemeral := sess.ephemeral
	durableDelivery := sess.durableDelivery
	binding := sess.binding
	suppressCheckpoints := sess.suppressCheckpoints
	terminalReportOwner := false
	if wasCurrentProcess && !sess.retired.Load() && !sess.terminalReportClaimed {
		sess.terminalReportClaimed = true
		sess.terminalReportDone = make(chan struct{})
		sess.terminalReportWillStop = !resetting && !ephemeral && !processSleeping
		terminalReportOwner = true
	}
	terminalReportDone := sess.terminalReportDone
	sess.mu.Unlock()

	// A delayed onExit from an older committed generation must not revoke the
	// current session token or start another continuation after replacement.
	if !wasCurrentProcess {
		slog.Warn("stale process exit dropped (replacement already committed)",
			"agent_id", ap.AgentID,
		)
		return
	}
	revokeSessionMCPToken(sess)

	defer d.publishSessionStateChanged(ap.AgentID)
	sess.mu.Lock()
	restartRetirementInFlight := sess.restartRetirementInFlight
	sess.mu.Unlock()
	retired := sess.retired.Load()
	if terminalReportOwner {
		if !resetting && !ephemeral && (!retired || restartRetirementInFlight) {
			// Publish the terminal stopped/idle turn-state WHILE sess.process is
			// still the current generation, so the AgentTeamReporterSink resolves idle
			// from the stopping/stopped Process (never an empty snapshot).
			d.reportExitStatus(binding.channelID, ap, binding.serverAgentID)
			slog.Info("current-generation exit published terminal idle",
				"agent_id", ap.AgentID,
				"channel_id", binding.channelID,
			)
		}
		if terminalReportDone != nil {
			close(terminalReportDone)
		}
		// Generation-fenced clear: only clear the old pointer if it is still the
		// exiting Process. A replacement that committed during the terminal report
		// is live and must not be cleared. When another path (retirement/restart)
		// owns the terminal report, that owner publishes stopped while this
		// Process is still current and clears it under its own fence.
		sess.mu.Lock()
		if sess.process == ap {
			sess.process = nil
		}
		sess.mu.Unlock()
	}
	if retired || restartRetirementInFlight {
		return
	}
	if resetting {
		d.fenceSessionMemoryWatcherForFinalRemoval(sess)
		d.stopMemoryWatcherAndRetry(sess)
		d.sessionManager.RemoveIfCurrent(sessionKey, sess)
		return
	}
	// The exited Process is the stable writer barrier. Finish any pending
	// legacy merge before an automatic continuation can Spawn and write the
	// resolved profile.
	d.finalizePendingLegacyPreferencesMigration(context.Background(), sess, true)

	if ephemeral && !durableDelivery {
		d.fenceSessionMemoryWatcherForFinalRemoval(sess)
		d.stopMemoryWatcherAndRetry(sess)
		d.sessionManager.RemoveIfCurrent(sessionKey, sess)
		return
	}

	if !ephemeral && !suppressCheckpoints {
		if ap.IsSleeping() {
			ap.RecordCheckpoint("flush", "Session paused. Recover needed context to continue current or pending work from this workspace; start with the configured memory index and the recovery files named in the Workspace & Memory instructions, then inspect files, git state, task context, or recent history if needed.")
		} else {
			ap.RecordCheckpoint("flush", "Process stopped. Recover needed context to continue current or pending work from this workspace; start with the configured memory index and the recovery files named in the Workspace & Memory instructions, then inspect files, git state, task context, or recent history if needed.")
		}
	}

	// Every unexpected-exit continuation competes in the same session-key start
	// state machine as a concurrently arriving delivery. Whichever path reserves
	// the generation first owns all launch publication and durable handoff work;
	// the loser must not launch a second Process for the same q1.
	startReservation, _, ownsStart := d.sessionManager.BeginStart(sessionKey, sess)
	if !ownsStart {
		sess.mu.Lock()
		restartRetirementInFlight = sess.restartRetirementInFlight
		sess.mu.Unlock()
		if restartRetirementInFlight {
			return
		}
		// The Process is terminal even though another start generation currently
		// owns manager publication. Permanently fence the old session's watcher
		// before returning: a failed replacement may otherwise restore a watcher
		// after this callback has set process=nil, while this callback cannot win a
		// second start generation to observe and drain that late publication.
		//
		// The fence/drain and replacement publication share sess.mu. Whichever
		// wins is therefore total: an already-published watcher is drained here,
		// while an in-flight restore is rejected and synchronously self-cleans.
		fenceSessionMemoryWatcherPublication(sess)
		d.stopMemoryWatcherAndRetry(sess)
		if ap.ShouldAutoContinueAcceptedDeliveries() {
			if _, live, err := oldestLiveRuntimeDeliveryOwnership(sess.workDir); err != nil || live {
				if err == nil {
					err = errors.New("runtime delivery ownership continuation is waiting for another session start")
				}
				d.scheduleRuntimeDeliveryOwnershipContinuationRetry(sess, err)
			}
		}
		return
	}
	defer d.sessionManager.FinishStart(startReservation)

	// Persist context recovery first, but never start its synthetic prompt ahead
	// of an older server-accepted runtime delivery. If a live ownership record
	// exists below, that record is handed off/reconciled first; the persisted
	// recovery entry is started after the ownership FIFO drains.
	contextRecoveryPersistenceFailed := false
	if needed, wakes := ap.ContextRecoveryNeeded(); needed {
		if len(wakes) == 0 {
			exactTarget := strings.TrimSpace(binding.targetID)
			if agentteamMeta, ok := binding.meta.(*types.AgentTeamMeta); ok {
				exactTarget = firstNonEmpty(exactTarget, strings.TrimSpace(agentteamMeta.Target))
			}
			if exactTarget != "" {
				wakes = []agent.WakeInfo{{Target: exactTarget}}
				slog.Warn("context recovery: no pending wake, using exact conversation target",
					"agent_id", ap.AgentID,
					"fallback_target", exactTarget,
				)
			}
		}
		if len(wakes) > 0 {
			if err := d.enqueueContextRecoveryContinuation(sess, wakes); err != nil {
				contextRecoveryPersistenceFailed = true
				d.emitRuntimeEvent(binding.channelID, binding.serverAgentID, types.RuntimeEvent{
					AgentID:    binding.serverAgentID,
					EventClass: consts.RuntimeEventClassWakeFailed,
					Target:     strings.TrimSpace(wakes[0].Target),
					Detail:     fmt.Sprintf("persist context recovery continuation: %v", err),
				})
				slog.Warn("persist context recovery continuation", "agent_id", ap.AgentID, "error", err)
			}
		} else {
			contextRecoveryPersistenceFailed = true
			d.emitRuntimeEvent(binding.channelID, binding.serverAgentID, types.RuntimeEvent{
				AgentID:    binding.serverAgentID,
				EventClass: consts.RuntimeEventClassWakeFailed,
				Detail:     "context recovery could not continue because no exact message target was available",
			})
			slog.Warn("context recovery: no exact target, cannot recover",
				"agent_id", ap.AgentID,
			)
		}
	}

	// Server-accepted stdin/protocol deliveries are owned by the durable
	// runtime-delivery journal, not the server outbox. Replay/reconcile them
	// immediately after an unexpected Process boundary; waiting for another
	// human message would strand a delivery the server has already forgotten.
	if ap.ShouldAutoContinueAcceptedDeliveries() {
		if restarted, err := d.tryStartRuntimeDeliveryOwnershipContinuationWithReservation(sess, startReservation); err != nil {
			slog.Warn("runtime delivery ownership continuation", "agent_id", ap.AgentID, "error", err)
			d.scheduleRuntimeDeliveryOwnershipContinuationRetry(sess, err)
			return
		} else if restarted {
			return
		} else if _, live, loadErr := oldestLiveRuntimeDeliveryOwnership(sess.workDir); loadErr != nil || live {
			// A concurrent continuation may already be publishing a Process, or a
			// recoverable launch path may have returned without starting one. Never
			// remove the session while accepted ownership is still live.
			if loadErr == nil {
				loadErr = errors.New("runtime delivery ownership continuation did not start")
			}
			d.scheduleRuntimeDeliveryOwnershipContinuationRetry(sess, loadErr)
			return
		}
	}
	if contextRecoveryPersistenceFailed {
		return
	}

	// This also picks up a recovery entry queued by the Process that previously
	// owned an accepted q1. Calling it unconditionally is cheap (a state read)
	// and makes the q1 -> synthetic-recovery ordering survive Process replacement.
	restarted, err := d.startContextRecoveryProcessWithReservation(sess, startReservation)
	if err != nil {
		slog.Warn("context recovery continuation", "agent_id", ap.AgentID, "error", err)
	}
	if restarted {
		// A launched recovery process owns the next lifecycle boundary even if
		// removing its marker returned an unknown persistence result.
		return
	}

	restarted, err = d.advanceDeferredContinuationWithReservation(sess, ap, startReservation)
	if err != nil {
		slog.Warn("advance deferred continuation", "agent_id", ap.AgentID, "error", err)
	}

	existing, ok := d.sessionManager.Get(sessionKey)
	if !ok {
		return
	}
	if restarted {
		return
	}
	if ap.IsSleeping() {
		if existing == sess {
			existing.mu.Lock()
			existing.process = nil
			existing.mu.Unlock()
		}
		return
	}
	d.fenceSessionMemoryWatcherForFinalRemoval(sess)
	d.stopMemoryWatcherAndRetry(sess)
	d.sessionManager.RemoveIfCurrent(sessionKey, sess)
}

func (d *Dispatcher) startRuntimeDeliveryOwnershipContinuationWithContext(
	ctx context.Context,
	sess *interactiveSession,
	startReservation *sessionStartReservation,
) (bool, error) {
	if ctx == nil {
		ctx = context.Background()
	}
	if err := ctx.Err(); err != nil {
		return false, err
	}
	sess.mu.Lock()
	if sess.process != nil {
		sess.mu.Unlock()
		return false, nil
	}
	ag := sess.agent
	rt := sess.runtime
	binding := sess.binding
	workDir := sess.workDir
	runtimeStartReplacement := sess.runtimeStartReplacement
	sess.mu.Unlock()

	oldest, ok, err := oldestLiveRuntimeDeliveryOwnership(workDir)
	if err != nil || !ok {
		return false, err
	}
	state, err := loadSessionResumeState(workDir)
	if err != nil {
		return false, err
	}
	currentPolicy := currentAgentRuntimePolicy(ag, rt)
	decision, err := applySessionResumeGuardToState(workDir, &state, currentPolicy)
	if err != nil {
		return false, err
	}
	logResumeGuardDecision(ag.ID, decision)
	model, reasoningEffort := resolveModelAndReasoning(&state, ag)
	if runtimeStartReplacement {
		// The saved values describe the Process being replaced. Once a newer
		// runtime launch configuration owns retirement, carrying those values
		// into its durable-delivery continuation would recreate the stale
		// fingerprint and make the incoming delivery restart it again forever.
		model = strings.TrimSpace(ag.Model)
		reasoningEffort = strings.TrimSpace(ag.ReasoningEffort)
	}
	savedSessionID := strings.TrimSpace(state.LastSessionID)
	if state.LastStartConfig != nil && strings.TrimSpace(state.LastStartConfig.SessionID) != "" {
		savedSessionID = strings.TrimSpace(state.LastStartConfig.SessionID)
	}

	handoff := types.InboxMessage{}
	msg := oldest.Message
	mode := promptModeDeferred
	wake := buildWakeFromDeferredMessage(msg)
	suppressIdleStart := false
	switch oldest.State {
	case agent.DeliveryOwnershipInflight:
		// Do not replay a transport-ambiguous body as a launch prompt. Start the
		// runtime without user input and let SetDeliveryOwnershipStore reconcile
		// clientUserMessageId before it releases later queued messages.
		msg = types.InboxMessage{}
		wake = nil
		mode = promptModeSession
		suppressIdleStart = true
	case agent.DeliveryOwnershipQueued:
		// A queued accepted body must not be a Spawn initial prompt: Spawn can
		// accept turn/start and still fail before Process/store publication. Start
		// a no-turn runtime first, then move queued -> pending and issue the
		// correlated request through DeliverForHandoff.
		handoff = msg
		binding = bindingWithDeferredAgentTeamMessage(binding, handoff)
		msg = types.InboxMessage{}
		wake = nil
		mode = promptModeSession
		suppressIdleStart = true
	default:
		// Only the explicitly defined queued phase permits replay. A malformed or
		// future ownership phase is still a live durable owner, but its replay
		// semantics are unknown; fail closed instead of speculatively treating it
		// as queued and duplicating a body the runtime may already have accepted.
		return false, fmt.Errorf("unsupported live runtime delivery ownership state %q for message %q", oldest.State, oldest.MessageID())
	}

	lc := &processLaunchContext{
		sess:                      sess,
		agent:                     ag,
		runtime:                   rt,
		binding:                   binding,
		msg:                       msg,
		wake:                      wake,
		requestContext:            ctx,
		mode:                      mode,
		pendingBehind:             max(0, len(state.RuntimeDeliveryOwnership)-1),
		suppressIdleStartTurn:     suppressIdleStart,
		ownershipHandoffMessageID: strings.TrimSpace(handoff.MessageID),
		startReservation:          startReservation,
		savedModel:                model,
		savedReasoningEffort:      reasoningEffort,
		savedSessionID:            savedSessionID,
		checkpointMsg:             checkpointMsgDeferred,
	}
	if err := d.resolve(lc); err != nil {
		d.abortSessionLaunch(sess, lc.startCfg)
		return false, err
	}
	process, err := d.launch(ctx, lc)
	if err != nil {
		d.abortSessionLaunch(sess, lc.startCfg)
		return false, err
	}
	if strings.TrimSpace(handoff.MessageID) != "" {
		if err := process.DeliverForHandoff(ctx, handoff); err != nil {
			d.revokeExecProcessAuthorization(process)
			process.Stop()
			return false, fmt.Errorf("deliver queued ownership continuation handoff: %w", err)
		}
	}
	return true, nil
}

func (d *Dispatcher) tryStartRuntimeDeliveryOwnershipContinuationWithContext(ctx context.Context, sess *interactiveSession) (bool, error) {
	if d == nil || sess == nil {
		return false, nil
	}
	if ctx == nil {
		ctx = context.Background()
	}
	if err := ctx.Err(); err != nil {
		return false, err
	}
	sess.mu.Lock()
	sessionKey := sess.sessionKey
	if strings.TrimSpace(sessionKey) == "" {
		sessionKey = sess.agentID
	}
	sess.mu.Unlock()
	reservation, _, ownsStart := d.sessionManager.BeginStart(sessionKey, sess)
	if !ownsStart {
		return false, errors.New("runtime delivery continuation lost the session start reservation")
	}
	defer d.sessionManager.FinishStart(reservation)
	return d.tryStartRuntimeDeliveryOwnershipContinuationWithContextAndReservation(ctx, sess, reservation)
}

func (d *Dispatcher) tryStartRuntimeDeliveryOwnershipContinuationWithReservation(
	sess *interactiveSession,
	startReservation *sessionStartReservation,
) (bool, error) {
	return d.tryStartRuntimeDeliveryOwnershipContinuationWithContextAndReservation(context.Background(), sess, startReservation)
}

func (d *Dispatcher) tryStartRuntimeDeliveryOwnershipContinuationWithContextAndReservation(
	ctx context.Context,
	sess *interactiveSession,
	startReservation *sessionStartReservation,
) (bool, error) {
	if d != nil && d.runtimeOwnershipContinuationStarter != nil {
		return d.runtimeOwnershipContinuationStarter(sess)
	}
	return d.startRuntimeDeliveryOwnershipContinuationWithContext(ctx, sess, startReservation)
}

const (
	runtimeOwnershipContinuationRetryAttempts = 8
	runtimeOwnershipContinuationRetryInitial  = 100 * time.Millisecond
	runtimeOwnershipContinuationRetryMax      = 5 * time.Second
)

func runtimeOwnershipContinuationRetryDelay(attempt int) time.Duration {
	if attempt <= 1 {
		return runtimeOwnershipContinuationRetryInitial
	}
	delay := runtimeOwnershipContinuationRetryInitial
	for i := 1; i < attempt; i++ {
		delay *= 2
		if delay >= runtimeOwnershipContinuationRetryMax {
			return runtimeOwnershipContinuationRetryMax
		}
	}
	return delay
}

func (d *Dispatcher) ownershipContinuationRetryDelay(attempt int) time.Duration {
	if d != nil && d.runtimeOwnershipContinuationDelay != nil {
		if delay := d.runtimeOwnershipContinuationDelay(attempt); delay >= 0 {
			return delay
		}
	}
	return runtimeOwnershipContinuationRetryDelay(attempt)
}

// launchRuntimeOwnershipRetryWorker registers one continuation worker with
// both the Dispatcher and its owning session before starting the goroutine.
// Shutdown closes the global registry before waiting, while reset blocks the
// session registry before waiting; the two guarded Add operations therefore
// cannot race either WaitGroup.Wait call.
func (d *Dispatcher) launchRuntimeOwnershipRetryWorker(
	sess *interactiveSession,
	run func(context.Context),
) bool {
	if d == nil || sess == nil || run == nil {
		return false
	}

	d.runtimeOwnershipRetryMu.Lock()
	if d.runtimeOwnershipRetryClosed {
		d.runtimeOwnershipRetryMu.Unlock()
		return false
	}
	parent := d.runtimeOwnershipRetryCtx
	if parent == nil {
		parent = context.Background()
	}
	d.runtimeOwnershipRetryWG.Add(1)
	d.runtimeOwnershipRetryMu.Unlock()

	sess.mu.Lock()
	if sess.ownershipContinuationRetryBlocked {
		sess.mu.Unlock()
		d.runtimeOwnershipRetryWG.Done()
		return false
	}
	if sess.ownershipContinuationRetryCtx == nil {
		sess.ownershipContinuationRetryCtx, sess.ownershipContinuationRetryCancel = context.WithCancel(parent)
	}
	ctx := sess.ownershipContinuationRetryCtx
	sess.ownershipContinuationRetryWG.Add(1)
	sess.mu.Unlock()

	go func() {
		defer d.runtimeOwnershipRetryWG.Done()
		defer sess.ownershipContinuationRetryWG.Done()
		run(ctx)
	}()
	return true
}

func waitRuntimeOwnershipRetryDelay(ctx context.Context, delay time.Duration) bool {
	if ctx == nil {
		ctx = context.Background()
	}
	if delay <= 0 {
		select {
		case <-ctx.Done():
			return false
		default:
			return true
		}
	}
	timer := time.NewTimer(delay)
	defer timer.Stop()
	select {
	case <-timer.C:
		return true
	case <-ctx.Done():
		return false
	}
}

// blockAndWaitRuntimeOwnershipRetries makes reset the final owner of every
// continuation worker already associated with sess. Once blocked, no worker
// can be added to the per-session WaitGroup, so a successful return is a hard
// no-late-write boundary for workspace removal.
func (d *Dispatcher) blockAndWaitRuntimeOwnershipRetries(ctx context.Context, sess *interactiveSession) error {
	if d == nil || sess == nil {
		return nil
	}
	sess.mu.Lock()
	sess.ownershipContinuationRetryBlocked = true
	sess.ownershipContinuationRetryGeneration++
	sess.ownershipContinuationRetryRunning = false
	cancel := sess.ownershipContinuationRetryCancel
	sess.mu.Unlock()
	if cancel != nil {
		cancel()
	}
	return waitForDispatcherWaitGroup(ctx, &sess.ownershipContinuationRetryWG)
}

func (d *Dispatcher) cancelRuntimeOwnershipRetryWorkers() {
	if d == nil {
		return
	}
	d.runtimeOwnershipRetryMu.Lock()
	if !d.runtimeOwnershipRetryClosed {
		d.runtimeOwnershipRetryClosed = true
		if d.runtimeOwnershipRetryCancel != nil {
			d.runtimeOwnershipRetryCancel()
		}
	}
	d.runtimeOwnershipRetryMu.Unlock()
}

func (d *Dispatcher) stopRuntimeOwnershipRetryWorkers(ctx context.Context) error {
	if d == nil {
		return nil
	}
	d.cancelRuntimeOwnershipRetryWorkers()
	return waitForDispatcherWaitGroup(ctx, &d.runtimeOwnershipRetryWG)
}

func waitForDispatcherWaitGroup(ctx context.Context, group *sync.WaitGroup) error {
	if group == nil {
		return nil
	}
	if ctx == nil {
		ctx = context.Background()
	}
	done := make(chan struct{})
	go func() {
		group.Wait()
		close(done)
	}()
	select {
	case <-done:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

// scheduleRuntimeDeliveryOwnershipContinuationRetry keeps a session ledger
// alive after a transient resolve/spawn failure. The server has already
// forgotten these accepted records, so removing the session here would strand
// them until unrelated traffic or a daemon restart.
func (d *Dispatcher) scheduleRuntimeDeliveryOwnershipContinuationRetry(sess *interactiveSession, initialErr error) {
	if d == nil || sess == nil {
		return
	}
	sess.mu.Lock()
	if sess.ownershipContinuationRetryRunning || sess.ownershipContinuationRetryBlocked || sess.retired.Load() {
		sess.mu.Unlock()
		return
	}
	sess.ownershipContinuationRetryRunning = true
	sess.ownershipContinuationRetryGeneration++
	generation := sess.ownershipContinuationRetryGeneration
	agentID := sess.agentID
	sess.mu.Unlock()

	started := d.launchRuntimeOwnershipRetryWorker(sess, func(retryCtx context.Context) {
		defer func() {
			sess.mu.Lock()
			if sess.ownershipContinuationRetryGeneration == generation {
				sess.ownershipContinuationRetryRunning = false
			}
			sess.mu.Unlock()
		}()
		lastErr := initialErr
		if agent.IsDeterministicRuntimeStartupError(lastErr) {
			d.failRuntimeDeliveryOwnershipContinuation(retryCtx, sess, generation, lastErr)
			return
		}
		for attempt := 1; attempt <= runtimeOwnershipContinuationRetryAttempts; attempt++ {
			if !waitRuntimeOwnershipRetryDelay(retryCtx, d.ownershipContinuationRetryDelay(attempt)) {
				return
			}
			sess.mu.Lock()
			active := sess.ownershipContinuationRetryRunning &&
				sess.ownershipContinuationRetryGeneration == generation &&
				!sess.retired.Load()
			sess.mu.Unlock()
			if !active {
				return
			}

			restarted, err := d.tryStartRuntimeDeliveryOwnershipContinuationWithContext(retryCtx, sess)
			if restarted {
				sess.mu.Lock()
				if sess.ownershipContinuationRetryGeneration == generation {
					sess.ownershipContinuationRetryRunning = false
				}
				sess.mu.Unlock()
				return
			}
			if err != nil {
				lastErr = err
				if agent.IsDeterministicRuntimeStartupError(err) {
					d.failRuntimeDeliveryOwnershipContinuation(retryCtx, sess, generation, lastErr)
					return
				}
			}
			if _, ok, loadErr := oldestLiveRuntimeDeliveryOwnership(func() string {
				sess.mu.Lock()
				defer sess.mu.Unlock()
				return sess.workDir
			}()); loadErr == nil && !ok {
				sess.mu.Lock()
				if sess.ownershipContinuationRetryGeneration == generation {
					sess.ownershipContinuationRetryRunning = false
				}
				sess.mu.Unlock()
				if restarted, err := d.tryStartContextRecoveryProcessWithContext(retryCtx, sess); err != nil {
					slog.Warn("start queued context recovery after ownership drain", "agent_id", agentID, "error", err)
				} else if restarted {
					return
				}
				return
			}
		}

		d.failRuntimeDeliveryOwnershipContinuation(retryCtx, sess, generation, lastErr)
		slog.Warn("runtime delivery ownership continuation retry exhausted", "agent_id", agentID, "error", lastErr)
	})
	if !started {
		sess.mu.Lock()
		if sess.ownershipContinuationRetryGeneration == generation {
			sess.ownershipContinuationRetryRunning = false
		}
		sess.mu.Unlock()
	}
}

func (d *Dispatcher) failRuntimeDeliveryOwnershipContinuation(ctx context.Context, sess *interactiveSession, generation uint64, cause error) {
	if ctx == nil {
		ctx = context.Background()
	}
	if err := ctx.Err(); err != nil {
		return
	}
	sess.mu.Lock()
	if !sess.ownershipContinuationRetryRunning || sess.ownershipContinuationRetryGeneration != generation {
		sess.mu.Unlock()
		return
	}
	workDir := sess.workDir
	binding := sess.binding
	sessionKey := sess.sessionKey
	if strings.TrimSpace(sessionKey) == "" {
		sessionKey = sess.agentID
	}
	sess.mu.Unlock()

	// Exhausting this retry worker is not proof that the delivery itself is
	// unowned: every failed attempt may merely have lost the session-key start
	// reservation to an ordinary delivery that is already restoring/reconciling
	// the same durable head. Reserve the key before terminalizing. If another
	// generation owns it, hand retry responsibility to whichever session is
	// current after that generation settles; never mark q1 failed underneath a
	// Process whose ownership store is being installed.
	terminalReservation, waitCh, ownsTerminalFence := d.sessionManager.BeginStart(sessionKey, sess)
	if !ownsTerminalFence {
		sess.mu.Lock()
		if sess.ownershipContinuationRetryGeneration == generation {
			sess.ownershipContinuationRetryRunning = false
		}
		sess.mu.Unlock()
		d.launchRuntimeOwnershipRetryWorker(sess, func(retryCtx context.Context) {
			d.retryRuntimeDeliveryOwnershipAfterCompetingStart(retryCtx, sessionKey, waitCh, cause)
		})
		return
	}
	terminalFenceHeld := true
	defer func() {
		if terminalFenceHeld {
			d.sessionManager.FinishStart(terminalReservation)
		}
	}()

	sess.mu.Lock()
	process := sess.process
	sess.mu.Unlock()
	if process != nil && process.AcceptingDeliveries() {
		sess.mu.Lock()
		if sess.ownershipContinuationRetryGeneration == generation {
			sess.ownershipContinuationRetryRunning = false
		}
		sess.mu.Unlock()
		return
	}

	var oldest agent.DeliveryOwnershipRecord
	for attempt := 1; ; attempt++ {
		if err := ctx.Err(); err != nil {
			return
		}
		var ok bool
		var err error
		oldest, ok, err = oldestLiveRuntimeDeliveryOwnership(workDir)
		if err == nil && ok {
			break
		}
		if err == nil && !ok {
			sess.mu.Lock()
			if sess.ownershipContinuationRetryGeneration == generation {
				sess.ownershipContinuationRetryRunning = false
			}
			sess.mu.Unlock()
			return
		}
		if sess.retired.Load() {
			return
		}
		slog.Error("load exhausted runtime continuation owner; retaining session ledger",
			"agent_id", sess.agentID, "error", err)
		if !waitRuntimeOwnershipRetryDelay(ctx, runtimeOwnershipContinuationRetryDelay(attempt)) {
			return
		}
	}
	if cause == nil {
		cause = errors.New("runtime continuation launch did not converge")
	}
	record := oldest
	record.State = agent.DeliveryOwnershipTerminal
	record.InflightReason = ""
	record.Outcome = agent.DeliveryOutcomeFailed
	record.UpdatedAt = time.Now().UTC()
	record.LastError = cause.Error()
	record.FailureCode = "runtime_continuation_launch_exhausted"
	record = agent.NormalizeDeliveryOwnershipRecordForPersistence(record)
	store := newSessionDeliveryOwnershipStore(workDir)
	for attempt := 1; ; attempt++ {
		if err := ctx.Err(); err != nil {
			return
		}
		var persistErr error
		if d.runtimeOwnershipTerminalPersist != nil {
			persistErr = d.runtimeOwnershipTerminalPersist(store, record)
		} else {
			_, persistErr = store.PutDeliveryOwnership(record)
		}
		if persistErr == nil {
			break
		} else {
			// Keep the live queued/uncertain record and session ledger until the
			// terminal transition is fsynced. A daemon restart will see that live
			// owner and resume continuation rather than losing an accepted body.
			slog.Error("persist exhausted runtime continuation failure; retaining live owner",
				"agent_id", sess.agentID, "message_id", record.MessageID(), "error", persistErr)
		}
		if sess.retired.Load() {
			return
		}
		if !waitRuntimeOwnershipRetryDelay(ctx, runtimeOwnershipContinuationRetryDelay(attempt)) {
			return
		}
	}
	// The terminal record is now durable. Release the start fence before
	// advancing q2 or starting queued context recovery; those paths must acquire
	// their own generation and would otherwise wait on this goroutine itself.
	d.sessionManager.FinishStart(terminalReservation)
	terminalFenceHeld = false
	if err := ctx.Err(); err != nil {
		return
	}
	failureDetail := fmt.Errorf("runtime delivery continuation failed after %d attempts: %w",
		runtimeOwnershipContinuationRetryAttempts, cause)
	if agent.IsDeterministicRuntimeStartupError(cause) {
		failureDetail = fmt.Errorf("runtime delivery continuation stopped on deterministic startup error: %w", cause)
	}
	d.emitWakeFailed(binding.channelID, binding.serverAgentID, oldest.Message, failureDetail)

	// Terminalizing the head record must not strand later accepted records. End
	// this head's bounded retry generation, then start a fresh generation for the
	// next live owner. scheduleRuntimeDeliveryOwnershipContinuationRetry always
	// launches its worker asynchronously, so a long FIFO cannot grow the current
	// goroutine's stack and every record receives its own independent attempt
	// budget.
	var (
		nextLive bool
		err      error
	)
	for attempt := 1; ; attempt++ {
		if err := ctx.Err(); err != nil {
			return
		}
		_, nextLive, err = oldestLiveRuntimeDeliveryOwnership(workDir)
		if err == nil {
			break
		}
		if sess.retired.Load() {
			return
		}
		slog.Error("load next runtime continuation owner; retaining session ledger",
			"agent_id", sess.agentID, "error", err)
		if !waitRuntimeOwnershipRetryDelay(ctx, runtimeOwnershipContinuationRetryDelay(attempt)) {
			return
		}
	}
	sess.mu.Lock()
	if sess.ownershipContinuationRetryGeneration == generation {
		sess.ownershipContinuationRetryRunning = false
	}
	sess.mu.Unlock()
	if nextLive {
		if err := ctx.Err(); err != nil {
			return
		}
		d.scheduleRuntimeDeliveryOwnershipContinuationRetry(sess,
			fmt.Errorf("continuing after terminal delivery %s: %w", oldest.MessageID(), cause))
		return
	}
	if err := ctx.Err(); err != nil {
		return
	}
	restarted, err := d.tryStartContextRecoveryProcessWithContext(ctx, sess)
	if err != nil {
		slog.Warn("start queued context recovery after terminal ownership", "agent_id", sess.agentID, "error", err)
	}
	if restarted {
		return
	}
	d.fenceSessionMemoryWatcherForFinalRemoval(sess)
	d.stopMemoryWatcherAndRetry(sess)
	d.sessionManager.RemoveIfCurrent(sessionKey, sess)
}

// retryRuntimeDeliveryOwnershipAfterCompetingStart runs only after an
// exhausted retry worker discovers that a different start generation owns the
// key. Publication installs the ownership store before releasing waitCh, so an
// accepting current Process is authoritative and needs no second starter. If
// the competing generation failed (or its Process already stopped), schedule a
// fresh bounded retry on the still-current session instead of failing its
// durable head from the stale session object.
func (d *Dispatcher) retryRuntimeDeliveryOwnershipAfterCompetingStart(ctx context.Context, sessionKey string, waitCh <-chan struct{}, cause error) {
	if ctx == nil {
		ctx = context.Background()
	}
	if waitCh != nil {
		select {
		case <-waitCh:
		case <-ctx.Done():
			return
		}
	}
	if ctx.Err() != nil || d == nil || d.sessionManager == nil || d.sessionManager.Closed() {
		return
	}
	current, ok := d.sessionManager.Get(sessionKey)
	if !ok || current == nil || current.retired.Load() {
		return
	}
	current.mu.Lock()
	process := current.process
	workDir := current.workDir
	current.mu.Unlock()
	if process != nil && process.AcceptingDeliveries() {
		return
	}
	if _, live, err := oldestLiveRuntimeDeliveryOwnership(workDir); err != nil || live {
		if err == nil {
			err = cause
		}
		if err == nil {
			err = errors.New("runtime delivery ownership remains live after competing start")
		}
		d.scheduleRuntimeDeliveryOwnershipContinuationRetry(current, err)
	}
}

// startContextRecoveryContinuation injects a synthetic context recovery
// message into the deferred delivery queue and immediately starts a new
// session. This works for all runtimes (including stdin-supporting ones like
// claude) because the process was already terminated and a brand new process
// is created with the recovery message as its initial prompt.
func (d *Dispatcher) startContextRecoveryContinuation(sess *interactiveSession, wakes []agent.WakeInfo) (bool, error) {
	if err := d.enqueueContextRecoveryContinuation(sess, wakes); err != nil {
		return false, err
	}
	return d.startContextRecoveryProcess(sess)
}

func (d *Dispatcher) tryStartContextRecoveryProcessWithContext(ctx context.Context, sess *interactiveSession) (bool, error) {
	if d == nil || sess == nil {
		return false, nil
	}
	if ctx == nil {
		ctx = context.Background()
	}
	if err := ctx.Err(); err != nil {
		return false, err
	}
	sess.mu.Lock()
	sessionKey := sess.sessionKey
	if strings.TrimSpace(sessionKey) == "" {
		sessionKey = sess.agentID
	}
	sess.mu.Unlock()
	reservation, _, ownsStart := d.sessionManager.BeginStart(sessionKey, sess)
	if !ownsStart {
		return false, errors.New("context recovery lost the session start reservation")
	}
	defer d.sessionManager.FinishStart(reservation)
	return d.startContextRecoveryProcessWithContextAndReservation(ctx, sess, reservation)
}

func aggregateContextRecoveryWakes(wakes []agent.WakeInfo) []agent.WakeInfo {
	byTarget := make(map[string]agent.WakeInfo, len(wakes))
	for _, raw := range wakes {
		wake := raw
		wake.TargetKey = strings.TrimSpace(wake.TargetKey)
		wake.Target = strings.TrimSpace(wake.Target)
		if wake.TargetKey == "" || wake.Target == "" || wake.DeliveryClass == agent.DeliveryClassSignalOnly {
			continue
		}
		current, ok := byTarget[wake.TargetKey]
		if !ok || recoveryWakeIsHigher(wake, current) {
			byTarget[wake.TargetKey] = wake
		}
	}

	keys := make([]string, 0, len(byTarget))
	for targetKey := range byTarget {
		keys = append(keys, targetKey)
	}
	sort.Strings(keys)
	aggregated := make([]agent.WakeInfo, 0, len(keys))
	for _, targetKey := range keys {
		aggregated = append(aggregated, byTarget[targetKey])
	}
	return aggregated
}

func recoveryWakeIsHigher(candidate, current agent.WakeInfo) bool {
	if candidate.TargetSeq != current.TargetSeq {
		return candidate.TargetSeq > current.TargetSeq
	}
	candidateTarget := strings.TrimSpace(candidate.Target)
	currentTarget := strings.TrimSpace(current.Target)
	if candidateTarget != currentTarget {
		return candidateTarget < currentTarget
	}
	return strings.TrimSpace(candidate.MessageID) < strings.TrimSpace(current.MessageID)
}

func contextRecoveryMarkerTarget(entry deferredDeliveryEntry) (types.InboxMessage, bool) {
	msg, ok := deferredInboxMessage(entry.Message)
	if !ok || !isContextRecoveryMessage(msg) || strings.TrimSpace(msg.TargetKey) == "" {
		return types.InboxMessage{}, false
	}
	return msg, true
}

// mergeContextRecoveryMarker retains an already-launched marker as a durable
// de-duplication fence, but never lets that fence hide a newer frontier for the
// same target. Pending entries are immutable: a higher frontier is represented
// by a second deterministic marker and is replayed only after the old marker's
// process boundary settles the old entry.
func mergeContextRecoveryMarker(entries []deferredDeliveryEntry, marker deferredDeliveryEntry) []deferredDeliveryEntry {
	markerMsg, ok := contextRecoveryMarkerTarget(marker)
	if !ok {
		return entries
	}
	markerID := strings.TrimSpace(marker.MessageID)
	if markerID == "" {
		markerID = deferredMessageID(marker.Message)
	}
	for _, entry := range entries {
		if strings.TrimSpace(entry.MessageID) == markerID || deferredMessageID(entry.Message) == markerID {
			return entries
		}
	}

	targetKey := strings.TrimSpace(markerMsg.TargetKey)
	pendingCovers := false
	existingCovers := false
	for _, entry := range entries {
		existingMsg, ok := contextRecoveryMarkerTarget(entry)
		if !ok || strings.TrimSpace(existingMsg.TargetKey) != targetKey {
			continue
		}
		if entry.ReplayPending {
			pendingCovers = pendingCovers || existingMsg.TargetSeq >= markerMsg.TargetSeq
			continue
		}
		existingCovers = existingCovers || existingMsg.TargetSeq >= markerMsg.TargetSeq
	}

	// A pending or available marker at this frontier already covers the retry.
	// Clean only non-pending stale markers; a pending entry is a live launch
	// fence and must remain byte-for-byte untouched.
	if pendingCovers || existingCovers {
		filtered := make([]deferredDeliveryEntry, 0, len(entries))
		for _, entry := range entries {
			existingMsg, ok := contextRecoveryMarkerTarget(entry)
			if ok && strings.TrimSpace(existingMsg.TargetKey) == targetKey &&
				!entry.ReplayPending && existingMsg.TargetSeq <= markerMsg.TargetSeq {
				continue
			}
			filtered = append(filtered, entry)
		}
		return filtered
	}

	// No marker covers the new frontier. Replace only older, not-yet-launched
	// entries; any ReplayPending marker remains as an independent fence.
	filtered := make([]deferredDeliveryEntry, 0, len(entries)+1)
	for _, entry := range entries {
		existingMsg, ok := contextRecoveryMarkerTarget(entry)
		if ok && strings.TrimSpace(existingMsg.TargetKey) == targetKey &&
			!entry.ReplayPending && existingMsg.TargetSeq < markerMsg.TargetSeq {
			continue
		}
		filtered = append(filtered, entry)
	}
	return append(filtered, marker)
}

func (d *Dispatcher) enqueueContextRecoveryContinuation(sess *interactiveSession, wakes []agent.WakeInfo) error {
	sess.mu.Lock()
	agentID := sess.agentID
	workDir := sess.workDir
	sess.mu.Unlock()

	if len(wakes) == 0 {
		return errors.New("enqueue context recovery continuation: no wake targets")
	}

	aggregated := aggregateContextRecoveryWakes(wakes)
	if len(aggregated) == 0 {
		return errors.New("enqueue context recovery continuation: no visible-reply wake has a canonical target key")
	}

	now := time.Now().UTC()
	markers := make([]deferredDeliveryEntry, 0, len(aggregated))
	for i, wake := range aggregated {
		if !parseWakeTargetIntoDeferredMessage(wake.Target, &deferredContinuationMessage{}) {
			return fmt.Errorf("enqueue context recovery continuation: target %d has invalid exact target %q", i, wake.Target)
		}
		msg := deferredContinuationMessage{
			MessageID: contextRecoveryContinuationID(agentID, wake),
			TargetKey: wake.TargetKey,
			TargetSeq: wake.TargetSeq,
			AuthorProfile: types.AuthorProfile{
				ActorType:   "system",
				DisplayName: "daemon",
				TargetID:    "daemon",
			},
			Content:     checkpointMsgRecovery,
			Timestamp:   now.Format(time.RFC3339),
			MessageType: "context_recovery",
		}
		if !parseWakeTargetIntoDeferredMessage(wake.Target, &msg) {
			return fmt.Errorf("enqueue context recovery continuation: target %d has invalid exact target %q", i, wake.Target)
		}
		raw, err := json.Marshal(msg)
		if err != nil {
			return fmt.Errorf("marshal context recovery message: %w", err)
		}
		markers = append(markers, deferredDeliveryEntry{
			MessageID:  msg.MessageID,
			AcceptedAt: now,
			Message:    append(json.RawMessage(nil), raw...),
		})
	}

	if err := updateSessionResumeState(workDir, func(state *sessionResumeState) error {
		entries := append([]deferredDeliveryEntry(nil), state.DeferredDeliveries...)
		for _, marker := range markers {
			entries = mergeContextRecoveryMarker(entries, marker)
		}
		normalizeDeferredDeliveryOrder(entries)
		state.DeferredDeliveries = entries
		return nil
	}); err != nil {
		return fmt.Errorf("persist context recovery deliveries: %w", err)
	}

	slog.Info("context recovery: injected continuation messages",
		"agent_id", agentID,
		"target_count", len(markers),
	)
	return nil
}

func contextRecoveryContinuationID(agentID string, wake agent.WakeInfo) string {
	hash := sha256.New()
	for _, value := range []string{
		strings.TrimSpace(agentID),
		strings.TrimSpace(wake.TargetKey),
		fmt.Sprintf("%d", wake.TargetSeq),
		strings.TrimSpace(wake.Target),
	} {
		_, _ = hash.Write([]byte(value))
		_, _ = hash.Write([]byte{0})
	}
	return "ctx-recovery-" + hex.EncodeToString(hash.Sum(nil)[:12])
}

// parseWakeTargetIntoDeferredMessage extracts channel type/name from a
// wake target string (e.g. "#channel:shortid", "#channel", "dm:@user:shortid").
func parseWakeTargetIntoDeferredMessage(target string, msg *deferredContinuationMessage) bool {
	if msg == nil {
		return false
	}
	channelType, channelName, parentChannelType, parentChannelName := agentteamTargetFields(target)
	if channelType == "" || channelName == "" {
		return false
	}
	if channelType == "thread" && (parentChannelType == "" || parentChannelName == "") {
		return false
	}
	msg.ChannelType = channelType
	msg.ChannelName = channelName
	msg.ParentChannelType = parentChannelType
	msg.ParentChannelName = parentChannelName
	return true
}

// startContextRecoveryProcess starts a fresh process for context recovery.
// This is used when startDeferredContinuation returns false (e.g. for
// stdin-supporting runtimes). It loads the saved state and delegates
// to the shared resolve + launch path.
func (d *Dispatcher) startContextRecoveryProcess(sess *interactiveSession) (bool, error) {
	return d.startContextRecoveryProcessWithContextAndReservation(context.Background(), sess, nil)
}

func (d *Dispatcher) startContextRecoveryProcessWithReservation(
	sess *interactiveSession,
	startReservation *sessionStartReservation,
) (bool, error) {
	return d.startContextRecoveryProcessWithContextAndReservation(context.Background(), sess, startReservation)
}

func (d *Dispatcher) startContextRecoveryProcessWithContextAndReservation(
	ctx context.Context,
	sess *interactiveSession,
	startReservation *sessionStartReservation,
) (bool, error) {
	if ctx == nil {
		ctx = context.Background()
	}
	if err := ctx.Err(); err != nil {
		return false, err
	}
	sess.mu.Lock()
	if sess.process != nil {
		sess.mu.Unlock()
		return false, nil
	}
	ag := sess.agent
	rt := sess.runtime
	binding := sess.binding
	agentID := sess.agentID
	workDir := sess.workDir
	sess.mu.Unlock()

	state, err := loadSessionResumeState(workDir)
	if err != nil {
		return false, err
	}
	if len(state.DeferredDeliveries) == 0 {
		return false, nil
	}
	currentPolicy := currentAgentRuntimePolicy(ag, rt)
	decision, err := applySessionResumeGuardToState(workDir, &state, currentPolicy)
	if err != nil {
		return false, err
	}
	logResumeGuardDecision(agentID, decision)

	// A pending replay belongs to the process that already claimed it. Wait for
	// that process boundary to settle the old marker before starting a newer
	// context-recovery frontier; otherwise the new marker could overtake the
	// old lifecycle and leave the fence stranded.
	if firstReplayPendingIndex(state.DeferredDeliveries) >= 0 {
		return false, nil
	}
	nextIndex := contextRecoveryReplayIndex(state.DeferredDeliveries)
	if nextIndex < 0 {
		return false, nil
	}

	msg, ok := deferredInboxMessage(state.DeferredDeliveries[nextIndex].Message)
	if !ok {
		return false, fmt.Errorf("context recovery delivery is missing wake message fields")
	}
	markerID := strings.TrimSpace(msg.MessageID)
	if err := updateSessionResumeState(workDir, func(latest *sessionResumeState) error {
		if !setDeferredDeliveryReplayPending(latest.DeferredDeliveries, markerID, true) {
			return fmt.Errorf("context recovery marker %q disappeared before launch", markerID)
		}
		return nil
	}); err != nil {
		return false, err
	}
	resetReplayPending := func() {
		_ = updateSessionResumeState(workDir, func(latest *sessionResumeState) error {
			setDeferredDeliveryReplayPending(latest.DeferredDeliveries, markerID, false)
			return nil
		})
	}
	binding = bindInitialDeferredLaunchMessage(sess, binding, msg)

	model, reasoningEffort := resolveModelAndReasoning(&state, ag)

	lc := &processLaunchContext{
		sess:                 sess,
		agent:                ag,
		runtime:              rt,
		binding:              binding,
		msg:                  msg,
		wake:                 buildWakeFromDeferredMessage(msg),
		requestContext:       ctx,
		mode:                 promptModeContextRecovery,
		skipResumeCheck:      true,
		pendingBehind:        0,
		savedModel:           model,
		savedReasoningEffort: reasoningEffort,
		startReservation:     startReservation,
		checkpointMsg:        checkpointMsgRecovery,
	}
	if err := d.resolve(lc); err != nil {
		resetReplayPending()
		d.abortSessionLaunch(sess, lc.startCfg)
		return false, err
	}
	if d.contextRecoveryLaunchObserver != nil {
		d.contextRecoveryLaunchObserver(lc.mode, lc.msg)
	}
	if _, err := d.launch(ctx, lc); err != nil {
		resetReplayPending()
		d.abortSessionLaunch(sess, lc.startCfg)
		return false, err
	}
	if isContextRecoveryMessage(msg) {
		if err := d.removeContextRecoveryMarker(workDir, msg.MessageID); err != nil {
			// The process is already launched with the marker claimed. Leave its
			// durable ReplayPending bit in place so a retry cannot launch the same
			// recovery marker twice; the next process boundary will settle it.
			return true, err
		}
	}
	return true, nil
}

func (d *Dispatcher) advanceDeferredContinuation(sess *interactiveSession, ap *agent.Process) (bool, error) {
	return d.advanceDeferredContinuationWithReservation(sess, ap, nil)
}

func (d *Dispatcher) advanceDeferredContinuationWithReservation(
	sess *interactiveSession,
	ap *agent.Process,
	startReservation *sessionStartReservation,
) (bool, error) {
	sess.mu.Lock()
	workDir := sess.workDir
	binding := sess.binding
	sess.mu.Unlock()
	state, err := loadSessionResumeState(workDir)
	if err != nil {
		return false, err
	}
	if len(state.DeferredDeliveries) == 0 {
		return false, nil
	}

	exitCode, _ := ap.ExitSummary()
	var (
		droppedMessage  types.InboxMessage
		droppedEntry    deferredDeliveryEntry
		droppedDelivery bool
	)
	err = updateSessionResumeState(workDir, func(state *sessionResumeState) error {
		replayIndex := firstReplayPendingIndex(state.DeferredDeliveries)
		if replayIndex < 0 {
			return nil
		}
		if exitCode == 0 {
			state.DeferredDeliveries = removeDeferredDeliveryAt(state.DeferredDeliveries, replayIndex)
		} else {
			entry := &state.DeferredDeliveries[replayIndex]
			entry.ReplayPending = false
			entry.ReplayAttempts++
			if entry.ReplayAttempts >= maxDeferredReplayAttempts {
				droppedEntry = *entry
				droppedMessage, droppedDelivery = deferredInboxMessage(entry.Message)
				state.DeferredDeliveries = removeDeferredDeliveryAt(state.DeferredDeliveries, replayIndex)
			}
		}
		return nil
	})
	if err != nil {
		return false, err
	}
	if droppedDelivery {
		d.emitRuntimeEvent(binding.channelID, binding.serverAgentID, types.RuntimeEvent{
			AgentID:    binding.serverAgentID,
			MessageID:  strings.TrimSpace(droppedMessage.MessageID),
			EventClass: consts.RuntimeEventClassWakeFailed,
			Target:     strings.TrimSpace(droppedMessage.Target),
			Detail: fmt.Sprintf(
				"deferred replay aborted after %d failed attempt(s); latest exit code=%d",
				droppedEntry.ReplayAttempts,
				exitCode,
			),
		})
		slog.Warn("dropping deferred delivery after repeated replay failures",
			"agent_id", ap.AgentID,
			"message_id", droppedEntry.MessageID,
			"attempts", droppedEntry.ReplayAttempts,
			"exit_code", exitCode)
	}

	// Context markers retain their dedicated fresh-session prompt mode. Check
	// them before the ordinary deferred replay path so a higher frontier that
	// followed an older pending marker is not downgraded to a generic replay.
	restarted, err := d.startContextRecoveryProcessWithReservation(sess, startReservation)
	if restarted || err != nil {
		return restarted, err
	}
	return d.startDeferredContinuationWithReservation(sess, startReservation)
}

func (d *Dispatcher) startDeferredContinuation(sess *interactiveSession) (bool, error) {
	return d.startDeferredContinuationWithReservation(sess, nil)
}

func (d *Dispatcher) startDeferredContinuationWithReservation(
	sess *interactiveSession,
	startReservation *sessionStartReservation,
) (bool, error) {
	sess.mu.Lock()
	if sess.process != nil {
		sess.mu.Unlock()
		return false, nil
	}
	ag := sess.agent
	rt := sess.runtime
	binding := sess.binding
	workDir := sess.workDir
	sess.mu.Unlock()

	state, err := loadSessionResumeState(workDir)
	if err != nil {
		return false, err
	}
	if state.LastStartConfig == nil || len(state.DeferredDeliveries) == 0 {
		return false, nil
	}
	currentPolicy := currentAgentRuntimePolicy(ag, rt)
	decision, err := applySessionResumeGuardToState(workDir, &state, currentPolicy)
	if err != nil {
		return false, err
	}
	logResumeGuardDecision(ag.ID, decision)

	runtimeName := string(rt.Type)
	if runtimeSupportsDeferredStdin(runtimeName) {
		return false, nil
	}

	var msg types.InboxMessage
	err = updateSessionResumeState(workDir, func(latest *sessionResumeState) error {
		if latest.LastStartConfig == nil || len(latest.DeferredDeliveries) == 0 {
			return nil
		}
		normalizeDeferredDeliveryOrder(latest.DeferredDeliveries)
		if firstReplayPendingIndex(latest.DeferredDeliveries) >= 0 {
			return nil
		}
		nextIndex := firstReplayAvailableIndex(latest.DeferredDeliveries)
		if nextIndex < 0 {
			return nil
		}
		var ok bool
		msg, ok = deferredInboxMessage(latest.DeferredDeliveries[nextIndex].Message)
		if !ok {
			return fmt.Errorf("deferred delivery %s is missing wake message fields", latest.DeferredDeliveries[nextIndex].MessageID)
		}
		latest.DeferredDeliveries[nextIndex].ReplayPending = true
		state = *latest
		return nil
	})
	if err != nil {
		return false, err
	}
	if strings.TrimSpace(msg.MessageID) == "" {
		return false, nil
	}
	binding = bindInitialDeferredLaunchMessage(sess, binding, msg)

	failReplay := func(err error) (bool, error) {
		messageID := strings.TrimSpace(msg.MessageID)
		_ = updateSessionResumeState(workDir, func(latest *sessionResumeState) error {
			for i := range latest.DeferredDeliveries {
				if strings.TrimSpace(latest.DeferredDeliveries[i].MessageID) == messageID {
					latest.DeferredDeliveries[i].ReplayPending = false
					break
				}
			}
			return nil
		})
		d.emitWakeFailed(binding.channelID, binding.serverAgentID, msg, err)
		return false, err
	}

	model, reasoningEffort := resolveModelAndReasoning(&state, ag)
	savedSessionID := strings.TrimSpace(state.LastStartConfig.SessionID)
	if savedSessionID == "" {
		savedSessionID = strings.TrimSpace(state.LastSessionID)
	}

	lc := &processLaunchContext{
		sess:                 sess,
		agent:                ag,
		runtime:              rt,
		binding:              binding,
		msg:                  msg,
		wake:                 buildWakeFromDeferredMessage(msg),
		requestContext:       context.Background(),
		mode:                 promptModeDeferred,
		pendingBehind:        len(state.DeferredDeliveries) - 1,
		savedModel:           model,
		savedReasoningEffort: reasoningEffort,
		savedSessionID:       savedSessionID,
		startReservation:     startReservation,
		checkpointMsg:        checkpointMsgDeferred,
	}
	if err := d.resolve(lc); err != nil {
		d.abortSessionLaunch(sess, lc.startCfg)
		return failReplay(err)
	}
	if _, err := d.launch(context.Background(), lc); err != nil {
		d.abortSessionLaunch(sess, lc.startCfg)
		return failReplay(err)
	}
	if isContextRecoveryMessage(msg) {
		if err := updateSessionResumeState(workDir, func(latest *sessionResumeState) error {
			latest.DeferredDeliveries = removeDeferredDeliveryByMessageID(latest.DeferredDeliveries, msg.MessageID)
			return nil
		}); err != nil {
			return false, err
		}
	}
	return true, nil
}