package fanout

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

	"github.com/stretchr/testify/require"

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

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

	now := time.Date(2026, 7, 29, 1, 0, 0, 0, time.UTC)
	appMessage := message.AppMessage{
		ID:          "msg_task_replay",
		Seq:         9,
		ChannelID:   "tsk_scope_replay",
		ChannelType: "task",
		MessageType: message.MessageTypeTaskDiscussion,
		AuthorProfile: message.AuthorProfile{
			ActorType:   message.AuthorActorTypeHuman,
			MemberID:    "mem_author",
			DisplayName: "Author",
		},
		Content:   "targeted Task discussion",
		CreatedAt: now,
	}
	delivery := message.MessageDelivery{
		MessageID:               appMessage.ID,
		Seq:                     appMessage.Seq,
		HumanRecipientMemberIDs: []string{"mem_reader"},
		HumanRecipientsTargeted: true,
		Message: message.ReceiveMessage{
			MessageID:   appMessage.ID,
			TargetKey:   appMessage.ChannelID,
			TargetSeq:   appMessage.Seq,
			ChannelType: "task",
		},
		Conversation: &message.DeliveryConversationProjection{
			ConversationID: "task:" + appMessage.ChannelID,
			ScopeID:        appMessage.ChannelID,
			Kind:           message.ConversationKindTask,
			SortAt:         now,
			Projection: map[string]any{
				"id":      "task:" + appMessage.ChannelID,
				"kind":    "task",
				"scopeId": appMessage.ChannelID,
				"route":   "/channels/chn_parent/tasks/9",
				"title":   "#t9 Repair replay",
				"task": map[string]any{
					"id":        "task_9",
					"scopeId":   appMessage.ChannelID,
					"channelId": "chn_parent",
					"number":    int64(9),
					"title":     "Repair replay",
					"status":    "in_progress",
					"route":     "/channels/chn_parent/tasks/9",
				},
			},
		},
	}
	payload, err := message.MarshalScopeFanoutPayload(appMessage, delivery)
	require.NoError(t, err)

	runtime, err := unreadservice.New(unreadservice.Config{})
	require.NoError(t, err)
	require.NoError(t, runtime.WarmScopeRoster("wsp_task_replay", appMessage.ChannelID, unreadservice.ScopeTypeTask, []unreadservice.RosterMember{
		{MemberID: "mem_author", UserID: "usr_author", Visibility: "public", NotificationState: "mute"},
		{MemberID: "mem_reader", UserID: "usr_reader", Visibility: "public", NotificationState: "mute"},
		{MemberID: "mem_bystander", UserID: "usr_bystander", Visibility: "public", NotificationState: "active"},
	}))
	unread := &capturingUnreadRuntime{runtime: runtime}
	realtime := &fakeRealtime{}
	adapter := NewScopeFanoutReplayAdapter(NewDeliveryFanout(nil, realtime, nil).WithUnreadRuntime(unread))
	outbox := message.ScopeFanoutOutboxRecord{
		WorkspaceID:  "wsp_task_replay",
		ScopeID:      appMessage.ChannelID,
		MessageID:    appMessage.ID,
		EventRefType: message.DeliverySourceRefMessage,
		EventRefID:   appMessage.ID,
		FanoutType:   message.ScopeOutboxFanoutScope,
		Payload:      payload,
	}

	require.NoError(t, adapter.Fanout(t.Context(), outbox))
	require.Equal(t, int64(1), runtime.ScopeUnreadCounts("wsp_task_replay", "mem_reader", nil)[appMessage.ChannelID])
	require.Zero(t, runtime.ScopeUnreadCounts("wsp_task_replay", "mem_bystander", nil)[appMessage.ChannelID])
	require.Len(t, realtime.messageNew, 1)
	require.Len(t, unread.messages, 1)
	require.Equal(t, []string{"mem_reader"}, unread.messages[0].RecipientMemberIDs)

	require.NoError(t, adapter.Fanout(t.Context(), outbox))
	require.Equal(t, int64(1), runtime.ScopeUnreadCounts("wsp_task_replay", "mem_reader", nil)[appMessage.ChannelID])
	require.Zero(t, runtime.ScopeUnreadCounts("wsp_task_replay", "mem_bystander", nil)[appMessage.ChannelID])
	require.Len(t, realtime.messageNew, 1, "stable message.created id must make replay idempotent")
	require.Len(t, unread.messages, 2, "both retries reapply the same idempotent runtime fact")
}

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

	writer := &capturingRecallStreamWriter{}
	runtime, err := unreadservice.New(unreadservice.Config{})
	require.NoError(t, err)
	unread := &capturingUnreadRuntime{runtime: runtime}
	now := time.Date(2026, 7, 29, 1, 5, 0, 0, time.UTC)
	appMessage := message.AppMessage{
		ID:          "msg_task_agent_only",
		Seq:         10,
		ChannelID:   "tsk_agent_only",
		ChannelType: "task",
		MessageType: message.MessageTypeTaskDiscussion,
		AuthorProfile: message.AuthorProfile{
			ActorType:   message.AuthorActorTypeHuman,
			MemberID:    "mem_actor",
			DisplayName: "Actor",
		},
		CreatedAt: now,
	}
	delivery := message.MessageDelivery{
		MessageID:               appMessage.ID,
		Seq:                     appMessage.Seq,
		HumanRecipientsTargeted: true,
		Message: message.ReceiveMessage{
			MessageID:   appMessage.ID,
			TargetKey:   appMessage.ChannelID,
			TargetSeq:   appMessage.Seq,
			ChannelType: "task",
		},
		Conversation: &message.DeliveryConversationProjection{
			ConversationID: "task:" + appMessage.ChannelID,
			ScopeID:        appMessage.ChannelID,
			Kind:           message.ConversationKindTask,
			SortAt:         now,
			Projection: map[string]any{
				"id":      "task:" + appMessage.ChannelID,
				"kind":    "task",
				"scopeId": appMessage.ChannelID,
				"route":   "/channels/chn_parent/tasks/10",
				"title":   "#t10 Agent only",
			},
		},
	}

	err = NewDeliveryFanout(nil, writer, nil).WithUnreadRuntime(unread).DeliverCommittedMessage(
		t.Context(),
		"wsp_agent_only",
		appMessage,
		delivery,
	)

	require.NoError(t, err)
	require.Len(t, writer.requests, 1)
	require.True(t, writer.requests[0].TargetedRecipients)
	require.Empty(t, writer.requests[0].RecipientMemberIDs)
	require.Empty(t, unread.messages, "an explicit empty human audience must not hydrate or account the Task scope")
}

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

	payload, err := message.MarshalScopeDeliveryIntentPayload(message.ScopeDeliveryIntentPayload{
		SourceRef: message.DeliverySourceRef{
			Type: message.DeliverySourceRefActivityEvent,
			ID:   "act_scope_replay",
		},
		Message: message.ReceiveMessage{
			MessageID: "act_scope_replay",
			TargetKey: "tsk_scope_replay",
			TargetSeq: 14,
			Content:   "activity wake",
		},
		WakeRecipients: []message.AgentRecipient{{AgentID: "agt_activity", MachineID: "mach_activity"}},
	})
	require.NoError(t, err)

	hub := &fakeHub{}
	realtime := &fakeRealtime{}
	adapter := NewScopeFanoutReplayAdapter(NewDeliveryFanout(hub, realtime, nil))

	err = adapter.Fanout(context.Background(), message.ScopeFanoutOutboxRecord{
		WorkspaceID:  "wsp_activity",
		EventRefType: message.DeliverySourceRefActivityEvent,
		EventRefID:   "act_scope_replay",
		FanoutType:   message.ScopeOutboxFanoutScope,
		Payload:      payload,
	})

	require.NoError(t, err)
	require.Equal(t, []string{"agt_activity"}, hub.notified)
	require.Equal(t, []int64{14}, hub.notifySeqs)
	require.Len(t, hub.notifyMessages, 1)
	require.Equal(t, "tsk_scope_replay", hub.notifyMessages[0].TargetKey)
	require.Equal(t, "activity wake", hub.notifyMessages[0].Content)
	require.Empty(t, realtime.messageNew)
	require.Zero(t, realtime.mentions)
}

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

	const (
		workspaceID = "wsp_activity_pending"
		scopeID     = "tsk_activity_pending"
		eventID     = "act_activity_pending"
	)
	now := time.Date(2026, 7, 29, 2, 15, 0, 0, time.UTC)
	payload, err := message.MarshalScopeDeliveryIntentPayload(message.ScopeDeliveryIntentPayload{
		SourceRef: message.DeliverySourceRef{
			Type: message.DeliverySourceRefActivityEvent,
			ID:   eventID,
		},
		Message: message.ReceiveMessage{
			MessageID:   eventID,
			TargetKey:   scopeID,
			TargetSeq:   17,
			ChannelType: message.ConversationKindTask,
			MessageType: "task.status_changed",
			Content:     "Changed status to in_review",
			AuthorProfile: message.AuthorProfile{
				MemberID: "mem_author",
			},
			Timestamp: now,
		},
		WakeRecipients: []message.AgentRecipient{{
			AgentID:       "agt_activity_pending",
			MachineID:     "mach_activity_pending",
			AgentMemberID: "mem_agent_activity_pending",
		}},
		HumanRecipientMemberIDs: []string{"mem_reader"},
		HumanRecipientsTargeted: true,
		Conversation: &message.DeliveryConversationProjection{
			ConversationID: "task:" + scopeID,
			ScopeID:        scopeID,
			Kind:           message.ConversationKindTask,
			SortAt:         now,
			Projection: map[string]any{
				"id":      "task:" + scopeID,
				"kind":    message.ConversationKindTask,
				"scopeId": scopeID,
				"title":   "#t17 Pending activity",
			},
		},
		AttentionPolicy: message.DeliveryAttentionWake,
	})
	require.NoError(t, err)
	outbox := message.ScopeFanoutOutboxRecord{
		WorkspaceID:  workspaceID,
		ScopeID:      scopeID,
		EventRefType: message.DeliverySourceRefActivityEvent,
		EventRefID:   eventID,
		FanoutType:   message.ScopeOutboxFanoutScope,
		Payload:      payload,
	}

	runtime, err := unreadservice.New(unreadservice.Config{})
	require.NoError(t, err)
	unread := &capturingUnreadRuntime{runtime: runtime}
	realtime := &fakeRealtime{}
	hub := &fakeHub{}
	adapter := NewScopeFanoutReplayAdapter(
		NewDeliveryFanout(hub, realtime, nil).WithUnreadRuntime(unread),
	)

	err = adapter.Fanout(t.Context(), outbox)
	require.ErrorIs(t, err, errTargetedTaskUnreadPending)
	require.Len(t, realtime.seenEventIDs, 1, "the durable Task row projection may precede runtime hydration")
	require.Empty(t, hub.notified, "pending unread must keep the outbox retryable before Agent wake")
	require.Len(t, unread.activities, 1)

	require.NoError(t, runtime.WarmScopeRoster(workspaceID, scopeID, unreadservice.ScopeTypeTask, []unreadservice.RosterMember{
		{MemberID: "mem_author", UserID: "usr_author", Visibility: "public", NotificationState: "mute"},
		{MemberID: "mem_reader", UserID: "usr_reader", Visibility: "public", NotificationState: "mute"},
	}))

	require.NoError(t, adapter.Fanout(t.Context(), outbox))
	require.Len(t, realtime.seenEventIDs, 1, "stable activity event ids make the retry idempotent")
	require.Equal(t, []string{"agt_activity_pending"}, hub.notified)
	require.Len(t, unread.activities, 2)
	require.Equal(t, int64(1), runtime.ScopeUnreadCounts(workspaceID, "mem_reader", nil)[scopeID])
}

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

	const (
		workspaceID = "wsp_activity_missing_unread"
		scopeID     = "tsk_activity_missing_unread"
		eventID     = "act_activity_missing_unread"
		outboxID    = "out_activity_missing_unread"
		leaseOwner  = "worker:activity_missing_unread"
	)
	payload, err := message.MarshalScopeDeliveryIntentPayload(message.ScopeDeliveryIntentPayload{
		SourceRef: message.DeliverySourceRef{
			Type: message.DeliverySourceRefActivityEvent,
			ID:   eventID,
		},
		Message: message.ReceiveMessage{
			MessageID:   eventID,
			TargetKey:   scopeID,
			TargetSeq:   18,
			ChannelType: message.ConversationKindTask,
			MessageType: "task.status_changed",
		},
		WakeRecipients: []message.AgentRecipient{{
			AgentID:   "agt_activity_missing_unread",
			MachineID: "mach_activity_missing_unread",
		}},
		HumanRecipientMemberIDs: []string{"mem_reader"},
		HumanRecipientsTargeted: true,
		Conversation: &message.DeliveryConversationProjection{
			ConversationID: "task:" + scopeID,
			ScopeID:        scopeID,
			Kind:           message.ConversationKindTask,
		},
	})
	require.NoError(t, err)
	outbox := message.ScopeFanoutOutboxRecord{
		ID:           outboxID,
		WorkspaceID:  workspaceID,
		ScopeID:      scopeID,
		EventRefType: message.DeliverySourceRefActivityEvent,
		EventRefID:   eventID,
		FanoutType:   message.ScopeOutboxFanoutScope,
		Payload:      payload,
		Status:       "leased_v2",
		LeaseOwner:   leaseOwner,
	}
	coordinator := &leaseTestCoordinator{
		record:       outbox,
		currentOwner: leaseOwner,
		receipts:     map[string]struct{}{},
	}
	hub := &fakeHub{}
	adapter := NewScopeFanoutReplayAdapter(
		NewDeliveryFanout(hub, &fakeRealtime{}, nil).
			WithScopeFanoutMarker(coordinator),
	)

	err = adapter.Fanout(t.Context(), outbox)
	require.ErrorIs(t, err, message.ErrScopeFanoutDeferred)
	require.ErrorIs(t, err, errTargetedTaskUnreadPending)
	require.Empty(t, hub.notified)
}

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

	for _, notifierFailure := range []string{"machine offline", "machine manually stopped"} {
		notifierFailure := notifierFailure
		t.Run(notifierFailure, func(t *testing.T) {
			t.Parallel()

			now := time.Date(2026, 7, 29, 1, 10, 0, 0, time.UTC)
			payload, err := message.MarshalScopeDeliveryIntentPayload(message.ScopeDeliveryIntentPayload{
				SourceRef: message.DeliverySourceRef{
					Type: message.DeliverySourceRefActivityEvent,
					ID:   "act_task_retry_" + notifierFailure,
				},
				Message: message.ReceiveMessage{
					MessageID:   "act_task_retry_" + notifierFailure,
					TargetKey:   "tsk_activity_retry",
					TargetSeq:   14,
					ChannelType: "task",
					MessageType: "task.status_changed",
					Content:     "Changed status to in_review",
					Timestamp:   now,
				},
				WakeRecipients: []message.AgentRecipient{{
					AgentID:       "agt_activity_retry",
					MachineID:     "mach_activity_retry",
					AgentMemberID: "mem_agent_activity_retry",
				}},
				HumanRecipientMemberIDs: []string{"mem_reader"},
				HumanRecipientsTargeted: true,
				Conversation: &message.DeliveryConversationProjection{
					ConversationID: "task:tsk_activity_retry",
					ScopeID:        "tsk_activity_retry",
					Kind:           message.ConversationKindTask,
					SortAt:         now,
					Projection: map[string]any{
						"id":      "task:tsk_activity_retry",
						"kind":    "task",
						"scopeId": "tsk_activity_retry",
						"route":   "/channels/chn_parent/tasks/14",
						"title":   "#t14 Retry activity",
						"task": map[string]any{
							"id":        "task_14",
							"scopeId":   "tsk_activity_retry",
							"channelId": "chn_parent",
							"number":    int64(14),
							"title":     "Retry activity",
							"status":    "in_review",
							"route":     "/channels/chn_parent/tasks/14",
						},
					},
				},
				AttentionPolicy: message.DeliveryAttentionWake,
			})
			require.NoError(t, err)

			runtime, err := unreadservice.New(unreadservice.Config{})
			require.NoError(t, err)
			require.NoError(t, runtime.WarmScopeRoster("wsp_activity_retry", "tsk_activity_retry", unreadservice.ScopeTypeTask, []unreadservice.RosterMember{
				{MemberID: "mem_reader", UserID: "usr_reader", Visibility: "public", NotificationState: "mute"},
				{MemberID: "mem_bystander", UserID: "usr_bystander", Visibility: "public", NotificationState: "active"},
			}))
			unread := &capturingUnreadRuntime{runtime: runtime}
			realtime := &fakeRealtime{}
			hub := &fakeHub{notifyErr: errors.New(notifierFailure)}
			adapter := NewScopeFanoutReplayAdapter(
				NewDeliveryFanout(hub, realtime, nil).WithUnreadRuntime(unread),
			)
			outbox := message.ScopeFanoutOutboxRecord{
				WorkspaceID:  "wsp_activity_retry",
				ScopeID:      "tsk_activity_retry",
				EventRefType: message.DeliverySourceRefActivityEvent,
				EventRefID:   "act_task_retry_" + notifierFailure,
				FanoutType:   message.ScopeOutboxFanoutScope,
				Payload:      payload,
			}

			err = adapter.Fanout(t.Context(), outbox)
			require.ErrorContains(t, err, notifierFailure)
			require.Equal(t, int64(1), runtime.ScopeUnreadCounts("wsp_activity_retry", "mem_reader", nil)["tsk_activity_retry"])
			require.Zero(t, runtime.ScopeUnreadCounts("wsp_activity_retry", "mem_bystander", nil)["tsk_activity_retry"])
			require.Len(t, realtime.messageNew, 0)
			require.Len(t, realtime.seenEventIDs, 1)
			require.Len(t, unread.activities, 1)

			// Retrying the same durable intent re-applies stable Human effects
			// idempotently while still surfacing the Agent handoff failure.
			err = adapter.Fanout(t.Context(), outbox)
			require.ErrorContains(t, err, notifierFailure)
			require.Equal(t, int64(1), runtime.ScopeUnreadCounts("wsp_activity_retry", "mem_reader", nil)["tsk_activity_retry"])
			require.Len(t, realtime.seenEventIDs, 1)
			require.Len(t, unread.activities, 2)
		})
	}
}

