package rewind

import (
	"context"
	"database/sql"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/core/domain"
	"github.com/tidbcloud/AgentTeam/internal/core/rewindcaps"
	"github.com/tidbcloud/AgentTeam/internal/core/rewindplan"
	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/store/access"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	"github.com/tidbcloud/AgentTeam/internal/store/model"
	"github.com/tidbcloud/AgentTeam/internal/store/testfixture"
)

func insertUnifiedDeliveredMessageFact(
	t *testing.T,
	ctx context.Context,
	queryer coredb.QueryExecer,
	workspaceID string,
	scopeID string,
	messageID string,
	agentMemberID string,
	seq int64,
	globalSeq int64,
	senderMemberID string,
	senderType string,
	messageType string,
	content string,
	createdAt time.Time,
) {
	t.Helper()
	require.NotEmpty(t, scopeID)

	mustExecRecovery(t, queryer, `
INSERT IGNORE INTO scopes (
    id, workspace_id, type, identity_key, latest_seq, latest_global_seq,
    latest_message_at, archived_until_seq, status, created_at, updated_at
) VALUES (
    $1, $2, 'channel', CONCAT('channel:', $1), $3, $4,
    $5, 0, 'active', $5, $5
)
`, scopeID, workspaceID, seq, globalSeq, createdAt)
	mustExecRecovery(t, queryer, `
INSERT IGNORE INTO scope_channel_profiles (
    scope_id, workspace_id, slug, display_name, description, visibility,
    owner_member_id, is_all_channel, created_at, updated_at
)
VALUES ($1, $2, $1, '', '', 'private', NULL, FALSE, $3, $3)
`, scopeID, workspaceID, createdAt)

	mustExecRecovery(t, queryer, `
INSERT IGNORE INTO scope_memberships (
    scope_id, member_id, visibility, owner_member_id, last_read_seq,
    last_read_global_seq, last_activity_global_seq, owner_context_dismissed_through_global_seq,
    effective_from_seq, effective_to_seq, created_at, updated_at, removed_at, archived_at
)
VALUES
    ($1, $2, 'public', NULL, 0, 0, 0, 0, 0, NULL, $4, $4, NULL, NULL),
    ($1, $3, 'public', NULL, 0, 0, 0, 0, 0, NULL, $4, $4, NULL, NULL)
`, scopeID, agentMemberID, senderMemberID, createdAt)

	mustExecRecovery(t, queryer, `
INSERT IGNORE INTO scope_notification_settings (
    scope_id, member_id, state, created_at, updated_at
)
VALUES
    ($1, $2, 'active', $4, $4),
    ($1, $3, 'active', $4, $4)
`, scopeID, agentMemberID, senderMemberID, createdAt)

	mustExecRecovery(t, queryer, `
INSERT IGNORE INTO scope_messages (
    id, workspace_id, scope_id, seq, workspace_global_seq,
    author_member_id, author_type, message_type, content, metadata, created_at, updated_at
) VALUES (
    $1, $2, $3, $4, $5,
    $6, $7, $8, $9, NULL, $10, $10
)
`, messageID, workspaceID, scopeID, seq, globalSeq, senderMemberID, senderType, messageType, content, createdAt)
}

func insertDeliveredWakeFanoutFact(
	t *testing.T,
	ctx context.Context,
	queryer coredb.QueryExecer,
	workspaceID string,
	scopeID string,
	messageID string,
	agentID string,
	agentMemberID string,
	seq int64,
	globalSeq int64,
	content string,
	createdAt time.Time,
) {
	t.Helper()
	payload, err := message.MarshalScopeFanoutPayload(
		message.AppMessage{ID: messageID, Seq: seq, ChannelID: scopeID, ChannelType: "channel", Content: content, CreatedAt: createdAt},
		message.MessageDelivery{
			ThreadID:  scopeID,
			Seq:       seq,
			GlobalSeq: globalSeq,
			MessageID: messageID,
			WakeRecipients: []message.AgentRecipient{{
				AgentID:         agentID,
				AgentMemberID:   agentMemberID,
				AutoWakeEnabled: true,
				DeliveryReason:  message.AgentDeliveryReasonDirectMention,
			}},
		},
	)
	require.NoError(t, err)
	record := message.BuildScopeMessageFanoutOutboxRecord(message.ScopeMessageFanoutOutboxRecordParams{
		WorkspaceID:   workspaceID,
		ScopeID:       scopeID,
		MessageID:     messageID,
		DedupeKey:     "scope:" + scopeID + ":message:" + messageID + ":fanout:" + message.ScopeOutboxFanoutScope,
		Payload:       payload,
		CreatedAt:     createdAt,
		NextAttemptAt: createdAt,
	})
	record.Status = "delivered"
	record.DeliveredAt = &createdAt
	require.NoError(t, message.NewScopeRepositories(queryer).Outbox.Insert(ctx, record))
}

func insertDeliveredWakeFanoutFactToThread(
	t *testing.T,
	ctx context.Context,
	queryer coredb.QueryExecer,
	workspaceID string,
	parentScopeID string,
	replyThreadScopeID string,
	messageID string,
	agentID string,
	agentMemberID string,
	seq int64,
	globalSeq int64,
	content string,
	createdAt time.Time,
) {
	t.Helper()
	payload, err := message.MarshalScopeFanoutPayload(
		message.AppMessage{ID: messageID, Seq: seq, ChannelID: parentScopeID, ChannelType: "channel", Content: content, CreatedAt: createdAt},
		message.MessageDelivery{
			ThreadID:  replyThreadScopeID,
			Seq:       seq,
			GlobalSeq: globalSeq,
			MessageID: messageID,
			Message: message.ReceiveMessage{
				MessageID:   messageID,
				TargetKey:   replyThreadScopeID,
				TargetSeq:   0,
				ChannelType: "thread",
				Content:     content,
				Timestamp:   createdAt,
			},
			WakeRecipients: []message.AgentRecipient{{
				AgentID:         agentID,
				AgentMemberID:   agentMemberID,
				AutoWakeEnabled: true,
				DeliveryReason:  message.AgentDeliveryReasonDirectMention,
			}},
		},
	)
	require.NoError(t, err)
	record := message.BuildScopeMessageFanoutOutboxRecord(message.ScopeMessageFanoutOutboxRecordParams{
		WorkspaceID:   workspaceID,
		ScopeID:       parentScopeID,
		MessageID:     messageID,
		DedupeKey:     "scope:" + parentScopeID + ":message:" + messageID + ":fanout:" + message.ScopeOutboxFanoutScope,
		Payload:       payload,
		CreatedAt:     createdAt,
		NextAttemptAt: createdAt,
	})
	record.Status = "delivered"
	record.DeliveredAt = &createdAt
	require.NoError(t, message.NewScopeRepositories(queryer).Outbox.Insert(ctx, record))
}

