package unread

import (
	"testing"
	"time"
)

func TestDefaultConfig(t *testing.T) {
	runtime, err := New(Config{})
	if err != nil {
		t.Fatalf("New(Config{}) returned error: %v", err)
	}

	cfg := runtime.Config()
	if cfg.SnapshotFlushInterval != 30*time.Second {
		t.Fatalf("SnapshotFlushInterval = %v, want 30s", cfg.SnapshotFlushInterval)
	}
	if cfg.SnapshotFlushBatchSize != 1000 {
		t.Fatalf("SnapshotFlushBatchSize = %d, want 1000", cfg.SnapshotFlushBatchSize)
	}
	if cfg.HydrateMaxConcurrency != 2 {
		t.Fatalf("HydrateMaxConcurrency = %d, want 2", cfg.HydrateMaxConcurrency)
	}
	if cfg.HydrateBatchSize != 100 {
		t.Fatalf("HydrateBatchSize = %d, want 100", cfg.HydrateBatchSize)
	}
	if cfg.RecoveryRetryInterval != 5*time.Second {
		t.Fatalf("RecoveryRetryInterval = %v, want 5s", cfg.RecoveryRetryInterval)
	}
	if cfg.EventQueueSize != 10000 {
		t.Fatalf("EventQueueSize = %d, want 10000", cfg.EventQueueSize)
	}
	if cfg.AccountingBatchSize != 500 {
		t.Fatalf("AccountingBatchSize = %d, want 500", cfg.AccountingBatchSize)
	}
	if cfg.MaxSyncAccountingMembers != 2000 {
		t.Fatalf("MaxSyncAccountingMembers = %d, want 2000", cfg.MaxSyncAccountingMembers)
	}
}

func TestNewRejectsInvalidBudgets(t *testing.T) {
	tests := []struct {
		name    string
		cfg     Config
		wantErr string
	}{
		{
			name:    "negative flush interval",
			cfg:     Config{SnapshotFlushInterval: -time.Second},
			wantErr: "snapshot flush interval must be positive",
		},
		{
			name:    "negative event queue",
			cfg:     Config{EventQueueSize: -1},
			wantErr: "event queue size must be positive",
		},
		{
			name:    "negative accounting batch",
			cfg:     Config{AccountingBatchSize: -1},
			wantErr: "accounting batch size must be positive",
		},
		{
			name:    "negative max sync accounting",
			cfg:     Config{MaxSyncAccountingMembers: -1},
			wantErr: "max sync accounting members must be positive",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			_, err := New(tt.cfg)
			if err == nil || err.Error() != tt.wantErr {
				t.Fatalf("New(%+v) error = %v, want %q", tt.cfg, err, tt.wantErr)
			}
		})
	}
}

func TestApplyMessageIncrementsEligibleMembersOnly(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.WarmScopeRoster("w1", "s1", ScopeTypeChannel, []RosterMember{
		{MemberID: "author", Visibility: "public", NotificationState: "active"},
		{MemberID: "eligible", UserID: "usr_eligible", Visibility: "public", NotificationState: "quiet"},
		{MemberID: "read", Visibility: "public", NotificationState: "active", LastMessageReadSeq: 7},
		{MemberID: "muted", Visibility: "public", NotificationState: "mute"},
		{MemberID: "private", Visibility: "private", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}

	result, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:       "w1",
		ScopeID:           "s1",
		ScopeType:         ScopeTypeChannel,
		Seq:               7,
		AuthorMemberID:    "author",
		MessageGeneration: 3,
	})
	if err != nil {
		t.Fatalf("ApplyMessage returned error: %v", err)
	}
	if result.Pending || result.Ignored {
		t.Fatalf("ApplyMessage result = %+v, want accounted", result)
	}
	if result.AccountedMembers != 1 {
		t.Fatalf("AccountedMembers = %d, want 1", result.AccountedMembers)
	}
	if len(result.Deltas) != 1 {
		t.Fatalf("Deltas length = %d, want 1: %+v", len(result.Deltas), result.Deltas)
	}
	got := result.Deltas[0]
	if got.StateVersion == 0 {
		t.Fatal("delta StateVersion = 0, want a causal live projection version")
	}
	got.StateVersion = 0
	if want := (UnreadDelta{
		WorkspaceID:                      "w1",
		UserID:                           "usr_eligible",
		MemberID:                         "eligible",
		ScopeID:                          "s1",
		ScopeType:                        ScopeTypeChannel,
		Count:                            1,
		HasUnread:                        true,
		WorkspaceHasUnread:               true,
		WorkspaceUnreadConversationCount: 1,
		ServingState:                     ServingStateReady,
	}); got != want {
		t.Fatalf("delta = %+v, want %+v", got, want)
	}
	if got := runtime.ScopeUnreadCounts("w1", "eligible", nil)["s1"]; got != 1 {
		t.Fatalf("eligible unread count = %d, want 1", got)
	}
	for _, memberID := range []string{"author", "read", "muted", "private"} {
		if got := runtime.ScopeUnreadCounts("w1", memberID, nil)["s1"]; got != 0 {
			t.Fatalf("%s unread count = %d, want 0", memberID, got)
		}
	}
	if !runtime.WorkspaceHasUnread("w1", "eligible") {
		t.Fatal("WorkspaceHasUnread for eligible = false, want true")
	}
}

func TestApplyMessageUsesNotifyPrincipalForSelfExclusion(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.WarmScopeRoster("w1", "s1", ScopeTypeChannel, []RosterMember{
		{MemberID: "author", Visibility: "public", NotificationState: "active"},
		{MemberID: "notify-principal", Visibility: "public", NotificationState: "active"},
		{MemberID: "recipient", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}

	result, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:             "w1",
		ScopeID:                 "s1",
		ScopeType:               ScopeTypeChannel,
		Seq:                     1,
		AuthorMemberID:          "author",
		NotifyPrincipalMemberID: "notify-principal",
	})
	if err != nil {
		t.Fatalf("ApplyMessage returned error: %v", err)
	}
	if result.AccountedMembers != 2 {
		t.Fatalf("AccountedMembers = %d, want author and recipient", result.AccountedMembers)
	}
	if got := runtime.ScopeUnreadCounts("w1", "notify-principal", nil)["s1"]; got != 0 {
		t.Fatalf("notify principal unread count = %d, want 0", got)
	}
	if got := runtime.ScopeUnreadCounts("w1", "author", nil)["s1"]; got != 1 {
		t.Fatalf("author unread count = %d, want 1 when notify principal differs", got)
	}
}

func TestApplyMessageAccountsInConfiguredBatches(t *testing.T) {
	runtime := newTestRuntime(t, Config{AccountingBatchSize: 2})
	if err := runtime.WarmScopeRoster("w1", "s1", ScopeTypeChannel, []RosterMember{
		{MemberID: "m1", Visibility: "public", NotificationState: "active"},
		{MemberID: "m2", Visibility: "public", NotificationState: "active"},
		{MemberID: "m3", Visibility: "public", NotificationState: "active"},
		{MemberID: "m4", Visibility: "public", NotificationState: "active"},
		{MemberID: "m5", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}

	result, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:    "w1",
		ScopeID:        "s1",
		ScopeType:      ScopeTypeChannel,
		Seq:            1,
		AuthorMemberID: "m1",
	})
	if err != nil {
		t.Fatalf("ApplyMessage returned error: %v", err)
	}
	if result.Batches != 3 {
		t.Fatalf("Batches = %d, want 3", result.Batches)
	}
	if result.AccountedMembers != 4 {
		t.Fatalf("AccountedMembers = %d, want 4", result.AccountedMembers)
	}
}

func TestApplyMessageRewoundDecrementsAndClampsMessageFrontier(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.WarmScopeRoster("w1", "s1", ScopeTypeChannel, []RosterMember{
		{MemberID: "author", Visibility: "public", NotificationState: "active"},
		{MemberID: "m1", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}
	for seq := int64(1); seq <= 2; seq++ {
		if _, err := runtime.ApplyMessage(MessageStored{
			WorkspaceID:    "w1",
			ScopeID:        "s1",
			ScopeType:      ScopeTypeChannel,
			Seq:            seq,
			AuthorMemberID: "author",
		}); err != nil {
			t.Fatalf("ApplyMessage seq %d returned error: %v", seq, err)
		}
	}
	if got := runtime.ScopeUnreadCounts("w1", "m1", nil)["s1"]; got != 2 {
		t.Fatalf("unread before rewind = %d, want 2", got)
	}

	rewound := MessageRewound{
		WorkspaceID:      "w1",
		RewindGeneration: 10,
		Scopes: []RewoundScope{{
			ScopeID:   "s1",
			LatestSeq: 1,
			RemovedMessages: []RewoundMessage{{
				ScopeID:        "s1",
				Seq:            2,
				AuthorMemberID: "author",
				MessageType:    "text",
			}},
		}},
	}
	result, err := runtime.ApplyMessageRewound(rewound)
	if err != nil {
		t.Fatalf("ApplyMessageRewound returned error: %v", err)
	}
	if result.Pending {
		t.Fatalf("ApplyMessageRewound result = %+v, want not pending", result)
	}
	if got := runtime.ScopeUnreadCounts("w1", "m1", nil)["s1"]; got != 1 {
		t.Fatalf("unread after rewind = %d, want 1", got)
	}
	rewound.RewindGeneration = 99
	if _, err := runtime.ApplyMessageRewound(rewound); err != nil {
		t.Fatalf("ApplyMessageRewound replay returned error: %v", err)
	}
	if got := runtime.ScopeUnreadCounts("w1", "m1", nil)["s1"]; got != 1 {
		t.Fatalf("unread after rewind replay = %d, want 1", got)
	}

	if _, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:    "w1",
		ScopeID:        "s1",
		ScopeType:      ScopeTypeChannel,
		Seq:            2,
		AuthorMemberID: "author",
	}); err != nil {
		t.Fatalf("ApplyMessage after rewind returned error: %v", err)
	}
	if got := runtime.ScopeUnreadCounts("w1", "m1", nil)["s1"]; got != 2 {
		t.Fatalf("unread after seq reuse = %d, want 2", got)
	}
}

func TestApplyMessageRewoundIgnoresTaskScopesWithoutCorruptingExactTail(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.WarmScopeRoster("w1", "tsk_1", ScopeTypeTask, []RosterMember{
		{MemberID: "author", Visibility: "public", NotificationState: "active"},
		{MemberID: "recipient", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}
	if _, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:        "w1",
		ScopeID:            "tsk_1",
		ScopeType:          ScopeTypeTask,
		Seq:                1,
		AuthorMemberID:     "author",
		RecipientMemberIDs: []string{"recipient"},
	}); err != nil {
		t.Fatalf("ApplyMessage returned error: %v", err)
	}

	result, err := runtime.ApplyMessageRewound(MessageRewound{
		WorkspaceID:      "w1",
		RewindGeneration: 1,
		Scopes: []RewoundScope{{
			ScopeID:   "tsk_1",
			LatestSeq: 0,
			RemovedMessages: []RewoundMessage{{
				ScopeID:        "tsk_1",
				Seq:            1,
				AuthorMemberID: "author",
				MessageType:    "task_discussion",
			}},
		}},
	})
	if err != nil {
		t.Fatalf("ApplyMessageRewound returned error: %v", err)
	}
	if !result.Ignored {
		t.Fatalf("ApplyMessageRewound result = %+v, want Task scope ignored", result)
	}
	if got := runtime.ScopeUnreadCounts("w1", "recipient", nil)["tsk_1"]; got != 1 {
		t.Fatalf("Task unread after rejected rewind = %d, want 1", got)
	}

	if err := runtime.ApplyReadCursor(ReadCursorAdvanced{
		WorkspaceID: "w1",
		MemberID:    "recipient",
		ScopeID:     "tsk_1",
		LastReadSeq: 0,
	}); err != nil {
		t.Fatalf("ApplyReadCursor returned error: %v", err)
	}
	if got := runtime.ScopeUnreadCounts("w1", "recipient", nil)["tsk_1"]; got != 1 {
		t.Fatalf("Task unread after ignored rewind and cursor 0 = %d, want 1", got)
	}
}

