package fanout

import (
	"context"
	"errors"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/api/dto"
	messageservice "github.com/tidbcloud/AgentTeam/internal/service/message"
	"github.com/tidbcloud/AgentTeam/internal/store/identity"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	unreadservice "github.com/tidbcloud/AgentTeam/internal/unread"
)

type capturingNotifier struct {
	recipients []message.AgentRecipient
	messages   []message.ReceiveMessage
}

func (notifier *capturingNotifier) NotifyAgentMessage(_ context.Context, recipient message.AgentRecipient, _ int64, message message.ReceiveMessage) error {
	notifier.recipients = append(notifier.recipients, recipient)
	notifier.messages = append(notifier.messages, message)
	return nil
}

type fakeReplyThreadRootPreloader struct {
	calls []message.AgentRecipient
}

func (preloader *fakeReplyThreadRootPreloader) PreloadReplyThreadRootForRecipient(_ context.Context, _ string, recipient message.AgentRecipient, msg message.ReceiveMessage) (message.ReceiveMessage, error) {
	preloader.calls = append(preloader.calls, recipient)
	metadata := message.CopyMessageMetadata(msg.Metadata)
	if len(metadata) == 0 {
		metadata = make(map[string]any, 1)
	}
	metadata[message.ReceiveMetadataReplyThreadRootKey] = map[string]any{
		"messageId":  "msg_root_1",
		"senderName": "andc5vq",
		"content":    "测试消息投递",
	}
	msg.Metadata = metadata
	return msg, nil
}

type capturingUnreadRuntime struct {
	messages          []unreadservice.MessageStored
	activities        []unreadservice.NonMessageActivityStored
	deltas            []unreadservice.UnreadDelta
	runtime           *unreadservice.Runtime
	applyMessageErrAt int
	applyMessageErr   error
	persistCalls      []taskUnreadPersistCall
	persistErr        error
}

type taskUnreadPersistCall struct {
	workspaceID string
	scopeID     string
	memberIDs   []string
}

func (runtime *capturingUnreadRuntime) ApplyNonMessageActivity(event unreadservice.NonMessageActivityStored) (unreadservice.ApplyResult, error) {
	runtime.activities = append(runtime.activities, event)
	if runtime.runtime != nil {
		return runtime.runtime.ApplyNonMessageActivity(event)
	}
	return unreadservice.ApplyResult{AccountedMembers: 1, Deltas: append([]unreadservice.UnreadDelta(nil), runtime.deltas...)}, nil
}

func (runtime *capturingUnreadRuntime) ApplyMessage(event unreadservice.MessageStored) (unreadservice.ApplyResult, error) {
	runtime.messages = append(runtime.messages, event)
	if runtime.applyMessageErrAt > 0 && len(runtime.messages) == runtime.applyMessageErrAt {
		return unreadservice.ApplyResult{}, runtime.applyMessageErr
	}
	if runtime.runtime != nil {
		return runtime.runtime.ApplyMessage(event)
	}
	return unreadservice.ApplyResult{AccountedMembers: 1, Deltas: append([]unreadservice.UnreadDelta(nil), runtime.deltas...)}, nil
}

func (runtime *capturingUnreadRuntime) ApplyMessageRewound(unreadservice.MessageRewound) (unreadservice.ApplyResult, error) {
	return unreadservice.ApplyResult{}, nil
}

func (runtime *capturingUnreadRuntime) ApplyScopeInvalidated(unreadservice.ScopeInvalidated) error {
	return nil
}

func (runtime *capturingUnreadRuntime) ApplyScopeInvalidatedWithDeltas(unreadservice.ScopeInvalidated) ([]unreadservice.UnreadDelta, error) {
	return nil, nil
}

func (runtime *capturingUnreadRuntime) PersistTaskUnreadSnapshots(_ context.Context, workspaceID string, scopeID string, memberIDs []string) error {
	runtime.persistCalls = append(runtime.persistCalls, taskUnreadPersistCall{
		workspaceID: workspaceID,
		scopeID:     scopeID,
		memberIDs:   append([]string(nil), memberIDs...),
	})
	return runtime.persistErr
}