// TestScopeFanoutReplayPreservesProviderTaskProjection proves the prior Senior
// review's authoritative requirement: when the durable outbox replays a
// provider-service Task discussion after a failed first fanout, the frozen
// payload must preserve notify principal, service author projection,
// contribution source, targeted Human unread/notification, and watcher Agent
// wake. The payload shape below mirrors exactly what
// appendServiceEventTaskDiscussionTx now persists through
// projectTaskScopeDiscussionDeliveryTx.
//
// Unlike a direct adapter double-call, this test drives the real
// pending_v2 lease/receipt coordinator path (leaseTestCoordinator +
// perAgentFailureNotifier). The first fanout fails after one Agent recipient
// (agt_a) has already accepted: receipt coordination records agt_a, and agt_b's
// handoff fails. On retry the coordinator skips the already-accepted agt_a and
// only retries the unaccepted agt_b handoff, so agt_a is woken exactly once
// while agt_b is woken twice (first attempt + retry).
func TestScopeFanoutReplayPreservesProviderTaskProjection(t *testing.T) {
	t.Parallel()

	const (
		workspaceID = "wsp_provider_task_replay"
		scopeID     = "tsk_provider_task_replay"
		messageID   = "msg_provider_task_replay"
		readerID    = "mem_provider_reader"
		watcherA    = "agt_provider_watcher_a"
		watcherAMem = "mem_provider_watcher_a"
		watcherB    = "agt_provider_watcher_b"
		watcherBMem = "mem_provider_watcher_b"
		principalID = "mem_provider_principal"
	)
	now := time.Date(2026, 8, 14, 4, 0, 0, 0, time.UTC)
	authorProfile := message.AuthorProfile{
		ActorType:   message.AuthorActorTypeService,
		PrincipalID: "app_install_1",
		MemberID:    "mem_provider_service",
		TargetID:    "github-bot",
		DisplayName: "GitHub",
		ServiceKind: message.AuthorServiceKindApp,
		ServiceSlug: "github",
	}.Normalize()
	source := message.ContributionSourceForServiceEventDelivery(message.ServiceEventDeliveryContributionSourceParams{
		DeliveryID:  "svced_provider_task_replay",
		Label:       "GitHub pull request subscription",
		EventID:     "evt_provider_task_replay",
		EventType:   "github.pull_request",
		EventAction: "comment_created",
	})
	appMessage := message.AppMessage{
		ID:            messageID,
		Seq:           1,
		ChannelID:     scopeID,
		ChannelType:   message.ConversationKindTask,
		MessageType:   message.MessageTypeTaskDiscussion,
		AuthorProfile: authorProfile,
		Source:        source,
		Content:       "provider event discussion",
		CreatedAt:     now,
	}
	delivery := message.MessageDelivery{
		MessageID:               messageID,
		Seq:                     1,
		ThreadID:                scopeID,
		NotifyPrincipalMemberID: principalID,
		HumanRecipientMemberIDs: []string{readerID},
		HumanRecipientsTargeted: true,
		WakeRecipients: []message.AgentRecipient{
			{AgentID: watcherA, AgentMemberID: watcherAMem, MachineID: "mac_provider_watcher_a", AutoWakeEnabled: true},
			{AgentID: watcherB, AgentMemberID: watcherBMem, MachineID: "mac_provider_watcher_b", AutoWakeEnabled: true},
		},
		Message: message.ReceiveMessage{
			MessageID:               messageID,
			TargetKey:               scopeID,
			TargetSeq:               1,
			ChannelType:             message.ConversationKindTask,
			NotifyPrincipalMemberID: principalID,
			AuthorProfile:           authorProfile,
			Source:                  source,
			Content:                 "provider event discussion",
			Timestamp:               now,
		},
		Conversation: &message.DeliveryConversationProjection{
			ConversationID: "task:" + scopeID,
			ScopeID:        scopeID,
			Kind:           message.ConversationKindTask,
			SortAt:         now,
			Projection: map[string]any{
				"id":      "task:" + scopeID,
				"kind":    "task",
				"scopeId": scopeID,
			},
		},
		MentionContext: message.MentionContext{ChannelType: message.ConversationKindTask},
	}
	delivery.AppMessage = &appMessage

	coordinator := &leaseTestCoordinator{
		available: true,
		receipts:  map[string]struct{}{},
		record: message.ScopeFanoutOutboxRecord{
			ID:           "out_provider_task_replay",
			WorkspaceID:  workspaceID,
			ScopeID:      scopeID,
			MessageID:    messageID,
			EventRefType: message.DeliverySourceRefMessage,
			EventRefID:   messageID,
			FanoutType:   message.ScopeOutboxFanoutScope,
			Status:       "pending_v2",
			MaxAttempts:  3,
		},
	}
	// agt_a accepts on every attempt; agt_b's first handoff fails. The retry
	// therefore skips agt_a (receipt present) and only re-attempts agt_b.
	notifier := &providerReplayCaptureNotifier{
		failCount: map[string]int{watcherB: 1},
		calls:     map[string]int{},
		messages:  map[string][]message.ReceiveMessage{},
	}
	runtime, err := unreadservice.New(unreadservice.Config{})
	require.NoError(t, err)
	require.NoError(t, runtime.WarmScopeRoster(workspaceID, scopeID, unreadservice.ScopeTypeTask, []unreadservice.RosterMember{
		{MemberID: readerID, UserID: "usr_provider_reader", Visibility: "public", NotificationState: "mute"},
	}))
	unread := &capturingUnreadRuntime{runtime: runtime}
	realtime := &fakeRealtime{}
	fanout := NewDeliveryFanout(notifier, realtime, nil).
		WithScopeFanoutMarker(coordinator).
		WithUnreadRuntime(unread).
		WithUnreadDeltaSink(realtime)

	// First fanout: unread + browser projection settle, agt_a accepts and its
	// receipt is recorded, then agt_b's handoff fails and the lease is released
	// for retry.
	err = fanout.DeliverCommittedMessage(t.Context(), workspaceID, appMessage, delivery)
	require.ErrorContains(t, err, "temporary daemon handoff failure")
	require.Equal(t, 1, coordinator.retries)
	require.Equal(t, 0, coordinator.completions)
	require.Equal(t, int64(1), runtime.ScopeUnreadCounts(workspaceID, readerID, nil)[scopeID], "targeted Human recipient receives unread before the failed Agent handoff")
	require.Len(t, realtime.messageNew, 1, "browser projection settles before the Agent handoff attempt")
	require.Len(t, coordinator.receipts, 1, "only the accepted watcher agent is recorded")
	_, accepted := coordinator.receipts[message.DeliverySourceRefMessage+"\x00"+messageID+"\x00"+watcherA]
	require.True(t, accepted, "the first watcher's accepted handoff must be receipted")

	// Retry: the already-accepted watcher A is skipped (no duplicate wake), and
	// only watcher B's unaccepted handoff is retried and accepted. The frozen
	// notify principal, service author projection, contribution source, targeted
	// unread, and browser projection all survive the retry.
	require.NoError(t, fanout.DeliverCommittedMessage(t.Context(), workspaceID, appMessage, delivery))
	require.Equal(t, 2, coordinator.claims)
	require.Equal(t, 1, coordinator.completions)
	require.Equal(t, map[string]int{watcherA: 1, watcherB: 2}, notifier.calls, "watcher A is woken exactly once; watcher B is retried")
	require.Len(t, coordinator.receipts, 2)
	require.Len(t, realtime.messageNew, 1, "browser projection is idempotent by stable message event id across retry")
	require.Equal(t, int64(1), runtime.ScopeUnreadCounts(workspaceID, readerID, nil)[scopeID], "targeted unread is idempotent across retry")
	require.Len(t, unread.messages, 2, "both attempts reapply the same idempotent targeted unread runtime fact")

	// The retried watcher B wake must carry the frozen projection exactly as the
	// durable payload froze it: notify principal, service author, contribution
	// source, and the Task channel type.
	require.Len(t, notifier.messages[watcherB], 2, "watcher B is woken on the failed first attempt and the successful retry")
	retried := notifier.messages[watcherB][1]
	require.Equal(t, principalID, retried.NotifyPrincipalMemberID, "retry must preserve the frozen notify principal on the Agent wake")
	require.Equal(t, authorProfile, retried.AuthorProfile, "retry must preserve the frozen service author projection")
	require.Equal(t, source, retried.Source, "retry must preserve the frozen contribution source")
	require.Equal(t, message.ConversationKindTask, retried.ChannelType)
	// watcher A's single accepted wake carries the same frozen projection.
	require.Len(t, notifier.messages[watcherA], 1)
	require.Equal(t, principalID, notifier.messages[watcherA][0].NotifyPrincipalMemberID)
	require.Equal(t, authorProfile, notifier.messages[watcherA][0].AuthorProfile)
	require.Equal(t, source, notifier.messages[watcherA][0].Source)
}

// providerReplayCaptureNotifier records every wake message per recipient while
// failing the first N handoffs for a configured recipient set, mirroring the
// perAgentFailureNotifier used by the lease/receipt coordinator tests.
type providerReplayCaptureNotifier struct {
	failCount map[string]int
	calls     map[string]int
	messages  map[string][]message.ReceiveMessage
}

func (notifier *providerReplayCaptureNotifier) NotifyAgentMessage(_ context.Context, recipient message.AgentRecipient, _ int64, msg message.ReceiveMessage) error {
	notifier.calls[recipient.AgentID]++
	notifier.messages[recipient.AgentID] = append(notifier.messages[recipient.AgentID], msg)
	if notifier.failCount[recipient.AgentID] > 0 {
		notifier.failCount[recipient.AgentID]--
		return errors.New("temporary daemon handoff failure")
	}
	return nil
}