func insertDeliveredDurableOnlyFanoutFact(
	t *testing.T,
	ctx context.Context,
	queryer coredb.QueryExecer,
	workspaceID string,
	scopeID string,
	messageID string,
	agentID string,
	agentMemberID string,
	seq int64,
	globalSeq int64,
	content string,
	createdAt time.Time,
) {
	t.Helper()
	payload, err := message.MarshalScopeFanoutPayload(
		message.AppMessage{ID: messageID, Seq: seq, ChannelID: scopeID, ChannelType: "channel", Content: content, CreatedAt: createdAt},
		message.MessageDelivery{
			ThreadID:  scopeID,
			Seq:       seq,
			GlobalSeq: globalSeq,
			MessageID: messageID,
			Recipients: []message.AgentRecipient{{
				AgentID:        agentID,
				AgentMemberID:  agentMemberID,
				DeliveryReason: message.AgentDeliveryReasonSpaceDefault,
			}},
		},
	)
	require.NoError(t, err)
	record := message.BuildScopeMessageFanoutOutboxRecord(message.ScopeMessageFanoutOutboxRecordParams{
		WorkspaceID:   workspaceID,
		ScopeID:       scopeID,
		MessageID:     messageID,
		DedupeKey:     "scope:" + scopeID + ":message:" + messageID + ":fanout:" + message.ScopeOutboxFanoutScope,
		Payload:       payload,
		CreatedAt:     createdAt,
		NextAttemptAt: createdAt,
	})
	record.Status = "delivered"
	record.DeliveredAt = &createdAt
	require.NoError(t, message.NewScopeRepositories(queryer).Outbox.Insert(ctx, record))
}

type recoveryPointFixture struct {
	ctx           context.Context
	pool          coredb.DB
	store         *Store
	workspaceID   string
	userID        string
	agentID       string
	agentMemberID string
	humanMemberID string
	messageID     string
}

func setupRecoveryPointFixture(t *testing.T, runtime string) recoveryPointFixture {
	t.Helper()

	ctx := context.Background()
	pool := startRewindRollbackDB(t, ctx)
	now := time.Now().UTC()

	workspaceID := "wsp_rewind"
	userID := "usr_rewind"
	humanMemberID := "mem_human"
	agentMemberID := "mem_agent"
	machineID := "mac_rewind"
	agentID := "agt_rewind"
	channelID := "chn_all"
	messageID := "msg_rewind"
	store := New(pool, rewindPlanningTestOptions(workspaceID, userID, humanMemberID, channelID))

	mustExecRecovery(t, pool, `
INSERT INTO users (id, email, name, display_name, avatar_url, is_admin, password_hash, email_verified_at, created_at, updated_at)
VALUES ($1, 'rewind@example.com', 'rewind', 'Rewind', '', TRUE, 'hash', $2, $2, $2)
`, userID, now)
	mustExecRecovery(t, pool, `
INSERT INTO workspaces (id, organization_id, slug, name, latest_message_seq, created_at, updated_at)
VALUES ($1, $2, 'rewind', 'Rewind', 1, $3, $3)
`, workspaceID, testfixture.SeedWorkspaceOrganization(t, ctx, pool, workspaceID, now), now)
	mustExecRecovery(t, pool, `
INSERT INTO members (id, workspace_id, type, display_name, user_id, workspace_role, created_at, updated_at)
VALUES ($1, $2, 'human', 'Human', $3, 'owner', $4, $4)
`, humanMemberID, workspaceID, userID, now)
	mustExecRecovery(t, pool, `
INSERT INTO members (id, workspace_id, type, display_name, created_at, updated_at)
VALUES ($1, $2, 'agent', 'Agent', $3, $3)
`, agentMemberID, workspaceID, now)
	mustExecRecovery(t, pool, `
INSERT INTO machines (id, workspace_id, created_by_user_id, name, api_key, created_at, updated_at)
VALUES ($1, $2, $3, 'Rewind machine', 'machine-key', $4, $4)
`, machineID, workspaceID, userID, now)
	mustExecRecovery(t, pool, `
INSERT INTO agents (
    id, workspace_id, machine_id, member_id, name, avatar_url, runtime, model, reasoning_effort,
    description, auth_token_hash, auto_wake_enabled, status, activity, activity_detail, created_at, updated_at
) VALUES (
    $1, $2, $3, $4, 'rewind-agent', '', $5, 'gpt-5-codex', '',
    '', 'hash', TRUE, 'active', 'online', '', $6, $6
)
`, agentID, workspaceID, machineID, agentMemberID, runtime, now)
	insertUnifiedDeliveredMessageFact(t, ctx, pool, workspaceID, channelID, messageID, agentMemberID, 1, 1, humanMemberID, "human", "", "hello", now)
	insertDeliveredWakeFanoutFact(t, ctx, pool, workspaceID, channelID, messageID, agentID, agentMemberID, 1, 1, "hello", now)

	return recoveryPointFixture{
		ctx:           ctx,
		pool:          pool,
		store:         store,
		workspaceID:   workspaceID,
		userID:        userID,
		agentID:       agentID,
		agentMemberID: agentMemberID,
		humanMemberID: humanMemberID,
		messageID:     messageID,
	}
}

