package fanout

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"strings"

	"github.com/tidbcloud/AgentTeam/internal/store/im"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	"github.com/tidbcloud/AgentTeam/internal/unread"
)

var errTargetedTaskUnreadPending = fmt.Errorf(
	"%w: targeted Task unread accounting is pending",
	message.ErrScopeFanoutDeferred,
)

// DeliveryFanout forwards persisted message deliveries to daemon agents and
// Workspace IM stream subscribers. It is shared by route domains that create
// messages outside the normal Workspace IM message create path.
type DeliveryFanout struct {
	notifier            AgentNotifier
	replyRootPreloader  ReplyThreadRootPreloader
	streamWriter        StreamEventWriter
	scopeFanoutMarker   ScopeFanoutDeliveryMarker
	scopeFanoutLeases   ScopeFanoutLeaseCoordinator
	scopeFanoutDeferrer ScopeFanoutLeaseDeferrer
	taskScopeLifecycle  TaskScopeLifecycleGuard
	unread              UnreadRuntime
	taskUnreadBarrier   TaskUnreadSnapshotBarrier
	unreadDeltaSink     UnreadDeltaSink
	logger              *slog.Logger
}

// NewDeliveryFanout builds a reusable delivery fanout adapter.
func NewDeliveryFanout(notifier AgentNotifier, streamWriter StreamEventWriter, logger *slog.Logger) DeliveryFanout {
	return DeliveryFanout{
		notifier:     notifier,
		streamWriter: streamWriter,
		logger:       logger,
	}
}

func (fanout DeliveryFanout) WithScopeFanoutMarker(marker ScopeFanoutDeliveryMarker) DeliveryFanout {
	fanout.scopeFanoutMarker = marker
	if coordinator, ok := marker.(ScopeFanoutLeaseCoordinator); ok {
		fanout.scopeFanoutLeases = coordinator
	}
	if deferrer, ok := marker.(ScopeFanoutLeaseDeferrer); ok {
		fanout.scopeFanoutDeferrer = deferrer
	}
	return fanout
}

func (fanout DeliveryFanout) WithTaskScopeLifecycleGuard(guard TaskScopeLifecycleGuard) DeliveryFanout {
	fanout.taskScopeLifecycle = guard
	return fanout
}

func (fanout DeliveryFanout) WithUnreadRuntime(runtime UnreadRuntime) DeliveryFanout {
	fanout.unread = runtime
	if barrier, ok := runtime.(TaskUnreadSnapshotBarrier); ok {
		fanout.taskUnreadBarrier = barrier
	}
	return fanout
}

func (fanout DeliveryFanout) WithUnreadDeltaSink(sink UnreadDeltaSink) DeliveryFanout {
	fanout.unreadDeltaSink = sink
	return fanout
}

func (fanout DeliveryFanout) WithReplyThreadRootPreloader(preloader ReplyThreadRootPreloader) DeliveryFanout {
	fanout.replyRootPreloader = preloader
	return fanout
}

func (fanout DeliveryFanout) deliverCommittedMessage(ctx context.Context, workspaceID string, appMessage message.AppMessage, delivery message.MessageDelivery) error {
	if delivery.AppMessage == nil {
		appMessageCopy := appMessage
		delivery.AppMessage = &appMessageCopy
	}
	_, err := fanout.withActiveTaskScopes(
		ctx,
		workspaceID,
		taskDeliveryTreeScopeIDs(appMessage, delivery),
		func(deliveryCtx context.Context) error {
			if err := fanout.renewScopeFanoutLease(deliveryCtx); err != nil {
				return fmt.Errorf("renew scope fanout before message delivery: %w", err)
			}
			// Ordinary messages do not have a frozen Human audience, so replay
			// deliberately suppresses unread re-projection. Account them before
			// the first stream attempt. Targeted Task messages have stable
			// recipients/event ids and are applied only after that recoverable
			// browser projection succeeds.
			unreadDeltas, err := fanout.applyUnreadMessageDeliveryPhase(
				deliveryCtx,
				workspaceID,
				appMessage,
				delivery,
				false,
			)
			if err != nil {
				return err
			}
			if err := fanout.publishDeliveryTree(deliveryCtx, workspaceID, appMessage, delivery); err != nil {
				return err
			}
			targetedTaskUnreadDeltas, err := fanout.applyUnreadMessageDeliveryPhase(
				deliveryCtx,
				workspaceID,
				appMessage,
				delivery,
				true,
			)
			unreadDeltas = append(unreadDeltas, targetedTaskUnreadDeltas...)
			if err != nil {
				// The stream is durable at this point. Publish deltas for earlier
				// successfully-accounted tree nodes before replay continues from
				// the failed targeted Task descendant.
				fanout.emitUnreadDeltas(workspaceID, unreadDeltas)
				return err
			}
			// The browser-visible message and unread runtime now both precede
			// daemon wake. A failed wake leaves the durable outbox pending
			// without exposing an Agent to a message clients cannot load.
			notifyErr := fanout.notifyAgentsStrict(deliveryCtx, workspaceID, delivery)
			followupErr := fanout.notifyFollowupAgentsStrict(deliveryCtx, workspaceID, delivery.FollowupDeliveries)
			fanout.emitUnreadDeltas(workspaceID, unreadDeltas)
			return errors.Join(notifyErr, followupErr)
		},
	)
	return err
}