func TestSendAndFanoutHumanMessagePreloadsReplyThreadRootForWakeRecipients(t *testing.T) {
	t.Parallel()

	store := newFakeMessageStore()
	store.sendTypeDelivery = message.MessageDelivery{
		Seq: 7,
		Message: message.ReceiveMessage{
			MessageID:   "msg_reply_1",
			ChannelType: "thread",
			Content:     "@agentteam-44-helper-1 请阅读本 thread 的所有消息",
		},
		WakeRecipients: []message.AgentRecipient{{
			AgentID:       "agt_1",
			AgentMemberID: "mem_agent_1",
		}},
	}

	notifier := &capturingNotifier{}
	preloader := &fakeReplyThreadRootPreloader{}
	fanout := NewDeliveryFanout(notifier, nil, nil).WithReplyThreadRootPreloader(preloader)

	_, _, err := SendAndFanoutHumanMessage(
		t.Context(),
		store,
		fanout,
		identity.User{ID: "usr_1", Name: "alice"},
		"wsp_1",
		"#test-pr-821:186226fe",
		"@agentteam-44-helper-1 请阅读本 thread 的所有消息",
		nil,
		messageservice.SendOptions{},
	)

	require.NoError(t, err)
	require.Len(t, preloader.calls, 1)
	require.Len(t, notifier.messages, 1)

	root, ok := notifier.messages[0].Metadata[message.ReceiveMetadataReplyThreadRootKey].(map[string]any)
	require.True(t, ok)
	require.Equal(t, "msg_root_1", root["messageId"])
	require.Equal(t, "测试消息投递", root["content"])
}

func TestDeliveryFanoutPreservesInterruptPolicy(t *testing.T) {
	t.Parallel()

	notifier := &capturingNotifier{}
	fanout := NewDeliveryFanout(notifier, nil, nil)
	fanout.NotifyAgents(t.Context(), message.MessageDelivery{
		Seq:             42,
		InterruptPolicy: message.AgentInterruptPolicyCurrentRun,
		Message: message.ReceiveMessage{
			MessageID:   "msg_interrupt",
			ChannelType: "channel",
			Content:     "switch now",
		},
		WakeRecipients: []message.AgentRecipient{{
			AgentID:         "agt_1",
			MachineID:       "mac_1",
			AutoWakeEnabled: true,
		}},
	})

	require.Len(t, notifier.messages, 1)
	require.Equal(t, message.AgentInterruptPolicyCurrentRun, notifier.messages[0].InterruptPolicy)
}

func TestDeliveryFanoutAttachesRecipientTeamContexts(t *testing.T) {
	t.Parallel()

	notifier := &capturingNotifier{}
	fanout := NewDeliveryFanout(notifier, nil, nil)
	fanout.NotifyAgents(t.Context(), message.MessageDelivery{
		Seq: 42,
		Message: message.ReceiveMessage{
			MessageID:   "msg_team_context",
			ChannelType: "channel",
			Content:     "@core-team please handle this",
		},
		WakeRecipients: []message.AgentRecipient{{
			AgentID:       "agt_team",
			AgentMemberID: "mem_agent",
			TeamContexts: []message.TeamDeliveryContext{{
				TeamID:                    "team_core",
				TeamHandle:                "core-team",
				TeamDisplayName:           "Core Team",
				TeamDescription:           "Core ownership",
				TeamMemberID:              "wtm_agent",
				RoleLabel:                 "Builder",
				RoleOverlayPromptSnapshot: "Build it.",
				RoleOverlayVersion:        9,
				MemberRoster: []message.TeamDeliveryRosterMember{{
					TeamMemberID:      "wtm_agent",
					ActorKind:         "agent",
					ActorID:           "mem_agent",
					DisplayName:       "AgentTeam Agent",
					Handle:            "agentteam-agent",
					RoleLabel:         "Builder",
					RoleOverlayPrompt: "Build it.",
					WakeOnTeamMention: true,
					DeliveryKind:      "wake",
					Self:              true,
				}},
			}},
		}},
	})

	require.Len(t, notifier.messages, 1)
	require.Len(t, notifier.messages[0].TeamContexts, 1)
	context := notifier.messages[0].TeamContexts[0]
	require.Equal(t, "team_core", context.TeamID)
	require.Equal(t, "core-team", context.TeamHandle)
	require.Equal(t, "wtm_agent", context.TeamMemberID)
	require.Equal(t, "Build it.", context.RoleOverlayPrompt)
	require.Len(t, context.MemberRoster, 1)
	require.Equal(t, "wtm_agent", context.MemberRoster[0].TeamMemberID)
	require.True(t, context.MemberRoster[0].Self)
	require.Equal(t, "Build it.", context.MemberRoster[0].RoleOverlayPrompt)
	require.True(t, context.MemberRoster[0].WakeOnTeamMention)
}

