package fanout

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

	"github.com/stretchr/testify/require"

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

type recordingScopeFanoutMarker struct {
	ids []string
	err error
}

func (marker *recordingScopeFanoutMarker) MarkScopeFanoutDeliveredForMessage(_ context.Context, messageID string) error {
	marker.ids = append(marker.ids, messageID)
	return marker.err
}

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

	ctx := context.Background()
	now := time.Date(2026, 5, 18, 8, 30, 0, 0, time.UTC)
	appMessage := message.AppMessage{
		ID:            "msg_scope_replay",
		Seq:           7,
		ChannelID:     "chn_scope_replay",
		AuthorProfile: message.AuthorProfile{ActorType: message.AuthorActorTypeHuman, DisplayName: "User"},
		Content:       "replay me",
		CreatedAt:     now,
	}
	delivery := message.MessageDelivery{
		MessageID: "msg_scope_replay",
		Seq:       7,
		Message: message.ReceiveMessage{
			MessageID: "msg_scope_replay",
			Content:   "replay me",
		},
		MentionedHumans: []message.HumanMentionRecipient{{UserID: "usr_mention", MemberID: "mem_mention"}},
		WakeRecipients:  []message.AgentRecipient{{AgentID: "agt_scope_replay", MachineID: "mach_scope_replay"}},
	}
	payload, err := message.MarshalScopeFanoutPayload(appMessage, delivery)
	require.NoError(t, err)

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

	err = adapter.Fanout(ctx, message.ScopeFanoutOutboxRecord{
		WorkspaceID:  "wsp_scope_replay",
		EventRefType: message.DeliverySourceRefMessage,
		FanoutType:   message.ScopeOutboxFanoutScope,
		Payload:      payload,
	})

	require.NoError(t, err)
	require.Equal(t, []string{"agt_scope_replay"}, hub.notified)
	require.Equal(t, []int64{7}, hub.notifySeqs)
	require.Len(t, hub.notifyMessages, 1)
	require.Equal(t, "chn_scope_replay", hub.notifyMessages[0].TargetKey)
	require.Equal(t, int64(7), hub.notifyMessages[0].TargetSeq)
	require.Empty(t, realtime.messageNew, "ordinary replay must not duplicate browser projection")
	require.Zero(t, realtime.mentions)
	require.Empty(t, unread.messages, "ordinary replay must not duplicate unread")
}

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

	const (
		workspaceID = "wsp_recoverable_stream"
		scopeID     = "chn_recoverable_stream"
		messageID   = "msg_recoverable_stream"
	)
	appMessage := message.AppMessage{
		ID: messageID, Seq: 8, ChannelID: scopeID, ChannelType: "channel",
		Content: "recover me", CreatedAt: time.Date(2026, 8, 3, 3, 30, 0, 0, time.UTC),
	}
	delivery := message.MessageDelivery{
		MessageID: messageID, Seq: appMessage.Seq,
		Message:                     message.ReceiveMessage{MessageID: messageID, TargetKey: scopeID, TargetSeq: appMessage.Seq, ChannelType: "channel"},
		RecoverableStreamProjection: true,
	}
	payload, err := message.MarshalScopeFanoutPayload(appMessage, delivery)
	require.NoError(t, err)

	realtime := &fakeRealtime{}
	unread := &capturingUnreadRuntime{}
	adapter := NewScopeFanoutReplayAdapter(NewDeliveryFanout(nil, realtime, nil).WithUnreadRuntime(unread))
	require.NoError(t, adapter.Fanout(t.Context(), message.ScopeFanoutOutboxRecord{
		WorkspaceID: workspaceID, ScopeID: scopeID, MessageID: messageID,
		EventRefType: message.DeliverySourceRefMessage, EventRefID: messageID,
		FanoutType: message.ScopeOutboxFanoutScope, Payload: payload,
	}))
	require.Len(t, realtime.messageNew, 1)
	require.Empty(t, unread.messages, "stream recovery must not claim exact ordinary unread replay")
}

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

	const (
		workspaceID = "wsp_ordinary_stream_retry"
		scopeID     = "chn_ordinary_stream_retry"
		messageID   = "msg_ordinary_stream_retry"
	)
	appMessage := message.AppMessage{
		ID:          messageID,
		Seq:         7,
		ChannelID:   scopeID,
		ChannelType: "channel",
		AuthorProfile: message.AuthorProfile{
			ActorType: message.AuthorActorTypeHuman,
			MemberID:  "mem_author",
		},
		Content:   "ordinary message",
		CreatedAt: time.Date(2026, 7, 29, 9, 30, 0, 0, time.UTC),
	}
	delivery := message.MessageDelivery{
		MessageID: messageID,
		Seq:       appMessage.Seq,
		Message: message.ReceiveMessage{
			MessageID:   messageID,
			TargetKey:   scopeID,
			TargetSeq:   appMessage.Seq,
			ChannelType: "channel",
		},
	}
	payload, err := message.MarshalScopeFanoutPayload(appMessage, delivery)
	require.NoError(t, err)

	runtime, err := unread.New(unread.Config{})
	require.NoError(t, err)
	require.NoError(t, runtime.WarmScopeRoster(workspaceID, scopeID, unread.ScopeTypeChannel, []unread.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)

	err = fanout.DeliverCommittedMessage(t.Context(), workspaceID, appMessage, delivery)
	require.ErrorIs(t, err, streamErr)
	require.Len(t, unread.messages, 1, "ordinary unread must be accounted before its non-replayable stream attempt")
	require.Equal(t, int64(1), runtime.ScopeUnreadCounts(workspaceID, "mem_reader", nil)[scopeID])
	require.Empty(t, realtime.unreadUpdated, "failed stream publication must not expose an unordered live delta")

	realtime.appendErr = nil
	adapter := NewScopeFanoutReplayAdapter(fanout)
	require.NoError(t, adapter.Fanout(t.Context(), message.ScopeFanoutOutboxRecord{
		WorkspaceID:  workspaceID,
		ScopeID:      scopeID,
		MessageID:    messageID,
		EventRefType: message.DeliverySourceRefMessage,
		EventRefID:   messageID,
		FanoutType:   message.ScopeOutboxFanoutScope,
		Payload:      payload,
	}))

	require.Len(t, unread.messages, 1, "ordinary replay must not broaden the historical audience")
	require.Equal(t, int64(1), runtime.ScopeUnreadCounts(workspaceID, "mem_reader", nil)[scopeID])
	require.Empty(t, realtime.unreadUpdated, "reconnect bootstrap repairs the already-accounted ordinary unread state")
}

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

	payload, err := message.MarshalScopeFanoutPayload(message.AppMessage{
		ID:        "msg_scope_retry",
		ChannelID: "chn_scope_retry",
	}, message.MessageDelivery{
		MessageID:      "msg_scope_retry",
		WakeRecipients: []message.AgentRecipient{{AgentID: "agt_scope_retry"}},
	})
	require.NoError(t, err)

	realtime := &fakeRealtime{}
	adapter := NewScopeFanoutReplayAdapter(NewDeliveryFanout(&fakeHub{notifyErr: errors.New("machine offline")}, realtime, nil))

	err = adapter.Fanout(context.Background(), message.ScopeFanoutOutboxRecord{
		WorkspaceID:  "wsp_scope_retry",
		EventRefType: message.DeliverySourceRefMessage,
		FanoutType:   message.ScopeOutboxFanoutScope,
		Payload:      payload,
	})

	require.Error(t, err)
	require.Contains(t, err.Error(), "machine offline")
	require.Empty(t, realtime.messageNew)
}

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

	payload, err := message.MarshalScopeFanoutPayload(message.AppMessage{
		ID:          "msg_workflow_failed",
		ChannelID:   "wfs_workflow_failed",
		ChannelType: message.WorkflowScopeType,
	}, message.MessageDelivery{
		MessageID: "msg_workflow_failed",
		Message: message.ReceiveMessage{
			ChannelType: message.WorkflowScopeType,
			ChannelName: "pln_workflow_failed",
		},
		WakeRecipients: []message.AgentRecipient{{AgentID: "agt_workflow_failed"}},
	})
	require.NoError(t, err)
	emitter := &capturingWorkflowWakeStatusEmitter{}
	adapter := NewScopeFanoutReplayAdapter(DeliveryFanout{})
	adapter.wakeStatusEmitter = emitter

	adapter.FanoutFailed(context.Background(), message.ScopeFanoutOutboxRecord{
		WorkspaceID:  "wsp_workflow_failed",
		ScopeID:      "wfs_workflow_failed",
		MessageID:    "msg_workflow_failed",
		EventRefType: message.DeliverySourceRefMessage,
		EventRefID:   "msg_workflow_failed",
		FanoutType:   message.ScopeOutboxFanoutScope,
		Payload:      payload,
	}, errors.New("machine is offline"))

	require.Len(t, emitter.calls, 1)
	call := emitter.calls[0]
	require.Equal(t, "wsp_workflow_failed", call.workspaceID)
	require.Equal(t, "wfs_workflow_failed", call.scopeID)
	require.Equal(t, "agt_workflow_failed", call.agentID)
	require.Equal(t, "msg_workflow_failed", call.messageID)
	require.Equal(t, "workflow:msg_workflow_failed", call.wakeToken)
	require.Equal(t, "workflow:pln_workflow_failed", call.target)
	require.Equal(t, "failed", call.status)
	require.Equal(t, "machine is offline", call.detail)
	_, err = time.Parse(time.RFC3339Nano, call.timestamp)
	require.NoError(t, err)
}

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

	const (
		workspaceID = "wsp_workflow_receipt"
		scopeID     = "wfs_workflow_receipt"
		messageID   = "msg_workflow_receipt"
		agentID     = "agt_workflow_receipt"
		leaseOwner  = "worker-workflow-receipt"
	)
	appMessage := message.AppMessage{
		ID: messageID, ChannelID: scopeID, ChannelType: message.WorkflowScopeType,
	}
	delivery := message.MessageDelivery{
		MessageID: messageID,
		Message: message.ReceiveMessage{
			MessageID: messageID, TargetKey: scopeID,
			ChannelType: message.WorkflowScopeType, ChannelName: "pln_workflow_receipt",
		},
		WakeRecipients: []message.AgentRecipient{{
			AgentID: agentID, MachineID: "mac_workflow_receipt",
			AgentMemberID: "mem_workflow_receipt", AutoWakeEnabled: true,
		}},
	}
	payload, err := message.MarshalScopeFanoutPayload(appMessage, delivery)
	require.NoError(t, err)
	outbox := message.ScopeFanoutOutboxRecord{
		ID: "out_workflow_receipt", WorkspaceID: workspaceID,
		ScopeID: scopeID, MessageID: messageID,
		EventRefType: message.DeliverySourceRefMessage, EventRefID: messageID,
		FanoutType: message.ScopeOutboxFanoutScope, Payload: payload,
		Status: "leased", LeaseOwner: leaseOwner,
	}
	coordinator := &leaseTestCoordinator{
		record: outbox, currentOwner: leaseOwner, receipts: map[string]struct{}{},
	}
	notifier := &perAgentFailureNotifier{
		failCount: map[string]int{agentID: 1}, calls: map[string]int{},
	}
	adapter := NewScopeFanoutReplayAdapter(
		NewDeliveryFanout(notifier, &fakeRealtime{}, nil).WithScopeFanoutMarker(coordinator),
	)

	require.ErrorContains(t, adapter.Fanout(t.Context(), outbox), "temporary daemon handoff failure")
	require.Empty(t, coordinator.receipts, "a failed daemon handoff must not authorize a Workflow reply")

	require.NoError(t, adapter.Fanout(t.Context(), outbox))
	require.Contains(
		t,
		coordinator.receipts,
		message.DeliverySourceRefMessage+"\x00"+messageID+"\x00"+agentID,
	)
	require.Equal(t, 2, notifier.calls[agentID])

	require.NoError(t, adapter.Fanout(t.Context(), outbox))
	require.Equal(t, 2, notifier.calls[agentID], "receipt replay must not redeliver an accepted Workflow request")
}

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

	payload, err := message.MarshalScopeDeliveryIntentPayload(message.ScopeDeliveryIntentPayload{
		SourceRef: message.DeliverySourceRef{
			Type: message.DeliverySourceRefActivityEvent,
			ID:   "act_payload",
		},
	})
	require.NoError(t, err)

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

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

	require.ErrorContains(t, err, "source ref mismatch")
}

