package rewind

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/core/domain"
	"github.com/tidbcloud/AgentTeam/internal/core/identifiers"
	"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/coordinator"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	"github.com/tidbcloud/AgentTeam/internal/store/model"
	"github.com/tidbcloud/AgentTeam/internal/store/storeerr"
)

// Store owns the rewind domain: plan preparation, validation, execution,
// and recovery-point management. It holds its own coredb.DB connection
// and receives cross-domain dependencies through Options.
type Store struct {
	db      coredb.DB
	options Options
}

// New creates a new rewind Store with the given database connection and options.
func New(db coredb.DB, options Options) *Store {
	return &Store{db: db, options: options}
}

func (s *Store) Validate() error { return s.options.Validate() }

func (o Options) Validate() error {
	var missing []string
	if o.LookupMessageChannelID == nil {
		missing = append(missing, "LookupMessageChannelID")
	}
	if o.ResolveAuthorizedHumanMessageTarget == nil {
		missing = append(missing, "ResolveAuthorizedHumanMessageTarget")
	}
	if o.RequireAgentOperation == nil {
		missing = append(missing, "RequireAgentOperation")
	}
	if o.ReplaceReplyThreadParticipants == nil {
		missing = append(missing, "ReplaceReplyThreadParticipants")
	}
	if o.MergeReplyThreadParticipantIDs == nil {
		missing = append(missing, "MergeReplyThreadParticipantIDs")
	}
	if len(missing) > 0 {
		return fmt.Errorf("rewind store: missing dependencies: %s", strings.Join(missing, ", "))
	}
	return nil
}

// Options declares cross-domain dependencies injected by the root Store.
type Options struct {
	// LookupMessageChannelID resolves the channel ID for a message.
	LookupMessageChannelID func(ctx context.Context, workspaceID string, messageID string) (string, error)

	// ResolveAuthorizedHumanMessageTarget authorizes a user for message-target access.
	// Wraps root Store's resolveAuthorizedHumanMessageTarget, pre-binding the three
	// read-action parameters (channelReadAction, dmReadAction, threadReadAction).
	ResolveAuthorizedHumanMessageTarget func(ctx context.Context, userID string, workspaceID string, channelID string) (access.HumanWorkspaceMember, message.ThreadAccess, error)

	// RequireAgentOperation authorizes a user to operate on an agent.
	RequireAgentOperation func(ctx context.Context, userID string, workspaceID string, agentID string) (access.HumanAgentAccess, error)

	// ReplaceReplyThreadParticipants replaces the participant list of a reply thread.
	ReplaceReplyThreadParticipants func(ctx context.Context, tx coredb.Tx, threadChannelID string, participantIDs []string) error

	// MergeReplyThreadParticipantIDs merges author and delivered participant IDs.
	MergeReplyThreadParticipantIDs func(authorIDs []string, deliveredIDs []string) []string
}

// ErrNotFound is re-exported from storeerr for convenience.
var ErrNotFound = storeerr.ErrNotFound

func badRequest(message string) error { return storeerr.BadRequest(message) }
func forbidden(message string) error  { return storeerr.Forbidden(message) }
func conflict(message string) error   { return storeerr.Conflict(message) }

func rollback(tx coredb.Tx, ctx context.Context) {
	if tx != nil {
		_ = tx.Rollback(ctx)
	}
}

func (store *Store) lookupMessageChannelID(ctx context.Context, workspaceID string, messageID string) (string, error) {
	if store == nil || store.options.LookupMessageChannelID == nil {
		return "", fmt.Errorf("rewind store lookup message channel id dependency is not configured")
	}
	return store.options.LookupMessageChannelID(ctx, workspaceID, messageID)
}

func (store *Store) resolveAuthorizedHumanMessageTarget(ctx context.Context, userID string, workspaceID string, channelID string) (access.HumanWorkspaceMember, message.ThreadAccess, error) {
	if store == nil || store.options.ResolveAuthorizedHumanMessageTarget == nil {
		return access.HumanWorkspaceMember{}, message.ThreadAccess{}, fmt.Errorf("rewind store resolve authorized human message target dependency is not configured")
	}
	return store.options.ResolveAuthorizedHumanMessageTarget(ctx, userID, workspaceID, channelID)
}

func (store *Store) requireAgentOperation(ctx context.Context, userID string, workspaceID string, agentID string) (access.HumanAgentAccess, error) {
	if store == nil || store.options.RequireAgentOperation == nil {
		return access.HumanAgentAccess{}, fmt.Errorf("rewind store require agent operation dependency is not configured")
	}
	return store.options.RequireAgentOperation(ctx, userID, workspaceID, agentID)
}

func (store *Store) replaceReplyThreadParticipants(ctx context.Context, tx coredb.Tx, threadChannelID string, participantIDs []string) error {
	if store == nil || store.options.ReplaceReplyThreadParticipants == nil {
		return fmt.Errorf("rewind store replace reply thread participants dependency is not configured")
	}
	return store.options.ReplaceReplyThreadParticipants(ctx, tx, threadChannelID, participantIDs)
}

func (store *Store) mergeReplyThreadParticipantIDs(authorIDs []string, deliveredIDs []string) []string {
	if store == nil || store.options.MergeReplyThreadParticipantIDs == nil {
		return message.MergeReplyThreadParticipantIDs(authorIDs, deliveredIDs)
	}
	return store.options.MergeReplyThreadParticipantIDs(authorIDs, deliveredIDs)
}

