// Package dispatcher dispatches server messages to the appropriate agent session.
package dispatcher

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	gittransport "github.com/go-git/go-git/v5/plumbing/transport/http"

	"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/agent/runtimeevents"
	"github.com/tidbcloud/AgentTeam/desktop/internal/agent/runtimeproc"
	"github.com/tidbcloud/AgentTeam/desktop/internal/api/exec"
	"github.com/tidbcloud/AgentTeam/desktop/internal/consts"
	"github.com/tidbcloud/AgentTeam/desktop/internal/identitynotice"
	"github.com/tidbcloud/AgentTeam/desktop/internal/memorywatcher"
	"github.com/tidbcloud/AgentTeam/desktop/internal/messagesend"
	"github.com/tidbcloud/AgentTeam/desktop/internal/metrics"
	"github.com/tidbcloud/AgentTeam/desktop/internal/migrations"
	"github.com/tidbcloud/AgentTeam/desktop/internal/nativememory"
	"github.com/tidbcloud/AgentTeam/desktop/internal/profile"
	"github.com/tidbcloud/AgentTeam/desktop/internal/token"
	"github.com/tidbcloud/AgentTeam/desktop/internal/types"
	"github.com/tidbcloud/AgentTeam/desktop/internal/vault"
	"github.com/tidbcloud/AgentTeam/desktop/internal/wire"
	"github.com/tidbcloud/AgentTeam/desktop/internal/wire/manager"
	"github.com/tidbcloud/AgentTeam/desktop/internal/workspace"
	"github.com/tidbcloud/AgentTeam/internal/core/domain"
)

// Compile-time checks that Dispatcher implements the narrow ServerConnectionContext
// port and the agentteam-local interface.
var (
	_ types.ServerConnectionContext = (*Dispatcher)(nil)
)

// Dispatcher dispatches server messages to Instructions or interactive agent
// sessions based on message content. It manages session lifecycle, agent/runtime
// resolution, and the MCP bridge state for interactive runtimes.
type Dispatcher struct {
	// pendingSendScopes binds the irreversible identity of a freshly minted MCP
	// bearer to its in-flight Process during the launch window before
	// PublishStart makes it the session's current process. Guarded by
	// pendingSendScopeMu; only the message_send scope port reads it.
	pendingSendScopeMu  sync.Mutex
	pendingSendScopes   map[string]pendingSendScopeBinding
	pendingTurnTargetMu sync.Mutex
	pendingTurnTargets  map[string]pendingTurnTargetBinding

	// Focused service facets. Embedded for API compatibility while moving
	// high-churn methods off the Dispatcher receiver.
	*runtimeProcessService
	*memoryService
	*skillService
	*knowledgeBaseService
	*execAuthorizationService
	*extensionRuntimeService

	// Channel-facing configuration and adapters.
	serverURL          string
	serverConnections  *manager.Manager
	agentOverrides     *AgentOverrides
	instructions       *InstructionHandler
	execAuthorizations *exec.RuntimeAuthorizationStore

	// Runtime session lifecycle and process ownership.
	sessionManager         *SessionManager
	runtimeProcessRegistry *runtimeproc.Registry
	runtimeProcessJanitor  *runtimeproc.Janitor
	runtimeJanitorOnce     sync.Once
	// Process callbacks outlive membership in sessionManager: onExit
	// intentionally clears that index before its final durable writes. Keep a
	// separate owner registry through callback completion so shutdown cannot miss
	// an already-unpublished callback.
	processExitCallbackMu     sync.Mutex
	processExitCallbackClosed bool
	processExitCallbackOwners map[*processExitCallbackOwner]struct{}
	// runtimeOwnershipRetryCtx owns every daemon continuation/reconciliation
	// retry goroutine. The launch mutex makes Add-vs-Wait safe: shutdown marks
	// the registry closed before taking a Wait snapshot, so no worker can appear
	// after the join has begun.
	runtimeOwnershipRetryMu     sync.Mutex
	runtimeOwnershipRetryWG     sync.WaitGroup
	runtimeOwnershipRetryCtx    context.Context
	runtimeOwnershipRetryCancel context.CancelFunc
	runtimeOwnershipRetryClosed bool
	// Runtime event fan-out and dashboard telemetry.
	runtimeEventBus       runtimeevents.Bus
	agentteamReporterSink *AgentTeamReporterSink
	trajectoryHub         *trajectoryHub
	trajectorySink        *TrajectorySink
	metricsSink           *MetricsSink

	// Optional memory watcher integration. When memoryAccess is nil the
	// dispatcher runs in pre-Phase-2.3 mode (no per-agent git sync).
	memoryAccess   MemoryAccessProvider
	memoryAccessMu sync.RWMutex
	// memorySyncStatuses is the daemon-local projection used when no active
	// watcher is published (lookup/clone failures, pre-clone, or teardown).
	// Lookup degradation takes precedence over an active watcher; otherwise the
	// active watcher snapshot wins. This state is never persisted.
	memorySyncStatusMu sync.RWMutex
	memorySyncStatuses map[string]memorywatcher.Status
	// memoryLookupStatuses overlays watcher/clone state while the current
	// binding cannot be resolved authoritatively. It is kept separate so an
	// active watcher's callbacks cannot hide access degradation or redirect an
	// explicit retry into a push-only attempt.
	memoryLookupStatuses map[string]memorywatcher.Status
	// memoryIdleRetries owns automatic recovery when no watcher exists or an
	// active session's authoritative binding lookup is degraded. Timers are per
	// agent, coalesced, and cancelled when recovery completes, access is revoked,
	// or dispatcher shutdown closes mutation admission.
	memoryIdleRetryMu     sync.Mutex
	memoryIdleRetries     map[string]*idleMemoryRetry
	memoryIdleRetryBlocks map[string]struct{}
	memoryIdleRetrySlots  chan struct{}
	memoryIdleRetryClosed bool
	// memoryLookupMu serializes authoritative Lookup+apply per agent. Provider
	// calls remain concurrent across agents, while a slow older Bound result
	// can never apply after a newer Unbound/Unknown result for the same agent.
	memoryLookupMu sync.Map
	// memoryWatcherMutationMu is the admission/publication fence for every path
	// that can construct and publish a session memory watcher. Shutdown closes
	// admission before retiring sessions, snapshots the already-admitted owners,
	// and waits for their clone/construct/publication-or-cleanup work before the
	// final watcher drain. Explicit owner channels avoid WaitGroup Add/Wait races.
	memoryWatcherMutationMu         sync.Mutex
	memoryWatcherMutationClosed     bool
	memoryWatcherMutationGeneration uint64
	memoryWatcherMutationOwners     map[*memoryWatcherMutationOwner]struct{}

	// Optional memory search registry. When nil the dispatcher runs without
	// search derivative reconciliation — equivalent to pre-Task-5c behavior.
	// Wired by EnableMemorySearch; daemon calls it from wireMemoryProvider so
	// tests that construct a bare Dispatcher via New("") keep the registry off.
	//
	// All reads MUST go through snapshotMemorySearch(); the pointer may be
	// swapped out by a subsequent EnableMemorySearch call (used in tests that
	// rebind the resolver/scan onto the same dispatcher).
	//
	// Replacement semantics (Round 10 blocker 1): EnableMemorySearch serializes
	// the full transaction (take-old → set-swap/nil → closeAll-old → publish-
	// next → close-swap) under memorySearchReplace so concurrent Enables
	// linearize by mutex acquisition order — the second Enable only proceeds
	// after the first has fully released (including closeAll), guaranteeing
	// the previous registry's per-binding owner locks are released before the
	// next registry's first getOrCreate. Refresh observes memorySearch=nil +
	// memorySearchSwap!=nil during replacement and bounded-waits on the swap
	// chan (via ctx) so the SAME Refresh call converges on the new registry
	// rather than degrading silently and waiting up to 30min for the next
	// periodic trigger.
	//
	// memorySearchClosed is set by Dispatcher.Close (under memorySearchReplace
	// + memorySearchMu) and prevents any in-flight or subsequent Enable from
	// publishing after shutdown.
	memorySearch        *memorySearchRegistry
	memorySearchMu      sync.RWMutex
	memorySearchReplace sync.Mutex
	memorySearchSwap    chan struct{}
	memorySearchClosed  bool

	// Optional managed skill sync. When skillSyncer is nil the dispatcher
	// runs without MeshX skill repository sync — equivalent to pre-MeshX-skill
	// behavior.
	skillSyncer   SkillSyncer
	skillSyncerMu sync.RWMutex

	// Test seam for daemon-owned accepted-delivery continuation retries.
	runtimeOwnershipContinuationStarter func(*interactiveSession) (bool, error)
	runtimeOwnershipContinuationDelay   func(int) time.Duration
	runtimeOwnershipTerminalPersist     func(agent.DeliveryOwnershipStore, agent.DeliveryOwnershipRecord) error
	// Narrow seams for deterministic context-recovery lifecycle tests. Production
	// leaves both nil: marker removal uses the durable resume-state transaction,
	// and launch observation is not part of runtime behavior.
	contextRecoveryMarkerRemoval  func(string, string) error
	contextRecoveryLaunchObserver func(promptMode, types.InboxMessage)
	// Test seam for deterministic detached recovery lifecycle ordering.
	recoveryPointDetachedRestorer func(context.Context, agent.DetachedRecoveryConfig, string, string) error
	// Test seam for detached recovery artifact cleanup routing.
	recoveryArtifactsDetachedCleaner func(context.Context, agent.DetachedRecoveryConfig, []agent.RecoveryArtifact) error
	// Test seam for deterministic memory-watcher publication races. Production
	// leaves it nil. It runs after Watcher.Start and before the admission /
	// generation / retired-session publication fence.
	memoryWatcherBeforePublish func(context.Context, *interactiveSession)
	// Test seam after shutdown has detached a published watcher from its session
	// but before FlushAndClose/final push completes. Production leaves it nil.
	memoryWatcherShutdownAfterDetach func(*interactiveSession)
	// Test seam after a final-removal path has fenced watcher publication but
	// before it drains the watcher and removes the manager entry.
	memoryWatcherFinalRemovalAfterFence func(*interactiveSession)
	// Test seam after a replacement has detached its old watcher but before that
	// watcher finishes FlushAndClose/final push. Production leaves it nil.
	memoryWatcherReplacementAfterOldDetach func(*interactiveSession)
	// Test seam before a binding refresh waits for an unpublished session start.
	memoryRefreshBeforeWaitForStart func()
	// Test seam for the session-workdir directory durability boundary. Production
	// uses the same directory fsync primitive as .daemon-state transactions.
	sessionWorkDirSync func(string) error

	// Optional managed KB sync. When knowledgeBaseSyncer is nil the dispatcher
	// does not clone MeshX KB repositories into agent kb/ directories.
	knowledgeBaseSyncer   KnowledgeBaseSyncer
	knowledgeBaseSyncerMu sync.RWMutex
	tokenAnalysisToken    func(string) (string, error)
}

type interactiveSession struct {
	mu sync.Mutex

	sessionKey string
	agentID    string
	agent      agent.Config
	runtime    runtimeconfig.Runtime
	binding    conversationBinding
	workDir    string
	startedAt  time.Time
	mcpToken   string

	process *agent.Process
	// processUsesWorkspaceRootMemory records the exact Memory prompt selected
	// for the current Process generation. RefreshMemoryAccess uses it to retire
	// only runtimes that were compiled before managed memory/MEMORY.md existed.
	processUsesWorkspaceRootMemory bool
	runtimeSessionID               string
	history                        []types.HistoryMessage
	nextSeq                        int64
	serverAgentID                  string
	activity                       string // latest activity: "online", "thinking", "working", etc.
	activityDetail                 string // latest activity detail: "Idle", tool name, etc.
	activeTaskTarget               types.ActiveTaskTarget
	ephemeral                      bool
	// durableDelivery keeps externally routed concrete workflow inputs on the
	// same two-phase ownership path as ordinary sessions while preserving the
	// workflow session's hidden/ephemeral UI semantics. Ephemeral must not mean
	// that an accepted server delivery can be sent from Spawn before a durable
	// owner and Process publication exist.
	durableDelivery     bool
	resetting           bool
	retired             atomic.Bool
	suppressCheckpoints bool
	// runtimeStartFingerprint is the immutable process-start identity for
	// runtime-affecting configuration. It intentionally does not change when
	// RefreshRunningAgentConfig updates sess.agent for display/read-model use.
	runtimeStartFingerprint              string
	ownershipContinuationRetryRunning    bool
	ownershipContinuationRetryGeneration uint64
	ownershipContinuationRetryBlocked    bool
	ownershipContinuationRetryCtx        context.Context
	ownershipContinuationRetryCancel     context.CancelFunc
	ownershipContinuationRetryWG         sync.WaitGroup
	// Intentional restart and the asynchronous Process exit callback share one
	// terminal telemetry claim. The restart owner may observe process=nil after
	// the exit callback has cleared it; terminalReportDone lets retirement wait
	// for that already-claimed report before replacement/removal, while
	// restartRetirementInFlight tells the callback not to terminal-fence the
	// same session that retirement may reuse for durable ownership continuation.
	terminalReportClaimed     bool
	terminalReportDone        chan struct{}
	terminalReportWillStop    bool
	restartRetirementInFlight bool
	// runtimeStartReplacement is set when a restart was requested because the
	// authoritative runtime launch configuration changed. Durable ownership
	// continuation must then prefer the replacement agent model/effort over the
	// previous Process's saved launch values. It remains set across a failed
	// continuation so the retry worker cannot resurrect the stale configuration.
	runtimeStartReplacement bool

	// memoryClone + watcher are non-nil only when the dispatcher has a
	// MemoryAccessProvider and the agent had a memory grant at session
	// start. Both are cleared under sess.mu by stopMemoryWatcher.
	// memoryPushAuth retains the provider-supplied git auth method so the
	// final push on session teardown can flush in-flight commits without
	// going back through the provider. In production this is a refreshable
	// AGS sync lease auth method.
	memoryClone      *vault.Clone
	watcher          *memorywatcher.Watcher
	memoryPushAuth   gittransport.AuthMethod
	memoryProtocol   MemorySyncProtocol
	memoryGeneration uint64
	// memoryContentVersionID is the exact native cloud head admitted for this
	// watcher. Same-generation restores must replace the watcher rather than be
	// mistaken for an unchanged binding.
	memoryContentVersionID string
	memoryNativeSync       *nativememory.Sync
	// memoryWatcherDrain tracks a watcher that has been detached from the three
	// published fields above but is still inside FlushAndClose/final push. A
	// concurrent/repeated teardown joins this owner instead of mistaking nil
	// fields for a completed drain and disabling memory search too early.
	memoryWatcherDrain *memoryWatcherDrain
	// memoryWatcherReplacementFenced prevents RefreshMemoryAccess from
	// reattaching an old session while a reserved replacement drains its watcher
	// and prepares the same clone for the new session.
	memoryWatcherReplacementFenced bool
	// memoryWatcherTerminalFenced permanently closes watcher publication once
	// the owning Process is terminal. It may be installed before a slow final
	// drain/removal, or when removal is deferred behind a competing start owner.
	// Whole-session retirement remains atomic with SessionManager publication /
	// removal so deliveries never spin on a retired-but-still-published pointer.
	memoryWatcherTerminalFenced bool
	// pendingLegacyPreferencesMigration is created after general local import is
	// durable. Its final legacy-source probe runs behind the last fallback writer
	// barrier and before the authoritative Process can dispatch its initial turn.
	pendingLegacyPreferencesMigration *legacyUserPreferencesMigrationFinalization

	// managedSkillPaths is the launch-time snapshot of managed skill repository
	// paths the runtime materializer should expose for this session. It is
	// populated from the SkillSyncer's cache for this agent's owning AgentTeam
	// source scope — never from the global profile.ManagedSkillRepositories()
	// records, which are an audit/sync log rather than an authority for current
	// bindings. Empty/nil means "no cached managed skills for this scope"
	// (e.g. unbound, revoked, not refreshed yet, or sync error).
	managedSkillPaths []string
}

func (is *interactiveSession) persistsRuntimeState() bool {
	return is != nil && (!is.ephemeral || is.durableDelivery)
}

type conversationBinding struct {
	channelID              string
	targetScopeType        string
	conversationKey        string
	targetID               string
	meta                   types.Metadata
	serverAgentID          string
	serverAgentName        string
	serverAgentDisplayName string
	serverRuntime          string
	serverModel            string
	serverReasoningEffort  string
	serverSessionID        string
	serverMaterialization  *types.WorkspaceMaterialization
	unreadSummary          map[string]int // unread counts for idle/resume prompts when no wake message is present
	activeTaskTarget       types.ActiveTaskTarget
}

// Dependencies are the required domain services owned by the daemon composition root.
type Dependencies struct {
	TokenAnalysisToken func(string) (string, error)
}

// New creates a Dispatcher with an isolated connection manager.
func New(serverURL string, dependencies Dependencies) (*Dispatcher, error) {
	d, err := newDispatcher(serverURL, dependencies)
	if err != nil {
		return nil, err
	}
	d.serverConnections = manager.New(d, wire.New)
	return d, nil
}

// NewWithConnectionManager creates a Dispatcher using the daemon-owned manager.
func NewWithConnectionManager(serverURL string, connections *manager.Manager, dependencies Dependencies) (*Dispatcher, error) {
	if connections == nil {
		return nil, fmt.Errorf("dispatcher connection manager is required")
	}
	d, err := newDispatcher(serverURL, dependencies)
	if err != nil {
		return nil, err
	}
	d.serverConnections = connections
	return d, nil
}

func newDispatcher(serverURL string, dependencies Dependencies) (*Dispatcher, error) {
	overrides := NewAgentOverrides()
	runtimeRegistry := newRuntimeProcessRegistry()
	runtimeOwnershipRetryCtx, runtimeOwnershipRetryCancel := context.WithCancel(context.Background())
	d := &Dispatcher{
		serverURL:                       strings.TrimRight(serverURL, "/"),
		agentOverrides:                  overrides,
		execAuthorizations:              exec.NewRuntimeAuthorizationStore(token.Default()),
		sessionManager:                  NewSessionManager(),
		runtimeProcessRegistry:          runtimeRegistry,
		memoryWatcherMutationGeneration: 1,
		memoryWatcherMutationOwners:     make(map[*memoryWatcherMutationOwner]struct{}),
		processExitCallbackOwners:       make(map[*processExitCallbackOwner]struct{}),
		runtimeOwnershipRetryCtx:        runtimeOwnershipRetryCtx,
		runtimeOwnershipRetryCancel:     runtimeOwnershipRetryCancel,
		runtimeEventBus:                 runtimeevents.NewBus(),
		trajectoryHub:                   newTrajectoryHub(),
		tokenAnalysisToken:              dependencies.TokenAnalysisToken,
	}
	d.initServices()
	d.runtimeProcessJanitor = newRuntimeProcessJanitor(d, runtimeRegistry)
	d.agentteamReporterSink = NewAgentTeamReporterSink(d.eventReporter, d.agentTurnStateForServer)
	d.trajectorySink = NewTrajectorySink(d.trajectoryHub)
	d.metricsSink = NewMetricsSink()
	d.instructions = NewInstructionHandler(d, overrides)
	return d, nil
}

// ExecAuthorizer exposes the capability store through the narrow api/exec
// Authorizer interface. It contains no business execution or dispatch method.
func (d *Dispatcher) ExecAuthorizer() *exec.RuntimeAuthorizationStore {
	if d == nil {
		return nil
	}
	return d.execAuthorizations
}

// teardownSessionProcess performs the authoritative cleanup for a session whose
// process is being stopped: flush the watcher, revoke runtime capabilities,
// stop the process, then finish any legacy import behind that stop barrier.
func (d *Dispatcher) teardownSessionProcess(sess *interactiveSession, process *agent.Process) {
	d.stopMemoryWatcherAndRetry(sess)
	d.stopSessionProcessAfterMemoryDrain(sess, process)
	d.finalizePendingLegacyPreferencesMigration(context.Background(), sess, true)
}

func (d *Dispatcher) stopSessionProcessAfterMemoryDrain(sess *interactiveSession, process *agent.Process) {
	revokeSessionMCPToken(sess)
	d.revokeExecProcessAuthorization(process)
	process.Stop()
}

// abortSessionLaunch performs cleanup when session startup fails before the
// process is created: flush the memory watcher, revoke the MCP token, and
// release the Exec runtime authorization.
func (d *Dispatcher) abortSessionLaunch(sess *interactiveSession, startCfg drivercontract.LaunchConfig) {
	d.stopMemoryWatcherAndRetry(sess)
	revokeSessionMCPToken(sess)
	d.revokeExecAuthorizationConfig(startCfg.Exec)
	revokeTokenAnalysisConfig(startCfg)
}

func waitForSessionStartOwners(ctx context.Context, owners []<-chan struct{}) error {
	if ctx == nil {
		ctx = context.Background()
	}
	for _, ownerDone := range owners {
		if ownerDone == nil {
			continue
		}
		select {
		case <-ownerDone:
		case <-ctx.Done():
			return ctx.Err()
		}
	}
	return nil
}

func (d *Dispatcher) waitForShutdownOwners(ctx context.Context, sessionOwners []<-chan struct{}, operation string) {
	if err := waitForSessionStartOwners(ctx, sessionOwners); err != nil {
		slog.Warn(operation+": session start owners did not quiesce before timeout", "error", err)
	}
	if err := d.waitForProcessExitCallbackOwners(ctx); err != nil {
		slog.Warn(operation+": process exit callbacks did not quiesce before timeout", "error", err)
	}
	if err := d.stopRuntimeOwnershipRetryWorkers(ctx); err != nil {
		slog.Warn(operation+": runtime ownership retry workers did not quiesce before timeout", "error", err)
	}
}