func TestDeliveryFanoutProjectsWorkflowCallbackAuthorityOnlyToFrozenResponder(t *testing.T) {
	t.Parallel()

	metadata := message.BuildWorkflowContextMetadata(message.WorkflowContextMetadataParams{
		WorkspaceID: "wsp_1", PlanID: "pln_1", PlanVersionID: "plv_1",
		RunID: "wri_1", EffectID: "wfx_1", NodeID: "calculate", SourceKind: "manual",
		CallbackExpected:  true,
		ResponderAgentRef: "calculator-a",
	})
	metadata[message.WorkflowContextMetadataResponderAgentIDKey] = "agt_selected"
	metadata["custom"] = "visible"

	notifier := &capturingNotifier{}
	fanout := NewDeliveryFanout(notifier, nil, nil)
	fanout.NotifyAgents(t.Context(), message.MessageDelivery{
		Seq: 42,
		Message: message.ReceiveMessage{
			MessageID: "msg_callback_projection", ChannelType: "channel",
			Content: "[calculator-a](agentteam://agent/agt_selected) and others inspect this", Metadata: metadata,
		},
		WakeRecipients: []message.AgentRecipient{
			{AgentID: "agt_selected", DeliveryReason: message.AgentDeliveryReasonDirectMention},
			{AgentID: "agt_default", DeliveryReason: message.AgentDeliveryReasonSpaceDefault},
			{AgentID: "agt_watcher", DeliveryReason: message.AgentDeliveryReasonTaskWatch},
			{AgentID: "agt_second_mention", DeliveryReason: message.AgentDeliveryReasonDirectMention},
		},
	})

	require.Len(t, notifier.messages, 4)
	for index, recipient := range notifier.recipients {
		projected := notifier.messages[index].Metadata
		require.Equal(t, "visible", projected["custom"])
		if recipient.AgentID == "agt_selected" {
			require.Equal(t, true, projected[message.WorkflowContextMetadataCallbackExpectedKey])
			require.Equal(t, "agt_selected", projected[message.WorkflowContextMetadataResponderAgentIDKey])
			continue
		}
		require.NotContains(t, projected, message.WorkflowContextMetadataCallbackExpectedKey)
		require.NotContains(t, projected, message.WorkflowContextMetadataRunIDKey)
		require.NotContains(t, projected, message.WorkflowContextMetadataResponderAgentIDKey)
	}
	require.Contains(t, metadata, message.WorkflowContextMetadataCallbackExpectedKey, "recipient projection must not mutate the shared delivery")
}

func TestDeliveryFanoutDoesNotInventTargetSeqForStartedReplyThreadRoot(t *testing.T) {
	t.Parallel()

	notifier := &capturingNotifier{}
	fanout := NewDeliveryFanout(notifier, nil, nil)
	fanout.NotifyAgents(t.Context(), message.MessageDelivery{
		Seq:       42,
		MessageID: "msg_thread_root",
		Message: message.ReceiveMessage{
			MessageID:   "msg_thread_root",
			TargetKey:   "chn_thread_1",
			ChannelType: "thread",
			Content:     "@agent handle this in a thread",
		},
		StartedReplyThread: &message.ReplyThreadSummary{ThreadChannelID: "chn_thread_1"},
		WakeRecipients: []message.AgentRecipient{{
			AgentID:         "agt_1",
			MachineID:       "mac_1",
			AutoWakeEnabled: true,
		}},
	})

	require.Len(t, notifier.messages, 1)
	require.Equal(t, "chn_thread_1", notifier.messages[0].TargetKey)
	require.Zero(t, notifier.messages[0].TargetSeq)
}

