package message

import (
	"context"
	"fmt"
	"sort"
	"strings"

	"github.com/tidbcloud/AgentTeam/internal/core/domain"
	"github.com/tidbcloud/AgentTeam/internal/core/routing"
	"github.com/tidbcloud/AgentTeam/internal/coredb"
)

const (
	AgentDeliveryReasonDMPeer                 = "dm_peer"
	AgentDeliveryReasonDirectMention          = "direct_mention"
	AgentDeliveryReasonTeamMention            = "team_mention"
	AgentDeliveryReasonThreadParticipant      = "thread_participant"
	AgentDeliveryReasonChannelBroadcastAll    = "channel_broadcast_all"
	AgentDeliveryReasonChannelBroadcastOnline = "channel_broadcast_online"
	AgentDeliveryReasonTaskCommandNotify      = "task_command_notify"
	AgentDeliveryReasonOwnerQuote             = "owner_quote"
	AgentDeliveryReasonOwnerPrivateContext    = "owner_private_context"
	AgentDeliveryReasonReaction               = "reaction"
	AgentDeliveryReasonSpaceDefault           = "space_default"
	AgentDeliveryReasonTaskWatch              = "task_watch"
	AgentDeliveryReasonTaskAssignee           = "task_assignee"
)

const (
	SpaceDeliveryResultWake   = "delivered_wake"
	SpaceDeliveryResultNoWake = "delivered_no_wake"
)

// MaxScopeDirectMentionTokens bounds per-message direct @mention routing.
const MaxScopeDirectMentionTokens = 32

type ChannelBroadcastMode string

const (
	ChannelBroadcastModeNone   ChannelBroadcastMode = ""
	ChannelBroadcastModeAll    ChannelBroadcastMode = "all"
	ChannelBroadcastModeOnline ChannelBroadcastMode = "online"
)

type AgentRecipient struct {
	AgentID         string
	MachineID       string
	AgentMemberID   string
	AutoWakeEnabled bool
	DeliveryReason  string
	MentionToken    string
	TeamContexts    []TeamDeliveryContext
}

type TeamDeliveryContext struct {
	TeamID                    string
	TeamHandle                string
	TeamDisplayName           string
	TeamDescription           string
	TeamMemberID              string
	Role                      string
	RoleLabel                 string
	RoleOverlayPromptSnapshot string
	RoleOverlayVersion        int64
	MemberRoster              []TeamDeliveryRosterMember
	// Workflow marks a Workflow-specific minimal context (frozen member snapshot,
	// empty roster). It must be rendered by the dedicated Workflow formatter and
	// never through the ordinary @Team roster/leader-update formatter.
	Workflow bool
}

type TeamDeliveryRosterMember struct {
	TeamMemberID      string
	ActorKind         string
	ActorID           string
	DisplayName       string
	Handle            string
	Role              string
	RoleLabel         string
	RoleOverlayPrompt string
	WakeOnTeamMention bool
	DeliveryKind      string
	Self              bool
}

type AgentRoutingPlan struct {
	DurableRecipients []AgentRecipient
	WakeRecipients    []AgentRecipient
}

type ResolvedMentionLink = routing.ResolvedMentionLink

func ParseMentionTokens(content string) []string {
	return routing.ParseMentionTokens(content)
}

func ParseDirectMentionTokens(content string) []string {
	return routing.ParseDirectMentionTokens(content)
}

func ParseDirectMentionTokensLimit(content string, limit int) []string {
	return routing.ParseDirectMentionTokensLimit(content, limit)
}

func ParseResolvedMentionLinks(content string) []ResolvedMentionLink {
	return routing.ParseResolvedMentionLinks(content)
}

func ParseControlledBroadcastMode(content string) (ChannelBroadcastMode, error) {
	mode, err := routing.ParseControlledBroadcastMode(content)
	if err != nil {
		return ChannelBroadcastModeNone, badRequest(err.Error())
	}
	return ChannelBroadcastMode(mode), nil
}

func ContainsControlledBroadcastToken(content string) bool {
	return routing.ContainsControlledBroadcastToken(content)
}

func FilterAgentRecipientsOwnedByUser(ctx context.Context, tx coredb.Tx, userID string, recipients []AgentRecipient) ([]AgentRecipient, error) {
	if len(recipients) == 0 {
		return nil, nil
	}

	memberIDs := make([]string, 0, len(recipients))
	seen := make(map[string]struct{}, len(recipients))
	for _, recipient := range recipients {
		memberID := strings.TrimSpace(recipient.AgentMemberID)
		if memberID == "" {
			continue
		}
		if _, ok := seen[memberID]; ok {
			continue
		}
		seen[memberID] = struct{}{}
		memberIDs = append(memberIDs, memberID)
	}
	if len(memberIDs) == 0 {
		return nil, nil
	}

	requestedMemberIDsCTE, requestedMemberIDArgs, _, err := coredb.BuildStringValuesCTE("requested_member_ids", "member_id", memberIDs, 2)
	if err != nil {
		return nil, err
	}

	query := fmt.Sprintf(`
WITH %s
SELECT a.member_id
FROM agents a
JOIN machines mac
  ON mac.workspace_id = a.workspace_id
 AND mac.id = a.machine_id
JOIN requested_member_ids ON requested_member_ids.member_id = a.member_id
WHERE mac.created_by_user_id = $1
  AND mac.status = 'active'
  AND a.status <> 'archived'
`, requestedMemberIDsCTE)
	rows, err := tx.Query(ctx, query, append([]any{strings.TrimSpace(userID)}, requestedMemberIDArgs...)...)
	if err != nil {
		return nil, fmt.Errorf("query owned agent recipients: %w", err)
	}
	defer rows.Close()

	ownedMemberIDs := make(map[string]struct{}, len(memberIDs))
	for rows.Next() {
		var memberID string
		if err := rows.Scan(&memberID); err != nil {
			return nil, fmt.Errorf("scan owned agent recipient: %w", err)
		}
		ownedMemberIDs[strings.TrimSpace(memberID)] = struct{}{}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate owned agent recipients: %w", err)
	}

	filtered := make([]AgentRecipient, 0, len(recipients))
	for _, recipient := range recipients {
		if _, ok := ownedMemberIDs[strings.TrimSpace(recipient.AgentMemberID)]; !ok {
			continue
		}
		filtered = append(filtered, recipient)
	}
	return filtered, nil
}

