package unread

import (
	"context"
	"errors"
	"strings"
	"testing"
	"time"
)

func TestServiceScopeUnreadCountsQueuesSnapshotHydrate(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 4})
	store := &fakeServiceStore{
		snapshots: []Snapshot{{
			WorkspaceID:        "w1",
			MemberID:           "m1",
			ScopeID:            "s1",
			MessageUnreadCount: 2,
			ServingState:       ServingStateSnapshotServing,
		}},
	}
	service := newTestService(t, runtime, store)

	counts := service.ScopeUnreadCounts("w1", "m1", []string{"s1"})
	if len(counts) != 0 {
		t.Fatalf("ScopeUnreadCounts before hydrate = %v, want empty", counts)
	}
	job := <-service.queue
	if job.kind != hydrateJobSnapshot {
		t.Fatalf("queued job kind = %s, want snapshot", job.kind)
	}

	service.runHydrateJob(context.Background(), job)
	service.finish(job)

	if got := service.ScopeUnreadCounts("w1", "m1", []string{"s1"})["s1"]; got != 2 {
		t.Fatalf("ScopeUnreadCounts after hydrate = %d, want 2", got)
	}
	if store.snapshotLoads != 1 {
		t.Fatalf("snapshotLoads = %d, want 1", store.snapshotLoads)
	}

	select {
	case job := <-service.queue:
		t.Fatalf("ScopeUnreadCounts queued hydrate for loaded scope: %+v", job)
	default:
	}
	if got := service.ScopeUnreadCounts("w1", "m1", []string{"s1"})["s1"]; got != 2 {
		t.Fatalf("ScopeUnreadCounts repeated read = %d, want 2", got)
	}
	if store.snapshotLoads != 1 {
		t.Fatalf("snapshotLoads after repeated read = %d, want 1", store.snapshotLoads)
	}
}

func TestServiceSnapshotHydrateSkipsEntryCreatedAfterQueue(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 4})
	store := &fakeServiceStore{
		snapshots: []Snapshot{{
			WorkspaceID:            "w1",
			MemberID:               "m1",
			ScopeID:                "s1",
			MessageUnreadCount:     7,
			ManualUnread:           false,
			ManualUnreadGeneration: 1,
			ServingState:           ServingStateSnapshotServing,
		}},
	}
	service := newTestService(t, runtime, store)

	if counts := service.ScopeUnreadCounts("w1", "m1", []string{"s1"}); len(counts) != 0 {
		t.Fatalf("ScopeUnreadCounts before hydrate = %v, want empty", counts)
	}
	job := <-service.queue

	if err := service.ApplyManualUnread(ManualUnreadChanged{
		WorkspaceID:            "w1",
		MemberID:               "m1",
		ScopeID:                "s1",
		ManualUnread:           true,
		ManualUnreadGeneration: 2,
	}); err != nil {
		t.Fatalf("ApplyManualUnread returned error: %v", err)
	}
	if err := service.runHydrateJob(context.Background(), job); err != nil {
		t.Fatalf("runHydrateJob returned error: %v", err)
	}
	service.finish(job)

	state, ok := runtime.ScopeStates("w1", "m1", []string{"s1"})["s1"]
	if !ok {
		t.Fatal("state created after queued hydrate was suppressed")
	}
	if state.ServingState != ServingStateReady || !state.ManualUnread || state.MessageUnreadCount != 0 {
		t.Fatalf("state after stale hydrate = %+v, want newer manual unread ready state", state)
	}
	if counts := service.ScopeUnreadCounts("w1", "m1", []string{"s1"}); len(counts) != 0 {
		t.Fatalf("ScopeUnreadCounts after stale hydrate = %v, want no stale snapshot count", counts)
	}
}

func TestServiceSnapshotMissDoesNotRepeatedlyHydrate(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 4})
	store := &fakeServiceStore{}
	service := newTestService(t, runtime, store)

	counts := service.ScopeUnreadCounts("w1", "m1", []string{"missing"})
	if len(counts) != 0 {
		t.Fatalf("ScopeUnreadCounts before hydrate = %v, want empty", counts)
	}
	job := <-service.queue
	service.runHydrateJob(context.Background(), job)
	service.finish(job)
	if store.snapshotLoads != 1 {
		t.Fatalf("snapshotLoads = %d, want 1", store.snapshotLoads)
	}

	counts = service.ScopeUnreadCounts("w1", "m1", []string{"missing"})
	if len(counts) != 0 {
		t.Fatalf("ScopeUnreadCounts after missing hydrate = %v, want empty", counts)
	}
	select {
	case job := <-service.queue:
		t.Fatalf("missing snapshot queued repeated hydrate: %+v", job)
	default:
	}
	if store.snapshotLoads != 1 {
		t.Fatalf("snapshotLoads after repeated read = %d, want 1", store.snapshotLoads)
	}
}

func TestServiceSnapshotMissSkipsEntryCreatedAfterQueue(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 4})
	store := &fakeServiceStore{}
	service := newTestService(t, runtime, store)

	if counts := service.ScopeUnreadCounts("w1", "m1", []string{"missing"}); len(counts) != 0 {
		t.Fatalf("ScopeUnreadCounts before hydrate = %v, want empty", counts)
	}
	job := <-service.queue

	if err := service.ApplyManualUnread(ManualUnreadChanged{
		WorkspaceID:            "w1",
		MemberID:               "m1",
		ScopeID:                "missing",
		ManualUnread:           true,
		ManualUnreadGeneration: 2,
	}); err != nil {
		t.Fatalf("ApplyManualUnread returned error: %v", err)
	}
	if err := service.runHydrateJob(context.Background(), job); err != nil {
		t.Fatalf("runHydrateJob returned error: %v", err)
	}
	service.finish(job)

	state, ok := runtime.ScopeStates("w1", "m1", []string{"missing"})["missing"]
	if !ok {
		t.Fatal("state created after queued snapshot miss was suppressed")
	}
	if state.ServingState != ServingStateReady || !state.ManualUnread {
		t.Fatalf("state after stale snapshot miss = %+v, want newer manual unread ready state", state)
	}
}

func TestServiceSnapshotHydrateClearsSnapshotUnreadCoveredByDurableCursor(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 4})
	store := &fakeServiceStore{
		snapshots: []Snapshot{{
			WorkspaceID:               "w1",
			MemberID:                  "m1",
			ScopeID:                   "s1",
			MessageUnreadCount:        1,
			LastMessageReadSeq:        0,
			LatestMessageAccountedSeq: 1,
			ServingState:              ServingStateSnapshotServing,
		}},
		metadata: []FrontierMetadata{{
			WorkspaceID:        "w1",
			MemberID:           "m1",
			ScopeID:            "s1",
			ScopeType:          ScopeTypeChannel,
			LatestMessageSeq:   1,
			LastMessageReadSeq: 1,
		}},
	}
	service := newTestService(t, runtime, store)

	_ = service.ScopeUnreadCounts("w1", "m1", []string{"s1"})
	job := <-service.queue
	service.runHydrateJob(context.Background(), job)
	service.finish(job)

	if counts := service.ScopeUnreadCounts("w1", "m1", []string{"s1"}); len(counts) != 0 {
		t.Fatalf("snapshot count after durable cursor validation = %v, want cleared unread", counts)
	}
	state, ok := runtime.ScopeStates("w1", "m1", []string{"s1"})["s1"]
	if !ok {
		t.Fatal("validated snapshot state missing")
	}
	if state.MessageUnreadCount != 0 || state.ManualUnread || state.ServingState != ServingStateReady {
		t.Fatalf("validated snapshot state = %+v, want ready zero-unread state", state)
	}
	if store.metadataLoads == 0 {
		t.Fatal("metadataLoads = 0, want durable cursor validation read")
	}
	if store.rosterLoads != 0 || store.messageTailLoads != 0 || store.activityTailLoads != 0 {
		t.Fatalf("unexpected tail/roster recovery reads: roster=%d messageTail=%d activityTail=%d", store.rosterLoads, store.messageTailLoads, store.activityTailLoads)
	}
	select {
	case job := <-service.queue:
		t.Fatalf("validated stale snapshot queued automatic recovery job: %+v", job)
	default:
	}
}

func TestServiceSnapshotHydrateKeepsSnapshotUnreadWhenDurableLatestAfterCursor(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 4})
	store := &fakeServiceStore{
		snapshots: []Snapshot{{
			WorkspaceID:               "w1",
			MemberID:                  "m1",
			ScopeID:                   "s1",
			MessageUnreadCount:        1,
			LastMessageReadSeq:        0,
			LatestMessageAccountedSeq: 1,
			ServingState:              ServingStateSnapshotServing,
		}},
		metadata: []FrontierMetadata{{
			WorkspaceID:        "w1",
			MemberID:           "m1",
			ScopeID:            "s1",
			ScopeType:          ScopeTypeChannel,
			LatestMessageSeq:   2,
			LastMessageReadSeq: 1,
		}},
	}
	service := newTestService(t, runtime, store)

	_ = service.ScopeUnreadCounts("w1", "m1", []string{"s1"})
	job := <-service.queue
	service.runHydrateJob(context.Background(), job)
	service.finish(job)

	if got := service.ScopeUnreadCounts("w1", "m1", []string{"s1"})["s1"]; got != 1 {
		t.Fatalf("snapshot count after incomplete durable frontier validation = %d, want preserved unread presence", got)
	}
	state, ok := runtime.ScopeStates("w1", "m1", []string{"s1"})["s1"]
	if !ok {
		t.Fatal("validated snapshot state missing")
	}
	if state.MessageUnreadCount != 1 || state.ServingState != ServingStateSnapshotServing {
		t.Fatalf("validated snapshot state = %+v, want snapshot-serving unread presence", state)
	}
	if !service.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread = false, want preserved unread presence while durable latest is after cursor")
	}
	if store.metadataLoads == 0 {
		t.Fatal("metadataLoads = 0, want durable frontier validation read")
	}
}

func TestServiceSnapshotHydrateKeepsDurableManualUnreadWhenCursorClearsCount(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 4})
	store := &fakeServiceStore{
		snapshots: []Snapshot{{
			WorkspaceID:               "w1",
			MemberID:                  "m1",
			ScopeID:                   "s1",
			MessageUnreadCount:        1,
			ManualUnread:              false,
			LastMessageReadSeq:        0,
			LatestMessageAccountedSeq: 1,
			ServingState:              ServingStateSnapshotServing,
		}},
		metadata: []FrontierMetadata{{
			WorkspaceID:            "w1",
			MemberID:               "m1",
			ScopeID:                "s1",
			ScopeType:              ScopeTypeChannel,
			LatestMessageSeq:       1,
			LastMessageReadSeq:     1,
			ManualUnread:           true,
			ManualUnreadGeneration: 3,
		}},
	}
	service := newTestService(t, runtime, store)

	_ = service.ScopeUnreadCounts("w1", "m1", []string{"s1"})
	job := <-service.queue
	service.runHydrateJob(context.Background(), job)
	service.finish(job)

	if counts := service.ScopeUnreadCounts("w1", "m1", []string{"s1"}); len(counts) != 0 {
		t.Fatalf("snapshot count after durable cursor validation = %v, want no numeric unread", counts)
	}
	state, ok := runtime.ScopeStates("w1", "m1", []string{"s1"})["s1"]
	if !ok {
		t.Fatal("validated manual-unread state missing")
	}
	if state.MessageUnreadCount != 0 || !state.ManualUnread || !state.HasUnread() || state.ServingState != ServingStateReady {
		t.Fatalf("validated manual-unread state = %+v, want ready manual-unread state without numeric count", state)
	}
	if !service.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread = false, want durable manual unread to preserve workspace presence")
	}
}