// DeliverCommittedMessage forwards a committed message through the agent and
// Workspace IM stream fan-out path and returns any fan-out error to the caller.
func (fanout DeliveryFanout) DeliverCommittedMessage(ctx context.Context, workspaceID string, appMessage message.AppMessage, delivery message.MessageDelivery) error {
	if handled, err := fanout.deliverInlineTaskMessage(ctx, workspaceID, appMessage, delivery); handled {
		return err
	}
	if err := fanout.deliverCommittedMessage(ctx, workspaceID, appMessage, delivery); err != nil {
		return err
	}
	fanout.markScopeFanoutDeliveredTree(ctx, delivery)
	return nil
}

// EmitCommittedMessage forwards a committed message through the same
// agent and browser stream fan-out path used by POST Workspace IM message create.
func (fanout DeliveryFanout) EmitCommittedMessage(ctx context.Context, workspaceID string, appMessage message.AppMessage, delivery message.MessageDelivery) {
	if err := fanout.DeliverCommittedMessage(ctx, workspaceID, appMessage, delivery); err != nil {
		fanout.logFanoutError("emit committed message", err)
		return
	}
}

// EmitCommittedHumanMessage forwards a committed human-authored message through
// the same agent and browser stream fan-out path used by POST Workspace IM message create.
func (fanout DeliveryFanout) EmitCommittedHumanMessage(ctx context.Context, workspaceID string, message message.AppMessage, delivery message.MessageDelivery) {
	fanout.EmitCommittedMessage(ctx, workspaceID, message, delivery)
}

// EmitCommittedMessageProjectionOnly publishes the browser message projection
// without ordinary unread accounting, user notification, Agent wake, or
// outbox completion. It is reserved for explicit NoOutbox presentation events
// such as channel membership transcript rows.
func (fanout DeliveryFanout) EmitCommittedMessageProjectionOnly(ctx context.Context, workspaceID string, appMessage message.AppMessage, delivery message.MessageDelivery) {
	delivery.SuppressBrowserNotification = true
	if err := fanout.publishDeliveryTree(ctx, workspaceID, appMessage, delivery); err != nil {
		fanout.logFanoutError("emit committed message projection only", err)
	}
}

func (fanout DeliveryFanout) applyUnreadMessageEvent(ctx context.Context, workspaceID string, appMessage message.AppMessage, delivery message.MessageDelivery) ([]unread.UnreadDelta, error) {
	if !message.IsHumanVisibleAppMessage(appMessage) {
		return nil, nil
	}
	if strings.TrimSpace(appMessage.ChannelType) == message.WorkflowScopeType ||
		strings.TrimSpace(delivery.Message.ChannelType) == message.WorkflowScopeType {
		return nil, nil
	}
	// A targeted delivery carries a frozen, authoritative audience. An explicit
	// empty audience means that this Task event is Agent-only; do not ask the
	// unread runtime to hydrate/account the whole scope. Besides broadening the
	// audience, doing so would leave a large Task permanently Pending because
	// there is no human audience to record as accounted after hydration.
	if delivery.HumanRecipientsTargeted && len(delivery.HumanRecipientMemberIDs) == 0 {
		return nil, nil
	}
	if targetedTaskMessageDelivery(appMessage, delivery) {
		if err := fanout.requireTargetedTaskUnreadBarrier(ctx, delivery.HumanRecipientMemberIDs); err != nil {
			return nil, err
		}
	}
	if fanout.unread == nil {
		return nil, nil
	}
	// Task message recipients are frozen in the durable outbox. Replaying the
	// same seq/generation is idempotent and must repair the post-commit crash
	// window where synchronous fanout never populated unread state.
	if delivery.IdempotentReplay && !targetedTaskMessageDelivery(appMessage, delivery) {
		return nil, nil
	}
	scopeID := strings.TrimSpace(delivery.Message.TargetKey)
	if scopeID == "" {
		scopeID = strings.TrimSpace(appMessage.ChannelID)
	}
	if scopeID == "" {
		scopeID = strings.TrimSpace(delivery.ThreadID)
	}
	seq := delivery.Seq
	if seq <= 0 {
		seq = appMessage.Seq
	}
	if scopeID == "" || seq <= 0 {
		return nil, nil
	}
	scopeType, ok := unreadScopeTypeForMessage(appMessage, delivery, scopeID)
	if !ok {
		return nil, nil
	}
	authorMemberID := strings.TrimSpace(appMessage.AuthorMemberID())
	if authorMemberID == "" {
		authorMemberID = strings.TrimSpace(delivery.Message.AuthorProfile.MemberID)
	}
	automationSource := appMessage.Source.Normalize()
	if automationSource.IsZero() {
		automationSource = delivery.Message.Source.Normalize()
	}
	event := unread.MessageStored{
		WorkspaceID:             strings.TrimSpace(workspaceID),
		ScopeID:                 scopeID,
		ScopeType:               scopeType,
		Seq:                     seq,
		AuthorMemberID:          authorMemberID,
		NotifyPrincipalMemberID: effectiveDeliveryNotifyPrincipal(delivery, authorMemberID),
		AllowAuthorUnread:       message.AutomationSourceIdentifiesRun(automationSource),
		MessageGeneration:       seq,
		RecipientMemberIDs:      append([]string(nil), delivery.HumanRecipientMemberIDs...),
	}
	result, err := fanout.unread.ApplyMessage(event)
	if err != nil {
		return nil, fmt.Errorf("apply unread runtime message event: %w", err)
	}
	if result.Pending && targetedTaskMessageDelivery(appMessage, delivery) {
		// A cold roster queues the immutable Task event but has not accounted it
		// yet. Keep the durable outbox retryable and, critically, stop before
		// Agent wake. The preceding browser stream projection is idempotent, so
		// replay can safely repeat it after roster recovery.
		return nil, fmt.Errorf(
			"%w: message scope %q seq %d",
			errTargetedTaskUnreadPending,
			scopeID,
			seq,
		)
	}
	if targetedTaskMessageDelivery(appMessage, delivery) {
		if err := fanout.persistTargetedTaskUnreadSnapshots(
			ctx,
			workspaceID,
			scopeID,
			delivery.HumanRecipientMemberIDs,
		); err != nil {
			return result.Deltas, err
		}
	}
	return result.Deltas, nil
}