// Close is the strong local shutdown boundary. It stops every interactive
// Process and joins its exit callback before returning.
func (d *Dispatcher) Close() {
	// Server connection admission is the outermost ingress fence. Close it before
	// any session teardown so no dashboard mutation or initial Start can
	// publish a connector after the shutdown snapshot.
	connectionShutdown := d.serverConnections.BeginShutdown()
	connectionShutdown.StartDrain(context.Background())
	trajectoryDone := d.trajectoryHub.BeginStop()
	memoryOwners := d.closeMemoryWatcherMutationAdmission()
	d.closeExecAuthorizationAdmission()
	d.closeProcessExitCallbackAdmission()
	startOwners, sessions := d.sessionManager.CloseAndSnapshot(true)
	d.cancelRuntimeOwnershipRetryWorkers()
	if d.runtimeEventBus != nil {
		d.runtimeEventBus.Close()
	}

	// A watcher can temporarily belong to an unpublished session start or an
	// in-flight access refresh/rebind. Join both owner registries before draining
	// the session set captured atomically with manager admission close; no
	// watcher publisher or concurrent remover can escape that boundary.
	_ = waitForMemoryWatcherQuiescence(context.Background(), memoryOwners, startOwners)

	sessionProcesses := make([]*agent.Process, 0, len(sessions))
	for _, sess := range sessions {
		sess.mu.Lock()
		process := sess.process
		sess.mu.Unlock()
		if process != nil {
			sessionProcesses = append(sessionProcesses, process)
			d.teardownSessionProcess(sess, process)
			continue
		}
		// Secondary drain (blocker 5): a session can have a watcher with no
		// process — pre-clone path, failed launch, or aborted start. Always call
		// the drain primitive rather than branching on sess.watcher: a concurrent
		// teardown may already have detached the fields while its FlushAndClose /
		// final push is still represented by memoryWatcherDrain. The idempotent
		// call joins that owner before search shutdown.
		d.stopMemoryWatcherAndRetry(sess)
	}

	// Memory watchers were drained above as part of session teardown (both the
	// process-bearing path and the watcher-only drain). Disable the search
	// registry AFTER that drain so no new Notify event arrives against a
	// closing reconciler. disableMemorySearch is idempotent and serialized
	// against any in-flight EnableMemorySearch via memorySearchReplace.
	d.disableMemorySearch()

	for _, process := range sessionProcesses {
		if !process.WaitForExit(5 * time.Second) {
			slog.Warn("dispatcher close: process did not exit before timeout", "agent_id", process.AgentID)
		}
	}
	// Close is the strong local teardown boundary used by tests and embedders.
	// StopForShutdown is the explicitly cancellable variant for daemon exit.
	d.waitForShutdownOwners(context.Background(), startOwners, "dispatcher close")
	// A launch can own an Exec authorization before its Process is published into the
	// session snapshot. Once launch admission is drained, sweep those orphanable
	// credentials and strongly join their rotation owners.
	d.revokeAllExecAuthorizations(context.Background())
	for _, process := range sessionProcesses {
		process.WaitForExitCallback(context.Background())
	}
	_ = connectionShutdown.Wait(context.Background())
	<-trajectoryDone
}

// ---------------------------------------------------------------------------
// Server runtime manager & MCP registry access
// ---------------------------------------------------------------------------

// ServerConnections returns the running AgentTeam server runtime accessor.
func (d *Dispatcher) ServerConnections() types.ServerConnectionAccessor {
	return d.serverConnections
}

// SubscribeSessionStateChanges subscribes to Dispatcher-owned session state
// changes. Consumers use the event bus instead of registering ad-hoc callbacks
// so future session-state observers stay decoupled from Dispatcher internals.
func (d *Dispatcher) SubscribeSessionStateChanges(subscriberName string) runtimeevents.Subscription {
	if d == nil || d.runtimeEventBus == nil {
		return nil
	}
	name := strings.TrimSpace(subscriberName)
	if name == "" {
		name = "session_state"
	}
	return d.runtimeEventBus.Subscribe(name, runtimeevents.FilterKind(runtimeevents.KindSessionStateChanged), 64, runtimeevents.DropOldest)
}

// SubscribeConsoleRefreshEvents subscribes to daemon state changes that require
// the dashboard console bundle to be refreshed.
func (d *Dispatcher) SubscribeConsoleRefreshEvents(subscriberName string) runtimeevents.Subscription {
	if d == nil || d.runtimeEventBus == nil {
		return nil
	}
	name := strings.TrimSpace(subscriberName)
	if name == "" {
		name = "console_refresh"
	}
	return d.runtimeEventBus.Subscribe(
		name,
		runtimeevents.FilterKind(
			runtimeevents.KindAgentCatalogChanged,
			runtimeevents.KindSessionStateChanged,
		),
		64,
		runtimeevents.DropOldest,
	)
}

func (d *Dispatcher) publishSessionStateChanged(agentID string) {
	if d == nil || d.runtimeEventBus == nil {
		return
	}
	d.runtimeEventBus.Publish(context.Background(), runtimeevents.NewEvent(
		runtimeevents.KindSessionStateChanged,
		runtimeevents.WithContext(runtimeevents.EventContext{
			AgentID: strings.TrimSpace(agentID),
		}),
	))
}

// PublishAgentCatalogChanged signals that AgentTeam-synced agent definitions have
// changed and dashboard aggregate data should be refetched.
func (d *Dispatcher) PublishAgentCatalogChanged(serverID string) {
	if d == nil || d.runtimeEventBus == nil {
		return
	}
	d.runtimeEventBus.Publish(context.Background(), runtimeevents.NewEvent(
		runtimeevents.KindAgentCatalogChanged,
		runtimeevents.WithContext(runtimeevents.EventContext{
			ServerID: strings.TrimSpace(serverID),
		}),
	))
}

// Start starts all enabled channels and begins accepting connections.
func (d *Dispatcher) Start(ctx context.Context) error {
	d.startRuntimeProcessJanitor(ctx)
	return d.serverConnections.Start(ctx)
}

// ---------------------------------------------------------------------------
// Trajectory fan-out (for dashboard SSE)
// ---------------------------------------------------------------------------

// SubscribeTrajectory returns a channel that receives trajectory events
// for the given agent, pre-filled with recently cached history.
func (d *Dispatcher) SubscribeTrajectory(agentID string) <-chan types.TrajectoryLogItem {
	if d == nil || d.trajectoryHub == nil {
		return nil
	}
	return d.trajectoryHub.Subscribe(agentID)
}

// SubscribeLiveTrajectory returns a channel that receives only new trajectory
// events for the given agent.
func (d *Dispatcher) SubscribeLiveTrajectory(agentID string) <-chan types.TrajectoryLogItem {
	if d == nil || d.trajectoryHub == nil {
		return nil
	}
	return d.trajectoryHub.SubscribeLive(agentID)
}

// SubscribeAllLiveTrajectory returns a channel that receives only new trajectory
// events for all agents.
func (d *Dispatcher) SubscribeAllLiveTrajectory() <-chan types.TrajectoryLogItem {
	if d == nil || d.trajectoryHub == nil {
		return nil
	}
	return d.trajectoryHub.SubscribeAllLive()
}

// UnsubscribeTrajectory removes and closes a trajectory subscriber channel.
func (d *Dispatcher) UnsubscribeTrajectory(agentID string, ch <-chan types.TrajectoryLogItem) {
	if d == nil || d.trajectoryHub == nil {
		return
	}
	d.trajectoryHub.Unsubscribe(agentID, ch)
}

// UnsubscribeAllTrajectory removes and closes an all-agent trajectory subscriber channel.
func (d *Dispatcher) UnsubscribeAllTrajectory(ch <-chan types.TrajectoryLogItem) {
	if d == nil || d.trajectoryHub == nil {
		return
	}
	d.trajectoryHub.UnsubscribeAll(ch)
}

// TrajectoryHistory returns recent trajectory items ordered newest-first.
func (d *Dispatcher) TrajectoryHistory(agentID string, limit int, before int64) []types.TrajectoryLogItem {
	if d == nil || d.trajectoryHub == nil {
		return nil
	}
	return d.trajectoryHub.History(agentID, limit, before)
}

// LatestTrajectoryByAgent returns the latest trajectory item for each active agent.
func (d *Dispatcher) LatestTrajectoryByAgent(agentIDs []string) map[string]types.TrajectoryLogItem {
	if d == nil || d.trajectoryHub == nil {
		return nil
	}
	return d.trajectoryHub.LatestByAgent(agentIDs)
}

// ---------------------------------------------------------------------------
// Message routing
// ---------------------------------------------------------------------------

// Dispatch routes an incoming server message to either Instructions (for
// slash commands) or dispatches it to an interactive agent session (for normal
// messages). It records aggregate server message stats for every message and per-agent
// stats once the target agent is known.
func (d *Dispatcher) Dispatch(ctx context.Context, ch types.ServerConnection, msg *types.Message) {
	_ = d.dispatch(ctx, ch, msg, dispatchModeDefault)
}

// DispatchForDeliveryAck routes a server-delivered message and reports whether
// the local agent handoff was accepted.
func (d *Dispatcher) DispatchForDeliveryAck(ctx context.Context, ch types.ServerConnection, msg *types.Message) error {
	if d == nil {
		return errors.New("dispatcher is nil")
	}
	return d.dispatch(ctx, ch, msg, dispatchModeDeliveryAck)
}

type dispatchMode int

const (
	dispatchModeDefault dispatchMode = iota
	dispatchModeDeliveryAck
)

var errDeliveryRequiresAcceptingProcess = errors.New("agent delivery requires an existing process accepting deliveries")

func (d *Dispatcher) dispatch(ctx context.Context, ch types.ServerConnection, msg *types.Message, mode dispatchMode) error {
	if msg == nil {
		return nil
	}
	if ctx == nil {
		ctx = context.Background()
	}

	slog.Debug("message received",
		"channel", msg.Input.Channel,
		"prompt_length", len([]rune(msg.Input.Prompt)))

	// agent:deliver is a handoff to the server-selected live Process, not a
	// command or cold-start path. In particular, a slash-prefixed delivered
	// body is runtime input and must not be interpreted by the daemon's local
	// instruction handler.
	if mode == dispatchModeDeliveryAck {
		agentID, err := d.deliverToExistingProcess(ctx, msg)
		recordInboundStats(msg, agentID)
		return err
	}

	handled, err := d.instructions.Handle(ctx, ch, msg)
	if err != nil {
		slog.Warn("instruction handling failed", "channel", msg.Input.Channel, "error", err)
	}
	if handled {
		recordInboundStats(msg, "")
		return err
	}

	return d.handleMessage(ctx, ch, msg, mode)
}

// Stop gracefully shuts down the Dispatcher by stopping all active sessions.
func (d *Dispatcher) Stop() {
	if d == nil {
		return
	}
	d.stopAll()
}

// StopForShutdown is the shutdown-optimized variant. It stops channels FIRST
// (to prevent reconnect races), then tears down all sessions in PARALLEL with
// short timeouts. This ensures the daemon exits within seconds rather than
// blocking on sequential per-session memory pushes.
func (d *Dispatcher) StopForShutdown(ctx context.Context) {
	if ctx == nil {
		ctx = context.Background()
	}
	shutdownCtx := ctx
	var cancelShutdown context.CancelFunc
	if _, bounded := shutdownCtx.Deadline(); !bounded {
		shutdownCtx, cancelShutdown = context.WithTimeout(shutdownCtx, 5*time.Second)
		defer cancelShutdown()
	}
	// Closing connector admission is synchronous and precedes every other
	// teardown fence. Actual connector drainage is non-blocking here and joins as
	// one task under shutdownCtx alongside Process and memory teardown.
	connectionShutdown := d.serverConnections.BeginShutdown()
	connectionShutdown.StartDrain(shutdownCtx)
	trajectoryDone := d.trajectoryHub.BeginStop()
	memoryOwners := d.closeMemoryWatcherMutationAdmission()
	d.closeExecAuthorizationAdmission()
	d.closeProcessExitCallbackAdmission()
	startOwners, sessions := d.sessionManager.CloseAndSnapshot(true)
	d.cancelRuntimeOwnershipRetryWorkers()
	if d.runtimeEventBus != nil {
		d.runtimeEventBus.Close()
	}

	// Every potentially blocking shutdown operation is launched independently and
	// observed under the same caller/default deadline. A connector or Process
	// that ignores cancellation may continue best-effort in its goroutine after
	// the deadline, but it cannot make this API exceed the total shutdown budget.
	// Channel tasks are appended first so reconnects are fenced as early as
	// possible instead of one stuck connector preventing all runtime teardown.
	tasks := make([]dispatcherShutdownTask, 0, len(sessions)+3)
	tasks = append(tasks, dispatcherShutdownTask{
		name: "server connection manager",
		run: func(ctx context.Context) {
			if err := connectionShutdown.Wait(ctx); err != nil {
				slog.Warn("dispatcher shutdown: server connections did not quiesce before deadline", "error", err)
			}
		},
	})
	tasks = append(tasks, dispatcherShutdownTask{
		name: "trajectory gc",
		run: func(ctx context.Context) {
			select {
			case <-trajectoryDone:
			case <-ctx.Done():
			}
		},
	})
	for _, sess := range sessions {
		sess.mu.Lock()
		process := sess.process
		sess.mu.Unlock()
		if process == nil {
			continue
		}
		sess := sess
		tasks = append(tasks, dispatcherShutdownTask{
			name: "interactive process " + strings.TrimSpace(process.AgentID),
			run: func(ctx context.Context) {
				revokeSessionMCPToken(sess)
				d.revokeExecProcessAuthorization(process)
				process.Stop()
				if !waitForProcessExitUntil(ctx, process) {
					slog.Warn("dispatcher shutdown: process did not exit before deadline", "agent_id", process.AgentID, "error", ctx.Err())
				}
				if !process.WaitForExitCallback(ctx) {
					slog.Warn("dispatcher shutdown: process exit callback did not finish before deadline", "agent_id", process.AgentID, "error", ctx.Err())
				}
			},
		})
	}
	tasks = append(tasks, dispatcherShutdownTask{
		name: "memory watcher owners, final drain, and search registry",
		run: func(ctx context.Context) {
			d.shutdownMemoryWatchersAfterOwnerDrain(ctx, memoryOwners, startOwners, sessions)
		},
	})
	tasks = append(tasks, dispatcherShutdownTask{
		name: "dispatcher reservation/retry/signal owners",
		run: func(ctx context.Context) {
			d.waitForShutdownOwners(ctx, startOwners, "dispatcher shutdown")
			// Launch admission is now quiescent. Revoke any Exec configuration
			// that was created before Process publication and join its rotator.
			d.revokeAllExecAuthorizations(ctx)
		},
	})

	if pending := waitForDispatcherShutdownTasks(shutdownCtx, tasks); len(pending) > 0 {
		slog.Warn("dispatcher shutdown deadline reached; cleanup continues best-effort", "pending", pending, "error", shutdownCtx.Err())
	}
}

type dispatcherShutdownTask struct {
	name string
	run  func(context.Context)
}

// waitForDispatcherShutdownTasks starts every shutdown operation concurrently
// and waits only until all finish or ctx is done. The completion channel is
// buffered so a task that ignores cancellation can finish after this function
// returns without blocking on result publication.
func waitForDispatcherShutdownTasks(ctx context.Context, tasks []dispatcherShutdownTask) []string {
	if ctx == nil {
		ctx = context.Background()
	}
	completed := make(chan int, len(tasks))
	pending := make(map[int]string, len(tasks))
	for index, task := range tasks {
		if task.run == nil {
			continue
		}
		name := strings.TrimSpace(task.name)
		if name == "" {
			name = fmt.Sprintf("shutdown task %d", index)
		}
		pending[index] = name
		go func(index int, run func(context.Context)) {
			run(ctx)
			completed <- index
		}(index, task.run)
	}

	for len(pending) > 0 {
		select {
		case index := <-completed:
			delete(pending, index)
		case <-ctx.Done():
			// Prefer completions that raced the deadline so diagnostics contain
			// only tasks that really remained outstanding at this boundary.
			for {
				select {
				case index := <-completed:
					delete(pending, index)
				default:
					result := make([]string, 0, len(pending))
					for _, name := range pending {
						result = append(result, name)
					}
					sort.Strings(result)
					return result
				}
			}
		}
	}
	return nil
}

func waitForProcessExitUntil(ctx context.Context, process *agent.Process) bool {
	if process == nil {
		return true
	}
	if ctx == nil {
		ctx = context.Background()
	}
	if process.WaitForExit(0) {
		return true
	}
	ticker := time.NewTicker(10 * time.Millisecond)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return process.WaitForExit(0)
		case <-ticker.C:
			if process.WaitForExit(0) {
				return true
			}
		}
	}
}

// ---------------------------------------------------------------------------
// Message dispatch
// ---------------------------------------------------------------------------

// handleMessage resolves an agent/runtime and delivers the inbound message
// to the bound interactive session.
func (d *Dispatcher) handleMessage(ctx context.Context, ch types.ServerConnection, msg *types.Message, mode dispatchMode) error {
	ag, rt, err := d.resolveAgentAndRuntime(msg)
	if err != nil {
		recordInboundStats(msg, "")
		if mode != dispatchModeDeliveryAck {
			d.replyError(ctx, ch, msg, err)
		}
		return err
	}

	recordInboundStats(msg, ag.ID)
	if _, err := d.deliverSession(ctx, ag, rt, msg, mode); err != nil {
		if mode != dispatchModeDeliveryAck {
			d.replyError(ctx, ch, msg, err)
		}
		return err
	}
	return nil
}

func deliveryAgentID(msg *types.Message) (string, error) {
	if msg == nil {
		return "", fmt.Errorf("%w: message is required", errDeliveryRequiresAcceptingProcess)
	}
	if agentID := strings.TrimSpace(msg.AgentID); agentID != "" {
		return agentID, nil
	}
	if meta, ok := msg.Input.Meta.(*types.AgentTeamMeta); ok {
		if agentID := strings.TrimSpace(meta.AgentID); agentID != "" {
			return agentID, nil
		}
	}
	return "", fmt.Errorf("%w: explicit agent id is missing", errDeliveryRequiresAcceptingProcess)
}

// deliverToExistingProcess admits agent:deliver only into the currently
// published Process generation. Profile/runtime resolution, start reservations,
// and Process creation belong to agent:start and are intentionally absent from
// this path.
func (d *Dispatcher) deliverToExistingProcess(ctx context.Context, msg *types.Message) (string, error) {
	lookupID, err := deliveryAgentID(msg)
	if err != nil {
		return "", err
	}
	session, ok := d.sessionManager.FindByLookupID(lookupID)
	if !ok || session == nil {
		return lookupID, fmt.Errorf("%w: agent %q has no current process", errDeliveryRequiresAcceptingProcess, lookupID)
	}
	binding, inbox, err := bindingFromSession(msg)
	if err != nil {
		return lookupID, err
	}

	session.mu.Lock()
	sessionKey := strings.TrimSpace(session.sessionKey)
	agentID := firstNonEmpty(strings.TrimSpace(session.agentID), strings.TrimSpace(session.agent.ID), lookupID)
	process := session.process
	retired := session.retired.Load()
	suppressCheckpoints := session.suppressCheckpoints
	workDir := session.workDir
	currentTarget := session.activeTaskTarget.Normalize()
	currentBinding := session.binding
	session.mu.Unlock()
	if sessionKey == "" {
		sessionKey = sessionKeyForBinding(agentID, conversationBinding{}).String()
	}

	if _, starting := d.sessionManager.StartInFlight(sessionKey); starting {
		return agentID, fmt.Errorf("%w: agent %q has a start in flight", errDeliveryRequiresAcceptingProcess, lookupID)
	}
	current, currentOK := d.sessionManager.Get(sessionKey)
	if !currentOK || current != session || retired || process == nil || !process.AcceptingDeliveries() {
		return agentID, fmt.Errorf("%w: agent %q has no current accepting process", errDeliveryRequiresAcceptingProcess, lookupID)
	}

	binding = existingProcessDeliveryBinding(currentBinding, binding)
	targetToPersist := binding.activeTaskTarget.Normalize()
	switch {
	case targetToPersist.Valid():
		binding.activeTaskTarget = targetToPersist
	case shouldCarrySessionActiveTaskTarget(currentTarget):
		binding.activeTaskTarget = currentTarget
	default:
		binding.activeTaskTarget = types.ActiveTaskTarget{}
	}
	inbox.ActiveTaskTarget = binding.activeTaskTarget
	inbox, _ = normalizeInboxWorkflowContext(inbox)

	urgent, err := handoffToExistingProcess(ctx, process, inbox, lookupID)
	if err != nil {
		return agentID, err
	}

	binding, inbox, ephemeral, committed := d.commitExistingProcessDelivery(
		sessionKey,
		session,
		process,
		binding,
		inbox,
		targetToPersist,
	)
	if !committed {
		return agentID, fmt.Errorf(
			"%w: agent %q process generation changed before handoff commit",
			errDeliveryRequiresAcceptingProcess,
			lookupID,
		)
	}
	if targetToPersist.Valid() {
		persistActiveTaskTarget(workDir, targetToPersist)
	}
	if !ephemeral {
		d.publishSessionStateChanged(agentID)
	}
	if urgent {
		d.recordUrgentDeliveryAccepted(sessionKey, agentID, binding, inbox)
		if content := normalizeCheckpointSummary(inbox.Content); content != "" && !suppressCheckpoints {
			process.RecordCheckpoint("delivery", urgentDeliveryActivityDetail(inbox))
		}
	} else {
		recordDeliveryCheckpoint(process, suppressCheckpoints, inbox)
	}
	return agentID, nil
}

// commitExistingProcessDelivery is the ACK linearization point. Process
// admission may already have fsynced ownership, but a start/retirement that
// wins before this commit must receive a rejected ACK so agent:start can
// recover the same stable message ID without loss or duplicate runtime input.
func (d *Dispatcher) commitExistingProcessDelivery(
	sessionKey string,
	session *interactiveSession,
	process *agent.Process,
	binding conversationBinding,
	inbox types.InboxMessage,
	targetToPersist types.ActiveTaskTarget,
) (conversationBinding, types.InboxMessage, bool, bool) {
	ephemeral := false
	committed := d.sessionManager.CommitDeliveryIfCurrent(sessionKey, session, process, func() {
		binding = existingProcessDeliveryBinding(session.binding, binding)
		switch {
		case targetToPersist.Valid():
			session.activeTaskTarget = targetToPersist
			binding.activeTaskTarget = targetToPersist
		case shouldCarrySessionActiveTaskTarget(session.activeTaskTarget.Normalize()):
			binding.activeTaskTarget = session.activeTaskTarget.Normalize()
		default:
			session.activeTaskTarget = types.ActiveTaskTarget{}
			binding.activeTaskTarget = types.ActiveTaskTarget{}
		}
		inbox.ActiveTaskTarget = binding.activeTaskTarget
		session.binding = binding
		ephemeral = session.ephemeral
		if !ephemeral {
			session.recordHistoryLocked(inbox, inbox.Content, nil, inbox.Timestamp)
		}
	})
	return binding, inbox, ephemeral, committed
}

