package app

import (
	"context"
	"fmt"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/core/domain"
	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/store/access"
	"github.com/tidbcloud/AgentTeam/internal/store/agent"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	"github.com/tidbcloud/AgentTeam/internal/store/tasktimeline"
)

const taskScopeFanoutInlineDeliveryGrace = 5 * time.Second

type taskScopeDiscussionDeliveryParams struct {
	WorkspaceID          string
	TaskScopeID          string
	MessageID            string
	AuthorMemberID       string
	AuthorType           domain.MemberType
	AuthorPrincipalID    string
	AuthorTargetID       string
	AuthorDisplayName    string
	AuthorMachineID      string
	AuthorServiceKind    string
	AuthorServiceSlug    string
	AuthorServiceBuiltin bool
	// NotifyPrincipalMemberID is the optional human principal attributed to a
	// service-authored Task discussion for wake/notification policy. It is
	// carried on the durable message and frozen into the outbox payload.
	NotifyPrincipalMemberID                string
	TeamInvokeTeamID                       string
	Source                                 message.ContributionSource
	RequireSingleWorkflowCallbackResponder bool
	WorkflowCallbackResponderAgentID       string
	WorkflowCallbackTaskWatchHandoff       *message.WorkflowCallbackTaskWatchHandoff
}

type taskScopeDiscussionMessageRecord struct {
	ID                 string
	WorkspaceID        string
	ScopeID            string
	Seq                int64
	WorkspaceGlobalSeq int64
	AuthorMemberID     string
	AuthorType         string
	ClientMessageID    string
	MessageType        string
	Content            string
	Metadata           map[string]any
	CreatedAt          time.Time
}

type taskScopeDiscussionRoutingParams struct {
	WorkspaceID      string
	TaskScopeID      string
	AuthorMemberID   string
	Content          string
	Metadata         map[string]any
	TeamInvokeTeamID string
}

type taskScopeDiscussionAudience struct {
	routingPlan             message.AgentRoutingPlan
	humanRecipientMemberIDs []string
	conversation            message.DeliveryConversationProjection
	mentionedHumans         []message.HumanMentionRecipient
}

func (store *Store) PreflightTaskScopeTeamInvokeForAgent(ctx context.Context, authorAgent agent.AuthenticatedAgent, taskScopeID string, content string, teamInvokeTeamID string) error {
	if strings.TrimSpace(teamInvokeTeamID) == "" {
		return nil
	}
	trimmedWorkspaceID := strings.TrimSpace(authorAgent.WorkspaceID)
	trimmedAuthorMemberID := strings.TrimSpace(authorAgent.MemberID)
	trimmedTaskScopeID := strings.TrimSpace(taskScopeID)
	if trimmedWorkspaceID == "" || trimmedAuthorMemberID == "" || trimmedTaskScopeID == "" {
		return badRequest("workspace, author member, and task scope are required")
	}
	return coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		_, err := store.taskScopeDiscussionAgentRoutingPlanForContent(ctx, tx, taskScopeDiscussionRoutingParams{
			WorkspaceID:      trimmedWorkspaceID,
			TaskScopeID:      trimmedTaskScopeID,
			AuthorMemberID:   trimmedAuthorMemberID,
			Content:          content,
			TeamInvokeTeamID: teamInvokeTeamID,
		})
		return err
	})
}