// PrepareRewindForUser builds a rewind plan for the given user and target message.
func (store *Store) PrepareRewindForUser(ctx context.Context, userID string, workspaceID string, messageID string) (Plan, error) {
	trimmedWorkspaceID := strings.TrimSpace(workspaceID)
	trimmedMessageID := strings.TrimSpace(messageID)
	if trimmedWorkspaceID == "" {
		return Plan{}, badRequest("workspace id is required")
	}
	if trimmedMessageID == "" {
		return Plan{}, badRequest("messageId is required")
	}

	channelID, err := store.lookupMessageChannelID(ctx, trimmedWorkspaceID, trimmedMessageID)
	if err != nil {
		return Plan{}, err
	}
	if strings.TrimSpace(channelID) == "" {
		return Plan{}, badRequest("message is not in hot history or is not visible")
	}

	membership, target, err := store.resolveAuthorizedHumanMessageTarget(ctx, strings.TrimSpace(userID), trimmedWorkspaceID, channelID)
	if err != nil {
		return Plan{}, badRequest("message is not in hot history or is not visible")
	}

	targetScopeID := message.ScopeIDForThreadAccess(target)
	targetMessage, archivedUntilSeq, err := LoadTargetMessage(ctx, store.db, trimmedWorkspaceID, targetScopeID, trimmedMessageID)
	if err != nil {
		if errors.Is(err, ErrNotFound) {
			return Plan{}, badRequest("message is not in hot history or is not visible")
		}
		return Plan{}, err
	}

	scope := ""
	switch {
	case target.ReplyThread != nil:
		scope = "reply"
	case target.ThreadType == domain.ThreadTypeDM:
		scope = "dm"
	case target.ThreadType == domain.ThreadTypeChannel:
		scope = "channel"
	default:
		return Plan{}, badRequest("rewind is only supported for DMs, reply threads, and channels")
	}

	if targetMessage.AuthorType != string(domain.MemberTypeHuman) {
		return Plan{}, forbidden("message must be your own human-authored message")
	}
	if strings.TrimSpace(targetMessage.AuthorMemberID) != strings.TrimSpace(membership.Member.ID) {
		return Plan{}, forbidden("message must be your own human-authored message")
	}
	if strings.TrimSpace(targetMessage.MessageType) == "system" {
		return Plan{}, badRequest("system messages cannot be rewound")
	}
	if targetMessage.Seq > 1 && targetMessage.Seq-1 <= archivedUntilSeq {
		return Plan{}, badRequest("rewind would cross the hot/cold boundary")
	}

	targetDeliveriesByMessageID, err := LoadSuffixDeliveries(ctx, store.db, trimmedWorkspaceID, targetScopeID, targetMessage.Seq)
	if err != nil {
		return Plan{}, err
	}
	if expandedTarget, expanded, err := store.expandCodexTurnRewindBoundary(
		ctx,
		trimmedWorkspaceID,
		targetScopeID,
		targetMessage,
		targetDeliveriesByMessageID[targetMessage.ID],
	); err != nil {
		return Plan{}, err
	} else if expanded {
		targetMessage = expandedTarget
		if targetMessage.AuthorType != string(domain.MemberTypeHuman) ||
			strings.TrimSpace(targetMessage.AuthorMemberID) != strings.TrimSpace(membership.Member.ID) {
			return Plan{}, conflict("rewind is blocked because Codex already consumed this message in a turn started by another author")
		}
		if targetMessage.Seq > 1 && targetMessage.Seq-1 <= archivedUntilSeq {
			return Plan{}, badRequest("rewind would cross the hot/cold boundary")
		}
	}

	suffixMessages, err := LoadSuffixMessages(ctx, store.db, trimmedWorkspaceID, targetScopeID, targetMessage.Seq)
	if err != nil {
		return Plan{}, err
	}
	if len(suffixMessages) == 0 {
		return Plan{}, badRequest("message is not in hot history or is not visible")
	}
	rewoundScopes := []RewoundScope{{
		ScopeID:         targetScopeID,
		LatestSeq:       rewindLatestSeqAfter(targetMessage.Seq),
		RemovedMessages: rewoundMessagesFromSuffix(targetScopeID, suffixMessages),
	}}

	deliveriesByMessageID, err := LoadSuffixDeliveries(ctx, store.db, trimmedWorkspaceID, targetScopeID, targetMessage.Seq)
	if err != nil {
		return Plan{}, err
	}

	agentPlans := make(map[string]*AgentPlan)
	pendingInterruptPlans := make(map[string]AgentPlan)
	suffixMessageIDs := make([]string, 0, len(suffixMessages))
	for _, message := range suffixMessages {
		suffixMessageIDs = append(suffixMessageIDs, message.ID)
		if message.AuthorType == string(domain.MemberTypeHuman) && strings.TrimSpace(message.AuthorMemberID) != strings.TrimSpace(membership.Member.ID) {
			return Plan{}, conflict("rewind is blocked because another human sent a later message")
		}

		if strings.TrimSpace(message.AuthorAgentID) != "" {
			agentPlan := rewindplan.AgentPlanForID(agentPlans, message.AuthorAgentID)
			if agentPlan.AgentID == "" {
				*agentPlan = AgentPlan{
					AgentID:              strings.TrimSpace(message.AuthorAgentID),
					MachineID:            strings.TrimSpace(message.AuthorMachineID),
					Runtime:              strings.TrimSpace(message.AuthorRuntime),
					Model:                strings.TrimSpace(message.AuthorModel),
					SessionID:            strings.TrimSpace(message.AuthorSessionID),
					FirstTouchMessageID:  message.ID,
					FirstTouchSeq:        message.Seq,
					FirstTouchGlobalSeq:  message.GlobalSeq,
					FirstTouchAt:         message.CreatedAt,
					ObservedAgentMessage: true,
				}
			} else {
				agentPlan.ObservedAgentMessage = true
			}
		} else if message.AuthorType == string(domain.MemberTypeAgent) && !message.AuthorAgentLive {
			return Plan{}, conflict("rewind is blocked because a related agent is no longer available for recovery")
		}

		for _, delivery := range deliveriesByMessageID[message.ID] {
			if !delivery.AgentRuntimeLive {
				return Plan{}, conflict("rewind is blocked because a related agent is no longer available for recovery")
			}
			agentPlan := rewindplan.AgentPlanForID(agentPlans, delivery.AgentID)
			if agentPlan.AgentID != "" {
				continue
			}
			*agentPlan = AgentPlan{
				AgentID:             strings.TrimSpace(delivery.AgentID),
				MachineID:           strings.TrimSpace(delivery.MachineID),
				Runtime:             strings.TrimSpace(delivery.Runtime),
				Model:               strings.TrimSpace(delivery.Model),
				SessionID:           strings.TrimSpace(delivery.SessionID),
				FirstTouchMessageID: message.ID,
				FirstTouchSeq:       message.Seq,
				FirstTouchGlobalSeq: delivery.GlobalSeq,
				FirstTouchAt:        delivery.DeliveredAt,
				FirstTouchDelivery:  true,
			}
		}
	}

	// For DM/channel surfaces, check for reply threads on suffix messages (fork guard).
	var cascadeReplyThreads []CascadeReplyThread
	if scope == "dm" || scope == "channel" {
		cascadeReplyThreads, err = store.loadCascadeReplyThreads(ctx, trimmedWorkspaceID, targetScopeID, suffixMessages, membership.Member.ID)
		if err != nil {
			return Plan{}, err
		}

		// Collect agents from cascade reply threads.
		for _, crt := range cascadeReplyThreads {
			crtDeliveries, err := LoadSuffixDeliveries(ctx, store.db, trimmedWorkspaceID, crt.CascadeScopeID(), 0)
			if err != nil {
				return Plan{}, err
			}
			crtMessages, err := LoadSuffixMessages(ctx, store.db, trimmedWorkspaceID, crt.CascadeScopeID(), 0)
			if err != nil {
				return Plan{}, err
			}
			rewoundScopes = append(rewoundScopes, RewoundScope{
				ScopeID:         crt.CascadeScopeID(),
				LatestSeq:       0,
				RemovedMessages: rewoundMessagesFromSuffix(crt.CascadeScopeID(), crtMessages),
			})
			for _, message := range crtMessages {
				if message.AuthorType == string(domain.MemberTypeHuman) && strings.TrimSpace(message.AuthorMemberID) != strings.TrimSpace(membership.Member.ID) {
					return Plan{}, conflict("rewind is blocked because another human sent a message in a reply thread")
				}
				if strings.TrimSpace(message.AuthorAgentID) != "" {
					agentPlan := rewindplan.AgentPlanForID(agentPlans, message.AuthorAgentID)
					if agentPlan.AgentID == "" {
						*agentPlan = AgentPlan{
							AgentID:              strings.TrimSpace(message.AuthorAgentID),
							MachineID:            strings.TrimSpace(message.AuthorMachineID),
							Runtime:              strings.TrimSpace(message.AuthorRuntime),
							Model:                strings.TrimSpace(message.AuthorModel),
							SessionID:            strings.TrimSpace(message.AuthorSessionID),
							FirstTouchMessageID:  message.ID,
							FirstTouchSeq:        message.Seq,
							FirstTouchGlobalSeq:  message.GlobalSeq,
							FirstTouchAt:         message.CreatedAt,
							ObservedAgentMessage: true,
						}
					} else if rewindplan.FirstTouchPrecedes(message.GlobalSeq, message.CreatedAt, agentPlan) {
						agentPlan.FirstTouchMessageID = message.ID
						agentPlan.FirstTouchSeq = message.Seq
						agentPlan.FirstTouchGlobalSeq = message.GlobalSeq
						agentPlan.FirstTouchAt = message.CreatedAt
						agentPlan.FirstTouchDelivery = false
						agentPlan.ObservedAgentMessage = true
					} else {
						agentPlan.ObservedAgentMessage = true
					}
				} else if message.AuthorType == string(domain.MemberTypeAgent) && !message.AuthorAgentLive {
					return Plan{}, conflict("rewind is blocked because a related agent is no longer available for recovery")
				}

				for _, delivery := range crtDeliveries[message.ID] {
					if !delivery.AgentRuntimeLive {
						return Plan{}, conflict("rewind is blocked because a related agent is no longer available for recovery")
					}
					agentPlan := rewindplan.AgentPlanForID(agentPlans, delivery.AgentID)
					if agentPlan.AgentID != "" {
						if rewindplan.FirstTouchPrecedes(delivery.GlobalSeq, delivery.DeliveredAt, agentPlan) {
							agentPlan.FirstTouchMessageID = message.ID
							agentPlan.FirstTouchSeq = message.Seq
							agentPlan.FirstTouchGlobalSeq = delivery.GlobalSeq
							agentPlan.FirstTouchAt = delivery.DeliveredAt
							agentPlan.FirstTouchDelivery = true
						}
						continue
					}
					*agentPlan = AgentPlan{
						AgentID:             strings.TrimSpace(delivery.AgentID),
						MachineID:           strings.TrimSpace(delivery.MachineID),
						Runtime:             strings.TrimSpace(delivery.Runtime),
						Model:               strings.TrimSpace(delivery.Model),
						SessionID:           strings.TrimSpace(delivery.SessionID),
						FirstTouchMessageID: message.ID,
						FirstTouchSeq:       message.Seq,
						FirstTouchGlobalSeq: delivery.GlobalSeq,
						FirstTouchAt:        delivery.DeliveredAt,
						FirstTouchDelivery:  true,
					}
				}
			}
		}
	}

	// Collect all scope IDs in rewind scope for cross-scope exclusion.
	rewindScopeIDs := []string{targetScopeID}
	for _, crt := range cascadeReplyThreads {
		rewindScopeIDs = append(rewindScopeIDs, crt.CascadeScopeID())
	}

	relatedAgents := make([]AgentPlan, 0, len(agentPlans))
	for _, agentPlan := range agentPlans {
		capability := rewindcaps.ForRuntime(agentPlan.Runtime)
		if !capability.Supported {
			return Plan{}, conflict("rewind is blocked because a related agent does not support rewind")
		}

		firstTouchScopeID := firstTouchScopeIDForRecoveryPoint(targetScopeID, cascadeReplyThreads, agentPlan.FirstTouchMessageID)

		recoveryThreadID, strategy, recoveryRef, err := store.loadRecoveryPointForFirstTouch(
			ctx,
			store.db,
			trimmedWorkspaceID,
			firstTouchScopeID,
			agentPlan,
		)
		if err != nil {
			if errors.Is(err, ErrNotFound) {
				if agentPlanCanRewindWithoutRecoveryPoint(agentPlan) {
					// A delivered wake row alone only means the server handed
					// the message to the daemon. If the agent has not produced
					// visible output in the rewind suffix, there is no visible
					// runtime side effect to restore. However the input may still
					// be pending inside the runtime transport, so only runtimes
					// with a keep-session interrupt/abort primitive may tombstone
					// without a prepared recovery point.
					if agentPlanNeedsPendingInterrupt(agentPlan) {
						pendingInterruptPlans[strings.TrimSpace(agentPlan.AgentID)] = *agentPlan
						continue
					}
					return Plan{}, conflict("rewind is blocked because a related agent does not support pending input interruption")
				}
				return Plan{}, conflict("rewind is blocked because a required recovery point is missing")
			}
			return Plan{}, err
		}
		if err := validateRecoveryPointForPlan(capability, recoveryThreadID, strategy, recoveryRef); err != nil {
			return Plan{}, conflict("rewind is blocked because a required recovery point is missing")
		}
		agentPlan.RecoveryThreadID = recoveryThreadID
		agentPlan.RecoveryStrategy = strategy
		agentPlan.RecoveryRef = recoveryRef

		conflicted, err := AgentHasCrossScopeActivityAfterExcluding(ctx, store.db, trimmedWorkspaceID, agentPlan.AgentID, rewindScopeIDs, agentPlan.FirstTouchAt)
		if err != nil {
			return Plan{}, err
		}
		if conflicted {
			return Plan{}, conflict("rewind is blocked because a related agent touched another conversation after first participation")
		}

		relatedAgents = append(relatedAgents, *agentPlan)
	}
	rewindplan.SortAgentPlans(relatedAgents)
	pendingInterruptAgents := make([]AgentPlan, 0, len(pendingInterruptPlans))
	for _, pending := range pendingInterruptPlans {
		pendingInterruptAgents = append(pendingInterruptAgents, pending)
	}
	rewindplan.SortAgentPlans(pendingInterruptAgents)

	replyThreadScopeID := ""
	replyParentMessageID := ""
	broadcastScopeID := strings.TrimSpace(channelID)
	if target.ReplyThread != nil {
		replyThreadScopeID = strings.TrimSpace(target.ChannelID)
		replyParentMessageID = strings.TrimSpace(target.ReplyThread.ParentMessageID)
		broadcastScopeID = strings.TrimSpace(target.ReplyThread.ParentChannelID)
	}

	return Plan{
		WorkspaceID:            trimmedWorkspaceID,
		ScopeID:                targetScopeID,
		BroadcastScopeID:       broadcastScopeID,
		TargetMessageID:        strings.TrimSpace(targetMessage.ID),
		TargetSeq:              targetMessage.Seq,
		TargetCreatedAt:        targetMessage.CreatedAt,
		Scope: scope,
		ReplyThreadScopeID:     replyThreadScopeID,
		ReplyParentMessageID:   replyParentMessageID,
		SuffixMessageIDs:       suffixMessageIDs,
		RelatedAgents:          relatedAgents,
		PendingInterruptAgents: pendingInterruptAgents,
		RewoundScopes:          rewoundScopes,
		CascadeReplyThreads:    cascadeReplyThreads,
	}, nil
}