func ValidateReplyThreadRoutingContent(content string) error {
	if ContainsControlledBroadcastToken(content) {
		return badRequest("threads do not support @all or @online")
	}
	return nil
}

func MergeAgentRecipients(recipients ...[]AgentRecipient) []AgentRecipient {
	merged := make(map[string]AgentRecipient)
	for index, group := range recipients {
		for _, recipient := range group {
			memberID := strings.TrimSpace(recipient.AgentMemberID)
			if memberID == "" {
				continue
			}
			if _, exists := merged[memberID]; exists && index == 0 {
				existing := merged[memberID]
				existing.TeamContexts = mergeTeamDeliveryContexts(existing.TeamContexts, recipient.TeamContexts)
				merged[memberID] = existing
				continue
			}
			if existing, exists := merged[memberID]; exists {
				recipient = mergeAgentRecipient(existing, recipient)
			}
			merged[memberID] = recipient
		}
	}
	if len(merged) == 0 {
		return nil
	}

	memberIDs := make([]string, 0, len(merged))
	for memberID := range merged {
		memberIDs = append(memberIDs, memberID)
	}
	sort.Strings(memberIDs)

	ordered := make([]AgentRecipient, 0, len(memberIDs))
	for _, memberID := range memberIDs {
		ordered = append(ordered, merged[memberID])
	}
	return ordered
}

func mergeAgentRecipient(existing AgentRecipient, incoming AgentRecipient) AgentRecipient {
	contexts := mergeTeamDeliveryContexts(existing.TeamContexts, incoming.TeamContexts)
	if existing.DeliveryReason == AgentDeliveryReasonDirectMention && incoming.DeliveryReason == AgentDeliveryReasonTeamMention {
		existing.TeamContexts = contexts
		return existing
	}
	if incoming.DeliveryReason == "" {
		incoming.DeliveryReason = existing.DeliveryReason
	}
	if incoming.MentionToken == "" {
		incoming.MentionToken = existing.MentionToken
	}
	if incoming.AgentID == "" {
		incoming.AgentID = existing.AgentID
	}
	if incoming.MachineID == "" {
		incoming.MachineID = existing.MachineID
	}
	if incoming.AgentMemberID == "" {
		incoming.AgentMemberID = existing.AgentMemberID
	}
	incoming.AutoWakeEnabled = incoming.AutoWakeEnabled || existing.AutoWakeEnabled
	incoming.TeamContexts = contexts
	return incoming
}

func mergeTeamDeliveryContexts(groups ...[]TeamDeliveryContext) []TeamDeliveryContext {
	merged := make(map[string]TeamDeliveryContext)
	for _, group := range groups {
		for _, context := range group {
			key := strings.TrimSpace(context.TeamID) + "\x00" + strings.TrimSpace(context.TeamMemberID)
			if key == "\x00" {
				continue
			}
			if _, ok := merged[key]; ok {
				continue
			}
			context.MemberRoster = append([]TeamDeliveryRosterMember(nil), context.MemberRoster...)
			merged[key] = context
		}
	}
	if len(merged) == 0 {
		return nil
	}
	keys := make([]string, 0, len(merged))
	for key := range merged {
		keys = append(keys, key)
	}
	sort.Strings(keys)
	contexts := make([]TeamDeliveryContext, 0, len(keys))
	for _, key := range keys {
		contexts = append(contexts, merged[key])
	}
	return contexts
}

// FinalizeAgentRoutingPlan normalizes a per-message routing plan before the
// writer persists the fanout payload. Wake recipients are always part of the
// durable recipient set: durable recipients define who may receive/read this
// message through agent delivery surfaces, while wake recipients are the subset
// that should be actively delivered to a daemon now.
func FinalizeAgentRoutingPlan(plan AgentRoutingPlan) AgentRoutingPlan {
	durableRecipients := MergeAgentRecipients(plan.DurableRecipients)
	wakeRecipients := MergeAgentRecipients(plan.WakeRecipients)
	if len(wakeRecipients) == 0 {
		return AgentRoutingPlan{DurableRecipients: durableRecipients}
	}

	durableMemberIDs := make(map[string]struct{}, len(durableRecipients))
	for _, recipient := range durableRecipients {
		memberID := strings.TrimSpace(recipient.AgentMemberID)
		if memberID == "" {
			continue
		}
		durableMemberIDs[memberID] = struct{}{}
	}
	missingDurableRecipients := make([]AgentRecipient, 0)
	for _, recipient := range wakeRecipients {
		memberID := strings.TrimSpace(recipient.AgentMemberID)
		if memberID == "" {
			continue
		}
		if _, ok := durableMemberIDs[memberID]; ok {
			continue
		}
		durableMemberIDs[memberID] = struct{}{}
		missingDurableRecipients = append(missingDurableRecipients, recipient)
	}
	if len(missingDurableRecipients) > 0 {
		durableRecipients = MergeAgentRecipients(durableRecipients, missingDurableRecipients)
	}
	return AgentRoutingPlan{DurableRecipients: durableRecipients, WakeRecipients: wakeRecipients}
}