// projectTaskScopeDiscussionDeliveryTx keeps the read and delivery projections aligned.
func (store *Store) projectTaskScopeDiscussionDeliveryTx(ctx context.Context, tx coredb.Tx, params taskScopeDiscussionDeliveryParams) (message.AppMessage, message.MessageDelivery, error) {
	record, err := loadTaskScopeDiscussionMessageForDelivery(ctx, tx, params.WorkspaceID, params.TaskScopeID, params.MessageID)
	if err != nil {
		return message.AppMessage{}, message.MessageDelivery{}, err
	}
	if strings.TrimSpace(record.AuthorMemberID) != strings.TrimSpace(params.AuthorMemberID) {
		return message.AppMessage{}, message.MessageDelivery{}, fmt.Errorf("task discussion author does not match scope message author")
	}
	authorProfile := message.AuthorProfile{
		ActorType:      string(params.AuthorType),
		PrincipalID:    strings.TrimSpace(params.AuthorPrincipalID),
		MemberID:       strings.TrimSpace(record.AuthorMemberID),
		TargetID:       firstNonEmpty(params.AuthorTargetID, record.AuthorMemberID),
		DisplayName:    firstNonEmpty(params.AuthorDisplayName, params.AuthorTargetID, record.AuthorMemberID),
		ServiceKind:    strings.TrimSpace(params.AuthorServiceKind),
		ServiceSlug:    strings.TrimSpace(params.AuthorServiceSlug),
		ServiceBuiltin: params.AuthorServiceBuiltin,
	}.Normalize()

	audience, err := store.resolveTaskScopeDiscussionAudience(ctx, tx, params, record)
	if err != nil {
		return message.AppMessage{}, message.MessageDelivery{}, err
	}

	attachmentsByMessageID, err := store.attachmentStore().InTx(tx).LoadAttachmentsForMessages(ctx, record.WorkspaceID, []string{record.ID})
	if err != nil {
		return message.AppMessage{}, message.MessageDelivery{}, err
	}
	notifyPrincipalMemberID := firstNonEmpty(strings.TrimSpace(params.NotifyPrincipalMemberID), record.AuthorMemberID)
	receiveMessage := message.ReceiveMessage{
		MessageID:               record.ID,
		TargetKey:               record.ScopeID,
		TargetSeq:               record.Seq,
		ChannelType:             "task",
		ChannelName:             record.ScopeID,
		ChannelDisplayName:      record.ScopeID,
		NotifyPrincipalMemberID: notifyPrincipalMemberID,
		AuthorProfile:           authorProfile,
		MessageType:             record.MessageType,
		Source:                  params.Source.Normalize(),
		Content:                 record.Content,
		Metadata:                message.CopyMessageMetadata(record.Metadata),
		Timestamp:               record.CreatedAt,
		Attachments:             attachmentsByMessageID[record.ID],
	}
	// A workflow_run/schedule_run source identifies an automation-run task
	// discussion written under the frozen Human authorization/audit principal.
	// The visible canonical author is the Workflow/Schedule service author, never
	// the authorizing Human.
	message.ApplyAutomationServiceAuthorToReceiveMessage(&receiveMessage)
	delivery := message.MessageDelivery{
		ThreadID:                record.ScopeID,
		Seq:                     record.Seq,
		GlobalSeq:               record.WorkspaceGlobalSeq,
		MessageID:               record.ID,
		ClientMessageID:         record.ClientMessageID,
		AuthorMachineID:         strings.TrimSpace(params.AuthorMachineID),
		NotifyPrincipalMemberID: notifyPrincipalMemberID,
		Message:                 receiveMessage,
		Recipients:              audience.routingPlan.DurableRecipients,
		WakeRecipients:          audience.routingPlan.WakeRecipients,
		HumanRecipientMemberIDs: append([]string(nil), audience.humanRecipientMemberIDs...),
		HumanRecipientsTargeted: true,
		Conversation:            &audience.conversation,
		MentionedHumans:         audience.mentionedHumans,
		MentionContext: message.MentionContext{
			ChannelType: "task",
		},
	}
	appMessage := appMessageFromDelivery(record.ScopeID, authorProfile, delivery)
	appMessage.TaskContext = tasktimeline.TaskNotificationRouteContext(delivery.Conversation)
	// Keep the task discussion read model and delivery projection on the authoritative automation author.
	if !params.Source.IsZero() {
		appMessage.Source = params.Source.Normalize()
	}
	// Keep the task discussion read model and its delivery projection consistent:
	// authoritative automation author, never the frozen Human principal.
	message.ApplyAutomationServiceAuthorToAppMessage(&appMessage)
	delivery.Message.AuthorProfile = appMessage.EffectiveAuthorProfile()
	if err := attachResolvedMentionsForSentAppMessage(ctx, tx, record.WorkspaceID, &appMessage); err != nil {
		return message.AppMessage{}, message.MessageDelivery{}, err
	}
	// Finalize service author presentation before freezing the immutable payload:
	// the stored outbox payload must decode to exactly the returned delivery.
	if strings.TrimSpace(string(params.AuthorType)) == message.AuthorActorTypeService {
		messages := []message.AppMessage{appMessage}
		if err := attachServiceAuthorProfilePresentationWithQueryer(ctx, tx, record.WorkspaceID, messages); err != nil {
			return message.AppMessage{}, message.MessageDelivery{}, err
		}
		appMessage = messages[0]
		attachAppMessageToDelivery(&delivery, &appMessage)
	}
	delivery.AppMessage = &appMessage
	payload, err := message.MarshalScopeFanoutPayload(appMessage, delivery)
	if err != nil {
		return message.AppMessage{}, message.MessageDelivery{}, err
	}
	repos := message.NewScopeRepositories(tx)
	if err := repos.Outbox.Insert(ctx, message.BuildScopeMessageFanoutOutboxRecord(message.ScopeMessageFanoutOutboxRecordParams{
		WorkspaceID:           record.WorkspaceID,
		ScopeID:               record.ScopeID,
		MessageID:             record.ID,
		Payload:               payload,
		TargetedHumanDelivery: true,
		CreatedAt:             record.CreatedAt,
		NextAttemptAt:         record.CreatedAt.Add(taskScopeFanoutInlineDeliveryGrace),
	})); err != nil {
		return message.AppMessage{}, message.MessageDelivery{}, err
	}
	if err := repos.Outbox.SetPayloadForMessage(ctx, record.ID, payload); err != nil {
		return message.AppMessage{}, message.MessageDelivery{}, err
	}
	if _, err := store.handleAppMentionDeliveriesForMessage(ctx, tx, record.AuthorMemberID, record.AuthorType, record.WorkspaceID, message.ThreadAccess{
		ThreadType: domain.ThreadTypeChannel,
		ChannelID:  record.ScopeID,
	}, record.ID, record.MessageType, record.Content); err != nil {
		return message.AppMessage{}, message.MessageDelivery{}, err
	}
	return appMessage, delivery, nil
}