func effectiveDeliveryNotifyPrincipal(delivery message.MessageDelivery, authorMemberID string) string {
	if principal := strings.TrimSpace(delivery.NotifyPrincipalMemberID); principal != "" {
		return principal
	}
	if principal := strings.TrimSpace(delivery.Message.NotifyPrincipalMemberID); principal != "" {
		return principal
	}
	return strings.TrimSpace(authorMemberID)
}

func (fanout DeliveryFanout) applyUnreadMessageDeliveriesPhase(
	ctx context.Context,
	workspaceID string,
	deliveries []message.MessageDelivery,
	targetedTaskPhase bool,
) ([]unread.UnreadDelta, error) {
	var deltas []unread.UnreadDelta
	for _, delivery := range deliveries {
		appMessage := message.AppMessage{}
		if delivery.AppMessage != nil {
			appMessage = *delivery.AppMessage
		}
		deliveryDeltas, err := fanout.applyUnreadMessageDeliveryPhase(
			ctx,
			workspaceID,
			appMessage,
			delivery,
			targetedTaskPhase,
		)
		deltas = append(deltas, deliveryDeltas...)
		if err != nil {
			return deltas, err
		}
	}
	return deltas, nil
}

func (fanout DeliveryFanout) applyUnreadMessageDeliveryPhase(
	ctx context.Context,
	workspaceID string,
	appMessage message.AppMessage,
	delivery message.MessageDelivery,
	targetedTaskPhase bool,
) ([]unread.UnreadDelta, error) {
	var deltas []unread.UnreadDelta
	if targetedTaskMessageDelivery(appMessage, delivery) == targetedTaskPhase {
		deliveryDeltas, err := fanout.applyUnreadMessageEvent(ctx, workspaceID, appMessage, delivery)
		deltas = append(deltas, deliveryDeltas...)
		if err != nil {
			if targetedTaskPhase {
				return deltas, err
			}
			// Ordinary message outboxes do not freeze their Human audience and
			// therefore deliberately suppress unread re-projection on replay.
			// Keep their historical best-effort behavior: a process-local
			// runtime failure must not stop this tree before its durable stream
			// projection or strand later ordinary nodes forever.
			fanout.logFanoutError("apply unread runtime message event", err)
		}
	}
	for _, followup := range delivery.FollowupDeliveries {
		followupMessage := message.AppMessage{}
		if followup.AppMessage != nil {
			followupMessage = *followup.AppMessage
		}
		followupDeltas, err := fanout.applyUnreadMessageDeliveryPhase(
			ctx,
			workspaceID,
			followupMessage,
			followup,
			targetedTaskPhase,
		)
		deltas = append(deltas, followupDeltas...)
		if err != nil {
			return deltas, err
		}
	}
	return deltas, nil
}

func unreadScopeTypeForMessage(appMessage message.AppMessage, delivery message.MessageDelivery, scopeID string) (unread.ScopeType, bool) {
	var resolved unread.ScopeType
	for _, value := range []string{appMessage.ChannelType, delivery.Message.ChannelType} {
		switch strings.TrimSpace(value) {
		case "":
			continue
		case "channel":
			if resolved != "" && resolved != unread.ScopeTypeChannel {
				return "", false
			}
			resolved = unread.ScopeTypeChannel
		case "dm":
			if resolved != "" && resolved != unread.ScopeTypeDM {
				return "", false
			}
			resolved = unread.ScopeTypeDM
		case "thread":
			if resolved != "" && resolved != unread.ScopeTypeThread {
				return "", false
			}
			resolved = unread.ScopeTypeThread
		case "task":
			if resolved != "" && resolved != unread.ScopeTypeTask {
				return "", false
			}
			resolved = unread.ScopeTypeTask
		case "service_account":
			if resolved != "" && resolved != unread.ScopeTypeServiceAccount {
				return "", false
			}
			resolved = unread.ScopeTypeServiceAccount
		default:
			return "", false
		}
	}
	if resolved != "" {
		return resolved, true
	}
	if strings.HasPrefix(strings.TrimSpace(scopeID), "tsk_") {
		return unread.ScopeTypeTask, true
	}
	if strings.HasPrefix(strings.TrimSpace(scopeID), "sasc_") {
		return unread.ScopeTypeServiceAccount, true
	}
	return "", false
}