// FilterAutoWakeAgentRecipients keeps only recipients whose notification
// settings should actively wake a runtime.
func FilterAutoWakeAgentRecipients(recipients []AgentRecipient) []AgentRecipient {
	filtered := make([]AgentRecipient, 0, len(recipients))
	for _, recipient := range recipients {
		if recipient.AutoWakeEnabled {
			filtered = append(filtered, recipient)
		}
	}
	return filtered
}

func loadReplyThreadParticipantIDs(ctx context.Context, queryer coredb.Queryer, threadChannelID string, parentMessageID string) ([]string, error) {
	trimmedThreadChannelID := strings.TrimSpace(threadChannelID)
	row := queryer.QueryRow(ctx, `
SELECT 1
FROM scopes
WHERE id = $1
  AND root_message_id = $2
  AND type = 'thread'
  AND status = 'active'
  AND archived_at IS NULL
LIMIT 1
`, trimmedThreadChannelID, strings.TrimSpace(parentMessageID))
	var exists int
	if err := row.Scan(&exists); err != nil {
		if coredb.IsNotFound(err) {
			return nil, notFound("reply thread not found")
		}
		return nil, fmt.Errorf("load reply thread participants: %w", err)
	}
	participantMap, err := loadReplyThreadParticipantMap(ctx, queryer, []string{trimmedThreadChannelID})
	if err != nil {
		return nil, err
	}
	return participantMap[trimmedThreadChannelID], nil
}

// scopeMentionAuthorityCTE is the single live owner-identity relation used by
// direct and persisted mention routing. An Agent author is never compared to a
// private membership owner as if its Agent member ID were a Human owner ID:
// its owner is recovered only from an active private host-Channel membership,
// then fenced by the active owner membership and machine creator relation.
const scopeMentionAuthorityCTE = `host_scope AS (
    SELECT s.id, s.workspace_id
    FROM scopes s
    JOIN scope_channel_profiles host_profile
      ON host_profile.scope_id = s.id
     AND host_profile.workspace_id = s.workspace_id
    WHERE s.id = $1
      AND s.type = 'channel'
      AND s.status = 'active'
      AND s.archived_at IS NULL
),
author_owner_identity AS (
    SELECT DISTINCT
        CASE
            WHEN author_member.type = 'human' THEN author_member.id
            ELSE author_private.owner_member_id
        END AS owner_member_id
    FROM host_scope
    JOIN members author_member
      ON author_member.workspace_id = host_scope.workspace_id
     AND author_member.id = $2
     AND author_member.status = 'active'
    LEFT JOIN agents author_agent
      ON author_member.type = 'agent'
     AND author_agent.workspace_id = host_scope.workspace_id
     AND author_agent.member_id = author_member.id
    LEFT JOIN machines author_machine
      ON author_machine.workspace_id = author_agent.workspace_id
     AND author_machine.id = author_agent.machine_id
    LEFT JOIN scope_memberships author_private
      ON author_member.type = 'agent'
     AND author_private.scope_id = host_scope.id
     AND author_private.member_id = author_member.id
     AND author_private.visibility = 'private'
     AND author_private.owner_member_id IS NOT NULL
     AND author_private.removed_at IS NULL
     AND author_private.archived_at IS NULL
     AND author_private.effective_to_seq IS NULL
     AND CURRENT_TIMESTAMP(6) >= COALESCE(author_private.authority_from_at, author_private.created_at)
     AND (
          author_private.authority_to_at IS NULL
          OR CURRENT_TIMESTAMP(6) < author_private.authority_to_at
     )
    WHERE (
        author_member.type = 'human'
        AND EXISTS (
            SELECT 1
            FROM scope_memberships author_host_membership
            WHERE author_host_membership.scope_id = host_scope.id
              AND author_host_membership.member_id = author_member.id
              AND author_host_membership.visibility = 'public'
              AND author_host_membership.owner_member_id IS NULL
              AND author_host_membership.removed_at IS NULL
              AND author_host_membership.archived_at IS NULL
              AND author_host_membership.effective_to_seq IS NULL
              AND CURRENT_TIMESTAMP(6) >= COALESCE(author_host_membership.authority_from_at, author_host_membership.created_at)
              AND (
                   author_host_membership.authority_to_at IS NULL
                   OR CURRENT_TIMESTAMP(6) < author_host_membership.authority_to_at
              )
        )
    ) OR (
        author_member.type = 'agent'
        AND author_agent.status <> 'archived'
        AND author_machine.status = 'active'
        AND author_private.owner_member_id IS NOT NULL
        AND EXISTS (
            SELECT 1
            FROM members author_owner_member
            JOIN scope_memberships author_owner_membership
              ON author_owner_membership.scope_id = host_scope.id
             AND author_owner_membership.member_id = author_owner_member.id
             AND author_owner_membership.visibility = 'public'
             AND author_owner_membership.owner_member_id IS NULL
             AND author_owner_membership.removed_at IS NULL
             AND author_owner_membership.archived_at IS NULL
             AND author_owner_membership.effective_to_seq IS NULL
             AND CURRENT_TIMESTAMP(6) >= COALESCE(author_owner_membership.authority_from_at, author_owner_membership.created_at)
             AND (
                  author_owner_membership.authority_to_at IS NULL
                  OR CURRENT_TIMESTAMP(6) < author_owner_membership.authority_to_at
             )
            WHERE author_owner_member.workspace_id = host_scope.workspace_id
              AND author_owner_member.id = author_private.owner_member_id
              AND author_owner_member.type = 'human'
              AND author_owner_member.status = 'active'
              AND author_owner_member.user_id IS NOT NULL
              AND author_machine.created_by_user_id = author_owner_member.user_id
        )
    )
)`