func TestApplyMessageRewoundSkipsUnloadedMemberEntries(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.BeginScopeRosterWarm("w1", "s1", ScopeTypeChannel); err != nil {
		t.Fatalf("BeginScopeRosterWarm returned error: %v", err)
	}
	if err := runtime.WarmScopeRosterPage("w1", "s1", ScopeTypeChannel, []RosterMember{
		{MemberID: "author", Visibility: "public", NotificationState: "active"},
		{MemberID: "m1", Visibility: "public", NotificationState: "active"},
		{MemberID: "m2", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRosterPage returned error: %v", err)
	}
	if err := runtime.FinishScopeRosterWarm("w1", "s1", ScopeTypeChannel); err != nil {
		t.Fatalf("FinishScopeRosterWarm returned error: %v", err)
	}
	if err := runtime.HydrateSnapshot(Snapshot{
		WorkspaceID:               "w1",
		MemberID:                  "m1",
		ScopeID:                   "s1",
		MessageUnreadCount:        1,
		LatestMessageAccountedSeq: 2,
		ServingState:              ServingStateReady,
	}); err != nil {
		t.Fatalf("HydrateSnapshot returned error: %v", err)
	}

	if _, err := runtime.ApplyMessageRewound(MessageRewound{
		WorkspaceID:      "w1",
		RewindGeneration: 10,
		Scopes: []RewoundScope{{
			ScopeID:   "s1",
			LatestSeq: 1,
			RemovedMessages: []RewoundMessage{{
				ScopeID:        "s1",
				Seq:            2,
				AuthorMemberID: "author",
				MessageType:    "text",
			}},
		}},
	}); err != nil {
		t.Fatalf("ApplyMessageRewound returned error: %v", err)
	}
	if got := runtime.ScopeUnreadCounts("w1", "m1", nil); len(got) != 0 {
		t.Fatalf("loaded member counts after rewind = %v, want empty", got)
	}
	if got := runtime.ScopeStates("w1", "m2", nil); len(got) != 0 {
		t.Fatalf("unloaded member states after rewind = %v, want empty", got)
	}
}

func TestApplyMessageRewoundDoesNotDecrementReadMessages(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.WarmScopeRoster("w1", "s1", ScopeTypeChannel, []RosterMember{
		{MemberID: "author", Visibility: "public", NotificationState: "active"},
		{MemberID: "m1", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}
	if _, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:    "w1",
		ScopeID:        "s1",
		ScopeType:      ScopeTypeChannel,
		Seq:            2,
		AuthorMemberID: "author",
	}); err != nil {
		t.Fatalf("ApplyMessage returned error: %v", err)
	}
	if err := runtime.ApplyReadCursor(ReadCursorAdvanced{
		WorkspaceID: "w1",
		MemberID:    "m1",
		ScopeID:     "s1",
		LastReadSeq: 2,
	}); err != nil {
		t.Fatalf("ApplyReadCursor returned error: %v", err)
	}

	if _, err := runtime.ApplyMessageRewound(MessageRewound{
		WorkspaceID:      "w1",
		RewindGeneration: 11,
		Scopes: []RewoundScope{{
			ScopeID:   "s1",
			LatestSeq: 1,
			RemovedMessages: []RewoundMessage{{
				ScopeID:        "s1",
				Seq:            2,
				AuthorMemberID: "author",
				MessageType:    "text",
			}},
		}},
	}); err != nil {
		t.Fatalf("ApplyMessageRewound returned error: %v", err)
	}
	if got := runtime.ScopeUnreadCounts("w1", "m1", nil)["s1"]; got != 0 {
		t.Fatalf("read unread after rewind = %d, want 0", got)
	}
}

func TestApplyMessageRewoundColdRosterSuppressesLoadedEntry(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.HydrateSnapshot(Snapshot{
		WorkspaceID:        "w1",
		MemberID:           "m1",
		ScopeID:            "s1",
		MessageUnreadCount: 2,
		ServingState:       ServingStateReady,
	}); err != nil {
		t.Fatalf("HydrateSnapshot returned error: %v", err)
	}

	result, err := runtime.ApplyMessageRewound(MessageRewound{
		WorkspaceID:      "w1",
		RewindGeneration: 12,
		Scopes: []RewoundScope{{
			ScopeID:   "s1",
			LatestSeq: 1,
			RemovedMessages: []RewoundMessage{{
				ScopeID:        "s1",
				Seq:            2,
				AuthorMemberID: "author",
				MessageType:    "text",
			}},
		}},
	})
	if err != nil {
		t.Fatalf("ApplyMessageRewound returned error: %v", err)
	}
	if !result.Pending {
		t.Fatalf("ApplyMessageRewound result = %+v, want pending", result)
	}
	if got := runtime.ScopeUnreadCounts("w1", "m1", nil); len(got) != 0 {
		t.Fatalf("cold roster rewind counts = %v, want suppressed", got)
	}
}

func TestApplyNonMessageActivityUsesActivityCursor(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.WarmScopeRoster("w1", "s1", ScopeTypeThread, []RosterMember{
		{MemberID: "actor", Visibility: "public", NotificationState: "active"},
		{MemberID: "eligible", Visibility: "public", NotificationState: "active", LastActivityReadSeq: 2},
		{MemberID: "read", Visibility: "public", NotificationState: "active", LastActivityReadSeq: 5},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}

	result, err := runtime.ApplyNonMessageActivity(NonMessageActivityStored{
		WorkspaceID:        "w1",
		ScopeID:            "s1",
		ScopeType:          ScopeTypeThread,
		Seq:                5,
		ActorMemberID:      "actor",
		ActivityGeneration: 9,
	})
	if err != nil {
		t.Fatalf("ApplyNonMessageActivity returned error: %v", err)
	}
	if result.AccountedMembers != 1 {
		t.Fatalf("AccountedMembers = %d, want 1", result.AccountedMembers)
	}
	states := runtime.ScopeStates("w1", "eligible", nil)
	state, ok := states["s1"]
	if !ok {
		t.Fatal("eligible state missing")
	}
	if state.ActivityUnreadCount != 1 || state.Count() != 1 {
		t.Fatalf("eligible state = %+v, want one activity unread", state)
	}
}

func TestColdRosterMessageOnlyEnqueuesPendingFrontier(t *testing.T) {
	runtime := newTestRuntime(t, Config{})

	result, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:       "w1",
		ScopeID:           "s1",
		ScopeType:         ScopeTypeChannel,
		Seq:               10,
		MessageGeneration: 4,
	})
	if err != nil {
		t.Fatalf("ApplyMessage returned error: %v", err)
	}
	if !result.Pending {
		t.Fatalf("ApplyMessage result = %+v, want pending", result)
	}
	if got := runtime.ScopeUnreadCounts("w1", "member", nil)["s1"]; got != 0 {
		t.Fatalf("unread count = %d, want 0", got)
	}
	pending := runtime.PendingFrontiers("w1")
	if len(pending) != 1 {
		t.Fatalf("len(PendingFrontiers) = %d, want 1", len(pending))
	}
	if pending[0].MessageSeq != 10 || pending[0].MessageGeneration != 4 {
		t.Fatalf("pending frontier = %+v, want message seq 10 generation 4", pending[0])
	}
}

func TestColdRosterMessageSuppressesLoadedReadyEntry(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.HydrateSnapshot(Snapshot{
		WorkspaceID:        "w1",
		MemberID:           "m1",
		ScopeID:            "s1",
		MessageUnreadCount: 3,
		ServingState:       ServingStateReady,
	}); err != nil {
		t.Fatalf("HydrateSnapshot returned error: %v", err)
	}

	result, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:       "w1",
		ScopeID:           "s1",
		ScopeType:         ScopeTypeChannel,
		Seq:               11,
		MessageGeneration: 11,
	})
	if err != nil {
		t.Fatalf("ApplyMessage returned error: %v", err)
	}
	if !result.Pending {
		t.Fatalf("ApplyMessage result = %+v, want pending", result)
	}
	if got := runtime.ScopeUnreadCounts("w1", "m1", nil)["s1"]; got != 0 {
		t.Fatalf("stale unread count = %d, want suppressed 0", got)
	}
}

func TestColdRosterActivitySuppressesLoadedReadyEntry(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.HydrateSnapshot(Snapshot{
		WorkspaceID:         "w1",
		MemberID:            "m1",
		ScopeID:             "s1",
		ActivityUnreadCount: 2,
		ServingState:        ServingStateReady,
	}); err != nil {
		t.Fatalf("HydrateSnapshot returned error: %v", err)
	}

	result, err := runtime.ApplyNonMessageActivity(NonMessageActivityStored{
		WorkspaceID:        "w1",
		ScopeID:            "s1",
		ScopeType:          ScopeTypeChannel,
		Seq:                9,
		ActivityGeneration: 9,
	})
	if err != nil {
		t.Fatalf("ApplyNonMessageActivity returned error: %v", err)
	}
	if !result.Pending {
		t.Fatalf("ApplyNonMessageActivity result = %+v, want pending", result)
	}
	if got := runtime.ScopeUnreadCounts("w1", "m1", nil)["s1"]; got != 0 {
		t.Fatalf("stale activity count = %d, want suppressed 0", got)
	}
}