// NotifyAgents forwards one message delivery to daemon agents.
func (fanout DeliveryFanout) NotifyAgents(ctx context.Context, delivery message.MessageDelivery) {
	if err := fanout.notifyAgentsStrict(ctx, "", delivery); err != nil {
		fanout.logFanoutError("notify agents", err)
		return
	}
	fanout.markScopeFanoutDelivered(ctx, delivery)
}

func (fanout DeliveryFanout) deliverAgentOnly(ctx context.Context, workspaceID string, delivery message.MessageDelivery) error {
	_, err := fanout.withActiveTaskScopes(
		ctx,
		workspaceID,
		taskDeliveryTreeScopeIDs(message.AppMessage{}, delivery),
		func(deliveryCtx context.Context) error {
			if err := fanout.notifyAgentsStrict(deliveryCtx, workspaceID, delivery); err != nil {
				return err
			}
			fanout.markScopeFanoutDelivered(ctx, delivery)
			return nil
		},
	)
	return err
}

// DeliverAgentOnly forwards an already-committed delivery to daemon agents
// without emitting browser-visible message stream events.
func (fanout DeliveryFanout) DeliverAgentOnly(ctx context.Context, workspaceID string, delivery message.MessageDelivery) error {
	return fanout.deliverAgentOnly(ctx, workspaceID, delivery)
}

func (fanout DeliveryFanout) EmitMessageUpdated(ctx context.Context, workspaceID string, appMessage message.AppMessage) {
	if fanout.streamWriter == nil || !message.IsHumanVisibleAppMessage(appMessage) {
		return
	}
	if err := publishMessageDeliveryUpdatedEvent(ctx, fanout.streamWriter, workspaceID, appMessage.ChannelID, appMessage.ID); err != nil {
		fanout.logFanoutError("publish message delivery update", err)
	}
}

// EmitMessageEdited publishes the authoritative message edit projection. It is
// distinct from EmitMessageUpdated, whose legacy contract only invalidates
// message-delivery diagnostics.
func (fanout DeliveryFanout) EmitMessageEdited(ctx context.Context, workspaceID string, appMessage message.AppMessage) {
	if err := fanout.emitMessageEditedStrict(ctx, workspaceID, appMessage, ""); err != nil {
		fanout.logFanoutError("publish message edit", err)
	}
}

func (fanout DeliveryFanout) emitMessageEditedStrict(ctx context.Context, workspaceID string, appMessage message.AppMessage, mutationID string) error {
	if fanout.streamWriter == nil || !message.IsHumanVisibleAppMessage(appMessage) {
		return nil
	}
	return publishMessageUpdatedEvent(ctx, fanout.streamWriter, workspaceID, appMessage.ChannelID, appMessage, messageMutationStreamEventID(imEventTypeMessageUpdated, mutationID, appMessage.ChannelID))
}

func (fanout DeliveryFanout) EmitMessageRecalled(ctx context.Context, workspaceID string, appMessage message.AppMessage, updateConversation bool) {
	if err := fanout.emitMessageRecalledStrict(ctx, workspaceID, appMessage, updateConversation, ""); err != nil {
		fanout.logFanoutError("publish message recall", err)
	}
}

func (fanout DeliveryFanout) emitMessageRecalledStrict(ctx context.Context, workspaceID string, appMessage message.AppMessage, updateConversation bool, mutationID string) error {
	if !message.IsHumanVisibleAppMessage(appMessage) {
		return nil
	}
	scopeID := strings.TrimSpace(appMessage.ChannelID)
	messageID := strings.TrimSpace(appMessage.ID)
	var publishErrors []error
	if fanout.streamWriter != nil {
		if err := publishMessageRecalledEvent(ctx, fanout.streamWriter, workspaceID, scopeID, appMessage, updateConversation, messageMutationStreamEventID(imEventTypeMessageRecalled, mutationID, scopeID)); err != nil {
			publishErrors = append(publishErrors, err)
		}
		if err := publishMessagePinsInvalidatedEvent(ctx, fanout.streamWriter, workspaceID, scopeID, []string{messageID}, messageMutationStreamEventID(imEventTypeMessagePinsInvalidated, mutationID, scopeID)); err != nil {
			publishErrors = append(publishErrors, err)
		}
	}
	if fanout.unread != nil && scopeID != "" {
		deltas, err := fanout.unread.ApplyScopeInvalidatedWithDeltas(unread.ScopeInvalidated{
			WorkspaceID: strings.TrimSpace(workspaceID), ScopeID: scopeID, Reason: "message_recall",
		})
		if err != nil {
			fanout.logFanoutError("apply unread runtime message recall", err)
			publishErrors = append(publishErrors, err)
			return errors.Join(publishErrors...)
		}
		fanout.emitUnreadDeltas(workspaceID, deltas)
	}
	return errors.Join(publishErrors...)
}

