package dispatcher

import (
	"context"
	"io"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"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/profile"
	"github.com/tidbcloud/AgentTeam/desktop/internal/types"
	"github.com/tidbcloud/AgentTeam/internal/core/rewindcaps"
)

func TestDetachedRecoveryConfigForAgentTeamAgentPopulatesProvider(t *testing.T) {
	configDir := t.TempDir()
	t.Setenv(profile.ConfigDirEnvKey, configDir)

	mgr := profile.Default()
	require.NoError(t, mgr.Agents().Add(agent.Config{
		ID:      "server-agent",
		Name:    "Server Agent",
		Runtime: "codex",
		Model:   "agent-default-model",
		BaseURL: "https://provider.example.com/v1",
		APIKey:  "sk-agent",
		WireAPI: "responses",
	}))

	d := mustNewDispatcher("http://server.test")
	cfg, err := d.detachedRecoveryConfigForAgentTeamAgent("server-agent", "codex", "sess_1", "/tmp/workdir", "gpt-5")
	require.NoError(t, err)
	require.NotNil(t, cfg.Provider)
	require.Equal(t, "server-agent", cfg.Provider.ID)
	require.Equal(t, "Server Agent", cfg.Provider.Name)
	require.Equal(t, "https://provider.example.com/v1", cfg.Provider.BaseURL)
	require.Equal(t, "sk-agent", cfg.Provider.APIKey)
	require.Equal(t, []string{"responses"}, cfg.Provider.WireAPIs)
	require.Equal(t, filepath.Join(profile.AgentWorkDir("server-agent"), ".runtime"), cfg.RuntimeStateDir)
	require.NotEqual(t, cfg.WorkDir, cfg.RuntimeStateDir)

	cfg, err = d.detachedRecoveryConfigForAgentTeamAgent("server-agent", "codex", "sess_1", "/tmp/workdir", "")
	require.NoError(t, err)
	require.Equal(t, "agent-default-model", cfg.Model)
	require.NotNil(t, cfg.Provider)
}

func TestDetachedRecoveryConfigForAgentTeamAgentMissingAgentLeavesProviderEmpty(t *testing.T) {
	configDir := t.TempDir()
	t.Setenv(profile.ConfigDirEnvKey, configDir)

	d := mustNewDispatcher("http://server.test")
	cfg, err := d.detachedRecoveryConfigForAgentTeamAgent("missing-agent", "codex", "sess_1", "/tmp/workdir", "gpt-5")
	require.NoError(t, err)
	require.Nil(t, cfg.Provider)
}

func TestGCRecoveryPointsUsesTargetSessionLocalAgentID(t *testing.T) {
	configDir := t.TempDir()
	t.Setenv(profile.ConfigDirEnvKey, configDir)

	const (
		localAgentID  = "local-agent"
		serverAgentID = "server-agent"
	)
	workDir := sessionWorkDir(localAgentID)
	d := mustNewDispatcher("http://server.test")
	d.sessionManager.Set(sessionKeyForBinding(localAgentID, conversationBinding{}).String(), &interactiveSession{
		agentID:       localAgentID,
		serverAgentID: serverAgentID,
		runtime:       runtimeconfig.Runtime{Type: runtimeconfig.TypeBuiltin},
		workDir:       workDir,
	})

	var (
		cleanupCfg       agent.DetachedRecoveryConfig
		cleanupArtifacts []agent.RecoveryArtifact
	)
	d.recoveryArtifactsDetachedCleaner = func(_ context.Context, cfg agent.DetachedRecoveryConfig, artifacts []agent.RecoveryArtifact) error {
		cleanupCfg = cfg
		cleanupArtifacts = append([]agent.RecoveryArtifact(nil), artifacts...)
		return nil
	}

	err := d.GCRecoveryPoints(serverAgentID, []types.RecoveryArtifactRequest{{
		Strategy:    rewindcaps.StrategyOpenCodeSessionRevert,
		RecoveryRef: "recovery-ref",
	}})

	require.NoError(t, err)
	require.Equal(t, localAgentID, cleanupCfg.AgentID)
	require.Equal(t, workDir, cleanupCfg.WorkDir)
	require.Equal(t, agentRuntimeStateDir(localAgentID), cleanupCfg.RuntimeStateDir)
	require.NotEqual(t, agentRuntimeStateDir(serverAgentID), cleanupCfg.RuntimeStateDir)
	require.Equal(t, []agent.RecoveryArtifact{{
		Strategy:    rewindcaps.StrategyOpenCodeSessionRevert,
		RecoveryRef: "recovery-ref",
	}}, cleanupArtifacts)
}