func rewoundMessagesFromSuffix(scopeID string, messages []SuffixMessage) []RewoundMessage {
	if len(messages) == 0 {
		return nil
	}
	facts := make([]RewoundMessage, 0, len(messages))
	scopeID = strings.TrimSpace(scopeID)
	for _, message := range messages {
		seq := message.Seq
		if seq <= 0 {
			continue
		}
		facts = append(facts, RewoundMessage{
			ScopeID:                 scopeID,
			Seq:                     seq,
			AuthorMemberID:          strings.TrimSpace(message.AuthorMemberID),
			NotifyPrincipalMemberID: strings.TrimSpace(message.NotifyPrincipalMemberID),
			AllowAuthorUnread:       message.AllowAuthorUnread,
			MessageType:             strings.TrimSpace(message.MessageType),
			CreatedAt:               message.CreatedAt,
		})
	}
	return facts
}

func rewindLatestSeqAfter(fromSeq int64) int64 {
	if fromSeq <= 1 {
		return 0
	}
	return fromSeq - 1
}

// ValidateRewindPlan validates that a rewind plan is still consistent with current state.
func (store *Store) ValidateRewindPlan(ctx context.Context, plan Plan) error {
	return coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		if err := LockRewindScopeBoundaries(ctx, tx, plan); err != nil {
			return err
		}
		if err := store.validateRewindPlanLocked(ctx, tx, plan); err != nil {
			return err
		}
		return nil
	})
}

// BeginRewindOperation creates a new rewind operation in preparing state.
func (store *Store) BeginRewindOperation(ctx context.Context, plan Plan) (string, error) {
	var operationID string
	if err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		if err := LockRewindScopeBoundaries(ctx, tx, plan); err != nil {
			return err
		}

		var existingID string
		err := tx.QueryRow(ctx, `
SELECT id
FROM rewind_operations
WHERE workspace_id = $1
  AND thread_id = $2
  AND status = $3
LIMIT 1
`, strings.TrimSpace(plan.WorkspaceID), plan.OperationScopeID(), rewindplan.OperationStatusPreparing.String()).Scan(&existingID)
		switch {
		case err == nil:
			return conflict("another rewind is already in progress")
		case !coredb.IsNotFound(err):
			return fmt.Errorf("check active rewind operation: %w", err)
		}

		generatedOperationID, err := identifiers.NewEntityID("rwd")
		if err != nil {
			return fmt.Errorf("generate rewind operation id: %w", err)
		}
		operationID = generatedOperationID

		agentIDs := make([]string, 0, len(plan.RelatedAgents))
		for _, related := range plan.RelatedAgents {
			agentID := strings.TrimSpace(related.AgentID)
			if agentID == "" {
				continue
			}
			agentIDs = append(agentIDs, agentID)
		}
		agentIDs = rewindplan.NormalizeNonEmptyStrings(agentIDs)
		sort.Strings(agentIDs)

		if _, err := tx.Exec(ctx, `
INSERT INTO rewind_operations (id, workspace_id, thread_id, message_id, status, failure_reason, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, '', NOW(), NOW())
`, operationID, strings.TrimSpace(plan.WorkspaceID), plan.OperationScopeID(), strings.TrimSpace(plan.TargetMessageID), rewindplan.OperationStatusPreparing.String()); err != nil {
			return fmt.Errorf("insert rewind operation: %w", err)
		}
		if err := InsertOperationAgents(ctx, tx, operationID, agentIDs); err != nil {
			return err
		}

		return nil
	}); err != nil {
		return "", err
	}
	return operationID, nil
}

// MarkRewindOperationFailed marks a rewind operation as failed.
func (store *Store) MarkRewindOperationFailed(ctx context.Context, operationID string, reason string) error {
	return SetOperationStatus(ctx, store.db, operationID, rewindplan.OperationStatusFailed, reason)
}