func (fanout DeliveryFanout) notifyAgentsStrict(ctx context.Context, workspaceID string, delivery message.MessageDelivery) error {
	var errs []error
	for _, recipient := range delivery.WakeRecipients {
		if fanout.notifier != nil {
			attempt, managed := scopeFanoutLeaseAttemptFromContext(ctx)
			var sourceEventType string
			var sourceEventID string
			if managed && fanout.scopeFanoutLeases != nil {
				if err := fanout.renewScopeFanoutLease(ctx); err != nil {
					errs = append(errs, fmt.Errorf("renew scope fanout before notifying agent %s: %w", strings.TrimSpace(recipient.AgentID), err))
					continue
				}
				var err error
				sourceEventType, sourceEventID, err = scopeFanoutSourceIdentity(attempt, delivery)
				if err != nil {
					errs = append(errs, err)
					continue
				}
				delivered, err := fanout.scopeFanoutLeases.ScopeFanoutAgentDelivered(
					ctx,
					sourceEventType,
					sourceEventID,
					recipient.AgentID,
				)
				if err != nil {
					errs = append(errs, fmt.Errorf("load agent %s scope fanout receipt: %w", strings.TrimSpace(recipient.AgentID), err))
					continue
				}
				if delivered {
					continue
				}
			}
			wakeMessage := receiveMessageForFanoutDelivery(delivery, recipient)
			wakeMessage = fanout.preloadReplyThreadRootForRecipient(ctx, workspaceID, recipient, wakeMessage)
			err := fanout.notifier.NotifyAgentMessage(ctx, recipient, delivery.Seq, wakeMessage)
			if err != nil {
				errs = append(errs, fmt.Errorf("notify agent %s: %w", strings.TrimSpace(recipient.AgentID), err))
				continue
			}
			if managed && fanout.scopeFanoutLeases != nil {
				if err := fanout.scopeFanoutLeases.RecordScopeFanoutAgentDelivered(
					ctx,
					attempt.Outbox.ID,
					attempt.Outbox.LeaseOwner,
					sourceEventType,
					sourceEventID,
					recipient.AgentID,
				); err != nil {
					errs = append(errs, fmt.Errorf("record agent %s scope fanout receipt: %w", strings.TrimSpace(recipient.AgentID), err))
				}
			}
		}
	}
	return errors.Join(errs...)
}

func (fanout DeliveryFanout) deliverDeliveryIntent(ctx context.Context, workspaceID string, payload message.ScopeDeliveryIntentPayload) error {
	scopeID, isTask := taskDeliveryIntentScope(payload)
	var taskScopes []string
	if isTask {
		taskScopes = []string{scopeID}
	}
	_, err := fanout.withActiveTaskScopes(ctx, workspaceID, taskScopes, func(deliveryCtx context.Context) error {
		return fanout.deliverActiveDeliveryIntent(deliveryCtx, workspaceID, payload)
	})
	return err
}

func (fanout DeliveryFanout) deliverActiveDeliveryIntent(ctx context.Context, workspaceID string, payload message.ScopeDeliveryIntentPayload) error {
	if err := fanout.renewScopeFanoutLease(ctx); err != nil {
		return err
	}
	delivery := message.MessageDelivery{
		ThreadID:       payload.Message.TargetKey,
		Seq:            payload.Message.TargetSeq,
		MessageID:      payload.Message.MessageID,
		Message:        payload.Message,
		Recipients:     append([]message.AgentRecipient(nil), payload.Recipients...),
		WakeRecipients: append([]message.AgentRecipient(nil), payload.WakeRecipients...),
	}
	if payload.Conversation != nil {
		delivery.Conversation = payload.Conversation
	}

	// The browser projection is durable and idempotent by source event id. Do
	// it before mutating process-local unread state so a transient IM-store
	// error cannot strand a red dot without a corresponding Inbox row.
	if err := publishDeliveryIntentConversationEvent(ctx, fanout.streamWriter, workspaceID, payload); err != nil {
		return err
	}

	var unreadDeltas []unread.UnreadDelta
	if payload.HumanRecipientsTargeted && len(payload.HumanRecipientMemberIDs) > 0 {
		if _, ok := scopeFanoutLeaseAttemptFromContext(ctx); ok {
			if fanout.unread == nil || fanout.taskUnreadBarrier == nil {
				return fmt.Errorf("%w: targeted Task unread barrier is not configured", errTargetedTaskUnreadPending)
			}
		}
	}
	if fanout.unread != nil && len(payload.HumanRecipientMemberIDs) > 0 {
		result, err := fanout.unread.ApplyNonMessageActivity(unread.NonMessageActivityStored{
			WorkspaceID:        strings.TrimSpace(workspaceID),
			ScopeID:            strings.TrimSpace(payload.Message.TargetKey),
			ScopeType:          unread.ScopeTypeTask,
			Seq:                payload.Message.TargetSeq,
			ActorMemberID:      strings.TrimSpace(payload.Message.AuthorProfile.MemberID),
			ActivityGeneration: payload.Message.TargetSeq,
			RecipientMemberIDs: append([]string(nil), payload.HumanRecipientMemberIDs...),
		})
		if err != nil {
			return fmt.Errorf("apply unread runtime task activity: %w", err)
		}
		if result.Pending {
			// Targeted Activity recipients are frozen in this durable v2
			// payload. Pending means the cold roster has queued, but not yet
			// accounted, the event; retry the outbox before Agent wake.
			return fmt.Errorf(
				"%w: activity scope %q seq %d",
				errTargetedTaskUnreadPending,
				strings.TrimSpace(payload.Message.TargetKey),
				payload.Message.TargetSeq,
			)
		}
		if err := fanout.persistTargetedTaskUnreadSnapshots(
			ctx,
			workspaceID,
			strings.TrimSpace(payload.Message.TargetKey),
			payload.HumanRecipientMemberIDs,
		); err != nil {
			fanout.emitUnreadDeltas(workspaceID, result.Deltas)
			return err
		}
		unreadDeltas = result.Deltas
	}
	notifyErr := fanout.notifyAgentsStrict(ctx, workspaceID, delivery)
	fanout.emitUnreadDeltas(workspaceID, unreadDeltas)
	return notifyErr
}