func TestLargeRosterMessageFallsBackToPendingFrontier(t *testing.T) {
	runtime := newTestRuntime(t, Config{MaxSyncAccountingMembers: 2})
	if err := runtime.WarmScopeRoster("w1", "s1", ScopeTypeChannel, []RosterMember{
		{MemberID: "m1", Visibility: "public", NotificationState: "active"},
		{MemberID: "m2", Visibility: "public", NotificationState: "active"},
		{MemberID: "m3", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}

	result, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:       "w1",
		ScopeID:           "s1",
		ScopeType:         ScopeTypeChannel,
		Seq:               2,
		MessageGeneration: 2,
	})
	if err != nil {
		t.Fatalf("ApplyMessage returned error: %v", err)
	}
	if !result.Pending {
		t.Fatalf("ApplyMessage result = %+v, want pending", result)
	}
	if got := runtime.ScopeUnreadCounts("w1", "m1", nil)["s1"]; got != 0 {
		t.Fatalf("m1 unread count = %d, want 0", got)
	}
	if len(runtime.PendingFrontiers("w1")) != 1 {
		t.Fatalf("PendingFrontiers length = %d, want 1", len(runtime.PendingFrontiers("w1")))
	}
	if runtime.ScopeRosterReady("w1", "s1") {
		t.Fatal("over-budget roster stayed ready and cannot trigger bounded recovery")
	}

	if err := runtime.WarmScopeRoster("w1", "s1", ScopeTypeChannel, []RosterMember{
		{MemberID: "m1", Visibility: "public", NotificationState: "active"},
		{MemberID: "m2", Visibility: "public", NotificationState: "active"},
		{MemberID: "m3", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster recovery returned error: %v", err)
	}
	result, err = runtime.ApplyPendingForScope("w1", "s1")
	if err != nil {
		t.Fatalf("ApplyPendingForScope returned error: %v", err)
	}
	if result.Pending || result.AccountedMembers != 3 {
		t.Fatalf("ApplyPendingForScope result = %+v, want recovered accounting without live sync budget", result)
	}
	if pending := runtime.PendingFrontiers("w1"); len(pending) != 0 {
		t.Fatalf("PendingFrontiers after recovered accounting = %+v, want empty", pending)
	}
}

func TestTaskScopeCountsOnlyFrozenMessageAndActivityRecipients(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.WarmScopeRoster("w1", "task-scope", ScopeTypeTask, []RosterMember{
		{MemberID: "author", Visibility: "public", NotificationState: "active"},
		{MemberID: "m1", Visibility: "public", NotificationState: "active"},
		{MemberID: "m2", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}

	result, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:        "w1",
		ScopeID:            "task-scope",
		ScopeType:          ScopeTypeTask,
		Seq:                1,
		AuthorMemberID:     "author",
		RecipientMemberIDs: []string{"m1"},
	})
	if err != nil {
		t.Fatalf("ApplyMessage returned error: %v", err)
	}
	if result.Ignored || result.AccountedMembers != 1 {
		t.Fatalf("ApplyMessage task result = %+v, want one frozen recipient", result)
	}
	if got := runtime.ScopeUnreadCounts("w1", "m1", nil)["task-scope"]; got != 1 {
		t.Fatalf("task scope generic unread count = %d, want 1", got)
	}
	// Unwatch affects future recipient compilation, not the already frozen
	// message audience. A roster refresh must preserve that unread.
	if err := runtime.WarmScopeRoster("w1", "task-scope", ScopeTypeTask, []RosterMember{
		{MemberID: "author", Visibility: "public", NotificationState: "mute"},
		{MemberID: "m1", Visibility: "public", NotificationState: "mute"},
		{MemberID: "m2", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster after Unwatch returned error: %v", err)
	}
	if got := runtime.ScopeUnreadCounts("w1", "m1", nil)["task-scope"]; got != 1 {
		t.Fatalf("task unread after Unwatch = %d, want frozen count 1", got)
	}

	activityResult, err := runtime.ApplyNonMessageActivity(NonMessageActivityStored{
		WorkspaceID:        "w1",
		ScopeID:            "task-scope",
		ScopeType:          ScopeTypeTask,
		Seq:                2,
		ActorMemberID:      "author",
		ActivityGeneration: 2,
		RecipientMemberIDs: []string{"m1"},
	})
	if err != nil {
		t.Fatalf("ApplyNonMessageActivity returned error: %v", err)
	}
	if activityResult.Ignored || activityResult.AccountedMembers != 1 {
		t.Fatalf("ApplyNonMessageActivity task result = %+v, want one frozen recipient", activityResult)
	}
	if got := runtime.ScopeUnreadCounts("w1", "m2", nil)["task-scope"]; got != 0 {
		t.Fatalf("non-recipient task unread count = %d, want 0", got)
	}
	if err := runtime.ApplyManualUnread(ManualUnreadChanged{
		WorkspaceID:  "w1",
		MemberID:     "m1",
		ScopeID:      "task-scope",
		ManualUnread: true,
	}); err != nil {
		t.Fatalf("ApplyManualUnread returned error: %v", err)
	}
	states := runtime.ScopeStates("w1", "m1", nil)
	state := states["task-scope"]
	if state.MessageUnreadCount != 1 || state.ActivityUnreadCount != 1 || state.Count() != 2 || !state.ManualUnread {
		t.Fatalf("task runtime state = %+v, want one message, one activity, and manual unread", state)
	}
	if !runtime.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread for task state = false, want true")
	}
	if got := runtime.WorkspaceUnreadConversationCount("w1", "m1"); got != 1 {
		t.Fatalf("WorkspaceUnreadConversationCount for task state = %d, want 1", got)
	}
}

func TestTaskFrozenRecipientMissingFromReadyRosterRehydratesBeforeAccounting(t *testing.T) {
	tests := []struct {
		name  string
		apply func(*Runtime) (ApplyResult, error)
	}{
		{
			name: "message",
			apply: func(runtime *Runtime) (ApplyResult, error) {
				return runtime.ApplyMessage(MessageStored{
					WorkspaceID:        "w1",
					ScopeID:            "tsk_1",
					ScopeType:          ScopeTypeTask,
					Seq:                11,
					AuthorMemberID:     "author",
					RecipientMemberIDs: []string{"new_assignee"},
				})
			},
		},
		{
			name: "activity",
			apply: func(runtime *Runtime) (ApplyResult, error) {
				return runtime.ApplyNonMessageActivity(NonMessageActivityStored{
					WorkspaceID:        "w1",
					ScopeID:            "tsk_1",
					ScopeType:          ScopeTypeTask,
					Seq:                12,
					ActorMemberID:      "author",
					RecipientMemberIDs: []string{"new_assignee"},
				})
			},
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			runtime := newTestRuntime(t, Config{})
			if err := runtime.WarmScopeRoster("w1", "tsk_1", ScopeTypeTask, []RosterMember{{
				MemberID: "author", Visibility: "public",
			}}); err != nil {
				t.Fatalf("WarmScopeRoster returned error: %v", err)
			}

			result, err := test.apply(runtime)
			if err != nil {
				t.Fatalf("live apply returned error: %v", err)
			}
			if !result.Pending || result.AccountedMembers != 0 {
				t.Fatalf("live apply result = %+v, want pending stale roster", result)
			}
			if runtime.ScopeRosterReady("w1", "tsk_1") {
				t.Fatal("stale Task roster stayed ready")
			}

			if err := runtime.WarmScopeRoster("w1", "tsk_1", ScopeTypeTask, []RosterMember{
				{MemberID: "author", Visibility: "public"},
				{MemberID: "new_assignee", UserID: "usr_assignee", Visibility: "public"},
			}); err != nil {
				t.Fatalf("WarmScopeRoster refreshed membership returned error: %v", err)
			}
			result, err = runtime.ApplyPendingForScope("w1", "tsk_1")
			if err != nil {
				t.Fatalf("ApplyPendingForScope returned error: %v", err)
			}
			if result.Pending || result.AccountedMembers != 1 {
				t.Fatalf("ApplyPendingForScope result = %+v, want one frozen recipient", result)
			}
			state := runtime.ScopeStates("w1", "new_assignee", []string{"tsk_1"})["tsk_1"]
			if state.Count() != 1 {
				t.Fatalf("new assignee Task unread state = %+v, want count 1", state)
			}
		})
	}
}

func TestTaskFrozenRecipientMissingAfterAuthoritativeRosterRefreshDoesNotAgentTeam(t *testing.T) {
	tests := []struct {
		name  string
		apply func(*Runtime) (ApplyResult, error)
	}{
		{
			name: "message",
			apply: func(runtime *Runtime) (ApplyResult, error) {
				return runtime.ApplyMessage(MessageStored{
					WorkspaceID:        "w1",
					ScopeID:            "tsk_1",
					ScopeType:          ScopeTypeTask,
					Seq:                21,
					AuthorMemberID:     "author",
					MessageGeneration:  21,
					RecipientMemberIDs: []string{"inactive_assignee"},
				})
			},
		},
		{
			name: "activity",
			apply: func(runtime *Runtime) (ApplyResult, error) {
				return runtime.ApplyNonMessageActivity(NonMessageActivityStored{
					WorkspaceID:        "w1",
					ScopeID:            "tsk_1",
					ScopeType:          ScopeTypeTask,
					Seq:                22,
					ActorMemberID:      "author",
					ActivityGeneration: 22,
					RecipientMemberIDs: []string{"inactive_assignee"},
				})
			},
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			runtime := newTestRuntime(t, Config{})
			if err := runtime.WarmScopeRoster("w1", "tsk_1", ScopeTypeTask, []RosterMember{{
				MemberID: "author", Visibility: "public",
			}}); err != nil {
				t.Fatalf("WarmScopeRoster returned error: %v", err)
			}
			result, err := test.apply(runtime)
			if err != nil {
				t.Fatalf("live apply returned error: %v", err)
			}
			if !result.Pending {
				t.Fatalf("live apply result = %+v, want pending stale roster", result)
			}

			// The authoritative reload may omit a recipient who became
			// inactive after the write-time audience was frozen.
			if err := runtime.WarmScopeRoster("w1", "tsk_1", ScopeTypeTask, []RosterMember{{
				MemberID: "author", Visibility: "public",
			}}); err != nil {
				t.Fatalf("WarmScopeRoster authoritative refresh returned error: %v", err)
			}
			result, err = runtime.ApplyPendingForScope("w1", "tsk_1")
			if err != nil {
				t.Fatalf("ApplyPendingForScope returned error: %v", err)
			}
			if result.Pending || result.AccountedMembers != 0 {
				t.Fatalf("recovered apply result = %+v, want inactive recipient dropped without another warm", result)
			}
			if !runtime.ScopeRosterReady("w1", "tsk_1") {
				t.Fatal("recovered apply coldified the authoritative roster again")
			}
			if pending := runtime.PendingFrontiers("w1"); len(pending) != 0 {
				t.Fatalf("pending frontiers after inactive recipient drop = %+v, want empty", pending)
			}
			result, err = test.apply(runtime)
			if err != nil {
				t.Fatalf("live outbox replay returned error: %v", err)
			}
			if !result.Ignored || result.Pending {
				t.Fatalf("live outbox replay result = %+v, want stable ignored completion", result)
			}
			if !runtime.ScopeRosterReady("w1", "tsk_1") {
				t.Fatal("stable live replay coldified the authoritative roster")
			}
		})
	}
}

func TestTaskOverBudgetRecoveredAudienceCompletesLiveReplay(t *testing.T) {
	tests := []struct {
		name  string
		apply func(*Runtime) (ApplyResult, error)
	}{
		{
			name: "message",
			apply: func(runtime *Runtime) (ApplyResult, error) {
				return runtime.ApplyMessage(MessageStored{
					WorkspaceID:        "w1",
					ScopeID:            "tsk_1",
					ScopeType:          ScopeTypeTask,
					Seq:                25,
					AuthorMemberID:     "author",
					MessageGeneration:  25,
					RecipientMemberIDs: []string{"m1", "m2"},
				})
			},
		},
		{
			name: "activity",
			apply: func(runtime *Runtime) (ApplyResult, error) {
				return runtime.ApplyNonMessageActivity(NonMessageActivityStored{
					WorkspaceID:        "w1",
					ScopeID:            "tsk_1",
					ScopeType:          ScopeTypeTask,
					Seq:                26,
					ActorMemberID:      "author",
					ActivityGeneration: 26,
					RecipientMemberIDs: []string{"m1", "m2"},
				})
			},
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			runtime := newTestRuntime(t, Config{MaxSyncAccountingMembers: 2})
			roster := []RosterMember{
				{MemberID: "author", UserID: "usr_author", Visibility: "public"},
				{MemberID: "m1", UserID: "usr_1", Visibility: "public"},
				{MemberID: "m2", UserID: "usr_2", Visibility: "public"},
			}
			if err := runtime.WarmScopeRoster("w1", "tsk_1", ScopeTypeTask, roster); err != nil {
				t.Fatalf("WarmScopeRoster returned error: %v", err)
			}
			result, err := test.apply(runtime)
			if err != nil {
				t.Fatalf("live over-budget apply returned error: %v", err)
			}
			if !result.Pending || result.Ignored {
				t.Fatalf("live over-budget apply result = %+v, want pending", result)
			}
			if runtime.ScopeRosterReady("w1", "tsk_1") {
				t.Fatal("over-budget Task roster stayed ready")
			}

			if err := runtime.WarmScopeRoster("w1", "tsk_1", ScopeTypeTask, roster); err != nil {
				t.Fatalf("WarmScopeRoster recovery returned error: %v", err)
			}
			result, err = runtime.ApplyPendingForScope("w1", "tsk_1")
			if err != nil {
				t.Fatalf("ApplyPendingForScope returned error: %v", err)
			}
			if result.Pending || result.Ignored || result.AccountedMembers != 2 {
				t.Fatalf("ApplyPendingForScope result = %+v, want two recovered recipients", result)
			}

			result, err = test.apply(runtime)
			if err != nil {
				t.Fatalf("live outbox replay returned error: %v", err)
			}
			if !result.Ignored || result.Pending {
				t.Fatalf("live outbox replay result = %+v, want exact audience ignored", result)
			}
			if !runtime.ScopeRosterReady("w1", "tsk_1") {
				t.Fatal("exact live replay coldified the recovered large Task roster")
			}
			for _, memberID := range []string{"m1", "m2"} {
				if got := runtime.ScopeUnreadCounts("w1", memberID, []string{"tsk_1"})["tsk_1"]; got != 1 {
					t.Fatalf("%s unread after exact live replay = %d, want 1", memberID, got)
				}
			}
		})
	}
}

func TestTaskSameSeqRecipientUnionStillWarmsNewAudience(t *testing.T) {
	tests := []struct {
		name  string
		apply func(*Runtime, []string) (ApplyResult, error)
	}{
		{
			name: "message",
			apply: func(runtime *Runtime, recipients []string) (ApplyResult, error) {
				return runtime.ApplyMessage(MessageStored{
					WorkspaceID:        "w1",
					ScopeID:            "tsk_1",
					ScopeType:          ScopeTypeTask,
					Seq:                31,
					AuthorMemberID:     "author",
					MessageGeneration:  31,
					RecipientMemberIDs: recipients,
				})
			},
		},
		{
			name: "activity",
			apply: func(runtime *Runtime, recipients []string) (ApplyResult, error) {
				return runtime.ApplyNonMessageActivity(NonMessageActivityStored{
					WorkspaceID:        "w1",
					ScopeID:            "tsk_1",
					ScopeType:          ScopeTypeTask,
					Seq:                32,
					ActorMemberID:      "author",
					ActivityGeneration: 32,
					RecipientMemberIDs: recipients,
				})
			},
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			runtime := newTestRuntime(t, Config{})
			if err := runtime.WarmScopeRoster("w1", "tsk_1", ScopeTypeTask, []RosterMember{
				{MemberID: "author", Visibility: "public"},
				{MemberID: "first", UserID: "usr_first", Visibility: "public"},
			}); err != nil {
				t.Fatalf("WarmScopeRoster returned error: %v", err)
			}
			result, err := test.apply(runtime, []string{"first"})
			if err != nil {
				t.Fatalf("initial apply returned error: %v", err)
			}
			if result.Pending || result.Ignored || result.AccountedMembers != 1 {
				t.Fatalf("initial apply result = %+v, want first recipient accounted", result)
			}

			result, err = test.apply(runtime, []string{"first", "second"})
			if err != nil {
				t.Fatalf("expanded audience apply returned error: %v", err)
			}
			if !result.Pending || result.Ignored {
				t.Fatalf("expanded audience result = %+v, want roster warm for new recipient", result)
			}
			if err := runtime.WarmScopeRoster("w1", "tsk_1", ScopeTypeTask, []RosterMember{
				{MemberID: "author", Visibility: "public"},
				{MemberID: "first", UserID: "usr_first", Visibility: "public"},
				{MemberID: "second", UserID: "usr_second", Visibility: "public"},
			}); err != nil {
				t.Fatalf("WarmScopeRoster expanded audience returned error: %v", err)
			}
			result, err = runtime.ApplyPendingForScope("w1", "tsk_1")
			if err != nil {
				t.Fatalf("ApplyPendingForScope returned error: %v", err)
			}
			if result.Pending || result.Ignored || result.AccountedMembers != 1 {
				t.Fatalf("ApplyPendingForScope result = %+v, want only new recipient accounted", result)
			}
			for _, memberID := range []string{"first", "second"} {
				if got := runtime.ScopeUnreadCounts("w1", memberID, []string{"tsk_1"})["tsk_1"]; got != 1 {
					t.Fatalf("%s unread after same-seq audience expansion = %d, want 1", memberID, got)
				}
			}
		})
	}
}

func TestTaskScopeUnreadFeedsWorkspaceAggregateAlongsideChannel(t *testing.T) {
	runtime := newTestRuntime(t, Config{})

	if err := runtime.WarmScopeRoster("w1", "tsk_1", ScopeTypeTask, []RosterMember{
		{MemberID: "author", Visibility: "public", NotificationState: "active"},
		{MemberID: "m1", UserID: "u1", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster task returned error: %v", err)
	}
	if err := runtime.WarmScopeRoster("w1", "chn_1", ScopeTypeChannel, []RosterMember{
		{MemberID: "author", Visibility: "public", NotificationState: "active"},
		{MemberID: "m1", UserID: "u1", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster channel returned error: %v", err)
	}

	taskResult, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:        "w1",
		ScopeID:            "tsk_1",
		ScopeType:          ScopeTypeTask,
		Seq:                1,
		AuthorMemberID:     "author",
		RecipientMemberIDs: []string{"m1"},
	})
	if err != nil {
		t.Fatalf("ApplyMessage task returned error: %v", err)
	}
	if taskResult.Ignored || taskResult.AccountedMembers != 1 {
		t.Fatalf("ApplyMessage task result = %+v, want one accounted member", taskResult)
	}
	if got := runtime.ScopeUnreadCounts("w1", "m1", nil)["tsk_1"]; got != 1 {
		t.Fatalf("task scope unread count = %d, want 1", got)
	}
	if !runtime.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread after task message = false, want true")
	}
	if got := runtime.WorkspaceUnreadConversationCount("w1", "m1"); got != 1 {
		t.Fatalf("WorkspaceUnreadConversationCount after task message = %d, want 1", got)
	}
	if len(taskResult.Deltas) != 1 {
		t.Fatalf("task ApplyMessage deltas = %d, want 1", len(taskResult.Deltas))
	}
	if !taskResult.Deltas[0].WorkspaceHasUnread || taskResult.Deltas[0].WorkspaceUnreadConversationCount != 1 {
		t.Fatalf("task delta workspace aggregate = hasUnread=%v count=%d, want true/1",
			taskResult.Deltas[0].WorkspaceHasUnread, taskResult.Deltas[0].WorkspaceUnreadConversationCount)
	}

	if err := runtime.ApplyManualUnread(ManualUnreadChanged{
		WorkspaceID:  "w1",
		MemberID:     "m1",
		ScopeID:      "tsk_1",
		ManualUnread: true,
	}); err != nil {
		t.Fatalf("ApplyManualUnread task returned error: %v", err)
	}
	if !runtime.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread after task manual unread = false, want true")
	}
	if got := runtime.WorkspaceUnreadConversationCount("w1", "m1"); got != 1 {
		t.Fatalf("WorkspaceUnreadConversationCount after task manual unread = %d, want 1", got)
	}

	channelResult, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:    "w1",
		ScopeID:        "chn_1",
		ScopeType:      ScopeTypeChannel,
		Seq:            1,
		AuthorMemberID: "author",
	})
	if err != nil {
		t.Fatalf("ApplyMessage channel returned error: %v", err)
	}
	if !runtime.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread after channel message = false, want true")
	}
	if got := runtime.WorkspaceUnreadConversationCount("w1", "m1"); got != 2 {
		t.Fatalf("WorkspaceUnreadConversationCount after channel message = %d, want 2", got)
	}
	if len(channelResult.Deltas) != 1 {
		t.Fatalf("channel ApplyMessage deltas = %d, want 1", len(channelResult.Deltas))
	}
	if !channelResult.Deltas[0].WorkspaceHasUnread || channelResult.Deltas[0].WorkspaceUnreadConversationCount != 2 {
		t.Fatalf("channel delta workspace aggregate = hasUnread=%v count=%d, want true/2",
			channelResult.Deltas[0].WorkspaceHasUnread, channelResult.Deltas[0].WorkspaceUnreadConversationCount)
	}

	if err := runtime.ApplyReadCursor(ReadCursorAdvanced{
		WorkspaceID: "w1",
		MemberID:    "m1",
		ScopeID:     "chn_1",
		LastReadSeq: 1,
	}); err != nil {
		t.Fatalf("ApplyReadCursor channel returned error: %v", err)
	}
	if !runtime.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread after channel read = false, want true (Task keeps aggregate alive)")
	}
	if got := runtime.WorkspaceUnreadConversationCount("w1", "m1"); got != 1 {
		t.Fatalf("WorkspaceUnreadConversationCount after channel read = %d, want 1", got)
	}

	if got := runtime.ScopeUnreadCounts("w1", "m1", nil)["tsk_1"]; got != 1 {
		t.Fatalf("task scope unread count after channel read = %d, want 1 (task per-scope state preserved)", got)
	}
}