func handoffToExistingProcess(
	ctx context.Context,
	process *agent.Process,
	inbox types.InboxMessage,
	lookupID string,
) (bool, error) {
	urgent := isInterruptingInboxMessage(inbox)
	if !urgent {
		err := process.DeliverForHandoff(ctx, inbox)
		if errors.Is(err, agent.ErrProcessNotAcceptingDelivery) {
			err = fmt.Errorf("%w: %v", errDeliveryRequiresAcceptingProcess, err)
		}
		return false, err
	}

	handled, err := process.InterruptAndDeliver(ctx, inbox)
	if errors.Is(err, agent.ErrProcessNotAcceptingDelivery) {
		err = fmt.Errorf("%w: %v", errDeliveryRequiresAcceptingProcess, err)
	}
	if err != nil {
		return true, err
	}
	if !handled {
		return true, fmt.Errorf(
			"%w: agent %q runtime cannot accept urgent delivery",
			errDeliveryRequiresAcceptingProcess,
			lookupID,
		)
	}
	return true, nil
}

// existingProcessDeliveryBinding keeps the immutable launch identity owned by
// the current Process while refreshing only the incoming message's routing
// path. agent:deliver carries no authoritative start configuration, so an
// empty (or stale) AgentTeamMeta must not erase the identity/runtime/session inputs a
// later agent:start, deferred continuation, or recovery launch will reuse.
func existingProcessDeliveryBinding(current, incoming conversationBinding) conversationBinding {
	incoming.serverAgentID = firstNonEmpty(strings.TrimSpace(current.serverAgentID), strings.TrimSpace(incoming.serverAgentID))
	incoming.serverAgentName = current.serverAgentName
	incoming.serverAgentDisplayName = current.serverAgentDisplayName
	incoming.serverRuntime = current.serverRuntime
	incoming.serverModel = current.serverModel
	incoming.serverReasoningEffort = current.serverReasoningEffort
	incoming.serverSessionID = current.serverSessionID
	incoming.serverMaterialization = current.serverMaterialization
	if len(incoming.unreadSummary) == 0 && len(current.unreadSummary) > 0 {
		incoming.unreadSummary = make(map[string]int, len(current.unreadSummary))
		for target, count := range current.unreadSummary {
			incoming.unreadSummary[target] = count
		}
	}
	return incoming
}

func recordInboundStats(msg *types.Message, agentID string) {
	if msg == nil {
		return
	}
	cs := metrics.DefaultServerMessageCollector()
	if cs == nil {
		return
	}
	if strings.TrimSpace(agentID) == "" {
		cs.RecordReceived(msg.Input.Channel)
		return
	}
	cs.RecordReceivedForAgent(msg.Input.Channel, agentID)
}

func (d *Dispatcher) resolveAgentAndRuntime(msg *types.Message) (agent.Config, runtimeconfig.Runtime, error) {
	mgr := profile.Default()

	agentID := strings.TrimSpace(msg.AgentID)

	// For agentteam messages, the server determines the target agent.
	if agentID == "" && msg.Input.Meta != nil {
		if lm, ok := msg.Input.Meta.(*types.AgentTeamMeta); ok {
			agentID = lm.AgentID
		}
	}

	// Check agent overrides from /agent command
	if agentID == "" && d.agentOverrides != nil {
		channelID := strings.TrimSpace(msg.Input.Channel)
		var convKey string
		if msg.Input.Meta != nil {
			convKey = conversationKeyForMetadata(msg.Input.Meta)
		}
		if convKey != "" {
			agentID = d.agentOverrides.Get(channelID, convKey)
		}
	}

	if agentID == "" {
		if channelCfg, ok := mgr.Servers().Get(msg.Input.Channel); ok {
			agentID = strings.TrimSpace(channelCfg.AgentID)
		}
	}

	// Fall back to the configured default agent.
	if agentID == "" {
		agentID = strings.TrimSpace(mgr.Agents().Default())
	}

	if agentID == "" {
		return agent.Config{}, runtimeconfig.Runtime{}, fmt.Errorf("no default agent configured for channel %q; use /agent <name> to select one", msg.Input.Channel)
	}

	ag, ok := mgr.Agents().Get(agentID)
	if !ok {
		return agent.Config{}, runtimeconfig.Runtime{}, fmt.Errorf("agent %q not found", agentID)
	}
	rt, ok := resolveConfiguredRuntime(mgr, ag.Runtime)
	if !ok {
		return agent.Config{}, runtimeconfig.Runtime{}, fmt.Errorf("runtime %q referenced by agent %q not found", ag.Runtime, ag.ID)
	}
	return ag, rt, nil
}

func resolveConfiguredRuntime(mgr *profile.Manager, runtimeName string) (runtimeconfig.Runtime, bool) {
	if mgr == nil {
		return runtimeconfig.Runtime{}, false
	}
	trimmed := strings.TrimSpace(runtimeName)
	if rt, ok := mgr.Runtimes().GetOrRefresh(trimmed); ok {
		return rt, true
	}
	normalized, err := runtimeconfig.ValidateType(trimmed)
	if err != nil {
		return runtimeconfig.Runtime{}, false
	}
	if normalized != runtimeconfig.RuntimeType(trimmed) {
		if rt, ok := mgr.Runtimes().GetOrRefresh(string(normalized)); ok {
			return rt, true
		}
	}
	if normalized == runtimeconfig.TypeBuiltin {
		rt, err := runtimeconfig.NormalizeRuntime(runtimeconfig.Runtime{Type: runtimeconfig.TypeBuiltin})
		return rt, err == nil
	}
	if rt, ok := mgr.Runtimes().GetOrRefresh(string(normalized)); ok {
		return rt, true
	}
	if normalized != runtimeconfig.TypeOpenCode {
		return runtimeconfig.Runtime{}, false
	}
	rt, err := runtimeconfig.NormalizeRuntime(runtimeconfig.Runtime{Type: runtimeconfig.TypeOpenCode})
	return rt, err == nil
}

func (d *Dispatcher) replyError(ctx context.Context, ch types.ServerConnection, msg *types.Message, err error) {
	if err == nil {
		return
	}
	errText := "agentteam agent dispatch failed: " + err.Error()
	_ = ch.Send(ctx, msg.Input.Meta, types.OutboundMessage{Content: errText})
}

// ---------------------------------------------------------------------------
// Session queries
// ---------------------------------------------------------------------------

// Session returns a read-only snapshot of one interactive session.
func (d *Dispatcher) Session(agentID string) (types.SessionSnapshot, bool) {
	session, ok := d.sessionManager.FindByLookupID(agentID)
	if !ok {
		return types.SessionSnapshot{}, false
	}

	session.mu.Lock()
	snapshot := d.buildSnapshot(session)
	session.mu.Unlock()
	return snapshot, true
}

// Sessions returns snapshots of all active interactive sessions.
func (d *Dispatcher) Sessions() []types.SessionSnapshot {
	sessions := d.sessionManager.Snapshot(false)

	result := make([]types.SessionSnapshot, 0, len(sessions))
	for _, sess := range sessions {
		sess.mu.Lock()
		snapshot := d.buildSnapshot(sess)
		sess.mu.Unlock()
		result = append(result, snapshot)
	}
	return result
}

func (d *Dispatcher) buildSnapshot(sess *interactiveSession) types.SessionSnapshot {
	status := "stopped"
	runtimeSessionID := strings.TrimSpace(sess.runtimeSessionID)
	turnState := ""
	if sess.process != nil {
		status = sess.process.State()
		turnState = sess.process.DerivedTurnState()
	}
	title := ""
	for i := len(sess.history) - 1; i >= 0; i-- {
		h := sess.history[i]
		if h.AuthorIdentity().ActorType == "human" && strings.TrimSpace(h.Content) != "" {
			title = strings.TrimSpace(h.Content)
			break
		}
	}
	return types.SessionSnapshot{
		SessionKey:       sess.sessionKey,
		AgentID:          sess.agentID,
		AgentConfigID:    sess.agent.ID,
		AgentName:        sess.agent.Name,
		Title:            title,
		ChannelID:        sess.binding.channelID,
		ConversationKey:  sess.binding.conversationKey,
		WorkDir:          sess.workDir,
		Status:           status,
		RuntimeSessionID: runtimeSessionID,
		TurnState:        turnState,
		Activity:         sess.activity,
		ActivityDetail:   sess.activityDetail,
		MessageCount:     len(sess.history),
		StartedAt:        sess.startedAt,
		ServerAgentID:    sess.serverAgentID,
		ActiveTaskTarget: sess.activeTaskTarget.Normalize(),
		Metadata:         sess.binding.meta,
	}
}

// ---------------------------------------------------------------------------
// Session lifecycle
// ---------------------------------------------------------------------------

// StopByAgent stops the non-ephemeral interactive session for the given agent.
// Normal dispatcher sessions are agent-scoped, so command handlers should use
// this instead of the latest conversation binding.
func (d *Dispatcher) StopByAgent(agentID string) bool {
	lookupID := strings.TrimSpace(agentID)
	if lookupID == "" {
		return false
	}
	targets := d.sessionManager.Each(func(sess *interactiveSession) bool {
		return !sess.ephemeral && sessionMatchesLookupID(sess, lookupID)
	})
	stopped := false
	for _, session := range targets {
		session.mu.Lock()
		process := session.process
		session.mu.Unlock()
		if process == nil {
			continue
		}
		d.teardownSessionProcess(session, process)
		stopped = true
	}
	return stopped
}

// ClearSessionResumeStateByAgent clears daemon-local resume metadata for an
// agent without deleting the workspace files.
func (d *Dispatcher) ClearSessionResumeStateByAgent(agentID string) error {
	lookupID := strings.TrimSpace(agentID)
	if lookupID == "" {
		return nil
	}
	workDirs := map[string]struct{}{
		sessionWorkDir(lookupID): {},
	}
	targets := d.sessionManager.Each(func(sess *interactiveSession) bool {
		return !sess.ephemeral && sessionMatchesLookupID(sess, lookupID)
	})
	for _, session := range targets {
		session.mu.Lock()
		workDir := strings.TrimSpace(session.workDir)
		session.mu.Unlock()
		if workDir != "" {
			workDirs[workDir] = struct{}{}
		}
	}
	for workDir := range workDirs {
		if err := defaultSessionResumeStateStore.Delete(workDir); err != nil {
			return err
		}
	}
	return nil
}

// RestartByAgent retires the current non-ephemeral runtime for the given agent
// without clearing its workspace. The next delivery or daemon continuation
// starts a fresh runtime with the latest prompt and MCP tool set.
func (d *Dispatcher) RestartByAgent(agentID string) bool {
	lookupID := strings.TrimSpace(agentID)
	if lookupID == "" {
		return false
	}
	targets := d.sessionManager.Each(func(sess *interactiveSession) bool {
		return !sess.ephemeral && sessionMatchesLookupID(sess, lookupID)
	})

	restarted := false
	for _, session := range targets {
		restarted = d.interruptSessionForRestart(session) || restarted
	}
	return restarted
}

// stopAll stops all running interactive sessions.
func (d *Dispatcher) stopAll() {
	connectionShutdown := d.serverConnections.BeginShutdown()
	connectionShutdown.StartDrain(context.Background())
	trajectoryDone := d.trajectoryHub.BeginStop()
	memoryOwners := d.closeMemoryWatcherMutationAdmission()
	d.closeExecAuthorizationAdmission()
	d.closeProcessExitCallbackAdmission()
	startOwners, sessions := d.sessionManager.CloseAndSnapshot(true)
	d.cancelRuntimeOwnershipRetryWorkers()
	_ = waitForMemoryWatcherQuiescence(context.Background(), memoryOwners, startOwners)
	processes := make([]*agent.Process, 0, len(sessions))
	for _, sess := range sessions {
		sess.mu.Lock()
		process := sess.process
		sess.mu.Unlock()
		if process != nil {
			processes = append(processes, process)
			d.teardownSessionProcess(sess, process)
		} else {
			d.stopMemoryWatcherAndRetry(sess)
		}
	}
	for _, process := range processes {
		if !process.WaitForExit(5 * time.Second) {
			slog.Warn("dispatcher stop: process did not exit before timeout", "agent_id", process.AgentID)
		}
	}
	d.disableMemorySearch()
	d.waitForShutdownOwners(context.Background(), startOwners, "dispatcher stop")
	d.revokeAllExecAuthorizations(context.Background())
	for _, process := range processes {
		process.WaitForExitCallback(context.Background())
	}
	_ = connectionShutdown.Wait(context.Background())
	<-trajectoryDone
}

// StopByServerAgentID stops all sessions matching the given agentteam server
// agent ID. Returns true if at least one session was found and stopped.
func (d *Dispatcher) StopByServerAgentID(serverAgentID string) bool {
	targets := d.sessionManager.Each(func(sess *interactiveSession) bool {
		return sess.serverAgentID == serverAgentID
	})

	stopped := false
	for _, sess := range targets {
		sess.mu.Lock()
		process := sess.process
		sess.mu.Unlock()
		if process != nil {
			d.teardownSessionProcess(sess, process)
			stopped = true
		}
	}
	return stopped
}

// InterruptByServerAgentID retires and stops communication sessions matching
// the given AgentTeam server agent ID so an interrupting delivery can start from the
// new authoritative input instead of being queued behind the old run. Lifecycle
// stop/reset instead reports runtime interruption/failure so controlled
// resume can remain available when the runtime session handle survives.
func (d *Dispatcher) InterruptByServerAgentID(serverAgentID string) bool {
	targets := d.sessionManager.Each(func(sess *interactiveSession) bool {
		return sess.serverAgentID == serverAgentID
	})

	interrupted := false
	for _, sess := range targets {
		interrupted = d.interruptSessionForRestart(sess) || interrupted
	}
	return interrupted
}

// InterruptActiveRunByServerAgentID interrupts the active runtime turn for
// sessions matching the given AgentTeam server agent ID without retiring the session
// or clearing daemon-local resume metadata.
func (d *Dispatcher) InterruptActiveRunByServerAgentID(serverAgentID string) bool {
	targets := d.sessionManager.Each(func(sess *interactiveSession) bool {
		return sess.serverAgentID == serverAgentID
	})

	interrupted := false
	for _, sess := range targets {
		sess.mu.Lock()
		leaderAgentID := sess.agentID
		process := sess.process
		sess.mu.Unlock()
		if process == nil {
			continue
		}
		handled, err := process.InterruptActiveRun(context.Background())
		if err != nil {
			slog.Warn("active run interrupt failed",
				"agent_id", leaderAgentID,
				"server_agent_id", serverAgentID,
				"error", err)
			continue
		}
		if !handled {
			continue
		}
		d.publishSessionStateChanged(leaderAgentID)
		interrupted = true
	}
	return interrupted
}

// SleepByServerAgentID signals all sessions matching the given agentteam server
// agent ID to sleep. Returns true if at least one session was found.
func (d *Dispatcher) SleepByServerAgentID(serverAgentID string) bool {
	targets := d.sessionManager.Each(func(sess *interactiveSession) bool {
		return sess.serverAgentID == serverAgentID
	})

	found := false
	for _, sess := range targets {
		sess.mu.Lock()
		process := sess.process
		sess.mu.Unlock()
		if process != nil {
			d.revokeExecProcessAuthorization(process)
			process.Sleep()
		}
		found = true
	}
	return found
}

// ResetByAgent stops running sessions for the given agent or exact session key
// and clears their workspace directories. Returns (true, nil) if any session
// was found and reset, (false, nil) if none exists.
func (d *Dispatcher) ResetByAgent(agentID string) (bool, error) {
	lookupID := strings.TrimSpace(agentID)
	if lookupID == "" {
		return false, nil
	}

	var targets []*interactiveSession
	if sess, ok := d.sessionManager.Get(lookupID); ok {
		sess.mu.Lock()
		ephemeral := sess.ephemeral
		sess.mu.Unlock()
		if !ephemeral {
			targets = append(targets, sess)
		}
	} else {
		targets = d.sessionManager.Each(func(sess *interactiveSession) bool {
			return !sess.ephemeral && sessionMatchesLookupID(sess, lookupID)
		})
	}

	if len(targets) == 0 {
		return false, nil
	}

	for _, session := range targets {
		if err := d.resetSession(session); err != nil {
			return false, err
		}
	}
	return true, nil
}

func (d *Dispatcher) resetSession(session *interactiveSession) error {
	if session == nil {
		return nil
	}

	resetCtx, cancelReset := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancelReset()

	// Reset participates in the exact same authoritative start generation as normal
	// delivery, agent:start, recovery, and daemon continuation. If another owner
	// is already replacing the Process, wait for that generation and reselect the
	// new current session before acquiring the reset fence; deleting the original
	// pointer's workdir would otherwise race the winner's first durable writes.
	target, reservation, sessionKey, err := d.reserveSessionReset(resetCtx, session)
	if err != nil {
		return err
	}
	defer d.sessionManager.FinishStart(reservation)

	target.mu.Lock()
	previousResetting := target.resetting
	previousSuppressCheckpoints := target.suppressCheckpoints
	target.resetting = true
	target.suppressCheckpoints = true
	agentID := target.agentID
	process := target.process
	workDir := target.workDir
	target.mu.Unlock()
	d.blockIdleMemoryRetry(agentID)
	resetRemovalCommitted := false
	defer func() {
		if resetRemovalCommitted {
			return
		}
		d.allowIdleMemoryRetry(agentID)
		// A timeout/error before RemoveIfCurrent must not leave a manager-visible
		// session permanently fenced from delivery or a later reset attempt.
		target.mu.Lock()
		target.resetting = previousResetting
		target.suppressCheckpoints = previousSuppressCheckpoints
		target.mu.Unlock()
	}()

	if process != nil {
		stopMemoryWatcher(target)
		d.stopSessionProcessAfterMemoryDrain(target, process)
		if !process.WaitForExit(5 * time.Second) {
			return fmt.Errorf("runtime did not stop before workspace reset")
		}
	}
	// Sleeping/watcher-only sessions have no Process for teardownSessionProcess
	// to key off, but their watcher still owns filesystem callbacks and a final
	// push. Drain it before removal and workspace purge; the call is idempotent
	// when the process-bearing branch already stopped it.
	stopMemoryWatcher(target)
	if err := d.blockAndWaitRuntimeOwnershipRetries(resetCtx, target); err != nil {
		return fmt.Errorf("runtime ownership retry did not stop before workspace reset: %w", err)
	}

	// Retire the selected current pointer while the reset generation remains
	// held. New deliveries can now observe no session, but StartInFlight keeps
	// them behind this reservation until workspace removal is complete.
	resetRemovalCommitted = d.sessionManager.RemoveIfCurrent(sessionKey, target)

	lookupMu := d.agentMemoryLookupMutex(agentID)
	lookupMu.Lock()
	defer lookupMu.Unlock()
	// RefreshMemoryAccess also owns lookupMu. Clearing state here means any
	// refresh that was already running publishes before this reset boundary;
	// a cancelled worker that was waiting on the lock observes its cancelled
	// context and exits without recreating the workspace.
	d.clearMemoryLookupStatus(agentID)
	d.clearMemorySyncStatus(agentID)
	d.blockIdleMemoryRetry(agentID)
	if reg, _ := d.snapshotMemorySearch(context.Background()); reg != nil && reg.manager != nil {
		reg.denyAgent(agentID)
		if err := reg.manager.PurgeAgent(context.Background(), agentID); err != nil {
			return fmt.Errorf("purge memory search before workspace reset: %w", err)
		}
		reg.dropAgentEntries(agentID)
	}

	if workDir != "" {
		if err := os.RemoveAll(workDir); err != nil {
			return err
		}
	}

	d.publishSessionStateChanged(agentID)
	return nil
}

func (d *Dispatcher) reserveSessionReset(
	ctx context.Context,
	initial *interactiveSession,
) (*interactiveSession, *sessionStartReservation, string, error) {
	if d == nil || d.sessionManager == nil || initial == nil {
		return nil, nil, "", fmt.Errorf("session reset target is required")
	}
	if ctx == nil {
		ctx = context.Background()
	}
	initial.mu.Lock()
	agentID := strings.TrimSpace(initial.agentID)
	if agentID == "" {
		agentID = strings.TrimSpace(initial.agent.ID)
	}
	initial.mu.Unlock()
	if agentID == "" {
		return nil, nil, "", fmt.Errorf("session reset target has no local agent id")
	}
	defaultSessionKey := sessionKeyForBinding(agentID, conversationBinding{}).String()
	fallback := initial

	for {
		current, exists := d.sessionManager.Get(defaultSessionKey)
		if exists && current != nil {
			fallback = current
		}
		var expected *interactiveSession
		if exists {
			expected = current
		}
		reservation, waitCh, owns := d.sessionManager.BeginStart(defaultSessionKey, expected)
		if owns {
			return fallback, reservation, defaultSessionKey, nil
		}
		if d.sessionManager.Closed() {
			return nil, nil, defaultSessionKey, errSessionManagerClosed
		}
		if err := waitForSessionStart(ctx, waitCh); err != nil {
			return nil, nil, defaultSessionKey, fmt.Errorf("wait for in-flight session start before reset: %w", err)
		}
		// A retired pointer can briefly remain visible between retirement and
		// RemoveIfCurrent. Its failed BeginStart returns an already-closed wait
		// channel, so yield through ctx rather than spinning while the remover
		// finishes its manager transaction.
		if current != nil && current.retired.Load() {
			timer := time.NewTimer(time.Millisecond)
			select {
			case <-timer.C:
			case <-ctx.Done():
				timer.Stop()
				return nil, nil, defaultSessionKey, ctx.Err()
			}
		}
	}
}