func receiveMessageForFanoutDelivery(delivery message.MessageDelivery, recipient message.AgentRecipient) message.ReceiveMessage {
	msg := delivery.Message
	if strings.TrimSpace(msg.MessageID) == "" {
		msg.MessageID = strings.TrimSpace(delivery.MessageID)
	}
	if strings.TrimSpace(msg.TargetKey) == "" {
		if delivery.AppMessage != nil && strings.TrimSpace(delivery.AppMessage.ChannelID) != "" {
			msg.TargetKey = strings.TrimSpace(delivery.AppMessage.ChannelID)
		} else if strings.TrimSpace(delivery.ThreadID) != "" {
			msg.TargetKey = strings.TrimSpace(delivery.ThreadID)
		}
	}
	if !msg.CarriesScopeTargetSeq() {
		msg.TargetSeq = 0
	} else if msg.TargetSeq <= 0 && !deliveryIsStartedReplyThreadRoot(delivery, msg) {
		if delivery.Seq > 0 {
			msg.TargetSeq = delivery.Seq
		} else if delivery.AppMessage != nil && delivery.AppMessage.Seq > 0 {
			msg.TargetSeq = delivery.AppMessage.Seq
		}
	}
	if strings.TrimSpace(msg.InterruptPolicy) == "" {
		msg.InterruptPolicy = strings.TrimSpace(delivery.InterruptPolicy)
	}
	msg.RequireLiveHandoff = msg.RequireLiveHandoff || delivery.RequireLiveHandoff
	if msg.Source.IsZero() && delivery.AppMessage != nil {
		source := delivery.AppMessage.Source.Normalize()
		if source.IsZero() {
			source = message.DefaultContributionSourceForAppMessage(*delivery.AppMessage)
		}
		msg.Source = source
	}
	// A workflow_run/schedule_run source identifies an automation-run message
	// written under the frozen Human authorization/audit principal. Deliveries
	// must present the canonical Workflow/Schedule service author, never that
	// Human, so agent wake prompts and browser sockets name the automation.
	message.ApplyAutomationServiceAuthorToReceiveMessage(&msg)
	return message.ReceiveMessageForAgentRecipient(msg, recipient)
}

func deliveryIsStartedReplyThreadRoot(delivery message.MessageDelivery, msg message.ReceiveMessage) bool {
	if delivery.StartedReplyThread == nil {
		return false
	}
	threadChannelID := strings.TrimSpace(delivery.StartedReplyThread.ThreadChannelID)
	if threadChannelID == "" {
		return false
	}
	if strings.TrimSpace(msg.TargetKey) != threadChannelID {
		return false
	}
	if messageID := strings.TrimSpace(delivery.MessageID); messageID != "" {
		return strings.TrimSpace(msg.MessageID) == messageID
	}
	return true
}

func (fanout DeliveryFanout) notifyReactionDelivery(ctx context.Context, delivery *message.ReactionDelivery) {
	if delivery == nil {
		return
	}
	if fanout.notifier == nil {
		return
	}
	if err := fanout.notifier.NotifyAgentMessage(ctx, delivery.Recipient, delivery.Seq, delivery.Message); err != nil && fanout.logger != nil {
		fanout.logger.Warn("notify reaction delivery", slog.String("agent_id", delivery.Recipient.AgentID), slog.Any("error", err))
	}
}

// DeliverReactionAdded forwards one committed reaction add to daemon agents and
// Workspace IM stream subscribers.
func (fanout DeliveryFanout) DeliverReactionAdded(ctx context.Context, workspaceID string, messageID string, result message.AddReactionResult, reactions []message.MessageReactionAggregate) {
	if !result.Changed {
		return
	}
	taskScopes := reactionTaskScopeIDs(result.ChannelID, result.ChannelType)
	_, err := fanout.withActiveTaskScopes(ctx, workspaceID, taskScopes, func(deliveryCtx context.Context) error {
		fanout.notifyReactionDelivery(deliveryCtx, result.Delivery)
		return publishReactionChangedEvent(deliveryCtx, fanout.streamWriter, workspaceID, result.ChannelID, messageID, reactions, result.Notification)
	})
	if err != nil {
		fanout.logFanoutError("guard and publish reaction add", err)
	}
}