func rewindPlanningTestOptions(workspaceID string, userID string, humanMemberID string, channelID string) Options {
	return Options{
		LookupMessageChannelID: func(context.Context, string, string) (string, error) {
			return channelID, nil
		},
		ResolveAuthorizedHumanMessageTarget: func(context.Context, string, string, string) (access.HumanWorkspaceMember, message.ThreadAccess, error) {
			membership := access.HumanWorkspaceMember{
				Member: model.Member{
					ID:          humanMemberID,
					WorkspaceID: workspaceID,
					Type:        domain.MemberTypeHuman,
					UserID:      &userID,
				},
				WorkspaceRole: "owner",
			}
			target := message.ThreadAccess{
				ThreadID:   channelID,
				ThreadType: domain.ThreadTypeChannel,
				ChannelID:  channelID,
			}
			return membership, target, nil
		},
		RequireAgentOperation: func(context.Context, string, string, string) (access.HumanAgentAccess, error) {
			return access.HumanAgentAccess{}, nil
		},
		ReplaceReplyThreadParticipants: func(context.Context, coredb.Tx, string, []string) error {
			return nil
		},
		MergeReplyThreadParticipantIDs: message.MergeReplyThreadParticipantIDs,
	}
}

func (fixture recoveryPointFixture) createDeliveredMessage(t *testing.T, scopeID string, messageID string, seq int64, content string) {
	t.Helper()

	now := time.Now().UTC()
	var nextGlobalSeq int64
	err := fixture.pool.QueryRow(fixture.ctx, `
SELECT COALESCE(MAX(workspace_global_seq), 0) + 1
FROM scope_messages
WHERE workspace_id = $1
	`, fixture.workspaceID).Scan(&nextGlobalSeq)
	require.NoError(t, err)

	insertUnifiedDeliveredMessageFact(t, fixture.ctx, fixture.pool, fixture.workspaceID, scopeID, messageID, fixture.agentMemberID, seq, nextGlobalSeq, fixture.humanMemberID, "human", "", content, now)
	insertDeliveredWakeFanoutFact(t, fixture.ctx, fixture.pool, fixture.workspaceID, scopeID, messageID, fixture.agentID, fixture.agentMemberID, seq, nextGlobalSeq, content, now)
}

func (fixture recoveryPointFixture) createAgentMessage(t *testing.T, scopeID string, messageID string, seq int64, globalSeq int64, content string, createdAt time.Time) {
	t.Helper()

	mustExecRecovery(t, fixture.pool, `
INSERT INTO scope_messages (
    id, workspace_id, scope_id, seq, workspace_global_seq,
    author_member_id, author_type, message_type, content, metadata, created_at, updated_at
) VALUES (
    $1, $2, $3, $4, $5,
    $6, 'agent', 'text', $7, NULL, $8, $8
)
`, messageID, fixture.workspaceID, scopeID, seq, globalSeq, fixture.agentMemberID, content, createdAt)
}

func TestRecordMessageRecoveryPointUpsertsDeliveredHotMessage(t *testing.T) {
	t.Parallel()
	fixture := setupRecoveryPointFixture(t, "codex")
	firstRecoveryRef, err := rewindcaps.EncodeCodexAppServerRecoveryRef("codex-thread-1", 0)
	require.NoError(t, err)
	secondRecoveryRef, err := rewindcaps.EncodeCodexAppServerRecoveryRef("codex-thread-1", 2)
	require.NoError(t, err)

	require.NoError(t, fixture.store.RecordMessageRecoveryPoint(
		fixture.ctx,
		fixture.workspaceID,
		fixture.messageID,
		fixture.agentID,
		"codex-thread-1",
		rewindcaps.StrategyCodexAppServerTurns,
		firstRecoveryRef,
	))
	require.NoError(t, fixture.store.RecordMessageRecoveryPoint(
		fixture.ctx,
		fixture.workspaceID,
		fixture.messageID,
		fixture.agentID,
		"codex-thread-1",
		rewindcaps.StrategyCodexAppServerTurns,
		secondRecoveryRef,
	))

	var recorded MessageRecoveryPoint
	err = fixture.pool.QueryRow(fixture.ctx, `
SELECT workspace_id, message_id, agent_id, thread_id, recovery_strategy, recovery_ref, prepared_at
FROM message_recovery_points
WHERE workspace_id = $1 AND message_id = $2 AND agent_id = $3
`, fixture.workspaceID, fixture.messageID, fixture.agentID).Scan(
		&recorded.WorkspaceID,
		&recorded.MessageID,
		&recorded.AgentID,
		&recorded.ThreadID,
		&recorded.Strategy,
		&recorded.RecoveryRef,
		&recorded.PreparedAt,
	)
	require.NoError(t, err)
	require.Equal(t, secondRecoveryRef, recorded.RecoveryRef)
	require.False(t, recorded.PreparedAt.IsZero())
}

func TestRecordMessageRecoveryPointAcceptsBuiltinOpenCodeRecoveryRef(t *testing.T) {
	t.Parallel()
	fixture := setupRecoveryPointFixture(t, "builtin")
	recoveryRef, err := rewindcaps.EncodeOpenCodeSessionRevertRecoveryRef("builtin-session-1", "msg_1")
	require.NoError(t, err)

	require.NoError(t, fixture.store.RecordMessageRecoveryPoint(
		fixture.ctx,
		fixture.workspaceID,
		fixture.messageID,
		fixture.agentID,
		"builtin-session-1",
		rewindcaps.StrategyOpenCodeSessionRevert,
		recoveryRef,
	))
}

func TestRecordMessageRecoveryPointAcceptsClaudeResumeAtRecoveryRef(t *testing.T) {
	t.Parallel()
	fixture := setupRecoveryPointFixture(t, "claude")
	recoveryRef, err := rewindcaps.EncodeClaudeSessionResumeAtRecoveryRef("claude-session-1", "uuid-1", rewindcaps.RecoveryBoundaryResumeAt)
	require.NoError(t, err)

	require.NoError(t, fixture.store.RecordMessageRecoveryPoint(
		fixture.ctx,
		fixture.workspaceID,
		fixture.messageID,
		fixture.agentID,
		"claude-session-1",
		rewindcaps.StrategyClaudeSessionResumeAt,
		recoveryRef,
	))
}