type workflowWakeStatusCall struct {
	workspaceID string
	scopeID     string
	agentID     string
	messageID   string
	wakeToken   string
	target      string
	status      string
	detail      string
	timestamp   string
}

type capturingWorkflowWakeStatusEmitter struct {
	calls []workflowWakeStatusCall
}

func (emitter *capturingWorkflowWakeStatusEmitter) EmitAgentWakeStatusForScope(
	workspaceID string,
	scopeID string,
	agentID string,
	messageID string,
	wakeToken string,
	target string,
	status string,
	_ string,
	detail string,
	timestamp string,
) {
	emitter.calls = append(emitter.calls, workflowWakeStatusCall{
		workspaceID: workspaceID,
		scopeID:     scopeID,
		agentID:     agentID,
		messageID:   messageID,
		wakeToken:   wakeToken,
		target:      target,
		status:      status,
		detail:      detail,
		timestamp:   timestamp,
	})
}

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

	payload, err := message.MarshalScopeDeliveryIntentPayload(message.ScopeDeliveryIntentPayload{
		SourceRef: message.DeliverySourceRef{
			Type: message.DeliverySourceRefActivityEvent,
			ID:   "act_with_message_id",
		},
	})
	require.NoError(t, err)

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

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

	require.ErrorContains(t, err, "must not include message_id")
}

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

	marker := &recordingScopeFanoutMarker{}
	fanout := NewDeliveryFanout(&fakeHub{}, &fakeRealtime{}, nil).WithScopeFanoutMarker(marker)
	appMessage := message.AppMessage{ID: "msg_root", ChannelID: "chn_1"}
	fanout.EmitCommittedMessage(context.Background(), "wsp_1", appMessage, message.MessageDelivery{
		MessageID: "msg_root",
		AppMessage: &message.AppMessage{
			ID:        "msg_root",
			ChannelID: "chn_1",
		},
		FollowupDeliveries: []message.MessageDelivery{{
			MessageID: "msg_followup",
			AppMessage: &message.AppMessage{
				ID:        "msg_followup",
				ChannelID: "chn_1",
			},
		}},
	})

	require.Equal(t, []string{"msg_root", "msg_followup"}, marker.ids)
}

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

	appMessage := message.AppMessage{
		ID:        "msg_retry_no_duplicate",
		Seq:       11,
		ChannelID: "chn_retry_no_duplicate",
		Content:   "hello",
		CreatedAt: time.Date(2026, 5, 18, 9, 0, 0, 0, time.UTC),
	}
	delivery := message.MessageDelivery{
		MessageID: "msg_retry_no_duplicate",
		Seq:       11,
		Message: message.ReceiveMessage{
			MessageID: "msg_retry_no_duplicate",
			Content:   "hello",
		},
		WakeRecipients: []message.AgentRecipient{{AgentID: "agt_retry_no_duplicate"}},
	}
	payload, err := message.MarshalScopeFanoutPayload(appMessage, delivery)
	require.NoError(t, err)

	marker := &recordingScopeFanoutMarker{}
	realtime := &fakeRealtime{}
	fanout := NewDeliveryFanout(&fakeHub{notifyErr: errors.New("machine offline")}, realtime, nil).WithScopeFanoutMarker(marker)
	fanout.EmitCommittedMessage(context.Background(), "wsp_1", appMessage, delivery)
	require.Equal(t, []string{"chn_retry_no_duplicate"}, realtime.messageNew)
	require.Empty(t, marker.ids)

	retryHub := &fakeHub{}
	adapter := NewScopeFanoutReplayAdapter(NewDeliveryFanout(retryHub, realtime, nil))
	err = adapter.Fanout(context.Background(), message.ScopeFanoutOutboxRecord{
		WorkspaceID:  "wsp_1",
		EventRefType: message.DeliverySourceRefMessage,
		FanoutType:   message.ScopeOutboxFanoutScope,
		Payload:      payload,
	})

	require.NoError(t, err)
	require.Equal(t, []string{"agt_retry_no_duplicate"}, retryHub.notified)
	require.Equal(t, []string{"chn_retry_no_duplicate"}, realtime.messageNew)
}

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

	realtime := &fakeRealtime{}
	marker := &recordingScopeFanoutMarker{}
	fanout := NewDeliveryFanout(&fakeHub{notifyErr: errors.New("machine offline")}, realtime, nil).
		WithScopeFanoutMarker(marker)
	fanout.DeliverFollowupMessages(context.Background(), "wsp_1", []message.MessageDelivery{{
		MessageID: "msg_followup_failed",
		AppMessage: &message.AppMessage{
			ID:        "msg_followup_failed",
			ChannelID: "chn_1",
		},
		WakeRecipients: []message.AgentRecipient{{AgentID: "agt_failed"}},
	}})

	require.Equal(t, []string{"chn_1"}, realtime.messageNew)
	require.Zero(t, realtime.mentions)
	require.Empty(t, marker.ids, "failed Agent wake must leave the durable outbox pending")
}

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

	// Captured v1 message-source shape emitted by pre-cutover nodes. Keep this
	// literal independent of the current marshaler so stored payload decoding
	// cannot regress silently when the in-memory structs evolve.
	payload := []byte(`{
	  "version": 1,
	  "appMessage": {
	    "ID": "msg_hidden_dispatch",
	    "Seq": 23,
	    "ChannelID": "tsk_hidden_dispatch",
	    "MessageType": "task_dispatch",
	    "Content": "Assigned #t23 rolling compatibility",
	    "Metadata": {"task": {"taskNumber": 23}}
	  },
	  "delivery": {
	    "MessageID": "msg_hidden_dispatch",
	    "Seq": 23,
	    "Message": {
	      "MessageID": "msg_hidden_dispatch",
	      "TargetKey": "tsk_hidden_dispatch",
	      "ChannelType": "task",
	      "MessageType": "task_dispatch",
	      "Content": "Assigned #t23 rolling compatibility",
	      "Metadata": {"task": {"taskNumber": 23}}
	    },
	    "WakeRecipients": [{
	      "AgentID": "agt_hidden",
	      "MachineID": "mach_hidden",
	      "AgentMemberID": "mem_hidden",
	      "AutoWakeEnabled": true,
	      "DeliveryReason": "task_command_notify"
	    }]
	  }
	}`)

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

	err := adapter.Fanout(context.Background(), message.ScopeFanoutOutboxRecord{
		WorkspaceID:  "wsp_hidden_dispatch",
		MessageID:    "msg_hidden_dispatch",
		ScopeID:      "tsk_hidden_dispatch",
		EventRefType: message.DeliverySourceRefMessage,
		EventRefID:   "msg_hidden_dispatch",
		FanoutType:   message.ScopeOutboxFanoutScope,
		Payload:      payload,
	})

	require.NoError(t, err)
	require.Equal(t, []string{"agt_hidden"}, hub.notified)
	require.Equal(t, []int64{23}, hub.notifySeqs)
	require.Len(t, hub.notifyMessages, 1)
	require.Equal(t, message.MessageTypeTaskDispatch, hub.notifyMessages[0].MessageType)
	require.Equal(t, "msg_hidden_dispatch", hub.notifyMessages[0].MessageID)
	require.Equal(t, "tsk_hidden_dispatch", hub.notifyMessages[0].TargetKey)
	require.Equal(t, int64(23), hub.notifyMessages[0].TargetSeq)
	require.Equal(t, "Assigned #t23 rolling compatibility", hub.notifyMessages[0].Content)
	require.Equal(t, float64(23), hub.notifyMessages[0].Metadata["task"].(map[string]any)["taskNumber"])
	require.Empty(t, realtime.messageNew)
	require.Zero(t, realtime.mentions)
	require.Empty(t, unread.messages)
	require.Empty(t, unread.activities)
}

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

	// Captured v1 delivery-intent shape emitted by the intermediate
	// dispatch_requested implementation. agentHandoff was removed later and
	// intentionally remains here to pin tolerant forward decoding.
	payload := []byte(`{
	  "version": 1,
	  "sourceRef": {"type": "activity_event", "id": "act_legacy_dispatch"},
	  "message": {
	    "MessageID": "act_legacy_dispatch",
	    "TargetKey": "tsk_legacy_dispatch",
	    "TargetSeq": 31,
	    "ChannelType": "task",
	    "MessageType": "task_dispatch",
	    "Content": "Started #t31 rolling compatibility",
	    "Metadata": {"task": {"taskNumber": 31}}
	  },
	  "wakeRecipients": [{
	    "AgentID": "agt_legacy_intent",
	    "MachineID": "mach_legacy_intent",
	    "AgentMemberID": "mem_legacy_intent",
	    "AutoWakeEnabled": true,
	    "DeliveryReason": "task_command_notify"
	  }],
	  "realtimePolicy": "none",
	  "attentionPolicy": "wake",
	  "agentHandoff": "standard"
	}`)

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

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

	require.NoError(t, err)
	require.Equal(t, []string{"agt_legacy_intent"}, hub.notified)
	require.Equal(t, []int64{31}, hub.notifySeqs)
	require.Len(t, hub.notifyMessages, 1)
	require.Equal(t, message.MessageTypeTaskDispatch, hub.notifyMessages[0].MessageType)
	require.Equal(t, "act_legacy_dispatch", hub.notifyMessages[0].MessageID)
	require.Equal(t, "tsk_legacy_dispatch", hub.notifyMessages[0].TargetKey)
	require.Equal(t, int64(31), hub.notifyMessages[0].TargetSeq)
	require.Equal(t, "Started #t31 rolling compatibility", hub.notifyMessages[0].Content)
	require.Equal(t, float64(31), hub.notifyMessages[0].Metadata["task"].(map[string]any)["taskNumber"])
	require.Empty(t, realtime.messageNew)
	require.Zero(t, realtime.mentions)
	require.Empty(t, unread.messages)
	require.Empty(t, unread.activities)
}