func (store *Store) resolveTaskScopeDiscussionAudience(ctx context.Context, tx coredb.Tx, params taskScopeDiscussionDeliveryParams, record taskScopeDiscussionMessageRecord) (taskScopeDiscussionAudience, error) {
	routingPlan, err := store.taskScopeDiscussionAgentRoutingPlan(ctx, tx, params, record)
	if err != nil {
		return taskScopeDiscussionAudience{}, err
	}
	if params.RequireSingleWorkflowCallbackResponder {
		responders, filterErr := filterActiveWorkflowCallbackResponders(ctx, tx, record.ScopeID, routingPlan.WakeRecipients)
		err = filterErr
		if err != nil {
			return taskScopeDiscussionAudience{}, err
		}
		selectedAgentID := strings.TrimSpace(params.WorkflowCallbackResponderAgentID)
		if selectedAgentID == "" {
			if message.ContainsControlledBroadcastToken(record.Content) {
				return taskScopeDiscussionAudience{}, badRequest("legacy workflow callback message does not support task broadcast mentions")
			}
			if len(responders) != 1 {
				return taskScopeDiscussionAudience{}, badRequest(fmt.Sprintf("legacy workflow callback message requires exactly one Agent responder; resolved %d", len(responders)))
			}
		} else {
			selectedFound := false
			for _, responder := range responders {
				if strings.TrimSpace(responder.AgentID) == selectedAgentID {
					selectedFound = true
					break
				}
			}
			if !selectedFound {
				return taskScopeDiscussionAudience{}, badRequest("workflow callback resolved mention did not resolve to the selected responder")
			}
		}
	}
	if err := ensureTaskScopeDiscussionParticipantDefaultUnwatched(ctx, tx, record.ScopeID, record.AuthorMemberID, record.CreatedAt); err != nil {
		return taskScopeDiscussionAudience{}, err
	}
	mentionedHumans, err := loadTaskScopeMentionedHumans(ctx, tx, record)
	if err != nil {
		return taskScopeDiscussionAudience{}, err
	}
	for _, recipient := range routingPlan.DurableRecipients {
		memberID := strings.TrimSpace(recipient.AgentMemberID)
		if memberID == "" {
			continue
		}
		if err := ensureTaskScopeDiscussionParticipantDefaultUnwatched(ctx, tx, record.ScopeID, memberID, record.CreatedAt); err != nil {
			return taskScopeDiscussionAudience{}, err
		}
	}
	for _, recipient := range mentionedHumans {
		memberID := strings.TrimSpace(recipient.MemberID)
		if memberID == "" {
			continue
		}
		if err := ensureTaskScopeDiscussionParticipantDefaultUnwatched(ctx, tx, record.ScopeID, memberID, record.CreatedAt); err != nil {
			return taskScopeDiscussionAudience{}, err
		}
	}
	if params.RequireSingleWorkflowCallbackResponder && params.WorkflowCallbackTaskWatchHandoff != nil {
		var responder message.AgentRecipient
		for _, candidate := range routingPlan.WakeRecipients {
			if strings.TrimSpace(candidate.AgentID) == strings.TrimSpace(params.WorkflowCallbackResponderAgentID) {
				responder = candidate
				break
			}
		}
		if err := store.applyWorkflowCallbackTaskWatchHandoffTx(
			ctx,
			tx,
			record.WorkspaceID,
			responder,
			*params.WorkflowCallbackTaskWatchHandoff,
			record.CreatedAt,
		); err != nil {
			return taskScopeDiscussionAudience{}, err
		}
	}
	attention, err := tasktimeline.LoadTaskAttentionSnapshotForScope(
		ctx,
		tx,
		record.WorkspaceID,
		record.ScopeID,
		record.AuthorMemberID,
		message.HumanMentionRecipientMemberIDs(mentionedHumans),
		routingPlan.DurableRecipients,
		"task.discussion_added",
		record.Content,
		record.CreatedAt,
	)
	if err != nil {
		return taskScopeDiscussionAudience{}, err
	}
	audience := taskScopeDiscussionAudience{
		routingPlan: message.AgentRoutingPlan{
			DurableRecipients: attention.AgentRecipients,
			WakeRecipients:    attention.AgentRecipients,
		},
		humanRecipientMemberIDs: attention.HumanRecipientMemberIDs,
		conversation:            attention.Conversation,
		mentionedHumans:         mentionedHumans,
	}
	return audience, nil
}