func LoadScopeMentionedAgentRecipients(ctx context.Context, tx coredb.Tx, scopeID string, authorMemberID string, content string) ([]AgentRecipient, error) {
	return LoadScopeMentionedAgentRecipientsForTokens(ctx, tx, scopeID, authorMemberID, ParseDirectMentionTokensLimit(content, MaxScopeDirectMentionTokens))
}

func LoadScopeMentionedAgentRecipientsForTokens(ctx context.Context, tx coredb.Tx, scopeID string, authorMemberID string, tokens []string) ([]AgentRecipient, error) {
	tokens = normalizeNonEmptyStrings(tokens)
	if len(tokens) > MaxScopeDirectMentionTokens {
		tokens = tokens[:MaxScopeDirectMentionTokens]
	}
	if len(tokens) == 0 {
		return nil, nil
	}
	mentionedTokensCTE, mentionedTokenArgs, _, err := coredb.BuildStringValuesCTE("mentioned_tokens", "token", tokens, 3)
	if err != nil {
		return nil, err
	}
	query := fmt.Sprintf(`
WITH %s,
%s
SELECT
    mentioned_tokens.token,
    a.id,
    a.machine_id,
    a.member_id,
    a.name,
    CASE WHEN sns.state = 'auto-wake' THEN 1 ELSE 0 END AS auto_wake_enabled
FROM host_scope
LEFT JOIN author_owner_identity
  ON TRUE
JOIN agents a
  ON a.workspace_id = host_scope.workspace_id
JOIN members target_member
  ON target_member.workspace_id = host_scope.workspace_id
 AND target_member.id = a.member_id
JOIN machines mac
  ON mac.workspace_id = a.workspace_id
 AND mac.id = a.machine_id
JOIN scope_memberships sm
  ON sm.scope_id = host_scope.id
 AND sm.member_id = a.member_id
 AND sm.removed_at IS NULL
 AND sm.archived_at IS NULL
 AND sm.effective_to_seq IS NULL
 AND CURRENT_TIMESTAMP(6) >= COALESCE(sm.authority_from_at, sm.created_at)
 AND (
      sm.authority_to_at IS NULL
      OR CURRENT_TIMESTAMP(6) < sm.authority_to_at
 )
LEFT JOIN members owner_member
  ON owner_member.workspace_id = host_scope.workspace_id
 AND owner_member.id = sm.owner_member_id
LEFT JOIN scope_memberships owner_host_membership
  ON sm.visibility = 'private'
 AND owner_host_membership.scope_id = host_scope.id
 AND owner_host_membership.member_id = owner_member.id
 AND owner_host_membership.visibility = 'public'
 AND owner_host_membership.owner_member_id IS NULL
 AND owner_host_membership.removed_at IS NULL
 AND owner_host_membership.archived_at IS NULL
 AND owner_host_membership.effective_to_seq IS NULL
 AND CURRENT_TIMESTAMP(6) >= COALESCE(owner_host_membership.authority_from_at, owner_host_membership.created_at)
 AND (
      owner_host_membership.authority_to_at IS NULL
      OR CURRENT_TIMESTAMP(6) < owner_host_membership.authority_to_at
 )
LEFT JOIN scope_notification_settings sns
  ON sns.scope_id = sm.scope_id
 AND sns.member_id = sm.member_id
JOIN mentioned_tokens
  ON (
      mentioned_tokens.token = LOWER(a.name)
      OR mentioned_tokens.token = LOWER(a.member_id)
      OR mentioned_tokens.token LIKE CONCAT(LOWER(NULLIF(a.name, '')), '%%')
      OR mentioned_tokens.token LIKE CONCAT(LOWER(NULLIF(a.member_id, '')), '%%')
  )
WHERE a.member_id <> $2
  AND target_member.type = 'agent'
  AND target_member.status = 'active'
  AND a.status <> 'archived'
  AND mac.status = 'active'
  AND (
      sm.visibility = 'public'
      OR (
          sm.visibility = 'private'
          AND sm.owner_member_id IS NOT NULL
          AND sm.owner_member_id = author_owner_identity.owner_member_id
          AND owner_member.type = 'human'
          AND owner_member.status = 'active'
          AND owner_member.user_id IS NOT NULL
          AND mac.created_by_user_id = owner_member.user_id
          AND owner_host_membership.member_id IS NOT NULL
      )
  )
ORDER BY mentioned_tokens.token ASC, a.name ASC, a.member_id ASC
`, mentionedTokensCTE, scopeMentionAuthorityCTE)
	args := append([]any{strings.TrimSpace(scopeID), strings.TrimSpace(authorMemberID)}, mentionedTokenArgs...)
	rows, err := tx.Query(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("load scope mentioned agent recipients: %w", err)
	}
	defer rows.Close()

	type mentionCandidate struct {
		recipient   AgentRecipient
		matchLength int
		rawToken    string
	}
	candidates := make([]mentionCandidate, 0, len(tokens))
	longestMatchByRawToken := make(map[string]int, len(tokens))
	for rows.Next() {
		var rawToken string
		var recipient AgentRecipient
		var agentName string
		if err := rows.Scan(&rawToken, &recipient.AgentID, &recipient.MachineID, &recipient.AgentMemberID, &agentName, &recipient.AutoWakeEnabled); err != nil {
			return nil, fmt.Errorf("scan scope mentioned agent recipient: %w", err)
		}
		mentionToken, ok := matchedMentionTargetToken(rawToken, []string{
			agentName,
			recipient.AgentMemberID,
		})
		if !ok {
			continue
		}
		normalizedRawToken := strings.ToLower(strings.TrimSpace(rawToken))
		matchLength := len(mentionToken)
		if current := longestMatchByRawToken[normalizedRawToken]; matchLength > current {
			longestMatchByRawToken[normalizedRawToken] = matchLength
		}
		recipient.DeliveryReason = AgentDeliveryReasonDirectMention
		recipient.MentionToken = mentionToken
		candidates = append(candidates, mentionCandidate{
			recipient:   recipient,
			matchLength: matchLength,
			rawToken:    normalizedRawToken,
		})
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate scope mentioned agent recipients: %w", err)
	}

	recipients := make([]AgentRecipient, 0, len(candidates))
	seenMemberIDs := make(map[string]struct{}, len(candidates))
	for _, candidate := range candidates {
		if candidate.matchLength != longestMatchByRawToken[candidate.rawToken] {
			continue
		}
		memberID := strings.TrimSpace(candidate.recipient.AgentMemberID)
		if memberID == "" {
			continue
		}
		if _, ok := seenMemberIDs[memberID]; ok {
			continue
		}
		seenMemberIDs[memberID] = struct{}{}
		recipients = append(recipients, candidate.recipient)
	}
	return recipients, nil
}