func TestServiceSnapshotHydrateTrustsFreshSnapshotAfterFrontierValidation(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 4})
	store := &fakeServiceStore{
		snapshots: []Snapshot{{
			WorkspaceID:               "w1",
			MemberID:                  "m1",
			ScopeID:                   "s1",
			MessageUnreadCount:        1,
			LatestMessageAccountedSeq: 1,
			ServingState:              ServingStateSnapshotServing,
		}},
		metadata: []FrontierMetadata{{
			WorkspaceID:      "w1",
			MemberID:         "m1",
			ScopeID:          "s1",
			ScopeType:        ScopeTypeChannel,
			LatestMessageSeq: 1,
		}},
	}
	service := newTestService(t, runtime, store)
	deltas := &fakeDeltaSink{}
	service.SetDeltaSink(deltas)

	_ = service.ScopeUnreadCounts("w1", "m1", []string{"s1"})
	job := <-service.queue
	service.runHydrateJob(context.Background(), job)
	service.finish(job)

	if got := service.ScopeUnreadCounts("w1", "m1", []string{"s1"})["s1"]; got != 1 {
		t.Fatalf("snapshot count after hydrate = %d, want trusted snapshot count 1", got)
	}
	state, ok := runtime.ScopeStates("w1", "m1", []string{"s1"})["s1"]
	if !ok {
		t.Fatal("hydrated snapshot state missing")
	}
	if state.ServingState != ServingStateSnapshotServing {
		t.Fatalf("ServingState = %s, want %s", state.ServingState, ServingStateSnapshotServing)
	}
	if store.metadataLoads == 0 {
		t.Fatal("metadataLoads = 0, want durable frontier validation read")
	}
	if store.rosterLoads != 0 || store.messageTailLoads != 0 || store.activityTailLoads != 0 {
		t.Fatalf("unexpected tail/roster recovery reads: roster=%d messageTail=%d activityTail=%d", store.rosterLoads, store.messageTailLoads, store.activityTailLoads)
	}
	if len(deltas.items) != 0 {
		t.Fatalf("recovery deltas = %+v, want none", deltas.items)
	}
	select {
	case job := <-service.queue:
		t.Fatalf("snapshot hydrate queued automatic recovery job: %+v", job)
	default:
	}
}

func TestServiceSnapshotHydratePaginatesExplicitScopeIDs(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 8, HydrateBatchSize: 2})
	store := &fakeServiceStore{
		snapshots: []Snapshot{
			{WorkspaceID: "w1", MemberID: "m1", ScopeID: "s1", MessageUnreadCount: 1, ServingState: ServingStateSnapshotServing},
			{WorkspaceID: "w1", MemberID: "m1", ScopeID: "s2", MessageUnreadCount: 2, ServingState: ServingStateSnapshotServing},
			{WorkspaceID: "w1", MemberID: "m1", ScopeID: "s3", MessageUnreadCount: 3, ServingState: ServingStateSnapshotServing},
		},
	}
	service := newTestService(t, runtime, store)

	_ = service.ScopeUnreadCounts("w1", "m1", []string{"s1", "s2", "s3"})
	job := <-service.queue
	service.runHydrateJob(context.Background(), job)
	service.finish(job)

	counts := runtime.ScopeUnreadCounts("w1", "m1", []string{"s1", "s2", "s3"})
	if counts["s1"] != 1 || counts["s2"] != 2 || counts["s3"] != 3 {
		t.Fatalf("paginated explicit counts = %v, want s1/s2/s3", counts)
	}
	if store.snapshotLoads != 2 {
		t.Fatalf("snapshotLoads = %d, want 2", store.snapshotLoads)
	}
}

func TestServiceUnscopedScopeUnreadCountsDoesNotHydrateSnapshots(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 8, HydrateBatchSize: 2})
	store := &fakeServiceStore{
		snapshots: []Snapshot{
			{WorkspaceID: "w1", MemberID: "m1", ScopeID: "s1", MessageUnreadCount: 1, ServingState: ServingStateSnapshotServing},
			{WorkspaceID: "w1", MemberID: "m1", ScopeID: "s2", MessageUnreadCount: 2, ServingState: ServingStateSnapshotServing},
			{WorkspaceID: "w1", MemberID: "m1", ScopeID: "s3", MessageUnreadCount: 3, ServingState: ServingStateSnapshotServing},
		},
	}
	service := newTestService(t, runtime, store)

	_ = service.ScopeUnreadCounts("w1", "m1", nil)
	select {
	case job := <-service.queue:
		t.Fatalf("unscoped ScopeUnreadCounts queued hydrate job: %+v", job)
	default:
	}
	if store.snapshotLoads != 0 {
		t.Fatalf("snapshotLoads = %d, want 0", store.snapshotLoads)
	}
}

func TestServiceWorkspaceHasUnreadHydratesMemberSnapshots(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 8, HydrateBatchSize: 2})
	store := &fakeServiceStore{
		snapshots: []Snapshot{
			{WorkspaceID: "w1", MemberID: "m1", ScopeID: "chn_1", MessageUnreadCount: 1, ServingState: ServingStateSnapshotServing},
		},
	}
	service := newTestService(t, runtime, store)

	if !service.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread after cold snapshot hydrate = false, want true")
	}
	if got := runtime.ScopeUnreadCounts("w1", "m1", []string{"chn_1"})["chn_1"]; got != 1 {
		t.Fatalf("hydrated channel scope count = %d, want 1", got)
	}
	if store.snapshotLoads != 1 {
		t.Fatalf("snapshotLoads = %d, want 1", store.snapshotLoads)
	}

	if !service.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread repeated read = false, want true")
	}
	if store.snapshotLoads != 1 {
		t.Fatalf("snapshotLoads after repeated presence read = %d, want 1", store.snapshotLoads)
	}
}

func TestServiceWorkspaceUnreadConversationCountHydratesBeforeReturning(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 8, HydrateBatchSize: 4})
	store := &fakeServiceStore{
		snapshots: []Snapshot{
			{WorkspaceID: "w1", MemberID: "m1", ScopeID: "scopeA", MessageUnreadCount: 0, ManualUnread: false, ServingState: ServingStateSnapshotServing},
			{WorkspaceID: "w1", MemberID: "m1", ScopeID: "scopeB", MessageUnreadCount: 2, ServingState: ServingStateSnapshotServing},
			{WorkspaceID: "w1", MemberID: "m1", ScopeID: "scopeC", ManualUnread: true, ServingState: ServingStateSnapshotServing},
		},
	}
	service := newTestService(t, runtime, store)

	if err := runtime.HydrateSnapshotIfAbsent(Snapshot{
		WorkspaceID:        "w1",
		MemberID:           "m1",
		ScopeID:            "scopeA",
		MessageUnreadCount: 1,
		ServingState:       ServingStateReady,
	}); err != nil {
		t.Fatalf("HydrateSnapshotIfAbsent scopeA returned error: %v", err)
	}
	if got := runtime.WorkspaceUnreadConversationCount("w1", "m1"); got != 1 {
		t.Fatalf("runtime partial aggregate before service hydrate = %d, want 1", got)
	}

	if got := service.WorkspaceUnreadConversationCount("w1", "m1"); got != 3 {
		t.Fatalf("service WorkspaceUnreadConversationCount = %d, want 3 after hydrating scopeB/scopeC", got)
	}
	if store.snapshotLoads != 1 {
		t.Fatalf("snapshotLoads after first authoritative count = %d, want 1", store.snapshotLoads)
	}

	if got := service.WorkspaceUnreadConversationCount("w1", "m1"); got != 3 {
		t.Fatalf("service WorkspaceUnreadConversationCount repeat = %d, want 3", got)
	}
	if store.snapshotLoads != 1 {
		t.Fatalf("snapshotLoads after repeat authoritative count = %d, want 1 (member already hydrated)", store.snapshotLoads)
	}
}

func TestServiceWorkspaceUnreadConversationCountHydratesOnlyVisibleInboxScopes(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 8, HydrateBatchSize: 4})
	store := &fakeServiceStore{
		snapshots: []Snapshot{
			{WorkspaceID: "w1", MemberID: "m1", ScopeID: "hidden", MessageUnreadCount: 2, ServingState: ServingStateSnapshotServing},
			{WorkspaceID: "w1", MemberID: "m1", ScopeID: "visible", ManualUnread: true, ServingState: ServingStateSnapshotServing},
		},
		visibleScopeIDs: []string{"visible"},
	}
	service := newTestService(t, runtime, store)

	if got := service.WorkspaceUnreadConversationCount("w1", "m1"); got != 1 {
		t.Fatalf("WorkspaceUnreadConversationCount = %d, want 1 visible Inbox scope", got)
	}
	states := service.WorkspaceStates("w1", "m1")
	if _, ok := states["hidden"]; ok {
		t.Fatalf("hidden scope was hydrated into workspace unread state: %+v", states["hidden"])
	}
	if _, ok := states["visible"]; !ok {
		t.Fatalf("visible scope missing from workspace unread state: %+v", states)
	}
	if store.visibleScopeLoads != 1 {
		t.Fatalf("visibleScopeLoads = %d, want 1", store.visibleScopeLoads)
	}
}

func TestServiceColdRestartKeepsActivityOnlyTaskHidden(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 8, HydrateBatchSize: 4})
	store := &fakeServiceStore{
		snapshots: []Snapshot{{
			WorkspaceID:         "w1",
			MemberID:            "m1",
			ScopeID:             "tsk_hidden",
			ScopeType:           ScopeTypeTask,
			ActivityUnreadCount: 1,
			ServingState:        ServingStateSnapshotServing,
		}},
		// A hidden Task is absent from the visible-scope page even when a stale
		// durable snapshot still records Activity unread after the hide.
		visibleScopeIDs: []string{},
	}
	service := newTestService(t, runtime, store)

	if got := service.WorkspaceUnreadConversationCount("w1", "m1"); got != 0 {
		t.Fatalf("WorkspaceUnreadConversationCount after cold restart = %d, want 0 for hidden Task", got)
	}
	if service.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread after cold restart = true, want false for hidden Task")
	}
	if states := service.WorkspaceStates("w1", "m1"); len(states) != 0 {
		t.Fatalf("WorkspaceStates after cold restart = %+v, want hidden Task omitted", states)
	}
	if store.visibleScopeLoads != 1 {
		t.Fatalf("visibleScopeLoads = %d, want 1", store.visibleScopeLoads)
	}
	if store.snapshotLoads != 0 {
		t.Fatalf("snapshotLoads = %d, want 0 because hidden Task is not a visible hydration candidate", store.snapshotLoads)
	}
}