func applyAgentTeamStartConfig(base agent.Config, cfg types.AgentStartConfig) agent.Config {
	startConfig := base
	if trimmedRuntime := strings.TrimSpace(cfg.Runtime); trimmedRuntime != "" {
		startConfig.Runtime = trimmedRuntime
	}
	if trimmedPermissionMode := strings.TrimSpace(cfg.PermissionMode); trimmedPermissionMode != "" {
		startConfig.PermissionMode = types.NormalizeAgentPermissionMode(trimmedPermissionMode)
	} else {
		startConfig.PermissionMode = types.NormalizeAgentPermissionMode(startConfig.PermissionMode)
	}
	if cfg.RuntimePolicy != nil {
		startConfig.RuntimePolicy = agentRuntimePolicyFromWire(cfg.RuntimePolicy)
	}
	startConfig.RuntimeDebugEnabled = cfg.RuntimeDebug != nil && cfg.RuntimeDebug.Enabled
	startConfig.Model = strings.TrimSpace(cfg.Model)
	startConfig.ReasoningEffort = strings.TrimSpace(cfg.ReasoningEffort)
	if trimmedName := strings.TrimSpace(cfg.Name); trimmedName != "" {
		startConfig.Name = trimmedName
	}
	if trimmedDescription := strings.TrimSpace(cfg.Description); trimmedDescription != "" {
		startConfig.Description = trimmedDescription
	}
	startConfig.LLMMode, startConfig.BaseURL, startConfig.APIKey, startConfig.Headers, startConfig.WireAPI =
		agent.NormalizeAgentLLMFields(cfg.LLMMode, cfg.BaseURL, cfg.APIKey, cfg.Headers, cfg.WireAPI)
	startConfig.CredentialKind = strings.TrimSpace(cfg.CredentialKind)
	return startConfig
}

func runtimeDebugConfigFromEnabled(enabled bool) *types.RuntimeDebugConfig {
	if !enabled {
		return nil
	}
	return &types.RuntimeDebugConfig{Enabled: true}
}

// StartAgentForAgentTeam resolves the agent config and runtime, then starts an
// interactive session for a agentteam agent:start event.
func (d *Dispatcher) StartAgentForAgentTeam(channelID, serverAgentID string, cfg types.AgentStartConfig, wake *types.Message) error {
	mgr := profile.Default()
	ag, ok := mgr.Agents().Get(serverAgentID)
	if !ok {
		err := fmt.Errorf("agent %q not found in local config", serverAgentID)
		slog.Warn("agentteam start: agent not found in local config", "server_agent_id", serverAgentID)
		return err
	}

	startConfig := applyAgentTeamStartConfig(ag, cfg)

	rt, ok := resolveConfiguredRuntime(mgr, startConfig.Runtime)
	if !ok {
		err := fmt.Errorf("runtime %q not found for agent %q", startConfig.Runtime, serverAgentID)
		slog.Warn("agentteam start: runtime not found for agent",
			"server_agent_id", serverAgentID, "runtime", startConfig.Runtime)
		return err
	}

	return d.startAgentForAgentTeamSession(context.Background(), startConfig, rt, channelID, serverAgentID, cfg, wake)
}

type startSessionOptions struct {
	ephemeral                  bool
	durableDelivery            bool
	suppressCheckpoints        bool
	suppressIdleStartTurn      bool
	requireProcessHandoff      bool
	startReservation           *sessionStartReservation
	authoritativeStartIdentity bool
}

type existingSessionDeliveryState struct {
	binding         conversationBinding
	msg             types.InboxMessage
	targetToPersist types.ActiveTaskTarget
	process         *agent.Process
	workDir         string
}

// DeliverSession derives the conversation binding from sess, resolves or starts the matching agent session, and delivers the inbound message.
func (d *Dispatcher) DeliverSession(ctx context.Context, ag agent.Config, rt runtimeconfig.Runtime, sess *types.Message) (string, error) {
	if d == nil {
		return "", errors.New("dispatcher is nil")
	}
	return d.deliverSession(ctx, ag, rt, sess, dispatchModeDefault)
}

func (d *Dispatcher) deliverSession(ctx context.Context, ag agent.Config, rt runtimeconfig.Runtime, sess *types.Message, mode dispatchMode) (string, error) {
	if sess == nil {
		return "", fmt.Errorf("session is required")
	}

	binding, msg, err := bindingFromSession(sess)
	if err != nil {
		return "", err
	}
	return d.deliverResolvedSession(ctx, sessionKeyForBinding(ag.ID, binding).String(), ag, rt, binding, msg, startSessionOptions{
		// Delivery ACK mode controls when the server forgets its outbox owner;
		// it must not control local admission reliability. Every concrete routed
		// message uses the same error-returning, durable Process handoff so the
		// default dispatch path cannot report success after Stop or persistence
		// rejection merely because Process.Deliver has no return value.
		requireProcessHandoff: strings.TrimSpace(msg.MessageID) != "",
	})
}

func (d *Dispatcher) deliverResolvedSession(
	ctx context.Context,
	sessionKey string,
	ag agent.Config,
	rt runtimeconfig.Runtime,
	binding conversationBinding,
	msg types.InboxMessage,
	options startSessionOptions,
) (string, error) {
	sessionKey, err := requireSessionKey(sessionKey)
	if err != nil {
		return "", err
	}
	agentID := strings.TrimSpace(ag.ID)
	if agentID == "" {
		return "", fmt.Errorf("agent id is required")
	}
	for {
		if d.sessionManager.Closed() {
			return "", errSessionManagerClosed
		}
		// Process publication is only the middle of a start generation: a
		// reserved queued delivery may still be moving through its durable
		// post-launch handoff. Keep later deliveries behind the whole generation.
		if waitCh, starting := d.sessionManager.StartInFlight(sessionKey); starting {
			if err := waitForSessionStart(ctx, waitCh); err != nil {
				if isWakeMessage(msg) {
					d.emitWakeFailed(binding.channelID, binding.serverAgentID, msg, err)
				}
				return "", err
			}
			continue
		}

		session, ok := d.sessionManager.Get(sessionKey)
		var expected *interactiveSession
		if ok {
			restarted, err := d.restartSessionIfRuntimeStartConfigChanged(session, ag, rt, binding)
			if err != nil {
				return "", err
			}
			if restarted {
				continue
			}
			delivery := prepareExistingSessionDelivery(session, binding, msg)
			binding = delivery.binding
			msg = delivery.msg
			if !session.ephemeral {
				session.recordHistory(msg, msg.Content, nil, msg.Timestamp)
				d.publishSessionStateChanged(agentID)
			}
			if delivery.targetToPersist.Valid() {
				persistActiveTaskTarget(delivery.workDir, delivery.targetToPersist)
			}
			if delivery.process == nil || !delivery.process.AcceptingDeliveries() {
				expected = session
			} else {
				// A continuation can reserve the key after the first check but before
				// publishing its Process into the existing session. Recheck after the
				// Process snapshot so q2 cannot pass its reserved q1 handoff.
				if waitCh, starting := d.sessionManager.StartInFlight(sessionKey); starting {
					if err := waitForSessionStart(ctx, waitCh); err != nil {
						if isWakeMessage(msg) {
							d.emitWakeFailed(binding.channelID, binding.serverAgentID, msg, err)
						}
						return "", err
					}
					continue
				}
				if isInterruptingInboxMessage(msg) {
					handled, err := delivery.process.InterruptAndDeliver(ctx, msg)
					if handled && err == nil {
						d.recordUrgentDeliveryAccepted(sessionKey, agentID, binding, msg)
						if content := normalizeCheckpointSummary(msg.Content); content != "" && !session.suppressCheckpoints {
							delivery.process.RecordCheckpoint("delivery", urgentDeliveryActivityDetail(msg))
						}
						return sessionKey, nil
					}
					if err != nil {
						slog.Warn("explicit interrupt direct delivery failed; falling back to process restart",
							"agent_id", agentID,
							"server_agent_id", binding.serverAgentID,
							"message_id", msg.MessageID,
							"error", err)
					}
					d.interruptSessionForRestart(session)
					continue
				}
				if options.requireProcessHandoff {
					if err := delivery.process.DeliverForHandoff(ctx, msg); err != nil {
						if errors.Is(err, agent.ErrProcessNotAcceptingDelivery) {
							expected = session
						} else {
							return "", err
						}
					} else {
						recordDeliveryCheckpoint(delivery.process, session.suppressCheckpoints, msg)
						return sessionKey, nil
					}
				} else {
					delivery.process.Deliver(msg)
					recordDeliveryCheckpoint(delivery.process, session.suppressCheckpoints, msg)
					return sessionKey, nil
				}
			}
		}

		reservation, waitCh, shouldStart := d.sessionManager.BeginStart(sessionKey, expected)
		if !shouldStart {
			if err := waitForSessionStart(ctx, waitCh); err != nil {
				if isWakeMessage(msg) {
					d.emitWakeFailed(binding.channelID, binding.serverAgentID, msg, err)
				}
				return "", err
			}
			continue
		}

		startOptions := options
		startOptions.startReservation = reservation
		session, err := d.startSession(ctx, sessionKey, ag, rt, binding, msg, startOptions)
		d.sessionManager.FinishStart(reservation)

		if err != nil {
			if isWakeMessage(msg) {
				d.emitWakeFailed(binding.channelID, binding.serverAgentID, msg, err)
			}
			return "", err
		}

		if !session.ephemeral {
			session.recordHistory(msg, msg.Content, nil, msg.Timestamp)
			d.publishSessionStateChanged(agentID)
		}
		if isInterruptingInboxMessage(msg) {
			d.recordUrgentDeliveryAccepted(sessionKey, agentID, binding, msg)
			if !session.suppressCheckpoints {
				session.mu.Lock()
				process := session.process
				session.mu.Unlock()
				if process != nil {
					process.RecordCheckpoint("delivery", urgentDeliveryActivityDetail(msg))
				}
			}
		}
		return sessionKey, nil
	}
}

func prepareExistingSessionDelivery(session *interactiveSession, binding conversationBinding, msg types.InboxMessage) existingSessionDeliveryState {
	session.mu.Lock()
	defer session.mu.Unlock()

	targetToPersist := binding.activeTaskTarget.Normalize()
	if targetToPersist.Valid() {
		session.activeTaskTarget = targetToPersist
	} else if currentTarget := session.activeTaskTarget.Normalize(); shouldCarrySessionActiveTaskTarget(currentTarget) {
		binding.activeTaskTarget = currentTarget
	} else {
		session.activeTaskTarget = types.ActiveTaskTarget{}
	}
	msg.ActiveTaskTarget = binding.activeTaskTarget
	msg, _ = normalizeInboxWorkflowContext(msg)
	session.binding = binding
	return existingSessionDeliveryState{
		binding:         binding,
		msg:             msg,
		targetToPersist: targetToPersist,
		process:         session.process,
		workDir:         session.workDir,
	}
}

func recordDeliveryCheckpoint(process *agent.Process, suppressCheckpoints bool, msg types.InboxMessage) {
	content := normalizeCheckpointSummary(msg.Content)
	if process == nil || suppressCheckpoints || content == "" {
		return
	}
	author := strings.TrimSpace(msg.AuthorIdentity().DisplayName)
	if author == "" {
		author = "unknown author"
	}
	location := strings.TrimSpace(msg.Target)
	if location == "" {
		location = msg.ChannelName
	}
	process.RecordCheckpoint("delivery", fmt.Sprintf("New message from %s in %s: %s", author, location, content))
}

func isInterruptingInboxMessage(msg types.InboxMessage) bool {
	return strings.TrimSpace(msg.InterruptPolicy) == "interrupt_current_run"
}

func urgentDeliveryActivityDetail(msg types.InboxMessage) string {
	return deliveryCheckpointSummary(msg, true)
}

func deliveryCheckpointSummary(msg types.InboxMessage, urgent bool) string {
	author := strings.TrimSpace(msg.AuthorIdentity().DisplayName)
	if author == "" {
		author = "unknown author"
	}
	location := strings.TrimSpace(msg.Target)
	if location == "" {
		location = strings.TrimSpace(msg.ChannelName)
	}
	kind := "New message"
	if urgent {
		kind = "Urgent message"
	}
	messageID := strings.TrimSpace(msg.MessageID)
	if messageID == "" {
		return fmt.Sprintf("%s from %s in %s; content omitted from local checkpoint and should be recovered from AgentTeam history if still visible.", kind, author, location)
	}
	return fmt.Sprintf("%s from %s in %s (message %s); content omitted from local checkpoint and should be recovered from AgentTeam history if still visible.", kind, author, location, messageID)
}

func (d *Dispatcher) recordUrgentDeliveryAccepted(sessionKey string, agentID string, binding conversationBinding, msg types.InboxMessage) {
	d.recordSessionActivity(sessionKey, agentID, binding, msg, "working", urgentDeliveryActivityDetail(msg))
}

func (d *Dispatcher) recordSessionActivity(sessionKey string, agentID string, binding conversationBinding, msg types.InboxMessage, activity string, detail string) {
	activity = strings.TrimSpace(activity)
	if activity == "" {
		return
	}
	detail = strings.TrimSpace(detail)
	changed := false
	d.sessionManager.WithSessionByKey(sessionKey, func(sess *interactiveSession) {
		sess.mu.Lock()
		changed = sess.activity != activity || sess.activityDetail != detail
		sess.activity = activity
		sess.activityDetail = detail
		sess.mu.Unlock()
	})
	if changed {
		d.publishSessionStateChanged(agentID)
	}
	d.publishRuntimeEvent(runtimeevents.NewDataEvent(
		runtimeevents.NewLifecycleData(runtimeevents.KindDeliveryAccepted, activity, detail),
		runtimeevents.WithContext(runtimeevents.EventContext{
			ServerID:      strings.TrimSpace(binding.channelID),
			AgentID:       strings.TrimSpace(agentID),
			ServerAgentID: strings.TrimSpace(binding.serverAgentID),
		}),
		runtimeevents.WithTrace(runtimeevents.EventTrace{
			MessageID: strings.TrimSpace(msg.MessageID),
			Target:    strings.TrimSpace(msg.Target),
		}),
	))
}

func (d *Dispatcher) interruptSessionForRestart(sess *interactiveSession) bool {
	return d.retireSessionForRestart(sess, true, nil, nil)
}

func (d *Dispatcher) interruptSessionProcessForRestart(sess *interactiveSession, process *agent.Process) bool {
	return d.retireSessionForRestart(sess, true, nil, &restartProcessGeneration{process: process})
}

type restartProcessGeneration struct{ process *agent.Process }

type runtimeStartReplacement struct {
	agent       agent.Config
	runtime     runtimeconfig.Runtime
	fingerprint string
}

func (d *Dispatcher) interruptSessionForRuntimeStartReplacement(
	sess *interactiveSession,
	ag agent.Config,
	rt runtimeconfig.Runtime,
	fingerprint string,
) bool {
	replacement := &runtimeStartReplacement{
		agent:       ag,
		runtime:     rt,
		fingerprint: strings.TrimSpace(fingerprint),
	}
	for sess != nil {
		if d.sessionManager.Closed() {
			return false
		}
		sess.mu.Lock()
		sessionKey := firstNonEmpty(strings.TrimSpace(sess.sessionKey), strings.TrimSpace(sess.agentID))
		currentFingerprint := strings.TrimSpace(sess.runtimeStartFingerprint)
		process := sess.process
		sess.mu.Unlock()
		if replacement.fingerprint != "" && currentFingerprint == replacement.fingerprint &&
			process != nil && process.AcceptingDeliveries() {
			return false
		}

		if waitCh, starting := d.sessionManager.StartInFlight(sessionKey); starting {
			if waitForSessionStart(context.Background(), waitCh) != nil {
				return false
			}
			current, ok := d.sessionManager.Get(sessionKey)
			if !ok {
				return false
			}
			sess = current
			continue
		}
		if d.retireSessionForRestart(sess, true, replacement, nil) {
			return true
		}

		// A start may have reserved the key after the snapshot above but before
		// retirement's BeginStart. Follow that generation to completion and then
		// re-evaluate its actually published fingerprint before returning. This
		// prevents a newer agent:start config from being acknowledged while an
		// older retry publishes a stale Process underneath it.
		if waitCh, starting := d.sessionManager.StartInFlight(sessionKey); starting {
			if waitForSessionStart(context.Background(), waitCh) != nil {
				return false
			}
		}
		current, ok := d.sessionManager.Get(sessionKey)
		if !ok {
			return false
		}
		if current == sess {
			// The competing generation may have published into the same retained
			// session and finished between BeginStart and StartInFlight. Recheck
			// its fingerprint at the top of the agentteam instead of mistaking pointer
			// identity for proof that no newer Process was published.
			continue
		}
		sess = current
	}
	return false
}

// detachSessionForPreparedRestart retires a published session whose Process
// lifecycle owner has already made it non-accepting and will terminate it after
// this callback returns. It must not call Process.Stop: live recovery invokes
// this while holding deliveryGate, and a public Stop would wait for the very
// lifecycle completion that the callback is responsible for allowing.
func (d *Dispatcher) detachSessionForPreparedRestart(sess *interactiveSession) bool {
	return d.retireSessionForRestart(sess, false, nil, nil)
}

func (d *Dispatcher) retireSessionForRestart(
	sess *interactiveSession,
	stopProcess bool,
	replacement *runtimeStartReplacement,
	expectedProcess *restartProcessGeneration,
) bool {
	if sess == nil {
		return false
	}
	if !stopProcess {
		return d.detachPreparedSessionForRestart(sess)
	}

	sess.mu.Lock()
	sessionKey := sess.sessionKey
	agentID := sess.agentID
	sess.mu.Unlock()
	if strings.TrimSpace(sessionKey) == "" {
		sessionKey = agentID
	}

	// Restart-like retirement and agent:deliver ACK commit share the same
	// generation fence. If retirement reserves first, the handoff commit is
	// rejected and server agent:start remains authoritative. If the handoff
	// commits first, this reservation owns replacement of the Process that now
	// carries the durable delivery owner.
	var reservation *sessionStartReservation
	var ownsRestart bool
	if expectedProcess != nil {
		reservation, _, ownsRestart = d.sessionManager.BeginStartForProcess(sessionKey, sess, expectedProcess.process)
	} else {
		reservation, _, ownsRestart = d.sessionManager.BeginStart(sessionKey, sess)
	}
	if !ownsRestart {
		return false
	}
	defer d.sessionManager.FinishStart(reservation)

	sess.mu.Lock()
	if sess.retired.Load() {
		sess.mu.Unlock()
		return false
	}
	if replacement != nil {
		sess.agent = replacement.agent
		sess.runtime = replacement.runtime
		sess.runtimeStartReplacement = true
	}
	process := sess.process
	binding := sess.binding
	ephemeral := sess.ephemeral
	serverAgentID := sess.serverAgentID
	terminalReportOwner := false
	if !sess.terminalReportClaimed {
		sess.terminalReportClaimed = true
		sess.terminalReportDone = make(chan struct{})
		sess.terminalReportWillStop = !ephemeral
		terminalReportOwner = true
	}
	terminalReportDone := sess.terminalReportDone
	terminalReportWillStop := sess.terminalReportWillStop
	sess.restartRetirementInFlight = true
	sess.mu.Unlock()

	revokeSessionMCPToken(sess)
	if process != nil {
		d.revokeExecProcessAuthorization(process)
		process.Stop()
	}
	// Process.Stop is the cutover barrier for legacy workspace-root Memory
	// writes. Re-probe and import the historical source only after that exact
	// generation can no longer mutate it, and before any ownership continuation
	// publishes an authoritative replacement Process.
	d.finalizePendingLegacyPreferencesMigration(context.Background(), sess, true)
	if !terminalReportOwner && terminalReportDone != nil {
		<-terminalReportDone
	}
	if !ephemeral && (terminalReportOwner || !terminalReportWillStop) {
		localAgentID := strings.TrimSpace(agentID)
		if process != nil {
			localAgentID = firstNonEmpty(strings.TrimSpace(process.AgentID), localAgentID)
		}
		// Publish the terminal stopped/idle turn-state WHILE sess.process is
		// still the retired generation, so the AgentTeamReporterSink resolves idle
		// instead of an empty snapshot after the pointer is cleared below.
		d.reportSessionStopped(
			binding.channelID,
			localAgentID,
			firstNonEmpty(strings.TrimSpace(binding.serverAgentID), strings.TrimSpace(serverAgentID)),
			"process retired for restart",
			false,
		)
	}
	// Generation-fenced clear: only clear the old pointer if it is still the
	// retired Process. A replacement that committed during the terminal report is
	// live and must not be cleared.
	sess.mu.Lock()
	if process != nil && sess.process == process {
		sess.process = nil
	}
	sess.mu.Unlock()
	if terminalReportOwner && terminalReportDone != nil {
		close(terminalReportDone)
	}

	// Process.Stop is a strong delivery-admission drain. Once it returns, every
	// handoff that raced this reservation has either failed before ownership or
	// left a complete live journal entry. A restart-like retirement must take
	// over that entry instead of removing the only session that can reconcile it.
	if d.retainRestartSessionForRuntimeOwnership(sess, reservation, agentID) {
		return true
	}

	d.fenceSessionMemoryWatcherForFinalRemoval(sess)
	d.stopMemoryWatcherAndRetry(sess)
	removed := d.sessionManager.RemoveIfCurrent(sessionKey, sess)
	d.publishSessionStateChanged(agentID)
	return removed
}

func (d *Dispatcher) retainRestartSessionForRuntimeOwnership(
	sess *interactiveSession,
	reservation *sessionStartReservation,
	agentID string,
) bool {
	restarted, err := d.tryStartRuntimeDeliveryOwnershipContinuationWithReservation(sess, reservation)
	if err != nil {
		slog.Warn("runtime delivery ownership continuation after session restart",
			"agent_id", agentID,
			"error", err,
		)
		d.scheduleRuntimeDeliveryOwnershipContinuationRetry(sess, err)
	} else if !restarted {
		_, live, loadErr := oldestLiveRuntimeDeliveryOwnership(sess.workDir)
		if loadErr != nil || live {
			if loadErr == nil {
				loadErr = errors.New("runtime delivery ownership continuation did not start after session restart")
			}
			d.scheduleRuntimeDeliveryOwnershipContinuationRetry(sess, loadErr)
			err = loadErr
		}
	}
	sess.mu.Lock()
	sess.restartRetirementInFlight = false
	sess.mu.Unlock()
	if restarted || err != nil {
		d.publishSessionStateChanged(agentID)
		return true
	}
	return false
}