// resolveTaskScopeWorkflowCallbackResponder applies the same Agent visibility
// boundary as an ordinary Task mention, but resolves the compiler-selected
// AgentRef rather than parsing message text. The ref may be an Agent ID, member
// ID, or workspace-unique Agent name. A new WorkItem responder is not
// necessarily a Task participant yet, so visibility is resolved from the
// Task's board channel; resolveTaskScopeDiscussionAudience then materializes
// the exact canonical responder into the Task scope in this same transaction.
func resolveTaskScopeWorkflowCallbackResponder(
	ctx context.Context,
	queryer coredb.Queryer,
	workspaceID string,
	taskScopeID string,
	responderAgentRef string,
	authorMemberID string,
) (message.AgentRecipient, error) {
	lookupAgentRef := strings.TrimPrefix(strings.TrimSpace(responderAgentRef), "@")
	rows, err := queryer.Query(ctx, `
SELECT
    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 tasks t
JOIN scopes task_scope
  ON task_scope.id = t.task_scope_id
 AND task_scope.workspace_id = $1
 AND task_scope.type = 'task'
 AND task_scope.status = 'active'
 AND task_scope.archived_at IS NULL
JOIN task_boards task_board
  ON task_board.id = t.board_id
 AND task_board.workspace_id = $1
 AND t.channel_id = task_board.channel_id
JOIN agents a
  ON a.workspace_id = $1
 AND (
      a.id = $3
      OR LOWER(a.member_id) = LOWER($3)
      OR LOWER(a.name) = LOWER($3)
 )
 AND a.status <> 'archived'
JOIN machines mac
  ON mac.workspace_id = a.workspace_id
 AND mac.id = a.machine_id
 AND mac.status = 'active'
JOIN scope_memberships sm
  ON sm.scope_id = task_board.channel_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
LEFT JOIN scope_memberships creator_private_agent
  ON creator_private_agent.scope_id = task_board.channel_id
 AND creator_private_agent.member_id = COALESCE(NULLIF(t.created_by_ref, ''), t.created_by_member_id)
 AND creator_private_agent.visibility = 'private'
 AND creator_private_agent.owner_member_id IS NOT NULL
 AND creator_private_agent.removed_at IS NULL
 AND creator_private_agent.archived_at IS NULL
 AND creator_private_agent.effective_to_seq IS NULL
LEFT JOIN scope_notification_settings sns
  ON sns.scope_id = sm.scope_id
 AND sns.member_id = sm.member_id
WHERE t.task_scope_id = $2
  AND a.member_id <> $4
  AND (
      sm.visibility = 'public'
      OR (
          sm.visibility = 'private'
          AND (
              sm.owner_member_id = $4
              OR sm.member_id = $4
              OR sm.owner_member_id = COALESCE(
                  NULLIF(t.execution_owner_member_id, ''),
                  NULLIF(creator_private_agent.owner_member_id, ''),
                  NULLIF(t.created_by_ref, ''),
                  t.created_by_member_id
              )
          )
      )
  )
ORDER BY
    CASE
        WHEN a.id = $3 THEN 0
        WHEN LOWER(a.member_id) = LOWER($3) THEN 1
        ELSE 2
    END,
    CASE WHEN sm.visibility = 'private' THEN 0 ELSE 1 END,
    sm.effective_from_seq DESC
`,
		strings.TrimSpace(workspaceID),
		strings.TrimSpace(taskScopeID),
		lookupAgentRef,
		strings.TrimSpace(authorMemberID),
	)
	if err != nil {
		return message.AgentRecipient{}, fmt.Errorf(
			"resolve task workflow callback responder: %w",
			err,
		)
	}
	defer rows.Close()
	return scanUniqueWorkflowCallbackResponder(
		rows,
		"workflow callback responder must be visible from the Task board",
	)
}