func TestRecordMessageRecoveryPointRejectsOpenCodeRecoveryRef(t *testing.T) {
	t.Parallel()
	fixture := setupRecoveryPointFixture(t, "opencode")

	err := fixture.store.RecordMessageRecoveryPoint(
		fixture.ctx,
		fixture.workspaceID,
		fixture.messageID,
		fixture.agentID,
		"opencode-session-1",
		rewindcaps.StrategyCodexAppServerTurns,
		`{"sessionId":"opencode-session-1"}`,
	)
	require.ErrorContains(t, err, `runtime "opencode" requires strategy "opencode_session_revert"`)
}

func TestRecordMessageRecoveryPointRejectsUndeliveredMessage(t *testing.T) {
	t.Parallel()
	fixture := setupRecoveryPointFixture(t, "codex")
	recoveryRef, err := rewindcaps.EncodeCodexAppServerRecoveryRef("codex-thread-1", 0)
	require.NoError(t, err)

	err = fixture.store.RecordMessageRecoveryPoint(
		fixture.ctx,
		fixture.workspaceID,
		"msg_missing",
		fixture.agentID,
		"codex-thread-1",
		rewindcaps.StrategyCodexAppServerTurns,
		recoveryRef,
	)
	require.ErrorContains(t, err, "matching delivered hot message not found")
}

func TestRecordMessageRecoveryPointRejectsMalformedRecoveryRef(t *testing.T) {
	t.Parallel()
	fixture := setupRecoveryPointFixture(t, "codex")

	err := fixture.store.RecordMessageRecoveryPoint(
		fixture.ctx,
		fixture.workspaceID,
		fixture.messageID,
		fixture.agentID,
		"codex-thread-1",
		rewindcaps.StrategyCodexAppServerTurns,
		`{"threadId":"codex-thread-1","turnCount":-1}`,
	)
	require.ErrorContains(t, err, "invalid turn count")
}

func TestRecordMessageRecoveryPointRejectsMismatchedRecoveryThreadID(t *testing.T) {
	t.Parallel()
	fixture := setupRecoveryPointFixture(t, "codex")
	recoveryRef, err := rewindcaps.EncodeCodexAppServerRecoveryRef("codex-thread-1", 0)
	require.NoError(t, err)

	err = fixture.store.RecordMessageRecoveryPoint(
		fixture.ctx,
		fixture.workspaceID,
		fixture.messageID,
		fixture.agentID,
		"codex-thread-2",
		rewindcaps.StrategyCodexAppServerTurns,
		recoveryRef,
	)
	require.ErrorContains(t, err, `thread id "codex-thread-1" does not match thread id "codex-thread-2"`)
}

func TestBeginRewindOperationStoresPrimaryScopeLifecycle(t *testing.T) {
	t.Parallel()
	fixture := setupRecoveryPointFixture(t, "codex")
	recoveryRef, err := rewindcaps.EncodeCodexAppServerRecoveryRef("codex-thread-1", 0)
	require.NoError(t, err)
	require.NoError(t, WriteMessageRecoveryPointWithExecer(
		fixture.ctx,
		fixture.pool,
		fixture.workspaceID,
		fixture.messageID,
		fixture.agentID,
		"codex-thread-1",
		rewindcaps.StrategyCodexAppServerTurns,
		recoveryRef,
		time.Now().UTC(),
	))

	plan, err := fixture.store.PrepareRewindForUser(fixture.ctx, fixture.userID, fixture.workspaceID, fixture.messageID)
	require.NoError(t, err)
	require.NotEmpty(t, plan.ScopeID)
	require.Equal(t, plan.ScopeID, plan.RewindScopeID())
	require.NoError(t, plan.ValidateSuffixBoundaries())

	operationID, err := fixture.store.BeginRewindOperation(fixture.ctx, plan)
	require.NoError(t, err)

	var storedScopeID string
	var storedStatus string
	require.NoError(t, fixture.pool.QueryRow(fixture.ctx, `
SELECT thread_id, status
FROM rewind_operations
WHERE id = $1
`, operationID).Scan(&storedScopeID, &storedStatus))
	require.Equal(t, plan.RewindScopeID(), storedScopeID)
	require.Equal(t, rewindplan.OperationStatusPreparing.String(), storedStatus)

	_, err = fixture.store.BeginRewindOperation(fixture.ctx, plan)
	require.ErrorContains(t, err, "another rewind is already in progress")
}

func TestPrepareRewindPendingDeliveryRequiresInterruptForTUIRuntimes(t *testing.T) {
	for _, runtimeName := range []string{"codex", "claude", "opencode"} {
		runtimeName := runtimeName
		t.Run(runtimeName, func(t *testing.T) {
			t.Parallel()
			fixture := setupRecoveryPointFixture(t, runtimeName)

			plan, err := fixture.store.PrepareRewindForUser(fixture.ctx, fixture.userID, fixture.workspaceID, fixture.messageID)

			require.NoError(t, err)
			require.Empty(t, plan.RelatedAgents)
			require.Len(t, plan.PendingInterruptAgents, 1)
			require.Equal(t, fixture.agentID, plan.PendingInterruptAgents[0].AgentID)
			require.Equal(t, runtimeName, plan.PendingInterruptAgents[0].Runtime)
		})
	}
}

func TestPrepareRewindPendingDeliveryBlocksWithoutInterruptSupport(t *testing.T) {
	t.Parallel()
	fixture := setupRecoveryPointFixture(t, "builtin")

	_, err := fixture.store.PrepareRewindForUser(fixture.ctx, fixture.userID, fixture.workspaceID, fixture.messageID)

	require.ErrorContains(t, err, "pending input interruption")
}