func TestReceiveMessageForFanoutDeliveryDoesNotInventTargetSeqForTaskActivity(t *testing.T) {
	t.Parallel()

	msg := receiveMessageForFanoutDelivery(message.MessageDelivery{
		Seq: 99,
		Message: message.ReceiveMessage{
			MessageID:   "tev_checkpoint",
			TargetKey:   "tsk_scope",
			TargetSeq:   88,
			ChannelType: "task",
			MessageType: "task.checkpoint_added",
			Source: message.ContributionSource{
				RefType: message.ContributionSourceRefTypeTaskActivity,
				RefID:   "tev_checkpoint",
			},
		},
	}, message.AgentRecipient{})

	require.Zero(t, msg.TargetSeq)
}

func TestDeliveryFanoutCopiesAppMessageSourceToAgentWake(t *testing.T) {
	t.Parallel()

	source := message.ContributionSource{
		Kind:      message.ContributionSourceKindScheduleRun,
		RefType:   message.ContributionSourceRefTypeAutomationRun,
		RefID:     "arun_wake_source",
		RunID:     "arun_wake_source",
		EventType: "cron",
		Label:     "daily source",
	}.Normalize()
	notifier := &capturingNotifier{}
	fanout := NewDeliveryFanout(notifier, nil, nil)

	fanout.EmitCommittedMessage(t.Context(), "wsp_1", message.AppMessage{
		ID:        "msg_wake_source",
		Seq:       12,
		ChannelID: "chn_wake_source",
		Source:    source,
	}, message.MessageDelivery{
		MessageID: "msg_wake_source",
		Seq:       12,
		Message: message.ReceiveMessage{
			MessageID: "msg_wake_source", Content: "source-backed wake", Source: source,
			AuthorProfile: message.AuthorProfile{
				ActorType: message.AuthorActorTypeHuman, MemberID: "mem_owner", DisplayName: "Owner",
			},
		},
		WakeRecipients: []message.AgentRecipient{{AgentID: "agt_wake_source"}},
	})

	require.Len(t, notifier.messages, 1)
	require.Equal(t, source, notifier.messages[0].Source)
	require.Equal(t, message.AuthorServiceKindSchedule, notifier.messages[0].AuthorProfile.ServiceKind)
}

func TestDeliveryFanoutFeedsUnreadRuntimeForCommittedVisibleMessage(t *testing.T) {
	t.Parallel()

	unread := &capturingUnreadRuntime{}
	fanout := NewDeliveryFanout(nil, nil, nil).WithUnreadRuntime(unread)

	fanout.EmitCommittedHumanMessage(t.Context(), "wsp_1", message.AppMessage{
		ID:          "msg_share_card",
		Seq:         12,
		ChannelID:   "chn_1",
		ChannelType: "channel",
		AuthorProfile: message.AuthorProfile{
			ActorType:   message.AuthorActorTypeHuman,
			MemberID:    "mem_author",
			DisplayName: "Alice",
		},
		MessageType: message.MessageTypeChannelShareCard,
	}, message.MessageDelivery{
		MessageID: "msg_share_card",
		Seq:       12,
		Message: message.ReceiveMessage{
			MessageID:   "msg_share_card",
			TargetKey:   "chn_1",
			ChannelType: "channel",
		},
	})

	require.Equal(t, []unreadservice.MessageStored{{
		WorkspaceID:             "wsp_1",
		ScopeID:                 "chn_1",
		ScopeType:               unreadservice.ScopeTypeChannel,
		Seq:                     12,
		AuthorMemberID:          "mem_author",
		NotifyPrincipalMemberID: "mem_author",
		MessageGeneration:       12,
	}}, unread.messages)
}