func TestServiceVisibleInboxScopeHydratePaginatesWithoutLoadingHiddenSnapshots(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 8, HydrateBatchSize: 2})
	store := &fakeServiceStore{
		snapshots: []Snapshot{
			{WorkspaceID: "w1", MemberID: "m1", ScopeID: "hidden", MessageUnreadCount: 9, ServingState: ServingStateSnapshotServing},
			{WorkspaceID: "w1", MemberID: "m1", ScopeID: "s1", MessageUnreadCount: 1, ServingState: ServingStateSnapshotServing},
			{WorkspaceID: "w1", MemberID: "m1", ScopeID: "s2", MessageUnreadCount: 1, ServingState: ServingStateSnapshotServing},
			{WorkspaceID: "w1", MemberID: "m1", ScopeID: "s3", ManualUnread: true, ServingState: ServingStateSnapshotServing},
		},
		visibleScopeIDs: []string{"s1", "s2", "s3"},
	}
	service := newTestService(t, runtime, store)

	if got := service.WorkspaceUnreadConversationCount("w1", "m1"); got != 3 {
		t.Fatalf("WorkspaceUnreadConversationCount = %d, want 3 visible Inbox scopes", got)
	}
	if store.visibleScopeLoads != 2 {
		t.Fatalf("visibleScopeLoads = %d, want 2 pages", store.visibleScopeLoads)
	}
	if store.snapshotLoads != 2 {
		t.Fatalf("snapshotLoads = %d, want 2 visible-scope batches", store.snapshotLoads)
	}
	if _, ok := service.WorkspaceStates("w1", "m1")["hidden"]; ok {
		t.Fatal("hidden scope was hydrated while paging visible Inbox scopes")
	}
}

func TestServiceUnscopedScopeStatesHydratesMemberSnapshotsForReplay(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 8, HydrateBatchSize: 2})
	store := &fakeServiceStore{
		snapshots: []Snapshot{
			{WorkspaceID: "w1", MemberID: "m1", ScopeID: "tsk_1", ManualUnread: true, ServingState: ServingStateSnapshotServing},
		},
	}
	service := newTestService(t, runtime, store)

	states := service.ScopeStates("w1", "m1", nil)
	state, ok := states["tsk_1"]
	if !ok {
		t.Fatalf("ScopeStates replay hydrate missing task-scope: %+v", states)
	}
	if state.ScopeType != ScopeTypeTask || !state.ManualUnread || !state.HasUnread() || state.ServingState != ServingStateSnapshotServing {
		t.Fatalf("task-scope state = %+v, want snapshot-serving manual unread", state)
	}
	if store.snapshotLoads != 1 {
		t.Fatalf("snapshotLoads = %d, want 1", store.snapshotLoads)
	}
	if !service.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread after replay hydrate = false, want true")
	}

	_ = service.ScopeStates("w1", "m1", nil)
	if store.snapshotLoads != 1 {
		t.Fatalf("snapshotLoads after repeated unscoped state read = %d, want 1", store.snapshotLoads)
	}
}

func TestServiceColdStartPreservesDurableTaskScopeTypeWithoutIDPrefix(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 8, HydrateBatchSize: 2})
	store := &fakeServiceStore{
		snapshots: []Snapshot{{
			WorkspaceID:        "w1",
			MemberID:           "m1",
			ScopeID:            "custom_task_scope",
			MessageUnreadCount: 4,
			ServingState:       ServingStateSnapshotServing,
		}},
		metadata: []FrontierMetadata{{
			WorkspaceID:      "w1",
			MemberID:         "m1",
			ScopeID:          "custom_task_scope",
			ScopeType:        ScopeTypeTask,
			LatestMessageSeq: 4,
		}},
	}
	service := newTestService(t, runtime, store)

	state, ok := service.ScopeStates("w1", "m1", nil)["custom_task_scope"]
	if !ok {
		t.Fatal("cold-start task scope state missing")
	}
	if state.ScopeType != ScopeTypeTask || state.Count() != 4 {
		t.Fatalf("cold-start task scope state = %+v, want task count 4", state)
	}
	if !service.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread = false, want true for durable Task scope")
	}
	if got := service.WorkspaceUnreadConversationCount("w1", "m1"); got != 1 {
		t.Fatalf("WorkspaceUnreadConversationCount = %d, want 1 for durable Task scope", got)
	}

	if err := runtime.WarmScopeRoster("w1", "custom_task_scope", "", []RosterMember{{
		MemberID: "m1",
		UserID:   "u1",
	}}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}
	deltas, err := service.ApplyScopeInvalidatedWithDeltas(ScopeInvalidated{
		WorkspaceID: "w1",
		ScopeID:     "custom_task_scope",
		MemberIDs:   []string{"m1"},
		Reason:      "test_scope_type",
	})
	if err != nil {
		t.Fatalf("ApplyScopeInvalidatedWithDeltas returned error: %v", err)
	}
	if len(deltas) != 1 || deltas[0].ScopeType != ScopeTypeTask {
		t.Fatalf("cold-start task delta = %+v, want scopeType task", deltas)
	}
}

func TestServiceMemberSnapshotHydratePaginatesAndMarksOnce(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 8, HydrateBatchSize: 2})
	store := &fakeServiceStore{
		snapshots: []Snapshot{
			{WorkspaceID: "w1", MemberID: "m1", ScopeID: "chn_a", MessageUnreadCount: 1, ServingState: ServingStateSnapshotServing},
			{WorkspaceID: "w1", MemberID: "m1", ScopeID: "chn_b", ManualUnread: true, ServingState: ServingStateSnapshotServing},
			{WorkspaceID: "w1", MemberID: "m1", ScopeID: "chn_c", MessageUnreadCount: 2, ServingState: ServingStateSnapshotServing},
		},
	}
	service := newTestService(t, runtime, store)

	if !service.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread after paginated member snapshot hydrate = false, want true")
	}
	states := service.ScopeStates("w1", "m1", nil)
	if len(states) != 3 {
		t.Fatalf("len(ScopeStates) = %d, want 3: %+v", len(states), states)
	}
	if store.snapshotLoads != 2 {
		t.Fatalf("snapshotLoads = %d, want 2 paginated member reads", store.snapshotLoads)
	}

	_ = service.ScopeStates("w1", "m1", nil)
	_ = service.WorkspaceHasUnread("w1", "m1")
	if store.snapshotLoads != 2 {
		t.Fatalf("snapshotLoads after repeated member reads = %d, want 2", store.snapshotLoads)
	}
}

func TestServiceTaskSnapshotActivityFeedsWorkspacePresence(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 8, HydrateBatchSize: 2})
	store := &fakeServiceStore{
		snapshots: []Snapshot{
			{WorkspaceID: "w1", MemberID: "m1", ScopeID: "tsk_1", ActivityUnreadCount: 4, ServingState: ServingStateSnapshotServing},
		},
	}
	service := newTestService(t, runtime, store)

	if !service.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread for task activity-only snapshot = false, want true")
	}
	states := service.ScopeStates("w1", "m1", nil)
	state := states["tsk_1"]
	if state.ScopeType != ScopeTypeTask || state.Count() != 4 || !state.HasUnread() {
		t.Fatalf("task activity-only snapshot state = %+v, want task unread count 4", state)
	}
	if store.snapshotLoads != 1 {
		t.Fatalf("snapshotLoads = %d, want 1", store.snapshotLoads)
	}
}

func TestServiceApplyMessageColdRosterQueuesRosterWarmAndReplaysPending(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 4})
	store := &fakeServiceStore{
		roster: []RosterMember{
			{MemberID: "author", Visibility: "public", NotificationState: "active"},
			{MemberID: "m1", Visibility: "public", NotificationState: "active"},
		},
	}
	service := newTestService(t, runtime, store)
	deltas := &fakeDeltaSink{}
	service.SetDeltaSink(deltas)

	result, err := service.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 cold roster", result)
	}
	var job hydrateJob
	select {
	case job = <-service.queue:
		if job.kind != hydrateJobRoster {
			t.Fatalf("queued job kind = %s, want roster", job.kind)
		}
	default:
		t.Fatal("cold roster did not queue roster warm job")
	}
	if err := service.runHydrateJob(context.Background(), job); err != nil {
		t.Fatalf("runHydrateJob returned error: %v", err)
	}
	service.finish(job)
	if store.rosterLoads != 1 || store.messageTailLoads != 0 {
		t.Fatalf("unexpected recovery reads: roster=%d messageTail=%d", store.rosterLoads, store.messageTailLoads)
	}
	if got := service.ScopeUnreadCounts("w1", "m1", []string{"s1"})["s1"]; got != 1 {
		t.Fatalf("ScopeUnreadCounts after roster warm = %d, want 1", got)
	}
	if len(deltas.items) != 1 || deltas.items[0].MemberID != "m1" || deltas.items[0].Count != 1 {
		t.Fatalf("unread deltas after pending replay = %+v, want m1 count 1", deltas.items)
	}
}

func TestServiceOverBudgetRosterWarmsAndCompletesWithRecoveryBudget(t *testing.T) {
	tests := []struct {
		name  string
		apply func(*Service) (ApplyResult, error)
	}{
		{
			name: "message",
			apply: func(service *Service) (ApplyResult, error) {
				return service.ApplyMessage(MessageStored{
					WorkspaceID:    "w1",
					ScopeID:        "s1",
					ScopeType:      ScopeTypeChannel,
					Seq:            7,
					AuthorMemberID: "author",
				})
			},
		},
		{
			name: "activity",
			apply: func(service *Service) (ApplyResult, error) {
				return service.ApplyNonMessageActivity(NonMessageActivityStored{
					WorkspaceID:   "w1",
					ScopeID:       "s1",
					ScopeType:     ScopeTypeChannel,
					Seq:           8,
					ActorMemberID: "author",
				})
			},
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			roster := []RosterMember{
				{MemberID: "author", UserID: "usr_author", Visibility: "public", NotificationState: "active"},
				{MemberID: "m1", UserID: "usr_1", Visibility: "public", NotificationState: "active"},
				{MemberID: "m2", UserID: "usr_2", Visibility: "public", NotificationState: "active"},
			}
			runtime := newTestRuntime(t, Config{
				EventQueueSize:           4,
				MaxSyncAccountingMembers: 2,
			})
			if err := runtime.WarmScopeRoster("w1", "s1", ScopeTypeChannel, roster); err != nil {
				t.Fatalf("WarmScopeRoster returned error: %v", err)
			}
			store := &fakeServiceStore{roster: roster}
			service := newTestService(t, runtime, store)

			result, err := test.apply(service)
			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 over-budget roster", result)
			}
			var job hydrateJob
			select {
			case job = <-service.queue:
				if job.kind != hydrateJobRoster {
					t.Fatalf("queued job kind = %s, want roster", job.kind)
				}
			default:
				t.Fatal("over-budget roster did not enqueue bounded warm")
			}
			if err := service.runHydrateJob(context.Background(), job); err != nil {
				t.Fatalf("runHydrateJob returned error: %v", err)
			}
			service.finish(job)
			if store.rosterLoads != 1 {
				t.Fatalf("rosterLoads = %d, want 1", store.rosterLoads)
			}
			for _, memberID := range []string{"m1", "m2"} {
				if got := service.ScopeUnreadCounts("w1", memberID, []string{"s1"})["s1"]; got != 1 {
					t.Fatalf("%s unread after recovered accounting = %d, want 1", memberID, got)
				}
			}
			if !runtime.ScopeRosterReady("w1", "s1") {
				t.Fatal("recovered apply coldified the refreshed large roster")
			}
		})
	}
}