// detachPreparedSessionForRestart is the recovery-only retirement path. The
// Process lifecycle owner already made the Process non-accepting while holding
// deliveryGate and recovery already proved that no live delivery owner exists,
// so this path must neither call public Process.Stop nor launch a continuation.
func (d *Dispatcher) detachPreparedSessionForRestart(sess *interactiveSession) bool {
	if sess == nil {
		return false
	}
	sess.mu.Lock()
	if !sess.retired.CompareAndSwap(false, true) {
		sess.mu.Unlock()
		return false
	}
	sessionKey := sess.sessionKey
	agentID := sess.agentID
	process := sess.process
	binding := sess.binding
	ephemeral := sess.ephemeral
	serverAgentID := sess.serverAgentID
	sess.mu.Unlock()
	if strings.TrimSpace(sessionKey) == "" {
		sessionKey = agentID
	}

	d.fenceSessionMemoryWatcherForFinalRemoval(sess)
	d.stopMemoryWatcherAndRetry(sess)
	revokeSessionMCPToken(sess)
	if process != nil {
		d.revokeExecProcessAuthorization(process)
		// Publish the terminal stopped/idle turn-state WHILE sess.process is
		// still the retired generation, so the AgentTeamReporterSink resolves idle
		// instead of an empty snapshot after the pointer is cleared below.
		if !ephemeral {
			d.reportSessionStopped(
				binding.channelID,
				firstNonEmpty(strings.TrimSpace(process.AgentID), strings.TrimSpace(agentID)),
				firstNonEmpty(strings.TrimSpace(binding.serverAgentID), strings.TrimSpace(serverAgentID)),
				"process retired after recovery",
				false,
			)
		}
		// Generation-fenced clear: only clear the old pointer if it is still the
		// retired Process. A replacement that committed during the terminal
		// report is live and must not be cleared.
		sess.mu.Lock()
		if sess.process == process {
			sess.process = nil
		}
		sess.mu.Unlock()
	}
	removed := d.sessionManager.RemoveIfCurrent(sessionKey, sess)
	d.publishSessionStateChanged(agentID)
	return removed
}