func TestTaskScopeInvalidatedSuppressesManualUnread(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.WarmScopeRoster("w1", "task-scope", ScopeTypeTask, []RosterMember{
		{MemberID: "author", Visibility: "public", NotificationState: "active"},
		{MemberID: "m1", UserID: "u1", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}
	if _, err := runtime.ApplyNonMessageActivity(NonMessageActivityStored{
		WorkspaceID:        "w1",
		ScopeID:            "task-scope",
		ScopeType:          ScopeTypeTask,
		Seq:                2,
		ActorMemberID:      "author",
		ActivityGeneration: 2,
	}); err != nil {
		t.Fatalf("ApplyNonMessageActivity returned error: %v", err)
	}
	if err := runtime.ApplyManualUnread(ManualUnreadChanged{
		WorkspaceID:  "w1",
		MemberID:     "m1",
		ScopeID:      "task-scope",
		ManualUnread: true,
	}); err != nil {
		t.Fatalf("ApplyManualUnread returned error: %v", err)
	}
	if got := runtime.ScopeUnreadCounts("w1", "m1", nil)["task-scope"]; got != 0 {
		t.Fatalf("runtime unread count = %d, want 0", got)
	}
	if !runtime.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread before stale = false, want true")
	}
	if rows := runtime.FlushDirtySnapshots(10, time.Unix(6, 0)); len(rows) != 1 {
		t.Fatalf("initial FlushDirtySnapshots returned %d rows, want 1", len(rows))
	}

	deltas, err := runtime.ApplyScopeInvalidatedWithDeltas(ScopeInvalidated{
		WorkspaceID: "w1",
		ScopeID:     "task-scope",
		MemberIDs:   []string{"m1"},
		Reason:      "task_done",
	})
	if err != nil {
		t.Fatalf("ApplyScopeInvalidatedWithDeltas returned error: %v", err)
	}
	if got := runtime.ScopeUnreadCounts("w1", "m1", nil)["task-scope"]; got != 0 {
		t.Fatalf("stale unread count = %d, want suppressed 0", got)
	}
	if runtime.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread after stale = true, want false")
	}
	if len(deltas) != 1 || deltas[0].Count != 0 || deltas[0].ManualUnread || deltas[0].HasUnread {
		t.Fatalf("suppression deltas = %+v, want one zero unread delta", deltas)
	}
	rows := runtime.FlushDirtySnapshots(10, time.Unix(8, 0))
	if len(rows) != 0 {
		t.Fatalf("FlushDirtySnapshots after task stale returned %d rows, want 0", len(rows))
	}
}

func TestTaskHideSuppressionReopensOnlyForLaterAttention(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.WarmScopeRoster("w1", "task-scope", ScopeTypeTask, []RosterMember{
		{MemberID: "author", Visibility: "public", NotificationState: "active"},
		{MemberID: "m1", UserID: "u1", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}
	if _, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:        "w1",
		ScopeID:            "task-scope",
		ScopeType:          ScopeTypeTask,
		Seq:                1,
		AuthorMemberID:     "author",
		RecipientMemberIDs: []string{"m1"},
	}); err != nil {
		t.Fatalf("ApplyMessage returned error: %v", err)
	}
	if _, err := runtime.ApplyNonMessageActivity(NonMessageActivityStored{
		WorkspaceID:        "w1",
		ScopeID:            "task-scope",
		ScopeType:          ScopeTypeTask,
		Seq:                2,
		ActorMemberID:      "author",
		RecipientMemberIDs: []string{"m1"},
	}); err != nil {
		t.Fatalf("ApplyNonMessageActivity returned error: %v", err)
	}
	if err := runtime.ApplyManualUnread(ManualUnreadChanged{
		WorkspaceID:  "w1",
		MemberID:     "m1",
		ScopeID:      "task-scope",
		ManualUnread: true,
	}); err != nil {
		t.Fatalf("ApplyManualUnread returned error: %v", err)
	}

	deltas, err := runtime.ApplyScopeInvalidatedWithDeltas(ScopeInvalidated{
		WorkspaceID:      "w1",
		ScopeID:          "task-scope",
		MemberIDs:        []string{"m1"},
		Reason:           "conversation_hide",
		Frontier:         1,
		ActivityFrontier: 2,
		FrontiersKnown:   true,
	})
	if err != nil {
		t.Fatalf("ApplyScopeInvalidatedWithDeltas returned error: %v", err)
	}
	if len(deltas) != 1 || deltas[0].HasUnread {
		t.Fatalf("hide deltas = %+v, want one zero projection", deltas)
	}
	if runtime.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread after hide = true, want false")
	}

	activityResult, err := runtime.ApplyNonMessageActivity(NonMessageActivityStored{
		WorkspaceID:        "w1",
		ScopeID:            "task-scope",
		ScopeType:          ScopeTypeTask,
		Seq:                3,
		ActorMemberID:      "author",
		RecipientMemberIDs: []string{"m1"},
	})
	if err != nil {
		t.Fatalf("ApplyNonMessageActivity after hide frontier returned error: %v", err)
	}
	if activityResult.AccountedMembers != 0 {
		t.Fatalf("Task Activity after hide result = %+v, want suppressed recipient", activityResult)
	}
	if runtime.WorkspaceHasUnread("w1", "m1") || runtime.WorkspaceUnreadConversationCount("w1", "m1") != 0 {
		t.Fatalf("Task Activity after hide restored workspace badge: states=%+v", runtime.ScopeStates("w1", "m1", nil))
	}
	if _, ok := runtime.ScopeStates("w1", "m1", []string{"task-scope"})["task-scope"]; ok {
		t.Fatalf("Task Activity after hide restored externally served state: %+v", runtime.ScopeStates("w1", "m1", nil))
	}

	oldResult, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:        "w1",
		ScopeID:            "task-scope",
		ScopeType:          ScopeTypeTask,
		Seq:                1,
		AuthorMemberID:     "author",
		RecipientMemberIDs: []string{"m1"},
	})
	if err != nil {
		t.Fatalf("ApplyMessage at hide frontier returned error: %v", err)
	}
	if oldResult.AccountedMembers != 0 || runtime.WorkspaceHasUnread("w1", "m1") {
		t.Fatalf("event at hide frontier reopened Task: result=%+v states=%+v", oldResult, runtime.ScopeStates("w1", "m1", nil))
	}

	newResult, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:        "w1",
		ScopeID:            "task-scope",
		ScopeType:          ScopeTypeTask,
		Seq:                2,
		AuthorMemberID:     "author",
		RecipientMemberIDs: []string{"m1"},
	})
	if err != nil {
		t.Fatalf("ApplyMessage after hide frontier returned error: %v", err)
	}
	if newResult.AccountedMembers != 1 {
		t.Fatalf("new Task message result = %+v, want one reopened recipient", newResult)
	}
	state := runtime.ScopeStates("w1", "m1", nil)["task-scope"]
	if state.MessageUnreadCount != 1 || state.ActivityUnreadCount != 0 || state.ManualUnread {
		t.Fatalf("reopened Task state = %+v, want only the later message unread", state)
	}
	if !runtime.WorkspaceHasUnread("w1", "m1") || runtime.WorkspaceUnreadConversationCount("w1", "m1") != 1 {
		t.Fatalf("later Task message did not restore workspace badge: state=%+v", state)
	}
}

func TestTaskActivityReadCursorClearsTargetedTaskActivityUnread(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.WarmScopeRoster("w1", "task-scope", ScopeTypeTask, []RosterMember{
		{MemberID: "author", Visibility: "public", NotificationState: "active"},
		{MemberID: "m1", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}
	if _, err := runtime.ApplyNonMessageActivity(NonMessageActivityStored{
		WorkspaceID:        "w1",
		ScopeID:            "task-scope",
		ScopeType:          ScopeTypeTask,
		Seq:                4,
		ActorMemberID:      "author",
		ActivityGeneration: 4,
		RecipientMemberIDs: []string{"m1"},
	}); err != nil {
		t.Fatalf("ApplyNonMessageActivity returned error: %v", err)
	}
	if got := runtime.ScopeUnreadCounts("w1", "m1", nil)["task-scope"]; got != 1 {
		t.Fatalf("task activity unread count before read = %d, want 1", got)
	}
	if err := runtime.ApplyActivityReadCursor(ActivityReadCursorAdvanced{
		WorkspaceID: "w1",
		MemberID:    "m1",
		ScopeID:     "task-scope",
		LastReadSeq: 4,
	}); err != nil {
		t.Fatalf("ApplyActivityReadCursor returned error: %v", err)
	}
	if got := runtime.ScopeUnreadCounts("w1", "m1", nil)["task-scope"]; got != 0 {
		t.Fatalf("task activity unread count after read = %d, want 0", got)
	}
}

func TestTaskScopeReadCursorClearsGenericMessageUnread(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.WarmScopeRoster("w1", "task-scope", ScopeTypeTask, []RosterMember{
		{MemberID: "m1", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}
	if _, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:        "w1",
		ScopeID:            "task-scope",
		ScopeType:          ScopeTypeTask,
		Seq:                3,
		AuthorMemberID:     "author",
		RecipientMemberIDs: []string{"m1"},
	}); err != nil {
		t.Fatalf("ApplyMessage returned error: %v", err)
	}
	if got := runtime.ScopeUnreadCounts("w1", "m1", nil)["task-scope"]; got != 1 {
		t.Fatalf("task scope unread count before read = %d, want 1", got)
	}
	if err := runtime.ApplyReadCursor(ReadCursorAdvanced{
		WorkspaceID: "w1",
		MemberID:    "m1",
		ScopeID:     "task-scope",
		LastReadSeq: 3,
	}); err != nil {
		t.Fatalf("ApplyReadCursor returned error: %v", err)
	}
	if got := runtime.ScopeUnreadCounts("w1", "m1", nil)["task-scope"]; got != 0 {
		t.Fatalf("task scope unread count after read = %d, want 0", got)
	}
}

func TestScopeInvalidatedSuppressesWithoutFlushingZeroSnapshot(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.WarmScopeRoster("w1", "s1", ScopeTypeChannel, []RosterMember{
		{MemberID: "m1", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}
	if _, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID: "w1",
		ScopeID:     "s1",
		ScopeType:   ScopeTypeChannel,
		Seq:         1,
	}); err != nil {
		t.Fatalf("ApplyMessage returned error: %v", err)
	}
	if err := runtime.ApplyManualUnread(ManualUnreadChanged{
		WorkspaceID:  "w1",
		MemberID:     "m1",
		ScopeID:      "s1",
		ManualUnread: true,
	}); err != nil {
		t.Fatalf("ApplyManualUnread returned error: %v", err)
	}
	if rows := runtime.FlushDirtySnapshots(10, time.Unix(1, 0)); len(rows) != 1 {
		t.Fatalf("initial FlushDirtySnapshots returned %d rows, want 1", len(rows))
	}

	if err := runtime.ApplyScopeInvalidated(ScopeInvalidated{
		WorkspaceID: "w1",
		ScopeID:     "s1",
		MemberIDs:   []string{"m1"},
		Reason:      "notification_changed",
	}); err != nil {
		t.Fatalf("ApplyScopeInvalidated returned error: %v", err)
	}
	if got := runtime.ScopeUnreadCounts("w1", "m1", nil); len(got) != 0 {
		t.Fatalf("invalidated counts = %v, want suppressed", got)
	}
	if runtime.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread after invalidation = true, want false")
	}
	rows := runtime.FlushDirtySnapshots(10, time.Unix(2, 0))
	if len(rows) != 0 {
		t.Fatalf("FlushDirtySnapshots after invalidation returned %d rows, want 0", len(rows))
	}
}

func TestScopeInvalidatedKeepsStaleRosterPendingReplaySuppressed(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.BeginScopeRosterWarm("w1", "s1", ScopeTypeThread); err != nil {
		t.Fatalf("BeginScopeRosterWarm returned error: %v", err)
	}
	result, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:    "w1",
		ScopeID:        "s1",
		ScopeType:      ScopeTypeThread,
		Seq:            10,
		AuthorMemberID: "author",
	})
	if err != nil {
		t.Fatalf("ApplyMessage returned error: %v", err)
	}
	if !result.Pending {
		t.Fatalf("ApplyMessage result = %+v, want pending", result)
	}
	if err := runtime.HydrateSnapshot(Snapshot{
		WorkspaceID:        "w1",
		MemberID:           "m1",
		ScopeID:            "s1",
		MessageUnreadCount: 1,
		ServingState:       ServingStateSnapshotServing,
	}); err != nil {
		t.Fatalf("HydrateSnapshot returned error: %v", err)
	}
	if !runtime.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread after snapshot = false, want true")
	}
	if err := runtime.ApplyScopeInvalidated(ScopeInvalidated{
		WorkspaceID: "w1",
		ScopeID:     "s1",
		MemberIDs:   []string{"m1"},
		Reason:      "thread_unsubscribe",
	}); err != nil {
		t.Fatalf("ApplyScopeInvalidated returned error: %v", err)
	}
	if runtime.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread after invalidation = true, want false")
	}
	if err := runtime.WarmScopeRosterPage("w1", "s1", ScopeTypeThread, []RosterMember{{
		MemberID:           "m1",
		UserID:             "u1",
		LastMessageReadSeq: 0,
		EffectiveFromSeq:   0,
		NotificationState:  "active",
		Visibility:         "public",
	}}); err != nil {
		t.Fatalf("WarmScopeRosterPage returned error: %v", err)
	}
	if err := runtime.FinishScopeRosterWarm("w1", "s1", ScopeTypeThread); err != nil {
		t.Fatalf("FinishScopeRosterWarm returned error: %v", err)
	}
	result, err = runtime.ApplyPendingForScope("w1", "s1")
	if err != nil {
		t.Fatalf("ApplyPendingForScope returned error: %v", err)
	}
	if len(result.Deltas) != 0 {
		t.Fatalf("ApplyPendingForScope deltas = %+v, want none for suppressed stale roster replay", result.Deltas)
	}
	if runtime.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread after stale pending replay = true, want false")
	}
	if got := runtime.ScopeUnreadCounts("w1", "m1", []string{"s1"})["s1"]; got != 0 {
		t.Fatalf("ScopeUnreadCounts after stale pending replay = %d, want 0", got)
	}
}