func (store *Store) taskScopeDiscussionAgentRoutingPlan(ctx context.Context, tx coredb.Tx, params taskScopeDiscussionDeliveryParams, record taskScopeDiscussionMessageRecord) (message.AgentRoutingPlan, error) {
	return store.taskScopeDiscussionAgentRoutingPlanForContent(ctx, tx, taskScopeDiscussionRoutingParams{
		WorkspaceID:      params.WorkspaceID,
		TaskScopeID:      record.ScopeID,
		AuthorMemberID:   record.AuthorMemberID,
		Content:          record.Content,
		Metadata:         record.Metadata,
		TeamInvokeTeamID: params.TeamInvokeTeamID,
	})
}

func (store *Store) taskScopeDiscussionAgentRoutingPlanForContent(ctx context.Context, tx coredb.Tx, params taskScopeDiscussionRoutingParams) (message.AgentRoutingPlan, error) {
	parentChannelID, err := loadTaskScopeParentChannelID(ctx, tx, params.WorkspaceID, params.TaskScopeID)
	if err != nil {
		return message.AgentRoutingPlan{}, err
	}
	threadAccess := message.ThreadAccess{
		ThreadType: domain.ThreadTypeChannel,
		ChannelID:  strings.TrimSpace(parentChannelID),
	}
	// Task memberships are notification/read-cursor state only. Mention
	// candidates are resolved from the canonical parent Channel roster so a
	// retained Task participant row cannot create a content-delivery bypass.
	mentionedRecipients, err := message.LoadScopeMentionedAgentRecipientsWithMetadata(ctx, tx, parentChannelID, params.AuthorMemberID, params.Content, params.Metadata)
	if err != nil {
		return message.AgentRoutingPlan{}, err
	}
	routingPlan := message.FinalizeAgentRoutingPlan(message.AgentRoutingPlan{
		DurableRecipients: mentionedRecipients,
		WakeRecipients:    mentionedRecipients,
	})
	return store.attachAgentTeamInvokeContexts(ctx, tx, params.WorkspaceID, threadAccess, params.AuthorMemberID, params.TeamInvokeTeamID, routingPlan)
}