func TestRestoreAgentRecoveryPointUsesDetachedRecoveryWhenSessionHasNoProcess(t *testing.T) {
	configDir := t.TempDir()
	t.Setenv(profile.ConfigDirEnvKey, configDir)

	legacyWorkDir := filepath.Join(configDir, "agents", "server-agent", "sessions", "sleeping-session")
	workDir := sessionWorkDir("server-agent")
	sessionKey := restoreSessionKeyForAgentTeamContext("agentteam", "server-agent", "sess_1")
	d := mustNewDispatcher("http://server.test")
	d.sessionManager.Set(sessionKey, &interactiveSession{
		sessionKey:    sessionKey,
		agentID:       "server-agent",
		serverAgentID: "server-agent",
		binding: conversationBinding{
			channelID:       "agentteam",
			serverAgentID:   "server-agent",
			serverSessionID: "sess_1",
		},
		runtime: runtimeconfig.Runtime{Type: runtimeconfig.TypeCodex},
		workDir: legacyWorkDir,
		process: nil,
	})

	err := d.RestoreAgentRecoveryPoint(
		"agentteam",
		"server-agent",
		"req_1",
		"codex",
		"",
		"sess_1",
		rewindcaps.StrategyCodexAppServerTurns,
		"not-json",
	)

	require.Error(t, err)
	require.Contains(t, err.Error(), "decode codex recovery ref")
	require.NotContains(t, err.Error(), "no running process")
	require.DirExists(t, workDir)
	require.NoDirExists(t, legacyWorkDir)
}

func TestRestoreAgentRecoveryPointUsesMatchingAgentTeamSessionWhenSessionIDMissing(t *testing.T) {
	configDir := t.TempDir()
	t.Setenv(profile.ConfigDirEnvKey, configDir)

	d := mustNewDispatcher("http://server.test")
	workDir := sessionWorkDir("server-agent")
	d.sessionManager.Set("active-session-key", &interactiveSession{
		sessionKey:    "active-session-key",
		agentID:       "server-agent",
		serverAgentID: "server-agent",
		binding: conversationBinding{
			channelID:       "agentteam",
			serverAgentID:   "server-agent",
			serverSessionID: "runtime-session-a",
		},
		runtime: runtimeconfig.Runtime{Type: runtimeconfig.TypeCodex},
		workDir: workDir,
		process: nil,
	})

	err := d.RestoreAgentRecoveryPoint(
		"agentteam",
		"server-agent",
		"req_1",
		"codex",
		"",
		"",
		rewindcaps.StrategyCodexAppServerTurns,
		"not-json",
	)

	require.Error(t, err)
	require.Contains(t, err.Error(), "decode codex recovery ref")
	require.NotContains(t, err.Error(), "no active session")
	require.DirExists(t, workDir)
}

func TestRestoreAgentRecoveryPointUsesMatchingAgentTeamSession(t *testing.T) {
	configDir := t.TempDir()
	t.Setenv(profile.ConfigDirEnvKey, configDir)

	d := mustNewDispatcher("http://server.test")
	keyB := restoreSessionKeyForAgentTeamContext("agentteam", "server-agent", "sess_b")
	workDirA := filepath.Join(configDir, "agents", "unrelated-agent")
	workDirB := sessionWorkDir("server-agent")
	for _, session := range []struct {
		key       string
		sessionID string
		workDir   string
	}{
		{key: "legacy-unrelated-session", sessionID: "sess_a", workDir: workDirA},
		{key: keyB, sessionID: "sess_b", workDir: workDirB},
	} {
		d.sessionManager.Set(session.key, &interactiveSession{
			sessionKey:    session.key,
			agentID:       "server-agent",
			serverAgentID: "server-agent",
			binding: conversationBinding{
				channelID:       "agentteam",
				serverAgentID:   "server-agent",
				serverSessionID: session.sessionID,
			},
			runtime: runtimeconfig.Runtime{Type: runtimeconfig.TypeCodex},
			workDir: session.workDir,
			process: nil,
		})
	}

	err := d.RestoreAgentRecoveryPoint(
		"agentteam",
		"server-agent",
		"req_1",
		"codex",
		"",
		"sess_b",
		rewindcaps.StrategyCodexAppServerTurns,
		"not-json",
	)

	require.Error(t, err)
	require.Contains(t, err.Error(), "decode codex recovery ref")
	require.DirExists(t, workDirB)
	_, statErr := os.Stat(workDirA)
	require.True(t, os.IsNotExist(statErr), "unexpected restore workdir %s exists", workDirA)
}

func TestRestoreAgentRecoveryPointRejectsMismatchedActiveAgentTeamSession(t *testing.T) {
	configDir := t.TempDir()
	t.Setenv(profile.ConfigDirEnvKey, configDir)

	d := mustNewDispatcher("http://server.test")
	key := restoreSessionKeyForAgentTeamContext("agentteam", "server-agent", "sess_current")
	d.sessionManager.Set(key, &interactiveSession{
		sessionKey:    key,
		agentID:       "server-agent",
		serverAgentID: "server-agent",
		binding: conversationBinding{
			channelID:       "agentteam",
			serverAgentID:   "server-agent",
			serverSessionID: "sess_current",
		},
		workDir: sessionWorkDir("server-agent"),
	})
	detachedCalled := false
	d.recoveryPointDetachedRestorer = func(context.Context, agent.DetachedRecoveryConfig, string, string) error {
		detachedCalled = true
		return nil
	}

	err := d.RestoreAgentRecoveryPoint(
		"agentteam", "server-agent", "req_stale", "codex", "", "sess_stale",
		rewindcaps.StrategyCodexAppServerTurns, "recovery-ref",
	)

	require.ErrorContains(t, err, "does not match recovery channel/session")
	require.False(t, detachedCalled, "a stale recovery request must not fall back to detached restore under the active session")
}