func TestPermanentScopeInvalidationBlocksLateStateResurrection(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.WarmScopeRoster("w1", "sasc_1", ScopeTypeServiceAccount, []RosterMember{
		{MemberID: "author", Visibility: "public", NotificationState: "active"},
		{MemberID: "m1", UserID: "u1", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}
	if _, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID: "w1", ScopeID: "sasc_1", ScopeType: ScopeTypeServiceAccount,
		Seq: 1, AuthorMemberID: "author", MessageGeneration: 1,
	}); err != nil {
		t.Fatalf("ApplyMessage returned error: %v", err)
	}
	// Hold a row that was already handed to a snapshot writer. Requeueing that
	// stale row after deletion must not restore the dirty marker.
	inflightRows := runtime.FlushDirtySnapshots(10, time.Unix(1, 0))
	if len(inflightRows) != 1 {
		t.Fatalf("FlushDirtySnapshots returned %d rows, want 1", len(inflightRows))
	}
	if err := runtime.ApplyManualUnread(ManualUnreadChanged{
		WorkspaceID: "w1", MemberID: "m1", ScopeID: "sasc_1", ManualUnread: true,
	}); err != nil {
		t.Fatalf("ApplyManualUnread returned error: %v", err)
	}
	runtime.MarkPendingFrontier(PendingFrontier{
		WorkspaceID: "w1", ScopeID: "sasc_1", ScopeType: ScopeTypeServiceAccount,
		MessageSeq: 1, ActivitySeq: 1,
	})

	deltas, err := runtime.ApplyScopeInvalidatedWithDeltas(ScopeInvalidated{
		WorkspaceID: "w1", ScopeID: "sasc_1", Reason: "service_account_delete", Permanent: true,
	})
	if err != nil {
		t.Fatalf("ApplyScopeInvalidatedWithDeltas returned error: %v", err)
	}
	if len(deltas) != 1 {
		t.Fatalf("permanent invalidation deltas = %+v, want one member delta", deltas)
	}
	if got := deltas[0]; got.Count != 0 || got.ManualUnread || got.HasUnread || got.WorkspaceHasUnread || got.WorkspaceUnreadConversationCount != 0 {
		t.Fatalf("permanent invalidation delta = %+v, want zero state", got)
	}
	assertPermanentlyInvalidatedScopeEmpty(t, runtime, "w1", "m1", "sasc_1")

	// Simulate callbacks and hydration work that were already in flight when the
	// service-account deletion committed. None may recreate an entry, roster,
	// frontier, aggregate, or dirty snapshot.
	for name, apply := range map[string]func() error{
		"manual unread": func() error {
			return runtime.ApplyManualUnread(ManualUnreadChanged{
				WorkspaceID: "w1", MemberID: "m1", ScopeID: "sasc_1", ManualUnread: true,
			})
		},
		"message read": func() error {
			return runtime.ApplyReadCursor(ReadCursorAdvanced{
				WorkspaceID: "w1", MemberID: "m1", ScopeID: "sasc_1", LastReadSeq: 9,
			})
		},
		"activity read": func() error {
			return runtime.ApplyActivityReadCursor(ActivityReadCursorAdvanced{
				WorkspaceID: "w1", MemberID: "m1", ScopeID: "sasc_1", LastReadSeq: 9,
			})
		},
		"snapshot hydrate": func() error {
			return runtime.HydrateSnapshot(Snapshot{
				WorkspaceID: "w1", MemberID: "m1", ScopeID: "sasc_1",
				MessageUnreadCount: 8, ManualUnread: true, ServingState: ServingStateReady,
			})
		},
		"snapshot hydrate if absent": func() error {
			return runtime.HydrateSnapshotIfAbsent(Snapshot{
				WorkspaceID: "w1", MemberID: "m1", ScopeID: "sasc_1",
				ActivityUnreadCount: 8, ServingState: ServingStateSnapshotServing,
			})
		},
		"roster warm": func() error {
			return runtime.WarmScopeRoster("w1", "sasc_1", ScopeTypeServiceAccount, []RosterMember{{
				MemberID: "m1", UserID: "u1", Visibility: "public", NotificationState: "active",
			}})
		},
		"roster warm begin": func() error {
			return runtime.BeginScopeRosterWarm("w1", "sasc_1", ScopeTypeServiceAccount)
		},
		"roster warm page": func() error {
			return runtime.WarmScopeRosterPage("w1", "sasc_1", ScopeTypeServiceAccount, []RosterMember{{
				MemberID: "m1", UserID: "u1", Visibility: "public", NotificationState: "active",
			}})
		},
		"roster warm finish": func() error {
			return runtime.FinishScopeRosterWarm("w1", "sasc_1", ScopeTypeServiceAccount)
		},
	} {
		if err := apply(); err != nil {
			t.Fatalf("late %s returned error: %v", name, err)
		}
	}

	messageResult, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID: "w1", ScopeID: "sasc_1", ScopeType: ScopeTypeServiceAccount,
		Seq: 10, AuthorMemberID: "author", MessageGeneration: 10,
	})
	if err != nil {
		t.Fatalf("late ApplyMessage returned error: %v", err)
	}
	if !messageResult.Ignored || messageResult.Pending || len(messageResult.Deltas) != 0 {
		t.Fatalf("late ApplyMessage result = %+v, want ignored", messageResult)
	}
	activityResult, err := runtime.ApplyNonMessageActivity(NonMessageActivityStored{
		WorkspaceID: "w1", ScopeID: "sasc_1", ScopeType: ScopeTypeServiceAccount,
		Seq: 11, ActorMemberID: "author", ActivityGeneration: 11,
	})
	if err != nil {
		t.Fatalf("late ApplyNonMessageActivity returned error: %v", err)
	}
	if !activityResult.Ignored || activityResult.Pending || len(activityResult.Deltas) != 0 {
		t.Fatalf("late ApplyNonMessageActivity result = %+v, want ignored", activityResult)
	}

	runtime.MarkScopeServingState("w1", "sasc_1", ServingStateReady, true)
	runtime.MarkEntryServingState("w1", "m1", "sasc_1", ServingStateReady, true)
	runtime.MarkEntriesUnavailable("w1", "m1", []string{"sasc_1"})
	runtime.MarkPendingFrontier(PendingFrontier{
		WorkspaceID: "w1", ScopeID: "sasc_1", MessageSeq: 12, ActivitySeq: 12,
	})
	runtime.MarkScopeRecovered(PendingFrontier{
		WorkspaceID: "w1", ScopeID: "sasc_1", MessageSeq: 12, ActivitySeq: 12,
	})
	runtime.RequeueDirtySnapshots(inflightRows)
	assertPermanentlyInvalidatedScopeEmpty(t, runtime, "w1", "m1", "sasc_1")
}

func TestPermanentScopeInvalidationClearsColdPendingAndBlocksLateRosterWarm(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 1})
	if err := runtime.BeginScopeRosterWarm("w1", "sasc_1", ScopeTypeServiceAccount); err != nil {
		t.Fatalf("BeginScopeRosterWarm returned error: %v", err)
	}
	messageResult, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID: "w1", ScopeID: "sasc_1", ScopeType: ScopeTypeServiceAccount,
		Seq: 4, AuthorMemberID: "author", MessageGeneration: 4,
	})
	if err != nil || !messageResult.Pending {
		t.Fatalf("cold ApplyMessage result = %+v, error = %v, want pending", messageResult, err)
	}
	messageResult, err = runtime.ApplyMessage(MessageStored{
		WorkspaceID: "w1", ScopeID: "sasc_1", ScopeType: ScopeTypeServiceAccount,
		Seq: 6, AuthorMemberID: "author", MessageGeneration: 6,
	})
	if err != nil || !messageResult.Pending {
		t.Fatalf("overflowing ApplyMessage result = %+v, error = %v, want pending", messageResult, err)
	}
	activityResult, err := runtime.ApplyNonMessageActivity(NonMessageActivityStored{
		WorkspaceID: "w1", ScopeID: "sasc_1", ScopeType: ScopeTypeServiceAccount,
		Seq: 5, ActorMemberID: "author", ActivityGeneration: 5,
	})
	if err != nil || !activityResult.Pending {
		t.Fatalf("cold ApplyNonMessageActivity result = %+v, error = %v, want pending", activityResult, err)
	}
	activityResult, err = runtime.ApplyNonMessageActivity(NonMessageActivityStored{
		WorkspaceID: "w1", ScopeID: "sasc_1", ScopeType: ScopeTypeServiceAccount,
		Seq: 7, ActorMemberID: "author", ActivityGeneration: 7,
	})
	if err != nil || !activityResult.Pending {
		t.Fatalf("overflowing ApplyNonMessageActivity result = %+v, error = %v, want pending", activityResult, err)
	}
	if len(runtime.PendingFrontiers("w1")) != 1 {
		t.Fatalf("PendingFrontiers before invalidation = %+v, want one", runtime.PendingFrontiers("w1"))
	}
	sh := runtime.shardForWorkspace("w1")
	sh.mu.Lock()
	_, overflowed := sh.pendingOverflow[scopeKey{workspaceID: "w1", scopeID: "sasc_1"}]
	sh.mu.Unlock()
	if !overflowed {
		t.Fatal("cold pending queue did not overflow as expected")
	}

	if err := runtime.ApplyScopeInvalidated(ScopeInvalidated{
		WorkspaceID: "w1", ScopeID: "sasc_1", Reason: "service_account_delete", Permanent: true,
	}); err != nil {
		t.Fatalf("ApplyScopeInvalidated returned error: %v", err)
	}
	if err := runtime.WarmScopeRosterPage("w1", "sasc_1", ScopeTypeServiceAccount, []RosterMember{{
		MemberID: "m1", UserID: "u1", Visibility: "public", NotificationState: "active",
	}}); err != nil {
		t.Fatalf("late WarmScopeRosterPage returned error: %v", err)
	}
	if err := runtime.FinishScopeRosterWarm("w1", "sasc_1", ScopeTypeServiceAccount); err != nil {
		t.Fatalf("late FinishScopeRosterWarm returned error: %v", err)
	}
	result, err := runtime.ApplyPendingForScope("w1", "sasc_1")
	if err != nil {
		t.Fatalf("ApplyPendingForScope returned error: %v", err)
	}
	if !result.Ignored || result.Pending || len(result.Deltas) != 0 {
		t.Fatalf("ApplyPendingForScope result = %+v, want ignored", result)
	}
	assertPermanentlyInvalidatedScopeEmpty(t, runtime, "w1", "m1", "sasc_1")
}

func TestReversibleScopeInvalidationStillAllowsLaterMutation(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.WarmScopeRoster("w1", "s1", ScopeTypeChannel, []RosterMember{{
		MemberID: "m1", UserID: "u1", Visibility: "public", NotificationState: "active",
	}}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}
	if err := runtime.ApplyScopeInvalidated(ScopeInvalidated{
		WorkspaceID: "w1", ScopeID: "s1", MemberIDs: []string{"m1"}, Reason: "notification_changed",
	}); err != nil {
		t.Fatalf("ApplyScopeInvalidated returned error: %v", err)
	}
	if err := runtime.ApplyManualUnread(ManualUnreadChanged{
		WorkspaceID: "w1", MemberID: "m1", ScopeID: "s1", ManualUnread: true,
	}); err != nil {
		t.Fatalf("ApplyManualUnread returned error: %v", err)
	}
	state, ok := runtime.ScopeStates("w1", "m1", []string{"s1"})["s1"]
	if !ok || !state.ManualUnread || state.ServingState != ServingStateReady {
		t.Fatalf("recovered reversible state = %+v, present = %v", state, ok)
	}
	if !runtime.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread after reversible mutation = false, want true")
	}
}