func TestDeliveryFanoutProjectionOnlySkipsUnreadAndAgentWake(t *testing.T) {
	t.Parallel()

	unread := &capturingUnreadRuntime{}
	notifier := &capturingNotifier{}
	realtime := &fakeRealtime{}
	fanout := NewDeliveryFanout(notifier, realtime, nil).WithUnreadRuntime(unread)

	fanout.EmitCommittedMessageProjectionOnly(t.Context(), "wsp_projection_only", message.AppMessage{
		ID:          "msg_member_event",
		Seq:         14,
		ChannelID:   "chn_projection_only",
		ChannelType: "channel",
		MessageType: message.MessageTypeSystem,
	}, message.MessageDelivery{
		MessageID: "msg_member_event",
		Seq:       14,
		Message: message.ReceiveMessage{
			MessageID:   "msg_member_event",
			TargetKey:   "chn_projection_only",
			ChannelType: "channel",
		},
		WakeRecipients: []message.AgentRecipient{{AgentID: "agt_must_not_wake"}},
	})

	require.Equal(t, []string{"chn_projection_only"}, realtime.messageNew)
	require.Equal(t, []bool{true}, realtime.messageNewSuppressBrowserNotifications)
	require.Empty(t, unread.messages)
	require.Empty(t, notifier.messages)
}

func TestDeliveryFanoutProjectsSilentExternalAppThreadReplyToUnreadAndRealtime(t *testing.T) {
	t.Parallel()

	unread := &capturingUnreadRuntime{}
	notifier := &capturingNotifier{}
	realtime := &fakeRealtime{}
	fanout := NewDeliveryFanout(notifier, realtime, nil).WithUnreadRuntime(unread)

	appMessage := message.AppMessage{
		ID:          "msg_external_bot_reply",
		Seq:         15,
		ChannelID:   "chn_external_bot_reply_thread",
		ChannelType: "thread",
		AuthorProfile: message.AuthorProfile{
			ActorType:   message.AuthorActorTypeService,
			MemberID:    "mem_external_bot",
			DisplayName: "Deploy App",
		},
		Content: "deployment started",
	}
	delivery := message.MessageDelivery{
		MessageID:                   appMessage.ID,
		Seq:                         appMessage.Seq,
		AppMessage:                  &appMessage,
		RecoverableStreamProjection: true,
		HumanRecipientsTargeted:     false,
		HumanRecipientMemberIDs:     nil,
		Message: message.ReceiveMessage{
			MessageID:   appMessage.ID,
			TargetKey:   appMessage.ChannelID,
			TargetSeq:   appMessage.Seq,
			ChannelType: "thread",
			AuthorProfile: message.AuthorProfile{
				ActorType: message.AuthorActorTypeService,
				MemberID:  "mem_external_bot",
			},
		},
		Recipients: []message.AgentRecipient{{AgentID: "agt_durable_observer"}},
		// External App runtime replies deliberately retain durable Agent
		// recipients while silent=true leaves the wake set empty.
		WakeRecipients: nil,
	}

	require.NoError(t, fanout.DeliverCommittedMessage(t.Context(), "wsp_external_bot_reply", appMessage, delivery))
	require.Equal(t, []string{appMessage.ChannelID}, realtime.messageNew)
	require.Len(t, unread.messages, 1)
	require.Equal(t, appMessage.ChannelID, unread.messages[0].ScopeID)
	require.Equal(t, unreadservice.ScopeTypeThread, unread.messages[0].ScopeType)
	require.Empty(t, notifier.messages)
}

func TestDeliveryFanoutEmitsUnreadRuntimeDeltasAfterMessageRealtime(t *testing.T) {
	t.Parallel()

	unread := &capturingUnreadRuntime{deltas: []unreadservice.UnreadDelta{{
		WorkspaceID:        "wsp_1",
		UserID:             "usr_reader",
		MemberID:           "mem_reader",
		ScopeID:            "chn_1",
		Count:              2,
		HasUnread:          true,
		WorkspaceHasUnread: true,
		ServingState:       unreadservice.ServingStateReady,
	}}}
	realtime := &fakeRealtime{}
	fanout := NewDeliveryFanout(nil, realtime, nil).
		WithUnreadRuntime(unread).
		WithUnreadDeltaSink(realtime)

	fanout.EmitCommittedHumanMessage(t.Context(), "wsp_1", message.AppMessage{
		ID:          "msg_delta",
		Seq:         13,
		ChannelID:   "chn_1",
		ChannelType: "channel",
		AuthorProfile: message.AuthorProfile{
			ActorType:   message.AuthorActorTypeHuman,
			MemberID:    "mem_author",
			DisplayName: "Alice",
		},
	}, message.MessageDelivery{
		MessageID: "msg_delta",
		Seq:       13,
		Message: message.ReceiveMessage{
			MessageID:   "msg_delta",
			TargetKey:   "chn_1",
			ChannelType: "channel",
		},
	})

	require.Equal(t, []string{"chn_1"}, realtime.messageNew)
	require.Equal(t, unread.deltas, realtime.unreadUpdated)
}