func TestServiceRefreshScopeRosterReloadsAndRestoresUnreadEligibility(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 4})
	store := &fakeServiceStore{
		roster: []RosterMember{
			{MemberID: "author", Visibility: "public", NotificationState: "active"},
			{MemberID: "m1", UserID: "u1", Visibility: "public", NotificationState: "active"},
		},
	}
	service := newTestService(t, runtime, store)
	if err := runtime.WarmScopeRoster("w1", "s1", ScopeTypeChannel, store.roster); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}
	applyMessage := func(seq int64) {
		t.Helper()
		result, err := service.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)
		}
	}
	count := func() int64 {
		t.Helper()
		return service.ScopeUnreadCounts("w1", "m1", []string{"s1"})["s1"]
	}

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

	// RefreshScopeRoster forces the roster cold and queues a bounded reload from
	// the store, which would be skipped by the plain enqueueRosterWarm path while
	// rosterReady was still true.
	if err := service.RefreshScopeRoster("w1", "s1", ScopeTypeChannel, []string{"m1"}); err != nil {
		t.Fatalf("RefreshScopeRoster returned error: %v", err)
	}
	if runtime.ScopeRosterReady("w1", "s1") {
		t.Fatal("ScopeRosterReady after RefreshScopeRoster = true, want false")
	}
	var job hydrateJob
	select {
	case job = <-service.queue:
		if job.kind != hydrateJobRoster || job.scopeID != "s1" {
			t.Fatalf("queued job = %+v, want roster reload for s1", job)
		}
	default:
		t.Fatal("RefreshScopeRoster did not queue a roster reload")
	}
	if err := service.runHydrateJob(context.Background(), job); err != nil {
		t.Fatalf("runHydrateJob returned error: %v", err)
	}
	service.finish(job)
	if !runtime.ScopeRosterReady("w1", "s1") {
		t.Fatal("ScopeRosterReady after reload = false, want true")
	}
	if store.rosterLoads == 0 {
		t.Fatal("rosterLoads = 0, want the store roster reloaded")
	}

	// The next eligible message regains exactly one unread point.
	result, err := service.ApplyMessage(MessageStored{
		WorkspaceID: "w1", ScopeID: "s1", ScopeType: ScopeTypeChannel,
		Seq: 3, AuthorMemberID: "author",
	})
	if err != nil {
		t.Fatalf("ApplyMessage(seq=3) returned error: %v", err)
	}
	if result.Pending || result.Ignored {
		t.Fatalf("ApplyMessage(seq=3) result = %+v, want applied", result)
	}
	if len(result.Deltas) != 1 || result.Deltas[0].MemberID != "m1" || result.Deltas[0].Count != 1 {
		t.Fatalf("reactivation deltas = %+v, want m1 count 1", result.Deltas)
	}
	if got := count(); got != 1 {
		t.Fatalf("unread after reactivation message = %d, want exactly 1", got)
	}
}

// fencedRosterStore blocks the first roster page load until the test releases
// it, so the test can deterministically interleave a refresh while an older
// hydrate is in flight. The first load returns the pre-refresh (mute) roster;
// every later load returns the post-refresh (active) roster.
type fencedRosterStore struct {
	*fakeServiceStore
	oldLoadStarted chan struct{}
	releaseOldLoad chan struct{}
	muteRoster     []RosterMember
	activeRoster   []RosterMember
	loads          int
}

func (store *fencedRosterStore) LoadScopeRosterPage(_ context.Context, _ string, _ string, _ string, _ int) ([]RosterMember, error) {
	store.loads++
	if store.loads == 1 {
		close(store.oldLoadStarted)
		<-store.releaseOldLoad
		return store.muteRoster, nil
	}
	return store.activeRoster, nil
}

// TestServiceRefreshScopeRosterNotSwallowedByInflightStaleHydrate reproduces the
// linearization bug deterministically: an older roster hydrate reads the pre-
// refresh mute roster and is still in flight (holding the roster job key) when
// active/quiet persists and RefreshScopeRoster runs. The refresh enqueue is
// deduplicated away by the in-flight key, so the fix must rely on the roster
// generation fence: the stale hydrate must not publish ready, and its finish
// must force the next generation from the store so the very next eligible
// message regains exactly one unread point.
func TestServiceRefreshScopeRosterNotSwallowedByInflightStaleHydrate(t *testing.T) {
	ctx := context.Background()
	runtime := newTestRuntime(t, Config{EventQueueSize: 8})
	store := &fencedRosterStore{
		fakeServiceStore: &fakeServiceStore{},
		oldLoadStarted:   make(chan struct{}),
		releaseOldLoad:   make(chan struct{}),
		muteRoster: []RosterMember{
			{MemberID: "author", Visibility: "public", NotificationState: "active"},
			{MemberID: "m1", UserID: "u1", Visibility: "public", NotificationState: "mute"},
		},
		activeRoster: []RosterMember{
			{MemberID: "author", Visibility: "public", NotificationState: "active"},
			{MemberID: "m1", UserID: "u1", Visibility: "public", NotificationState: "active"},
		},
	}
	service := newTestService(t, runtime, store)
	deltas := &fakeDeltaSink{}
	service.SetDeltaSink(deltas)

	// Start the old hydrate as if a cold-warm or prior refresh had enqueued it.
	// Marking it inflight is what makes RefreshScopeRoster's enqueue drop.
	oldJob := hydrateJob{kind: hydrateJobRoster, workspaceID: "w1", scopeID: "s1", scopeType: ScopeTypeChannel}
	service.enqueue(oldJob)
	queued := <-service.queue
	if queued.key() != oldJob.key() {
		t.Fatalf("queued job = %+v, want old roster hydrate", queued)
	}

	var oldErr error
	oldDone := make(chan struct{})
	go func() {
		defer close(oldDone)
		oldErr = service.runHydrateJob(ctx, queued)
		service.finish(queued)
	}()

	// The old hydrate captured the pre-refresh generation and is blocked at the
	// store load where it would read the mute roster.
	<-store.oldLoadStarted

	// active/quiet persists and the refresh runs while the same job key is
	// inflight: this enqueue is deduplicated away.
	if err := service.RefreshScopeRoster("w1", "s1", ScopeTypeChannel, []string{"m1"}); err != nil {
		t.Fatalf("RefreshScopeRoster returned error: %v", err)
	}
	if runtime.ScopeRosterReady("w1", "s1") {
		t.Fatal("ScopeRosterReady immediately after refresh = true, want false")
	}

	// Release the old hydrate. It must not publish the stale mute roster.
	close(store.releaseOldLoad)
	<-oldDone
	if oldErr != nil {
		t.Fatalf("old hydrate returned error: %v", oldErr)
	}
	if runtime.ScopeRosterReady("w1", "s1") {
		t.Fatal("stale mute hydrate published ready after refresh")
	}

	// The refresh is not lost: the old hydrate's finish must force the next
	// generation from the store, which reads the active roster.
	var nextJob hydrateJob
	select {
	case nextJob = <-service.queue:
		if nextJob.kind != hydrateJobRoster || nextJob.scopeID != "s1" {
			t.Fatalf("next-generation job = %+v, want roster hydrate for s1", nextJob)
		}
	default:
		t.Fatal("refresh was swallowed: no next-generation roster hydrate queued")
	}
	if err := service.runHydrateJob(ctx, nextJob); err != nil {
		t.Fatalf("next-generation hydrate returned error: %v", err)
	}
	service.finish(nextJob)
	if !runtime.ScopeRosterReady("w1", "s1") {
		t.Fatal("ScopeRosterReady after next-generation hydrate = false, want true")
	}
	if store.loads != 2 {
		t.Fatalf("roster page loads = %d, want 2 (stale old load + next generation)", store.loads)
	}

	// The next eligible message regains exactly one unread point.
	result, err := service.ApplyMessage(MessageStored{
		WorkspaceID: "w1", ScopeID: "s1", ScopeType: ScopeTypeChannel,
		Seq: 3, AuthorMemberID: "author",
	})
	if err != nil {
		t.Fatalf("ApplyMessage returned error: %v", err)
	}
	if result.Pending || result.Ignored {
		t.Fatalf("ApplyMessage result = %+v, want applied", result)
	}
	if len(result.Deltas) != 1 || result.Deltas[0].MemberID != "m1" || result.Deltas[0].Count != 1 {
		t.Fatalf("reactivation deltas = %+v, want m1 count 1", result.Deltas)
	}
	if got := service.ScopeUnreadCounts("w1", "m1", []string{"s1"})["s1"]; got != 1 {
		t.Fatalf("unread after reactivation message = %d, want exactly 1", got)
	}
}

// TestServiceRefreshGenerationSurvivesHydrateErrorAndRetry verifies an error or
// retry path never clears an unsatisfied refresh generation: the roster stays
// non-ready after a failed hydrate and the scheduled retry satisfies the same
// refresh generation from the store.
func TestServiceRefreshGenerationSurvivesHydrateErrorAndRetry(t *testing.T) {
	ctx := context.Background()
	runtime := newTestRuntime(t, Config{EventQueueSize: 8})
	store := &fakeServiceStore{
		roster: []RosterMember{
			{MemberID: "author", Visibility: "public", NotificationState: "active"},
			{MemberID: "m1", UserID: "u1", Visibility: "public", NotificationState: "active"},
		},
		rosterErr: errors.New("temporary roster read failure"),
	}
	service := newTestService(t, runtime, store)

	if err := service.RefreshScopeRoster("w1", "s1", ScopeTypeChannel, []string{"m1"}); err != nil {
		t.Fatalf("RefreshScopeRoster returned error: %v", err)
	}
	var job hydrateJob
	select {
	case job = <-service.queue:
	default:
		t.Fatal("RefreshScopeRoster did not queue a roster reload")
	}
	err := service.runHydrateJob(ctx, job)
	if err == nil {
		t.Fatal("runHydrateJob returned nil for transient roster failure")
	}
	service.finish(job, err)

	// The failed hydrate must not clear the pending refresh: the roster stays
	// non-ready and the refresh generation is still unsatisfied.
	if runtime.ScopeRosterReady("w1", "s1") {
		t.Fatal("ScopeRosterReady after failed hydrate = true, want false (refresh unsatisfied)")
	}

	// Pull the scheduled retry deterministically without relying on backoff
	// wall-clock timing, then satisfy the same refresh generation.
	service.mu.Lock()
	retryState, ok := service.retry[job.key()]
	service.mu.Unlock()
	if !ok {
		t.Fatal("failed hydrate did not schedule a retry for the refresh generation")
	}
	store.rosterErr = nil
	service.enqueue(retryState.job)
	retryJob := <-service.queue
	if retryJob.key() != job.key() {
		t.Fatalf("retry job = %+v, want %+v", retryJob, job)
	}
	if err := service.runHydrateJob(ctx, retryJob); err != nil {
		t.Fatalf("retry runHydrateJob returned error: %v", err)
	}
	service.finish(retryJob)
	if !runtime.ScopeRosterReady("w1", "s1") {
		t.Fatal("ScopeRosterReady after retry = false, want true (refresh satisfied)")
	}

	result, err := service.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 {
		t.Fatalf("ApplyMessage result = %+v, want applied", result)
	}
	if len(result.Deltas) != 1 || result.Deltas[0].MemberID != "m1" || result.Deltas[0].Count != 1 {
		t.Fatalf("deltas after retry = %+v, want m1 count 1", result.Deltas)
	}
}

func TestServiceApplyMessageWarmRosterUpdatesMemoryOnly(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 4})
	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)
	}
	store := &fakeServiceStore{}
	service := newTestService(t, runtime, store)

	result, err := service.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.AccountedMembers != 1 {
		t.Fatalf("ApplyMessage result = %+v, want one in-memory account", result)
	}
	if got := service.ScopeUnreadCounts("w1", "m1", []string{"s1"})["s1"]; got != 1 {
		t.Fatalf("ScopeUnreadCounts = %d, want 1", got)
	}
	select {
	case job := <-service.queue:
		t.Fatalf("warm roster queued recovery job: %+v", job)
	default:
	}
}