func TestPrepareRewindAllowsWakeDeliveredAgentBeforeVisibleOutput(t *testing.T) {
	t.Parallel()
	fixture := setupRecoveryPointFixture(t, "codex")

	plan, err := fixture.store.PrepareRewindForUser(fixture.ctx, fixture.userID, fixture.workspaceID, fixture.messageID)

	require.NoError(t, err)
	require.Empty(t, plan.RelatedAgents)
	require.Len(t, plan.PendingInterruptAgents, 1)
	require.Equal(t, fixture.agentID, plan.PendingInterruptAgents[0].AgentID)

	operationID, err := fixture.store.BeginRewindOperation(fixture.ctx, plan)
	require.NoError(t, err)
	require.NoError(t, fixture.store.ApplyRewind(fixture.ctx, operationID, plan))
	require.True(t, rewindMessageIsRewound(t, fixture.ctx, fixture.pool, fixture.messageID))
}

func TestPrepareRewindCodexPendingActiveDeliveryRequiresInterrupt(t *testing.T) {
	t.Parallel()
	fixture := setupRecoveryPointFixture(t, "codex")
	fixture.createDeliveredMessage(t, "chn_all", "msg_rewind_pending_steer", 2, "pending steer")

	plan, err := fixture.store.PrepareRewindForUser(fixture.ctx, fixture.userID, fixture.workspaceID, "msg_rewind_pending_steer")

	require.NoError(t, err)
	require.Equal(t, "msg_rewind_pending_steer", plan.TargetMessageID)
	require.Equal(t, []string{"msg_rewind_pending_steer"}, plan.SuffixMessageIDs)
	require.Empty(t, plan.RelatedAgents)
	require.Len(t, plan.PendingInterruptAgents, 1)
	require.Equal(t, fixture.agentID, plan.PendingInterruptAgents[0].AgentID)
	require.Equal(t, "msg_rewind_pending_steer", plan.PendingInterruptAgents[0].FirstTouchMessageID)

	recoveryRef, err := rewindcaps.EncodeCodexAppServerRecoveryRef("codex-thread-1", 0)
	require.NoError(t, err)
	require.NoError(t, WriteMessageRecoveryPointWithExecer(
		fixture.ctx,
		fixture.pool,
		fixture.workspaceID,
		"msg_rewind_pending_steer",
		fixture.agentID,
		"codex-thread-1",
		rewindcaps.StrategyCodexAppServerTurns,
		recoveryRef,
		time.Now().UTC(),
	))

	err = fixture.store.ValidateRewindPlan(fixture.ctx, plan)
	require.ErrorContains(t, err, "newer activity changed the conversation")
}

func TestPrepareRewindCodexAckedActiveDeliveryExpandsToTurnBoundary(t *testing.T) {
	t.Parallel()
	fixture := setupRecoveryPointFixture(t, "codex")
	fixture.createDeliveredMessage(t, "chn_all", "msg_rewind_seen_steer", 2, "seen steer")
	recoveryRef, err := rewindcaps.EncodeCodexAppServerRecoveryRef("codex-thread-1", 0)
	require.NoError(t, err)
	for _, messageID := range []string{fixture.messageID, "msg_rewind_seen_steer"} {
		require.NoError(t, WriteMessageRecoveryPointWithExecer(
			fixture.ctx,
			fixture.pool,
			fixture.workspaceID,
			messageID,
			fixture.agentID,
			"codex-thread-1",
			rewindcaps.StrategyCodexAppServerTurns,
			recoveryRef,
			time.Now().UTC(),
		))
	}

	plan, err := fixture.store.PrepareRewindForUser(fixture.ctx, fixture.userID, fixture.workspaceID, "msg_rewind_seen_steer")

	require.NoError(t, err)
	require.Equal(t, fixture.messageID, plan.TargetMessageID)
	require.Equal(t, int64(1), plan.TargetSeq)
	require.Equal(t, []string{fixture.messageID, "msg_rewind_seen_steer"}, plan.SuffixMessageIDs)
	require.Empty(t, plan.PendingInterruptAgents)
	require.Len(t, plan.RelatedAgents, 1)
	require.Equal(t, fixture.agentID, plan.RelatedAgents[0].AgentID)
	require.Equal(t, fixture.messageID, plan.RelatedAgents[0].FirstTouchMessageID)
	require.Equal(t, recoveryRef, plan.RelatedAgents[0].RecoveryRef)
}

func TestPrepareRewindIgnoresDurableOnlyNoWakeAgentMembershipWithoutRecoveryPoint(t *testing.T) {
	t.Parallel()
	fixture := setupRecoveryPointFixture(t, "codex")

	_, err := fixture.pool.Exec(fixture.ctx, `
DELETE FROM scope_fanout_outbox
WHERE workspace_id = $1 AND scope_id = $2 AND message_id = $3
`, fixture.workspaceID, "chn_all", fixture.messageID)
	require.NoError(t, err)
	insertDeliveredDurableOnlyFanoutFact(
		t,
		fixture.ctx,
		fixture.pool,
		fixture.workspaceID,
		"chn_all",
		fixture.messageID,
		fixture.agentID,
		fixture.agentMemberID,
		1,
		1,
		"hello",
		time.Now().UTC(),
	)

	plan, err := fixture.store.PrepareRewindForUser(fixture.ctx, fixture.userID, fixture.workspaceID, fixture.messageID)

	require.NoError(t, err)
	require.Empty(t, plan.RelatedAgents)
}

func TestPrepareRewindRejectsAgentReplyMissingRecoveryPointAsRequiredState(t *testing.T) {
	t.Parallel()
	fixture := setupRecoveryPointFixture(t, "codex")
	now := time.Now().UTC()

	fixture.createAgentMessage(t, "chn_all", "msg_agent_reply_missing_recovery", 2, 2, "agent reply", now.Add(time.Second))

	_, err := fixture.store.PrepareRewindForUser(fixture.ctx, fixture.userID, fixture.workspaceID, fixture.messageID)

	require.ErrorContains(t, err, "required recovery point is missing")
}