func LoadScopeMentionedAgentRecipientsWithMetadata(ctx context.Context, tx coredb.Tx, scopeID string, authorMemberID string, content string, metadata map[string]any) ([]AgentRecipient, error) {
	tokens := ParseDirectMentionTokens(content)
	_, tokenSet := normalizedDirectMentionTokens(content)
	persistedMentions := persistedAgentMentionsForTokens(metadata, tokenSet)

	boundTokens := make([]string, 0, len(persistedMentions))
	for _, mention := range persistedMentions {
		boundTokens = append(boundTokens, mention.Token)
	}
	liveTokens := directMentionTokensExcluding(tokens, boundTokens)

	contentRecipients, err := LoadScopeMentionedAgentRecipientsForTokens(ctx, tx, scopeID, authorMemberID, liveTokens)
	if err != nil {
		return nil, err
	}
	metadataRecipients, err := loadScopePersistedMentionedAgentRecipients(ctx, tx, scopeID, authorMemberID, persistedMentions)
	if err != nil {
		return nil, err
	}
	return MergeAgentRecipients(contentRecipients, metadataRecipients), nil
}

// LoadScopePersistedMentionedAgentRecipientsWithMetadata resolves direct @mentions
// only through persisted mention metadata. It returns ok=false when any live
// token would need current-name routing to resolve, so callers can fail closed
// for old replay paths that no longer have their original fanout payload.
func LoadScopePersistedMentionedAgentRecipientsWithMetadata(ctx context.Context, tx coredb.Tx, scopeID string, authorMemberID string, content string, metadata map[string]any) ([]AgentRecipient, bool, error) {
	directTokens, tokenSet := normalizedDirectMentionTokens(content)
	if len(directTokens) == 0 {
		return nil, false, nil
	}
	persistedByToken := persistedMentionsByDirectToken(persistedAgentMentionsForTokens(metadata, tokenSet), directTokens)
	if len(persistedByToken) != len(directTokens) {
		return nil, false, nil
	}

	mentions := make([]PersistedMention, 0, len(persistedByToken))
	expectedAgentIDs := make(map[string]struct{}, len(persistedByToken))
	for _, directToken := range directTokens {
		persistedMentions := persistedByToken[directToken]
		if len(persistedMentions) == 0 {
			return nil, false, nil
		}
		for _, mention := range persistedMentions {
			agentID := strings.TrimSpace(mention.ID)
			if agentID == "" {
				return nil, false, nil
			}
			if _, ok := expectedAgentIDs[agentID]; ok {
				continue
			}
			expectedAgentIDs[agentID] = struct{}{}
			mentions = append(mentions, mention)
		}
	}

	recipients, err := loadScopePersistedMentionedAgentRecipients(ctx, tx, scopeID, authorMemberID, mentions)
	if err != nil {
		return nil, false, err
	}
	resolvedAgentIDs := make(map[string]struct{}, len(recipients))
	for _, recipient := range recipients {
		agentID := strings.TrimSpace(recipient.AgentID)
		if agentID != "" {
			resolvedAgentIDs[agentID] = struct{}{}
		}
	}
	if len(resolvedAgentIDs) != len(expectedAgentIDs) {
		return nil, false, nil
	}
	return MergeAgentRecipients(recipients), true, nil
}

func directMentionTokensExcluding(tokens []string, excluded []string) []string {
	if len(tokens) == 0 {
		return nil
	}
	if len(excluded) == 0 {
		return append([]string(nil), tokens...)
	}
	excludedSet := make(map[string]struct{}, len(excluded))
	for _, token := range excluded {
		normalized := strings.ToLower(strings.TrimSpace(token))
		if normalized != "" {
			excludedSet[normalized] = struct{}{}
		}
	}
	filtered := make([]string, 0, len(tokens))
	for _, token := range tokens {
		normalized := strings.ToLower(strings.TrimSpace(token))
		if normalized == "" {
			continue
		}
		if _, ok := excludedSet[normalized]; ok {
			continue
		}
		excludedByTarget := false
		for excludedToken := range excludedSet {
			if MentionTokenMatchesTarget(normalized, excludedToken) {
				excludedByTarget = true
				break
			}
		}
		if excludedByTarget {
			continue
		}
		filtered = append(filtered, normalized)
	}
	return filtered
}