// DeliverReactionRemoved forwards one committed reaction removal to Workspace IM
// stream subscribers.
func (fanout DeliveryFanout) DeliverReactionRemoved(ctx context.Context, workspaceID string, messageID string, result message.RemoveReactionResult, reactions []message.MessageReactionAggregate) {
	if !result.Changed {
		return
	}
	taskScopes := reactionTaskScopeIDs(result.ChannelID, result.ChannelType)
	_, err := fanout.withActiveTaskScopes(ctx, workspaceID, taskScopes, func(deliveryCtx context.Context) error {
		return publishReactionChangedEvent(deliveryCtx, fanout.streamWriter, workspaceID, result.ChannelID, messageID, reactions, nil)
	})
	if err != nil {
		fanout.logFanoutError("guard and publish reaction remove", err)
	}
}

func reactionTaskScopeIDs(channelID string, channelType string) []string {
	channelID = strings.TrimSpace(channelID)
	if strings.TrimSpace(channelType) != message.ConversationKindTask && !strings.HasPrefix(channelID, "tsk_") {
		return nil
	}
	return []string{channelID}
}

func (fanout DeliveryFanout) notifyFollowupAgentsStrict(ctx context.Context, workspaceID string, deliveries []message.MessageDelivery) error {
	var errs []error
	for _, delivery := range deliveries {
		if err := fanout.notifyAgentsStrict(ctx, workspaceID, delivery); err != nil {
			errs = append(errs, err)
		}
		if err := fanout.notifyFollowupAgentsStrict(ctx, workspaceID, delivery.FollowupDeliveries); err != nil {
			errs = append(errs, err)
		}
	}
	return errors.Join(errs...)
}

// DeliverFollowupMessages emits app-created follow-up deliveries as one
// synchronous unit so the outbox is marked only after agent and browser stream fan-out.
func (fanout DeliveryFanout) DeliverFollowupMessages(ctx context.Context, workspaceID string, deliveries []message.MessageDelivery) {
	// Task-scoped follow-up deliveries carry a frozen audience in a pending_v2
	// outbox row. Only the lease-coordinated inline path (and the durable replay
	// worker) may complete those rows; the ordinary batch path below can only
	// mark v1 `pending` rows delivered and would otherwise leave the v2 row for
	// the replay worker to deliver a second time. Route them through
	// DeliverCommittedMessage so first fanout is exactly-once just like ordinary
	// Task discussion, while non-Task deliveries keep the existing batch policy.
	var taskDeliveries []message.MessageDelivery
	var ordinaryDeliveries []message.MessageDelivery
	for _, delivery := range deliveries {
		appMessage := message.AppMessage{}
		if delivery.AppMessage != nil {
			appMessage = *delivery.AppMessage
		}
		if len(taskDeliveryTreeScopeIDs(appMessage, delivery)) > 0 {
			taskDeliveries = append(taskDeliveries, delivery)
		} else {
			ordinaryDeliveries = append(ordinaryDeliveries, delivery)
		}
	}
	for _, delivery := range taskDeliveries {
		appMessage := message.AppMessage{}
		if delivery.AppMessage != nil {
			appMessage = *delivery.AppMessage
		}
		if err := fanout.DeliverCommittedMessage(ctx, workspaceID, appMessage, delivery); err != nil {
			fanout.logFanoutError("deliver task follow-up message", err)
		}
	}
	if len(ordinaryDeliveries) == 0 {
		return
	}
	deliveries = ordinaryDeliveries
	taskScopes := taskDeliveryTreesScopeIDs(deliveries)
	_, err := fanout.withActiveTaskScopes(ctx, workspaceID, taskScopes, func(deliveryCtx context.Context) error {
		unreadDeltas, err := fanout.applyUnreadMessageDeliveriesPhase(deliveryCtx, workspaceID, deliveries, false)
		if err != nil {
			return err
		}
		if err := fanout.publishDeliveryTrees(deliveryCtx, workspaceID, deliveries); err != nil {
			return fmt.Errorf("publish follow-up messages: %w", err)
		}
		targetedTaskUnreadDeltas, err := fanout.applyUnreadMessageDeliveriesPhase(deliveryCtx, workspaceID, deliveries, true)
		unreadDeltas = append(unreadDeltas, targetedTaskUnreadDeltas...)
		if err != nil {
			// The stream is durable, so preserve live repair for siblings that
			// were accounted before one targeted Task descendant failed.
			fanout.emitUnreadDeltas(workspaceID, unreadDeltas)
			return err
		}
		// As with the root committed-message path, browser-visible durable
		// projection must precede daemon wake. Otherwise a transient stream
		// failure can wake an Agent before the message is visible and the
		// pending outbox replay can wake it a second time.
		if err := fanout.notifyFollowupAgentsStrict(deliveryCtx, workspaceID, deliveries); err != nil {
			return fmt.Errorf("deliver follow-up messages: %w", err)
		}
		fanout.emitUnreadDeltas(workspaceID, unreadDeltas)
		fanout.markScopeFanoutDeliveredTrees(ctx, deliveries)
		return nil
	})
	if err != nil {
		fanout.logFanoutError("guard and deliver follow-up messages", err)
	}
}