func TestRecordMessageRecoveryPointAcceptsStartedThreadWakeDelivery(t *testing.T) {
	t.Parallel()
	fixture := setupRecoveryPointFixture(t, "codex")
	now := time.Now().UTC()
	const childScopeID = "thr_rewind_started_thread_record"

	seedRewindReplyThreadScope(t, fixture.pool, fixture.workspaceID, "chn_all", childScopeID, fixture.messageID, 0, 1, now.Add(time.Second))
	_, err := fixture.pool.Exec(fixture.ctx, `
DELETE FROM scope_memberships
WHERE scope_id = $1 AND member_id = $2
`, "chn_all", fixture.agentMemberID)
	require.NoError(t, err)
	_, err = fixture.pool.Exec(fixture.ctx, `
DELETE FROM scope_fanout_outbox
WHERE workspace_id = $1 AND scope_id = $2 AND message_id = $3
`, fixture.workspaceID, "chn_all", fixture.messageID)
	require.NoError(t, err)
	insertDeliveredWakeFanoutFactToThread(
		t,
		fixture.ctx,
		fixture.pool,
		fixture.workspaceID,
		"chn_all",
		childScopeID,
		fixture.messageID,
		fixture.agentID,
		fixture.agentMemberID,
		1,
		1,
		"hello",
		now,
	)
	recoveryRef, err := rewindcaps.EncodeCodexAppServerRecoveryRef("codex-started-thread", 1)
	require.NoError(t, err)

	require.NoError(t, fixture.store.RecordMessageRecoveryPoint(
		fixture.ctx,
		fixture.workspaceID,
		fixture.messageID,
		fixture.agentID,
		"codex-started-thread",
		rewindcaps.StrategyCodexAppServerTurns,
		recoveryRef,
	))

	points, err := fixture.store.ListAllAgentRecoveryPoints(fixture.ctx, fixture.workspaceID, fixture.agentID)
	require.NoError(t, err)
	require.Len(t, points, 1)
	require.Equal(t, fixture.messageID, points[0].MessageID)
	require.Equal(t, recoveryRef, points[0].RecoveryRef)
}

func TestDeleteAgentRecoveryPointsOutsideReplyThreadKeepsRootMessagePoint(t *testing.T) {
	t.Parallel()
	fixture := setupRecoveryPointFixture(t, "codex")
	now := time.Now().UTC()
	const childScopeID = "thr_rewind_started_thread_keep_root_point"

	seedRewindReplyThreadScope(t, fixture.pool, fixture.workspaceID, "chn_all", childScopeID, fixture.messageID, 0, 1, now.Add(time.Second))
	rootRef, err := rewindcaps.EncodeCodexAppServerRecoveryRef("codex-started-thread", 1)
	require.NoError(t, err)
	otherRef, err := rewindcaps.EncodeCodexAppServerRecoveryRef("codex-other-thread", 2)
	require.NoError(t, err)
	require.NoError(t, WriteMessageRecoveryPointWithExecer(
		fixture.ctx,
		fixture.pool,
		fixture.workspaceID,
		fixture.messageID,
		fixture.agentID,
		"codex-started-thread",
		rewindcaps.StrategyCodexAppServerTurns,
		rootRef,
		now,
	))
	fixture.createDeliveredMessage(t, "chn_other", "msg_rewind_other_point", 1, "other")
	require.NoError(t, WriteMessageRecoveryPointWithExecer(
		fixture.ctx,
		fixture.pool,
		fixture.workspaceID,
		"msg_rewind_other_point",
		fixture.agentID,
		"codex-other-thread",
		rewindcaps.StrategyCodexAppServerTurns,
		otherRef,
		now.Add(time.Second),
	))

	pruned, err := fixture.store.DeleteAgentRecoveryPointsOutsideThread(fixture.ctx, fixture.workspaceID, fixture.agentID, childScopeID)

	require.NoError(t, err)
	require.Len(t, pruned, 1)
	require.Equal(t, "msg_rewind_other_point", pruned[0].MessageID)
	points, err := fixture.store.ListAllAgentRecoveryPoints(fixture.ctx, fixture.workspaceID, fixture.agentID)
	require.NoError(t, err)
	require.Len(t, points, 1)
	require.Equal(t, fixture.messageID, points[0].MessageID)
	require.Equal(t, rootRef, points[0].RecoveryRef)
}