func loadScopePersistedMentionedAgentRecipients(ctx context.Context, tx coredb.Tx, scopeID string, authorMemberID string, mentions []PersistedMention) ([]AgentRecipient, error) {
	if len(mentions) == 0 {
		return nil, nil
	}

	agentIDs := make([]string, 0, len(mentions))
	for _, mention := range mentions {
		agentIDs = append(agentIDs, mention.ID)
	}
	requestedAgentsCTE, requestedAgentArgs, _, err := coredb.BuildStringValuesCTE("requested_agents", "agent_id", normalizeNonEmptyStrings(agentIDs), 3)
	if err != nil {
		return nil, err
	}
	rows, err := tx.Query(ctx, fmt.Sprintf(`
WITH %s,
%s
SELECT
    a.id,
    a.machine_id,
    a.member_id,
    CASE WHEN sns.state = 'auto-wake' THEN 1 ELSE 0 END AS auto_wake_enabled
FROM requested_agents
JOIN host_scope
  ON TRUE
LEFT JOIN author_owner_identity
  ON TRUE
JOIN agents a
  ON a.workspace_id = host_scope.workspace_id
 AND a.id = requested_agents.agent_id
JOIN members target_member
  ON target_member.workspace_id = host_scope.workspace_id
 AND target_member.id = a.member_id
JOIN machines mac
  ON mac.workspace_id = a.workspace_id
 AND mac.id = a.machine_id
JOIN scope_memberships sm
  ON sm.scope_id = host_scope.id
 AND sm.member_id = a.member_id
 AND sm.removed_at IS NULL
 AND sm.archived_at IS NULL
 AND sm.effective_to_seq IS NULL
 AND CURRENT_TIMESTAMP(6) >= COALESCE(sm.authority_from_at, sm.created_at)
 AND (
     sm.authority_to_at IS NULL
     OR CURRENT_TIMESTAMP(6) < sm.authority_to_at
 )
LEFT JOIN members owner_member
  ON owner_member.workspace_id = host_scope.workspace_id
 AND owner_member.id = sm.owner_member_id
LEFT JOIN scope_memberships owner_host_membership
  ON sm.visibility = 'private'
 AND owner_host_membership.scope_id = host_scope.id
 AND owner_host_membership.member_id = owner_member.id
 AND owner_host_membership.visibility = 'public'
 AND owner_host_membership.owner_member_id IS NULL
 AND owner_host_membership.removed_at IS NULL
 AND owner_host_membership.archived_at IS NULL
 AND owner_host_membership.effective_to_seq IS NULL
 AND CURRENT_TIMESTAMP(6) >= COALESCE(owner_host_membership.authority_from_at, owner_host_membership.created_at)
 AND (
     owner_host_membership.authority_to_at IS NULL
     OR CURRENT_TIMESTAMP(6) < owner_host_membership.authority_to_at
 )
LEFT JOIN scope_notification_settings sns
  ON sns.scope_id = sm.scope_id
 AND sns.member_id = sm.member_id
WHERE a.member_id <> $2
  AND target_member.type = 'agent'
  AND target_member.status = 'active'
  AND a.status <> 'archived'
  AND mac.status = 'active'
  AND (
      sm.visibility = 'public'
      OR (
          sm.visibility = 'private'
          AND sm.owner_member_id IS NOT NULL
          AND sm.owner_member_id = author_owner_identity.owner_member_id
          AND owner_member.type = 'human'
          AND owner_member.status = 'active'
          AND owner_member.user_id IS NOT NULL
          AND mac.created_by_user_id = owner_member.user_id
          AND owner_host_membership.member_id IS NOT NULL
      )
  )
ORDER BY a.name ASC, a.member_id ASC
`, requestedAgentsCTE, scopeMentionAuthorityCTE), append([]any{strings.TrimSpace(scopeID), strings.TrimSpace(authorMemberID)}, requestedAgentArgs...)...)
	if err != nil {
		return nil, fmt.Errorf("load scope persisted mentioned agent recipients: %w", err)
	}
	defer rows.Close()

	recipients := make([]AgentRecipient, 0, len(agentIDs))
	for rows.Next() {
		var recipient AgentRecipient
		if err := rows.Scan(&recipient.AgentID, &recipient.MachineID, &recipient.AgentMemberID, &recipient.AutoWakeEnabled); err != nil {
			return nil, fmt.Errorf("scan scope persisted mentioned agent recipient: %w", err)
		}
		recipient.DeliveryReason = AgentDeliveryReasonDirectMention
		recipients = append(recipients, recipient)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate scope persisted mentioned agent recipients: %w", err)
	}
	return recipients, nil
}

func loadScopeAgentRecipients(ctx context.Context, tx coredb.Tx, scopeID string, authorMemberID string, deliveryReason string) ([]AgentRecipient, []AgentRecipient, error) {
	return loadScopeAgentRecipientsWithOptions(ctx, tx, scopeID, authorMemberID, deliveryReason, false)
}

func loadScopeBroadcastAgentRecipients(ctx context.Context, tx coredb.Tx, scopeID string, authorMemberID string, deliveryReason string) ([]AgentRecipient, []AgentRecipient, error) {
	return loadScopeAgentRecipientsWithOptions(ctx, tx, scopeID, authorMemberID, deliveryReason, true)
}