// MarkRewindOperationRepairRequired marks a rewind operation as requiring repair.
func (store *Store) MarkRewindOperationRepairRequired(ctx context.Context, operationID string, reason string) error {
	return SetOperationStatus(ctx, store.db, operationID, rewindplan.OperationStatusRepairRequired, reason)
}

// ApplyRewind executes a validated rewind plan.
func (store *Store) ApplyRewind(ctx context.Context, operationID string, plan Plan) error {
	return coordinator.New(store.db).Run(ctx, "apply rewind workflow", func(tx coredb.Tx) error {
		if err := LockRewindScopeBoundaries(ctx, tx, plan); err != nil {
			return err
		}
		if err := store.validateRewindPlanLocked(ctx, tx, plan); err != nil {
			return err
		}

		messageIDs := plan.SuffixMessageIDs
		if len(messageIDs) == 0 {
			return badRequest("rewind suffix is empty")
		}

		// Cascade-rewind reply threads hanging off suffix messages before
		// tombstoning the suffix itself.
		for _, crt := range plan.CascadeReplyThreads {
			if err := RewindCascadeReplyThread(ctx, tx, plan.WorkspaceID, crt); err != nil {
				return err
			}
		}

		var fromSeq int64
		if err := tx.QueryRow(ctx, `
SELECT seq
FROM scope_messages
WHERE workspace_id = $1
  AND scope_id = $2
  AND id = $3
`, strings.TrimSpace(plan.WorkspaceID), rewindPlanScopeID(plan), strings.TrimSpace(plan.TargetMessageID)).Scan(&fromSeq); err != nil {
			return fmt.Errorf("load rewind scope from seq: %w", err)
		}
		if _, err := message.RewindScopeSuffixTx(ctx, tx, message.ScopeRewindSuffixRequest{
			WorkspaceID:       strings.TrimSpace(plan.WorkspaceID),
			ScopeID:           rewindPlanScopeID(plan),
			FromSeq:           fromSeq,
			RewoundAt:         time.Now().UTC(),
			SuppressionReason: message.ScopeRewindSuppressionReason,
		}); err != nil {
			return fmt.Errorf("rewind scope suffix: %w", err)
		}

		if plan.Scope == "reply" {
			if err := store.recomputeReplyThreadState(ctx, tx, plan); err != nil {
				return err
			}
		}

		if _, err := tx.Exec(ctx, `
UPDATE rewind_operations
SET status = $2,
    failure_reason = '',
    updated_at = NOW()
WHERE id = $1
`, strings.TrimSpace(operationID), rewindplan.OperationStatusCompleted.String()); err != nil {
			return fmt.Errorf("complete rewind operation: %w", err)
		}

		return nil
	})
}

// --- Internal helpers ---

func rewindPlanScopeID(plan Plan) string {
	return plan.RewindScopeID()
}

func (store *Store) validateRewindPlanLocked(ctx context.Context, queryer Queryer, plan Plan) error {
	currentSuffixMessageIDs, err := LoadCurrentSuffixMessageIDs(ctx, queryer, plan.WorkspaceID, rewindPlanScopeID(plan), plan.TargetMessageID)
	if err != nil {
		return err
	}
	if !rewindplan.SameStringSlice(currentSuffixMessageIDs, plan.SuffixMessageIDs) {
		return conflict("rewind is blocked because newer activity changed the conversation")
	}

	// Validate cascade reply thread message IDs haven't drifted.
	for _, crt := range plan.CascadeReplyThreads {
		currentCRTMessageIDs, err := LoadCurrentSuffixMessageIDs(ctx, queryer, plan.WorkspaceID, crt.CascadeScopeID(), "")
		if err != nil {
			return err
		}
		if !rewindplan.SameStringSlice(currentCRTMessageIDs, crt.MessageIDs) {
			return conflict("rewind is blocked because newer activity changed the conversation")
		}
	}

	// Collect all scope IDs in rewind scope for cross-scope exclusion.
	rewindScopeIDs := []string{rewindPlanScopeID(plan)}
	for _, crt := range plan.CascadeReplyThreads {
		rewindScopeIDs = append(rewindScopeIDs, crt.CascadeScopeID())
	}

	for _, relatedAgent := range plan.RelatedAgents {
		capability := rewindcaps.ForRuntime(relatedAgent.Runtime)
		if !capability.Supported {
			return conflict("rewind is blocked because a related agent does not support rewind")
		}

		firstTouchScopeID := firstTouchScopeIDForRecoveryPoint(rewindPlanScopeID(plan), plan.CascadeReplyThreads, relatedAgent.FirstTouchMessageID)
		recoveryThreadID, strategy, recoveryRef, err := store.loadRecoveryPointForFirstTouch(
			ctx,
			queryer,
			plan.WorkspaceID,
			firstTouchScopeID,
			&relatedAgent,
		)
		if err != nil {
			if errors.Is(err, ErrNotFound) {
				return conflict("rewind is blocked because a required recovery point is missing")
			}
			return err
		}
		if err := validateRecoveryPointForPlan(capability, recoveryThreadID, strategy, recoveryRef); err != nil {
			return conflict("rewind is blocked because a required recovery point is missing")
		}
		if recoveryThreadID != strings.TrimSpace(relatedAgent.RecoveryThreadID) ||
			strategy != strings.TrimSpace(relatedAgent.RecoveryStrategy) ||
			recoveryRef != strings.TrimSpace(relatedAgent.RecoveryRef) {
			return conflict("rewind is blocked because a required recovery point is missing")
		}

		conflicted, err := AgentHasCrossScopeActivityAfterExcluding(
			ctx,
			queryer,
			plan.WorkspaceID,
			relatedAgent.AgentID,
			rewindScopeIDs,
			relatedAgent.FirstTouchAt,
		)
		if err != nil {
			return err
		}
		if conflicted {
			return conflict("rewind is blocked because a related agent touched another conversation after first participation")
		}
	}

	for _, pendingAgent := range plan.PendingInterruptAgents {
		if !agentPlanNeedsPendingInterrupt(&pendingAgent) {
			return conflict("rewind is blocked because a related agent does not support pending input interruption")
		}
		_, _, _, err := LoadMessageRecoveryPoint(ctx, queryer, plan.WorkspaceID, pendingAgent.FirstTouchMessageID, pendingAgent.AgentID)
		if err == nil {
			return conflict("rewind is blocked because newer activity changed the conversation")
		}
		if !errors.Is(err, ErrNotFound) {
			return err
		}
	}

	return nil
}

func (store *Store) expandCodexTurnRewindBoundary(ctx context.Context, workspaceID string, scopeID string, targetMessage TargetMessage, deliveries []model.RewindDeliveryRow) (TargetMessage, bool, error) {
	if len(deliveries) == 0 {
		return TargetMessage{}, false, nil
	}
	boundary := targetMessage
	for _, delivery := range deliveries {
		if strings.TrimSpace(delivery.Runtime) != rewindcaps.RuntimeCodex {
			continue
		}
		recoveryThreadID, strategy, recoveryRef, err := LoadMessageRecoveryPoint(ctx, store.db, workspaceID, targetMessage.ID, delivery.AgentID)
		if err != nil {
			if errors.Is(err, ErrNotFound) {
				continue
			}
			return TargetMessage{}, false, err
		}
		if strings.TrimSpace(strategy) != rewindcaps.StrategyCodexAppServerTurns {
			continue
		}
		candidate, err := LoadEarliestMessageForRecoveryPoint(
			ctx,
			store.db,
			workspaceID,
			scopeID,
			delivery.AgentID,
			recoveryThreadID,
			strategy,
			recoveryRef,
			targetMessage.Seq,
		)
		if err != nil {
			if errors.Is(err, ErrNotFound) {
				continue
			}
			return TargetMessage{}, false, err
		}
		if candidate.Seq > 0 && candidate.Seq < boundary.Seq {
			boundary = candidate
		}
	}
	if boundary.ID == "" || boundary.Seq == targetMessage.Seq {
		return TargetMessage{}, false, nil
	}
	return boundary, true, nil
}

func validateRecoveryPointForPlan(capability rewindcaps.Capability, recoveryThreadID string, strategy string, recoveryRef string) error {
	if strings.TrimSpace(strategy) != strings.TrimSpace(capability.Strategy) {
		return fmt.Errorf("recovery strategy mismatch")
	}
	return rewindcaps.ValidatePreparedRecoveryPoint(strategy, recoveryThreadID, recoveryRef)
}