func TestDeliveryFanoutStopsBeforeAgentNotifyWhenMessageRealtimeFails(t *testing.T) {
	t.Parallel()

	streamErr := errors.New("stream unavailable")
	stream := &fakeRealtime{appendErr: streamErr}
	notifier := &capturingNotifier{}
	unread := &capturingUnreadRuntime{deltas: []unreadservice.UnreadDelta{{
		WorkspaceID:        "wsp_1",
		MemberID:           "mem_reader",
		ScopeID:            "tsk_1",
		Count:              1,
		HasUnread:          true,
		WorkspaceHasUnread: true,
		ServingState:       unreadservice.ServingStateReady,
	}}}
	unreadSink := &fakeRealtime{}
	lifecycle := &taskScopeLifecycleGuardStub{active: true}
	fanout := NewDeliveryFanout(notifier, stream, nil).
		WithTaskScopeLifecycleGuard(lifecycle).
		WithUnreadRuntime(unread).
		WithUnreadDeltaSink(unreadSink)

	err := fanout.DeliverCommittedMessage(
		t.Context(),
		"wsp_1",
		taskLifecycleAppMessage("tsk_1"),
		taskLifecycleDelivery("tsk_1"),
	)

	require.ErrorIs(t, err, streamErr)
	require.Equal(t, 1, lifecycle.callbackCalls)
	require.Empty(t, unread.messages, "failed stream append must not advance unread accounting")
	require.Empty(t, unreadSink.unreadUpdated, "unread deltas remain after successful stream publication")
	require.Empty(t, notifier.messages, "a failed stream append must not wake an agent")
}

func TestDeliveryFanoutRetryAfterMessageRealtimeFailureEmitsUnreadDelta(t *testing.T) {
	t.Parallel()

	runtime, err := unreadservice.New(unreadservice.Config{})
	require.NoError(t, err)
	require.NoError(t, runtime.WarmScopeRoster("wsp_retry", "tsk_retry", unreadservice.ScopeTypeTask, []unreadservice.RosterMember{
		{MemberID: "mem_author", UserID: "usr_author", Visibility: "public", NotificationState: "active"},
		{MemberID: "mem_reader", UserID: "usr_reader", Visibility: "public", NotificationState: "active"},
	}))
	unread := &capturingUnreadRuntime{runtime: runtime}
	streamErr := errors.New("stream unavailable")
	realtime := &fakeRealtime{appendErr: streamErr}
	fanout := NewDeliveryFanout(nil, realtime, nil).
		WithUnreadRuntime(unread).
		WithUnreadDeltaSink(realtime)
	appMessage := taskLifecycleAppMessage("tsk_retry")
	delivery := taskLifecycleDelivery("tsk_retry")

	err = fanout.DeliverCommittedMessage(t.Context(), "wsp_retry", appMessage, delivery)
	require.ErrorIs(t, err, streamErr)
	require.Empty(t, unread.messages)
	require.Empty(t, realtime.unreadUpdated)
	require.Zero(t, runtime.ScopeUnreadCounts("wsp_retry", "mem_reader", nil)["tsk_retry"])

	// Durable outbox replay uses the same immutable sequence and stable stream
	// event id. It must be able to perform unread accounting and emit the delta
	// after the transient stream failure clears.
	realtime.appendErr = nil
	delivery.IdempotentReplay = true
	require.NoError(t, fanout.DeliverCommittedMessage(t.Context(), "wsp_retry", appMessage, delivery))

	require.Len(t, unread.messages, 1)
	require.Equal(t, []string{"tsk_retry"}, realtime.messageNew)
	require.Len(t, realtime.unreadUpdated, 1)
	require.Equal(t, "mem_reader", realtime.unreadUpdated[0].MemberID)
	require.Equal(t, "tsk_retry", realtime.unreadUpdated[0].ScopeID)
	require.Equal(t, int64(1), realtime.unreadUpdated[0].Count)
	require.Equal(t, int64(1), runtime.ScopeUnreadCounts("wsp_retry", "mem_reader", nil)["tsk_retry"])

	require.NoError(t, fanout.DeliverCommittedMessage(t.Context(), "wsp_retry", appMessage, delivery))
	require.Len(t, realtime.messageNew, 1, "stable message event id must dedupe later replay")
	require.Len(t, realtime.unreadUpdated, 1, "same-generation replay must not emit a duplicate unread delta")
	require.Equal(t, int64(1), runtime.ScopeUnreadCounts("wsp_retry", "mem_reader", nil)["tsk_retry"])
}