func loadTaskScopeParentChannelID(ctx context.Context, queryer coredb.QueryRower, workspaceID string, taskScopeID string) (string, error) {
	resolved, err := access.ResolveActiveTaskBoardChannel(ctx, queryer, workspaceID, taskScopeID)
	if err != nil {
		return "", mapTaskScopeMessageError(err)
	}
	return strings.TrimSpace(resolved.ChannelID), nil
}

func loadTaskScopeMentionedHumans(ctx context.Context, tx coredb.Tx, record taskScopeDiscussionMessageRecord) ([]message.HumanMentionRecipient, error) {
	channelID, err := loadTaskScopeParentChannelID(ctx, tx, record.WorkspaceID, record.ScopeID)
	if err != nil {
		return nil, fmt.Errorf("load task discussion parent channel for human mentions: %w", err)
	}
	mentionHints := make([]message.MessageMentionHint, 0)
	for _, link := range message.ParseResolvedMentionLinks(record.Content) {
		if strings.TrimSpace(link.Type) != "human" {
			continue
		}
		mentionHints = append(mentionHints, message.MessageMentionHint{
			Type:  link.Type,
			ID:    link.ID,
			Token: link.Token,
		})
	}
	return message.LoadMentionedHumanRecipientsForTarget(
		ctx,
		tx,
		record.WorkspaceID,
		record.AuthorMemberID,
		message.ThreadAccess{ThreadType: domain.ThreadTypeChannel, ChannelID: strings.TrimSpace(channelID)},
		record.Content,
		mentionHints,
	)
}

func loadTaskScopeDiscussionMessageForDelivery(ctx context.Context, queryer coredb.Queryer, workspaceID string, taskScopeID string, messageID string) (taskScopeDiscussionMessageRecord, error) {
	var record taskScopeDiscussionMessageRecord
	var rawMetadata []byte
	if err := queryer.QueryRow(ctx, `
SELECT id, workspace_id, scope_id, seq, workspace_global_seq, author_member_id, author_type,
       COALESCE(client_message_id, ''),
       message_type, content, metadata, created_at
FROM scope_messages
WHERE workspace_id = $1
  AND scope_id = $2
  AND id = $3
  AND message_type = $4
`, strings.TrimSpace(workspaceID), strings.TrimSpace(taskScopeID), strings.TrimSpace(messageID), message.MessageTypeTaskDiscussion).Scan(
		&record.ID,
		&record.WorkspaceID,
		&record.ScopeID,
		&record.Seq,
		&record.WorkspaceGlobalSeq,
		&record.AuthorMemberID,
		&record.AuthorType,
		&record.ClientMessageID,
		&record.MessageType,
		&record.Content,
		&rawMetadata,
		&record.CreatedAt,
	); err != nil {
		if coredb.IsNotFound(err) {
			return taskScopeDiscussionMessageRecord{}, ErrNotFound
		}
		return taskScopeDiscussionMessageRecord{}, fmt.Errorf("load task scope discussion message: %w", err)
	}
	metadata, err := message.DecodeMessageMetadata(rawMetadata)
	if err != nil {
		return taskScopeDiscussionMessageRecord{}, err
	}
	record.Metadata = metadata
	return record, nil
}