func agentPlanCanRewindWithoutRecoveryPoint(agentPlan *AgentPlan) bool {
	return agentPlan != nil && agentPlan.FirstTouchDelivery && !agentPlan.ObservedAgentMessage
}

func agentPlanNeedsPendingInterrupt(agentPlan *AgentPlan) bool {
	if !agentPlanCanRewindWithoutRecoveryPoint(agentPlan) ||
		strings.TrimSpace(agentPlan.AgentID) == "" ||
		strings.TrimSpace(agentPlan.MachineID) == "" {
		return false
	}
	switch strings.TrimSpace(agentPlan.Runtime) {
	case rewindcaps.RuntimeCodex, rewindcaps.RuntimeClaude, rewindcaps.RuntimeOpenCode:
		return true
	default:
		return false
	}
}

func firstTouchScopeIDForRecoveryPoint(defaultScopeID string, cascadeReplyThreads []CascadeReplyThread, firstTouchMessageID string) string {
	trimmedMessageID := strings.TrimSpace(firstTouchMessageID)
	for _, crt := range cascadeReplyThreads {
		for _, msgID := range crt.MessageIDs {
			if strings.TrimSpace(msgID) == trimmedMessageID {
				return strings.TrimSpace(crt.CascadeScopeID())
			}
		}
	}
	return strings.TrimSpace(defaultScopeID)
}

func (store *Store) loadRecoveryPointForFirstTouch(ctx context.Context, queryer Queryer, workspaceID string, scopeID string, agentPlan *AgentPlan) (string, string, string, error) {
	if agentPlan == nil {
		return "", "", "", ErrNotFound
	}

	recoveryThreadID, strategy, recoveryRef, err := LoadMessageRecoveryPoint(
		ctx,
		queryer,
		workspaceID,
		agentPlan.FirstTouchMessageID,
		agentPlan.AgentID,
	)
	if err == nil {
		return recoveryThreadID, strategy, recoveryRef, nil
	}
	if !errors.Is(err, ErrNotFound) {
		return "", "", "", err
	}
	if agentPlan.FirstTouchSeq <= 1 {
		return "", "", "", ErrNotFound
	}

	return LoadLatestRecoveryPointBeforeSeq(ctx, queryer, workspaceID, scopeID, agentPlan.AgentID, agentPlan.FirstTouchSeq)
}

// loadCascadeReplyThreads finds reply thread scopes on suffix messages that need cascade tombstone.
// It enforces the fork guard: suffix messages (other than the last non-system
// message) with reply threads block rewind. Rewind notices and other system
// bookkeeping messages can trail the target message without turning a simple
// tail reply-thread cascade into a fork.
func (store *Store) loadCascadeReplyThreads(ctx context.Context, workspaceID string, parentScopeID string, suffixMessages []SuffixMessage, currentMemberID string) ([]CascadeReplyThread, error) {
	suffixMessageIDs := suffixMessageIDsFromMessages(suffixMessages)
	if len(suffixMessageIDs) == 0 {
		return nil, nil
	}

	suffixMessageIDsCTE, suffixMessageIDArgs, _, err := coredb.BuildStringValuesCTE("suffix_message_ids", "message_id", rewindplan.NormalizeNonEmptyStrings(suffixMessageIDs), 3)
	if err != nil {
		return nil, err
	}

	query := fmt.Sprintf(`
WITH %s
SELECT child.id AS thread_channel_id,
       child.root_message_id AS parent_message_id,
       child.id AS thread_id
FROM scopes parent
JOIN scope_messages root
  ON root.workspace_id = parent.workspace_id
 AND root.scope_id = parent.id
JOIN suffix_message_ids suffix
  ON suffix.message_id = root.id
JOIN scopes child
  ON child.workspace_id = parent.workspace_id
 AND child.parent_scope_id = parent.id
 AND child.root_message_id = root.id
 AND child.type = 'thread'
 AND child.status = 'active'
 AND child.archived_at IS NULL
WHERE parent.workspace_id = $1
  AND parent.id = $2
  AND parent.status = 'active'
  AND parent.archived_at IS NULL
  AND `+message.MessageNotRewoundSQL("root")+`
ORDER BY child.created_at ASC, child.id ASC
`, suffixMessageIDsCTE)
	args := append([]any{strings.TrimSpace(workspaceID), strings.TrimSpace(parentScopeID)}, suffixMessageIDArgs...)
	rows, err := store.db.Query(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("load cascade reply threads: %w", err)
	}
	defer rows.Close()

	type replyThreadInfo struct {
		threadChannelID string
		parentMessageID string
		threadID        string
	}
	var replyThreadInfos []replyThreadInfo
	for rows.Next() {
		var info replyThreadInfo
		if err := rows.Scan(&info.threadChannelID, &info.parentMessageID, &info.threadID); err != nil {
			return nil, fmt.Errorf("scan cascade reply thread: %w", err)
		}
		replyThreadInfos = append(replyThreadInfos, info)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate cascade reply threads: %w", err)
	}

	if len(replyThreadInfos) == 0 {
		return nil, nil
	}

	// Fork guard: only the last non-system business suffix message is allowed to
	// have a reply thread. Structured rewind notices were excluded before this
	// point; any remaining generic system rows are not user-authored branch
	// points.
	lastSuffixMessageID := lastNonSystemSuffixMessageID(suffixMessages)
	if lastSuffixMessageID == "" {
		return nil, nil
	}
	for _, info := range replyThreadInfos {
		if strings.TrimSpace(info.parentMessageID) != strings.TrimSpace(lastSuffixMessageID) {
			return nil, conflict("rewind is blocked because a message in the suffix has a reply thread; please rewind it first")
		}
	}

	threadChannelIDs := make([]string, 0, len(replyThreadInfos))
	for _, info := range replyThreadInfos {
		threadChannelIDs = append(threadChannelIDs, info.threadChannelID)
	}
	messagesByThreadID, err := loadCascadeReplyThreadMessageIDs(ctx, store.db, workspaceID, threadChannelIDs)
	if err != nil {
		return nil, err
	}

	result := make([]CascadeReplyThread, 0, len(replyThreadInfos))
	for _, info := range replyThreadInfos {
		result = append(result, CascadeReplyThread{
			ScopeID:         strings.TrimSpace(info.threadChannelID),
			ParentMessageID: strings.TrimSpace(info.parentMessageID),
			MessageIDs:      messagesByThreadID[strings.TrimSpace(info.threadChannelID)],
		})
	}

	return result, nil
}

func suffixMessageIDsFromMessages(messages []SuffixMessage) []string {
	ids := make([]string, 0, len(messages))
	for _, message := range messages {
		if id := strings.TrimSpace(message.ID); id != "" {
			ids = append(ids, id)
		}
	}
	return ids
}

func lastNonSystemSuffixMessageID(messages []SuffixMessage) string {
	for index := len(messages) - 1; index >= 0; index-- {
		if strings.TrimSpace(messages[index].MessageType) == message.MessageTypeSystem {
			continue
		}
		if id := strings.TrimSpace(messages[index].ID); id != "" {
			return id
		}
	}
	return ""
}

func loadCascadeReplyThreadMessageIDs(ctx context.Context, queryer Queryer, workspaceID string, threadChannelIDs []string) (map[string][]string, error) {
	normalizedThreadIDs := rewindplan.NormalizeNonEmptyStrings(threadChannelIDs)
	messagesByThreadID := make(map[string][]string, len(normalizedThreadIDs))
	for _, threadID := range normalizedThreadIDs {
		messagesByThreadID[threadID] = nil
	}
	if len(normalizedThreadIDs) == 0 {
		return messagesByThreadID, nil
	}

	threadIDsCTE, threadIDArgs, _, err := coredb.BuildStringValuesCTE("cascade_thread_ids", "thread_channel_id", normalizedThreadIDs, 2)
	if err != nil {
		return nil, err
	}
	query := fmt.Sprintf(`
WITH %s
SELECT requested.thread_channel_id,
       m.id AS message_id
FROM cascade_thread_ids requested
JOIN scopes child
  ON child.id = requested.thread_channel_id
 AND child.workspace_id = $1
 AND child.type = 'thread'
 AND child.status = 'active'
 AND child.archived_at IS NULL
JOIN scope_messages m
  ON m.workspace_id = child.workspace_id
 AND m.scope_id = child.id
 AND `+message.MessageNotRewoundSQL("m")+`
ORDER BY requested.thread_channel_id ASC, m.seq ASC
`, threadIDsCTE)
	args := append([]any{strings.TrimSpace(workspaceID)}, threadIDArgs...)
	rows, err := queryer.Query(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("load cascade reply thread messages: %w", err)
	}
	defer rows.Close()

	for rows.Next() {
		var threadID string
		var messageID string
		if err := rows.Scan(&threadID, &messageID); err != nil {
			return nil, fmt.Errorf("scan cascade reply thread message: %w", err)
		}
		trimmedThreadID := strings.TrimSpace(threadID)
		if trimmedThreadID == "" {
			continue
		}
		messagesByThreadID[trimmedThreadID] = append(messagesByThreadID[trimmedThreadID], strings.TrimSpace(messageID))
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate cascade reply thread messages: %w", err)
	}
	return messagesByThreadID, nil
}