func TestDeliveryFanoutMessageCreatedIncludesResolvedMentions(t *testing.T) {
	t.Parallel()

	realtime := &fakeRealtime{}
	fanout := NewDeliveryFanout(nil, realtime, nil)

	fanout.EmitCommittedMessage(t.Context(), "wsp_1", message.AppMessage{
		ID:        "msg_mention",
		Seq:       21,
		ChannelID: "chn_1",
		Content:   "hello @agent-name",
		AuthorProfile: message.AuthorProfile{
			ActorType:   message.AuthorActorTypeAgent,
			MemberID:    "mem_agent",
			DisplayName: "Agent Author",
		},
		ResolvedMentions: []message.ResolvedMention{{
			Token:         "agent-name",
			Type:          "agent",
			ID:            "agt_1",
			DisplayName:   "Agent Display",
			ResolvedToken: "agent-name",
		}},
	}, message.MessageDelivery{
		MessageID: "msg_mention",
		Seq:       21,
		Message: message.ReceiveMessage{
			MessageID: "msg_mention",
			TargetKey: "chn_1",
		},
	})

	require.Len(t, realtime.messageNewMessages, 1)
	require.Equal(t, []message.ResolvedMention{{
		Token:         "agent-name",
		Type:          "agent",
		ID:            "agt_1",
		DisplayName:   "Agent Display",
		ResolvedToken: "agent-name",
	}}, realtime.messageNewMessages[0].ResolvedMentions)
}

func TestDeliveryFanoutFeedsGenericUnreadForTaskScopeMessage(t *testing.T) {
	t.Parallel()

	unread := &capturingUnreadRuntime{}
	fanout := NewDeliveryFanout(nil, nil, nil).WithUnreadRuntime(unread)

	fanout.EmitCommittedHumanMessage(t.Context(), "wsp_1", message.AppMessage{
		ID:          "msg_task",
		Seq:         7,
		ChannelID:   "tsk_1",
		ChannelType: "task",
	}, message.MessageDelivery{
		MessageID: "msg_task",
		Seq:       7,
		Message: message.ReceiveMessage{
			MessageID:   "msg_task",
			TargetKey:   "tsk_1",
			ChannelType: "task",
		},
	})

	require.Equal(t, []unreadservice.MessageStored{{
		WorkspaceID:       "wsp_1",
		ScopeID:           "tsk_1",
		ScopeType:         unreadservice.ScopeTypeTask,
		Seq:               7,
		MessageGeneration: 7,
	}}, unread.messages)
}

func TestConversationIndexFromServiceAccountMessageBuildsInboxProjection(t *testing.T) {
	t.Parallel()

	createdAt := time.Date(2026, 7, 11, 9, 30, 0, 0, time.UTC)
	item := conversationIndexFromMessage(dto.AppMessage{
		ID:          "msg_service_account",
		Seq:         4,
		ChannelID:   "sasc_news",
		ChannelType: "service_account",
		AuthorProfile: &dto.AuthorProfile{
			ActorType:   "service",
			PrincipalID: "sacct_news",
			MemberID:    "mem_service_news",
			DisplayName: "Workspace News",
			AvatarURL:   "https://example.com/news.png",
			ServiceKind: "service_account",
			ServiceSlug: "workspace-news",
		},
		Content:   "A service-account announcement",
		CreatedAt: createdAt,
	})

	require.NotNil(t, item)
	require.Equal(t, "serviceAccount:sasc_news", item.ConversationID)
	require.Equal(t, "sasc_news", item.ScopeID)
	require.Equal(t, "service_account", item.Kind)
	require.Equal(t, createdAt, item.SortAt)
	require.Equal(t, "Workspace News", item.Projection["title"])
	require.Equal(t, "workspace-news", item.Projection["subtitle"])
	require.Equal(t, "service", item.Projection["peerType"])
	require.Equal(t, "sacct_news", item.Projection["peerId"])
	require.Equal(t, "/inbox/service-accounts/sacct_news", item.Projection["route"])
	require.Equal(t, "A service-account announcement", item.Projection["preview"])
	require.Equal(t, "active", item.Projection["subscriptionState"])
	require.Equal(t, createdAt, item.Projection["lastMessageAt"])
	require.IsType(t, dto.AppMessage{}, item.Projection["lastMessage"])
}