func loadScopeAgentRecipientsWithOptions(ctx context.Context, tx coredb.Tx, scopeID string, authorMemberID string, deliveryReason string, includePrivateMemberships bool) ([]AgentRecipient, []AgentRecipient, error) {
	rows, err := tx.Query(ctx, `
SELECT
    a.id,
    a.machine_id,
    a.member_id,
    CASE WHEN sns.state = 'auto-wake' THEN 1 ELSE 0 END AS auto_wake_enabled
FROM agents a
JOIN machines mac ON mac.workspace_id = a.workspace_id AND mac.id = a.machine_id
JOIN scope_memberships sm
  ON sm.scope_id = $1
 AND sm.member_id = a.member_id
 AND sm.removed_at IS NULL
 AND (sm.visibility = 'public' OR ($3 AND sm.visibility = 'private'))
LEFT JOIN scope_notification_settings sns
  ON sns.scope_id = sm.scope_id
 AND sns.member_id = sm.member_id
WHERE a.member_id <> $2
  AND a.status <> 'archived'
  AND mac.status = 'active'
ORDER BY a.member_id ASC
`, strings.TrimSpace(scopeID), strings.TrimSpace(authorMemberID), includePrivateMemberships)
	if err != nil {
		return nil, nil, fmt.Errorf("load scope agent recipients: %w", err)
	}
	defer rows.Close()

	allRecipients := make([]AgentRecipient, 0)
	wakeRecipients := make([]AgentRecipient, 0)
	for rows.Next() {
		var recipient AgentRecipient
		if err := rows.Scan(&recipient.AgentID, &recipient.MachineID, &recipient.AgentMemberID, &recipient.AutoWakeEnabled); err != nil {
			return nil, nil, fmt.Errorf("scan scope agent recipient: %w", err)
		}
		recipient.DeliveryReason = deliveryReason
		allRecipients = append(allRecipients, recipient)
		wakeRecipients = append(wakeRecipients, recipient)
	}
	if err := rows.Err(); err != nil {
		return nil, nil, fmt.Errorf("iterate scope agent recipients: %w", err)
	}
	return allRecipients, wakeRecipients, nil
}

func LoadScopeDefaultAgentRecipients(ctx context.Context, tx coredb.Tx, scopeID string, authorMemberID string) ([]AgentRecipient, error) {
	recipients, _, err := loadScopeAgentRecipients(ctx, tx, scopeID, authorMemberID, AgentDeliveryReasonSpaceDefault)
	return recipients, err
}

func LoadScopeThreadParticipantAgentRecipients(ctx context.Context, tx coredb.Tx, scopeID string, authorMemberID string) ([]AgentRecipient, error) {
	recipients, _, err := loadScopeAgentRecipients(ctx, tx, scopeID, authorMemberID, AgentDeliveryReasonThreadParticipant)
	return recipients, err
}

func LoadOwnedScopeBroadcastRoutingPlan(ctx context.Context, tx coredb.Tx, userID string, scopeID string, authorMemberID string, content string) (AgentRoutingPlan, error) {
	broadcastMode, err := ParseControlledBroadcastMode(content)
	if err != nil {
		return AgentRoutingPlan{}, err
	}
	switch broadcastMode {
	case ChannelBroadcastModeAll:
		recipients, wakeRecipients, err := loadScopeBroadcastAgentRecipients(ctx, tx, scopeID, authorMemberID, AgentDeliveryReasonChannelBroadcastAll)
		if err != nil {
			return AgentRoutingPlan{}, err
		}
		ownedRecipients, err := FilterAgentRecipientsOwnedByUser(ctx, tx, userID, recipients)
		if err != nil {
			return AgentRoutingPlan{}, err
		}
		ownedWakeRecipients, err := FilterAgentRecipientsOwnedByUser(ctx, tx, userID, wakeRecipients)
		if err != nil {
			return AgentRoutingPlan{}, err
		}
		return AgentRoutingPlan{DurableRecipients: ownedRecipients, WakeRecipients: ownedWakeRecipients}, nil
	case ChannelBroadcastModeOnline:
		_, onlineRecipients, err := loadScopeBroadcastAgentRecipients(ctx, tx, scopeID, authorMemberID, AgentDeliveryReasonChannelBroadcastOnline)
		if err != nil {
			return AgentRoutingPlan{}, err
		}
		ownedOnlineRecipients, err := FilterAgentRecipientsOwnedByUser(ctx, tx, userID, onlineRecipients)
		if err != nil {
			return AgentRoutingPlan{}, err
		}
		return AgentRoutingPlan{DurableRecipients: ownedOnlineRecipients, WakeRecipients: ownedOnlineRecipients}, nil
	default:
		return AgentRoutingPlan{}, nil
	}
}