// recomputeReplyThreadState trims suffix membership facts and rebuilds participants after rewind.
func (store *Store) recomputeReplyThreadState(ctx context.Context, tx coredb.Tx, plan Plan) error {
	rows, err := tx.Query(ctx, `
SELECT
    CASE
        WHEN m.author_type = 'human' THEN COALESCE(mem.user_id, '')
        WHEN m.author_type = 'agent' THEN COALESCE(a.id, '')
        ELSE ''
    END AS author_participant_id
FROM scope_messages m
JOIN scopes child
  ON child.id = m.scope_id
 AND child.workspace_id = m.workspace_id
 AND child.type = 'thread'
 AND child.status = 'active'
 AND child.archived_at IS NULL
JOIN members mem ON mem.id = m.author_member_id
LEFT JOIN agents a ON a.workspace_id = m.workspace_id AND a.member_id = m.author_member_id AND m.author_type = 'agent'
WHERE m.workspace_id = $1
  AND m.scope_id = $2
ORDER BY m.seq ASC
`, strings.TrimSpace(plan.WorkspaceID), strings.TrimSpace(plan.ReplyThreadStateScopeID()))
	if err != nil {
		return fmt.Errorf("load remaining reply thread messages: %w", err)
	}
	authorParticipantIDs := make([]string, 0)
	for rows.Next() {
		var authorParticipantID string
		if err := rows.Scan(&authorParticipantID); err != nil {
			rows.Close()
			return fmt.Errorf("scan remaining reply thread message: %w", err)
		}
		authorParticipantIDs = append(authorParticipantIDs, strings.TrimSpace(authorParticipantID))
	}
	if err := rows.Err(); err != nil {
		rows.Close()
		return fmt.Errorf("iterate remaining reply thread messages: %w", err)
	}
	rows.Close()

	if err := trimRewindReplyThreadScopeFacts(ctx, tx, plan); err != nil {
		return err
	}

	deliveredParticipantIDs := make([]string, 0)
	deliveredRows, err := tx.Query(ctx, `
SELECT DISTINCT a.id AS participant_id
FROM scope_memberships sm
JOIN agents a
  ON a.member_id = sm.member_id
WHERE sm.scope_id = $1
  AND sm.removed_at IS NULL
  AND sm.archived_at IS NULL
  AND sm.effective_to_seq IS NULL
  AND ($3 = 0 OR sm.effective_from_seq < $3)
  AND a.workspace_id = $2
`, strings.TrimSpace(plan.ReplyThreadStateScopeID()), strings.TrimSpace(plan.WorkspaceID), plan.TargetSeq)
	if err != nil {
		return fmt.Errorf("load remaining reply thread agent participants: %w", err)
	}

	for deliveredRows.Next() {
		var participantID string
		if err := deliveredRows.Scan(&participantID); err != nil {
			deliveredRows.Close()
			return fmt.Errorf("scan remaining reply thread agent participant: %w", err)
		}
		deliveredParticipantIDs = append(deliveredParticipantIDs, strings.TrimSpace(participantID))
	}
	if err := deliveredRows.Err(); err != nil {
		deliveredRows.Close()
		return fmt.Errorf("iterate remaining reply thread agent participants: %w", err)
	}
	deliveredRows.Close()

	participantIDs := store.mergeReplyThreadParticipantIDs(authorParticipantIDs, deliveredParticipantIDs)

	if err := store.replaceReplyThreadParticipants(ctx, tx, strings.TrimSpace(plan.ReplyThreadStateScopeID()), participantIDs); err != nil {
		return fmt.Errorf("update rewind reply thread participants: %w", err)
	}

	return nil
}

func trimRewindReplyThreadScopeFacts(ctx context.Context, tx coredb.Tx, plan Plan) error {
	if plan.TargetSeq > 0 {
		if _, err := tx.Exec(ctx, `
UPDATE scope_memberships
SET unread_generation = unread_generation + 1,
    effective_to_seq = effective_from_seq,
    updated_at = NOW()
WHERE scope_id = $1
  AND effective_from_seq >= $2
  AND effective_to_seq IS NULL
`, strings.TrimSpace(plan.ReplyThreadStateScopeID()), plan.TargetSeq); err != nil {
			return fmt.Errorf("trim rewind reply thread memberships: %w", err)
		}
	}

	return nil
}

// --- Recovery point methods ---

// RecordMessageRecoveryPoint records a recovery point for an agent message.
func (store *Store) RecordMessageRecoveryPoint(ctx context.Context, workspaceID string, messageID string, agentID string, threadID string, strategy string, recoveryRef string) error {
	_, err := store.RecordMessageRecoveryPointAndPruneOtherThreads(ctx, workspaceID, messageID, agentID, threadID, strategy, recoveryRef)
	return err
}

// RecordMessageRecoveryPointAndPruneOtherThreads records a recovery point and prunes
// recovery points from other threads for the same agent.
func (store *Store) RecordMessageRecoveryPointAndPruneOtherThreads(ctx context.Context, workspaceID string, messageID string, agentID string, threadID string, strategy string, recoveryRef string) ([]MessageRecoveryPoint, error) {
	trimmedWorkspaceID := strings.TrimSpace(workspaceID)
	trimmedMessageID := strings.TrimSpace(messageID)
	trimmedAgentID := strings.TrimSpace(agentID)
	trimmedThreadID := strings.TrimSpace(threadID)
	trimmedStrategy := strings.TrimSpace(strategy)
	trimmedRecoveryRef := strings.TrimSpace(recoveryRef)

	switch {
	case trimmedWorkspaceID == "":
		return nil, fmt.Errorf("workspace id is required")
	case trimmedMessageID == "":
		return nil, fmt.Errorf("message id is required")
	case trimmedAgentID == "":
		return nil, fmt.Errorf("agent id is required")
	case trimmedThreadID == "":
		return nil, fmt.Errorf("thread id is required")
	case trimmedStrategy == "":
		return nil, fmt.Errorf("strategy is required")
	case trimmedRecoveryRef == "":
		return nil, fmt.Errorf("recovery ref is required")
	}

	runtimeValue, err := store.loadAgentRuntimeForRecoveryPoint(ctx, trimmedWorkspaceID, trimmedAgentID)
	if err != nil {
		return nil, err
	}
	capability := rewindcaps.ForRuntime(runtimeValue)
	if !capability.RequiresPreparedPoint {
		return nil, fmt.Errorf("record message recovery point: runtime %q does not support prepared rewind recovery points", runtimeValue)
	}
	if trimmedStrategy != capability.Strategy {
		return nil, fmt.Errorf("record message recovery point: runtime %q requires strategy %q", runtimeValue, capability.Strategy)
	}
	if err := rewindcaps.ValidatePreparedRecoveryPoint(trimmedStrategy, trimmedThreadID, trimmedRecoveryRef); err != nil {
		return nil, fmt.Errorf("record message recovery point: %w", err)
	}

	sourceExists, err := store.deliveredMessageRecoverySourceExists(ctx, trimmedWorkspaceID, trimmedMessageID, trimmedAgentID)
	if err != nil {
		return nil, err
	}
	if !sourceExists {
		return nil, fmt.Errorf("record message recovery point: matching delivered hot message not found")
	}
	if err := WriteMessageRecoveryPointWithExecer(ctx, store.db, trimmedWorkspaceID, trimmedMessageID, trimmedAgentID, trimmedThreadID, trimmedStrategy, trimmedRecoveryRef, time.Now().UTC()); err != nil {
		return nil, fmt.Errorf("record message recovery point: %w", err)
	}

	conversationScopeID, err := store.loadDeliveredConversationScopeID(ctx, trimmedWorkspaceID, trimmedMessageID, trimmedAgentID)
	if err != nil {
		return nil, err
	}

	pruned, err := store.DeleteAgentRecoveryPointsOutsideThread(ctx, trimmedWorkspaceID, trimmedAgentID, conversationScopeID)
	if err != nil {
		return nil, err
	}
	return pruned, nil
}