func TestPrepareRewindCascadesStartedReplyThreadAgentReplyWithRootRecoveryPoint(t *testing.T) {
	t.Parallel()
	fixture := setupRecoveryPointFixture(t, "codex")
	now := time.Now().UTC()
	const (
		childScopeID   = "thr_rewind_started_thread_agent_reply"
		childMessageID = "msg_rewind_started_thread_agent_reply"
	)

	seedRewindReplyThreadScope(t, fixture.pool, fixture.workspaceID, "chn_all", childScopeID, fixture.messageID, 1, 2, now.Add(time.Second))
	fixture.createAgentMessage(t, childScopeID, childMessageID, 1, 2, "16", now.Add(2*time.Second))
	_, err := fixture.pool.Exec(fixture.ctx, `
DELETE FROM scope_fanout_outbox
WHERE workspace_id = $1 AND scope_id = $2 AND message_id = $3
`, fixture.workspaceID, "chn_all", fixture.messageID)
	require.NoError(t, err)
	insertDeliveredWakeFanoutFactToThread(
		t,
		fixture.ctx,
		fixture.pool,
		fixture.workspaceID,
		"chn_all",
		childScopeID,
		fixture.messageID,
		fixture.agentID,
		fixture.agentMemberID,
		1,
		1,
		"8+8=?",
		now,
	)
	recoveryRef, err := rewindcaps.EncodeCodexAppServerRecoveryRef("codex-started-thread", 1)
	require.NoError(t, err)
	require.NoError(t, fixture.store.RecordMessageRecoveryPoint(
		fixture.ctx,
		fixture.workspaceID,
		fixture.messageID,
		fixture.agentID,
		"codex-started-thread",
		rewindcaps.StrategyCodexAppServerTurns,
		recoveryRef,
	))
	pruned, err := fixture.store.DeleteAgentRecoveryPointsOutsideThread(fixture.ctx, fixture.workspaceID, fixture.agentID, childScopeID)
	require.NoError(t, err)
	require.Empty(t, pruned)

	plan, err := fixture.store.PrepareRewindForUser(fixture.ctx, fixture.userID, fixture.workspaceID, fixture.messageID)

	require.NoError(t, err)
	require.Len(t, plan.CascadeReplyThreads, 1)
	require.Equal(t, childScopeID, plan.CascadeReplyThreads[0].CascadeScopeID())
	require.Equal(t, []string{childMessageID}, plan.CascadeReplyThreads[0].MessageIDs)
	require.Len(t, plan.RelatedAgents, 1)
	require.Equal(t, fixture.agentID, plan.RelatedAgents[0].AgentID)
	require.Equal(t, fixture.messageID, plan.RelatedAgents[0].FirstTouchMessageID)
	require.Equal(t, recoveryRef, plan.RelatedAgents[0].RecoveryRef)

	operationID, err := fixture.store.BeginRewindOperation(fixture.ctx, plan)
	require.NoError(t, err)
	require.NoError(t, fixture.store.ApplyRewind(fixture.ctx, operationID, plan))
	require.True(t, rewindMessageIsRewound(t, fixture.ctx, fixture.pool, fixture.messageID))
	require.True(t, rewindMessageIsRewound(t, fixture.ctx, fixture.pool, childMessageID))
}

func TestPrepareRewindCascadesEmptyStartedReplyThreadBeforeAgentOutput(t *testing.T) {
	t.Parallel()
	fixture := setupRecoveryPointFixture(t, "codex")
	now := time.Now().UTC()
	const childScopeID = "thr_rewind_started_thread_empty"

	seedRewindReplyThreadScope(t, fixture.pool, fixture.workspaceID, "chn_all", childScopeID, fixture.messageID, 0, 1, now.Add(time.Second))
	_, err := fixture.pool.Exec(fixture.ctx, `
DELETE FROM scope_fanout_outbox
WHERE workspace_id = $1 AND scope_id = $2 AND message_id = $3
`, fixture.workspaceID, "chn_all", fixture.messageID)
	require.NoError(t, err)
	insertDeliveredWakeFanoutFactToThread(
		t,
		fixture.ctx,
		fixture.pool,
		fixture.workspaceID,
		"chn_all",
		childScopeID,
		fixture.messageID,
		fixture.agentID,
		fixture.agentMemberID,
		1,
		1,
		"8+8=?",
		now,
	)

	plan, err := fixture.store.PrepareRewindForUser(fixture.ctx, fixture.userID, fixture.workspaceID, fixture.messageID)

	require.NoError(t, err)
	require.Empty(t, plan.RelatedAgents)
	require.Len(t, plan.CascadeReplyThreads, 1)
	require.Equal(t, childScopeID, plan.CascadeReplyThreads[0].CascadeScopeID())
	require.Empty(t, plan.CascadeReplyThreads[0].MessageIDs)
	operationID, err := fixture.store.BeginRewindOperation(fixture.ctx, plan)
	require.NoError(t, err)
	require.NoError(t, fixture.store.ApplyRewind(fixture.ctx, operationID, plan))
	require.True(t, rewindMessageIsRewound(t, fixture.ctx, fixture.pool, fixture.messageID))

	var archivedAt sql.NullTime
	require.NoError(t, fixture.pool.QueryRow(fixture.ctx, `
SELECT archived_at
FROM scopes
WHERE workspace_id = $1 AND id = $2
`, fixture.workspaceID, childScopeID).Scan(&archivedAt))
	require.True(t, archivedAt.Valid)
}

func TestPrepareRewindRejectsMalformedStoredRecoveryPoint(t *testing.T) {
	t.Parallel()
	fixture := setupRecoveryPointFixture(t, "codex")
	require.NoError(t, WriteMessageRecoveryPointWithExecer(
		fixture.ctx,
		fixture.pool,
		fixture.workspaceID,
		fixture.messageID,
		fixture.agentID,
		"codex-thread-1",
		rewindcaps.StrategyCodexAppServerTurns,
		`{"threadId":"codex-thread-1","turnCount":-1}`,
		time.Now().UTC(),
	))

	_, err := fixture.store.PrepareRewindForUser(fixture.ctx, fixture.userID, fixture.workspaceID, fixture.messageID)

	require.ErrorContains(t, err, "required recovery point is missing")
}

func TestValidateRewindPlanRejectsChangedStoredRecoveryPoint(t *testing.T) {
	t.Parallel()
	fixture := setupRecoveryPointFixture(t, "codex")
	recoveryRef, err := rewindcaps.EncodeCodexAppServerRecoveryRef("codex-thread-1", 0)
	require.NoError(t, err)
	require.NoError(t, WriteMessageRecoveryPointWithExecer(
		fixture.ctx,
		fixture.pool,
		fixture.workspaceID,
		fixture.messageID,
		fixture.agentID,
		"codex-thread-1",
		rewindcaps.StrategyCodexAppServerTurns,
		recoveryRef,
		time.Now().UTC(),
	))
	plan, err := fixture.store.PrepareRewindForUser(fixture.ctx, fixture.userID, fixture.workspaceID, fixture.messageID)
	require.NoError(t, err)
	require.Len(t, plan.RelatedAgents, 1)

	require.NoError(t, WriteMessageRecoveryPointWithExecer(
		fixture.ctx,
		fixture.pool,
		fixture.workspaceID,
		fixture.messageID,
		fixture.agentID,
		"codex-thread-1",
		rewindcaps.StrategyCodexAppServerTurns,
		`{"threadId":"codex-thread-1","turnCount":-1}`,
		time.Now().UTC(),
	))

	err = fixture.store.ValidateRewindPlan(fixture.ctx, plan)

	require.ErrorContains(t, err, "required recovery point is missing")
}