func TestConversationIndexFromTaskMessageReturnsNil(t *testing.T) {
	t.Parallel()

	item := conversationIndexFromMessage(dto.AppMessage{
		ID:          "msg_task_fanout",
		Seq:         7,
		ChannelID:   "tsk_fanout_index",
		ChannelType: message.ConversationKindTask,
		Content:     "task body",
		CreatedAt:   time.Date(2026, 7, 15, 12, 0, 0, 0, time.UTC),
	})

	require.Nil(t, item, "task messages must not materialize an Inbox conversation index row via the fanout path")
}

func TestPublishMessageCreatedEventForTaskScopeStillEmitsEventButNoProjection(t *testing.T) {
	t.Parallel()

	writer := &capturingRecallStreamWriter{}
	appMessage := message.AppMessage{
		ID:          "msg_task_stream",
		Seq:         11,
		ChannelID:   "tsk_stream_index",
		ChannelType: message.ConversationKindTask,
		Content:     "task discussion",
		CreatedAt:   time.Date(2026, 7, 15, 12, 0, 0, 0, time.UTC),
	}

	require.NoError(t, publishMessageCreatedEvent(t.Context(), writer, "wsp_1", "tsk_stream_index", appMessage, message.MessageDelivery{
		MessageID: appMessage.ID,
		Seq:       appMessage.Seq,
		Message: message.ReceiveMessage{
			MessageID:   appMessage.ID,
			TargetKey:   "tsk_stream_index",
			ChannelType: message.ConversationKindTask,
		},
	}))

	require.Len(t, writer.requests, 1, "task stream event must still be written for realtime consumers")
	require.Nil(t, writer.requests[0].Conversation, "task message must not attach an Inbox projection for the fanout path")
}

func TestDeliveryFanoutRedactsServiceAccountPublisherFromSharedEvent(t *testing.T) {
	t.Parallel()

	realtime := &fakeRealtime{}
	fanout := NewDeliveryFanout(nil, realtime, nil)
	appMessage := message.AppMessage{
		ID:          "msg_service_account_private_publisher",
		Seq:         5,
		ChannelID:   "sasc_news",
		ChannelType: "service_account",
		MessageType: message.MessageTypeServiceAccountPost,
		AuthorProfile: message.AuthorProfile{
			ActorType:   message.AuthorActorTypeService,
			PrincipalID: "sacct_news",
			MemberID:    "mem_service_news",
			DisplayName: "Workspace News",
			ServiceKind: message.AuthorServiceKindServiceAccount,
		},
		Metadata: map[string]any{
			"serviceAccountPost": map[string]any{"title": "Public title"},
			"serviceAccountPublisher": map[string]any{
				"bylineActorRef": map[string]any{"memberId": "mem_private_publisher"},
			},
		},
	}

	fanout.EmitCommittedMessage(t.Context(), "wsp_1", appMessage, message.MessageDelivery{
		MessageID: appMessage.ID,
		Seq:       appMessage.Seq,
		Message: message.ReceiveMessage{
			MessageID:   appMessage.ID,
			TargetKey:   appMessage.ChannelID,
			ChannelType: appMessage.ChannelType,
		},
	})

	require.Len(t, realtime.messageNewMessages, 1)
	require.NotContains(t, realtime.messageNewMessages[0].Metadata, "serviceAccountPublisher")
	require.Equal(t, map[string]any{"title": "Public title"}, realtime.messageNewMessages[0].Metadata["serviceAccountPost"])
}