func ensureTaskScopeDiscussionParticipantDefaultUnwatched(ctx context.Context, execer coredb.QueryExecer, scopeID string, memberID string, createdAt time.Time) error {
	trimmedScopeID := strings.TrimSpace(scopeID)
	trimmedMemberID := strings.TrimSpace(memberID)
	if trimmedScopeID == "" || trimmedMemberID == "" {
		return nil
	}
	if err := message.EnsureScopeMembership(ctx, execer, trimmedScopeID, trimmedMemberID, "public", "", "", 0, 0, createdAt); err != nil {
		return err
	}
	state, err := taskScopeDiscussionDefaultUnwatchedNotificationStateForMember(ctx, execer, trimmedMemberID)
	if err != nil {
		return err
	}
	if state == "" {
		return nil
	}
	now := createdAt.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	if _, err := execer.Exec(ctx, `
INSERT INTO scope_notification_settings (
  scope_id, member_id, state, task_watch_state,
  created_at, updated_at, unread_generation
) VALUES ($1, $2, $3, $3, $4, $4, 1)
ON DUPLICATE KEY UPDATE
  unread_generation = CASE
    WHEN task_watch_state IS NULL THEN unread_generation + 1
    ELSE unread_generation
  END,
  updated_at = CASE
    WHEN task_watch_state IS NULL THEN VALUES(updated_at)
    ELSE updated_at
  END,
  task_watch_state = COALESCE(task_watch_state, VALUES(task_watch_state))
`, trimmedScopeID, trimmedMemberID, state, now); err != nil {
		return fmt.Errorf("insert task discussion default notification setting: %w", err)
	}
	return nil
}

func (store *Store) applyWorkflowCallbackTaskWatchHandoffTx(
	ctx context.Context,
	tx coredb.Tx,
	workspaceID string,
	responder message.AgentRecipient,
	handoff message.WorkflowCallbackTaskWatchHandoff,
	activatedAt time.Time,
) error {
	workspaceID = strings.TrimSpace(workspaceID)
	taskScopeID := strings.TrimSpace(handoff.TaskScopeID)
	currentAgentID := strings.TrimSpace(responder.AgentID)
	currentMemberID := strings.TrimSpace(responder.AgentMemberID)
	if workspaceID == "" || taskScopeID == "" || currentAgentID == "" || currentMemberID == "" {
		return badRequest("workflow callback Task watch handoff is incomplete")
	}
	active, err := message.LockActiveTaskScopeForMutation(ctx, tx, workspaceID, taskScopeID)
	if err != nil {
		return err
	}
	if !active {
		return conflict("workflow callback Task workScope is unavailable")
	}
	now := activatedAt.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	if err := ensureTaskScopeDiscussionParticipantDefaultUnwatched(ctx, tx, taskScopeID, currentMemberID, now); err != nil {
		return err
	}
	// Watch the successor first. If a later predecessor lookup fails, the whole
	// message transaction rolls back, so observers never see a half handoff.
	if err := message.UpsertTaskWatchNotificationState(ctx, tx, currentMemberID, taskScopeID, "auto-wake", now); err != nil {
		return err
	}

	seen := map[string]struct{}{currentAgentID: {}}
	for _, rawAgentID := range handoff.PredecessorAgentIDs {
		agentID := strings.TrimSpace(rawAgentID)
		if agentID == "" {
			continue
		}
		if _, duplicate := seen[agentID]; duplicate {
			continue
		}
		seen[agentID] = struct{}{}
		var memberID string
		err := tx.QueryRow(ctx, `
SELECT a.member_id
FROM agents a
JOIN scope_memberships sm
  ON sm.scope_id = $3
 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
WHERE a.workspace_id = $1
  AND a.id = $2
  AND a.status <> 'archived'
LIMIT 1
`, workspaceID, agentID, taskScopeID).Scan(&memberID)
		if coredb.IsNotFound(err) {
			continue
		}
		if err != nil {
			return fmt.Errorf("load workflow callback predecessor Task participant: %w", err)
		}
		if err := message.UpsertTaskWatchNotificationState(ctx, tx, memberID, taskScopeID, "no-wake", now); err != nil {
			return err
		}
	}
	return nil
}

func taskScopeDiscussionDefaultUnwatchedNotificationStateForMember(ctx context.Context, queryer coredb.Queryer, memberID string) (string, error) {
	var memberType string
	if err := queryer.QueryRow(ctx, `
SELECT type
FROM members
WHERE id = $1
`, strings.TrimSpace(memberID)).Scan(&memberType); err != nil {
		return "", fmt.Errorf("load task discussion notification member type: %w", err)
	}
	if strings.TrimSpace(memberType) == string(domain.MemberTypeAgent) {
		return "no-wake", nil
	}
	return "mute", nil
}