func waitForSessionStart(ctx context.Context, waitCh <-chan struct{}) error {
	if waitCh == nil {
		return nil
	}
	if ctx == nil {
		ctx = context.Background()
	}
	select {
	case <-waitCh:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

// PrepareRuntimeMessageToolDeliveries establishes recovery boundaries for
// messages included in a daemon-local MCP result. The result's final transport
// hop has not committed yet, so this is intentionally a preflight only; exact
// runtime/Exec notification happens from the Delivered receipt.
func (d *Dispatcher) PrepareRuntimeMessageToolDeliveries(agentID string, messages []types.InboxMessage) int {
	session, err := d.lookupSession(agentID)
	if err != nil {
		return 0
	}

	session.mu.Lock()
	process := session.process
	session.mu.Unlock()
	if process == nil {
		return 0
	}
	return process.PrepareRuntimeMessageDeliveryRecoveries(messages, agent.RuntimeDeliveryPathToolOutput)
}

// InboxMessagesForFreshnessPage returns one bounded page of pending inbound
// messages for one target freshness key without consuming unrelated deliveries.
func (d *Dispatcher) InboxMessagesForFreshnessPage(agentID string, targetKey string, throughSeq int64, limit int) (types.InboxReadPage, error) {
	return d.InboxMessagesForFreshnessPageForRuntime(agentID, targetKey, throughSeq, limit, "")
}

// InboxMessagesForFreshnessPageForRuntime binds freshness claims to the exact
// runtime session that owns the final MCP response hop.
func (d *Dispatcher) InboxMessagesForFreshnessPageForRuntime(agentID string, targetKey string, throughSeq int64, limit int, runtimeSessionID string) (types.InboxReadPage, error) {
	session, err := d.lookupSession(agentID)
	if err != nil {
		return types.InboxReadPage{}, err
	}

	session.mu.Lock()
	process := session.process
	session.mu.Unlock()
	if process == nil {
		return types.InboxReadPage{}, fmt.Errorf("interactive session %q is not ready", agentID)
	}
	return process.InboxMessagesForFreshnessPageForRuntime(targetKey, throughSeq, limit, runtimeSessionID), nil
}

// CurrentTurnMessageID returns the message ID of the current turn's triggering
// delivery for the given agent. It mirrors InboxMessagesForFreshnessPageForRuntime
// for session/process resolution and returns "" when the session or process is
// unavailable. The message ID is daemon-authoritative and must never be sourced
// from model-controlled arguments.
func (d *Dispatcher) CurrentTurnMessageID(agentID string) string {
	session, err := d.lookupSession(agentID)
	if err != nil {
		return ""
	}
	session.mu.Lock()
	process := session.process
	session.mu.Unlock()
	if process == nil {
		return ""
	}
	return process.CurrentTurnMessageID()
}

// CurrentTurnDeliveryTarget returns the canonical target reference of the
// current turn's triggering delivery, fenced to the process generation that
// owns the given Exec authorization. When the session's current process does
// not carry exactly that authorization — for example a replaced process's
// still-valid credential asking after its successor started — it returns ""
// so consumers fail closed instead of reading the successor's turn. It is
// daemon-authoritative turn state used for conversation-scoped defaults (for
// example the Exec task-create board or source-thread target).
func (d *Dispatcher) CurrentTurnDeliveryTarget(agentID, authorizationID string) string {
	authorizationID = strings.TrimSpace(authorizationID)
	if authorizationID == "" {
		return ""
	}
	// Launch-window path: first-turn Exec calls may race inside Spawn before
	// the Process is published. The in-flight Process bound to this exact
	// authorization answers, gated by a start reservation that must stay
	// current on both sides of the read.
	if binding, owned := d.pendingTurnTarget(authorizationID); owned {
		if !d.sessionManager.startReservationCurrent(binding.reservation) {
			return ""
		}
		target := binding.process.CurrentTurnDeliveryTarget()
		if current, stillOwned := d.pendingTurnTarget(authorizationID); stillOwned &&
			current.process == binding.process &&
			d.sessionManager.startReservationCurrent(current.reservation) {
			return target
		}
		// Fall through: publication may have landed between the reads.
	}
	session, err := d.lookupSession(agentID)
	if err != nil {
		return ""
	}
	session.mu.Lock()
	process := session.process
	session.mu.Unlock()
	if process == nil {
		return ""
	}
	if process.ExecAuthorizationID() != authorizationID {
		return ""
	}
	return process.CurrentTurnDeliveryTarget()
}

// AcknowledgeRuntimeMessageToolDeliveriesForSession is the exact runtime-session
// fenced variant used by deferred Codex MCP callbacks.
func (d *Dispatcher) AcknowledgeRuntimeMessageToolDeliveriesForSession(agentID string, receipts []string, outcome types.MCPToolDeliveryOutcome, runtimeSessionID string) {
	session, err := d.lookupSession(agentID)
	if err != nil {
		return
	}
	session.mu.Lock()
	process := session.process
	session.mu.Unlock()
	if process == nil {
		return
	}
	process.AcknowledgeRuntimeMessageToolDeliveriesForSession(receipts, outcome, runtimeSessionID)
}

// PendingRuntimeInputCoversFreshness reports whether an ack-aware direct input
// is waiting on an acknowledgement and, once acked, can cover the requested
// target frontier without routing the runtime to message_history_read.
func (d *Dispatcher) PendingRuntimeInputCoversFreshness(agentID string, targetKey string, throughSeq int64) bool {
	session, err := d.lookupSession(agentID)
	if err != nil {
		return false
	}

	session.mu.Lock()
	process := session.process
	session.mu.Unlock()
	if process == nil {
		return false
	}
	return process.PendingRuntimeInputCoversFreshness(targetKey, throughSeq)
}

// ReconcileInboxMessages removes local inbox entries that the runtime already
// consumed through another authoritative delivery.
func (d *Dispatcher) ReconcileInboxMessages(agentID string, messageIDs []string) error {
	session, err := d.lookupSession(agentID)
	if err != nil {
		return err
	}

	session.mu.Lock()
	process := session.process
	session.mu.Unlock()
	if process == nil {
		return fmt.Errorf("interactive session %q is not ready", agentID)
	}
	process.ReconcileInboxMessages(messageIDs)
	return nil
}

// ReconcileRuntimeAckedDeliveries suppresses message IDs already consumed or
// acknowledged by the runtime so authoritative unread replay does not redeliver
// them.
func (d *Dispatcher) ReconcileRuntimeAckedDeliveries(agentID string, messageIDs []string) ([]string, error) {
	session, err := d.lookupSession(agentID)
	if err != nil {
		return nil, err
	}

	session.mu.Lock()
	process := session.process
	session.mu.Unlock()
	if process == nil {
		return nil, fmt.Errorf("interactive session %q is not ready", agentID)
	}
	return process.ReconcileRuntimeAckedMessages(messageIDs), nil
}

// RememberMessageFreshnessObserved records AgentTeam target frontiers that have
// become visible to the runtime through MCP or delivery prompts.
func (d *Dispatcher) RememberMessageFreshnessObserved(agentID string, messages []types.InboxMessage) bool {
	session, err := d.lookupSession(agentID)
	if err != nil {
		return false
	}

	session.mu.Lock()
	process := session.process
	session.mu.Unlock()
	if process == nil {
		return false
	}
	process.RememberObservedFrontiers(messages)
	return true
}

// MessageFreshnessSnapshot returns the observed frontier for targetKey in the
// live session identified by agentID.
func (d *Dispatcher) MessageFreshnessSnapshot(agentID string, targetKey string) (types.MessageFreshnessFrontier, bool) {
	session, err := d.lookupSession(agentID)
	if err != nil {
		return types.MessageFreshnessFrontier{}, false
	}

	session.mu.Lock()
	process := session.process
	session.mu.Unlock()
	if process == nil {
		return types.MessageFreshnessFrontier{}, false
	}
	return process.MessageFreshnessSnapshot(targetKey)
}

// SendScope returns the turn send-scope of the exact process
// bound to the caller's bearer for the session identified by agentID. The
// pending path serves the unpublished launch window through the launch-scope
// registry, so a first-turn tool call during Spawn reads its own process's
// scope. The published path requires the bearer token, the exact process, and
// the current non-retired session to match before AND after the snapshot.
// ok=false means the caller must fail closed.
func (d *Dispatcher) SendScope(ctx context.Context, agentID string) (messagesend.Scope, bool) {
	identity, bound := messagesend.BearerID(ctx)
	if !bound {
		return messagesend.Scope{}, false
	}
	// Pending launch path: the registry binds the caller's bearer to its
	// in-flight Process and start reservation until publication hands it
	// over. An invalidated reservation fails closed before and after the
	// snapshot.
	if binding, owned := d.pendingSendScope(identity); owned {
		if !d.sessionManager.startReservationCurrent(binding.reservation) {
			return messagesend.Scope{}, false
		}
		process := binding.process
		scope, live, generation := process.SendScopeWithGeneration()
		if !live {
			return messagesend.Scope{}, false
		}
		// Sandwich fence: the binding must be unchanged before AND after the
		// Process-side atomic validation, so an invalidation landing between
		// the two binding reads can never authorize via a stale bearer.
		if !process.ValidateSendScope(generation) {
			return messagesend.Scope{}, false
		}
		if current, stillOwned := d.pendingSendScope(identity); stillOwned &&
			current.process == process &&
			d.sessionManager.startReservationCurrent(current.reservation) &&
			process.ValidateSendScope(generation) {
			return scope, true
		}
		// Fall through: the launch may have published between the two reads;
		// the published path below re-validates atomically.
	}
	session, err := d.lookupSession(agentID)
	if err != nil || session == nil {
		return messagesend.Scope{}, false
	}
	if session.retired.Load() {
		return messagesend.Scope{}, false
	}
	session.mu.Lock()
	token := session.mcpToken
	process := session.process
	session.mu.Unlock()
	if process == nil || messagesend.TokenIdentity(token) != identity {
		return messagesend.Scope{}, false
	}
	scope, live, generation := process.SendScopeWithGeneration()
	if !live {
		return messagesend.Scope{}, false
	}
	// Re-verify the binding after the snapshot: the session must still be the
	// current, non-retired owner of the same token and process.
	if session.retired.Load() {
		return messagesend.Scope{}, false
	}
	session.mu.Lock()
	current := session.process
	currentToken := session.mcpToken
	session.mu.Unlock()
	if current != process || messagesend.TokenIdentity(currentToken) != identity {
		return messagesend.Scope{}, false
	}
	// Sandwich fence: atomic Process validation between two binding reads —
	// a replacement landing after the first check invalidates the second.
	if !process.ValidateSendScope(generation) {
		return messagesend.Scope{}, false
	}
	currentSession, err := d.lookupSession(agentID)
	if err != nil || currentSession != session || currentSession.retired.Load() {
		return messagesend.Scope{}, false
	}
	if !process.ValidateSendScope(generation) {
		return messagesend.Scope{}, false
	}
	return scope, true
}

// bindPendingScope binds a freshly minted MCP bearer identity to its
// in-flight, not-yet-published Process and its start reservation so
// first-turn tool calls during Spawn read their own scope; an invalidated
// reservation fails the binding closed.
func (d *Dispatcher) bindPendingScope(identity string, process *agent.Process, reservation *sessionStartReservation) {
	identity = strings.TrimSpace(identity)
	if identity == "" || process == nil {
		return
	}
	d.pendingSendScopeMu.Lock()
	if d.pendingSendScopes == nil {
		d.pendingSendScopes = make(map[string]pendingSendScopeBinding)
	}
	d.pendingSendScopes[identity] = pendingSendScopeBinding{process: process, reservation: reservation}
	d.pendingSendScopeMu.Unlock()
}

// pendingSendScopeBinding is one launch-window binding: the in-flight Process
// plus the start reservation that must still be current for the binding to be
// honored.
type pendingSendScopeBinding struct {
	process     *agent.Process
	reservation *sessionStartReservation
}

// pendingTurnTargetBinding is the Exec-authorization-keyed launch-window
// binding for turn-delivery targets: the in-flight Process answers task
// create defaults before publication, gated by the same start reservation.
type pendingTurnTargetBinding struct {
	process     *agent.Process
	reservation *sessionStartReservation
}

func (d *Dispatcher) bindPendingTurnTarget(authorizationID string, process *agent.Process, reservation *sessionStartReservation) {
	authorizationID = strings.TrimSpace(authorizationID)
	if authorizationID == "" || process == nil {
		return
	}
	d.pendingTurnTargetMu.Lock()
	if d.pendingTurnTargets == nil {
		d.pendingTurnTargets = make(map[string]pendingTurnTargetBinding)
	}
	d.pendingTurnTargets[authorizationID] = pendingTurnTargetBinding{process: process, reservation: reservation}
	d.pendingTurnTargetMu.Unlock()
}

func (d *Dispatcher) unbindPendingTurnTarget(authorizationID string) {
	d.pendingTurnTargetMu.Lock()
	delete(d.pendingTurnTargets, authorizationID)
	d.pendingTurnTargetMu.Unlock()
}

func (d *Dispatcher) pendingTurnTarget(authorizationID string) (pendingTurnTargetBinding, bool) {
	d.pendingTurnTargetMu.Lock()
	defer d.pendingTurnTargetMu.Unlock()
	binding, ok := d.pendingTurnTargets[authorizationID]
	return binding, ok
}

// unbindPendingScope removes the launch-window binding after publication or
// failure.
func (d *Dispatcher) unbindPendingScope(identity string) {
	d.pendingSendScopeMu.Lock()
	delete(d.pendingSendScopes, identity)
	d.pendingSendScopeMu.Unlock()
}

func (d *Dispatcher) pendingSendScope(identity string) (pendingSendScopeBinding, bool) {
	d.pendingSendScopeMu.Lock()
	defer d.pendingSendScopeMu.Unlock()
	binding, ok := d.pendingSendScopes[identity]
	return binding, ok
}

// RememberMessageFreshnessHeld stores the refresh hint returned by a held send.
func (d *Dispatcher) RememberMessageFreshnessHeld(agentID string, targetKey string, seenUpToSeq int64, latestSeq int64) bool {
	session, err := d.lookupSession(agentID)
	if err != nil {
		return false
	}

	session.mu.Lock()
	process := session.process
	session.mu.Unlock()
	if process == nil {
		return false
	}
	process.RememberFreshnessHeld(targetKey, seenUpToSeq, latestSeq)
	return true
}

// PrepareVisibleReply returns a non-persistent delivery correlation token for
// an explicit reaction binding. It is not used by ordinary visible-reply
// settlement; the optional selector is only a message/target lookup hint.
func (d *Dispatcher) PrepareVisibleReply(agentID string, selectors ...string) string {
	session, err := d.lookupSession(agentID)
	if err != nil {
		return ""
	}
	session.mu.Lock()
	process := session.process
	session.mu.Unlock()
	if process == nil {
		return ""
	}
	return process.PrepareVisibleReply(selectors...)
}

// ConfirmVisibleReplyForTarget settles all acknowledged delivery obligations covered
// by the server's resolved target freshness frontier. This path intentionally
// does not select or inspect a local delivery token.
func (d *Dispatcher) ConfirmVisibleReplyForTarget(agentID, targetKey string, seenUpToSeq int64, messageID string) bool {
	session, err := d.lookupSession(agentID)
	if err != nil {
		return false
	}
	session.mu.Lock()
	process := session.process
	session.mu.Unlock()
	if process == nil {
		return false
	}
	return process.ConfirmVisibleReplyForTarget(targetKey, seenUpToSeq, messageID)
}

func (d *Dispatcher) ConfirmReactionReply(agentID, wakeToken, messageID string) bool {
	session, err := d.lookupSession(agentID)
	if err != nil {
		return false
	}
	session.mu.Lock()
	process := session.process
	session.mu.Unlock()
	if process == nil {
		return false
	}
	return process.ConfirmReactionReply(wakeToken, messageID)
}

// startAgentForAgentTeamSession creates an interactive session and spawns the
// agent process for a agentteam agent:start event.
func (d *Dispatcher) startAgentForAgentTeamSession(
	ctx context.Context,
	ag agent.Config,
	rt runtimeconfig.Runtime,
	channelID string,
	serverAgentID string,
	serverConfig types.AgentStartConfig,
	wake *types.Message,
) error {
	if d.sessionManager.Closed() {
		return errSessionManagerClosed
	}
	binding, msg, err := agentStartBindingAndMessage(channelID, serverAgentID, serverConfig, wake)
	if err != nil {
		return err
	}

	agentID := strings.TrimSpace(ag.ID)
	sessionKey := sessionKeyForBinding(agentID, binding).String()

	existing, _ := d.sessionManager.Get(sessionKey)
	reservation, waitCh, shouldStart := d.sessionManager.BeginStart(sessionKey, existing)
	if !shouldStart {
		if d.sessionManager.Closed() {
			return errSessionManagerClosed
		}
		if err := waitForSessionStart(ctx, waitCh); err != nil {
			if isWakeMessage(msg) {
				d.emitWakeFailed(binding.channelID, binding.serverAgentID, msg, err)
			}
			return err
		}
		if isWakeMessage(msg) {
			_, err := d.deliverResolvedSession(ctx, sessionKey, ag, rt, binding, msg, startSessionOptions{
				requireProcessHandoff: strings.TrimSpace(msg.MessageID) != "",
			})
			return err
		}
		if !d.sessionHasLiveProcess(sessionKey) {
			return fmt.Errorf("agent %q did not start after waiting for in-flight start", serverAgentID)
		}
		return nil
	}

	if existing != nil {
		existing.mu.Lock()
		proc := existing.process
		existing.mu.Unlock()
		if proc != nil {
			revokeSessionMCPToken(existing)
			d.revokeExecProcessAuthorization(proc)
			proc.Stop()
		}
	}

	session, err := d.startSession(ctx, sessionKey, ag, rt, binding, msg, startSessionOptions{
		suppressIdleStartTurn:      len(serverConfig.UnreadSummary) == 0,
		requireProcessHandoff:      strings.TrimSpace(msg.MessageID) != "",
		startReservation:           reservation,
		authoritativeStartIdentity: true,
	})
	d.sessionManager.FinishStart(reservation)
	if err != nil {
		slog.Warn("failed to start agent on agent:start",
			"agent_id", serverAgentID, "error", err)
		d.emitRuntimeStartFailed(channelID, serverAgentID, msg, err)
		return err
	}

	if isWakeMessage(msg) {
		session.recordHistory(msg, msg.Content, msg.Attachments(), msg.Timestamp)
	}
	d.publishSessionStateChanged(agentID)

	return nil
}

func (d *Dispatcher) sessionHasLiveProcess(sessionKey string) bool {
	session, ok := d.sessionManager.Get(sessionKey)
	if !ok || session == nil {
		return false
	}
	session.mu.Lock()
	process := session.process
	session.mu.Unlock()
	return process != nil
}

func agentStartBindingAndMessage(channelID, serverAgentID string, serverConfig types.AgentStartConfig, wake *types.Message) (conversationBinding, types.InboxMessage, error) {
	binding := conversationBinding{
		channelID:              channelID,
		serverAgentID:          serverAgentID,
		serverAgentName:        serverConfig.Name,
		serverAgentDisplayName: serverConfig.DisplayName,
		serverRuntime:          serverConfig.Runtime,
		serverModel:            serverConfig.Model,
		serverReasoningEffort:  serverConfig.ReasoningEffort,
		serverSessionID:        serverConfig.SessionID,
		serverMaterialization:  serverConfig.Materialization,
		unreadSummary:          serverConfig.UnreadSummary,
	}
	if wake == nil {
		return binding, types.InboxMessage{
			AuthorProfile: types.AuthorProfile{
				ActorType:   "system",
				DisplayName: "system",
				TargetID:    "system",
			},
			Content:   "",
			Timestamp: time.Now().Format(time.RFC3339),
		}, nil
	}
	wakeBinding, msg, err := bindingFromSession(wake)
	if err != nil {
		return conversationBinding{}, types.InboxMessage{}, err
	}
	if strings.TrimSpace(wakeBinding.channelID) == "" {
		wakeBinding.channelID = channelID
	}
	wakeBinding.serverAgentID = serverAgentID
	wakeBinding.serverAgentName = serverConfig.Name
	wakeBinding.serverAgentDisplayName = serverConfig.DisplayName
	wakeBinding.serverRuntime = serverConfig.Runtime
	wakeBinding.serverModel = serverConfig.Model
	wakeBinding.serverReasoningEffort = serverConfig.ReasoningEffort
	wakeBinding.serverSessionID = serverConfig.SessionID
	wakeBinding.serverMaterialization = serverConfig.Materialization
	if len(serverConfig.UnreadSummary) > 0 {
		wakeBinding.unreadSummary = serverConfig.UnreadSummary
	}
	return wakeBinding, msg, nil
}

// ---------------------------------------------------------------------------
// Internal session helpers
// ---------------------------------------------------------------------------

func (d *Dispatcher) lookupSession(agentID string) (*interactiveSession, error) {
	session, ok := d.sessionManager.FindByLookupID(agentID)
	if !ok {
		return nil, fmt.Errorf("interactive session %q not found", agentID)
	}
	return session, nil
}

func (d *Dispatcher) startSession(
	ctx context.Context,
	sessionKey string,
	ag agent.Config,
	rt runtimeconfig.Runtime,
	binding conversationBinding,
	msg types.InboxMessage,
	options startSessionOptions,
) (*interactiveSession, error) {
	sessionKey, err := requireSessionKey(sessionKey)
	if err != nil {
		return nil, err
	}
	agentID := strings.TrimSpace(ag.ID)
	if agentID == "" {
		return nil, fmt.Errorf("agent id is required")
	}
	workDir := sessionWorkDirForKey(agentID, sessionKey, options.ephemeral)
	if err := d.ensureSessionWorkDirDurable(workDir); err != nil {
		return nil, fmt.Errorf("create durable agent workdir: %w", err)
	}
	if _, err := migrations.MigrateIdentityDocuments(workDir); err != nil {
		return nil, fmt.Errorf("migrate identity document: %w", err)
	}
	var ownershipLaunchIDs []string
	var ownershipHandoff types.InboxMessage
	ownershipWaitForReconcile := false
	ownershipLaunchCommitted := false
	ownershipLaunchAttempted := false
	defer func() {
		if !ownershipLaunchCommitted && !ownershipLaunchAttempted {
			rollbackRuntimeDeliveryOwnershipLaunch(workDir, ownershipLaunchIDs)
		}
	}()
	persistsRuntimeState := !options.ephemeral || options.durableDelivery
	if persistsRuntimeState {
		restored, err := prepareRuntimeDeliveryOwnershipLaunch(workDir, msg, options.requireProcessHandoff)
		if err != nil {
			return nil, err
		}
		ownershipLaunchIDs = append(ownershipLaunchIDs, restored.NewlyPersistedIDs...)
		for _, reclaimed := range restored.Reclaimed {
			d.emitWakeFailed(binding.channelID, binding.serverAgentID, reclaimed.Message,
				fmt.Errorf("older runtime delivery was failed to keep newer delivery available: %s", reclaimed.MessageID()))
		}
		if restored.HandoffPersisted {
			ownershipHandoff = restored.Message
			options.suppressIdleStartTurn = true
		}
		if restored.WaitForReconcile {
			ownershipWaitForReconcile = true
			options.suppressIdleStartTurn = true
		}
	}
	seedSessionMemoryFromAgentRoot(agentID, workDir)
	if binding.serverMaterialization != nil {
		if err := workspace.MaterializeAgentWorkspace(workDir, *binding.serverMaterialization); err != nil {
			slog.Warn("materialize workspace failed", "agent_id", agentID, "error", err)
		}
	}
	memoryReport, err := initMemoryMD(workDir, agentID, ag)
	if err != nil {
		return nil, fmt.Errorf("init MEMORY.md: %w", err)
	}
	if !options.ephemeral {
		d.sendMemoryLayoutWarning(ctx, binding, memoryReport)
	}
	// Construct the session struct first so attachMemoryWatcher can pin its
	// ContextResolver closure against it. The watcher is started before
	// d.resolve / d.launch so any memory writes the runtime makes during
	// startup are observable from the very first debounce window.
	sess := &interactiveSession{
		sessionKey:          sessionKey,
		agentID:             agentID,
		agent:               ag,
		runtime:             rt,
		binding:             binding,
		workDir:             workDir,
		startedAt:           time.Now(),
		serverAgentID:       binding.serverAgentID,
		ephemeral:           options.ephemeral,
		durableDelivery:     options.durableDelivery,
		suppressCheckpoints: options.suppressCheckpoints,
	}
	var previousSession *interactiveSession
	if options.startReservation != nil {
		previousSession = options.startReservation.expected
	}
	memoryReplacement := d.attachMemoryWatcherReplacing(ctx, sess, previousSession, agentID, workDir)
	memoryReplacementCommitted := false
	defer func() {
		if memoryReplacementCommitted {
			return
		}
		// Some pre-resolve failure paths (for example runtime-session ID
		// allocation) return before abortSessionLaunch has an Exec config.
		// Keep watcher rollback total and idempotent here.
		d.stopMemoryWatcherAndRetry(sess)
		if memoryReplacement == nil {
			return
		}
		restoreCtx, cancelRestore := context.WithTimeout(context.Background(), 60*time.Second)
		defer cancelRestore()
		d.restoreReplacedMemoryWatcher(restoreCtx, memoryReplacement)
	}()
	if err := d.finalizeLegacyPreferencesBeforeSpawn(ctx, sess, previousSession); err != nil {
		return nil, err
	}
	// Launch reads only the dispatcher skill cache; profile records are not
	// binding authority for this agent.
	sess.managedSkillPaths = d.attachSkillSync(ctx, agentID)
	if shouldAllocateAgentTeamRuntimeSession(binding) {
		runtimeSessionID, err := newAgentTeamRuntimeSessionID()
		if err != nil {
			return nil, err
		}
		sess.runtimeSessionID = runtimeSessionID
	}
	if persistedTarget := loadPersistedActiveTaskTarget(workDir); shouldCarrySessionActiveTaskTarget(persistedTarget) && !binding.activeTaskTarget.Valid() {
		binding.activeTaskTarget = persistedTarget
		msg.ActiveTaskTarget = persistedTarget
		sess.binding.activeTaskTarget = persistedTarget
	}
	if activeTarget := binding.activeTaskTarget.Normalize(); activeTarget.Valid() {
		sess.activeTaskTarget = activeTarget
		msg.ActiveTaskTarget = activeTarget
		persistActiveTaskTarget(workDir, activeTarget)
	}
	msg, _ = normalizeInboxWorkflowContext(msg)

	launchMsg := msg
	if strings.TrimSpace(ownershipHandoff.MessageID) != "" || ownershipWaitForReconcile {
		launchMsg = types.InboxMessage{}
		// This launch establishes a transport/session only; the message body
		// remains daemon-owned until delivery or reconciliation acknowledges it.
	}
	lc := &processLaunchContext{
		sess:                       sess,
		agent:                      ag,
		runtime:                    rt,
		binding:                    binding,
		msg:                        launchMsg,
		requestContext:             ctx,
		mode:                       promptModeSession,
		suppressIdleStartTurn:      options.suppressIdleStartTurn,
		ownershipHandoffMessageID:  strings.TrimSpace(ownershipHandoff.MessageID),
		startReservation:           options.startReservation,
		authoritativeStartIdentity: options.authoritativeStartIdentity,
		checkpointMsg:              "", // set after resolve when session ID is known
	}
	if err := d.resolve(lc); err != nil {
		d.abortSessionLaunch(sess, lc.startCfg)
		return nil, err
	}

	// Clear pending identity notice on fresh starts (no existing session).
	if strings.TrimSpace(lc.startCfg.SessionID) == "" {
		if err := identitynotice.Clear(workDir); err != nil {
			slog.Warn("clear pending identity notice for fresh start", "agent_id", agentID, "error", err)
		}
	}

	if strings.TrimSpace(lc.startCfg.SessionID) != "" {
		lc.checkpointMsg = checkpointMsgResume
	} else {
		lc.checkpointMsg = checkpointMsgFreshSession
	}

	// The caller owns the bounded start operation, not the lifetime of the
	// Process it creates. d.launch detaches this cancellation immediately before
	// publication; a cancellation after that boundary is handled by the durable
	// handoff and explicit Process.Stop paths instead of looking like an
	// unexpected child exit.
	processCtx, cancelProcess := context.WithCancel(context.WithoutCancel(ctx))
	stopStartCancellation := context.AfterFunc(ctx, cancelProcess)
	processLifetimeDetached := false
	lc.detachStartCancellation = func() error {
		startErr := ctx.Err()
		if !stopStartCancellation() && startErr == nil {
			startErr = context.Canceled
		}
		if startErr != nil {
			cancelProcess()
			return startErr
		}
		processLifetimeDetached = true
		return nil
	}
	defer func() {
		if processLifetimeDetached {
			return
		}
		stopStartCancellation()
		cancelProcess()
	}()

	ownershipLaunchAttempted = true
	process, err := d.launch(processCtx, lc)
	if err != nil {
		d.abortSessionLaunch(sess, lc.startCfg)
		return nil, err
	}
	if strings.TrimSpace(ownershipHandoff.MessageID) != "" {
		if err := process.DeliverForHandoff(ctx, ownershipHandoff); err != nil {
			d.abortSessionLaunch(sess, lc.startCfg)
			process.Stop()
			return nil, fmt.Errorf("deliver durable initial runtime handoff: %w", err)
		}
	}

	ownershipLaunchCommitted = true
	memoryReplacementCommitted = true
	return sess, nil
}

func persistSessionOnInit(workDir string) func(drivercontract.ParsedEvent) {
	return func(ev drivercontract.ParsedEvent) {
		if ev.Kind == consts.ParsedEventKindSessionInit && strings.TrimSpace(ev.SessionID) != "" {
			persistSessionStateAsync(workDir, ev.SessionID)
		}
	}
}

func (d *Dispatcher) sendMemoryLayoutWarning(ctx context.Context, binding conversationBinding, report workspace.MemoryLayoutReport) {
	if strings.TrimSpace(report.Warning) == "" {
		return
	}
	ch, ok := d.serverConnections.ServerConnection(binding.channelID)
	if !ok || ch == nil || binding.meta == nil {
		slog.Warn("MEMORY.md over budget after compaction",
			"server_id", binding.channelID,
			"characters", report.MemoryIndexCharacters,
			"limit", report.MemoryIndexLimit)
		return
	}
	if err := ch.Send(ctx, binding.meta, types.OutboundMessage{Content: report.Warning}); err != nil {
		slog.Warn("send MEMORY.md over-budget warning",
			"server_id", binding.channelID,
			"characters", report.MemoryIndexCharacters,
			"limit", report.MemoryIndexLimit,
			"error", err)
	}
}

// SetActiveTaskTarget refreshes the task-native routing target for one live
// interactive session.
func (d *Dispatcher) SetActiveTaskTarget(agentID string, target types.ActiveTaskTarget) bool {
	target = target.Normalize()
	if !target.Valid() {
		return false
	}
	session, err := d.lookupSession(agentID)
	if err != nil {
		return false
	}
	session.mu.Lock()
	session.activeTaskTarget = target
	session.binding.activeTaskTarget = target
	workDir := session.workDir
	session.mu.Unlock()
	persistActiveTaskTarget(workDir, target)
	return true
}

// normalizeInboxWorkflowContext resolves the trusted transport metadata once
// and clears the raw metadata copy from the inbox message. Workflow callbacks
// carry no expiry: selection and authorization depend only on the frozen
// callback identity, which remains valid until the Agent replies or an
// infrastructure-level delivery failure resolves the effect.
func normalizeInboxWorkflowContext(msg types.InboxMessage) (types.InboxMessage, types.WorkflowContext) {
	context := msg.ResolvedWorkflowContext().Normalize()
	msg.WorkflowContext = context
	msg.Metadata = types.MetadataWithoutWorkflowContext(msg.Metadata)
	return msg, context
}

func loadPersistedActiveTaskTarget(workDir string) types.ActiveTaskTarget {
	state, err := loadSessionResumeState(workDir)
	if err != nil {
		slog.Debug("load persisted active task target", "work_dir", workDir, "error", err)
		return types.ActiveTaskTarget{}
	}
	return state.ActiveTaskTarget.Normalize()
}

func persistActiveTaskTarget(workDir string, target types.ActiveTaskTarget) {
	target = target.Normalize()
	if !target.Valid() || strings.TrimSpace(workDir) == "" {
		return
	}
	if err := updateSessionResumeState(workDir, func(state *sessionResumeState) error {
		state.ActiveTaskTarget = target
		return nil
	}); err != nil {
		slog.Debug("persist active task target", "work_dir", workDir, "error", err)
	}
}

func shouldCarrySessionActiveTaskTarget(target types.ActiveTaskTarget) bool {
	target = target.Normalize()
	return target.Valid() && strings.TrimSpace(target.Source) != "human_selected"
}

func persistSessionStateAsync(workDir, sessionID string) {
	if err := updateSessionResumeState(workDir, func(state *sessionResumeState) error {
		trimmedSessionID := strings.TrimSpace(sessionID)
		state.LastSessionID = trimmedSessionID
		if state.LastStartConfig != nil {
			state.LastStartConfig.SessionID = trimmedSessionID
		}
		return nil
	}); err != nil {
		slog.Debug("persist session state", "error", err)
	}
}

// persistStartConfigState establishes an unsafe launch epoch before runtime
// callbacks are attached. Each Codex Process gets a strictly newer durable
// epoch, while transition version starts at zero for that epoch.
func persistStartConfigState(workDir, runtime, _serverURL, sessionID string, cfg *types.AgentStartConfig, runtimePolicy ...types.RuntimePolicySnapshot) (uint64, error) {
	return persistStartConfigStateForLaunch(workDir, runtime, _serverURL, sessionID, cfg, true, runtimePolicy...)
}

func persistStartConfigStateForLaunch(
	workDir, runtime, _serverURL, sessionID string,
	cfg *types.AgentStartConfig,
	preservePersistedIdentity bool,
	runtimePolicy ...types.RuntimePolicySnapshot,
) (uint64, error) {
	cfgCopy := cloneAgentStartConfig(cfg)
	trimmedSessionID := strings.TrimSpace(sessionID)
	if cfgCopy != nil && cfgCopy.SessionID == "" {
		cfgCopy.SessionID = trimmedSessionID
	}
	var epoch uint64
	err := updateSessionResumeState(workDir, func(state *sessionResumeState) error {
		// Launch persistence rebuilds a resume-only AgentStartConfig and must not
		// become a competing identity writer. Identity reconciliation owns the
		// complete Name/DisplayName pair, so carry that pair across both initial
		// and deferred launch overwrites once it has been persisted.
		if preservePersistedIdentity && cfgCopy != nil && state.LastStartConfig != nil {
			name := strings.TrimSpace(state.LastStartConfig.Name)
			displayName := strings.TrimSpace(state.LastStartConfig.DisplayName)
			if name != "" && displayName != "" {
				cfgCopy.Name = name
				cfgCopy.DisplayName = displayName
			}
		}
		state.LastStartConfig = cfgCopy
		state.LastSessionID = trimmedSessionID
		state.SessionResumeSafe = false
		if strings.EqualFold(strings.TrimSpace(runtime), string(runtimeconfig.TypeCodex)) {
			state.SessionResumeSafetyEpoch++
			state.SessionResumeSafetyVersion = 0
			epoch = state.SessionResumeSafetyEpoch
		}
		if len(runtimePolicy) > 0 {
			persistRuntimePolicyOnState(state, runtimePolicy[0])
		}
		return nil
	})
	return epoch, err
}

// PersistResumeSafety applies one versioned transition within a Codex Process
// launch epoch. A callback from an older Process, or an older transition in the
// same Process, is a no-op rather than overwriting a newer unsafe boundary.
func PersistResumeSafety(workDir, _serverURL, sessionID string, epoch, version uint64, safe bool) (bool, error) {
	applied := false
	err := updateSessionResumeState(workDir, func(state *sessionResumeState) error {
		if epoch == 0 || state.SessionResumeSafetyEpoch != epoch || version < state.SessionResumeSafetyVersion {
			return nil
		}
		// A safe=true callback is based on a Process-local quiescence snapshot.
		// Accepting a delivery and persisting its live ownership can race that
		// callback after the snapshot but before this transaction. Re-evaluate the
		// durable journal while holding the state lock: queued, pending, or
		// uncertain input is an unsafe restart boundary regardless of callback
		// ordering. Consume a newer callback version while retaining false so the
		// same stale snapshot cannot become authoritative after the owner later
		// reaches a terminal phase.
		if safe && sessionResumeStateHasLiveDeliveryOwnership(*state) {
			state.SessionResumeSafe = false
			if version > state.SessionResumeSafetyVersion {
				trimmedSessionID := strings.TrimSpace(sessionID)
				state.LastSessionID = trimmedSessionID
				state.SessionResumeSafetyVersion = version
				if state.LastStartConfig != nil && trimmedSessionID != "" {
					state.LastStartConfig.SessionID = trimmedSessionID
				}
			}
			return nil
		}
		if version == state.SessionResumeSafetyVersion {
			// Equal versions are idempotent only. A conflicting equal-version
			// callback indicates stale/corrupt ordering and must fail closed.
			applied = state.SessionResumeSafe == safe && strings.TrimSpace(state.LastSessionID) == strings.TrimSpace(sessionID)
			return nil
		}
		trimmedSessionID := strings.TrimSpace(sessionID)
		state.LastSessionID = trimmedSessionID
		state.SessionResumeSafe = safe
		state.SessionResumeSafetyVersion = version
		if state.LastStartConfig != nil && trimmedSessionID != "" {
			state.LastStartConfig.SessionID = trimmedSessionID
		}
		applied = true
		return nil
	})
	return applied, err
}

func sessionResumeStateHasLiveDeliveryOwnership(state sessionResumeState) bool {
	for _, record := range state.RuntimeDeliveryOwnership {
		if runtimeDeliveryOwnershipIsLive(record) {
			return true
		}
	}
	return false
}

const maxSessionHistory = 2000

func (is *interactiveSession) recordHistory(msg types.InboxMessage, content string, attachments []types.MessageAttachment, timestamp string) {
	is.mu.Lock()
	defer is.mu.Unlock()
	is.recordHistoryLocked(msg, content, attachments, timestamp)
}

func (is *interactiveSession) recordHistoryLocked(msg types.InboxMessage, content string, attachments []types.MessageAttachment, timestamp string) {
	is.nextSeq++
	authorIdentity := msg.AuthorIdentity()
	authorProfile := msg.AuthorProfile
	if authorProfile == (types.AuthorProfile{}) && strings.TrimSpace(authorIdentity.DisplayName) != "" {
		authorProfile = types.AuthorProfile{
			ActorType:   strings.TrimSpace(authorIdentity.ActorType),
			PrincipalID: strings.TrimSpace(authorIdentity.PrincipalID),
			MemberID:    strings.TrimSpace(authorIdentity.MemberID),
			TargetID:    strings.TrimSpace(authorIdentity.TargetID),
			DisplayName: strings.TrimSpace(authorIdentity.DisplayName),
		}
	}
	is.history = append(is.history, types.HistoryMessage{
		Seq:           is.nextSeq,
		AuthorProfile: authorProfile,
		Content:       content,
		CreatedAt:     timestamp,
		Attachments:   attachments,
	})
	if len(is.history) > maxSessionHistory {
		is.history = is.history[len(is.history)-maxSessionHistory:]
	}
}

func bindingFromSession(sess *types.Message) (conversationBinding, types.InboxMessage, error) {
	meta := sess.Input.Meta
	if meta == nil {
		return conversationBinding{}, types.InboxMessage{}, fmt.Errorf("session %q is missing channel metadata", sess.ID)
	}

	normalized := types.NormalizeMetadata(meta)
	binding := conversationBinding{
		channelID:       strings.TrimSpace(sess.Input.Channel),
		conversationKey: conversationKeyForMetadata(meta),
		targetID:        normalized.TargetID,
		meta:            meta,
	}

	var attachmentIDs, attachmentNames string

	if sc, ok := meta.(types.SessionConfigurator); ok {
		binding.serverAgentID = sc.ServerAgentID()
		binding.serverRuntime = sc.ServerRuntime()
		binding.serverModel = sc.ServerModel()
		binding.serverReasoningEffort = sc.ServerReasoningEffort()
		binding.serverSessionID = sc.ServerSessionID()
		attachmentIDs = sc.GetAttachmentIDs()
		attachmentNames = sc.GetAttachmentNames()
	}

	authorTarget := firstNonEmpty(normalized.AuthorTargetID, binding.targetID, "user")
	msg := types.InboxMessage{
		MessageID:       firstNonEmpty(normalized.MessageID, sess.ID),
		Target:          inboxMessageTarget(meta),
		ChannelName:     binding.channelID,
		AuthorProfile:   types.AuthorProfile{ActorType: "human", PrincipalID: authorTarget, TargetID: authorTarget, DisplayName: authorTarget},
		Content:         formatInboundContent(sess),
		Timestamp:       parseMessageTimestamp(normalized.RawTimestamp),
		AttachmentIDs:   attachmentIDs,
		AttachmentNames: attachmentNames,
	}

	// Extract agentteam-specific wake metadata.
	if lm, ok := meta.(*types.AgentTeamMeta); ok {
		if len(lm.UnreadSummary) > 0 {
			binding.unreadSummary = lm.UnreadSummary
		}
		msg.TargetKey = strings.TrimSpace(lm.TargetKey)
		msg.TargetSeq = lm.TargetSeq
		msg.MessageType = strings.TrimSpace(lm.MessageType)
		msg.InterruptPolicy = strings.TrimSpace(lm.InterruptPolicy)
		if lm.AuthorProfile != (types.AuthorProfile{}) {
			msg.AuthorProfile = lm.AuthorProfile
		}
		msg.Source = lm.Source
		msg.WorkflowContext = lm.WorkflowContext.Normalize()
		if !msg.WorkflowContext.Valid() {
			msg.WorkflowContext = types.WorkflowContextFromMetadata(lm.Metadata)
		}
		msg.Metadata = types.MetadataWithoutWorkflowContext(lm.Metadata)
		applyAgentTeamTargetFields(&msg, lm.Target, binding.channelID)
		msg.ChannelType = firstNonEmpty(strings.TrimSpace(lm.TargetChannelType), msg.ChannelType)
		msg.ChannelName = firstNonEmpty(strings.TrimSpace(lm.TargetChannelName), msg.ChannelName)
		msg.ParentChannelType = firstNonEmpty(strings.TrimSpace(lm.ParentTargetChannelType), msg.ParentChannelType)
		msg.ParentChannelName = firstNonEmpty(strings.TrimSpace(lm.ParentTargetChannelName), msg.ParentChannelName)
		msg.ChannelDisplayName = strings.TrimSpace(lm.ChannelDisplayName)
		msg.ParentChannelDisplayName = strings.TrimSpace(lm.ParentChannelDisplayName)
		msg.ActiveTaskTarget = lm.ActiveTaskTarget.Normalize()
		msg.AttachmentList = lm.MessageAttachments()
		msg.TeamContexts = append([]types.TeamContext(nil), lm.TeamContexts...)
	}
	if strings.TrimSpace(msg.ChannelType) == "" {
		applyAgentTeamTargetFields(&msg, msg.Target, binding.channelID)
	}
	binding.targetScopeType = strings.TrimSpace(msg.ChannelType)
	binding.activeTaskTarget = msg.ResolvedActiveTaskTarget()
	msg.WorkflowContext = msg.ResolvedWorkflowContext()
	msg.ActiveTaskTarget = binding.activeTaskTarget
	return binding, msg, nil
}

func formatInboundContent(sess *types.Message) string {
	content := strings.TrimSpace(sess.Input.Prompt)
	if content == "" {
		content = "[empty message]"
	}

	if sess.Input.Meta == nil {
		return content
	}

	if md, ok := sess.Input.Meta.(types.MediaDecorator); ok {
		if annotation := md.MediaAnnotation(); annotation != "" {
			return content + "\n\n" + annotation
		}
	}

	return content
}

func inboxMessageTarget(meta types.Metadata) string {
	agentteamMeta, ok := meta.(*types.AgentTeamMeta)
	if !ok {
		return ""
	}
	return strings.TrimSpace(agentteamMeta.Target)
}

func applyAgentTeamTargetFields(msg *types.InboxMessage, target string, fallbackChannelName string) {
	if msg == nil {
		return
	}
	channelType, channelName, parentChannelType, parentChannelName := agentteamTargetFields(target)
	if strings.TrimSpace(msg.ChannelType) == "" {
		msg.ChannelType = channelType
	}
	if strings.TrimSpace(msg.ChannelName) == "" {
		msg.ChannelName = firstNonEmpty(channelName, strings.TrimSpace(fallbackChannelName))
	}
	if strings.TrimSpace(msg.ParentChannelType) == "" {
		msg.ParentChannelType = parentChannelType
	}
	if strings.TrimSpace(msg.ParentChannelName) == "" {
		msg.ParentChannelName = parentChannelName
	}
}

func agentteamTargetFields(target string) (channelType, channelName, parentChannelType, parentChannelName string) {
	parsed, err := domain.ParseMessageTargetReference(target)
	if err != nil {
		return "", "", "", ""
	}
	switch parsed.Kind {
	case domain.MessageTargetKindChannel:
		return "channel", parsed.Thread.ChannelSlug, "", ""
	case domain.MessageTargetKindDM:
		return "dm", parsed.Thread.PeerMemberID, "", ""
	case domain.MessageTargetKindReplyThread:
		parentType := "channel"
		parentName := parsed.Thread.ChannelSlug
		if parsed.Thread.Type == domain.ThreadTypeDM {
			parentType = "dm"
			parentName = parsed.Thread.PeerMemberID
		}
		return "thread", parsed.Thread.ShortID, parentType, parentName
	case domain.MessageTargetKindTask:
		return "task", parsed.TaskScopeID, "", ""
	case domain.MessageTargetKindWorkflow:
		return "workflow", parsed.WorkflowPlanID, "", ""
	case domain.MessageTargetKindTeam:
		if strings.TrimSpace(parsed.Team.ShortID) == "" {
			return "channel", parsed.Team.TeamID, "", ""
		}
		return "thread", parsed.Team.ShortID, "channel", parsed.Team.TeamID
	}
	return "", "", "", ""
}

func parseMessageTimestamp(timestamp string) string {
	if msText := strings.TrimSpace(timestamp); msText != "" {
		if ms, err := strconv.ParseInt(msText, 10, 64); err == nil && ms > 0 {
			return time.UnixMilli(ms).Format(time.RFC3339)
		}
	}
	return time.Now().Format(time.RFC3339)
}

func seedSessionMemoryFromAgentRoot(agentID, workDir string) {
	agentRoot := sessionWorkDir(agentID)
	if strings.TrimSpace(agentRoot) == "" || strings.TrimSpace(workDir) == "" {
		return
	}
	if agentRoot == workDir {
		seedAgentRootFromLatestSession(agentID, agentRoot)
		return
	}
	source := filepath.Join(agentRoot, "MEMORY.md")
	target := filepath.Join(workDir, "MEMORY.md")
	if _, err := os.Stat(target); err == nil {
		return
	}
	content, err := os.ReadFile(source)
	if err != nil {
		return
	}
	if err := os.MkdirAll(filepath.Dir(target), 0o755); err != nil {
		return
	}
	if err := os.WriteFile(target, content, 0o644); err != nil {
		slog.Debug("seed session MEMORY.md from agent root", "agent_id", agentID, "error", err)
	}
}

func seedAgentRootFromLatestSession(agentID, agentRoot string) {
	if _, err := os.Stat(filepath.Join(agentRoot, "MEMORY.md")); err == nil {
		return
	}

	latestDir := latestLegacySessionDir(agentRoot)
	if latestDir == "" {
		return
	}
	copyFileIfMissing(filepath.Join(latestDir, "MEMORY.md"), filepath.Join(agentRoot, "MEMORY.md"))
	copyFileIfMissing(filepath.Join(latestDir, sessionResumeStateFilename), filepath.Join(agentRoot, sessionResumeStateFilename))
	for _, dir := range []string{"state", "kb", "archive"} {
		copyDirIfMissing(filepath.Join(latestDir, dir), filepath.Join(agentRoot, dir))
	}
	slog.Debug("seed agent root memory from latest session", "agent_id", agentID, "source", latestDir)
}

func latestLegacySessionDir(agentRoot string) string {
	sessionDirs, err := os.ReadDir(filepath.Join(agentRoot, "sessions"))
	if err != nil {
		return ""
	}
	var latestDir string
	var latestModTime time.Time
	for _, entry := range sessionDirs {
		if !entry.IsDir() {
			continue
		}
		sessionDir := filepath.Join(agentRoot, "sessions", entry.Name())
		info, err := os.Stat(filepath.Join(sessionDir, "MEMORY.md"))
		if err != nil {
			continue
		}
		if latestDir == "" || info.ModTime().After(latestModTime) {
			latestDir = sessionDir
			latestModTime = info.ModTime()
		}
	}
	return latestDir
}

func copyFileIfMissing(source, target string) {
	if _, err := os.Stat(target); err == nil {
		return
	}
	content, err := os.ReadFile(source)
	if err != nil {
		return
	}
	if err := os.MkdirAll(filepath.Dir(target), 0o755); err != nil {
		return
	}
	if err := os.WriteFile(target, content, 0o644); err != nil {
		slog.Debug("copy session file", "source", source, "target", target, "error", err)
	}
}

func copyDirIfMissing(source, target string) {
	if _, err := os.Stat(target); err == nil {
		return
	}
	if info, err := os.Stat(source); err != nil || !info.IsDir() {
		return
	}
	if err := filepath.Walk(source, func(path string, info os.FileInfo, err error) error {
		if err != nil {
			return nil
		}
		rel, err := filepath.Rel(source, path)
		if err != nil {
			return nil
		}
		dst := filepath.Join(target, rel)
		if info.IsDir() {
			return os.MkdirAll(dst, info.Mode().Perm())
		}
		if !info.Mode().IsRegular() {
			return nil
		}
		copyFileIfMissing(path, dst)
		return nil
	}); err != nil {
		slog.Debug("copy session directory", "source", source, "target", target, "error", err)
	}
}

func applyCodexAppServerResumePolicy(workDir string, startCfg *drivercontract.LaunchConfig, currentFingerprint string) {
	if startCfg == nil || strings.TrimSpace(startCfg.SessionID) == "" {
		return
	}

	state, err := loadSessionResumeState(workDir)
	if err != nil {
		slog.Debug("load session resume state for codex app-server implementation policy", "error", err)
	}

	savedFingerprint := ""
	if state.LastStartConfig != nil {
		savedFingerprint = strings.TrimSpace(state.LastStartConfig.InstructionsFingerprint)
	}
	currentFingerprint = strings.TrimSpace(currentFingerprint)
	if currentFingerprint == "" && startCfg.CompiledPrompt != nil {
		currentFingerprint = codexAppServerInstructionsFingerprint(startCfg.CompiledPrompt.LaunchSurface("baseInstructions"), startCfg.CompiledPrompt.LaunchSurface("developerInstructions"))
	}
	sessionID := startCfg.SessionID

	switch {
	case savedFingerprint == "":
		// Local codex-cli 0.120.0 accepts resume-time instruction fields in the
		// schema but keeps executing the legacy thread's original instructions.
		// Force a clean cutover so pre-split sessions do not keep stale rules.
		startCfg.SessionID = ""
		slog.Info("dropping saved codex session: legacy thread cannot refresh instructions on resume",
			"work_dir", workDir, "session_id", sessionID)
	case !codexAppServerInstructionsFingerprintMatches(savedFingerprint, currentFingerprint):
		slog.Info("dropping saved codex session: instructions changed",
			"work_dir", workDir, "session_id", sessionID)
		startCfg.SessionID = ""
	}
}

func (d *Dispatcher) validateSessionResume(workDir, sessionID, runtime, serverURL, currentFingerprint string) string {
	// Keep the call shape while deliberately ignoring transport identity. A
	// port:0 restart, proxy change, or server URL change must not strand the
	// provider thread.
	_ = serverURL
	sessionID = strings.TrimSpace(sessionID)
	if sessionID == "" {
		return ""
	}

	state, err := loadSessionResumeState(workDir)
	if err != nil {
		slog.Info("dropping saved session: resume state unavailable",
			"runtime", runtime, "session_id", sessionID, "error", err)
		return ""
	}

	savedFingerprint := ""
	if state.LastStartConfig != nil {
		savedFingerprint = strings.TrimSpace(state.LastStartConfig.InstructionsFingerprint)
	}
	if trimmedCurrentFingerprint := strings.TrimSpace(currentFingerprint); trimmedCurrentFingerprint != "" {
		switch {
		case savedFingerprint == "":
			slog.Info("dropping saved session: instruction fingerprint unavailable",
				"runtime", runtime, "session_id", sessionID)
			return ""
		case runtimeconfig.EffectiveRuntimeType(runtimeconfig.RuntimeType(runtime)) == runtimeconfig.TypeCodex &&
			!codexAppServerInstructionsFingerprintMatches(savedFingerprint, trimmedCurrentFingerprint):
			slog.Info("dropping saved session: instructions changed",
				"runtime", runtime, "session_id", sessionID)
			return ""
		case runtimeconfig.EffectiveRuntimeType(runtimeconfig.RuntimeType(runtime)) != runtimeconfig.TypeCodex &&
			savedFingerprint != trimmedCurrentFingerprint:
			slog.Info("dropping saved session: instructions changed",
				"runtime", runtime, "session_id", sessionID)
			return ""
		}
	}

	// Runtime authority owns whether this provider session exists and can be
	// resumed. Codex SessionResumeSafe is a delivery ownership/reconciliation
	// fence, not a provider-thread existence guard. An unsafe thread is still
	// handed to the runtime's authoritative resume/read path; an uncertain
	// input remains owned by the durable journal and is never blindly replayed.
	return sessionID
}

func cloneStringMap(input map[string]string) map[string]string {
	if len(input) == 0 {
		return nil
	}
	result := make(map[string]string, len(input))
	for key, value := range input {
		result[key] = value
	}
	return result
}

func cloneStringMapForWrite(input map[string]string) map[string]string {
	result := cloneStringMap(input)
	if result == nil {
		return map[string]string{}
	}
	return result
}

func normalizeCheckpointSummary(input string) string {
	normalized := strings.Join(strings.Fields(strings.TrimSpace(input)), " ")
	if normalized == "" {
		return ""
	}
	runes := []rune(normalized)
	if len(runes) <= 240 {
		return normalized
	}
	return string(runes[:240]) + "…"
}

func firstNonEmpty(values ...string) string {
	for _, value := range values {
		if strings.TrimSpace(value) != "" {
			return value
		}
	}
	return ""
}

func (d *Dispatcher) eventReporter(channelID string) types.EventReporter {
	ch, ok := d.serverConnections.ServerConnection(channelID)
	if !ok {
		return nil
	}
	reporter, ok := ch.(types.EventReporter)
	if !ok {
		return nil
	}
	return reporter
}

// agentTurnStateForServer resolves the daemon-authoritative real turn execution
// state AND the owning RuntimeSessionID for one agent as one immutable
// observation from a single locked SessionSnapshot, or a zero observation when
// the agent has no running session. It reads the live session's Process state so
// delivery/turn events can publish the current active/queued/idle signal — and
// the exact session that owns it — to the server Hub overlay for Workflow
// callback recovery. Both values come from the same generation so a torn
// old-idle + new-session pair is impossible.
func (d *Dispatcher) agentTurnStateForServer(serverID, agentID string) TurnStateObservation {
	if d == nil {
		return TurnStateObservation{}
	}
	snapshot, ok := d.Session(strings.TrimSpace(agentID))
	if !ok {
		return TurnStateObservation{}
	}
	return TurnStateObservation{
		TurnState: strings.TrimSpace(snapshot.TurnState),
		SessionID: strings.TrimSpace(snapshot.RuntimeSessionID),
	}
}

func (d *Dispatcher) resolveMCPConfigs(channelID, agentID, serverAgentID, sessionID, authToken string, rt runtimeconfig.RuntimeType) []types.NamedMCPConfig {
	// The MCP bridge is routed by server connection ID in the URL, so keep
	// using the session's bound server ID even if the in-memory manager does
	// not have a live instance for it.
	resolvedServerID := strings.TrimSpace(channelID)
	if ch, ok := d.serverConnections.ServerConnection(resolvedServerID); ok {
		if chID := strings.TrimSpace(ch.ID()); chID != "" {
			resolvedServerID = chID
		}
	}

	agentteamMCP := types.MCPConfig{
		URL:     mcpAgentTeamEndpoint(d.serverURL, resolvedServerID, agentID),
		Headers: mcpAuthHeaders(authToken),
	}

	configs := []types.NamedMCPConfig{agentteamMCP.ToNamed("agentteam")}

	reservedMCPNames := namedMCPConfigSet(configs)
	configs = append(configs, externalMCPConfigs(profile.Default().MCPServers().All(), reservedMCPNames)...)

	return configs
}

func namedMCPConfigSet(configs []types.NamedMCPConfig) map[string]struct{} {
	names := make(map[string]struct{}, len(configs))
	for _, cfg := range configs {
		name := strings.TrimSpace(cfg.Name)
		if name != "" {
			names[name] = struct{}{}
		}
	}
	return names
}

func externalMCPConfigs(servers []profile.MCPServer, reserved map[string]struct{}) []types.NamedMCPConfig {
	configs := make([]types.NamedMCPConfig, 0, len(servers))
	for _, server := range servers {
		if !server.Enabled {
			continue
		}
		name := strings.TrimSpace(server.ID)
		if name == "" {
			continue
		}
		if _, exists := reserved[name]; exists {
			continue
		}
		config, ok := externalMCPConfig(server)
		if !ok {
			continue
		}
		reserved[name] = struct{}{}
		configs = append(configs, config)
	}
	return configs
}

func externalMCPConfig(server profile.MCPServer) (types.NamedMCPConfig, bool) {
	base := types.NamedMCPConfig{
		Name: strings.TrimSpace(server.ID),
	}
	switch server.Transport {
	case profile.MCPServerTransportHTTP:
		if server.HTTP == nil || strings.TrimSpace(server.HTTP.URL) == "" {
			return types.NamedMCPConfig{}, false
		}
		base.Transport = types.MCPTransportHTTP
		base.URL = strings.TrimSpace(server.HTTP.URL)
		base.Headers = resolveExternalMCPHeaders(server.HTTP.Headers)
		return base, true
	case profile.MCPServerTransportStdio:
		if server.Stdio == nil || strings.TrimSpace(server.Stdio.Command) == "" {
			return types.NamedMCPConfig{}, false
		}
		base.Transport = types.MCPTransportStdio
		base.Command = strings.TrimSpace(server.Stdio.Command)
		base.Args = append([]string(nil), server.Stdio.Args...)
		base.CWD = strings.TrimSpace(server.Stdio.CWD)
		base.Env = externalMCPEnvMap(server.Stdio.ResolvedEnv())
		return base, true
	default:
		return types.NamedMCPConfig{}, false
	}
}

func resolveExternalMCPHeaders(headers []string) map[string]string {
	if len(headers) == 0 {
		return nil
	}
	resolved := make(map[string]string, len(headers))
	for _, entry := range headers {
		name, value, ok := strings.Cut(entry, ":")
		if !ok {
			continue
		}
		name = strings.TrimSpace(name)
		expanded := profile.ExpandRuntimeEnv(strings.TrimSpace(value))
		if expanded == "" {
			continue
		}
		resolved[name] = expanded
	}
	if len(resolved) == 0 {
		return nil
	}
	return resolved
}

func externalMCPEnvMap(entries []string) map[string]string {
	if len(entries) == 0 {
		return nil
	}
	result := make(map[string]string, len(entries))
	for _, entry := range entries {
		name, value, ok := strings.Cut(entry, "=")
		if !ok || strings.TrimSpace(name) == "" {
			continue
		}
		result[name] = value
	}
	if len(result) == 0 {
		return nil
	}
	return result
}

func (d *Dispatcher) findProcess(agentID string) (*agent.Process, bool) {
	sess, ok := d.sessionManager.FindByLookupID(agentID)
	if !ok || sess == nil {
		return nil, false
	}
	sess.mu.Lock()
	process := sess.process
	sess.mu.Unlock()
	if process == nil {
		return nil, false
	}
	return process, true
}

func (d *Dispatcher) publishRuntimeEvent(ev runtimeevents.Event) {
	ev = runtimeEventWithDefaults(ev)
	if d.runtimeEventBus != nil {
		d.runtimeEventBus.Publish(context.Background(), ev)
	}
	if d.agentteamReporterSink != nil {
		d.agentteamReporterSink.Handle(ev)
	}
	if d.trajectorySink != nil {
		d.trajectorySink.Handle(ev)
	}
	if d.metricsSink != nil {
		d.metricsSink.Handle(ev)
	}
}

func (d *Dispatcher) reportExitStatus(channelID string, ap *agent.Process, serverAgentID string) {
	agentID := serverAgentID
	if agentID == "" {
		agentID = ap.AgentID
	}
	exitCode, _ := ap.ExitSummary()
	if ap.IsSleeping() {
		d.publishRuntimeEvent(runtimeevents.NewDataEvent(
			runtimeevents.NewLifecycleData(
				runtimeevents.KindSessionSleeping,
				"sleeping",
				"",
			),
			runtimeevents.WithContext(runtimeevents.EventContext{
				ServerID:      channelID,
				AgentID:       ap.AgentID,
				ServerAgentID: agentID,
			}),
		))
	} else if exitCode == 0 {
		d.reportSessionStopped(channelID, ap.AgentID, agentID, "process exited normally", false)
	} else {
		d.reportSessionStopped(channelID, ap.AgentID, agentID, fmt.Sprintf("exited with code %d", exitCode), true)
	}
}

// reportSessionStopped publishes the authoritative terminal lifecycle projection.
// Intentional retirement uses this helper instead of reading ExitSummary before
// the asynchronously terminated child has necessarily populated its exit code.
func (d *Dispatcher) reportSessionStopped(channelID, localAgentID, serverAgentID, detail string, warn bool) {
	opts := []runtimeevents.EventOption{
		runtimeevents.WithContext(runtimeevents.EventContext{
			ServerID:      strings.TrimSpace(channelID),
			AgentID:       strings.TrimSpace(localAgentID),
			ServerAgentID: firstNonEmpty(strings.TrimSpace(serverAgentID), strings.TrimSpace(localAgentID)),
		}),
	}
	if warn {
		opts = append(opts, runtimeevents.WithSeverity(runtimeevents.SeverityWarn))
	}
	d.publishRuntimeEvent(runtimeevents.NewDataEvent(
		runtimeevents.NewLifecycleData(
			runtimeevents.KindSessionStopped,
			"offline",
			strings.TrimSpace(detail),
		),
		opts...,
	))
}

func (d *Dispatcher) emitRuntimeEvent(channelID, serverAgentID string, event types.RuntimeEvent) {
	d.publishRuntimeEvent(runtimeEventFromTypesRuntimeEvent(channelID, serverAgentID, event))
}

func (d *Dispatcher) emitTrajectoryEntries(channelID, agentID, sessionID string, entries []types.TrajectoryEntry) {
	trimmedAgentID := strings.TrimSpace(agentID)
	timestamp := time.Now().UnixMilli()
	if d.trajectoryHub != nil && trimmedAgentID != "" && len(entries) > 0 {
		d.trajectoryHub.BroadcastAt(trimmedAgentID, entries, timestamp)
	}
	reporter := d.eventReporter(channelID)
	if reporter == nil || len(entries) == 0 {
		return
	}
	reporter.Emit(types.TrajectoryEvent{
		AgentID:   trimmedAgentID,
		SessionID: strings.TrimSpace(sessionID),
		Timestamp: timestamp,
		Entries:   entries,
	})
}

func (d *Dispatcher) emitRuntimeStartFailed(channelID, serverAgentID string, msg types.InboxMessage, err error) {
	if err == nil {
		return
	}
	d.emitRuntimeEvent(channelID, serverAgentID, types.RuntimeEvent{
		AgentID:    serverAgentID,
		MessageID:  strings.TrimSpace(msg.MessageID),
		EventClass: consts.RuntimeEventClassRuntimeStartFailed,
		Target:     strings.TrimSpace(msg.Target),
		Detail:     strings.TrimSpace(err.Error()),
	})
}

func (d *Dispatcher) emitWakeFailed(channelID, serverAgentID string, msg types.InboxMessage, err error) {
	if err == nil {
		return
	}
	d.emitRuntimeEvent(channelID, serverAgentID, types.RuntimeEvent{
		AgentID:    serverAgentID,
		MessageID:  strings.TrimSpace(msg.MessageID),
		EventClass: consts.RuntimeEventClassWakeFailed,
		Target:     strings.TrimSpace(msg.Target),
		Detail:     strings.TrimSpace(err.Error()),
	})
}

// buildWakeStatusCallback returns a callback that converts agent-level
// WakeStatusEvent values into normalized runtime events and emits them through
// the runtime event sinks.
func (d *Dispatcher) buildWakeStatusCallback(channelID, serverAgentID string, sessions ...*interactiveSession) func(agent.WakeStatusEvent) {
	if len(sessions) > 0 && sessions[0] != nil && sessions[0].ephemeral {
		return nil
	}
	if d.eventReporter(channelID) == nil {
		return nil
	}
	return func(ev agent.WakeStatusEvent) {
		d.publishRuntimeEvent(runtimeEventFromWakeStatus(channelID, serverAgentID, ev))
	}
}

func (d *Dispatcher) buildRuntimeEventCallback(channelID, serverAgentID string) func(agent.RuntimeEvent) {
	if d.eventReporter(channelID) == nil {
		return nil
	}
	return func(ev agent.RuntimeEvent) {
		d.publishRuntimeEvent(runtimeEventFromAgentRuntimeEvent(channelID, serverAgentID, ev))
	}
}

func (d *Dispatcher) buildEventCallback(channelID, sessionMapKey, eventAgentID, workDir string, sessions ...*interactiveSession) func(drivercontract.ParsedEvent) {
	var sess *interactiveSession
	if len(sessions) > 0 {
		sess = sessions[0]
	}
	reportRuntimeEvents := sess == nil || !sess.ephemeral
	var persistFn func(drivercontract.ParsedEvent)
	if sess == nil || sess.persistsRuntimeState() {
		persistFn = persistSessionOnInit(workDir)
	}
	recordCheckpoints := sess == nil || !sess.suppressCheckpoints
	return func(ev drivercontract.ParsedEvent) {
		if persistFn != nil {
			persistFn(ev)
		}

		agentID := strings.TrimSpace(eventAgentID)
		if agentID == "" {
			agentID = sessionMapKey
		}
		if agentID == "" {
			agentID = ev.SessionID
		}

		runtimeEvent, ok := runtimeEventFromParsedEvent(channelID, agentID, ev)
		if !ok {
			return
		}
		activity, detail := runtimeEventActivity(runtimeEvent)

		switch ev.Kind {
		case consts.ParsedEventKindTurnEnd:
			if recordCheckpoints {
				if ap, ok := d.findProcess(sessionMapKey); ok && ap.WorkDir() != "" {
					ap.RecordCheckpoint("flush", "Waiting for the next message after the latest completed turn.")
				}
			}
		case consts.ParsedEventKindRuntimeIdle:
			if recordCheckpoints {
				if ap, ok := d.findProcess(sessionMapKey); ok && ap.WorkDir() != "" {
					ap.RecordCheckpoint("flush", "Waiting for the next message after codex app-server returned to idle.")
				}
			}
		case consts.ParsedEventKindRuntimeCompacting:
			if recordCheckpoints {
				if ap, ok := d.findProcess(sessionMapKey); ok && ap.WorkDir() != "" {
					ap.RecordCheckpoint("compact", "Compacting context after the latest completed turn.")
				}
			}
		case consts.ParsedEventKindRuntimeCompacted:
			if recordCheckpoints {
				if ap, ok := d.findProcess(sessionMapKey); ok && ap.WorkDir() != "" {
					ap.RecordCheckpoint("compact", firstNonEmpty(detail, "Context compacted."))
				}
			}
		}

		// Update session activity for dashboard polling.
		if activity != "" {
			changed := false
			d.sessionManager.WithSessionByKey(sessionMapKey, func(sess *interactiveSession) {
				sess.mu.Lock()
				changed = sess.activity != activity || sess.activityDetail != detail
				sess.activity = activity
				sess.activityDetail = detail
				sess.mu.Unlock()
			})
			if changed {
				d.publishSessionStateChanged(agentID)
			}
		}

		if reportRuntimeEvents {
			d.publishRuntimeEvent(runtimeEvent)
		}
	}
}

func classifyEventTool(toolName string) string {
	name := toolName
	for _, prefix := range []string{"mcp__agentteam__", "mcp_agentteam_"} {
		if strings.HasPrefix(name, prefix) {
			name = strings.TrimPrefix(name, prefix)
			break
		}
	}
	if name == toolName {
		return ""
	}

	switch name {
	case "message_send":
		return "send"
	}
	return ""
}

func toolDisplayName(toolName string) string {
	name := toolName
	for _, prefix := range []string{"mcp__agentteam__", "mcp_agentteam_"} {
		if strings.HasPrefix(name, prefix) {
			name = strings.TrimPrefix(name, prefix)
			break
		}
	}

	switch name {
	case "message_attachment_view":
		return "Viewing file…"
	case "message_history_read":
		return "Reading history…"
	case "task_list":
		return "Listing tasks…"
	case "task_get":
		return "Reading task…"
	case "task_create":
		return "Creating tasks…"
	case "task_update":
		return "Updating task…"
	case "task_checkpoint":
		return "Adding task checkpoint…"
	}

	if name != toolName {
		return ""
	}

	if strings.HasPrefix(toolName, "collaboration.") {
		switch strings.TrimPrefix(toolName, "collaboration.") {
		case "spawn_agent":
			return "Spawning an agent…"
		case "send_input":
			return "Sending input to an agent…"
		case "resume_agent":
			return "Resuming an agent…"
		case "wait":
			return "Waiting for an agent…"
		case "close_agent":
			return "Closing an agent…"
		case "agent_tool":
			return "Coordinating agents…"
		}
	}

	switch toolName {
	case "Read", "read_file", "file_read":
		return "Reading file…"
	case "view":
		return "Viewing file…"
	case "LS", "ls", "list_files", "list_directory", "list_dir":
		return "Listing files…"
	case "Write", "write", "write_file", "file_write":
		return "Writing file…"
	case "Edit", "edit", "MultiEdit", "multiedit", "edit_file", "file_change":
		return "Editing file…"
	case "Bash", "bash", "shell", "command_execution", "exec_command":
		return "Running command…"
	case "write_stdin", "job_output":
		return "Waiting for command output…"
	case "Glob", "glob":
		return "Searching files…"
	case "Grep", "grep":
		return "Searching content…"
	case "TodoWrite", "todos", "todo_write":
		return "Updating TODOs…"
	case "download":
		return "Downloading file…"
	case "fetch", "web_fetch", "agentic_fetch":
		return "Fetching URL…"
	case "web_search":
		return "Searching web…"
	case "sourcegraph":
		return "Searching content…"
	case "job_kill":
		return "Stopping command…"
	case "list_mcp_resources":
		return "Listing MCP resources…"
	case "read_mcp_resource":
		return "Reading MCP resource…"
	case "lsp_diagnostics":
		return "Reading diagnostics…"
	case "lsp_references":
		return "Finding references…"
	case "lsp_restart":
		return "Restarting LSP…"
	}
	return ""
}

// RestoreAgentRecoveryPoint restores a previously captured recovery point for
// the agent identified by serverAgentID. The requestID is included for logging
// and correlation purposes.
func (d *Dispatcher) RestoreAgentRecoveryPoint(channelID, serverAgentID, requestID, runtime, model, sessionID, strategy, recoveryRef string) error {
	capability, err := agent.ValidateRecoveryCapability(runtime, strategy)
	if err != nil {
		return err
	}

	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()

	localAgentID := recoveryLocalAgentID(serverAgentID)
	defaultSessionKey := sessionKeyForBinding(localAgentID, conversationBinding{}).String()
	if strings.TrimSpace(defaultSessionKey) == "" {
		return fmt.Errorf("no active session found for server agent %s", serverAgentID)
	}

	// Target selection and recovery reservation are one SessionManager
	// operation. The reservation occupies the same generation slot as every
	// Process start, so delivery/config-restart cannot publish a replacement
	// between the snapshot below and live/detached recovery finalization.
	var (
		targetSession       *interactiveSession
		recoveryReservation *sessionStartReservation
	)
	for {
		var waitCh <-chan struct{}
		var acquired bool
		targetSession, recoveryReservation, waitCh, acquired = d.sessionManager.BeginRecovery(
			defaultSessionKey,
			func(sess *interactiveSession) bool {
				return recoverySessionMatchesAgent(sess, localAgentID, serverAgentID)
			},
		)
		if acquired {
			break
		}
		if d.sessionManager.Closed() {
			return errSessionManagerClosed
		}
		if err := waitForSessionStart(ctx, waitCh); err != nil {
			return err
		}
	}
	defer d.sessionManager.FinishStart(recoveryReservation)
	defaultSessionKey = recoveryReservation.key

	// A production agent has one default runtime across AgentTeam scopes, but a
	// delayed recovery request must not rewind that runtime after it has moved to
	// a different server session. Validate the resolved target under the same
	// recovery reservation used for restore; treating a mismatch as "detached"
	// would mutate the shared workdir underneath the live Process.
	if targetSession != nil && !recoverySessionMatchesAgentTeamContext(targetSession, channelID, serverAgentID, sessionID) {
		return fmt.Errorf("active session for server agent %s does not match recovery channel/session", serverAgentID)
	}

	workDir := sessionWorkDirForKey(localAgentID, defaultSessionKey, false)
	var proc *agent.Process
	if targetSession != nil {
		targetSession.mu.Lock()
		if id := strings.TrimSpace(targetSession.agentID); id != "" {
			localAgentID = id
		}
		proc = targetSession.process
		if existingWorkDir := strings.TrimSpace(targetSession.workDir); proc != nil && existingWorkDir != "" {
			workDir = existingWorkDir
		}
		targetSession.mu.Unlock()
	}
	if proc == nil {
		// A sleeping legacy session can still carry a conversation-hashed
		// workdir. Detached recovery must use the same resolved agent root as a
		// production start after daemon restart, otherwise rollback and the next
		// wake would mutate two different state journals.
		workDir = sessionWorkDirForKey(localAgentID, defaultSessionKey, false)
	}
	if strings.TrimSpace(workDir) == "" {
		return fmt.Errorf("no active session found for server agent %s", serverAgentID)
	}

	detachedCfg, err := d.detachedRecoveryConfigForAgentTeamAgent(localAgentID, capability.Runtime, sessionID, workDir, model)
	if err != nil {
		return err
	}

	if proc == nil {
		if err := requireNoLiveRuntimeDeliveryOwnershipForRecovery(workDir); err != nil {
			return err
		}
		err := d.restoreRecoveryPointDetached(ctx, detachedCfg, capability.Strategy, recoveryRef)
		if err == nil {
			err = d.discardCodexResumeAfterRecovery(nil, workDir, capability.Runtime, false)
		}
		return err
	}
	if !proc.IsIdle() {
		handled, interruptErr := proc.InterruptActiveRun(ctx)
		if interruptErr != nil {
			return interruptErr
		}
		if !handled || !proc.IsIdle() {
			return agent.ErrAgentBusy
		}
		d.publishSessionStateChanged(localAgentID)
	}

	if proc.SupportsLiveRecovery(capability.Strategy) {
		if runtimeconfig.RuntimeType(strings.TrimSpace(capability.Runtime)) == runtimeconfig.TypeCodex {
			// Keep the Process delivery gate from native rollback through durable
			// resume cleanup and retirement. A wake is therefore ordered wholly
			// before rollback (which rejects as busy) or wholly after the old Process
			// becomes non-accepting; it can never be acknowledged in the gap.
			err = proc.RestoreRecoveryPointAndFinalize(ctx, capability.Strategy, recoveryRef, func() error {
				return d.discardCodexResumeAfterPreparedRecovery(targetSession, workDir, capability.Runtime)
			})
		} else {
			err = proc.RestoreRecoveryPoint(ctx, capability.Strategy, recoveryRef)
		}
		if err == nil {
			return nil
		}
		if !errors.Is(err, agent.ErrAgentBusy) {
			return err
		}
		return err
	}

	if err := proc.SleepForRecovery(ctx); err != nil {
		return err
	}

	if !proc.WaitForExit(5 * time.Second) {
		return fmt.Errorf("runtime %q did not stop before detached recovery", capability.Runtime)
	}

	err = d.restoreRecoveryPointDetached(ctx, detachedCfg, capability.Strategy, recoveryRef)
	if err == nil {
		err = d.retireSessionAfterDetachedRecovery(targetSession, workDir, capability.Runtime, false)
	}
	return err
}

// retireSessionAfterDetachedRecovery finalizes a successful detached restore.
// The Process was already stopped via SleepForRecovery + WaitForExit, so the
// stopped generation must be retired/detached under the held recovery
// reservation so the immediately following wake can create/resume a replacement
// process from the restored session. Codex additionally clears its resumable
// app-server thread state; other runtimes (Pi, Kimi, Claude) clear only the
// stopped session pointer and preserve the shared ownership ledger. A failure
// leaves the prior state fail-closed (the stopped Process stays retired).
func (d *Dispatcher) retireSessionAfterDetachedRecovery(sess *interactiveSession, workDir string, runtime string, stopPrepared bool) error {
	if runtimeconfig.RuntimeType(strings.TrimSpace(runtime)) == runtimeconfig.TypeCodex {
		return d.discardCodexResumeAfterRecovery(sess, workDir, runtime, stopPrepared)
	}
	// Non-Codex detached runtimes: detach the stopped session so the next wake
	// can start a replacement. detachSessionForPreparedRestart CAS-retires the
	// session and generation-fenced clears the stopped Process pointer without
	// taking a new start reservation or calling public Process.Stop, which is
	// safe under the held recovery reservation.
	if sess != nil {
		d.detachSessionForPreparedRestart(sess)
	}
	return nil
}

func (d *Dispatcher) restoreRecoveryPointDetached(
	ctx context.Context,
	cfg agent.DetachedRecoveryConfig,
	strategy string,
	recoveryRef string,
) error {
	if d != nil && d.recoveryPointDetachedRestorer != nil {
		return d.recoveryPointDetachedRestorer(ctx, cfg, strategy, recoveryRef)
	}
	return agent.RestoreRecoveryPointDetached(ctx, cfg, strategy, recoveryRef)
}

// discardCodexResumeAfterPreparedRecovery is used only after Process has won
// the stop transition while still holding deliveryGate. The dispatcher detaches
// the session but leaves termination/completion to that existing owner.
func (d *Dispatcher) discardCodexResumeAfterPreparedRecovery(sess *interactiveSession, workDir string, runtime string) error {
	return d.discardCodexResumeAfterRecovery(sess, workDir, runtime, true)
}

func (d *Dispatcher) discardCodexResumeAfterRecovery(sess *interactiveSession, workDir string, runtime string, stopPrepared bool) error {
	if runtimeconfig.RuntimeType(strings.TrimSpace(runtime)) != runtimeconfig.TypeCodex {
		return nil
	}
	if strings.TrimSpace(workDir) != "" {
		if err := updateSessionResumeState(workDir, func(state *sessionResumeState) error {
			// Native rollback invalidates only the resumable app-server thread. Keep
			// the shared state ledger intact: accepted runtime-delivery ownership,
			// deferred messages, active task, and runtime policy all belong to the
			// daemon rather than to the reverted Codex thread.
			state.LastSessionID = ""
			state.SessionResumeSafe = false
			state.SessionResumeSafetyEpoch++
			state.SessionResumeSafetyVersion = 0
			if state.LastStartConfig != nil {
				state.LastStartConfig = cloneAgentStartConfig(state.LastStartConfig)
				state.LastStartConfig.SessionID = ""
			}
			return nil
		}); err != nil {
			return fmt.Errorf("clear codex resume state after recovery: %w", err)
		}
	}
	if sess == nil {
		return nil
	}
	// codex-app-server rollback may retain reverted turns as internal
	// interrupted history. Stop and retire the local process after native restore
	// so the next wake starts from AgentTeam's rewound visible transcript instead of
	// reusing a contaminated app-server thread.
	if stopPrepared {
		d.detachSessionForPreparedRestart(sess)
	} else {
		d.interruptSessionForRestart(sess)
	}
	return nil
}

func (d *Dispatcher) detachedRecoveryConfigForAgentTeamAgent(agentID, runtime, sessionID, workDir, model string) (agent.DetachedRecoveryConfig, error) {
	cfg := agent.DetachedRecoveryConfig{
		AgentID:         strings.TrimSpace(agentID),
		Runtime:         strings.TrimSpace(runtime),
		SessionID:       strings.TrimSpace(sessionID),
		WorkDir:         strings.TrimSpace(workDir),
		RuntimeStateDir: agentRuntimeStateDir(agentID),
		Model:           strings.TrimSpace(model),
	}
	if rt, ok := profile.Default().Runtimes().GetOrRefresh(cfg.Runtime); ok {
		cfg.Binary = strings.TrimSpace(rt.Binary)
		cfg.Env = runtimeconfig.CloneStringMap(rt.Env)
	}
	agentConfig, ok := profile.Default().Agents().Get(cfg.AgentID)
	if !ok {
		return cfg, nil
	}
	cfg.PermissionMode = types.NormalizeAgentPermissionMode(agentConfig.PermissionMode)
	cfg.RuntimeDebugEnabled = agentConfig.RuntimeDebugEnabled
	if strings.TrimSpace(cfg.Model) == "" {
		cfg.Model = strings.TrimSpace(agentConfig.Model)
	}

	llmResolution, err := profile.ResolveAgentLLM(agentConfig, runtimeconfig.RuntimeType(cfg.Runtime))
	if err != nil {
		return agent.DetachedRecoveryConfig{}, fmt.Errorf("resolve detached recovery provider for agent %q: %w", cfg.AgentID, err)
	}
	if llmResolution.Upstream == nil {
		return cfg, nil
	}

	cfg.Provider = &drivercontract.ProviderConfig{
		ID:             llmResolution.Upstream.ID,
		Name:           llmResolution.Upstream.Name,
		BaseURL:        strings.TrimSpace(llmResolution.Upstream.BaseURL),
		APIKey:         strings.TrimSpace(llmResolution.Upstream.APIKey),
		CredentialKind: strings.TrimSpace(agentConfig.CredentialKind),
		EnvKey:         strings.TrimSpace(llmResolution.Upstream.EnvKey),
		Headers:        llmResolution.Upstream.Headers,
		WireAPIs:       append([]string(nil), llmResolution.Upstream.WireAPIs...),
	}
	return cfg, nil
}

func restoreSessionKeyForAgentTeamContext(channelID, serverAgentID, sessionID string) string {
	trimmedAgentID := strings.TrimSpace(serverAgentID)
	if trimmedAgentID == "" {
		return ""
	}
	// Production interactive sessions are agent-scoped. Recovery must derive
	// exactly the same key so daemon restart cannot create a second hashed state
	// journal for the same runtime thread.
	return sessionKeyForBinding(trimmedAgentID, conversationBinding{}).String()
}

func recoveryLocalAgentID(serverAgentID string) string {
	serverAgentID = strings.TrimSpace(serverAgentID)
	if serverAgentID == "" {
		return ""
	}
	if configured, ok := profile.Default().Agents().Get(serverAgentID); ok {
		if localAgentID := strings.TrimSpace(configured.ID); localAgentID != "" {
			return localAgentID
		}
	}
	return serverAgentID
}

func recoverySessionMatchesAgentTeamContext(sess *interactiveSession, channelID, serverAgentID, sessionID string) bool {
	if sess == nil || sess.ephemeral {
		return false
	}
	trimmedAgentID := strings.TrimSpace(serverAgentID)
	trimmedServerID := strings.TrimSpace(channelID)
	trimmedSessionID := strings.TrimSpace(sessionID)
	sess.mu.Lock()
	defer sess.mu.Unlock()
	if strings.TrimSpace(sess.serverAgentID) != trimmedAgentID {
		return false
	}
	if trimmedServerID != "" && strings.TrimSpace(sess.binding.channelID) != trimmedServerID {
		return false
	}
	if trimmedSessionID != "" && strings.TrimSpace(sess.binding.serverSessionID) != trimmedSessionID {
		return false
	}
	return true
}

func recoverySessionMatchesAgent(sess *interactiveSession, localAgentID, serverAgentID string) bool {
	if sess == nil || sess.ephemeral {
		return false
	}
	trimmedLocalID := strings.TrimSpace(localAgentID)
	trimmedServerID := strings.TrimSpace(serverAgentID)
	sess.mu.Lock()
	defer sess.mu.Unlock()
	return (trimmedLocalID != "" && (strings.TrimSpace(sess.agentID) == trimmedLocalID || strings.TrimSpace(sess.agent.ID) == trimmedLocalID)) ||
		(trimmedServerID != "" && strings.TrimSpace(sess.serverAgentID) == trimmedServerID)
}

func requireNoLiveRuntimeDeliveryOwnershipForRecovery(workDir string) error {
	state, err := loadSessionResumeState(workDir)
	if err != nil {
		return fmt.Errorf("load runtime delivery ownership before detached recovery: %w", err)
	}
	for _, record := range state.RuntimeDeliveryOwnership {
		if runtimeDeliveryOwnershipIsLive(record) {
			return agent.ErrAgentBusy
		}
	}
	return nil
}

// GCRecoveryPoints garbage-collects stale recovery artifacts for the agent
// identified by serverAgentID.
func (d *Dispatcher) GCRecoveryPoints(serverAgentID string, requests []types.RecoveryArtifactRequest) error {
	if len(requests) == 0 {
		return nil
	}

	targetSession, ok := d.sessionManager.FirstByServerAgentID(serverAgentID)
	if !ok {
		return nil
	}

	targetSession.mu.Lock()
	localAgentID := strings.TrimSpace(targetSession.agentID)
	if localAgentID == "" {
		localAgentID = strings.TrimSpace(targetSession.agent.ID)
	}
	workDir := targetSession.workDir
	runtimeName := string(targetSession.runtime.Type)
	targetSession.mu.Unlock()

	if strings.TrimSpace(workDir) == "" || localAgentID == "" {
		return nil
	}

	artifacts := make([]agent.RecoveryArtifact, 0, len(requests))
	for _, r := range requests {
		if strings.TrimSpace(r.Strategy) == "" || strings.TrimSpace(r.RecoveryRef) == "" {
			continue
		}
		artifacts = append(artifacts, agent.RecoveryArtifact{
			Strategy:    r.Strategy,
			RecoveryRef: r.RecoveryRef,
		})
	}

	if len(artifacts) == 0 {
		return nil
	}

	cfg, err := d.detachedRecoveryConfigForAgentTeamAgent(localAgentID, runtimeName, "", workDir, "")
	if err != nil {
		return err
	}

	cleaner := agent.CleanupRecoveryArtifactsDetached
	if d.recoveryArtifactsDetachedCleaner != nil {
		cleaner = d.recoveryArtifactsDetachedCleaner
	}
	return cleaner(context.Background(), cfg, artifacts)
}