type activeRestoreCodexDriver struct {
	done       chan struct{}
	doneOnce   sync.Once
	name       string
	strategy   string
	interrupts []string
	restores   []string
}

type blockingRestoreRuntimeDriver struct {
	*activeRestoreCodexDriver
	restoreStarted     chan struct{}
	restoreRelease     chan struct{}
	restoreStartedOnce sync.Once
	restoreReleaseOnce sync.Once
}

func newBlockingRestoreRuntimeDriver(name, strategy string) *blockingRestoreRuntimeDriver {
	return &blockingRestoreRuntimeDriver{
		activeRestoreCodexDriver: newActiveRestoreRuntimeDriver(name, strategy),
		restoreStarted:           make(chan struct{}),
		restoreRelease:           make(chan struct{}),
	}
}

func (driver *blockingRestoreRuntimeDriver) RestoreRecoveryPoint(ctx context.Context, _ string, recoveryRef string) error {
	driver.restores = append(driver.restores, recoveryRef)
	driver.restoreStartedOnce.Do(func() { close(driver.restoreStarted) })
	select {
	case <-driver.restoreRelease:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

func (driver *blockingRestoreRuntimeDriver) releaseRestore() {
	driver.restoreReleaseOnce.Do(func() { close(driver.restoreRelease) })
}

func newActiveRestoreCodexDriver() *activeRestoreCodexDriver {
	return &activeRestoreCodexDriver{done: make(chan struct{}), name: "codex", strategy: rewindcaps.StrategyCodexAppServerTurns}
}

func newActiveRestoreRuntimeDriver(name string, strategy string) *activeRestoreCodexDriver {
	return &activeRestoreCodexDriver{done: make(chan struct{}), name: name, strategy: strategy}
}

func (driver *activeRestoreCodexDriver) Name() string { return driver.name }

func (driver *activeRestoreCodexDriver) SupportsStdinNotification() bool { return true }

func (driver *activeRestoreCodexDriver) MCPToolPrefix() string { return "mcp__agentteam__" }

func (driver *activeRestoreCodexDriver) Spawn(context.Context, drivercontract.LaunchConfig) (drivercontract.SpawnResult, error) {
	return drivercontract.SpawnResult{
		Stdout: io.NopCloser(strings.NewReader("")),
		Wait: func() error {
			<-driver.done
			return nil
		},
		Terminate: func() {
			driver.doneOnce.Do(func() { close(driver.done) })
		},
	}, nil
}

func (driver *activeRestoreCodexDriver) ParseLine(string) []drivercontract.ParsedEvent { return nil }

func (driver *activeRestoreCodexDriver) EncodeStdinMessage(string, string) []byte { return nil }

func (driver *activeRestoreCodexDriver) PrepareRecoveryPoint() (string, string, bool, error) {
	return driver.strategy, "recovery-ref", true, nil
}

func (driver *activeRestoreCodexDriver) RestoreRecoveryPoint(_ context.Context, _ string, recoveryRef string) error {
	driver.restores = append(driver.restores, recoveryRef)
	return nil
}

func (driver *activeRestoreCodexDriver) InterruptActiveRun(_ context.Context, sessionID string) error {
	driver.interrupts = append(driver.interrupts, sessionID)
	return nil
}

func TestRestoreAgentRecoveryPointInterruptsBusyCodexSessionBeforeLiveRestore(t *testing.T) {
	configDir := t.TempDir()
	t.Setenv(profile.ConfigDirEnvKey, configDir)

	driver := newActiveRestoreCodexDriver()
	workDir := filepath.Join(configDir, "agents", "server-agent", "sessions", "active-session")
	require.NoError(t, os.MkdirAll(workDir, 0o755))
	proc, err := agent.NewProcess(
		context.Background(),
		driver,
		testLaunchConfig(drivercontract.LaunchConfig{AgentID: "agent-local", RuntimeType: runtimeconfig.TypeCodex, SessionID: "sess_1", WorkDir: workDir}, testLaunchSurfaces{Prompt: "active turn"}),
		agent.ProcessOptions{},
	)
	require.NoError(t, err)
	proc.SessionID = "sess_1"
	// Production Codex processes receive this durable callback from
	// Dispatcher.launch before publication. This test constructs Process
	// directly, so install an accepting persistence seam before exercising the
	// new fail-closed interrupt/restore prerequisites.
	require.NoError(t, proc.SetOnResumeSafety(func(uint64, bool, string) error { return nil }))
	t.Cleanup(proc.Sleep)
	require.False(t, proc.IsIdle())

	d := mustNewDispatcher("http://server.test")
	sessionKey := restoreSessionKeyForAgentTeamContext("agentteam", "server-agent", "sess_1")
	d.sessionManager.Set(sessionKey, &interactiveSession{
		sessionKey:    sessionKey,
		agentID:       "agent-local",
		serverAgentID: "server-agent",
		binding: conversationBinding{
			channelID:       "agentteam",
			serverAgentID:   "server-agent",
			serverSessionID: "sess_1",
		},
		runtime: runtimeconfig.Runtime{Type: runtimeconfig.TypeCodex},
		workDir: workDir,
		process: proc,
	})

	err = d.RestoreAgentRecoveryPoint(
		"agentteam",
		"server-agent",
		"req_1",
		"codex",
		"",
		"sess_1",
		rewindcaps.StrategyCodexAppServerTurns,
		"recovery-ref",
	)

	require.NoError(t, err)
	require.Equal(t, []string{"sess_1"}, driver.interrupts)
	require.Equal(t, []string{"recovery-ref"}, driver.restores)
}

func TestRestoreAgentRecoveryPointInterruptsBusyOpenCodeSessionBeforeLiveRestore(t *testing.T) {
	configDir := t.TempDir()
	t.Setenv(profile.ConfigDirEnvKey, configDir)

	driver := newActiveRestoreRuntimeDriver("opencode", rewindcaps.StrategyOpenCodeSessionRevert)
	workDir := filepath.Join(configDir, "agents", "server-agent", "sessions", "active-opencode")
	proc, err := agent.NewProcess(
		context.Background(),
		driver,
		testLaunchConfig(drivercontract.LaunchConfig{AgentID: "agent-local", RuntimeType: runtimeconfig.TypeOpenCode, SessionID: "sess_1", WorkDir: workDir}, testLaunchSurfaces{Prompt: "active turn"}),
		agent.ProcessOptions{},
	)
	require.NoError(t, err)
	proc.SessionID = "sess_1"
	t.Cleanup(proc.Sleep)
	require.False(t, proc.IsIdle())

	d := mustNewDispatcher("http://server.test")
	sessionKey := restoreSessionKeyForAgentTeamContext("agentteam", "server-agent", "sess_1")
	d.sessionManager.Set(sessionKey, &interactiveSession{
		sessionKey:    sessionKey,
		agentID:       "agent-local",
		serverAgentID: "server-agent",
		binding: conversationBinding{
			channelID:       "agentteam",
			serverAgentID:   "server-agent",
			serverSessionID: "sess_1",
		},
		runtime: runtimeconfig.Runtime{Type: runtimeconfig.TypeOpenCode},
		workDir: workDir,
		process: proc,
	})

	err = d.RestoreAgentRecoveryPoint(
		"agentteam", "server-agent", "req_1", "opencode", "", "sess_1",
		rewindcaps.StrategyOpenCodeSessionRevert, "recovery-ref",
	)

	require.NoError(t, err)
	require.Equal(t, []string{"sess_1"}, driver.interrupts)
	require.Equal(t, []string{"recovery-ref"}, driver.restores)
}

func TestRestoreAgentRecoveryPointLiveRestoreFencesConfigRestartOnResolvedLocalKey(t *testing.T) {
	configDir := t.TempDir()
	t.Setenv(profile.ConfigDirEnvKey, configDir)

	driver := newBlockingRestoreRuntimeDriver("opencode", rewindcaps.StrategyOpenCodeSessionRevert)
	workDir := sessionWorkDir("agent-local")
	proc, err := agent.NewProcess(
		context.Background(),
		driver,
		drivercontract.LaunchConfig{
			AgentID:     "agent-local",
			RuntimeType: runtimeconfig.TypeOpenCode,
			SessionID:   "sess_1",
			WorkDir:     workDir,
		},
		agent.ProcessOptions{},
	)
	require.NoError(t, err)
	proc.SessionID = "sess_1"
	t.Cleanup(func() {
		driver.releaseRestore()
		proc.Sleep()
	})
	require.True(t, proc.IsIdle())

	liveAgent := agent.Config{ID: "agent-local", Runtime: "opencode", Model: "model-before"}
	liveRuntime := runtimeconfig.Runtime{Name: "opencode", Type: runtimeconfig.TypeOpenCode}
	liveFingerprint, err := currentRuntimeStartFingerprint(liveAgent, liveRuntime)
	require.NoError(t, err)

	d := mustNewDispatcher("http://server.test")
	defaultSessionKey := sessionKeyForBinding("agent-local", conversationBinding{}).String()
	sess := &interactiveSession{
		sessionKey:              defaultSessionKey,
		agentID:                 "agent-local",
		agent:                   liveAgent,
		serverAgentID:           "server-agent",
		runtime:                 liveRuntime,
		runtimeStartFingerprint: liveFingerprint,
		binding: conversationBinding{
			channelID:       "agentteam",
			serverAgentID:   "server-agent",
			serverSessionID: "sess_1",
		},
		workDir: workDir,
		process: proc,
	}
	d.sessionManager.Set(defaultSessionKey, sess)

	restoreDone := make(chan error, 1)
	go func() {
		restoreDone <- d.RestoreAgentRecoveryPoint(
			"agentteam", "server-agent", "req_restore", "opencode", "", "sess_1",
			rewindcaps.StrategyOpenCodeSessionRevert, "recovery-ref",
		)
	}()
	select {
	case <-driver.restoreStarted:
	case <-time.After(5 * time.Second):
		t.Fatal("live restore did not enter the runtime")
	}

	// The recovery request arrived with a server id different from the local
	// process id. BeginRecovery must still reserve the production local key;
	// otherwise this changed configuration would retire the Process mid-restore.
	_, starting := d.sessionManager.StartInFlight(defaultSessionKey)
	require.True(t, starting)
	changedAgent := liveAgent
	changedAgent.Model = "model-after"
	deliveryCtx, cancelDelivery := context.WithCancel(context.Background())
	deliveryDone := make(chan error, 1)
	go func() {
		_, deliveryErr := d.deliverResolvedSession(
			deliveryCtx,
			defaultSessionKey,
			changedAgent,
			liveRuntime,
			conversationBinding{
				channelID:       "agentteam",
				serverAgentID:   "server-agent",
				serverSessionID: "sess_1",
			},
			types.InboxMessage{MessageID: "wake-crossing-restore", Content: "new wake"},
			startSessionOptions{requireProcessHandoff: true},
		)
		deliveryDone <- deliveryErr
	}()
	select {
	case deliveryErr := <-deliveryDone:
		t.Fatalf("configuration restart crossed live restore: %v", deliveryErr)
	default:
	}
	require.False(t, sess.retired.Load())
	require.True(t, proc.AcceptingDeliveries())

	cancelDelivery()
	select {
	case deliveryErr := <-deliveryDone:
		require.ErrorIs(t, deliveryErr, context.Canceled)
	case <-time.After(5 * time.Second):
		t.Fatal("blocked delivery did not observe cancellation")
	}
	require.False(t, sess.retired.Load())
	require.True(t, proc.AcceptingDeliveries())

	driver.releaseRestore()
	select {
	case restoreErr := <-restoreDone:
		require.NoError(t, restoreErr)
	case <-time.After(5 * time.Second):
		t.Fatal("live restore did not finish")
	}
	require.False(t, sess.retired.Load())
	require.True(t, proc.AcceptingDeliveries())
}

func TestRestoreAgentRecoveryPointInterruptsBusyClaudeSessionBeforeDetachedRestore(t *testing.T) {
	configDir := t.TempDir()
	t.Setenv(profile.ConfigDirEnvKey, configDir)

	driver := newActiveRestoreRuntimeDriver("claude", rewindcaps.StrategyClaudeSessionResumeAt)
	workDir := filepath.Join(configDir, "agents", "server-agent", "sessions", "active-claude")
	proc, err := agent.NewProcess(
		context.Background(),
		driver,
		testLaunchConfig(drivercontract.LaunchConfig{AgentID: "agent-local", RuntimeType: runtimeconfig.TypeClaudeCode, SessionID: "sess_1", WorkDir: workDir}, testLaunchSurfaces{Prompt: "active turn"}),
		agent.ProcessOptions{},
	)
	require.NoError(t, err)
	proc.SessionID = "sess_1"
	t.Cleanup(proc.Sleep)
	require.False(t, proc.IsIdle())

	recoveryRef, err := rewindcaps.EncodeClaudeSessionResumeAtRecoveryRef("sess_1", "", rewindcaps.RecoveryBoundaryBeforeFirstInput)
	require.NoError(t, err)

	d := mustNewDispatcher("http://server.test")
	sessionKey := restoreSessionKeyForAgentTeamContext("agentteam", "server-agent", "sess_1")
	d.sessionManager.Set(sessionKey, &interactiveSession{
		sessionKey:    sessionKey,
		agentID:       "agent-local",
		serverAgentID: "server-agent",
		binding: conversationBinding{
			channelID:       "agentteam",
			serverAgentID:   "server-agent",
			serverSessionID: "sess_1",
		},
		runtime: runtimeconfig.Runtime{Type: runtimeconfig.TypeClaudeCode},
		workDir: workDir,
		process: proc,
	})

	err = d.RestoreAgentRecoveryPoint(
		"agentteam", "server-agent", "req_1", "claude", "", "sess_1",
		rewindcaps.StrategyClaudeSessionResumeAt, recoveryRef,
	)

	require.NoError(t, err)
	require.Equal(t, []string{"sess_1"}, driver.interrupts)
	require.FileExists(t, filepath.Join(workDir, ".daemon-claude-recovery.json"))
}

func TestRestoreAgentRecoveryPointUsesResolvedAgentRootAndCleansResumeWhenSessionIsMissing(t *testing.T) {
	configDir := t.TempDir()
	t.Setenv(profile.ConfigDirEnvKey, configDir)

	expectedWorkDir := sessionWorkDir("server-agent")
	require.NoError(t, os.MkdirAll(expectedWorkDir, 0o755))
	require.NoError(t, saveSessionResumeState(expectedWorkDir, sessionResumeState{
		LastSessionID:              "thread-before-restore",
		SessionResumeSafe:          true,
		SessionResumeSafetyEpoch:   4,
		SessionResumeSafetyVersion: 2,
		LastStartConfig: &types.AgentStartConfig{
			Runtime:   "codex",
			SessionID: "thread-before-restore",
		},
	}))

	d := mustNewDispatcher("http://server.test")
	var detachedCfg agent.DetachedRecoveryConfig
	d.recoveryPointDetachedRestorer = func(_ context.Context, cfg agent.DetachedRecoveryConfig, strategy, recoveryRef string) error {
		detachedCfg = cfg
		require.Equal(t, rewindcaps.StrategyCodexAppServerTurns, strategy)
		require.Equal(t, "recovery-ref", recoveryRef)
		return nil
	}

	err := d.RestoreAgentRecoveryPoint(
		"agentteam",
		"server-agent",
		"req_1",
		"codex",
		"",
		"sess_1",
		rewindcaps.StrategyCodexAppServerTurns,
		"recovery-ref",
	)

	require.NoError(t, err)
	require.Equal(t, "server-agent", detachedCfg.AgentID)
	require.Equal(t, expectedWorkDir, detachedCfg.WorkDir)
	require.DirExists(t, expectedWorkDir)
	require.NoDirExists(t, filepath.Join(expectedWorkDir, "sessions"), "recovery must not create a conversation-hashed state root")

	state, err := loadSessionResumeState(expectedWorkDir)
	require.NoError(t, err)
	require.Empty(t, state.LastSessionID)
	require.False(t, state.SessionResumeSafe)
	require.Equal(t, uint64(5), state.SessionResumeSafetyEpoch)
	require.Zero(t, state.SessionResumeSafetyVersion)
	require.NotNil(t, state.LastStartConfig)
	require.Empty(t, state.LastStartConfig.SessionID)
}

func TestRestoreAgentRecoveryPointDetachedRejectsEveryLiveDurableDeliveryOwner(t *testing.T) {
	configDir := t.TempDir()
	t.Setenv(profile.ConfigDirEnvKey, configDir)

	for _, owner := range []struct {
		name   string
		state  agent.DeliveryOwnershipState
		reason agent.DeliveryOwnershipInflightReason
	}{
		{name: "queued", state: agent.DeliveryOwnershipQueued},
		{name: "dispatching", state: agent.DeliveryOwnershipInflight, reason: agent.DeliveryInflightDispatching},
		{name: "outcome_unknown", state: agent.DeliveryOwnershipInflight, reason: agent.DeliveryInflightOutcomeUnknown},
	} {
		t.Run(owner.name, func(t *testing.T) {
			agentID := "detached-owner-" + owner.name
			workDir := sessionWorkDir(agentID)
			require.NoError(t, os.MkdirAll(workDir, 0o755))
			require.NoError(t, saveSessionResumeState(workDir, sessionResumeState{
				RuntimeDeliveryOwnershipVersion: runtimeDeliveryOwnershipVersion,
				RuntimeDeliveryOwnership: []agent.DeliveryOwnershipRecord{{
					Message:        types.InboxMessage{MessageID: "owned-wake"},
					State:          owner.state,
					InflightReason: owner.reason,
				}},
			}))

			d := mustNewDispatcher("http://server.test")
			detachedCalled := false
			d.recoveryPointDetachedRestorer = func(context.Context, agent.DetachedRecoveryConfig, string, string) error {
				detachedCalled = true
				return nil
			}

			err := d.RestoreAgentRecoveryPoint(
				"agentteam", agentID, "req_owner", "codex", "", "sess_1",
				rewindcaps.StrategyCodexAppServerTurns, "recovery-ref",
			)
			require.ErrorIs(t, err, agent.ErrAgentBusy)
			require.False(t, detachedCalled)
			_, starting := d.sessionManager.StartInFlight(agentID)
			require.False(t, starting, "failed recovery must release its reservation")
		})
	}
}

func TestRestoreAgentRecoveryPointDetachedFencesNewWakeStartThroughCleanup(t *testing.T) {
	configDir := t.TempDir()
	t.Setenv(profile.ConfigDirEnvKey, configDir)

	agentID := "detached-new-wake"
	workDir := sessionWorkDir(agentID)
	require.NoError(t, os.MkdirAll(workDir, 0o755))
	require.NoError(t, saveSessionResumeState(workDir, sessionResumeState{
		LastSessionID: "thread-before-restore",
		LastStartConfig: &types.AgentStartConfig{
			Runtime:   "codex",
			SessionID: "thread-before-restore",
		},
	}))

	d := mustNewDispatcher("http://server.test")
	restoreStarted := make(chan struct{})
	restoreRelease := make(chan struct{})
	d.recoveryPointDetachedRestorer = func(ctx context.Context, cfg agent.DetachedRecoveryConfig, _, _ string) error {
		require.Equal(t, workDir, cfg.WorkDir)
		close(restoreStarted)
		select {
		case <-restoreRelease:
			return nil
		case <-ctx.Done():
			return ctx.Err()
		}
	}

	restoreDone := make(chan error, 1)
	go func() {
		restoreDone <- d.RestoreAgentRecoveryPoint(
			"agentteam", agentID, "req_restore", "codex", "", "sess_1",
			rewindcaps.StrategyCodexAppServerTurns, "recovery-ref",
		)
	}()
	select {
	case <-restoreStarted:
	case <-time.After(5 * time.Second):
		t.Fatal("detached restore did not start")
	}

	defaultSessionKey := sessionKeyForBinding(agentID, conversationBinding{}).String()
	startReservation, waitCh, shouldStart := d.sessionManager.BeginStart(defaultSessionKey, nil)
	require.False(t, shouldStart, "new wake start must wait behind recovery")
	require.Nil(t, startReservation)

	type startResult struct {
		reservation *sessionStartReservation
		acquired    bool
	}
	startDone := make(chan startResult, 1)
	go func() {
		if err := waitForSessionStart(context.Background(), waitCh); err != nil {
			startDone <- startResult{}
			return
		}
		reservation, _, acquired := d.sessionManager.BeginStart(defaultSessionKey, nil)
		startDone <- startResult{reservation: reservation, acquired: acquired}
	}()
	select {
	case result := <-startDone:
		t.Fatalf("new wake acquired start generation before restore cleanup: %+v", result)
	default:
	}

	close(restoreRelease)
	select {
	case err := <-restoreDone:
		require.NoError(t, err)
	case <-time.After(5 * time.Second):
		t.Fatal("detached restore did not finish")
	}
	var result startResult
	select {
	case result = <-startDone:
	case <-time.After(5 * time.Second):
		t.Fatal("new wake start did not acquire after recovery")
	}
	require.True(t, result.acquired)
	require.NotNil(t, result.reservation)
	d.sessionManager.FinishStart(result.reservation)

	state, err := loadSessionResumeState(workDir)
	require.NoError(t, err)
	require.Empty(t, state.LastSessionID, "start generation must open only after resume cleanup is durable")
	require.NotNil(t, state.LastStartConfig)
	require.Empty(t, state.LastStartConfig.SessionID)
}

func TestSessionManagerRestartOwnerBlocksDetachedRecoveryReservation(t *testing.T) {
	sm := NewSessionManager()
	key := "agent-restart-owner"
	restartReservation, restartWait, acquired := sm.BeginStart(key, nil)
	require.True(t, acquired)

	target, recoveryReservation, recoveryWait, recoveryAcquired := sm.BeginRecovery(key, nil)
	require.Nil(t, target)
	require.Nil(t, recoveryReservation)
	require.False(t, recoveryAcquired)
	require.Equal(t, restartWait, recoveryWait)

	sm.FinishStart(restartReservation)
	target, recoveryReservation, _, recoveryAcquired = sm.BeginRecovery(key, nil)
	require.Nil(t, target)
	require.True(t, recoveryAcquired)
	require.NotNil(t, recoveryReservation)
	sm.FinishStart(recoveryReservation)
}

func TestDiscardCodexResumeAfterRecoveryClearsOnlyResumeFieldsAndPreservesDeliveryOwner(t *testing.T) {
	workDir := t.TempDir()
	require.NoError(t, saveSessionResumeState(workDir, sessionResumeState{
		LastSessionID:              "thread-old",
		SessionResumeSafe:          true,
		SessionResumeSafetyEpoch:   7,
		SessionResumeSafetyVersion: 3,
		LastStartConfig: &types.AgentStartConfig{
			Name:      "Codex Agent",
			Runtime:   "codex",
			Model:     "gpt-test",
			SessionID: "thread-old",
		},
		DeferredDeliveries: []deferredDeliveryEntry{{MessageID: "deferred-1", Seq: 1}},
		RuntimeDeliveryOwnership: []agent.DeliveryOwnershipRecord{{
			Message: types.InboxMessage{
				MessageID: "wake-owned",
				Target:    "#ops:thread",
				Content:   "accepted while recovery completed",
			},
			State: agent.DeliveryOwnershipQueued,
		}},
	}))

	d := mustNewDispatcher("http://server.test")
	t.Cleanup(d.Close)
	sessionKey := "session-key"
	sess := &interactiveSession{
		sessionKey:    sessionKey,
		agentID:       "agent-local",
		serverAgentID: "server-agent",
		workDir:       workDir,
	}
	d.sessionManager.Set(sessionKey, sess)
	continuationStarted := 0
	d.runtimeOwnershipContinuationStarter = func(got *interactiveSession) (bool, error) {
		require.Same(t, sess, got)
		continuationStarted++
		return true, nil
	}

	require.NoError(t, d.discardCodexResumeAfterRecovery(sess, workDir, "codex", false))

	state, err := loadSessionResumeState(workDir)
	require.NoError(t, err)
	require.Empty(t, state.LastSessionID)
	require.False(t, state.SessionResumeSafe)
	require.Equal(t, uint64(8), state.SessionResumeSafetyEpoch, "cleanup must fence callbacks from the retired Process")
	require.Zero(t, state.SessionResumeSafetyVersion)
	require.NotNil(t, state.LastStartConfig)
	require.Empty(t, state.LastStartConfig.SessionID)
	require.Equal(t, "gpt-test", state.LastStartConfig.Model, "non-resume launch configuration must survive rollback")
	require.Len(t, state.DeferredDeliveries, 1)
	require.Equal(t, "deferred-1", state.DeferredDeliveries[0].MessageID)
	require.Len(t, state.RuntimeDeliveryOwnership, 1)
	require.Equal(t, "wake-owned", state.RuntimeDeliveryOwnership[0].MessageID())
	require.True(t, state.RuntimeDeliveryOwnership[0].IsQueued())
	retained, ok := d.sessionManager.Get(sessionKey)
	require.True(t, ok, "accepted delivery ownership must retain its restart ledger")
	require.Same(t, sess, retained)
	require.Equal(t, 1, continuationStarted)

	// Process retirement must not strand an accepted wake. The next launch can
	// still recover the exact body from the preserved ownership journal.
	restored, restoreErr := prepareRuntimeDeliveryOwnershipLaunch(workDir, types.InboxMessage{}, true)
	require.NoError(t, restoreErr)
	require.True(t, restored.HandoffPersisted)
	require.Equal(t, "wake-owned", restored.Message.MessageID)
	require.Equal(t, "accepted while recovery completed", restored.Message.Content)
}

func TestDiscardCodexResumeAfterRecoveryIgnoresOtherRuntimes(t *testing.T) {
	workDir := t.TempDir()
	require.NoError(t, saveSessionResumeState(workDir, sessionResumeState{
		LastSessionID:     "session-keep",
		SessionResumeSafe: true,
	}))

	d := mustNewDispatcher("http://server.test")
	require.NoError(t, d.discardCodexResumeAfterRecovery(nil, workDir, "claude", false))

	state, err := loadSessionResumeState(workDir)
	require.NoError(t, err)
	require.Equal(t, "session-keep", state.LastSessionID)
}

// TestRestoreAgentRecoveryPointDetachedRetiresStoppedPiSession verifies that a
// successful Pi detached restore (proc != nil, non-live recovery) retires the
// stopped session so the immediately following wake can start a replacement
// process from the restored session, without requiring a daemon restart.
func TestRestoreAgentRecoveryPointDetachedRetiresStoppedPiSession(t *testing.T) {
	configDir := t.TempDir()
	t.Setenv(profile.ConfigDirEnvKey, configDir)

	driver := newActiveRestoreRuntimeDriver("pi", rewindcaps.StrategyPiSessionTreeRewind)
	workDir := sessionWorkDir("server-agent")
	require.NoError(t, os.MkdirAll(workDir, 0o755))
	proc, err := agent.NewProcess(
		context.Background(),
		driver,
		drivercontract.LaunchConfig{
			AgentID:     "agent-local",
			RuntimeType: runtimeconfig.TypePi,
			SessionID:   "sess_1",
			WorkDir:     workDir,
		},
		agent.ProcessOptions{},
	)
	require.NoError(t, err)
	proc.SessionID = "sess_1"
	t.Cleanup(proc.Sleep)
	require.True(t, proc.IsIdle(), "Pi process must be idle so the detached branch is reached")

	recoveryRef, err := rewindcaps.EncodePiSessionTreeRewindRef("sess_1", "entry-1", rewindcaps.RecoveryBoundaryResumeAt)
	require.NoError(t, err)

	d := mustNewDispatcher("http://server.test")
	detachedCalled := false
	d.recoveryPointDetachedRestorer = func(_ context.Context, _ agent.DetachedRecoveryConfig, strategy, _ string) error {
		detachedCalled = true
		require.Equal(t, rewindcaps.StrategyPiSessionTreeRewind, strategy)
		return nil
	}
	sessionKey := restoreSessionKeyForAgentTeamContext("agentteam", "server-agent", "sess_1")
	target := &interactiveSession{
		sessionKey:    sessionKey,
		agentID:       "agent-local",
		serverAgentID: "server-agent",
		binding: conversationBinding{
			channelID:       "agentteam",
			serverAgentID:   "server-agent",
			serverSessionID: "sess_1",
		},
		runtime: runtimeconfig.Runtime{Type: runtimeconfig.TypePi},
		workDir: workDir,
		process: proc,
	}
	d.sessionManager.Set(sessionKey, target)

	err = d.RestoreAgentRecoveryPoint(
		"agentteam", "server-agent", "req_1", "pi", "", "sess_1",
		rewindcaps.StrategyPiSessionTreeRewind, recoveryRef,
	)

	require.NoError(t, err)
	require.True(t, detachedCalled, "Pi detached restore executed")

	// The stopped Pi generation must be retired so the next wake starts a
	// replacement: the session process pointer is cleared and the session is
	// removed from the manager.
	target.mu.Lock()
	clearedProcess := target.process
	retired := target.retired.Load()
	target.mu.Unlock()
	require.Nil(t, clearedProcess, "stopped Pi process pointer cleared after detached restore")
	require.True(t, retired, "Pi session retired after detached restore")

	_, found := d.sessionManager.Get(sessionKey)
	require.False(t, found, "retired Pi session removed so the next wake can start a replacement")
}