func LoadScopeAgentRoutingPlan(ctx context.Context, tx coredb.Tx, authorMemberID string, target ThreadAccess, content string, allowControlledBroadcast bool) (AgentRoutingPlan, error) {
	scopeID := scopeIDForThreadTarget(target)
	if scopeID == "" {
		return AgentRoutingPlan{}, fmt.Errorf("scope id is required for message routing")
	}
	if target.ReplyThread != nil {
		if err := ValidateReplyThreadRoutingContent(content); err != nil {
			return AgentRoutingPlan{}, err
		}
	}
	if target.ThreadType == domain.ThreadTypeDM {
		recipients, err := loadScopeDMPeerAgentRecipients(ctx, tx, scopeID, authorMemberID)
		if err != nil {
			return AgentRoutingPlan{}, err
		}
		return AgentRoutingPlan{DurableRecipients: recipients, WakeRecipients: recipients}, nil
	}
	if allowControlledBroadcast && target.ReplyThread == nil {
		broadcastMode, err := ParseControlledBroadcastMode(content)
		if err != nil {
			return AgentRoutingPlan{}, err
		}
		switch broadcastMode {
		case ChannelBroadcastModeAll:
			recipients, wakeRecipients, err := loadScopeAgentRecipients(ctx, tx, scopeID, authorMemberID, AgentDeliveryReasonChannelBroadcastAll)
			if err != nil {
				return AgentRoutingPlan{}, err
			}
			return AgentRoutingPlan{DurableRecipients: recipients, WakeRecipients: wakeRecipients}, nil
		case ChannelBroadcastModeOnline:
			_, onlineRecipients, err := loadScopeAgentRecipients(ctx, tx, scopeID, authorMemberID, AgentDeliveryReasonChannelBroadcastOnline)
			if err != nil {
				return AgentRoutingPlan{}, err
			}
			return AgentRoutingPlan{DurableRecipients: onlineRecipients, WakeRecipients: onlineRecipients}, nil
		}
	}
	mentionScopeID := scopeID
	if target.ReplyThread != nil && strings.TrimSpace(target.ReplyThread.ParentChannelID) != "" {
		mentionScopeID = strings.TrimSpace(target.ReplyThread.ParentChannelID)
	}
	mentionedRecipients, err := LoadScopeMentionedAgentRecipients(ctx, tx, mentionScopeID, authorMemberID, content)
	if err != nil {
		return AgentRoutingPlan{}, err
	}
	var defaultRecipients []AgentRecipient
	if target.ReplyThread != nil {
		defaultRecipients, err = LoadScopeThreadParticipantAgentRecipients(ctx, tx, scopeID, authorMemberID)
	} else {
		defaultRecipients, err = LoadScopeDefaultAgentRecipients(ctx, tx, scopeID, authorMemberID)
	}
	if err != nil {
		return AgentRoutingPlan{}, err
	}
	recipients := MergeAgentRecipients(defaultRecipients, mentionedRecipients)
	wakeRecipients := MergeAgentRecipients(FilterAutoWakeAgentRecipients(defaultRecipients), mentionedRecipients)
	return AgentRoutingPlan{DurableRecipients: recipients, WakeRecipients: wakeRecipients}, nil
}

func loadScopeDMPeerAgentRecipients(ctx context.Context, tx coredb.Tx, scopeID string, authorMemberID string) ([]AgentRecipient, error) {
	rows, err := tx.Query(ctx, `
SELECT a.id, a.machine_id, a.member_id, CASE WHEN sns.state = 'auto-wake' THEN 1 ELSE 0 END
FROM agents a
JOIN machines mac
  ON mac.workspace_id = a.workspace_id
 AND mac.id = a.machine_id
JOIN scope_memberships sm
  ON sm.scope_id = $1
 AND sm.member_id = a.member_id
 AND sm.removed_at IS NULL
LEFT JOIN scope_notification_settings sns
  ON sns.scope_id = sm.scope_id
 AND sns.member_id = sm.member_id
WHERE a.member_id <> $2
  AND a.status <> 'archived'
  AND mac.status = 'active'
ORDER BY a.member_id ASC
`, strings.TrimSpace(scopeID), strings.TrimSpace(authorMemberID))
	if err != nil {
		return nil, fmt.Errorf("load scope dm agent recipients: %w", err)
	}
	defer rows.Close()
	recipients := make([]AgentRecipient, 0, 1)
	for rows.Next() {
		var recipient AgentRecipient
		if err := rows.Scan(&recipient.AgentID, &recipient.MachineID, &recipient.AgentMemberID, &recipient.AutoWakeEnabled); err != nil {
			return nil, fmt.Errorf("scan scope dm agent recipient: %w", err)
		}
		recipient.DeliveryReason = AgentDeliveryReasonDMPeer
		recipients = append(recipients, recipient)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate scope dm agent recipients: %w", err)
	}
	return recipients, nil
}

func FilterAgentIDsByScopeMembers(ctx context.Context, tx coredb.Tx, scopeID string, agentIDs []string) ([]string, error) {
	normalizedAgentIDs := normalizeNonEmptyStrings(agentIDs)
	if len(normalizedAgentIDs) == 0 {
		return nil, nil
	}
	requestedAgentIDsCTE, requestedAgentIDArgs, _, err := coredb.BuildStringValuesCTE("requested_agent_ids", "agent_id", normalizedAgentIDs, 2)
	if err != nil {
		return nil, err
	}
	query := fmt.Sprintf(`
WITH %s
SELECT a.id
FROM requested_agent_ids
JOIN agents a ON a.id = requested_agent_ids.agent_id
JOIN machines mac
  ON mac.workspace_id = a.workspace_id
 AND mac.id = a.machine_id
JOIN scope_memberships sm
  ON sm.scope_id = $1
 AND sm.member_id = a.member_id
 AND sm.removed_at IS NULL
WHERE a.status <> 'archived'
  AND mac.status = 'active'
ORDER BY a.id ASC
`, requestedAgentIDsCTE)
	rows, err := tx.Query(ctx, query, append([]any{strings.TrimSpace(scopeID)}, requestedAgentIDArgs...)...)
	if err != nil {
		return nil, fmt.Errorf("filter agent ids by scope members: %w", err)
	}
	defer rows.Close()
	filtered := make([]string, 0, len(normalizedAgentIDs))
	for rows.Next() {
		var agentID string
		if err := rows.Scan(&agentID); err != nil {
			return nil, fmt.Errorf("scan filtered scope agent id: %w", err)
		}
		filtered = append(filtered, strings.TrimSpace(agentID))
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate filtered scope agent ids: %w", err)
	}
	return filtered, nil
}

func scopeIDForThreadTarget(target ThreadAccess) string {
	if strings.TrimSpace(target.ChannelID) != "" {
		return strings.TrimSpace(target.ChannelID)
	}
	return strings.TrimSpace(target.DMThreadID)
}