func (fanout DeliveryFanout) withActiveTaskScopes(
	ctx context.Context,
	workspaceID string,
	scopeIDs []string,
	deliver func(context.Context) error,
) (bool, error) {
	if deliver == nil {
		return false, fmt.Errorf("Task delivery callback is required")
	}
	if len(scopeIDs) == 0 || fanout.taskScopeLifecycle == nil {
		return true, deliver(ctx)
	}
	workspaceID = strings.TrimSpace(workspaceID)
	if workspaceID == "" {
		return false, fmt.Errorf("Task delivery workspace is required")
	}
	protectedCtx := im.WithTaskLifecycleProtectedScopes(ctx, workspaceID, scopeIDs)
	active, err := fanout.taskScopeLifecycle.WithActiveTaskScopes(ctx, workspaceID, scopeIDs, func() error {
		return deliver(protectedCtx)
	})
	if err != nil {
		return false, fmt.Errorf("guard Task delivery scopes: %w", err)
	}
	return active, nil
}

func (fanout DeliveryFanout) preloadReplyThreadRootForRecipient(ctx context.Context, workspaceID string, recipient message.AgentRecipient, msg message.ReceiveMessage) message.ReceiveMessage {
	if fanout.replyRootPreloader == nil || strings.TrimSpace(workspaceID) == "" {
		return msg
	}
	decorated, err := fanout.replyRootPreloader.PreloadReplyThreadRootForRecipient(ctx, workspaceID, recipient, msg)
	if err != nil {
		fanout.logFanoutError("preload reply thread root", fmt.Errorf("agent %s: %w", strings.TrimSpace(recipient.AgentID), err))
		return msg
	}
	return decorated
}

func (fanout DeliveryFanout) publishDeliveryTrees(ctx context.Context, workspaceID string, deliveries []message.MessageDelivery) error {
	for _, delivery := range deliveries {
		appMessage := message.AppMessage{}
		if delivery.AppMessage != nil {
			appMessage = *delivery.AppMessage
		}
		if err := fanout.publishDeliveryTree(ctx, workspaceID, appMessage, delivery); err != nil {
			return err
		}
	}
	return nil
}

func (fanout DeliveryFanout) emitUnreadDeltas(workspaceID string, deltas []unread.UnreadDelta) {
	if fanout.unreadDeltaSink == nil || len(deltas) == 0 {
		return
	}
	for _, delta := range deltas {
		fanout.unreadDeltaSink.EmitUnreadUpdated(workspaceID, delta)
	}
}

func (fanout DeliveryFanout) publishDeliveryTree(ctx context.Context, workspaceID string, appMessage message.AppMessage, delivery message.MessageDelivery) error {
	if err := fanout.publishSingleDelivery(ctx, workspaceID, appMessage, delivery); err != nil {
		return err
	}
	return fanout.publishDeliveryTrees(ctx, workspaceID, delivery.FollowupDeliveries)
}

func (fanout DeliveryFanout) publishSingleDelivery(ctx context.Context, workspaceID string, appMessage message.AppMessage, delivery message.MessageDelivery) error {
	if fanout.streamWriter == nil {
		return nil
	}
	if strings.TrimSpace(appMessage.ID) != "" && message.IsHumanVisibleAppMessage(appMessage) {
		if strings.TrimSpace(appMessage.ChannelType) == message.WorkflowScopeType {
			publishWorkflowMessageCreatedLiveEvent(
				fanout.streamWriter, workspaceID, appMessage.ChannelID, appMessage,
			)
			return nil
		}
		return publishMessageCreatedEvent(ctx, fanout.streamWriter, workspaceID, appMessage.ChannelID, appMessage, delivery)
	}
	if delivery.ReplyThreadUpdated != nil {
		return publishThreadUpdatedEvent(ctx, fanout.streamWriter, workspaceID, *delivery.ReplyThreadUpdated)
	}
	return nil
}

func (fanout DeliveryFanout) markScopeFanoutDeliveredTrees(ctx context.Context, deliveries []message.MessageDelivery) {
	for _, delivery := range deliveries {
		fanout.markScopeFanoutDeliveredTree(ctx, delivery)
	}
}

func (fanout DeliveryFanout) markScopeFanoutDeliveredTree(ctx context.Context, delivery message.MessageDelivery) {
	fanout.markScopeFanoutDelivered(ctx, delivery)
	fanout.markScopeFanoutDeliveredTrees(ctx, delivery.FollowupDeliveries)
}

func (fanout DeliveryFanout) markScopeFanoutDelivered(ctx context.Context, delivery message.MessageDelivery) {
	if fanout.scopeFanoutMarker == nil || delivery.IdempotentReplay {
		return
	}
	// A coordinated inline delivery owns a lease and must complete it through
	// CompleteScopeFanoutLease. The legacy marker only accepts pending v1 rows;
	// invoking it here would race the lease protocol and, for targeted v2 rows,
	// leave a misleading error before the durable unread barrier runs.
	if _, coordinated := scopeFanoutLeaseAttemptFromContext(ctx); coordinated {
		return
	}
	messageID := strings.TrimSpace(delivery.MessageID)
	if messageID == "" {
		return
	}
	if err := fanout.scopeFanoutMarker.MarkScopeFanoutDeliveredForMessage(ctx, messageID); err != nil {
		fanout.logFanoutError("mark scope fanout delivered", err)
	}
}

func (fanout DeliveryFanout) logFanoutError(message string, err error) {
	if err == nil || fanout.logger == nil {
		return
	}
	fanout.logger.Warn(message, slog.Any("error", err))
}