func TestNotificationMuteClearsAndRefreshRosterRestoresEligibility(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.WarmScopeRoster("w1", "s1", ScopeTypeChannel, []RosterMember{
		{MemberID: "author", Visibility: "public", NotificationState: "active"},
		{MemberID: "m1", UserID: "u1", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}
	applyMessage := func(seq int64) ApplyResult {
		t.Helper()
		result, err := runtime.ApplyMessage(MessageStored{
			WorkspaceID: "w1", ScopeID: "s1", ScopeType: ScopeTypeChannel,
			Seq: seq, AuthorMemberID: "author",
		})
		if err != nil {
			t.Fatalf("ApplyMessage(seq=%d) returned error: %v", seq, err)
		}
		if result.Pending || result.Ignored {
			t.Fatalf("ApplyMessage(seq=%d) result = %+v, want applied", seq, result)
		}
		return result
	}
	count := func() int64 {
		t.Helper()
		state, ok := runtime.ScopeStates("w1", "m1", []string{"s1"})["s1"]
		if !ok {
			return 0
		}
		return state.Count()
	}

	// Pre-mute message accounts for m1.
	applyMessage(1)
	if got := count(); got != 1 {
		t.Fatalf("unread after first message = %d, want 1", got)
	}
	if !runtime.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("workspace has no unread after first message")
	}

	// Mute clears current unread, suppresses the entry, and returns a zero delta.
	deltas, err := runtime.ApplyNotificationMuted(ScopeInvalidated{
		WorkspaceID: "w1", ScopeID: "s1", MemberIDs: []string{"m1"}, Reason: "channel_notification_self",
	})
	if err != nil {
		t.Fatalf("ApplyNotificationMuted returned error: %v", err)
	}
	if len(deltas) != 1 || deltas[0].Count != 0 || deltas[0].HasUnread || deltas[0].ServingState != ServingStateStale {
		t.Fatalf("mute deltas = %+v, want one zero stale delta", deltas)
	}
	if got := count(); got != 0 {
		t.Fatalf("unread after mute = %d, want 0", got)
	}
	if runtime.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("workspace has unread after mute")
	}

	// A message posted while muted stays uncounted.
	applyMessage(2)
	if got := count(); got != 0 {
		t.Fatalf("unread after muted message = %d, want 0", got)
	}

	// Re-activation refreshes the roster baseline from the store. The refresh
	// bumps the roster generation, so the reload must capture it via Begin before
	// warming pages and publishing.
	if err := runtime.InvalidateScopeRosterForRefresh("w1", "s1", ScopeTypeChannel, []string{"m1"}); err != nil {
		t.Fatalf("InvalidateScopeRosterForRefresh returned error: %v", err)
	}
	if runtime.ScopeRosterReady("w1", "s1") {
		t.Fatal("ScopeRosterReady after refresh invalidation = true, want false")
	}
	if err := runtime.BeginScopeRosterWarm("w1", "s1", ScopeTypeChannel); err != nil {
		t.Fatalf("BeginScopeRosterWarm returned error: %v", err)
	}
	if err := runtime.WarmScopeRosterPage("w1", "s1", ScopeTypeChannel, []RosterMember{
		{MemberID: "author", Visibility: "public", NotificationState: "active"},
		{MemberID: "m1", UserID: "u1", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRosterPage returned error: %v", err)
	}
	if err := runtime.FinishScopeRosterWarm("w1", "s1", ScopeTypeChannel); err != nil {
		t.Fatalf("FinishScopeRosterWarm returned error: %v", err)
	}

	// The next eligible message regains exactly one unread point: the
	// mute-period message is not replayed and the pre-mute unread is not revived.
	applyMessage(3)
	if got := count(); got != 1 {
		t.Fatalf("unread after reactivation message = %d, want exactly 1", got)
	}
	if !runtime.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("workspace has no unread after reactivation message")
	}
	state, ok := runtime.ScopeStates("w1", "m1", []string{"s1"})["s1"]
	if !ok || state.ServingState != ServingStateReady || state.MessageUnreadCount != 1 {
		t.Fatalf("reactivated state = %+v, present = %v, want ready count 1", state, ok)
	}
}

func TestNotificationMuteThenQuietReactivationKeepsFutureEligibility(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.WarmScopeRoster("w1", "s1", ScopeTypeThread, []RosterMember{
		{MemberID: "author", Visibility: "public", NotificationState: "active"},
		{MemberID: "m1", UserID: "u1", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}
	apply := func(seq int64) {
		t.Helper()
		result, err := runtime.ApplyMessage(MessageStored{
			WorkspaceID: "w1", ScopeID: "s1", ScopeType: ScopeTypeThread,
			Seq: seq, AuthorMemberID: "author",
		})
		if err != nil {
			t.Fatalf("ApplyMessage(seq=%d) returned error: %v", seq, err)
		}
		if result.Pending || result.Ignored {
			t.Fatalf("ApplyMessage(seq=%d) result = %+v, want applied", seq, result)
		}
	}
	count := func() int64 {
		t.Helper()
		state, ok := runtime.ScopeStates("w1", "m1", []string{"s1"})["s1"]
		if !ok {
			return 0
		}
		return state.Count()
	}

	apply(1)
	if _, err := runtime.ApplyNotificationMuted(ScopeInvalidated{
		WorkspaceID: "w1", ScopeID: "s1", MemberIDs: []string{"m1"}, Reason: "thread_notification_self",
	}); err != nil {
		t.Fatalf("ApplyNotificationMuted returned error: %v", err)
	}
	apply(2)
	if got := count(); got != 0 {
		t.Fatalf("unread after muted message = %d, want 0", got)
	}

	// Quiet re-activation: quiet is still unread-eligible, and only the next
	// message may produce a point.
	if err := runtime.InvalidateScopeRosterForRefresh("w1", "s1", ScopeTypeThread, []string{"m1"}); err != nil {
		t.Fatalf("InvalidateScopeRosterForRefresh returned error: %v", err)
	}
	if err := runtime.BeginScopeRosterWarm("w1", "s1", ScopeTypeThread); err != nil {
		t.Fatalf("BeginScopeRosterWarm returned error: %v", err)
	}
	if err := runtime.WarmScopeRosterPage("w1", "s1", ScopeTypeThread, []RosterMember{
		{MemberID: "author", Visibility: "public", NotificationState: "active"},
		{MemberID: "m1", UserID: "u1", Visibility: "public", NotificationState: "quiet"},
	}); err != nil {
		t.Fatalf("WarmScopeRosterPage returned error: %v", err)
	}
	if err := runtime.FinishScopeRosterWarm("w1", "s1", ScopeTypeThread); err != nil {
		t.Fatalf("FinishScopeRosterWarm returned error: %v", err)
	}
	apply(3)
	if got := count(); got != 1 {
		t.Fatalf("unread after quiet reactivation message = %d, want exactly 1", got)
	}
}

func TestNotificationMuteBeforeEntrySkipsSubsequentMessages(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.BeginScopeRosterWarm("w1", "s1", ScopeTypeChannel); err != nil {
		t.Fatalf("BeginScopeRosterWarm returned error: %v", err)
	}
	// The page-wise warm path does not create entries; m1 has no loaded entry
	// yet when the mute lands.
	if err := runtime.WarmScopeRosterPage("w1", "s1", ScopeTypeChannel, []RosterMember{
		{MemberID: "author", Visibility: "public", NotificationState: "active"},
		{MemberID: "m1", UserID: "u1", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRosterPage returned error: %v", err)
	}
	if err := runtime.FinishScopeRosterWarm("w1", "s1", ScopeTypeChannel); err != nil {
		t.Fatalf("FinishScopeRosterWarm returned error: %v", err)
	}
	if _, err := runtime.ApplyNotificationMuted(ScopeInvalidated{
		WorkspaceID: "w1", ScopeID: "s1", MemberIDs: []string{"m1"}, Reason: "channel_notification_self",
	}); err != nil {
		t.Fatalf("ApplyNotificationMuted returned error: %v", err)
	}
	result, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID: "w1", ScopeID: "s1", ScopeType: ScopeTypeChannel,
		Seq: 1, AuthorMemberID: "author",
	})
	if err != nil {
		t.Fatalf("ApplyMessage returned error: %v", err)
	}
	if result.Pending || result.Ignored || result.AccountedMembers != 0 {
		t.Fatalf("ApplyMessage result = %+v, want muted member skipped", result)
	}
	state, ok := runtime.ScopeStates("w1", "m1", []string{"s1"})["s1"]
	if ok && state.HasUnread() {
		t.Fatalf("muted member with no prior entry gained unread: %+v", state)
	}
	if runtime.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("workspace has unread for a muted member with no prior entry")
	}
}

func assertPermanentlyInvalidatedScopeEmpty(t *testing.T, runtime *Runtime, workspaceID, memberID, scopeID string) {
	t.Helper()
	if states := runtime.ScopeStates(workspaceID, memberID, []string{scopeID}); len(states) != 0 {
		t.Fatalf("ScopeStates = %+v, want empty", states)
	}
	if counts := runtime.ScopeUnreadCounts(workspaceID, memberID, []string{scopeID}); len(counts) != 0 {
		t.Fatalf("ScopeUnreadCounts = %+v, want empty", counts)
	}
	if runtime.WorkspaceHasUnread(workspaceID, memberID) {
		t.Fatal("WorkspaceHasUnread = true, want false")
	}
	if got := runtime.WorkspaceUnreadConversationCount(workspaceID, memberID); got != 0 {
		t.Fatalf("WorkspaceUnreadConversationCount = %d, want 0", got)
	}
	if rows := runtime.FlushDirtySnapshots(100, time.Unix(99, 0)); len(rows) != 0 {
		t.Fatalf("FlushDirtySnapshots = %+v, want empty", rows)
	}
	if pending := runtime.PendingFrontiers(workspaceID); len(pending) != 0 {
		t.Fatalf("PendingFrontiers = %+v, want empty", pending)
	}
	if pending := runtime.PendingFrontiersAll(); len(pending) != 0 {
		t.Fatalf("PendingFrontiersAll = %+v, want empty", pending)
	}
	if _, ok := runtime.PendingFrontierForScope(workspaceID, scopeID); ok {
		t.Fatal("PendingFrontierForScope present, want absent")
	}
	if missing := runtime.missingScopeIDs(workspaceID, memberID, []string{scopeID}); len(missing) != 0 {
		t.Fatalf("missingScopeIDs = %+v, want empty for tombstone", missing)
	}
	if runtime.ScopeRosterReady(workspaceID, scopeID) {
		t.Fatal("ScopeRosterReady = true, want false")
	}
	if messageSeq, activitySeq, ok := runtime.ScopeAccountedFrontier(workspaceID, scopeID); ok || messageSeq != 0 || activitySeq != 0 {
		t.Fatalf("ScopeAccountedFrontier = (%d, %d, %v), want zero/absent", messageSeq, activitySeq, ok)
	}

	sh := runtime.shardForWorkspace(workspaceID)
	sh.mu.Lock()
	defer sh.mu.Unlock()
	key := scopeKey{workspaceID: workspaceID, scopeID: scopeID}
	if !sh.isScopePermanentlyInvalidatedKey(key) {
		t.Fatal("permanent scope tombstone missing")
	}
	_, hasOverflow := sh.pendingOverflow[key]
	if sh.scopes[key] != nil || sh.pending[key].WorkspaceID != "" || len(sh.pendingMessages[key]) != 0 || len(sh.pendingActivities[key]) != 0 || hasOverflow {
		t.Fatalf("scope runtime state survived permanent invalidation: scope=%+v pending=%+v messages=%+v activities=%+v",
			sh.scopes[key], sh.pending[key], sh.pendingMessages[key], sh.pendingActivities[key])
	}
	for entryKey := range sh.entries {
		if entryKey.workspaceID == workspaceID && entryKey.scopeID == scopeID {
			t.Fatalf("entry survived permanent invalidation: %+v", entryKey)
		}
	}
	for entryKey := range sh.dirty {
		if entryKey.workspaceID == workspaceID && entryKey.scopeID == scopeID {
			t.Fatalf("dirty marker survived permanent invalidation: %+v", entryKey)
		}
	}
	for entryKey := range sh.suppressedRoster {
		if entryKey.workspaceID == workspaceID && entryKey.scopeID == scopeID {
			t.Fatalf("suppressed roster marker survived permanent invalidation: %+v", entryKey)
		}
	}
}

func TestColdRosterPendingMessageReplaysAfterRosterWarm(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	result, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:    "w1",
		ScopeID:        "s1",
		ScopeType:      ScopeTypeChannel,
		Seq:            1,
		AuthorMemberID: "author",
	})
	if err != nil {
		t.Fatalf("ApplyMessage returned error: %v", err)
	}
	if !result.Pending {
		t.Fatalf("ApplyMessage result = %+v, want pending", result)
	}
	if err := runtime.WarmScopeRoster("w1", "s1", ScopeTypeChannel, []RosterMember{
		{MemberID: "author", Visibility: "public", NotificationState: "active"},
		{MemberID: "m1", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}

	result, err = runtime.ApplyPendingForScope("w1", "s1")
	if err != nil {
		t.Fatalf("ApplyPendingForScope returned error: %v", err)
	}
	if result.Pending || result.AccountedMembers != 1 {
		t.Fatalf("ApplyPendingForScope result = %+v, want one accounted member", result)
	}
	if got := runtime.ScopeUnreadCounts("w1", "m1", []string{"s1"})["s1"]; got != 1 {
		t.Fatalf("ScopeUnreadCounts after pending replay = %d, want 1", got)
	}
	if pending := runtime.PendingFrontiers("w1"); len(pending) != 0 {
		t.Fatalf("PendingFrontiers after replay = %+v, want empty", pending)
	}
}

func TestSnapshotServingIsServedButNotDirty(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.HydrateSnapshot(Snapshot{
		WorkspaceID:        "w1",
		MemberID:           "m1",
		ScopeID:            "s1",
		MessageUnreadCount: 2,
		ServingState:       ServingStateSnapshotServing,
	}); err != nil {
		t.Fatalf("HydrateSnapshot returned error: %v", err)
	}

	if got := runtime.ScopeUnreadCounts("w1", "m1", nil)["s1"]; got != 2 {
		t.Fatalf("snapshot unread count = %d, want 2", got)
	}
	if !runtime.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread snapshot = false, want true")
	}
	if rows := runtime.FlushDirtySnapshots(10, time.Unix(1, 0)); len(rows) != 0 {
		t.Fatalf("FlushDirtySnapshots after hydrate returned %d rows, want 0", len(rows))
	}
}

func TestDirtySnapshotsAreCoalesced(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.WarmScopeRoster("w1", "s1", ScopeTypeChannel, []RosterMember{
		{MemberID: "m1", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}
	for seq := int64(1); seq <= 3; seq++ {
		if _, err := runtime.ApplyMessage(MessageStored{
			WorkspaceID: "w1",
			ScopeID:     "s1",
			ScopeType:   ScopeTypeChannel,
			Seq:         seq,
		}); err != nil {
			t.Fatalf("ApplyMessage seq %d returned error: %v", seq, err)
		}
	}

	rows := runtime.FlushDirtySnapshots(10, time.Unix(9, 0))
	if len(rows) != 1 {
		t.Fatalf("FlushDirtySnapshots returned %d rows, want 1", len(rows))
	}
	row := rows[0]
	if row.MessageUnreadCount != 3 {
		t.Fatalf("snapshot MessageUnreadCount = %d, want 3", row.MessageUnreadCount)
	}
	if row.UpdatedAt != time.Unix(9, 0) {
		t.Fatalf("snapshot UpdatedAt = %v, want unix 9", row.UpdatedAt)
	}
	if rows := runtime.FlushDirtySnapshots(10, time.Unix(10, 0)); len(rows) != 0 {
		t.Fatalf("second FlushDirtySnapshots returned %d rows, want 0", len(rows))
	}
}

func TestReadCursorClearsMessageAndManualUnread(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.WarmScopeRoster("w1", "s1", ScopeTypeChannel, []RosterMember{
		{MemberID: "m1", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}
	if _, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID: "w1",
		ScopeID:     "s1",
		ScopeType:   ScopeTypeChannel,
		Seq:         1,
	}); err != nil {
		t.Fatalf("ApplyMessage returned error: %v", err)
	}
	if err := runtime.ApplyManualUnread(ManualUnreadChanged{
		WorkspaceID:  "w1",
		MemberID:     "m1",
		ScopeID:      "s1",
		ManualUnread: true,
	}); err != nil {
		t.Fatalf("ApplyManualUnread returned error: %v", err)
	}
	if !runtime.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread before read = false, want true")
	}

	if err := runtime.ApplyReadCursor(ReadCursorAdvanced{
		WorkspaceID: "w1",
		MemberID:    "m1",
		ScopeID:     "s1",
		LastReadSeq: 1,
	}); err != nil {
		t.Fatalf("ApplyReadCursor returned error: %v", err)
	}
	if got := runtime.ScopeUnreadCounts("w1", "m1", nil)["s1"]; got != 0 {
		t.Fatalf("unread count after read = %d, want 0", got)
	}
	if runtime.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread after read = true, want false")
	}
}

func TestReadCursorsRetainAlreadyAccountedTaskEventsAfterCursor(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.WarmScopeRoster("w1", "tsk_1", ScopeTypeTask, []RosterMember{
		{MemberID: "recipient", Visibility: "public"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}
	if _, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:        "w1",
		ScopeID:            "tsk_1",
		ScopeType:          ScopeTypeTask,
		Seq:                11,
		RecipientMemberIDs: []string{"recipient"},
	}); err != nil {
		t.Fatalf("ApplyMessage returned error: %v", err)
	}
	if _, err := runtime.ApplyNonMessageActivity(NonMessageActivityStored{
		WorkspaceID:        "w1",
		ScopeID:            "tsk_1",
		ScopeType:          ScopeTypeTask,
		Seq:                11,
		RecipientMemberIDs: []string{"recipient"},
	}); err != nil {
		t.Fatalf("ApplyNonMessageActivity returned error: %v", err)
	}

	if err := runtime.ApplyReadCursor(ReadCursorAdvanced{
		WorkspaceID: "w1",
		MemberID:    "recipient",
		ScopeID:     "tsk_1",
		LastReadSeq: 10,
	}); err != nil {
		t.Fatalf("ApplyReadCursor through 10 returned error: %v", err)
	}
	if err := runtime.ApplyActivityReadCursor(ActivityReadCursorAdvanced{
		WorkspaceID: "w1",
		MemberID:    "recipient",
		ScopeID:     "tsk_1",
		LastReadSeq: 10,
	}); err != nil {
		t.Fatalf("ApplyActivityReadCursor through 10 returned error: %v", err)
	}
	state := runtime.ScopeStates("w1", "recipient", []string{"tsk_1"})["tsk_1"]
	if state.MessageUnreadCount != 1 || state.ActivityUnreadCount != 1 || state.Count() != 2 {
		t.Fatalf("state after delayed cursor through 10 = %+v, want one message and one activity unread", state)
	}

	if err := runtime.ApplyReadCursor(ReadCursorAdvanced{
		WorkspaceID: "w1",
		MemberID:    "recipient",
		ScopeID:     "tsk_1",
		LastReadSeq: 11,
	}); err != nil {
		t.Fatalf("ApplyReadCursor through 11 returned error: %v", err)
	}
	if err := runtime.ApplyActivityReadCursor(ActivityReadCursorAdvanced{
		WorkspaceID: "w1",
		MemberID:    "recipient",
		ScopeID:     "tsk_1",
		LastReadSeq: 11,
	}); err != nil {
		t.Fatalf("ApplyActivityReadCursor through 11 returned error: %v", err)
	}
	state = runtime.ScopeStates("w1", "recipient", []string{"tsk_1"})["tsk_1"]
	if state.MessageUnreadCount != 0 || state.ActivityUnreadCount != 0 || state.Count() != 0 {
		t.Fatalf("state after cursor through latest = %+v, want no unread", state)
	}
}

func TestTaskUnreadAcceptsReversedCallbacksAndDeduplicatesReplay(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.WarmScopeRoster("w1", "tsk_1", ScopeTypeTask, []RosterMember{{
		MemberID:   "recipient",
		Visibility: "public",
	}}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}

	for _, seq := range []int64{12, 10, 11, 10} {
		result, err := runtime.ApplyMessage(MessageStored{
			WorkspaceID:        "w1",
			ScopeID:            "tsk_1",
			ScopeType:          ScopeTypeTask,
			Seq:                seq,
			RecipientMemberIDs: []string{"recipient"},
		})
		if err != nil {
			t.Fatalf("ApplyMessage seq %d returned error: %v", seq, err)
		}
		wantAccounted := 1
		if seq == 10 && result.AccountedMembers == 0 {
			wantAccounted = 0 // the second seq 10 is an idempotent replay
		}
		if result.AccountedMembers != wantAccounted {
			t.Fatalf("ApplyMessage seq %d result = %+v, want accounted=%d", seq, result, wantAccounted)
		}
	}
	for _, seq := range []int64{22, 20, 21, 20} {
		if _, err := runtime.ApplyNonMessageActivity(NonMessageActivityStored{
			WorkspaceID:        "w1",
			ScopeID:            "tsk_1",
			ScopeType:          ScopeTypeTask,
			Seq:                seq,
			RecipientMemberIDs: []string{"recipient"},
		}); err != nil {
			t.Fatalf("ApplyNonMessageActivity seq %d returned error: %v", seq, err)
		}
	}

	state := runtime.ScopeStates("w1", "recipient", []string{"tsk_1"})["tsk_1"]
	if state.MessageUnreadCount != 3 || state.ActivityUnreadCount != 3 {
		t.Fatalf("reversed Task callbacks produced state %+v, want three unique events in each sequence lane", state)
	}
	if err := runtime.ApplyReadCursor(ReadCursorAdvanced{
		WorkspaceID: "w1",
		MemberID:    "recipient",
		ScopeID:     "tsk_1",
		LastReadSeq: 10,
	}); err != nil {
		t.Fatalf("ApplyReadCursor returned error: %v", err)
	}
	if err := runtime.ApplyActivityReadCursor(ActivityReadCursorAdvanced{
		WorkspaceID: "w1",
		MemberID:    "recipient",
		ScopeID:     "tsk_1",
		LastReadSeq: 20,
	}); err != nil {
		t.Fatalf("ApplyActivityReadCursor returned error: %v", err)
	}
	state = runtime.ScopeStates("w1", "recipient", []string{"tsk_1"})["tsk_1"]
	if state.MessageUnreadCount != 2 || state.ActivityUnreadCount != 2 {
		t.Fatalf("cursor over sorted reversed tails produced state %+v, want two newer events in each lane", state)
	}
}

func TestTaskHideArrivingAfterLaterActivityStillSuppressesTask(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.WarmScopeRoster("w1", "tsk_1", ScopeTypeTask, []RosterMember{
		{MemberID: "actor", Visibility: "public"},
		{MemberID: "recipient", UserID: "u1", Visibility: "public"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}
	for _, seq := range []int64{5, 6} {
		if _, err := runtime.ApplyNonMessageActivity(NonMessageActivityStored{
			WorkspaceID:        "w1",
			ScopeID:            "tsk_1",
			ScopeType:          ScopeTypeTask,
			Seq:                seq,
			ActorMemberID:      "actor",
			RecipientMemberIDs: []string{"recipient"},
		}); err != nil {
			t.Fatalf("ApplyNonMessageActivity seq %d returned error: %v", seq, err)
		}
	}

	deltas, err := runtime.ApplyScopeInvalidatedWithDeltas(ScopeInvalidated{
		WorkspaceID:      "w1",
		ScopeID:          "tsk_1",
		MemberIDs:        []string{"recipient"},
		Reason:           "conversation_hide",
		Frontier:         0,
		ActivityFrontier: 5,
		FrontiersKnown:   true,
	})
	if err != nil {
		t.Fatalf("ApplyScopeInvalidatedWithDeltas returned error: %v", err)
	}
	if len(deltas) != 1 || deltas[0].HasUnread || deltas[0].Count != 0 || deltas[0].ServingState != ServingStateStale {
		t.Fatalf("delayed hide deltas = %+v, want one stale zero projection", deltas)
	}
	if _, ok := runtime.ScopeStates("w1", "recipient", []string{"tsk_1"})["tsk_1"]; ok {
		t.Fatalf("later Activity superseded delayed hide: states=%+v", runtime.ScopeStates("w1", "recipient", nil))
	}
	if runtime.WorkspaceHasUnread("w1", "recipient") || runtime.WorkspaceUnreadConversationCount("w1", "recipient") != 0 {
		t.Fatal("later Activity kept workspace badge after delayed hide")
	}

	for _, seq := range []int64{4, 7} {
		result, err := runtime.ApplyNonMessageActivity(NonMessageActivityStored{
			WorkspaceID:        "w1",
			ScopeID:            "tsk_1",
			ScopeType:          ScopeTypeTask,
			Seq:                seq,
			ActorMemberID:      "actor",
			RecipientMemberIDs: []string{"recipient"},
		})
		if err != nil {
			t.Fatalf("ApplyNonMessageActivity seq %d returned error: %v", seq, err)
		}
		if result.AccountedMembers != 0 {
			t.Fatalf("Task Activity seq %d result = %+v, want suppressed while hidden", seq, result)
		}
	}
	if runtime.WorkspaceHasUnread("w1", "recipient") {
		t.Fatal("Task Activity callback restored hidden workspace badge")
	}
	if _, ok := runtime.ScopeStates("w1", "recipient", []string{"tsk_1"})["tsk_1"]; ok {
		t.Fatalf("Task Activity callback restored hidden externally served state: %+v", runtime.ScopeStates("w1", "recipient", nil))
	}
}

func TestTaskHideArrivingAfterLaterMessageIsSuperseded(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.WarmScopeRoster("w1", "tsk_1", ScopeTypeTask, []RosterMember{
		{MemberID: "actor", Visibility: "public"},
		{MemberID: "recipient", UserID: "u1", Visibility: "public"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}
	for _, seq := range []int64{5, 6} {
		if _, err := runtime.ApplyMessage(MessageStored{
			WorkspaceID:        "w1",
			ScopeID:            "tsk_1",
			ScopeType:          ScopeTypeTask,
			Seq:                seq,
			AuthorMemberID:     "actor",
			RecipientMemberIDs: []string{"recipient"},
		}); err != nil {
			t.Fatalf("ApplyMessage seq %d returned error: %v", seq, err)
		}
	}

	deltas, err := runtime.ApplyScopeInvalidatedWithDeltas(ScopeInvalidated{
		WorkspaceID:      "w1",
		ScopeID:          "tsk_1",
		MemberIDs:        []string{"recipient"},
		Reason:           "conversation_hide",
		Frontier:         5,
		ActivityFrontier: 0,
		FrontiersKnown:   true,
	})
	if err != nil {
		t.Fatalf("ApplyScopeInvalidatedWithDeltas returned error: %v", err)
	}
	if len(deltas) != 1 || !deltas[0].HasUnread || deltas[0].Count != 1 || deltas[0].ServingState != ServingStateReady {
		t.Fatalf("superseded hide deltas = %+v, want current ready seq 6 message", deltas)
	}
	state, ok := runtime.ScopeStates("w1", "recipient", []string{"tsk_1"})["tsk_1"]
	if !ok || state.MessageUnreadCount != 1 || state.ActivityUnreadCount != 0 || state.ServingState != ServingStateReady {
		t.Fatalf("state after delayed hide = %+v, want only later message retained and ready", state)
	}
	if !runtime.WorkspaceHasUnread("w1", "recipient") || runtime.WorkspaceUnreadConversationCount("w1", "recipient") != 1 {
		t.Fatal("later Task message did not supersede delayed hide in workspace aggregate")
	}

	result, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:        "w1",
		ScopeID:            "tsk_1",
		ScopeType:          ScopeTypeTask,
		Seq:                4,
		AuthorMemberID:     "actor",
		RecipientMemberIDs: []string{"recipient"},
	})
	if err != nil {
		t.Fatalf("ApplyMessage delayed seq 4 returned error: %v", err)
	}
	if result.AccountedMembers != 0 {
		t.Fatalf("delayed pre-hide message result = %+v, want ignored by advanced hide cursor", result)
	}
}

func TestReadCursorClearsHydratedBaselineButRetainsNewerRuntimeEvent(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.HydrateSnapshot(Snapshot{
		WorkspaceID:                "w1",
		MemberID:                   "recipient",
		ScopeID:                    "tsk_1",
		ScopeType:                  ScopeTypeTask,
		MessageUnreadCount:         2,
		ActivityUnreadCount:        2,
		LastMessageReadSeq:         8,
		LastActivityReadSeq:        8,
		LatestMessageAccountedSeq:  10,
		LatestActivityAccountedSeq: 10,
		ServingState:               ServingStateReady,
	}); err != nil {
		t.Fatalf("HydrateSnapshot returned error: %v", err)
	}
	if err := runtime.WarmScopeRoster("w1", "tsk_1", ScopeTypeTask, []RosterMember{
		{MemberID: "recipient", LastMessageReadSeq: 8, LastActivityReadSeq: 8, Visibility: "public"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}
	if _, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:        "w1",
		ScopeID:            "tsk_1",
		ScopeType:          ScopeTypeTask,
		Seq:                11,
		RecipientMemberIDs: []string{"recipient"},
	}); err != nil {
		t.Fatalf("ApplyMessage seq 11 returned error: %v", err)
	}
	if _, err := runtime.ApplyNonMessageActivity(NonMessageActivityStored{
		WorkspaceID:        "w1",
		ScopeID:            "tsk_1",
		ScopeType:          ScopeTypeTask,
		Seq:                11,
		RecipientMemberIDs: []string{"recipient"},
	}); err != nil {
		t.Fatalf("ApplyNonMessageActivity seq 11 returned error: %v", err)
	}
	if err := runtime.ApplyReadCursor(ReadCursorAdvanced{
		WorkspaceID: "w1",
		MemberID:    "recipient",
		ScopeID:     "tsk_1",
		LastReadSeq: 10,
	}); err != nil {
		t.Fatalf("ApplyReadCursor returned error: %v", err)
	}
	if err := runtime.ApplyActivityReadCursor(ActivityReadCursorAdvanced{
		WorkspaceID: "w1",
		MemberID:    "recipient",
		ScopeID:     "tsk_1",
		LastReadSeq: 10,
	}); err != nil {
		t.Fatalf("ApplyActivityReadCursor returned error: %v", err)
	}
	state := runtime.ScopeStates("w1", "recipient", []string{"tsk_1"})["tsk_1"]
	if state.MessageUnreadCount != 1 || state.ActivityUnreadCount != 1 {
		t.Fatalf("state after baseline cursors = %+v, want newer seq 11 retained for message and activity", state)
	}
}

func TestTaskUnreadSeqTrackingIsBoundedAndConservativelyClearsCompactedEvents(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	requireNoError := func(action string, err error) {
		t.Helper()
		if err != nil {
			t.Fatalf("%s returned error: %v", action, err)
		}
	}
	requireNoError("WarmScopeRoster", runtime.WarmScopeRoster("w1", "tsk_1", ScopeTypeTask, []RosterMember{{
		MemberID:   "recipient",
		Visibility: "public",
	}}))

	for seq := int64(1); seq <= 600; seq++ {
		_, err := runtime.ApplyMessage(MessageStored{
			WorkspaceID:        "w1",
			ScopeID:            "tsk_1",
			ScopeType:          ScopeTypeTask,
			Seq:                seq,
			RecipientMemberIDs: []string{"recipient"},
		})
		requireNoError("ApplyMessage", err)
		_, err = runtime.ApplyNonMessageActivity(NonMessageActivityStored{
			WorkspaceID:        "w1",
			ScopeID:            "tsk_1",
			ScopeType:          ScopeTypeTask,
			Seq:                seq,
			RecipientMemberIDs: []string{"recipient"},
		})
		requireNoError("ApplyNonMessageActivity", err)
	}

	sh := runtime.shardForWorkspace("w1")
	sh.mu.Lock()
	current := sh.entries[entryKey{workspaceID: "w1", memberID: "recipient", scopeID: "tsk_1"}]
	if current == nil {
		sh.mu.Unlock()
		t.Fatal("Task unread entry was not created")
	}
	messageTailLength := len(current.messageUnreadSeqs)
	activityTailLength := len(current.activityUnreadSeqs)
	messageBaselineCount := current.messageUnreadBaselineCount
	activityBaselineCount := current.activityUnreadBaselineCount
	scope := sh.scopes[scopeKey{workspaceID: "w1", scopeID: "tsk_1"}]
	if scope == nil {
		sh.mu.Unlock()
		t.Fatal("Task scope state was not created")
	}
	messageAudienceLength := len(scope.taskMessageAudiences)
	activityAudienceLength := len(scope.taskActivityAudiences)
	sh.mu.Unlock()
	if messageTailLength > 2*maxTrackedTaskUnreadSeqs || activityTailLength > 2*maxTrackedTaskUnreadSeqs {
		t.Fatalf("tracked Task tails = message %d, activity %d; want each <= %d", messageTailLength, activityTailLength, 2*maxTrackedTaskUnreadSeqs)
	}
	if messageAudienceLength > 2*maxTrackedTaskUnreadSeqs || activityAudienceLength > 2*maxTrackedTaskUnreadSeqs {
		t.Fatalf("tracked Task audiences = message %d, activity %d; want each <= %d", messageAudienceLength, activityAudienceLength, 2*maxTrackedTaskUnreadSeqs)
	}
	if messageBaselineCount == 0 || activityBaselineCount == 0 {
		t.Fatalf("compacted Task baselines = message %d, activity %d; want positive", messageBaselineCount, activityBaselineCount)
	}

	// Cursor 100 falls inside the opaque compacted baseline. Retaining that
	// entire baseline is a conservative overcount, but it must never erase an
	// event that may be newer than the cursor.
	requireNoError("ApplyReadCursor through compacted baseline", runtime.ApplyReadCursor(ReadCursorAdvanced{
		WorkspaceID: "w1",
		MemberID:    "recipient",
		ScopeID:     "tsk_1",
		LastReadSeq: 100,
	}))
	requireNoError("ApplyActivityReadCursor through compacted baseline", runtime.ApplyActivityReadCursor(ActivityReadCursorAdvanced{
		WorkspaceID: "w1",
		MemberID:    "recipient",
		ScopeID:     "tsk_1",
		LastReadSeq: 100,
	}))
	state := runtime.ScopeStates("w1", "recipient", []string{"tsk_1"})["tsk_1"]
	if state.MessageUnreadCount != 600 || state.ActivityUnreadCount != 600 {
		t.Fatalf("state after cursor inside opaque baseline = %+v, want conservative retention", state)
	}

	// Once the cursor covers the baseline frontier, exact retained tail seqs
	// allow only events 551..600 to remain.
	requireNoError("ApplyReadCursor through 550", runtime.ApplyReadCursor(ReadCursorAdvanced{
		WorkspaceID: "w1",
		MemberID:    "recipient",
		ScopeID:     "tsk_1",
		LastReadSeq: 550,
	}))
	requireNoError("ApplyActivityReadCursor through 550", runtime.ApplyActivityReadCursor(ActivityReadCursorAdvanced{
		WorkspaceID: "w1",
		MemberID:    "recipient",
		ScopeID:     "tsk_1",
		LastReadSeq: 550,
	}))
	state = runtime.ScopeStates("w1", "recipient", []string{"tsk_1"})["tsk_1"]
	if state.MessageUnreadCount != 50 || state.ActivityUnreadCount != 50 {
		t.Fatalf("state after cursor 550 = %+v, want exact seqs 551..600", state)
	}
}

func TestMessageRewindUpdatesHydratedBaselineBeforeLaterRead(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.HydrateSnapshot(Snapshot{
		WorkspaceID:               "w1",
		MemberID:                  "recipient",
		ScopeID:                   "s1",
		ScopeType:                 ScopeTypeChannel,
		MessageUnreadCount:        3,
		LastMessageReadSeq:        8,
		EffectiveFromSeq:          1,
		LatestMessageAccountedSeq: 11,
		ServingState:              ServingStateReady,
	}); err != nil {
		t.Fatalf("HydrateSnapshot returned error: %v", err)
	}
	if err := runtime.WarmScopeRoster("w1", "s1", ScopeTypeChannel, []RosterMember{
		{MemberID: "recipient", LastMessageReadSeq: 8, EffectiveFromSeq: 1, Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}
	if _, err := runtime.ApplyMessageRewound(MessageRewound{
		WorkspaceID:      "w1",
		RewindGeneration: 2,
		Scopes: []RewoundScope{{
			ScopeID:   "s1",
			LatestSeq: 10,
			RemovedMessages: []RewoundMessage{{
				ScopeID:        "s1",
				Seq:            11,
				AuthorMemberID: "author",
			}},
		}},
	}); err != nil {
		t.Fatalf("ApplyMessageRewound returned error: %v", err)
	}
	if state := runtime.ScopeStates("w1", "recipient", []string{"s1"})["s1"]; state.MessageUnreadCount != 2 {
		t.Fatalf("MessageUnreadCount after rewind = %d, want 2", state.MessageUnreadCount)
	}
	if err := runtime.ApplyReadCursor(ReadCursorAdvanced{
		WorkspaceID: "w1",
		MemberID:    "recipient",
		ScopeID:     "s1",
		LastReadSeq: 10,
	}); err != nil {
		t.Fatalf("ApplyReadCursor returned error: %v", err)
	}
	if state := runtime.ScopeStates("w1", "recipient", []string{"s1"})["s1"]; state.MessageUnreadCount != 0 {
		t.Fatalf("MessageUnreadCount after read through rewound frontier = %d, want 0", state.MessageUnreadCount)
	}
}

func TestKnownManualUnreadGenerationRejectsStaleReadClear(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.ApplyManualUnread(ManualUnreadChanged{
		WorkspaceID:            "w1",
		MemberID:               "m1",
		ScopeID:                "s1",
		ManualUnread:           true,
		ManualUnreadGeneration: 12,
	}); err != nil {
		t.Fatalf("ApplyManualUnread returned error: %v", err)
	}
	if err := runtime.ApplyReadCursor(ReadCursorAdvanced{
		WorkspaceID:                 "w1",
		MemberID:                    "m1",
		ScopeID:                     "s1",
		LastReadSeq:                 10,
		ManualUnreadGeneration:      11,
		ManualUnreadGenerationKnown: true,
	}); err != nil {
		t.Fatalf("ApplyReadCursor stale generation returned error: %v", err)
	}
	if state := runtime.ScopeStates("w1", "m1", []string{"s1"})["s1"]; !state.ManualUnread {
		t.Fatal("stale read cursor cleared newer manual unread")
	}
	if err := runtime.ApplyReadCursor(ReadCursorAdvanced{
		WorkspaceID:                 "w1",
		MemberID:                    "m1",
		ScopeID:                     "s1",
		LastReadSeq:                 10,
		ManualUnreadGeneration:      12,
		ManualUnreadGenerationKnown: true,
	}); err != nil {
		t.Fatalf("ApplyReadCursor current generation returned error: %v", err)
	}
	if state := runtime.ScopeStates("w1", "m1", []string{"s1"})["s1"]; state.ManualUnread {
		t.Fatal("current read cursor did not clear manual unread")
	}
}

func TestServiceAccountScopeCountsUnreadLikeOrdinaryConversation(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.WarmScopeRoster("w1", "sasc_1", ScopeTypeServiceAccount, []RosterMember{
		{MemberID: "publisher", Visibility: "public", NotificationState: "active"},
		{MemberID: "recipient", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}
	if _, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:    "w1",
		ScopeID:        "sasc_1",
		ScopeType:      ScopeTypeServiceAccount,
		Seq:            1,
		AuthorMemberID: "publisher",
	}); err != nil {
		t.Fatalf("ApplyMessage returned error: %v", err)
	}
	if got := runtime.ScopeUnreadCounts("w1", "recipient", nil)["sasc_1"]; got != 1 {
		t.Fatalf("recipient unread count = %d, want 1", got)
	}
	if got := runtime.ScopeUnreadCounts("w1", "publisher", nil)["sasc_1"]; got != 0 {
		t.Fatalf("publisher unread count = %d, want 0", got)
	}
	if got := scopeTypeForSnapshotScopeID("sasc_1"); got != ScopeTypeServiceAccount {
		t.Fatalf("snapshot scope type = %q, want %q", got, ScopeTypeServiceAccount)
	}
}

func newTestRuntime(t *testing.T, cfg Config) *Runtime {
	t.Helper()
	runtime, err := New(cfg)
	if err != nil {
		t.Fatalf("New returned error: %v", err)
	}
	return runtime
}

// TestApplyMessageKeepsWorkspaceCountAfterOtherScopeRead verifies that marking
// one scope read while another scope is still unread preserves the workspace
// aggregate on the emitted delta. The live socket serializes the field
// unconditionally, so any construction site that forgets to populate it would
// send an authoritative zero and blank the workspace badge even though scope B
// still has unread messages.
func TestApplyMessageKeepsWorkspaceCountAfterOtherScopeRead(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	if err := runtime.WarmScopeRoster("w1", "scopeA", ScopeTypeChannel, []RosterMember{
		{MemberID: "author", UserID: "usr_author", Visibility: "public", NotificationState: "active"},
		{MemberID: "eligible", UserID: "usr_eligible", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster scopeA returned error: %v", err)
	}
	if err := runtime.WarmScopeRoster("w1", "scopeB", ScopeTypeChannel, []RosterMember{
		{MemberID: "author", UserID: "usr_author", Visibility: "public", NotificationState: "active"},
		{MemberID: "eligible", UserID: "usr_eligible", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster scopeB returned error: %v", err)
	}

	if _, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:    "w1",
		ScopeID:        "scopeA",
		ScopeType:      ScopeTypeChannel,
		Seq:            1,
		AuthorMemberID: "author",
	}); err != nil {
		t.Fatalf("ApplyMessage scopeA returned error: %v", err)
	}
	if _, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:    "w1",
		ScopeID:        "scopeB",
		ScopeType:      ScopeTypeChannel,
		Seq:            1,
		AuthorMemberID: "author",
	}); err != nil {
		t.Fatalf("ApplyMessage scopeB returned error: %v", err)
	}

	if got := runtime.WorkspaceUnreadConversationCount("w1", "eligible"); got != 2 {
		t.Fatalf("workspace unread conversation count after both unread = %d, want 2", got)
	}

	if err := runtime.ApplyReadCursor(ReadCursorAdvanced{
		WorkspaceID: "w1",
		MemberID:    "eligible",
		ScopeID:     "scopeA",
		LastReadSeq: 1,
	}); err != nil {
		t.Fatalf("ApplyReadCursor returned error: %v", err)
	}

	if got := runtime.WorkspaceUnreadConversationCount("w1", "eligible"); got != 1 {
		t.Fatalf("workspace unread conversation count after marking scopeA read = %d, want 1", got)
	}

	result, err := runtime.ApplyMessage(MessageStored{
		WorkspaceID:    "w1",
		ScopeID:        "scopeA",
		ScopeType:      ScopeTypeChannel,
		Seq:            2,
		AuthorMemberID: "author",
	})
	if err != nil {
		t.Fatalf("ApplyMessage scopeA seq 2 returned error: %v", err)
	}
	if len(result.Deltas) != 1 {
		t.Fatalf("Deltas length = %d, want 1: %+v", len(result.Deltas), result.Deltas)
	}
	delta := result.Deltas[0]
	if delta.WorkspaceUnreadConversationCount != 2 {
		t.Fatalf("delta WorkspaceUnreadConversationCount = %d, want 2 (scopeA and scopeB both unread)", delta.WorkspaceUnreadConversationCount)
	}
	if !delta.WorkspaceHasUnread {
		t.Fatal("delta WorkspaceHasUnread = false, want true")
	}
}