// DeleteAgentRecoveryPointsOutsideThread deletes recovery points for an agent outside the given thread.
func (store *Store) DeleteAgentRecoveryPointsOutsideThread(ctx context.Context, workspaceID string, agentID string, keepThreadID string) ([]MessageRecoveryPoint, error) {
	trimmedWorkspaceID := strings.TrimSpace(workspaceID)
	trimmedAgentID := strings.TrimSpace(agentID)
	trimmedKeepThreadID := strings.TrimSpace(keepThreadID)

	switch {
	case trimmedWorkspaceID == "":
		return nil, fmt.Errorf("workspace id is required")
	case trimmedAgentID == "":
		return nil, fmt.Errorf("agent id is required")
	case trimmedKeepThreadID == "":
		return nil, fmt.Errorf("keep thread id is required")
	}

	keepScopeID, err := resolveRewindConversationScopeID(ctx, store.db, trimmedWorkspaceID, trimmedKeepThreadID)
	if err != nil {
		return nil, err
	}

	var points []MessageRecoveryPoint
	if err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		rows, err := tx.Query(ctx, `
SELECT mrp.workspace_id, mrp.message_id, mrp.agent_id, mrp.thread_id, mrp.recovery_strategy, mrp.recovery_ref, mrp.prepared_at
FROM message_recovery_points AS mrp
JOIN scope_messages AS m
  ON m.workspace_id = mrp.workspace_id
 AND m.id = mrp.message_id
JOIN scopes AS scope
  ON scope.id = m.scope_id
 AND scope.workspace_id = m.workspace_id
 AND scope.status = 'active'
 AND scope.archived_at IS NULL
WHERE mrp.workspace_id = $1
  AND mrp.agent_id = $2
  AND NOT (
      m.scope_id = $3
      OR EXISTS (
          SELECT 1
          FROM scopes keep_scope
          WHERE keep_scope.workspace_id = mrp.workspace_id
            AND keep_scope.id = $3
            AND keep_scope.type = 'thread'
            AND keep_scope.status = 'active'
            AND keep_scope.archived_at IS NULL
            AND keep_scope.parent_scope_id = m.scope_id
            AND keep_scope.root_message_id = m.id
      )
  )
ORDER BY mrp.prepared_at ASC, mrp.message_id ASC
FOR UPDATE
`, trimmedWorkspaceID, trimmedAgentID, keepScopeID)
		if err != nil {
			return fmt.Errorf("select agent recovery points outside thread: %w", err)
		}
		selectedPoints, err := ScanMessageRecoveryPoints(rows)
		if err != nil {
			return fmt.Errorf("scan selected agent recovery points outside thread: %w", err)
		}
		points = selectedPoints
		if len(points) == 0 {
			return nil
		}

		messageIDs := make([]string, 0, len(points))
		for _, point := range points {
			messageIDs = append(messageIDs, point.MessageID)
		}

		deleteQuery, deleteArgs, err := BuildDeleteRecoveryPointsByMessageIDsQuery(trimmedWorkspaceID, trimmedAgentID, messageIDs)
		if err != nil {
			return fmt.Errorf("delete agent recovery points outside thread: %w", err)
		}
		commandTag, err := tx.Exec(ctx, deleteQuery, deleteArgs...)
		if err != nil {
			return fmt.Errorf("delete agent recovery points outside thread: %w", err)
		}
		if commandTag.RowsAffected() != int64(len(points)) {
			return fmt.Errorf("delete agent recovery points outside thread: unexpected rows affected")
		}

		return nil
	}); err != nil {
		return nil, err
	}
	return points, nil
}

func resolveRewindConversationScopeID(ctx context.Context, queryer Queryer, workspaceID string, conversationID string) (string, error) {
	row := queryer.QueryRow(ctx, `
SELECT id
FROM scopes
WHERE workspace_id = $1
  AND id = $2
  AND status = 'active'
  AND archived_at IS NULL
LIMIT 1
`, strings.TrimSpace(workspaceID), strings.TrimSpace(conversationID))

	var scopeID string
	if err := row.Scan(&scopeID); err != nil {
		if coredb.IsNotFound(err) {
			return "", fmt.Errorf("resolve rewind conversation scope: conversation not found")
		}
		return "", fmt.Errorf("resolve rewind conversation scope: %w", err)
	}
	trimmedScopeID := strings.TrimSpace(scopeID)
	if trimmedScopeID == "" {
		return "", fmt.Errorf("resolve rewind conversation scope: scope id is required")
	}
	return trimmedScopeID, nil
}

// DeleteAllAgentRecoveryPoints deletes all recovery points for an agent in a workspace.
func (store *Store) DeleteAllAgentRecoveryPoints(ctx context.Context, workspaceID string, agentID string) ([]MessageRecoveryPoint, error) {
	trimmedWorkspaceID := strings.TrimSpace(workspaceID)
	trimmedAgentID := strings.TrimSpace(agentID)

	switch {
	case trimmedWorkspaceID == "":
		return nil, fmt.Errorf("workspace id is required")
	case trimmedAgentID == "":
		return nil, fmt.Errorf("agent id is required")
	}

	var points []MessageRecoveryPoint
	if err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		rows, err := tx.Query(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 agent_id = $2
ORDER BY prepared_at ASC, message_id ASC
FOR UPDATE
`, trimmedWorkspaceID, trimmedAgentID)
		if err != nil {
			return fmt.Errorf("select all agent recovery points: %w", err)
		}
		selectedPoints, err := ScanMessageRecoveryPoints(rows)
		if err != nil {
			return fmt.Errorf("scan selected agent recovery points: %w", err)
		}
		points = selectedPoints
		if len(points) == 0 {
			return nil
		}

		commandTag, err := tx.Exec(ctx, `
DELETE FROM message_recovery_points
WHERE workspace_id = $1
  AND agent_id = $2
`, trimmedWorkspaceID, trimmedAgentID)
		if err != nil {
			return fmt.Errorf("delete all agent recovery points: %w", err)
		}
		if commandTag.RowsAffected() != int64(len(points)) {
			return fmt.Errorf("delete all agent recovery points: unexpected rows affected")
		}

		return nil
	}); err != nil {
		return nil, err
	}
	return points, nil
}

// ListAllAgentRecoveryPoints lists all recovery points for an agent in a workspace.
func (store *Store) ListAllAgentRecoveryPoints(ctx context.Context, workspaceID string, agentID string) ([]MessageRecoveryPoint, error) {
	trimmedWorkspaceID := strings.TrimSpace(workspaceID)
	trimmedAgentID := strings.TrimSpace(agentID)

	switch {
	case trimmedWorkspaceID == "":
		return nil, fmt.Errorf("workspace id is required")
	case trimmedAgentID == "":
		return nil, fmt.Errorf("agent id is required")
	}

	rows, err := store.db.Query(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 agent_id = $2
ORDER BY prepared_at ASC
`, trimmedWorkspaceID, trimmedAgentID)
	if err != nil {
		return nil, fmt.Errorf("list all agent recovery points: %w", err)
	}
	points, err := ScanMessageRecoveryPoints(rows)
	if err != nil {
		return nil, fmt.Errorf("scan listed agent recovery points: %w", err)
	}
	return points, nil
}

// DeleteAllAgentRecoveryPointsForWorkspace authorizes and deletes all recovery points for an agent.
func (store *Store) DeleteAllAgentRecoveryPointsForWorkspace(ctx context.Context, userID string, workspaceID string, agentID string) ([]MessageRecoveryPoint, error) {
	if _, err := store.requireAgentOperation(ctx, userID, workspaceID, agentID); err != nil {
		return nil, err
	}
	return store.DeleteAllAgentRecoveryPoints(ctx, workspaceID, agentID)
}