func TestValidateRewindPlanRejectsSuffixDrift(t *testing.T) {
	t.Parallel()
	fixture := setupRecoveryPointFixture(t, "codex")
	recoveryRef, err := rewindcaps.EncodeCodexAppServerRecoveryRef("codex-thread-1", 0)
	require.NoError(t, err)
	require.NoError(t, WriteMessageRecoveryPointWithExecer(
		fixture.ctx,
		fixture.pool,
		fixture.workspaceID,
		fixture.messageID,
		fixture.agentID,
		"codex-thread-1",
		rewindcaps.StrategyCodexAppServerTurns,
		recoveryRef,
		time.Now().UTC(),
	))
	plan, err := fixture.store.PrepareRewindForUser(fixture.ctx, fixture.userID, fixture.workspaceID, fixture.messageID)
	require.NoError(t, err)
	require.Equal(t, []string{fixture.messageID}, plan.SuffixMessageIDs)

	fixture.createDeliveredMessage(t, "chn_all", "msg_rewind_suffix_drift", 2, "newer message")

	err = fixture.store.ValidateRewindPlan(fixture.ctx, plan)

	require.ErrorContains(t, err, "newer activity changed the conversation")
}

func TestRecordMessageRecoveryPointAndPruneOtherThreadsDeletesOlderConversationPoints(t *testing.T) {
	t.Parallel()
	fixture := setupRecoveryPointFixture(t, "codex")
	fixture.createDeliveredMessage(t, "chn_other", "msg_other", 1, "other")

	otherRecoveryRef, err := rewindcaps.EncodeCodexAppServerRecoveryRef("codex-thread-other", 0)
	require.NoError(t, err)
	require.NoError(t, fixture.store.RecordMessageRecoveryPoint(
		fixture.ctx,
		fixture.workspaceID,
		"msg_other",
		fixture.agentID,
		"codex-thread-other",
		rewindcaps.StrategyCodexAppServerTurns,
		otherRecoveryRef,
	))

	currentRecoveryRef, err := rewindcaps.EncodeCodexAppServerRecoveryRef("codex-thread-current", 0)
	require.NoError(t, err)
	pruned, err := fixture.store.RecordMessageRecoveryPointAndPruneOtherThreads(
		fixture.ctx,
		fixture.workspaceID,
		fixture.messageID,
		fixture.agentID,
		"codex-thread-current",
		rewindcaps.StrategyCodexAppServerTurns,
		currentRecoveryRef,
	)
	require.NoError(t, err)
	require.Len(t, pruned, 1)
	require.Equal(t, "msg_other", pruned[0].MessageID)
	require.Equal(t, otherRecoveryRef, pruned[0].RecoveryRef)

	var remainingMessageIDs []string
	rows, err := fixture.pool.Query(fixture.ctx, `
SELECT message_id
FROM message_recovery_points
WHERE workspace_id = $1 AND agent_id = $2
ORDER BY message_id ASC
`, fixture.workspaceID, fixture.agentID)
	require.NoError(t, err)
	defer rows.Close()
	for rows.Next() {
		var messageID string
		require.NoError(t, rows.Scan(&messageID))
		remainingMessageIDs = append(remainingMessageIDs, messageID)
	}
	require.NoError(t, rows.Err())
	require.Equal(t, []string{fixture.messageID}, remainingMessageIDs)
}

func TestDeleteAllAgentRecoveryPointsRemovesAndReturnsRows(t *testing.T) {
	t.Parallel()
	fixture := setupRecoveryPointFixture(t, "codex")
	now := time.Now().UTC()
	var nextGlobalSeq int64
	err := fixture.pool.QueryRow(fixture.ctx, `
SELECT COALESCE(MAX(workspace_global_seq), 0) + 1
FROM scope_messages
WHERE workspace_id = $1
`, fixture.workspaceID).Scan(&nextGlobalSeq)
	require.NoError(t, err)
	insertUnifiedDeliveredMessageFact(t, fixture.ctx, fixture.pool, fixture.workspaceID, "chn_all", "msg_other_delete", fixture.agentMemberID, 2, nextGlobalSeq, fixture.humanMemberID, "human", "", "other", now)

	firstRef, err := rewindcaps.EncodeCodexAppServerRecoveryRef("codex-thread-1", 1)
	require.NoError(t, err)
	secondRef, err := rewindcaps.EncodeCodexAppServerRecoveryRef("codex-thread-2", 2)
	require.NoError(t, err)

	require.NoError(t, fixture.store.RecordMessageRecoveryPoint(
		fixture.ctx,
		fixture.workspaceID,
		fixture.messageID,
		fixture.agentID,
		"codex-thread-1",
		rewindcaps.StrategyCodexAppServerTurns,
		firstRef,
	))
	require.NoError(t, fixture.store.RecordMessageRecoveryPoint(
		fixture.ctx,
		fixture.workspaceID,
		"msg_other_delete",
		fixture.agentID,
		"codex-thread-2",
		rewindcaps.StrategyCodexAppServerTurns,
		secondRef,
	))

	deleted, err := fixture.store.DeleteAllAgentRecoveryPoints(fixture.ctx, fixture.workspaceID, fixture.agentID)
	require.NoError(t, err)
	require.Len(t, deleted, 2)

	var remaining int
	err = fixture.pool.QueryRow(fixture.ctx, `
SELECT COUNT(*)
FROM message_recovery_points
WHERE workspace_id = $1 AND agent_id = $2
`, fixture.workspaceID, fixture.agentID).Scan(&remaining)
	require.NoError(t, err)
	require.Zero(t, remaining)
}

func mustExecRecovery(t *testing.T, pool interface {
	Exec(context.Context, string, ...any) (coredb.Result, error)
}, query string, args ...any) {
	t.Helper()

	_, err := pool.Exec(context.Background(), query, args...)
	require.NoError(t, err)
}