func TestServiceApplyTaskActivityColdRosterReplaysFrozenRecipients(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 4})
	store := &fakeServiceStore{
		roster: []RosterMember{
			{MemberID: "actor", Visibility: "public", NotificationState: "active"},
			{MemberID: "m1", UserID: "u1", Visibility: "public", NotificationState: "active"},
		},
	}
	service := newTestService(t, runtime, store)
	deltas := &fakeDeltaSink{}
	service.SetDeltaSink(deltas)

	result, err := service.ApplyNonMessageActivity(NonMessageActivityStored{
		WorkspaceID:        "w1",
		ScopeID:            "task_scope_1",
		ScopeType:          ScopeTypeTask,
		Seq:                9,
		ActorMemberID:      "actor",
		ActivityGeneration: 9,
		RecipientMemberIDs: []string{"m1"},
	})
	if err != nil {
		t.Fatalf("ApplyNonMessageActivity returned error: %v", err)
	}
	if result.Ignored || !result.Pending {
		t.Fatalf("ApplyNonMessageActivity result = %+v, want pending roster recovery", result)
	}
	var job hydrateJob
	select {
	case job = <-service.queue:
		if job.kind != hydrateJobRoster {
			t.Fatalf("queued job kind = %s, want roster", job.kind)
		}
	default:
		t.Fatal("task activity did not queue roster recovery")
	}
	if err := service.runHydrateJob(context.Background(), job); err != nil {
		t.Fatalf("runHydrateJob returned error: %v", err)
	}
	service.finish(job)
	if store.rosterLoads != 1 || store.activityTailLoads != 0 {
		t.Fatalf("unexpected recovery reads: roster=%d activityTail=%d", store.rosterLoads, store.activityTailLoads)
	}
	if got := service.ScopeUnreadCounts("w1", "m1", []string{"task_scope_1"})["task_scope_1"]; got != 1 {
		t.Fatalf("task activity unread after replay = %d, want 1", got)
	}
	if len(deltas.items) != 1 || deltas.items[0].MemberID != "m1" || deltas.items[0].Count != 1 {
		t.Fatalf("unread deltas after task activity replay = %+v, want m1 count 1", deltas.items)
	}
}

func TestServiceTaskPendingReplayMergesRecipientSnapshotAcrossRestart(t *testing.T) {
	type eventKind struct {
		name  string
		apply func(*Runtime, int64) (ApplyResult, error)
		live  func(*Service, int64) (ApplyResult, error)
		count func(ScopeUnreadState) int64
	}
	eventKinds := []eventKind{
		{
			name: "message",
			apply: func(runtime *Runtime, seq int64) (ApplyResult, error) {
				return runtime.ApplyMessage(MessageStored{
					WorkspaceID:        "w1",
					ScopeID:            "tsk_1",
					ScopeType:          ScopeTypeTask,
					Seq:                seq,
					AuthorMemberID:     "actor",
					MessageGeneration:  seq,
					RecipientMemberIDs: []string{"m1"},
				})
			},
			live: func(service *Service, seq int64) (ApplyResult, error) {
				return service.ApplyMessage(MessageStored{
					WorkspaceID:        "w1",
					ScopeID:            "tsk_1",
					ScopeType:          ScopeTypeTask,
					Seq:                seq,
					AuthorMemberID:     "actor",
					MessageGeneration:  seq,
					RecipientMemberIDs: []string{"m1"},
				})
			},
			count: func(state ScopeUnreadState) int64 {
				return state.MessageUnreadCount
			},
		},
		{
			name: "activity",
			apply: func(runtime *Runtime, seq int64) (ApplyResult, error) {
				return runtime.ApplyNonMessageActivity(NonMessageActivityStored{
					WorkspaceID:        "w1",
					ScopeID:            "tsk_1",
					ScopeType:          ScopeTypeTask,
					Seq:                seq,
					ActorMemberID:      "actor",
					ActivityGeneration: seq,
					RecipientMemberIDs: []string{"m1"},
				})
			},
			live: func(service *Service, seq int64) (ApplyResult, error) {
				return service.ApplyNonMessageActivity(NonMessageActivityStored{
					WorkspaceID:        "w1",
					ScopeID:            "tsk_1",
					ScopeType:          ScopeTypeTask,
					Seq:                seq,
					ActorMemberID:      "actor",
					ActivityGeneration: seq,
					RecipientMemberIDs: []string{"m1"},
				})
			},
			count: func(state ScopeUnreadState) int64 {
				return state.ActivityUnreadCount
			},
		},
	}
	scenarios := []struct {
		name              string
		snapshotThrough   int64
		pendingSeq        int64
		activeRecipient   bool
		wantCount         int64
		wantSnapshotLoads int
	}{
		{
			name:              "historical_snapshot_plus_pending",
			snapshotThrough:   3,
			pendingSeq:        4,
			activeRecipient:   true,
			wantCount:         4,
			wantSnapshotLoads: 1,
		},
		{
			name:              "snapshot_already_includes_pending",
			snapshotThrough:   4,
			pendingSeq:        4,
			activeRecipient:   true,
			wantCount:         4,
			wantSnapshotLoads: 1,
		},
		{
			name:              "snapshot_miss_uses_zero_baseline",
			snapshotThrough:   0,
			pendingSeq:        1,
			activeRecipient:   true,
			wantCount:         1,
			wantSnapshotLoads: 1,
		},
		{
			name:              "inactive_recipient_is_dropped",
			snapshotThrough:   3,
			pendingSeq:        4,
			activeRecipient:   false,
			wantCount:         0,
			wantSnapshotLoads: 0,
		},
	}

	for _, kind := range eventKinds {
		for _, scenario := range scenarios {
			t.Run(kind.name+"/"+scenario.name, func(t *testing.T) {
				config := Config{EventQueueSize: 8, HydrateBatchSize: 1}
				roster := []RosterMember{
					{MemberID: "actor", UserID: "usr_actor", Visibility: "public"},
					{MemberID: "m1", UserID: "usr_reader", Visibility: "public"},
				}

				var snapshots []Snapshot
				if scenario.snapshotThrough > 0 {
					runtime1 := newTestRuntime(t, config)
					if err := runtime1.WarmScopeRoster("w1", "tsk_1", ScopeTypeTask, roster); err != nil {
						t.Fatalf("runtime1 WarmScopeRoster returned error: %v", err)
					}
					for seq := int64(1); seq <= scenario.snapshotThrough; seq++ {
						result, err := kind.apply(runtime1, seq)
						if err != nil {
							t.Fatalf("runtime1 apply seq %d returned error: %v", seq, err)
						}
						if result.Pending || result.Ignored || result.AccountedMembers != 1 {
							t.Fatalf("runtime1 apply seq %d result = %+v, want one account", seq, result)
						}
					}
					snapshots = runtime1.FlushDirtySnapshots(8, time.Now().UTC())
					if len(snapshots) != 1 {
						t.Fatalf("runtime1 snapshots = %+v, want recipient snapshot", snapshots)
					}
					snapshots[0].ServingState = ServingStateSnapshotServing
				}

				recoveryRoster := roster
				if !scenario.activeRecipient {
					recoveryRoster = roster[:1]
				}
				metadata := []FrontierMetadata{{
					WorkspaceID:             "w1",
					MemberID:                "m1",
					ScopeID:                 "tsk_1",
					ScopeType:               ScopeTypeTask,
					LatestMessageSeq:        scenario.pendingSeq,
					LatestActivitySeq:       scenario.pendingSeq,
					ScopeMessageGeneration:  scenario.pendingSeq,
					ScopeActivityGeneration: scenario.pendingSeq,
					MembershipVisibility:    "public",
				}}
				store := &fakeServiceStore{
					snapshots: snapshots,
					metadata:  metadata,
					roster:    recoveryRoster,
				}
				runtime2 := newTestRuntime(t, config)
				service := newTestService(t, runtime2, store)

				result, err := kind.live(service, scenario.pendingSeq)
				if err != nil {
					t.Fatalf("runtime2 initial apply returned error: %v", err)
				}
				if result.Ignored || !result.Pending {
					t.Fatalf("runtime2 initial apply result = %+v, want cold pending", result)
				}
				job := <-service.queue
				if job.kind != hydrateJobRoster {
					t.Fatalf("queued job kind = %s, want roster", job.kind)
				}
				if err := service.runHydrateJob(context.Background(), job); err != nil {
					t.Fatalf("runHydrateJob returned error: %v", err)
				}
				service.finish(job)

				if store.snapshotLoads != scenario.wantSnapshotLoads {
					t.Fatalf("snapshotLoads = %d, want %d", store.snapshotLoads, scenario.wantSnapshotLoads)
				}
				state, exists := runtime2.ScopeStates("w1", "m1", []string{"tsk_1"})["tsk_1"]
				if scenario.activeRecipient {
					if !exists || kind.count(state) != scenario.wantCount || state.Count() != scenario.wantCount {
						t.Fatalf("runtime2 recovered state = %+v (exists %v), want count %d", state, exists, scenario.wantCount)
					}
				} else if exists {
					t.Fatalf("inactive recipient recovered stale snapshot state: %+v", state)
				}

				result, err = kind.live(service, scenario.pendingSeq)
				if err != nil {
					t.Fatalf("runtime2 exact live retry returned error: %v", err)
				}
				if !result.Ignored || result.Pending {
					t.Fatalf("runtime2 exact live retry result = %+v, want stable ignored completion", result)
				}
				select {
				case followup := <-service.queue:
					t.Fatalf("exact retry queued unexpected hydrate job: %+v", followup)
				default:
				}
			})
		}
	}
}