// ListAllAgentRecoveryPointsForWorkspace authorizes and lists all recovery points for an agent.
func (store *Store) ListAllAgentRecoveryPointsForWorkspace(ctx context.Context, userID string, workspaceID string, agentID string) ([]MessageRecoveryPoint, error) {
	if _, err := store.requireAgentOperation(ctx, userID, workspaceID, agentID); err != nil {
		return nil, err
	}
	return store.ListAllAgentRecoveryPoints(ctx, workspaceID, agentID)
}

func (store *Store) loadAgentRuntimeForRecoveryPoint(ctx context.Context, workspaceID string, agentID string) (string, error) {
	var runtimeValue string
	err := store.db.QueryRow(ctx, `
SELECT runtime
FROM agents
WHERE workspace_id = $1 AND id = $2
`, workspaceID, agentID).Scan(&runtimeValue)
	if err != nil {
		if coredb.IsNotFound(err) {
			return "", fmt.Errorf("record message recovery point: agent not found")
		}
		return "", fmt.Errorf("load agent runtime for recovery point: %w", err)
	}
	return strings.TrimSpace(runtimeValue), nil
}

func (store *Store) deliveredMessageRecoverySourceExists(ctx context.Context, workspaceID string, messageID string, agentID string) (bool, error) {
	var exists bool
	err := store.db.QueryRow(ctx, `
SELECT EXISTS (
    SELECT 1
    FROM agents a
    JOIN scope_messages m
      ON m.workspace_id = a.workspace_id
     AND m.id = $2
    JOIN scopes scope
      ON scope.id = m.scope_id
     AND scope.workspace_id = m.workspace_id
     AND scope.status = 'active'
     AND scope.archived_at IS NULL
    WHERE a.id = $3
      AND a.workspace_id = $1
      AND m.workspace_id = $1
      AND (
          m.author_member_id = a.member_id
          OR EXISTS (
              SELECT 1
              FROM scope_memberships sm
              WHERE sm.scope_id = m.scope_id
                AND sm.member_id = a.member_id
                AND sm.removed_at IS NULL
                AND sm.effective_from_seq <= m.seq
                AND (sm.effective_to_seq IS NULL OR sm.effective_to_seq >= m.seq)
          )
      )
)
`, workspaceID, messageID, agentID).Scan(&exists)
	if err != nil {
		return false, fmt.Errorf("load delivered message recovery source: %w", err)
	}
	if exists {
		return true, nil
	}
	_, wakeDelivered, err := store.loadDeliveredWakeConversationScopeID(ctx, workspaceID, messageID, agentID)
	if err != nil {
		return false, err
	}
	return wakeDelivered, nil
}

func (store *Store) loadDeliveredConversationScopeID(ctx context.Context, workspaceID string, messageID string, agentID string) (string, error) {
	if scopeID, ok, err := store.loadDeliveredWakeConversationScopeID(ctx, workspaceID, messageID, agentID); ok || err != nil {
		return scopeID, err
	}

	row := store.db.QueryRow(ctx, `
SELECT DISTINCT m.scope_id
FROM agents a
JOIN scope_messages m
  ON m.workspace_id = a.workspace_id
 AND m.id = $2
JOIN scopes scope
  ON scope.id = m.scope_id
 AND scope.workspace_id = m.workspace_id
 AND scope.status = 'active'
 AND scope.archived_at IS NULL
WHERE a.workspace_id = $1
  AND a.id = $3
  AND (
      m.author_member_id = a.member_id
      OR EXISTS (
          SELECT 1
          FROM scope_memberships sm
          WHERE sm.scope_id = m.scope_id
            AND sm.member_id = a.member_id
            AND sm.removed_at IS NULL
            AND sm.effective_from_seq <= m.seq
            AND (sm.effective_to_seq IS NULL OR sm.effective_to_seq >= m.seq)
      )
  )
LIMIT 1
`, strings.TrimSpace(workspaceID), strings.TrimSpace(messageID), strings.TrimSpace(agentID))

	var scopeID string
	if err := row.Scan(&scopeID); err != nil {
		if coredb.IsNotFound(err) {
			return "", fmt.Errorf("load delivered conversation scope: matching delivered hot message not found")
		}
		return "", fmt.Errorf("load delivered conversation scope: %w", err)
	}
	trimmedScopeID := strings.TrimSpace(scopeID)
	if trimmedScopeID == "" {
		return "", fmt.Errorf("load delivered conversation scope: scope id is required")
	}
	return trimmedScopeID, nil
}

type deliveredWakePayloadRow struct {
	ScopeID string `db:"scope_id"`
	Payload []byte `db:"payload"`
}

func (store *Store) loadDeliveredWakeConversationScopeID(ctx context.Context, workspaceID string, messageID string, agentID string) (string, bool, error) {
	trimmedWorkspaceID := strings.TrimSpace(workspaceID)
	trimmedMessageID := strings.TrimSpace(messageID)
	trimmedAgentID := strings.TrimSpace(agentID)
	if trimmedWorkspaceID == "" || trimmedMessageID == "" || trimmedAgentID == "" {
		return "", false, nil
	}

	var rows []deliveredWakePayloadRow
	if err := coredb.Select(ctx, store.db, &rows, `
SELECT
    sfo.scope_id AS scope_id,
    COALESCE(CAST(sfo.payload AS CHAR), '') AS payload
FROM scope_fanout_outbox sfo
JOIN scope_messages m
  ON m.workspace_id = sfo.workspace_id
 AND m.id = sfo.message_id
JOIN scopes scope
  ON scope.id = m.scope_id
 AND scope.workspace_id = m.workspace_id
 AND scope.status = 'active'
 AND scope.archived_at IS NULL
WHERE sfo.workspace_id = $1
  AND sfo.message_id = $2
  AND sfo.fanout_type = 'scope_fanout'
  AND sfo.status = 'delivered'
  AND sfo.suppressed_at IS NULL
  AND m.workspace_id = $1
  AND `+message.MessageNotRewoundSQL("m")+`
ORDER BY sfo.delivered_at ASC, sfo.id ASC
`, trimmedWorkspaceID, trimmedMessageID); err != nil {
		return "", false, fmt.Errorf("load delivered wake recovery source: %w", err)
	}

	for _, row := range rows {
		wakeRecipients, err := wakeRecipientsFromScopeFanoutPayload(row.Payload)
		if err != nil {
			return "", false, fmt.Errorf("load delivered wake recovery source for %s: %w", trimmedMessageID, err)
		}
		if !wakeRecipientsIncludeAgent(wakeRecipients, trimmedAgentID) {
			continue
		}
		conversationScopeID, err := deliveredWakePayloadConversationScopeID(row.Payload, row.ScopeID)
		if err != nil {
			return "", false, fmt.Errorf("load delivered wake conversation scope for %s: %w", trimmedMessageID, err)
		}
		resolvedScopeID, err := resolveRewindConversationScopeID(ctx, store.db, trimmedWorkspaceID, conversationScopeID)
		if err != nil {
			return "", false, err
		}
		return resolvedScopeID, true, nil
	}
	return "", false, nil
}

func wakeRecipientsIncludeAgent(recipients []message.AgentRecipient, agentID string) bool {
	trimmedAgentID := strings.TrimSpace(agentID)
	if trimmedAgentID == "" {
		return false
	}
	for _, recipient := range recipients {
		if strings.TrimSpace(recipient.AgentID) == trimmedAgentID {
			return true
		}
	}
	return false
}

func deliveredWakePayloadConversationScopeID(payload []byte, fallbackScopeID string) (string, error) {
	if len(payload) > 0 {
		if fanoutPayload, err := message.UnmarshalScopeFanoutPayload(payload); err == nil {
			if scopeID := firstNonEmptyString(
				fanoutPayload.Delivery.ThreadID,
				fanoutPayload.Delivery.Message.TargetKey,
				fanoutPayload.AppMessage.ChannelID,
			); scopeID != "" {
				return scopeID, nil
			}
		}
		if intentPayload, err := message.UnmarshalScopeDeliveryIntentPayload(payload); err == nil {
			if scopeID := strings.TrimSpace(intentPayload.Message.TargetKey); scopeID != "" {
				return scopeID, nil
			}
		}
	}
	if scopeID := strings.TrimSpace(fallbackScopeID); scopeID != "" {
		return scopeID, nil
	}
	return "", fmt.Errorf("delivered wake conversation scope id is required")
}

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