func TestServicePersistTaskUnreadSnapshotsSurvivesCrashBeforeBackgroundFlush(t *testing.T) {
	const (
		workspaceID = "w1"
		scopeID     = "tsk_snapshot_completion_barrier"
		memberID    = "m1"
	)
	runtime1 := newTestRuntime(t, Config{EventQueueSize: 8})
	if err := runtime1.WarmScopeRoster(workspaceID, scopeID, ScopeTypeTask, []RosterMember{
		{MemberID: "actor", UserID: "usr_actor", Visibility: "public"},
		{MemberID: memberID, UserID: "usr_reader", Visibility: "public"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}
	result, err := runtime1.ApplyMessage(MessageStored{
		WorkspaceID:        workspaceID,
		ScopeID:            scopeID,
		ScopeType:          ScopeTypeTask,
		Seq:                1,
		AuthorMemberID:     "actor",
		MessageGeneration:  1,
		RecipientMemberIDs: []string{memberID},
	})
	if err != nil {
		t.Fatalf("ApplyMessage returned error: %v", err)
	}
	if result.Pending || result.AccountedMembers != 1 {
		t.Fatalf("ApplyMessage result = %+v, want one accounted member", result)
	}

	durableStore := &fakeServiceStore{}
	service1 := newTestService(t, runtime1, durableStore)
	if err := service1.PersistTaskUnreadSnapshots(
		context.Background(),
		workspaceID,
		scopeID,
		[]string{memberID},
	); err != nil {
		t.Fatalf("PersistTaskUnreadSnapshots returned error: %v", err)
	}
	if len(durableStore.upserted) != 1 ||
		durableStore.upserted[0].MessageUnreadCount != 1 ||
		durableStore.upserted[0].LatestMessageAccountedSeq != 1 {
		t.Fatalf("persisted snapshots = %+v, want one message-unread snapshot", durableStore.upserted)
	}
	// Simulate an abrupt process loss before the periodic dirty flush. The
	// second owner sees only the row written by the targeted completion
	// barrier; no pending outbox replay is available after completion.
	restartStore := &fakeServiceStore{
		snapshots: append([]Snapshot(nil), durableStore.upserted...),
		metadata: []FrontierMetadata{{
			WorkspaceID:             workspaceID,
			MemberID:                memberID,
			ScopeID:                 scopeID,
			ScopeType:               ScopeTypeTask,
			LatestMessageSeq:        1,
			ScopeMessageGeneration:  1,
			MembershipVisibility:    "public",
			ScopeActivityGeneration: 0,
		}},
	}
	runtime2 := newTestRuntime(t, Config{EventQueueSize: 8})
	service2 := newTestService(t, runtime2, restartStore)
	if err := service2.hydrateSnapshotJob(context.Background(), hydrateJob{
		kind:        hydrateJobSnapshot,
		workspaceID: workspaceID,
		memberID:    memberID,
		scopeIDs:    []string{scopeID},
	}); err != nil {
		t.Fatalf("restart hydrateSnapshotJob returned error: %v", err)
	}
	recovered := service2.ScopeStates(workspaceID, memberID, []string{scopeID})[scopeID]
	if recovered.MessageUnreadCount != 1 || !recovered.HasUnread() {
		t.Fatalf("recovered state = %+v, want durable unread", recovered)
	}
}

func TestServiceTaskSnapshotBarrierIncludesEventQueuedDuringHydrate(t *testing.T) {
	runtime := newTestRuntime(t, Config{
		EventQueueSize:      8,
		HydrateBatchSize:    1,
		AccountingBatchSize: 1,
	})
	store := &fakeServiceStore{
		roster: []RosterMember{
			{MemberID: "actor", UserID: "usr_actor", Visibility: "public"},
			{MemberID: "m1", UserID: "usr_1", Visibility: "public"},
			{MemberID: "m2", UserID: "usr_2", Visibility: "public"},
		},
		snapshots: []Snapshot{
			{
				WorkspaceID:               "w1",
				MemberID:                  "m1",
				ScopeID:                   "tsk_1",
				ScopeType:                 ScopeTypeTask,
				MessageUnreadCount:        2,
				LatestMessageAccountedSeq: 2,
				ServingState:              ServingStateSnapshotServing,
			},
			{
				WorkspaceID:                "w1",
				MemberID:                   "m2",
				ScopeID:                    "tsk_1",
				ScopeType:                  ScopeTypeTask,
				ActivityUnreadCount:        5,
				LatestActivityAccountedSeq: 5,
				ServingState:               ServingStateSnapshotServing,
			},
		},
		metadata: []FrontierMetadata{
			{
				WorkspaceID:          "w1",
				MemberID:             "m1",
				ScopeID:              "tsk_1",
				ScopeType:            ScopeTypeTask,
				LatestMessageSeq:     3,
				MembershipVisibility: "public",
			},
			{
				WorkspaceID:          "w1",
				MemberID:             "m2",
				ScopeID:              "tsk_1",
				ScopeType:            ScopeTypeTask,
				LatestActivitySeq:    6,
				MembershipVisibility: "public",
			},
		},
	}
	service := newTestService(t, runtime, store)
	var injectedResult ApplyResult
	var injectedErr error
	store.afterSnapshotLoad = func() {
		injectedResult, injectedErr = service.ApplyNonMessageActivity(NonMessageActivityStored{
			WorkspaceID:        "w1",
			ScopeID:            "tsk_1",
			ScopeType:          ScopeTypeTask,
			Seq:                6,
			ActorMemberID:      "actor",
			ActivityGeneration: 6,
			RecipientMemberIDs: []string{"m2"},
		})
	}

	result, err := service.ApplyMessage(MessageStored{
		WorkspaceID:        "w1",
		ScopeID:            "tsk_1",
		ScopeType:          ScopeTypeTask,
		Seq:                3,
		AuthorMemberID:     "actor",
		MessageGeneration:  3,
		RecipientMemberIDs: []string{"m1"},
	})
	if err != nil {
		t.Fatalf("initial message returned error: %v", err)
	}
	if !result.Pending || result.Ignored {
		t.Fatalf("initial message result = %+v, want cold pending", result)
	}
	job := <-service.queue
	if err := service.runHydrateJob(context.Background(), job); err != nil {
		t.Fatalf("runHydrateJob returned error: %v", err)
	}
	service.finish(job)

	if injectedErr != nil {
		t.Fatalf("injected activity returned error: %v", injectedErr)
	}
	if !injectedResult.Pending || injectedResult.Ignored {
		t.Fatalf("injected activity result = %+v, want pending behind snapshot barrier", injectedResult)
	}
	if store.snapshotLoads != 2 {
		t.Fatalf("snapshotLoads = %d, want both frozen recipients hydrated", store.snapshotLoads)
	}
	m1 := runtime.ScopeStates("w1", "m1", []string{"tsk_1"})["tsk_1"]
	if m1.MessageUnreadCount != 3 || m1.ActivityUnreadCount != 0 {
		t.Fatalf("m1 recovered state = %+v, want message baseline 2 + pending 1", m1)
	}
	m2 := runtime.ScopeStates("w1", "m2", []string{"tsk_1"})["tsk_1"]
	if m2.MessageUnreadCount != 0 || m2.ActivityUnreadCount != 6 {
		t.Fatalf("m2 recovered state = %+v, want activity baseline 5 + injected 1", m2)
	}
	select {
	case followup := <-service.queue:
		t.Fatalf("snapshot barrier left a redundant hydrate job: %+v", followup)
	default:
	}
}

func TestServiceTaskSnapshotBarrierMergesUnresolvedControlCallbacks(t *testing.T) {
	type testCase struct {
		name                    string
		snapshot                *Snapshot
		messageCursor           int64
		activityCursor          int64
		manualUnread            bool
		manualUnreadGeneration  int64
		clearManualWithRead     bool
		pendingMessageSeq       int64
		pendingActivitySeq      int64
		wantMessageUnreadCount  int64
		wantActivityUnreadCount int64
		wantManualUnread        bool
	}
	tests := []testCase{
		{
			name: "partial_message_cursor_preserves_opaque_baseline",
			snapshot: &Snapshot{
				MessageUnreadCount:        3,
				LatestMessageAccountedSeq: 3,
				ManualUnread:              false,
				ManualUnreadGeneration:    1,
			},
			messageCursor:          1,
			manualUnread:           true,
			manualUnreadGeneration: 2,
			pendingMessageSeq:      4,
			wantMessageUnreadCount: 4,
			wantManualUnread:       true,
		},
		{
			name: "durable_partial_read_count_adds_exact_pending",
			snapshot: &Snapshot{
				MessageUnreadCount:        2,
				LatestMessageAccountedSeq: 3,
				LastMessageReadSeq:        1,
			},
			messageCursor:          1,
			pendingMessageSeq:      4,
			wantMessageUnreadCount: 3,
		},
		{
			name: "cursor_covering_message_baseline_clears_it",
			snapshot: &Snapshot{
				MessageUnreadCount:        3,
				LatestMessageAccountedSeq: 3,
			},
			messageCursor:          3,
			pendingMessageSeq:      4,
			wantMessageUnreadCount: 1,
		},
		{
			name: "partial_activity_cursor_preserves_opaque_baseline",
			snapshot: &Snapshot{
				ActivityUnreadCount:        3,
				LatestActivityAccountedSeq: 3,
			},
			activityCursor:          1,
			pendingActivitySeq:      4,
			wantActivityUnreadCount: 4,
		},
		{
			name: "read_cursor_unversioned_manual_clear_beats_snapshot",
			snapshot: &Snapshot{
				MessageUnreadCount:        3,
				LatestMessageAccountedSeq: 3,
				ManualUnread:              true,
				ManualUnreadGeneration:    7,
			},
			messageCursor:          1,
			clearManualWithRead:    true,
			pendingMessageSeq:      4,
			wantMessageUnreadCount: 4,
			wantManualUnread:       false,
		},
		{
			name: "unversioned_manual_callback_beats_versioned_snapshot",
			snapshot: &Snapshot{
				MessageUnreadCount:        3,
				LatestMessageAccountedSeq: 3,
				ManualUnread:              false,
				ManualUnreadGeneration:    7,
			},
			manualUnread:           true,
			pendingMessageSeq:      4,
			wantMessageUnreadCount: 4,
			wantManualUnread:       true,
		},
		{
			name:                   "explicit_snapshot_miss_resolves_zero_baseline",
			snapshot:               nil,
			messageCursor:          1,
			manualUnread:           true,
			pendingMessageSeq:      2,
			wantMessageUnreadCount: 1,
			wantManualUnread:       true,
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			config := Config{EventQueueSize: 8, HydrateBatchSize: 1}
			runtime := newTestRuntime(t, config)

			if test.messageCursor > 0 {
				if err := runtime.ApplyReadCursor(ReadCursorAdvanced{
					WorkspaceID:                 "w1",
					MemberID:                    "m1",
					ScopeID:                     "tsk_1",
					LastReadSeq:                 test.messageCursor,
					ManualUnreadGenerationKnown: !test.clearManualWithRead,
				}); err != nil {
					t.Fatalf("ApplyReadCursor returned error: %v", err)
				}
			}
			if test.activityCursor > 0 {
				if err := runtime.ApplyActivityReadCursor(ActivityReadCursorAdvanced{
					WorkspaceID: "w1",
					MemberID:    "m1",
					ScopeID:     "tsk_1",
					LastReadSeq: test.activityCursor,
				}); err != nil {
					t.Fatalf("ApplyActivityReadCursor returned error: %v", err)
				}
			}
			if test.manualUnread {
				if err := runtime.ApplyManualUnread(ManualUnreadChanged{
					WorkspaceID:            "w1",
					MemberID:               "m1",
					ScopeID:                "tsk_1",
					ManualUnread:           true,
					ManualUnreadGeneration: test.manualUnreadGeneration,
				}); err != nil {
					t.Fatalf("ApplyManualUnread returned error: %v", err)
				}
			}
			if rows := runtime.FlushDirtySnapshots(8, time.Now().UTC()); len(rows) != 0 {
				t.Fatalf("unresolved control-only dirty rows flushed before snapshot barrier: %+v", rows)
			}

			var snapshots []Snapshot
			if test.snapshot != nil {
				snapshot := *test.snapshot
				snapshot.WorkspaceID = "w1"
				snapshot.MemberID = "m1"
				snapshot.ScopeID = "tsk_1"
				snapshot.ScopeType = ScopeTypeTask
				snapshot.MembershipVisibility = "public"
				snapshot.ServingState = ServingStateSnapshotServing
				snapshots = []Snapshot{snapshot}
			}
			latestMessageSeq := test.pendingMessageSeq
			if latestMessageSeq == 0 && test.snapshot != nil {
				latestMessageSeq = test.snapshot.LatestMessageAccountedSeq
			}
			latestActivitySeq := test.pendingActivitySeq
			if latestActivitySeq == 0 && test.snapshot != nil {
				latestActivitySeq = test.snapshot.LatestActivityAccountedSeq
			}
			store := &fakeServiceStore{
				roster: []RosterMember{
					{MemberID: "actor", UserID: "usr_actor", Visibility: "public"},
					{
						MemberID:            "m1",
						UserID:              "usr_reader",
						LastMessageReadSeq:  test.messageCursor,
						LastActivityReadSeq: test.activityCursor,
						Visibility:          "public",
					},
				},
				snapshots: snapshots,
				metadata: []FrontierMetadata{{
					WorkspaceID:          "w1",
					MemberID:             "m1",
					ScopeID:              "tsk_1",
					ScopeType:            ScopeTypeTask,
					LatestMessageSeq:     latestMessageSeq,
					LatestActivitySeq:    latestActivitySeq,
					LastMessageReadSeq:   test.messageCursor,
					LastActivityReadSeq:  test.activityCursor,
					MembershipVisibility: "public",
					ManualUnread:         test.snapshot != nil && test.snapshot.ManualUnread,
					ManualUnreadGeneration: func() int64 {
						if test.snapshot == nil {
							return 0
						}
						return test.snapshot.ManualUnreadGeneration
					}(),
				}},
			}
			service := newTestService(t, runtime, store)

			var result ApplyResult
			var err error
			if test.pendingMessageSeq > 0 {
				result, err = service.ApplyMessage(MessageStored{
					WorkspaceID:        "w1",
					ScopeID:            "tsk_1",
					ScopeType:          ScopeTypeTask,
					Seq:                test.pendingMessageSeq,
					AuthorMemberID:     "actor",
					MessageGeneration:  test.pendingMessageSeq,
					RecipientMemberIDs: []string{"m1"},
				})
			} else {
				result, err = service.ApplyNonMessageActivity(NonMessageActivityStored{
					WorkspaceID:        "w1",
					ScopeID:            "tsk_1",
					ScopeType:          ScopeTypeTask,
					Seq:                test.pendingActivitySeq,
					ActorMemberID:      "actor",
					ActivityGeneration: test.pendingActivitySeq,
					RecipientMemberIDs: []string{"m1"},
				})
			}
			if err != nil {
				t.Fatalf("initial targeted event returned error: %v", err)
			}
			if !result.Pending || result.Ignored {
				t.Fatalf("initial targeted result = %+v, want pending unresolved baseline", result)
			}
			job := <-service.queue
			if err := service.runHydrateJob(context.Background(), job); err != nil {
				t.Fatalf("runHydrateJob returned error: %v", err)
			}
			service.finish(job)

			state := runtime.ScopeStates("w1", "m1", []string{"tsk_1"})["tsk_1"]
			if state.MessageUnreadCount != test.wantMessageUnreadCount ||
				state.ActivityUnreadCount != test.wantActivityUnreadCount ||
				state.ManualUnread != test.wantManualUnread {
				t.Fatalf("resolved state = %+v, want message=%d activity=%d manual=%v",
					state,
					test.wantMessageUnreadCount,
					test.wantActivityUnreadCount,
					test.wantManualUnread,
				)
			}
			if test.name == "unversioned_manual_callback_beats_versioned_snapshot" {
				if err := runtime.ApplyManualUnread(ManualUnreadChanged{
					WorkspaceID:            "w1",
					MemberID:               "m1",
					ScopeID:                "tsk_1",
					ManualUnread:           false,
					ManualUnreadGeneration: 4,
				}); err != nil {
					t.Fatalf("stale versioned ApplyManualUnread returned error: %v", err)
				}
				state = runtime.ScopeStates("w1", "m1", []string{"tsk_1"})["tsk_1"]
				if !state.ManualUnread {
					t.Fatalf("stale generation reversed later unversioned manual mutation: %+v", state)
				}
			}
			rows := runtime.FlushDirtySnapshots(8, time.Now().UTC())
			if len(rows) != 1 {
				t.Fatalf("resolved dirty rows = %+v, want one flushable snapshot", rows)
			}
			if rows[0].MessageUnreadCount != test.wantMessageUnreadCount ||
				rows[0].ActivityUnreadCount != test.wantActivityUnreadCount ||
				rows[0].ManualUnread != test.wantManualUnread {
				t.Fatalf("resolved flushed snapshot = %+v, want merged control state", rows[0])
			}
		})
	}
}

func TestServiceTaskControlOnlyCallbackResolvesSnapshotBeforeServeOrFlush(t *testing.T) {
	t.Run("durable snapshot", func(t *testing.T) {
		runtime := newTestRuntime(t, Config{EventQueueSize: 8})
		store := &fakeServiceStore{
			snapshots: []Snapshot{{
				WorkspaceID:               "w1",
				MemberID:                  "m1",
				ScopeID:                   "tsk_1",
				ScopeType:                 ScopeTypeTask,
				MessageUnreadCount:        3,
				LatestMessageAccountedSeq: 3,
				ManualUnread:              true,
				ManualUnreadGeneration:    7,
				MembershipVisibility:      "public",
				ServingState:              ServingStateSnapshotServing,
			}},
			metadata: []FrontierMetadata{{
				WorkspaceID:             "w1",
				MemberID:                "m1",
				ScopeID:                 "tsk_1",
				ScopeType:               ScopeTypeTask,
				LatestMessageSeq:        3,
				LastMessageReadSeq:      1,
				MembershipVisibility:    "public",
				ManualUnread:            true,
				ManualUnreadGeneration:  7,
				ScopeMessageGeneration:  3,
				ScopeActivityGeneration: 0,
			}},
		}
		service := newTestService(t, runtime, store)

		if err := service.ApplyReadCursor(ReadCursorAdvanced{
			WorkspaceID:                 "w1",
			MemberID:                    "m1",
			ScopeID:                     "tsk_1",
			LastReadSeq:                 1,
			ManualUnreadGenerationKnown: false,
		}); err != nil {
			t.Fatalf("ApplyReadCursor returned error: %v", err)
		}
		if state := service.ScopeStates("w1", "m1", []string{"tsk_1"}); len(state) != 0 {
			t.Fatalf("unresolved scoped state served before hydrate: %+v", state)
		}
		if rows := runtime.FlushDirtySnapshots(8, time.Now().UTC()); len(rows) != 0 {
			t.Fatalf("unresolved control-only rows flushed: %+v", rows)
		}

		job := <-service.queue
		if job.kind != hydrateJobSnapshot || len(job.scopeIDs) != 1 || job.scopeIDs[0] != "tsk_1" {
			t.Fatalf("control-only callback queued job = %+v, want scoped snapshot hydrate", job)
		}
		if err := service.runHydrateJob(context.Background(), job); err != nil {
			t.Fatalf("runHydrateJob returned error: %v", err)
		}
		service.finish(job)

		state := service.ScopeStates("w1", "m1", []string{"tsk_1"})["tsk_1"]
		if state.MessageUnreadCount != 3 || state.ManualUnread {
			t.Fatalf("resolved scoped state = %+v, want opaque count 3 and later manual clear", state)
		}
		rows := runtime.FlushDirtySnapshots(8, time.Now().UTC())
		if len(rows) != 1 || rows[0].MessageUnreadCount != 3 || rows[0].ManualUnread {
			t.Fatalf("resolved control state flush = %+v, want merged durable row", rows)
		}
	})

	t.Run("explicit miss", func(t *testing.T) {
		runtime := newTestRuntime(t, Config{EventQueueSize: 8})
		store := &fakeServiceStore{}
		service := newTestService(t, runtime, store)

		if err := service.ApplyManualUnread(ManualUnreadChanged{
			WorkspaceID:  "w1",
			MemberID:     "m1",
			ScopeID:      "tsk_1",
			ManualUnread: true,
		}); err != nil {
			t.Fatalf("ApplyManualUnread returned error: %v", err)
		}
		if state := service.ScopeStates("w1", "m1", []string{"tsk_1"}); len(state) != 0 {
			t.Fatalf("unresolved manual state served before miss resolution: %+v", state)
		}
		job := <-service.queue
		if err := service.runHydrateJob(context.Background(), job); err != nil {
			t.Fatalf("runHydrateJob returned error: %v", err)
		}
		service.finish(job)

		state := service.ScopeStates("w1", "m1", []string{"tsk_1"})["tsk_1"]
		if !state.ManualUnread || state.Count() != 0 {
			t.Fatalf("explicit-miss resolved state = %+v, want preserved manual unread", state)
		}
		rows := runtime.FlushDirtySnapshots(8, time.Now().UTC())
		if len(rows) != 1 || !rows[0].ManualUnread {
			t.Fatalf("explicit-miss resolved flush = %+v, want one manual row", rows)
		}
	})
}

func TestServiceTaskControlOnlySnapshotFailureRetriesWhileUnresolved(t *testing.T) {
	runtime := newTestRuntime(t, Config{
		EventQueueSize:        8,
		RecoveryRetryInterval: time.Millisecond,
	})
	store := &fakeServiceStore{
		snapshotErr: errors.New("temporary snapshot read failure"),
		snapshots: []Snapshot{{
			WorkspaceID:               "w1",
			MemberID:                  "m1",
			ScopeID:                   "tsk_1",
			ScopeType:                 ScopeTypeTask,
			MessageUnreadCount:        2,
			LatestMessageAccountedSeq: 2,
			MembershipVisibility:      "public",
			ServingState:              ServingStateSnapshotServing,
		}},
	}
	service := newTestService(t, runtime, store)
	if err := service.ApplyManualUnread(ManualUnreadChanged{
		WorkspaceID:  "w1",
		MemberID:     "m1",
		ScopeID:      "tsk_1",
		ManualUnread: true,
	}); err != nil {
		t.Fatalf("ApplyManualUnread returned error: %v", err)
	}
	job := <-service.queue
	err := service.runHydrateJob(context.Background(), job)
	if err == nil {
		t.Fatal("runHydrateJob returned nil for transient snapshot failure")
	}
	service.finish(job, err)

	store.snapshotErr = nil
	service.retryHydrateJobs()
	retryJob := <-service.queue
	if retryJob.key() != job.key() {
		t.Fatalf("retry job = %+v, want %+v", retryJob, job)
	}
	if err := service.runHydrateJob(context.Background(), retryJob); err != nil {
		t.Fatalf("retry runHydrateJob returned error: %v", err)
	}
	service.finish(retryJob)
	state := service.ScopeStates("w1", "m1", []string{"tsk_1"})["tsk_1"]
	if state.MessageUnreadCount != 2 || !state.ManualUnread {
		t.Fatalf("state after successful retry = %+v, want durable count plus manual mutation", state)
	}
}

func TestServiceApplyTaskActivityAddsToExistingTaskUnread(t *testing.T) {
	runtime := newTestRuntime(t, Config{EventQueueSize: 4})
	if err := runtime.WarmScopeRoster("w1", "task_scope_1", ScopeTypeTask, []RosterMember{
		{MemberID: "actor", UserID: "usr_actor", Visibility: "public", NotificationState: "active"},
		{MemberID: "m1", UserID: "usr_reader", Visibility: "public", NotificationState: "active"},
	}); err != nil {
		t.Fatalf("WarmScopeRoster returned error: %v", err)
	}
	service := newTestService(t, runtime, &fakeServiceStore{})
	deltas := &fakeDeltaSink{}
	service.SetDeltaSink(deltas)

	if err := service.ApplyManualUnread(ManualUnreadChanged{
		WorkspaceID:  "w1",
		MemberID:     "m1",
		ScopeID:      "task_scope_1",
		ManualUnread: true,
	}); err != nil {
		t.Fatalf("ApplyManualUnread returned error: %v", err)
	}
	before := runtime.ScopeStates("w1", "m1", []string{"task_scope_1"})["task_scope_1"]

	result, err := service.ApplyNonMessageActivity(NonMessageActivityStored{
		WorkspaceID:        "w1",
		ScopeID:            "task_scope_1",
		ScopeType:          ScopeTypeTask,
		Seq:                9,
		ActorMemberID:      "actor",
		ActivityGeneration: 9,
		RecipientMemberIDs: []string{"m1"},
	})
	if err != nil {
		t.Fatalf("ApplyNonMessageActivity returned error: %v", err)
	}
	if result.Ignored || result.Pending || result.AccountedMembers != 1 {
		t.Fatalf("ApplyNonMessageActivity result = %+v, want one targeted account", result)
	}
	after := runtime.ScopeStates("w1", "m1", []string{"task_scope_1"})["task_scope_1"]
	if after.MessageUnreadCount != before.MessageUnreadCount || after.ActivityUnreadCount != 1 || after.ManualUnread != before.ManualUnread {
		t.Fatalf("task state after typed activity = %+v, want one activity unread on %+v", after, before)
	}
	if !service.WorkspaceHasUnread("w1", "m1") {
		t.Fatal("WorkspaceHasUnread after task activity = false, want true")
	}
	select {
	case job := <-service.queue:
		t.Fatalf("task activity with warm roster queued recovery job: %+v", job)
	default:
	}
	if len(result.Deltas) != 1 || result.Deltas[0].Count != 1 || !result.Deltas[0].ManualUnread {
		t.Fatalf("task activity result deltas = %+v, want targeted unread", result.Deltas)
	}
	if len(deltas.items) != 0 {
		t.Fatalf("warm in-memory apply should not emit through recovery sink: %+v", deltas.items)
	}
}

func newTestService(t *testing.T, runtime *Runtime, store SnapshotStore) *Service {
	t.Helper()
	service, err := NewService(runtime, store, nil)
	if err != nil {
		t.Fatalf("NewService returned error: %v", err)
	}
	return service
}

type fakeDeltaSink struct {
	items []UnreadDelta
}

func (fake *fakeDeltaSink) EmitUnreadUpdated(_ string, delta UnreadDelta) {
	fake.items = append(fake.items, delta)
}

type fakeServiceStore struct {
	snapshots       []Snapshot
	metadata        []FrontierMetadata
	visibleScopeIDs []string
	roster          []RosterMember
	messageTail     []MessageStored
	activityTail    []NonMessageActivityStored

	snapshotErr       error
	metadataErr       error
	rosterErr         error
	upserted          []Snapshot
	upsertErr         error
	afterRosterLoad   func()
	afterSnapshotLoad func()
	snapshotLoads     int
	metadataLoads     int
	visibleScopeLoads int
	rosterLoads       int
	messageTailLoads  int
	activityTailLoads int
}

func (store *fakeServiceStore) LoadVisibleConversationScopeIDsPage(_ context.Context, workspaceID string, memberID string, afterScopeID string, limit int) ([]string, error) {
	store.visibleScopeLoads++
	source := store.visibleScopeIDs
	if source == nil {
		source = make([]string, 0, len(store.snapshots))
		for _, snapshot := range store.snapshots {
			if strings.TrimSpace(snapshot.WorkspaceID) == strings.TrimSpace(workspaceID) &&
				strings.TrimSpace(snapshot.MemberID) == strings.TrimSpace(memberID) {
				source = append(source, snapshot.ScopeID)
			}
		}
	}
	page := make([]string, 0, limit)
	for _, scopeID := range source {
		scopeID = strings.TrimSpace(scopeID)
		if scopeID == "" || scopeID <= strings.TrimSpace(afterScopeID) {
			continue
		}
		page = append(page, scopeID)
		if len(page) >= limit {
			break
		}
	}
	return page, nil
}

func (store *fakeServiceStore) UpsertSnapshots(_ context.Context, snapshots []Snapshot) error {
	if store.upsertErr != nil {
		return store.upsertErr
	}
	store.upserted = append(store.upserted, snapshots...)
	return nil
}

func (store *fakeServiceStore) UpsertTaskBarrierSnapshots(_ context.Context, snapshots []Snapshot) error {
	if store.upsertErr != nil {
		return store.upsertErr
	}
	store.upserted = append(store.upserted, snapshots...)
	return nil
}

func (store *fakeServiceStore) LoadSnapshotPage(_ context.Context, workspaceID string, memberID string, scopeIDs []string, afterScopeID string, limit int) ([]Snapshot, error) {
	if store.snapshotErr != nil {
		return nil, store.snapshotErr
	}
	store.snapshotLoads++
	if hook := store.afterSnapshotLoad; hook != nil {
		store.afterSnapshotLoad = nil
		hook()
	}
	allowed := stringSet(scopeIDs)
	page := make([]Snapshot, 0, limit)
	for _, item := range store.snapshots {
		if strings.TrimSpace(item.WorkspaceID) != strings.TrimSpace(workspaceID) || strings.TrimSpace(item.MemberID) != strings.TrimSpace(memberID) {
			continue
		}
		scopeID := strings.TrimSpace(item.ScopeID)
		if len(allowed) > 0 {
			if _, ok := allowed[scopeID]; !ok {
				continue
			}
		} else if scopeID <= strings.TrimSpace(afterScopeID) {
			continue
		}
		page = append(page, item)
		if len(page) >= limit {
			break
		}
	}
	return page, nil
}

func (store *fakeServiceStore) LoadFrontierMetadataPage(_ context.Context, workspaceID string, memberID string, scopeIDs []string, afterScopeID string, limit int) ([]FrontierMetadata, error) {
	if store.metadataErr != nil {
		return nil, store.metadataErr
	}
	store.metadataLoads++
	source := store.metadata
	if source == nil {
		source = store.metadataFromSnapshots()
	}
	allowed := stringSet(scopeIDs)
	page := make([]FrontierMetadata, 0, limit)
	for _, item := range source {
		if strings.TrimSpace(item.WorkspaceID) != strings.TrimSpace(workspaceID) || strings.TrimSpace(item.MemberID) != strings.TrimSpace(memberID) {
			continue
		}
		scopeID := strings.TrimSpace(item.ScopeID)
		if len(allowed) > 0 {
			if _, ok := allowed[scopeID]; !ok {
				continue
			}
		} else if scopeID <= strings.TrimSpace(afterScopeID) {
			continue
		}
		if item.ScopeMessageGeneration == 0 {
			item.ScopeMessageGeneration = item.LatestMessageSeq
		}
		if item.ScopeActivityGeneration == 0 {
			item.ScopeActivityGeneration = item.LatestActivitySeq
		}
		page = append(page, item)
		if len(page) >= limit {
			break
		}
	}
	return page, nil
}

func (store *fakeServiceStore) metadataFromSnapshots() []FrontierMetadata {
	rows := make([]FrontierMetadata, 0, len(store.snapshots))
	for _, snapshot := range store.snapshots {
		rows = append(rows, FrontierMetadata{
			WorkspaceID:             snapshot.WorkspaceID,
			MemberID:                snapshot.MemberID,
			ScopeID:                 snapshot.ScopeID,
			ScopeType:               scopeTypeForSnapshotScopeID(snapshot.ScopeID),
			LatestMessageSeq:        maxInt64(snapshot.LatestMessageAccountedSeq, snapshot.LastMessageReadSeq),
			LatestActivitySeq:       maxInt64(snapshot.LatestActivityAccountedSeq, snapshot.LastActivityReadSeq),
			ScopeMessageGeneration:  snapshot.ScopeMessageGeneration,
			ScopeActivityGeneration: snapshot.ScopeActivityGeneration,
			LastMessageReadSeq:      snapshot.LastMessageReadSeq,
			LastActivityReadSeq:     snapshot.LastActivityReadSeq,
			EffectiveFromSeq:        snapshot.EffectiveFromSeq,
			NotificationState:       snapshot.NotificationState,
			MembershipVisibility:    snapshot.MembershipVisibility,
			ManualUnread:            snapshot.ManualUnread,
			MembershipGeneration:    snapshot.MembershipGeneration,
			NotificationGeneration:  snapshot.NotificationGeneration,
			ManualUnreadGeneration:  snapshot.ManualUnreadGeneration,
			RewindGeneration:        snapshot.RewindGeneration,
		})
	}
	return rows
}

func (store *fakeServiceStore) LoadScopeRosterPage(_ context.Context, _ string, _ string, afterMemberID string, limit int) ([]RosterMember, error) {
	if store.rosterErr != nil {
		return nil, store.rosterErr
	}
	store.rosterLoads++
	if hook := store.afterRosterLoad; hook != nil {
		store.afterRosterLoad = nil
		hook()
	}
	page := make([]RosterMember, 0, limit)
	afterMemberID = strings.TrimSpace(afterMemberID)
	for _, member := range store.roster {
		memberID := strings.TrimSpace(member.MemberID)
		if afterMemberID != "" && memberID <= afterMemberID {
			continue
		}
		page = append(page, member)
		if len(page) >= limit {
			break
		}
	}
	return page, nil
}

func (store *fakeServiceStore) LoadMessageTailPage(_ context.Context, _ string, _ string, afterSeq int64, limit int) ([]MessageStored, error) {
	store.messageTailLoads++
	page := make([]MessageStored, 0, limit)
	for _, item := range store.messageTail {
		if item.Seq <= afterSeq {
			continue
		}
		if item.MessageGeneration == 0 {
			item.MessageGeneration = item.Seq
		}
		page = append(page, item)
		if len(page) >= limit {
			break
		}
	}
	return page, nil
}

func (store *fakeServiceStore) LoadActivityTailPage(_ context.Context, _ string, _ string, afterSeq int64, limit int) ([]NonMessageActivityStored, error) {
	store.activityTailLoads++
	page := make([]NonMessageActivityStored, 0, limit)
	for _, item := range store.activityTail {
		if item.Seq <= afterSeq {
			continue
		}
		if item.ActivityGeneration == 0 {
			item.ActivityGeneration = item.Seq
		}
		page = append(page, item)
		if len(page) >= limit {
			break
		}
	}
	return page, nil
}

func stringSet(values []string) map[string]struct{} {
	values = compactJobStrings(values)
	if len(values) == 0 {
		return nil
	}
	set := make(map[string]struct{}, len(values))
	for _, value := range values {
		set[value] = struct{}{}
	}
	return set
}

func TestServiceFlushDirtySnapshots(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	store := &fakeServiceStore{}
	service := newTestService(t, runtime, store)
	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)
	}

	service.flushOnce(context.Background(), 10)

	if len(store.upserted) != 1 {
		t.Fatalf("len(upserted) = %d, want 1", len(store.upserted))
	}
	if store.upserted[0].UpdatedAt.After(time.Now().Add(time.Second)) {
		t.Fatalf("UpdatedAt = %v, want current time", store.upserted[0].UpdatedAt)
	}
}

func TestServiceFlushFailureRequeuesDirtySnapshots(t *testing.T) {
	runtime := newTestRuntime(t, Config{})
	store := &fakeServiceStore{upsertErr: context.Canceled}
	service := newTestService(t, runtime, store)
	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)
	}

	service.flushOnce(context.Background(), 10)
	rows := runtime.FlushDirtySnapshots(10, time.Unix(2, 0))

	if len(rows) != 1 {
		t.Fatalf("dirty rows after failed flush = %d, want 1", len(rows))
	}
}