package unread

import (
	"fmt"
	"sort"
	"strings"
	"time"
)

// WarmScopeRoster installs the complete write-side roster for one scope.
func (r *Runtime) WarmScopeRoster(workspaceID, scopeID string, scopeType ScopeType, members []RosterMember) error {
	workspaceID = cleanID(workspaceID)
	scopeID = cleanID(scopeID)
	if workspaceID == "" || scopeID == "" {
		return fmt.Errorf("workspace id and scope id are required")
	}

	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return fmt.Errorf("runtime is not initialized")
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()

	if sh.isScopePermanentlyInvalidated(workspaceID, scopeID) {
		return nil
	}
	state := sh.scope(workspaceID, scopeID)
	state.rosterReady = true
	state.scopeType = scopeType
	state.members = make(map[string]RosterMember, len(members))
	state.memberIDs = make([]string, 0, len(members))
	for _, member := range members {
		member, ok := normalizeRosterMember(member)
		if !ok {
			continue
		}
		if _, exists := state.members[member.MemberID]; !exists {
			state.memberIDs = append(state.memberIDs, member.MemberID)
		}
		state.members[member.MemberID] = member
		key := entryKey{workspaceID: workspaceID, memberID: member.MemberID, scopeID: scopeID}
		_, entryAlreadyExists := sh.entries[key]
		sh.updateEntry(key, func(current *entry) {
			applyRosterMember(current, member)
			if scopeType != "" {
				current.state.ScopeType = scopeType
			}
			// WarmScopeRoster is the complete, synchronous write-side install
			// used by an already-live owner. A newly created entry therefore
			// has an intentional zero baseline. The page-wise restart path uses
			// WarmScopeRosterPage and must resolve snapshots before publication.
			if !entryAlreadyExists {
				current.snapshotResolved = true
			}
			if current.state.ServingState == "" || current.state.ServingState == ServingStateHydrating || current.state.ServingState == ServingStateUnavailable {
				current.state.ServingState = ServingStateReady
			}
		}, false)
	}
	return nil
}

// BeginScopeRosterWarm starts a bounded page-wise roster refresh for one scope
// and captures the current roster generation. If a concurrent refresh bumps the
// generation after this call, this warm becomes stale and its pages must not be
// published.
func (r *Runtime) BeginScopeRosterWarm(workspaceID, scopeID string, scopeType ScopeType) error {
	workspaceID = cleanID(workspaceID)
	scopeID = cleanID(scopeID)
	if workspaceID == "" || scopeID == "" {
		return fmt.Errorf("workspace id and scope id are required")
	}

	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return fmt.Errorf("runtime is not initialized")
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()

	if sh.isScopePermanentlyInvalidated(workspaceID, scopeID) {
		return nil
	}
	state := sh.scope(workspaceID, scopeID)
	state.rosterReady = false
	state.rosterWarmGeneration = state.rosterGeneration
	state.scopeType = scopeType
	state.members = make(map[string]RosterMember)
	state.memberIDs = nil
	return nil
}

// WarmScopeRosterPage appends one roster page during a bounded roster refresh.
// If a refresh bumped the roster generation after BeginScopeRosterWarm captured
// it, the page is dropped: it was read before the refresh linearization point
// and must not contribute to an authoritative roster.
func (r *Runtime) WarmScopeRosterPage(workspaceID, scopeID string, scopeType ScopeType, members []RosterMember) error {
	workspaceID = cleanID(workspaceID)
	scopeID = cleanID(scopeID)
	if workspaceID == "" || scopeID == "" {
		return fmt.Errorf("workspace id and scope id are required")
	}

	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return fmt.Errorf("runtime is not initialized")
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()

	if sh.isScopePermanentlyInvalidated(workspaceID, scopeID) {
		return nil
	}
	state := sh.scope(workspaceID, scopeID)
	if state.rosterWarmGeneration != state.rosterGeneration {
		return nil
	}
	state.rosterReady = false
	state.scopeType = scopeType
	if state.members == nil {
		state.members = make(map[string]RosterMember, len(members))
	}
	for _, member := range members {
		member, ok := normalizeRosterMember(member)
		if !ok {
			continue
		}
		if _, exists := state.members[member.MemberID]; !exists {
			state.memberIDs = append(state.memberIDs, member.MemberID)
		}
		state.members[member.MemberID] = member
		key := entryKey{workspaceID: workspaceID, memberID: member.MemberID, scopeID: scopeID}
		if current := sh.entries[key]; current != nil {
			sh.updateEntry(key, func(current *entry) {
				applyRosterMember(current, member)
				if scopeType != "" {
					current.state.ScopeType = scopeType
				}
			}, false)
		}
	}
	return nil
}

// FinishScopeRosterWarm marks a page-wise roster refresh complete. If a refresh
// bumped the roster generation after the warm captured it, the publish is
// refused so a stale hydrate cannot install an authoritative roster.
func (r *Runtime) FinishScopeRosterWarm(workspaceID, scopeID string, scopeType ScopeType) error {
	workspaceID = cleanID(workspaceID)
	scopeID = cleanID(scopeID)
	if workspaceID == "" || scopeID == "" {
		return fmt.Errorf("workspace id and scope id are required")
	}

	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return fmt.Errorf("runtime is not initialized")
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()

	if sh.isScopePermanentlyInvalidated(workspaceID, scopeID) {
		return nil
	}
	state := sh.scope(workspaceID, scopeID)
	if state.rosterWarmGeneration != state.rosterGeneration {
		return nil
	}
	state.scopeType = scopeType
	if state.members == nil {
		state.members = make(map[string]RosterMember)
	}
	state.rosterReady = true
	return nil
}

// ScopeRosterWarmStale reports whether the in-flight roster warm for a scope is
// stale: a refresh bumped the roster generation after BeginScopeRosterWarm
// captured it, so the warm's pages were read before the refresh linearization
// point and must not be published.
func (r *Runtime) ScopeRosterWarmStale(workspaceID, scopeID string) bool {
	workspaceID = cleanID(workspaceID)
	scopeID = cleanID(scopeID)
	if workspaceID == "" || scopeID == "" {
		return false
	}
	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return false
	}
	sh.mu.Lock()
	defer sh.mu.Unlock()
	if sh.isScopePermanentlyInvalidated(workspaceID, scopeID) {
		return false
	}
	state := sh.scope(workspaceID, scopeID)
	return state.rosterWarmGeneration != state.rosterGeneration
}

// finishScopeRosterWarmAfterTaskSnapshots atomically publishes a refreshed
// roster only after every currently queued, active Task attention recipient
// has an in-memory entry. Service resolves each returned page from the durable
// snapshot store (or records an explicit snapshot miss) and retries this
// barrier. Keeping rosterReady false while it does so prevents a concurrent
// live event from creating a zero-baseline entry ahead of snapshot hydrate.
//
// stale reports that a refresh bumped the roster generation after this warm
// captured it at BeginScopeRosterWarm: the loaded pages predate the refresh
// linearization point and must not be published as an authoritative roster. The
// Service must abandon this hydrate without publishing and let Service.finish
// force the next generation from the store.
func (r *Runtime) finishScopeRosterWarmAfterTaskSnapshots(
	workspaceID, scopeID string,
	scopeType ScopeType,
	limit int,
) (missingMemberIDs []string, ready bool, stale bool, err error) {
	workspaceID = cleanID(workspaceID)
	scopeID = cleanID(scopeID)
	if workspaceID == "" || scopeID == "" {
		return nil, false, false, fmt.Errorf("workspace id and scope id are required")
	}
	if limit <= 0 {
		return nil, false, false, fmt.Errorf("snapshot hydrate limit must be positive")
	}

	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return nil, false, false, fmt.Errorf("runtime is not initialized")
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()

	if sh.isScopePermanentlyInvalidated(workspaceID, scopeID) {
		return nil, true, false, nil
	}
	state := sh.scope(workspaceID, scopeID)
	if state.rosterWarmGeneration != state.rosterGeneration {
		return nil, false, true, nil
	}
	if scopeType != "" {
		state.scopeType = scopeType
	}
	if state.members == nil {
		state.members = make(map[string]RosterMember)
	}

	key := scopeKey{workspaceID: workspaceID, scopeID: scopeID}
	isTask := state.scopeType == ScopeTypeTask
	if !isTask {
		for _, event := range sh.pendingMessages[key] {
			if event.ScopeType == ScopeTypeTask {
				isTask = true
				break
			}
		}
	}
	if !isTask {
		for _, event := range sh.pendingActivities[key] {
			if event.ScopeType == ScopeTypeTask {
				isTask = true
				break
			}
		}
	}
	if !isTask {
		state.rosterReady = true
		return nil, true, false, nil
	}

	candidates := make(map[string]struct{})
	addRecipients := func(recipientMemberIDs []string) {
		for _, memberID := range recipientMemberIDs {
			memberID = cleanID(memberID)
			member, active := state.members[memberID]
			if memberID == "" || !active || !eligibleTaskAttentionMember(member) {
				continue
			}
			entry := sh.entries[entryKey{workspaceID: workspaceID, memberID: memberID, scopeID: scopeID}]
			if entry == nil || !entry.snapshotResolved {
				candidates[memberID] = struct{}{}
			}
		}
	}
	for _, event := range sh.pendingMessages[key] {
		addRecipients(event.RecipientMemberIDs)
	}
	for _, event := range sh.pendingActivities[key] {
		addRecipients(event.RecipientMemberIDs)
	}
	if len(candidates) == 0 {
		state.rosterReady = true
		return nil, true, false, nil
	}

	missingMemberIDs = make([]string, 0, len(candidates))
	for memberID := range candidates {
		missingMemberIDs = append(missingMemberIDs, memberID)
	}
	sort.Strings(missingMemberIDs)
	if len(missingMemberIDs) > limit {
		missingMemberIDs = missingMemberIDs[:limit]
	}
	return missingMemberIDs, false, false, nil
}

// ApplyMessage applies a committed human-visible message event to memory.
func (r *Runtime) ApplyMessage(event MessageStored) (ApplyResult, error) {
	if !supportedScopeType(event.ScopeType) {
		return ApplyResult{Ignored: true}, nil
	}
	return r.applyMessage(event, true)
}

// ApplyNonMessageActivity applies a committed high-signal durable activity event.
func (r *Runtime) ApplyNonMessageActivity(event NonMessageActivityStored) (ApplyResult, error) {
	if !supportedScopeType(event.ScopeType) {
		return ApplyResult{Ignored: true}, nil
	}
	return r.applyActivity(event, true)
}

// ApplyRecoveredMessage applies a bounded durable message tail item during
// recovery. It uses the same accounting rules as ApplyMessage but does not
// re-enqueue large warm rosters; the caller already runs under recovery budget.
func (r *Runtime) ApplyRecoveredMessage(event MessageStored) (ApplyResult, error) {
	if !supportedScopeType(event.ScopeType) {
		return ApplyResult{Ignored: true}, nil
	}
	return r.applyMessage(event, false)
}

// ApplyRecoveredActivity applies a bounded durable activity tail item during
// recovery without turning a large roster back into pending hot-path work.
func (r *Runtime) ApplyRecoveredActivity(event NonMessageActivityStored) (ApplyResult, error) {
	if !supportedScopeType(event.ScopeType) {
		return ApplyResult{Ignored: true}, nil
	}
	return r.applyActivity(event, false)
}

// ApplyMessageRewound applies committed rewind removal facts to loaded memory.
func (r *Runtime) ApplyMessageRewound(event MessageRewound) (ApplyResult, error) {
	event.WorkspaceID = cleanID(event.WorkspaceID)
	if event.WorkspaceID == "" {
		return ApplyResult{}, fmt.Errorf("workspace id is required")
	}

	var result ApplyResult
	for _, scope := range event.Scopes {
		scope = normalizeRewoundScope(scope)
		if scope.ScopeID == "" {
			continue
		}
		scopeResult, err := r.applyMessageRewoundScope(event.WorkspaceID, scope, event.RewindGeneration)
		if err != nil {
			return result, err
		}
		result.AccountedMembers += scopeResult.AccountedMembers
		result.Batches += scopeResult.Batches
		result.Ignored = result.Ignored || scopeResult.Ignored
		result.Pending = result.Pending || scopeResult.Pending
	}
	return result, nil
}

func (r *Runtime) applyMessageRewoundScope(workspaceID string, scope RewoundScope, generation int64) (ApplyResult, error) {
	total, pending, ignored, err := r.prepareRewindAccounting(workspaceID, scope)
	if err != nil || pending || ignored {
		if ignored {
			return ApplyResult{Ignored: true}, err
		}
		return ApplyResult{Pending: pending}, err
	}

	result := ApplyResult{Batches: batchesFor(total, r.config.AccountingBatchSize)}
	for start := 0; start < total; start += r.config.AccountingBatchSize {
		members, ok := r.rosterBatch(workspaceID, scope.ScopeID, start, r.config.AccountingBatchSize)
		if !ok {
			if !r.markRewindPending(workspaceID, scope.ScopeID) {
				return ApplyResult{Ignored: true}, nil
			}
			return ApplyResult{Pending: true}, nil
		}
		accounted, deltas, ignored := r.applyMessageRewoundBatch(workspaceID, scope, generation, members)
		if ignored {
			return ApplyResult{Ignored: true}, nil
		}
		result.AccountedMembers += accounted
		result.Deltas = append(result.Deltas, deltas...)
	}
	return result, nil
}

func (r *Runtime) prepareRewindAccounting(workspaceID string, scope RewoundScope) (int, bool, bool, error) {
	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return 0, false, false, fmt.Errorf("runtime is not initialized")
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()

	key := scopeKey{workspaceID: workspaceID, scopeID: scope.ScopeID}
	if sh.isScopePermanentlyInvalidatedKey(key) {
		return 0, false, true, nil
	}
	state := sh.scopes[key]
	if state == nil || !state.rosterReady {
		sh.markScopeEntriesState(workspaceID, scope.ScopeID, ServingStateHydrating, false)
		return 0, true, false, nil
	}
	// Task discussions are append-only durable activity and are deliberately
	// excluded from the rewind target lookup. Reject a defensive direct runtime
	// call too: Task unread uses exact post-snapshot seq tails, while generic
	// suffix rewind accounting cannot update those tails safely.
	if state.scopeType == ScopeTypeTask {
		return 0, false, true, nil
	}
	return len(state.memberIDs), false, false, nil
}

func (r *Runtime) applyMessageRewoundBatch(workspaceID string, scope RewoundScope, generation int64, members []RosterMember) (int, []UnreadDelta, bool) {
	if len(members) == 0 {
		return 0, nil, false
	}
	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return 0, nil, false
	}

	changedEntries := 0
	deltas := make([]UnreadDelta, 0)
	sh.mu.Lock()
	defer sh.mu.Unlock()
	if sh.isScopePermanentlyInvalidated(workspaceID, scope.ScopeID) {
		return 0, nil, true
	}
	for _, member := range members {
		if !eligibleForUnread(member) {
			continue
		}
		key := entryKey{workspaceID: workspaceID, memberID: member.MemberID, scopeID: scope.ScopeID}
		if sh.entries[key] == nil {
			continue
		}
		changed := false
		sh.updateEntry(key, func(current *entry) {
			if !servesExternally(current.state.ServingState) {
				return
			}
			applyRosterMember(current, member)
			accountedSeq := current.latestMessageAccountedSeq
			lastReadSeq := current.lastMessageReadSeq
			effectiveFromSeq := current.effectiveFromSeq
			var decrement int64
			for _, message := range scope.RemovedMessages {
				if message.Seq <= 0 || message.Seq > accountedSeq {
					continue
				}
				if !message.AllowAuthorUnread && effectiveRewoundMessageNotifyPrincipal(message) == member.MemberID {
					continue
				}
				if message.Seq <= lastReadSeq || message.Seq < effectiveFromSeq {
					continue
				}
				decrement++
			}
			if decrement > current.state.MessageUnreadCount {
				decrement = current.state.MessageUnreadCount
			}
			if decrement > 0 {
				current.state.MessageUnreadCount -= decrement
				changed = true
			}
			if current.lastMessageReadSeq > scope.LatestSeq {
				current.lastMessageReadSeq = scope.LatestSeq
				changed = true
			}
			if current.latestMessageAccountedSeq > scope.LatestSeq {
				current.latestMessageAccountedSeq = scope.LatestSeq
				changed = true
			}
			if generation > current.rewindGeneration {
				current.rewindGeneration = generation
				changed = true
			}
		}, false)
		if changed {
			sh.dirty[key] = struct{}{}
			changedEntries++
			if delta, ok := sh.deltaForEntry(key); ok {
				deltas = append(deltas, delta)
			}
		}
	}
	return changedEntries, deltas, false
}

func (r *Runtime) markRewindPending(workspaceID string, scopeID string) bool {
	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return false
	}
	sh.mu.Lock()
	defer sh.mu.Unlock()
	if sh.isScopePermanentlyInvalidated(workspaceID, scopeID) {
		return false
	}
	sh.markScopeEntriesState(workspaceID, scopeID, ServingStateHydrating, false)
	return true
}

func (r *Runtime) applyMessage(event MessageStored, enforceSyncBudget bool) (ApplyResult, error) {
	event.WorkspaceID = cleanID(event.WorkspaceID)
	event.ScopeID = cleanID(event.ScopeID)
	event.AuthorMemberID = cleanID(event.AuthorMemberID)
	event.NotifyPrincipalMemberID = cleanID(event.NotifyPrincipalMemberID)
	event.RecipientMemberIDs = normalizeRecipientMemberIDs(event.RecipientMemberIDs)
	if event.NotifyPrincipalMemberID == "" {
		event.NotifyPrincipalMemberID = event.AuthorMemberID
	}
	if event.WorkspaceID == "" || event.ScopeID == "" {
		return ApplyResult{}, fmt.Errorf("workspace id and scope id are required")
	}

	total, pending, ignored, err := r.prepareMessageAccounting(event, enforceSyncBudget)
	if err != nil || pending || ignored {
		if ignored {
			return ApplyResult{Ignored: true}, err
		}
		return ApplyResult{Pending: pending}, err
	}

	result := ApplyResult{Batches: batchesFor(total, r.config.AccountingBatchSize)}
	for start := 0; start < total; start += r.config.AccountingBatchSize {
		members, ok := r.rosterBatch(event.WorkspaceID, event.ScopeID, start, r.config.AccountingBatchSize)
		if !ok {
			if !r.markMessagePending(event) {
				return ApplyResult{Ignored: true}, nil
			}
			return ApplyResult{Pending: true}, nil
		}
		accounted, deltas, ignored := r.applyMessageBatch(event, members)
		if ignored {
			return ApplyResult{Ignored: true}, nil
		}
		result.AccountedMembers += accounted
		result.Deltas = append(result.Deltas, deltas...)
	}
	if !r.markScopeMessageAccounted(event) {
		return ApplyResult{Ignored: true}, nil
	}
	return result, nil
}

func (r *Runtime) prepareMessageAccounting(event MessageStored, enforceSyncBudget bool) (int, bool, bool, error) {
	sh := r.shardForWorkspace(event.WorkspaceID)
	if sh == nil {
		return 0, false, false, fmt.Errorf("runtime is not initialized")
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()

	key := scopeKey{workspaceID: event.WorkspaceID, scopeID: event.ScopeID}
	if sh.isScopePermanentlyInvalidatedKey(key) {
		return 0, false, true, nil
	}
	scope := sh.scopes[key]
	if scope == nil || !scope.rosterReady {
		sh.coalescePendingMessage(event)
		sh.queuePendingMessage(event, r.config.EventQueueSize)
		sh.markScopeEntriesState(event.WorkspaceID, event.ScopeID, ServingStateHydrating, false)
		return 0, true, false, nil
	}
	if enforceSyncBudget && taskMessageAudienceAlreadyAccounted(scope, event) {
		return 0, false, true, nil
	}
	if enforceSyncBudget && taskAttentionNeedsRosterRecovery(
		sh,
		scope,
		event.WorkspaceID,
		event.ScopeID,
		event.ScopeType,
		event.RecipientMemberIDs,
	) {
		// Task attention freezes its Human audience in the write transaction.
		// A ready owner-local roster can still predate the membership created
		// for a newly assigned recipient, or come from a restart where that
		// recipient's durable unread snapshot has not been loaded yet. Turn only
		// the live path cold so Service reloads the roster and resolves the exact
		// frozen recipients' snapshots before accounting the event. Recovery
		// already follows that barrier; a recipient omitted by the authoritative
		// roster is inactive and is deliberately dropped rather than requeued.
		scope.rosterReady = false
		sh.coalescePendingMessage(event)
		sh.queuePendingMessage(event, r.config.EventQueueSize)
		sh.markScopeEntriesState(event.WorkspaceID, event.ScopeID, ServingStateHydrating, false)
		return 0, true, false, nil
	}
	total := len(scope.memberIDs)
	if enforceSyncBudget && total > r.config.MaxSyncAccountingMembers {
		scope.rosterReady = false
		sh.coalescePendingMessage(event)
		sh.queuePendingMessage(event, r.config.EventQueueSize)
		sh.markScopeEntriesState(event.WorkspaceID, event.ScopeID, ServingStateHydrating, false)
		return 0, true, false, nil
	}
	return total, false, false, nil
}

func (r *Runtime) applyMessageBatch(event MessageStored, members []RosterMember) (int, []UnreadDelta, bool) {
	if len(members) == 0 {
		return 0, nil, false
	}

	sh := r.shardForWorkspace(event.WorkspaceID)
	if sh == nil {
		return 0, nil, false
	}

	accounted := 0
	deltas := make([]UnreadDelta, 0)
	sh.mu.Lock()
	defer sh.mu.Unlock()
	if sh.isScopePermanentlyInvalidated(event.WorkspaceID, event.ScopeID) {
		return 0, nil, true
	}
	for _, member := range members {
		key := entryKey{workspaceID: event.WorkspaceID, memberID: member.MemberID, scopeID: event.ScopeID}
		if sh.skipSuppressedRosterMember(key, member, event.Seq, false) {
			continue
		}
		if !eligibleForMessage(member, event) {
			continue
		}
		applied := false
		sh.updateEntry(key, func(current *entry) {
			applyRosterMember(current, member)
			if event.ScopeType != "" {
				current.state.ScopeType = event.ScopeType
			}
			if event.ScopeType == ScopeTypeTask || current.state.isTaskScope() {
				if event.Seq <= current.lastMessageReadSeq || !current.appendTaskMessageUnreadSeq(event.Seq) {
					return
				}
			} else {
				if event.Seq <= current.latestMessageAccountedSeq {
					return
				}
				current.state.MessageUnreadCount++
			}
			current.state.ServingState = ServingStateReady
			current.latestMessageAccountedSeq = maxInt64(current.latestMessageAccountedSeq, event.Seq)
			current.scopeMessageGeneration = maxInt64(current.scopeMessageGeneration, event.MessageGeneration)
			applied = true
		}, true)
		if applied {
			accounted++
			if delta, ok := sh.deltaForEntry(key); ok {
				deltas = append(deltas, delta)
			}
		}
	}
	return accounted, deltas, false
}

func (r *Runtime) markMessagePending(event MessageStored) bool {
	sh := r.shardForWorkspace(event.WorkspaceID)
	if sh == nil {
		return false
	}
	sh.mu.Lock()
	defer sh.mu.Unlock()
	if sh.isScopePermanentlyInvalidated(event.WorkspaceID, event.ScopeID) {
		return false
	}
	sh.coalescePendingMessage(event)
	sh.queuePendingMessage(event, r.config.EventQueueSize)
	sh.markScopeEntriesState(event.WorkspaceID, event.ScopeID, ServingStateHydrating, false)
	return true
}

func (r *Runtime) applyActivity(event NonMessageActivityStored, enforceSyncBudget bool) (ApplyResult, error) {
	event.WorkspaceID = cleanID(event.WorkspaceID)
	event.ScopeID = cleanID(event.ScopeID)
	event.ActorMemberID = cleanID(event.ActorMemberID)
	event.RecipientMemberIDs = normalizeRecipientMemberIDs(event.RecipientMemberIDs)
	if event.WorkspaceID == "" || event.ScopeID == "" {
		return ApplyResult{}, fmt.Errorf("workspace id and scope id are required")
	}

	total, pending, ignored, err := r.prepareActivityAccounting(event, enforceSyncBudget)
	if err != nil || pending || ignored {
		if ignored {
			return ApplyResult{Ignored: true}, err
		}
		return ApplyResult{Pending: pending}, err
	}

	result := ApplyResult{Batches: batchesFor(total, r.config.AccountingBatchSize)}
	for start := 0; start < total; start += r.config.AccountingBatchSize {
		members, ok := r.rosterBatch(event.WorkspaceID, event.ScopeID, start, r.config.AccountingBatchSize)
		if !ok {
			if !r.markActivityPending(event) {
				return ApplyResult{Ignored: true}, nil
			}
			return ApplyResult{Pending: true}, nil
		}
		accounted, deltas, ignored := r.applyActivityBatch(event, members)
		if ignored {
			return ApplyResult{Ignored: true}, nil
		}
		result.AccountedMembers += accounted
		result.Deltas = append(result.Deltas, deltas...)
	}
	if !r.markScopeActivityAccounted(event) {
		return ApplyResult{Ignored: true}, nil
	}
	return result, nil
}

func (r *Runtime) prepareActivityAccounting(event NonMessageActivityStored, enforceSyncBudget bool) (int, bool, bool, error) {
	sh := r.shardForWorkspace(event.WorkspaceID)
	if sh == nil {
		return 0, false, false, fmt.Errorf("runtime is not initialized")
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()

	key := scopeKey{workspaceID: event.WorkspaceID, scopeID: event.ScopeID}
	if sh.isScopePermanentlyInvalidatedKey(key) {
		return 0, false, true, nil
	}
	scope := sh.scopes[key]
	if scope == nil || !scope.rosterReady {
		sh.coalescePendingActivity(event)
		sh.queuePendingActivity(event, r.config.EventQueueSize)
		sh.markScopeEntriesState(event.WorkspaceID, event.ScopeID, ServingStateHydrating, false)
		return 0, true, false, nil
	}
	if enforceSyncBudget && taskActivityAudienceAlreadyAccounted(scope, event) {
		return 0, false, true, nil
	}
	if enforceSyncBudget && taskAttentionNeedsRosterRecovery(
		sh,
		scope,
		event.WorkspaceID,
		event.ScopeID,
		event.ScopeType,
		event.RecipientMemberIDs,
	) {
		scope.rosterReady = false
		sh.coalescePendingActivity(event)
		sh.queuePendingActivity(event, r.config.EventQueueSize)
		sh.markScopeEntriesState(event.WorkspaceID, event.ScopeID, ServingStateHydrating, false)
		return 0, true, false, nil
	}
	total := len(scope.memberIDs)
	if enforceSyncBudget && total > r.config.MaxSyncAccountingMembers {
		scope.rosterReady = false
		sh.coalescePendingActivity(event)
		sh.queuePendingActivity(event, r.config.EventQueueSize)
		sh.markScopeEntriesState(event.WorkspaceID, event.ScopeID, ServingStateHydrating, false)
		return 0, true, false, nil
	}
	return total, false, false, nil
}

func taskAttentionNeedsRosterRecovery(
	sh *shard,
	scope *scopeState,
	workspaceID string,
	scopeID string,
	eventScopeType ScopeType,
	recipientMemberIDs []string,
) bool {
	if scope == nil ||
		(eventScopeType != ScopeTypeTask && scope.scopeType != ScopeTypeTask) ||
		len(recipientMemberIDs) == 0 {
		return false
	}
	for _, memberID := range recipientMemberIDs {
		memberID = cleanID(memberID)
		member, exists := scope.members[memberID]
		if memberID == "" {
			continue
		}
		if !exists {
			return true
		}
		if !eligibleTaskAttentionMember(member) {
			continue
		}
		if sh == nil {
			return true
		}
		entry := sh.entries[entryKey{
			workspaceID: workspaceID,
			memberID:    memberID,
			scopeID:     scopeID,
		}]
		if entry == nil || !entry.snapshotResolved {
			return true
		}
	}
	return false
}

func taskMessageAudienceAlreadyAccounted(scope *scopeState, event MessageStored) bool {
	if scope == nil ||
		scope.latestMessageAccountedSeq < event.Seq ||
		(event.MessageGeneration > 0 && scope.scopeMessageGeneration < event.MessageGeneration) {
		return false
	}
	return taskAudienceContains(scope.taskMessageAudiences[event.Seq], event.RecipientMemberIDs)
}

func taskActivityAudienceAlreadyAccounted(scope *scopeState, event NonMessageActivityStored) bool {
	if scope == nil ||
		scope.latestActivityAccountedSeq < event.Seq ||
		(event.ActivityGeneration > 0 && scope.scopeActivityGeneration < event.ActivityGeneration) {
		return false
	}
	return taskAudienceContains(scope.taskActivityAudiences[event.Seq], event.RecipientMemberIDs)
}

func taskAudienceContains(accounted map[string]struct{}, recipientMemberIDs []string) bool {
	if len(accounted) == 0 || len(recipientMemberIDs) == 0 {
		return false
	}
	for _, memberID := range recipientMemberIDs {
		if _, exists := accounted[memberID]; !exists {
			return false
		}
	}
	return true
}

func (r *Runtime) applyActivityBatch(event NonMessageActivityStored, members []RosterMember) (int, []UnreadDelta, bool) {
	if len(members) == 0 {
		return 0, nil, false
	}

	sh := r.shardForWorkspace(event.WorkspaceID)
	if sh == nil {
		return 0, nil, false
	}

	accounted := 0
	deltas := make([]UnreadDelta, 0)
	sh.mu.Lock()
	defer sh.mu.Unlock()
	if sh.isScopePermanentlyInvalidated(event.WorkspaceID, event.ScopeID) {
		return 0, nil, true
	}
	for _, member := range members {
		key := entryKey{workspaceID: event.WorkspaceID, memberID: member.MemberID, scopeID: event.ScopeID}
		if sh.skipSuppressedRosterMember(key, member, event.Seq, true) {
			continue
		}
		if !eligibleForActivity(member, event) {
			continue
		}
		applied := false
		sh.updateEntry(key, func(current *entry) {
			applyRosterMember(current, member)
			if event.ScopeType != "" {
				current.state.ScopeType = event.ScopeType
			}
			if event.ScopeType == ScopeTypeTask || current.state.isTaskScope() {
				if event.Seq <= current.lastActivityReadSeq || !current.appendTaskActivityUnreadSeq(event.Seq) {
					return
				}
			} else {
				if event.Seq <= current.latestActivityAccountedSeq {
					return
				}
				current.state.ActivityUnreadCount++
			}
			current.state.ServingState = ServingStateReady
			current.latestActivityAccountedSeq = maxInt64(current.latestActivityAccountedSeq, event.Seq)
			current.scopeActivityGeneration = maxInt64(current.scopeActivityGeneration, event.ActivityGeneration)
			applied = true
		}, true)
		if applied {
			accounted++
			if delta, ok := sh.deltaForEntry(key); ok {
				deltas = append(deltas, delta)
			}
		}
	}
	return accounted, deltas, false
}

func (r *Runtime) markActivityPending(event NonMessageActivityStored) bool {
	sh := r.shardForWorkspace(event.WorkspaceID)
	if sh == nil {
		return false
	}
	sh.mu.Lock()
	defer sh.mu.Unlock()
	if sh.isScopePermanentlyInvalidated(event.WorkspaceID, event.ScopeID) {
		return false
	}
	sh.coalescePendingActivity(event)
	sh.queuePendingActivity(event, r.config.EventQueueSize)
	sh.markScopeEntriesState(event.WorkspaceID, event.ScopeID, ServingStateHydrating, false)
	return true
}

// ScopeRosterReady reports whether a scope has a complete in-memory roster.
func (r *Runtime) ScopeRosterReady(workspaceID, scopeID string) bool {
	ready, _ := r.scopeRosterStatus(workspaceID, scopeID)
	return ready
}

func (r *Runtime) scopeRosterStatus(workspaceID, scopeID string) (ready bool, permanentlyInvalidated bool) {
	workspaceID = cleanID(workspaceID)
	scopeID = cleanID(scopeID)
	if workspaceID == "" || scopeID == "" {
		return false, false
	}
	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return false, false
	}
	sh.mu.Lock()
	defer sh.mu.Unlock()
	key := scopeKey{workspaceID: workspaceID, scopeID: scopeID}
	if sh.isScopePermanentlyInvalidatedKey(key) {
		return false, true
	}
	scope := sh.scopes[key]
	return scope != nil && scope.rosterReady, false
}

func (r *Runtime) scopePermanentlyInvalidated(workspaceID, scopeID string) bool {
	_, permanentlyInvalidated := r.scopeRosterStatus(workspaceID, scopeID)
	return permanentlyInvalidated
}

// ApplyPendingForScope applies process-local events accumulated while a scope
// roster was cold. It never reads durable tails; crash-window recovery remains
// snapshot-only in V1.
func (r *Runtime) ApplyPendingForScope(workspaceID, scopeID string) (ApplyResult, error) {
	workspaceID = cleanID(workspaceID)
	scopeID = cleanID(scopeID)
	if workspaceID == "" || scopeID == "" {
		return ApplyResult{}, fmt.Errorf("workspace id and scope id are required")
	}
	ready, permanentlyInvalidated := r.scopeRosterStatus(workspaceID, scopeID)
	if permanentlyInvalidated {
		return ApplyResult{Ignored: true}, nil
	}
	if !ready {
		return ApplyResult{Pending: true}, nil
	}

	messages, activities, permanentlyInvalidated := r.drainPendingScopeEvents(workspaceID, scopeID)
	if permanentlyInvalidated {
		return ApplyResult{Ignored: true}, nil
	}
	sort.Slice(messages, func(i, j int) bool {
		return messages[i].Seq < messages[j].Seq
	})
	sort.Slice(activities, func(i, j int) bool {
		return activities[i].Seq < activities[j].Seq
	})

	var result ApplyResult
	for _, event := range messages {
		applied, err := r.ApplyRecoveredMessage(event)
		if err != nil {
			return result, err
		}
		if applied.Ignored {
			return ApplyResult{Ignored: true}, nil
		}
		result.AccountedMembers += applied.AccountedMembers
		result.Batches += applied.Batches
		result.Pending = result.Pending || applied.Pending
		result.Deltas = append(result.Deltas, applied.Deltas...)
	}
	for _, event := range activities {
		applied, err := r.ApplyRecoveredActivity(event)
		if err != nil {
			return result, err
		}
		if applied.Ignored {
			return ApplyResult{Ignored: true}, nil
		}
		result.AccountedMembers += applied.AccountedMembers
		result.Batches += applied.Batches
		result.Pending = result.Pending || applied.Pending
		result.Deltas = append(result.Deltas, applied.Deltas...)
	}
	if r.scopePermanentlyInvalidated(workspaceID, scopeID) {
		return ApplyResult{Ignored: true}, nil
	}
	return result, nil
}

func (r *Runtime) markScopeMessageAccounted(event MessageStored) bool {
	sh := r.shardForWorkspace(event.WorkspaceID)
	if sh == nil {
		return false
	}
	sh.mu.Lock()
	defer sh.mu.Unlock()

	if sh.isScopePermanentlyInvalidated(event.WorkspaceID, event.ScopeID) {
		return false
	}
	scope := sh.scope(event.WorkspaceID, event.ScopeID)
	isTask := event.ScopeType == ScopeTypeTask || scope.scopeType == ScopeTypeTask
	if event.ScopeType != "" {
		scope.scopeType = event.ScopeType
	}
	scope.latestMessageAccountedSeq = maxInt64(scope.latestMessageAccountedSeq, event.Seq)
	scope.scopeMessageGeneration = maxInt64(scope.scopeMessageGeneration, event.MessageGeneration)
	if isTask {
		scope.taskMessageAudiences = recordTaskAccountedAudience(
			scope.taskMessageAudiences,
			event.Seq,
			event.RecipientMemberIDs,
		)
	}
	return true
}

func (r *Runtime) markScopeActivityAccounted(event NonMessageActivityStored) bool {
	sh := r.shardForWorkspace(event.WorkspaceID)
	if sh == nil {
		return false
	}
	sh.mu.Lock()
	defer sh.mu.Unlock()

	if sh.isScopePermanentlyInvalidated(event.WorkspaceID, event.ScopeID) {
		return false
	}
	scope := sh.scope(event.WorkspaceID, event.ScopeID)
	isTask := event.ScopeType == ScopeTypeTask || scope.scopeType == ScopeTypeTask
	if event.ScopeType != "" {
		scope.scopeType = event.ScopeType
	}
	scope.latestActivityAccountedSeq = maxInt64(scope.latestActivityAccountedSeq, event.Seq)
	scope.scopeActivityGeneration = maxInt64(scope.scopeActivityGeneration, event.ActivityGeneration)
	if isTask {
		scope.taskActivityAudiences = recordTaskAccountedAudience(
			scope.taskActivityAudiences,
			event.Seq,
			event.RecipientMemberIDs,
		)
	}
	return true
}

func recordTaskAccountedAudience(
	audiences map[int64]map[string]struct{},
	seq int64,
	recipientMemberIDs []string,
) map[int64]map[string]struct{} {
	if seq <= 0 || len(recipientMemberIDs) == 0 {
		return audiences
	}
	if audiences == nil {
		audiences = make(map[int64]map[string]struct{})
	}
	recipients := audiences[seq]
	if recipients == nil {
		recipients = make(map[string]struct{}, len(recipientMemberIDs))
		audiences[seq] = recipients
	}
	for _, memberID := range recipientMemberIDs {
		if memberID = cleanID(memberID); memberID != "" {
			recipients[memberID] = struct{}{}
		}
	}
	if len(audiences) <= 2*maxTrackedTaskUnreadSeqs {
		return audiences
	}
	seqs := make([]int64, 0, len(audiences))
	for trackedSeq := range audiences {
		seqs = append(seqs, trackedSeq)
	}
	sort.Slice(seqs, func(i, j int) bool { return seqs[i] < seqs[j] })
	for _, trackedSeq := range seqs[:len(seqs)-maxTrackedTaskUnreadSeqs] {
		delete(audiences, trackedSeq)
	}
	return audiences
}

// ApplyReadCursor clears message unread state through the committed cursor.
// Events may reach this owner before the transaction callback that carries an
// older read cursor. Retaining their exact seqs prevents that delayed callback
// from erasing causally newer unread work.
func (r *Runtime) ApplyReadCursor(event ReadCursorAdvanced) error {
	event.WorkspaceID = cleanID(event.WorkspaceID)
	event.MemberID = cleanID(event.MemberID)
	event.ScopeID = cleanID(event.ScopeID)
	if event.WorkspaceID == "" || event.MemberID == "" || event.ScopeID == "" {
		return fmt.Errorf("workspace id, member id, and scope id are required")
	}

	sh := r.shardForWorkspace(event.WorkspaceID)
	if sh == nil {
		return fmt.Errorf("runtime is not initialized")
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()

	key := entryKey{workspaceID: event.WorkspaceID, memberID: event.MemberID, scopeID: event.ScopeID}
	sh.updateEntry(key, func(current *entry) {
		readSeq := maxInt64(current.lastMessageReadSeq, event.LastReadSeq)
		if current.state.isTaskScope() {
			current.clearMessageUnreadThrough(readSeq)
		} else {
			current.state.MessageUnreadCount = 0
		}
		if !event.ManualUnreadGenerationKnown || event.ManualUnreadGeneration >= current.manualUnreadGeneration {
			current.state.ManualUnread = false
			current.manualUnreadMutationSet = true
			if event.ManualUnreadGenerationKnown {
				current.manualUnreadGeneration = event.ManualUnreadGeneration
			}
		}
		current.state.ServingState = ServingStateReady
		current.lastMessageReadSeq = readSeq
		current.latestMessageAccountedSeq = maxInt64(current.latestMessageAccountedSeq, event.LastReadSeq)
		current.membershipGeneration = maxInt64(current.membershipGeneration, event.MembershipGeneration)
	}, true)
	return nil
}

// ApplyActivityReadCursor clears activity unread state through the committed
// cursor while retaining already-accounted activity after that cursor.
func (r *Runtime) ApplyActivityReadCursor(event ActivityReadCursorAdvanced) error {
	event.WorkspaceID = cleanID(event.WorkspaceID)
	event.MemberID = cleanID(event.MemberID)
	event.ScopeID = cleanID(event.ScopeID)
	if event.WorkspaceID == "" || event.MemberID == "" || event.ScopeID == "" {
		return fmt.Errorf("workspace id, member id, and scope id are required")
	}

	sh := r.shardForWorkspace(event.WorkspaceID)
	if sh == nil {
		return fmt.Errorf("runtime is not initialized")
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()

	key := entryKey{workspaceID: event.WorkspaceID, memberID: event.MemberID, scopeID: event.ScopeID}
	sh.updateEntry(key, func(current *entry) {
		readSeq := maxInt64(current.lastActivityReadSeq, event.LastReadSeq)
		if current.state.isTaskScope() {
			current.clearActivityUnreadThrough(readSeq)
		} else {
			current.state.ActivityUnreadCount = 0
		}
		current.state.ServingState = ServingStateReady
		current.lastActivityReadSeq = readSeq
		current.latestActivityAccountedSeq = maxInt64(current.latestActivityAccountedSeq, event.LastReadSeq)
		current.membershipGeneration = maxInt64(current.membershipGeneration, event.MembershipGeneration)
	}, true)
	return nil
}

// ApplyManualUnread loads a committed manual unread marker into memory.
func (r *Runtime) ApplyManualUnread(event ManualUnreadChanged) error {
	event.WorkspaceID = cleanID(event.WorkspaceID)
	event.MemberID = cleanID(event.MemberID)
	event.ScopeID = cleanID(event.ScopeID)
	if event.WorkspaceID == "" || event.MemberID == "" || event.ScopeID == "" {
		return fmt.Errorf("workspace id, member id, and scope id are required")
	}

	sh := r.shardForWorkspace(event.WorkspaceID)
	if sh == nil {
		return fmt.Errorf("runtime is not initialized")
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()

	key := entryKey{workspaceID: event.WorkspaceID, memberID: event.MemberID, scopeID: event.ScopeID}
	sh.updateEntry(key, func(current *entry) {
		if event.ManualUnreadGeneration > 0 && event.ManualUnreadGeneration < current.manualUnreadGeneration {
			return
		}
		current.state.ManualUnread = event.ManualUnread
		current.state.ServingState = ServingStateReady
		current.manualUnreadGeneration = maxInt64(current.manualUnreadGeneration, event.ManualUnreadGeneration)
		current.manualUnreadMutationSet = true
	}, true)
	return nil
}

// ApplyScopeInvalidated suppresses loaded state affected by a committed
// mutation whose exact unread delta is not available at the API boundary.
func (r *Runtime) ApplyScopeInvalidated(event ScopeInvalidated) error {
	_, err := r.ApplyScopeInvalidatedWithDeltas(event)
	return err
}

// ApplyScopeInvalidatedWithDeltas suppresses loaded state and returns
// realtime-safe zero deltas for entries that can identify a user.
func (r *Runtime) ApplyScopeInvalidatedWithDeltas(event ScopeInvalidated) ([]UnreadDelta, error) {
	event.WorkspaceID = cleanID(event.WorkspaceID)
	event.ScopeID = cleanID(event.ScopeID)
	if event.WorkspaceID == "" || event.ScopeID == "" {
		return nil, fmt.Errorf("workspace id and scope id are required")
	}

	sh := r.shardForWorkspace(event.WorkspaceID)
	if sh == nil {
		return nil, fmt.Errorf("runtime is not initialized")
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()

	var deltas []UnreadDelta
	if event.Permanent {
		scope := scopeKey{workspaceID: event.WorkspaceID, scopeID: event.ScopeID}
		sh.permanentScopeTombstones[scope] = struct{}{}
		for key := range sh.entries {
			if key.workspaceID != event.WorkspaceID || key.scopeID != event.ScopeID {
				continue
			}
			if delta, ok := sh.suppressEntryWithDeltaUnchecked(key, ServingStateStale); ok {
				deltas = append(deltas, delta)
			}
			delete(sh.entries, key)
			delete(sh.dirty, key)
			delete(sh.suppressedRoster, key)
		}
		// Defensive cleanup covers markers whose entry was already absent.
		for key := range sh.dirty {
			if key.workspaceID == event.WorkspaceID && key.scopeID == event.ScopeID {
				delete(sh.dirty, key)
			}
		}
		for key := range sh.suppressedRoster {
			if key.workspaceID == event.WorkspaceID && key.scopeID == event.ScopeID {
				delete(sh.suppressedRoster, key)
			}
		}
		delete(sh.scopes, scope)
		delete(sh.pending, scope)
		delete(sh.pendingMessages, scope)
		delete(sh.pendingActivities, scope)
		delete(sh.pendingOverflow, scope)
		return deltas, nil
	}
	if len(event.MemberIDs) == 0 {
		for key := range sh.entries {
			if key.workspaceID != event.WorkspaceID || key.scopeID != event.ScopeID {
				continue
			}
			if sh.applyTaskHideFrontiers(key, event) {
				if delta, ok := sh.deltaForEntry(key); ok {
					deltas = append(deltas, delta)
				}
				continue
			}
			sh.rememberSuppressedRosterMember(key, event.Frontier, event.ActivityFrontier, event.FrontiersKnown)
			if delta, ok := sh.suppressEntryWithDelta(key, ServingStateStale); ok {
				deltas = append(deltas, delta)
			}
		}
		return deltas, nil
	}
	for _, memberID := range event.MemberIDs {
		memberID = cleanID(memberID)
		if memberID == "" {
			continue
		}
		key := entryKey{workspaceID: event.WorkspaceID, memberID: memberID, scopeID: event.ScopeID}
		if sh.applyTaskHideFrontiers(key, event) {
			if delta, ok := sh.deltaForEntry(key); ok {
				deltas = append(deltas, delta)
			}
			continue
		}
		sh.rememberSuppressedRosterMember(key, event.Frontier, event.ActivityFrontier, event.FrontiersKnown)
		if delta, ok := sh.suppressEntryWithDelta(key, ServingStateStale); ok {
			deltas = append(deltas, delta)
		}
	}
	return deltas, nil
}

// ApplyNotificationMuted applies a committed notification mute for one member
// of a scope. Mute clears the member's currently loaded unread (message,
// activity, and manual), suppresses the entries, and records the pre-mute
// roster baseline so messages posted while the scope stays muted are not
// counted. The cleared entries are marked dirty so the zero state is persisted
// durably; a later re-activation must refresh the roster rather than revive the
// pre-mute unread.
func (r *Runtime) ApplyNotificationMuted(event ScopeInvalidated) ([]UnreadDelta, error) {
	event.WorkspaceID = cleanID(event.WorkspaceID)
	event.ScopeID = cleanID(event.ScopeID)
	if event.WorkspaceID == "" || event.ScopeID == "" {
		return nil, fmt.Errorf("workspace id and scope id are required")
	}
	sh := r.shardForWorkspace(event.WorkspaceID)
	if sh == nil {
		return nil, fmt.Errorf("runtime is not initialized")
	}
	sh.mu.Lock()
	defer sh.mu.Unlock()
	if sh.isScopePermanentlyInvalidated(event.WorkspaceID, event.ScopeID) {
		return nil, nil
	}
	var deltas []UnreadDelta
	if len(event.MemberIDs) == 0 {
		for key := range sh.entries {
			if key.workspaceID != event.WorkspaceID || key.scopeID != event.ScopeID {
				continue
			}
			if delta, ok := sh.clearEntryForNotificationMute(key); ok {
				deltas = append(deltas, delta)
			}
		}
		return deltas, nil
	}
	for _, memberID := range event.MemberIDs {
		memberID = cleanID(memberID)
		if memberID == "" {
			continue
		}
		key := entryKey{workspaceID: event.WorkspaceID, memberID: memberID, scopeID: event.ScopeID}
		if delta, ok := sh.clearEntryForNotificationMute(key); ok {
			deltas = append(deltas, delta)
		}
	}
	return deltas, nil
}

// InvalidateScopeRosterForRefresh clears the notification-suppression roster
// markers for one or more members and forces the scope roster out of the ready
// state so the Service reloads it from the store. It is the re-activation path
// for a notification self-state change away from mute: the reloaded roster
// carries the newly persisted active/quiet state that replaces the prior mute
// baseline, restoring unread eligibility for future messages without replaying
// messages posted while muted or reviving the pre-mute unread.
//
// The roster generation is bumped atomically here. Any page-wise warm that
// captured an older generation becomes stale and is refused at its publish
// fence, so a hydrate that read pre-persist mute rows cannot install an
// authoritative roster after this linearization point.
func (r *Runtime) InvalidateScopeRosterForRefresh(workspaceID, scopeID string, scopeType ScopeType, memberIDs []string) error {
	workspaceID = cleanID(workspaceID)
	scopeID = cleanID(scopeID)
	if workspaceID == "" || scopeID == "" {
		return fmt.Errorf("workspace id and scope id are required")
	}
	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return fmt.Errorf("runtime is not initialized")
	}
	sh.mu.Lock()
	defer sh.mu.Unlock()
	if sh.isScopePermanentlyInvalidated(workspaceID, scopeID) {
		return nil
	}
	if len(memberIDs) > 0 {
		for _, memberID := range memberIDs {
			memberID = cleanID(memberID)
			if memberID == "" {
				continue
			}
			delete(sh.suppressedRoster, entryKey{workspaceID: workspaceID, memberID: memberID, scopeID: scopeID})
		}
	} else {
		for key := range sh.suppressedRoster {
			if key.workspaceID == workspaceID && key.scopeID == scopeID {
				delete(sh.suppressedRoster, key)
			}
		}
	}
	state := sh.scope(workspaceID, scopeID)
	state.rosterGeneration++
	state.rosterReady = false
	if scopeType != "" {
		state.scopeType = scopeType
	}
	return nil
}

// HydrateSnapshot loads a snapshot row into memory as a best-effort hint.
func (r *Runtime) HydrateSnapshot(row Snapshot) error {
	return r.hydrateSnapshot(row, false)
}

// HydrateSnapshotIfAbsent loads a snapshot row only when the entry has not been
// created by a newer runtime update since hydrate was requested.
func (r *Runtime) HydrateSnapshotIfAbsent(row Snapshot) error {
	return r.hydrateSnapshot(row, true)
}

func (r *Runtime) hydrateSnapshot(row Snapshot, absentOnly bool) error {
	row.WorkspaceID = cleanID(row.WorkspaceID)
	row.MemberID = cleanID(row.MemberID)
	row.ScopeID = cleanID(row.ScopeID)
	if row.WorkspaceID == "" || row.MemberID == "" || row.ScopeID == "" {
		return fmt.Errorf("workspace id, member id, and scope id are required")
	}
	if row.ServingState == "" {
		row.ServingState = ServingStateSnapshotServing
	}

	sh := r.shardForWorkspace(row.WorkspaceID)
	if sh == nil {
		return fmt.Errorf("runtime is not initialized")
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()

	key := entryKey{workspaceID: row.WorkspaceID, memberID: row.MemberID, scopeID: row.ScopeID}
	if sh.isScopePermanentlyInvalidated(row.WorkspaceID, row.ScopeID) {
		delete(sh.dirty, key)
		return nil
	}
	existing := sh.entries[key]
	mergeUnresolvedTaskControls := false
	if absentOnly && existing != nil {
		isTask := row.ScopeType == ScopeTypeTask || existing.state.isTaskScope()
		if !isTask {
			if scope := sh.scopes[scopeKey{workspaceID: row.WorkspaceID, scopeID: row.ScopeID}]; scope != nil {
				isTask = scope.scopeType == ScopeTypeTask
			}
		}
		if !isTask || existing.snapshotResolved {
			return nil
		}
		mergeUnresolvedTaskControls = true
	}
	sh.updateEntry(key, func(current *entry) {
		if !mergeUnresolvedTaskControls {
			installSnapshotRow(current, row)
			current.snapshotResolved = true
			current.manualUnreadMutationSet = false
			return
		}

		// Only control callbacks can create a Task entry while it is unresolved:
		// targeted message/activity accounting keeps the roster cold. Install
		// the durable opaque baseline, then overlay those later controls. A
		// cursor inside an opaque baseline cannot be split safely and is kept
		// conservatively; a cursor covering its latest seq clears it exactly.
		previousUserID := current.userID
		previousLastMessageReadSeq := current.lastMessageReadSeq
		previousLastActivityReadSeq := current.lastActivityReadSeq
		previousEffectiveFromSeq := current.effectiveFromSeq
		previousLatestMessageAccountedSeq := current.latestMessageAccountedSeq
		previousLatestActivityAccountedSeq := current.latestActivityAccountedSeq
		previousNotificationState := current.notificationState
		previousMembershipVisibility := current.membershipVisibility
		previousScopeMessageGeneration := current.scopeMessageGeneration
		previousScopeActivityGeneration := current.scopeActivityGeneration
		previousMembershipGeneration := current.membershipGeneration
		previousNotificationGeneration := current.notificationGeneration
		previousManualUnread := current.state.ManualUnread
		previousManualUnreadGeneration := current.manualUnreadGeneration
		previousManualUnreadMutationSet := current.manualUnreadMutationSet
		previousRewindGeneration := current.rewindGeneration

		installSnapshotRow(current, row)
		if previousUserID != "" {
			current.userID = previousUserID
		}
		current.lastMessageReadSeq = maxInt64(current.lastMessageReadSeq, previousLastMessageReadSeq)
		current.lastActivityReadSeq = maxInt64(current.lastActivityReadSeq, previousLastActivityReadSeq)
		current.effectiveFromSeq = maxInt64(current.effectiveFromSeq, previousEffectiveFromSeq)
		current.latestMessageAccountedSeq = maxInt64(current.latestMessageAccountedSeq, previousLatestMessageAccountedSeq)
		current.latestActivityAccountedSeq = maxInt64(current.latestActivityAccountedSeq, previousLatestActivityAccountedSeq)
		current.scopeMessageGeneration = maxInt64(current.scopeMessageGeneration, previousScopeMessageGeneration)
		current.scopeActivityGeneration = maxInt64(current.scopeActivityGeneration, previousScopeActivityGeneration)
		if previousMembershipGeneration >= current.membershipGeneration && previousMembershipVisibility != "" {
			current.membershipVisibility = previousMembershipVisibility
		}
		current.membershipGeneration = maxInt64(current.membershipGeneration, previousMembershipGeneration)
		if previousNotificationGeneration >= current.notificationGeneration && previousNotificationState != "" {
			current.notificationState = previousNotificationState
		}
		current.notificationGeneration = maxInt64(current.notificationGeneration, previousNotificationGeneration)
		if previousManualUnreadMutationSet &&
			(previousManualUnreadGeneration == 0 ||
				previousManualUnreadGeneration >= current.manualUnreadGeneration) {
			current.state.ManualUnread = previousManualUnread
		}
		current.manualUnreadGeneration = maxInt64(current.manualUnreadGeneration, previousManualUnreadGeneration)
		current.manualUnreadMutationSet = previousManualUnreadMutationSet
		current.rewindGeneration = maxInt64(current.rewindGeneration, previousRewindGeneration)
		current.clearMessageUnreadThrough(current.lastMessageReadSeq)
		current.clearActivityUnreadThrough(current.lastActivityReadSeq)
		current.snapshotResolved = true
	}, false)
	return nil
}

func installSnapshotRow(current *entry, row Snapshot) {
	scopeType := row.ScopeType
	if scopeType == "" {
		scopeType = scopeTypeForSnapshotScopeID(row.ScopeID)
	}
	current.state = ScopeUnreadState{
		WorkspaceID:         row.WorkspaceID,
		MemberID:            row.MemberID,
		ScopeID:             row.ScopeID,
		ScopeType:           scopeType,
		MessageUnreadCount:  maxInt64(row.MessageUnreadCount, 0),
		ActivityUnreadCount: maxInt64(row.ActivityUnreadCount, 0),
		ManualUnread:        row.ManualUnread,
		ServingState:        row.ServingState,
	}
	if current.state.isTaskScope() {
		current.messageUnreadBaselineCount = current.state.MessageUnreadCount
		current.messageUnreadBaselineSeq = row.LatestMessageAccountedSeq
		current.messageUnreadSeqs = nil
		current.activityUnreadBaselineCount = current.state.ActivityUnreadCount
		current.activityUnreadBaselineSeq = row.LatestActivityAccountedSeq
		current.activityUnreadSeqs = nil
	} else {
		current.messageUnreadBaselineCount = 0
		current.messageUnreadBaselineSeq = 0
		current.messageUnreadSeqs = nil
		current.activityUnreadBaselineCount = 0
		current.activityUnreadBaselineSeq = 0
		current.activityUnreadSeqs = nil
	}
	current.lastMessageReadSeq = row.LastMessageReadSeq
	current.lastActivityReadSeq = row.LastActivityReadSeq
	current.effectiveFromSeq = row.EffectiveFromSeq
	current.latestMessageAccountedSeq = row.LatestMessageAccountedSeq
	current.latestActivityAccountedSeq = row.LatestActivityAccountedSeq
	current.notificationState = cleanState(row.NotificationState)
	current.membershipVisibility = cleanState(row.MembershipVisibility)
	current.scopeMessageGeneration = row.ScopeMessageGeneration
	current.scopeActivityGeneration = row.ScopeActivityGeneration
	current.membershipGeneration = row.MembershipGeneration
	current.notificationGeneration = row.NotificationGeneration
	current.manualUnreadGeneration = row.ManualUnreadGeneration
	current.rewindGeneration = row.RewindGeneration
}

// ScopeStates returns memory-backed states for the requested scopes. Not-ready
// entries are suppressed from the result.
func (r *Runtime) ScopeStates(workspaceID, memberID string, scopeIDs []string) map[string]ScopeUnreadState {
	workspaceID = cleanID(workspaceID)
	memberID = cleanID(memberID)
	if workspaceID == "" || memberID == "" {
		return map[string]ScopeUnreadState{}
	}

	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return map[string]ScopeUnreadState{}
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()

	results := make(map[string]ScopeUnreadState)
	if len(scopeIDs) == 0 {
		for key, current := range sh.entries {
			if key.workspaceID != workspaceID || key.memberID != memberID ||
				sh.isScopePermanentlyInvalidated(key.workspaceID, key.scopeID) ||
				(current.state.isTaskScope() && !current.snapshotResolved) ||
				!servesExternally(current.state.ServingState) {
				continue
			}
			results[key.scopeID] = current.state
		}
		return results
	}
	for _, scopeID := range scopeIDs {
		scopeID = cleanID(scopeID)
		if scopeID == "" {
			continue
		}
		if sh.isScopePermanentlyInvalidated(workspaceID, scopeID) {
			continue
		}
		current := sh.entries[entryKey{workspaceID: workspaceID, memberID: memberID, scopeID: scopeID}]
		if current == nil ||
			(current.state.isTaskScope() && !current.snapshotResolved) ||
			!servesExternally(current.state.ServingState) {
			continue
		}
		results[scopeID] = current.state
	}
	return results
}

// CurrentUnreadDelta atomically snapshots one scope and its workspace
// aggregate. Live emitters must prefer this over composing ScopeStates and
// Workspace* reads across separate shard-lock acquisitions.
func (r *Runtime) CurrentUnreadDelta(workspaceID, userID, memberID, scopeID string, fallbackScopeType ScopeType) (UnreadDelta, bool) {
	workspaceID = cleanID(workspaceID)
	userID = cleanID(userID)
	memberID = cleanID(memberID)
	scopeID = cleanID(scopeID)
	if workspaceID == "" || memberID == "" || scopeID == "" {
		return UnreadDelta{}, false
	}
	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return UnreadDelta{}, false
	}
	sh.mu.Lock()
	defer sh.mu.Unlock()
	if sh.isScopePermanentlyInvalidated(workspaceID, scopeID) {
		return UnreadDelta{}, false
	}
	key := entryKey{workspaceID: workspaceID, memberID: memberID, scopeID: scopeID}
	current := sh.entries[key]
	if current != nil && current.state.isTaskScope() && !current.snapshotResolved {
		return UnreadDelta{}, false
	}
	delta, ok := sh.deltaForEntry(key)
	if !ok || !servesExternally(delta.ServingState) {
		return UnreadDelta{}, false
	}
	delta.UserID = userID
	if delta.ScopeType == "" {
		delta.ScopeType = fallbackScopeType
	}
	return delta, true
}

// ScopeUnreadCounts returns compatibility count values for ready or
// snapshot-serving entries. Hydrating, stale, and unavailable entries are
// suppressed instead of triggering a database fallback.
func (r *Runtime) ScopeUnreadCounts(workspaceID, memberID string, scopeIDs []string) map[string]int64 {
	states := r.ScopeStates(workspaceID, memberID, scopeIDs)
	counts := make(map[string]int64, len(states))
	for scopeID, state := range states {
		if count := state.Count(); count > 0 {
			counts[scopeID] = count
		}
	}
	return counts
}

func (r *Runtime) missingScopeIDs(workspaceID, memberID string, scopeIDs []string) []string {
	workspaceID = cleanID(workspaceID)
	memberID = cleanID(memberID)
	if r == nil || workspaceID == "" || memberID == "" || len(scopeIDs) == 0 {
		return nil
	}

	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return nil
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()

	seen := make(map[string]struct{}, len(scopeIDs))
	missing := make([]string, 0)
	for _, scopeID := range scopeIDs {
		scopeID = cleanID(scopeID)
		if scopeID == "" {
			continue
		}
		if _, ok := seen[scopeID]; ok {
			continue
		}
		seen[scopeID] = struct{}{}
		if sh.isScopePermanentlyInvalidated(workspaceID, scopeID) {
			continue
		}
		current := sh.entries[entryKey{workspaceID: workspaceID, memberID: memberID, scopeID: scopeID}]
		if current == nil || (current.state.isTaskScope() && !current.snapshotResolved) {
			missing = append(missing, scopeID)
		}
	}
	return missing
}

func (r *Runtime) taskSnapshotNeedsHydrate(workspaceID, memberID, scopeID string) bool {
	workspaceID = cleanID(workspaceID)
	memberID = cleanID(memberID)
	scopeID = cleanID(scopeID)
	if r == nil || workspaceID == "" || memberID == "" || scopeID == "" {
		return false
	}
	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return false
	}
	sh.mu.Lock()
	defer sh.mu.Unlock()
	if sh.isScopePermanentlyInvalidated(workspaceID, scopeID) {
		return false
	}
	current := sh.entries[entryKey{workspaceID: workspaceID, memberID: memberID, scopeID: scopeID}]
	return current != nil && current.state.isTaskScope() && !current.snapshotResolved
}

// taskSnapshotRowsForMembers snapshots the exact loaded Task recipients used by
// one durable attention delivery. It is intentionally non-destructive: the
// ordinary background flush still owns dirty-marker cleanup, while the
// delivery barrier only needs a stable value to persist before completing the
// outbox lease.
func (r *Runtime) taskSnapshotRowsForMembers(
	workspaceID string,
	scopeID string,
	memberIDs []string,
	now time.Time,
) ([]Snapshot, error) {
	workspaceID = cleanID(workspaceID)
	scopeID = cleanID(scopeID)
	memberIDs = normalizeRecipientMemberIDs(memberIDs)
	if r == nil || workspaceID == "" || scopeID == "" {
		return nil, fmt.Errorf("workspace id and scope id are required")
	}
	if len(memberIDs) == 0 {
		return nil, nil
	}

	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return nil, fmt.Errorf("runtime is not initialized")
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()
	if sh.isScopePermanentlyInvalidated(workspaceID, scopeID) {
		return nil, nil
	}

	rows := make([]Snapshot, 0, len(memberIDs))
	for _, memberID := range memberIDs {
		key := entryKey{workspaceID: workspaceID, memberID: memberID, scopeID: scopeID}
		current := sh.entries[key]
		if current == nil {
			// Frozen recipients that became inactive are deliberately omitted
			// by Task accounting and have no new unread state to persist.
			continue
		}
		if !current.state.isTaskScope() {
			return nil, fmt.Errorf("scope %q is not a Task unread entry", scopeID)
		}
		if !current.snapshotResolved {
			return nil, fmt.Errorf("Task unread snapshot for member %q is unresolved", memberID)
		}
		rows = append(rows, current.snapshotRow(now))
	}
	return rows, nil
}

// WorkspaceHasUnread returns the loaded in-memory workspace red-dot aggregate.
func (r *Runtime) WorkspaceHasUnread(workspaceID, memberID string) bool {
	workspaceID = cleanID(workspaceID)
	memberID = cleanID(memberID)
	if workspaceID == "" || memberID == "" {
		return false
	}

	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return false
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()
	return sh.aggregates[memberKey{workspaceID: workspaceID, memberID: memberID}] > 0
}

// WorkspaceUnreadConversationCount returns the authoritative number of
// scopes currently contributing to the workspace red-dot aggregate
// (Count()>0 || ManualUnread). This is the same value published on the
// snapshot and replayed through runtime.unread_updated events so clients
// can replace the badge counter directly without recomputing from a
// partial scope map.
func (r *Runtime) WorkspaceUnreadConversationCount(workspaceID, memberID string) int64 {
	workspaceID = cleanID(workspaceID)
	memberID = cleanID(memberID)
	if workspaceID == "" || memberID == "" {
		return 0
	}

	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return 0
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()
	return int64(sh.aggregates[memberKey{workspaceID: workspaceID, memberID: memberID}])
}

// MarkScopeServingState updates all loaded entries for a scope to a serving
// state. It is used by recovery to suppress exact counts while a pending
// durable frontier is being replayed.
func (r *Runtime) MarkScopeServingState(workspaceID string, scopeID string, state ServingState, markDirty bool) {
	workspaceID = cleanID(workspaceID)
	scopeID = cleanID(scopeID)
	if workspaceID == "" || scopeID == "" {
		return
	}

	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()
	if sh.isScopePermanentlyInvalidated(workspaceID, scopeID) {
		return
	}
	sh.markScopeEntriesState(workspaceID, scopeID, state, markDirty)
}

// MarkEntryServingState updates one loaded member/scope entry state.
func (r *Runtime) MarkEntryServingState(workspaceID string, memberID string, scopeID string, state ServingState, markDirty bool) {
	workspaceID = cleanID(workspaceID)
	memberID = cleanID(memberID)
	scopeID = cleanID(scopeID)
	if workspaceID == "" || memberID == "" || scopeID == "" {
		return
	}

	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return
	}

	key := entryKey{workspaceID: workspaceID, memberID: memberID, scopeID: scopeID}
	sh.mu.Lock()
	defer sh.mu.Unlock()
	if sh.isScopePermanentlyInvalidated(workspaceID, scopeID) {
		delete(sh.dirty, key)
		return
	}
	if sh.entries[key] == nil {
		return
	}
	sh.updateEntry(key, func(current *entry) {
		current.state.ServingState = state
	}, markDirty)
}

// MarkEntriesUnavailable records explicit snapshot misses so warm reads do not
// repeatedly enqueue the same bounded hydrate work.
func (r *Runtime) MarkEntriesUnavailable(workspaceID string, memberID string, scopeIDs []string) {
	workspaceID = cleanID(workspaceID)
	memberID = cleanID(memberID)
	scopeIDs = compactJobStrings(scopeIDs)
	if workspaceID == "" || memberID == "" || len(scopeIDs) == 0 {
		return
	}

	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()
	for _, scopeID := range scopeIDs {
		key := entryKey{workspaceID: workspaceID, memberID: memberID, scopeID: scopeID}
		if sh.isScopePermanentlyInvalidated(workspaceID, scopeID) {
			delete(sh.dirty, key)
			continue
		}
		sh.updateEntry(key, func(current *entry) {
			// A bounded store miss is itself authoritative baseline provenance.
			// Preserve any newer control-only callback state already present;
			// only a genuinely absent entry needs an unavailable serving marker.
			if sh.entries[key] == nil || current.state.ServingState == ServingStateHydrating {
				current.state.ServingState = ServingStateUnavailable
			}
			current.snapshotResolved = true
		}, false)
	}
}

// FlushDirtySnapshots returns and clears up to limit dirty snapshot rows.
func (r *Runtime) FlushDirtySnapshots(limit int, now time.Time) []Snapshot {
	if r == nil || limit <= 0 {
		return nil
	}

	rows := make([]Snapshot, 0, limit)
	for _, sh := range r.shards {
		sh.mu.Lock()
		for key := range sh.dirty {
			if len(rows) >= limit {
				break
			}
			if sh.isScopePermanentlyInvalidated(key.workspaceID, key.scopeID) {
				delete(sh.dirty, key)
				continue
			}
			current := sh.entries[key]
			if current == nil {
				delete(sh.dirty, key)
				continue
			}
			if current.state.isTaskScope() && !current.snapshotResolved {
				// A read/manual callback can beat cold-start snapshot hydrate.
				// Never let that control-only zero baseline overwrite an older
				// durable Task snapshot; keep it dirty until hydrate/explicit
				// miss resolves the baseline and overlays the callback.
				continue
			}
			rows = append(rows, current.snapshotRow(now))
			delete(sh.dirty, key)
		}
		sh.mu.Unlock()
		if len(rows) >= limit {
			break
		}
	}
	return rows
}

// RequeueDirtySnapshots marks rows returned by FlushDirtySnapshots dirty again
// after a failed snapshot write. It preserves current memory as the serving
// source and only restores the flush backlog marker.
func (r *Runtime) RequeueDirtySnapshots(rows []Snapshot) {
	if r == nil || len(rows) == 0 {
		return
	}
	for _, row := range rows {
		workspaceID := cleanID(row.WorkspaceID)
		memberID := cleanID(row.MemberID)
		scopeID := cleanID(row.ScopeID)
		if workspaceID == "" || memberID == "" || scopeID == "" {
			continue
		}
		sh := r.shardForWorkspace(workspaceID)
		if sh == nil {
			continue
		}
		key := entryKey{workspaceID: workspaceID, memberID: memberID, scopeID: scopeID}
		sh.mu.Lock()
		if sh.isScopePermanentlyInvalidated(workspaceID, scopeID) {
			delete(sh.dirty, key)
		} else if sh.entries[key] != nil {
			sh.dirty[key] = struct{}{}
		}
		sh.mu.Unlock()
	}
}

// PendingFrontiers returns coalesced pending frontier work.
func (r *Runtime) PendingFrontiers(workspaceID string) []PendingFrontier {
	workspaceID = cleanID(workspaceID)
	if workspaceID == "" {
		return nil
	}

	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return nil
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()

	pending := make([]PendingFrontier, 0, len(sh.pending))
	for key, frontier := range sh.pending {
		if key.workspaceID == workspaceID && !sh.isScopePermanentlyInvalidatedKey(key) {
			pending = append(pending, frontier)
		}
	}
	return pending
}

// PendingFrontiersAll returns coalesced pending frontier work across all
// workspace shards.
func (r *Runtime) PendingFrontiersAll() []PendingFrontier {
	if r == nil {
		return nil
	}
	pending := make([]PendingFrontier, 0)
	for _, sh := range r.shards {
		sh.mu.Lock()
		for key, frontier := range sh.pending {
			if sh.isScopePermanentlyInvalidatedKey(key) {
				continue
			}
			pending = append(pending, frontier)
		}
		sh.mu.Unlock()
	}
	return pending
}

// PendingFrontierForScope returns the coalesced pending frontier for one scope.
func (r *Runtime) PendingFrontierForScope(workspaceID string, scopeID string) (PendingFrontier, bool) {
	workspaceID = cleanID(workspaceID)
	scopeID = cleanID(scopeID)
	if workspaceID == "" || scopeID == "" {
		return PendingFrontier{}, false
	}

	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return PendingFrontier{}, false
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()
	key := scopeKey{workspaceID: workspaceID, scopeID: scopeID}
	if sh.isScopePermanentlyInvalidatedKey(key) {
		return PendingFrontier{}, false
	}
	frontier, ok := sh.pending[key]
	return frontier, ok
}

// MarkPendingFrontier coalesces durable frontier work discovered by metadata
// checks or event backpressure.
func (r *Runtime) MarkPendingFrontier(frontier PendingFrontier) {
	frontier.WorkspaceID = cleanID(frontier.WorkspaceID)
	frontier.ScopeID = cleanID(frontier.ScopeID)
	if frontier.WorkspaceID == "" || frontier.ScopeID == "" {
		return
	}

	sh := r.shardForWorkspace(frontier.WorkspaceID)
	if sh == nil {
		return
	}

	key := scopeKey{workspaceID: frontier.WorkspaceID, scopeID: frontier.ScopeID}
	sh.mu.Lock()
	defer sh.mu.Unlock()
	if sh.isScopePermanentlyInvalidatedKey(key) {
		delete(sh.pending, key)
		delete(sh.pendingMessages, key)
		delete(sh.pendingActivities, key)
		delete(sh.pendingOverflow, key)
		return
	}
	current := sh.pending[key]
	current.WorkspaceID = frontier.WorkspaceID
	current.ScopeID = frontier.ScopeID
	if current.ScopeType == "" {
		current.ScopeType = frontier.ScopeType
	}
	current.MessageSeq = maxInt64(current.MessageSeq, frontier.MessageSeq)
	current.ActivitySeq = maxInt64(current.ActivitySeq, frontier.ActivitySeq)
	current.MessageGeneration = maxInt64(current.MessageGeneration, frontier.MessageGeneration)
	current.ActivityGeneration = maxInt64(current.ActivityGeneration, frontier.ActivityGeneration)
	sh.pending[key] = current
}

// ScopeAccountedFrontier returns the accounted message and activity sequence
// for one scope. A scope-level recovered frontier is authoritative because some
// events legitimately update no member entry, such as author-only or ineligible
// roster events.
func (r *Runtime) ScopeAccountedFrontier(workspaceID string, scopeID string) (messageSeq int64, activitySeq int64, ok bool) {
	workspaceID = cleanID(workspaceID)
	scopeID = cleanID(scopeID)
	if workspaceID == "" || scopeID == "" {
		return 0, 0, false
	}

	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return 0, 0, false
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()
	if sh.isScopePermanentlyInvalidated(workspaceID, scopeID) {
		return 0, 0, false
	}
	messageFromScope := false
	activityFromScope := false
	if scope := sh.scopes[scopeKey{workspaceID: workspaceID, scopeID: scopeID}]; scope != nil {
		messageSeq = scope.latestMessageAccountedSeq
		activitySeq = scope.latestActivityAccountedSeq
		messageFromScope = messageSeq > 0
		activityFromScope = activitySeq > 0
		ok = messageFromScope || activityFromScope
	}
	for key, current := range sh.entries {
		if key.workspaceID != workspaceID || key.scopeID != scopeID || current == nil {
			continue
		}
		ok = true
		if !messageFromScope && messageSeq == 0 {
			messageSeq = current.latestMessageAccountedSeq
		}
		if !activityFromScope && activitySeq == 0 {
			activitySeq = current.latestActivityAccountedSeq
		}
		if !messageFromScope && current.latestMessageAccountedSeq < messageSeq {
			messageSeq = current.latestMessageAccountedSeq
		}
		if !activityFromScope && current.latestActivityAccountedSeq < activitySeq {
			activitySeq = current.latestActivityAccountedSeq
		}
	}
	return messageSeq, activitySeq, ok
}

// MarkScopeRecovered records that bounded recovery has caught one scope up to
// the supplied durable frontier, even when no member entry changed.
func (r *Runtime) MarkScopeRecovered(frontier PendingFrontier) {
	workspaceID := cleanID(frontier.WorkspaceID)
	scopeID := cleanID(frontier.ScopeID)
	if workspaceID == "" || scopeID == "" {
		return
	}

	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()
	if sh.isScopePermanentlyInvalidated(workspaceID, scopeID) {
		return
	}
	scope := sh.scope(workspaceID, scopeID)
	if frontier.ScopeType != "" {
		scope.scopeType = frontier.ScopeType
	}
	scope.latestMessageAccountedSeq = maxInt64(scope.latestMessageAccountedSeq, frontier.MessageSeq)
	scope.latestActivityAccountedSeq = maxInt64(scope.latestActivityAccountedSeq, frontier.ActivitySeq)
	scope.scopeMessageGeneration = maxInt64(scope.scopeMessageGeneration, frontier.MessageGeneration)
	scope.scopeActivityGeneration = maxInt64(scope.scopeActivityGeneration, frontier.ActivityGeneration)
}

// ClearPendingFrontier removes a pending frontier after recovery catches up to
// at least the observed values. Newer coalesced work is left in place.
func (r *Runtime) ClearPendingFrontier(frontier PendingFrontier) {
	workspaceID := cleanID(frontier.WorkspaceID)
	scopeID := cleanID(frontier.ScopeID)
	if workspaceID == "" || scopeID == "" {
		return
	}

	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return
	}

	key := scopeKey{workspaceID: workspaceID, scopeID: scopeID}
	sh.mu.Lock()
	defer sh.mu.Unlock()
	if sh.isScopePermanentlyInvalidatedKey(key) {
		delete(sh.pending, key)
		delete(sh.pendingMessages, key)
		delete(sh.pendingActivities, key)
		delete(sh.pendingOverflow, key)
		return
	}
	current, ok := sh.pending[key]
	if !ok {
		return
	}
	if current.MessageSeq > frontier.MessageSeq ||
		current.ActivitySeq > frontier.ActivitySeq ||
		current.MessageGeneration > frontier.MessageGeneration ||
		current.ActivityGeneration > frontier.ActivityGeneration {
		return
	}
	delete(sh.pending, key)
}

func (r *Runtime) rosterBatch(workspaceID string, scopeID string, start int, limit int) ([]RosterMember, bool) {
	if limit <= 0 {
		limit = 1
	}
	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return nil, false
	}

	sh.mu.Lock()
	defer sh.mu.Unlock()

	key := scopeKey{workspaceID: workspaceID, scopeID: scopeID}
	if sh.isScopePermanentlyInvalidatedKey(key) {
		return nil, false
	}
	scope := sh.scopes[key]
	if scope == nil || !scope.rosterReady {
		return nil, false
	}
	if start >= len(scope.memberIDs) {
		return nil, true
	}
	end := start + limit
	if end > len(scope.memberIDs) {
		end = len(scope.memberIDs)
	}
	members := make([]RosterMember, 0, end-start)
	for _, memberID := range scope.memberIDs[start:end] {
		member, ok := scope.members[memberID]
		if ok {
			members = append(members, member)
		}
	}
	return members, true
}

func (sh *shard) scope(workspaceID, scopeID string) *scopeState {
	key := scopeKey{workspaceID: workspaceID, scopeID: scopeID}
	state := sh.scopes[key]
	if state == nil {
		state = &scopeState{members: make(map[string]RosterMember)}
		sh.scopes[key] = state
	}
	return state
}

func (sh *shard) isScopePermanentlyInvalidated(workspaceID, scopeID string) bool {
	return sh.isScopePermanentlyInvalidatedKey(scopeKey{workspaceID: workspaceID, scopeID: scopeID})
}

func (sh *shard) isScopePermanentlyInvalidatedKey(key scopeKey) bool {
	if sh == nil || len(sh.permanentScopeTombstones) == 0 {
		return false
	}
	_, ok := sh.permanentScopeTombstones[key]
	return ok
}

func normalizeRosterMember(member RosterMember) (RosterMember, bool) {
	member.MemberID = cleanID(member.MemberID)
	if member.MemberID == "" {
		return RosterMember{}, false
	}
	member.UserID = cleanID(member.UserID)
	member.NotificationState = cleanState(member.NotificationState)
	member.Visibility = cleanState(member.Visibility)
	return member, true
}

func normalizeRewoundScope(scope RewoundScope) RewoundScope {
	scope.ScopeID = cleanID(scope.ScopeID)
	if scope.LatestSeq < 0 {
		scope.LatestSeq = 0
	}
	if len(scope.RemovedMessages) == 0 {
		return scope
	}
	messages := make([]RewoundMessage, 0, len(scope.RemovedMessages))
	for _, message := range scope.RemovedMessages {
		message.ScopeID = cleanID(message.ScopeID)
		if message.ScopeID == "" {
			message.ScopeID = scope.ScopeID
		}
		if message.ScopeID != scope.ScopeID || message.Seq <= 0 {
			continue
		}
		message.AuthorMemberID = cleanID(message.AuthorMemberID)
		message.NotifyPrincipalMemberID = cleanID(message.NotifyPrincipalMemberID)
		if message.NotifyPrincipalMemberID == "" {
			message.NotifyPrincipalMemberID = message.AuthorMemberID
		}
		message.MessageType = cleanState(message.MessageType)
		messages = append(messages, message)
	}
	scope.RemovedMessages = messages
	return scope
}

func (sh *shard) updateEntry(key entryKey, update func(*entry), markDirty bool) {
	if sh.isScopePermanentlyInvalidated(key.workspaceID, key.scopeID) {
		delete(sh.dirty, key)
		return
	}
	sh.updateEntryUnchecked(key, update, markDirty)
}

func (sh *shard) updateEntryUnchecked(key entryKey, update func(*entry), markDirty bool) {
	current := sh.entries[key]
	if current == nil {
		current = &entry{
			state: ScopeUnreadState{
				WorkspaceID:  key.workspaceID,
				MemberID:     key.memberID,
				ScopeID:      key.scopeID,
				ServingState: ServingStateHydrating,
			},
		}
		sh.entries[key] = current
	}

	member := memberKey{workspaceID: key.workspaceID, memberID: key.memberID}
	before := servesExternally(current.state.ServingState) && current.state.HasUnread()
	previousStateVersion := current.state.StateVersion
	update(current)
	if current.state.StateVersion <= previousStateVersion {
		current.state.StateVersion = previousStateVersion + 1
	}
	current.state.WorkspaceID = key.workspaceID
	current.state.MemberID = key.memberID
	current.state.ScopeID = key.scopeID
	if current.state.ScopeType == "" {
		if scope := sh.scopes[scopeKey{workspaceID: key.workspaceID, scopeID: key.scopeID}]; scope != nil {
			current.state.ScopeType = scope.scopeType
		}
	}
	if current.state.ScopeType == "" {
		current.state.ScopeType = scopeTypeForSnapshotScopeID(key.scopeID)
	}
	after := servesExternally(current.state.ServingState) && current.state.HasUnread()
	switch {
	case !before && after:
		sh.aggregates[member]++
	case before && !after:
		if sh.aggregates[member] > 0 {
			sh.aggregates[member]--
		}
	}
	if markDirty {
		sh.dirty[key] = struct{}{}
	}
}

func (sh *shard) markScopeEntriesState(workspaceID, scopeID string, state ServingState, markDirty bool) {
	if sh.isScopePermanentlyInvalidated(workspaceID, scopeID) {
		return
	}
	for key := range sh.entries {
		if key.workspaceID != workspaceID || key.scopeID != scopeID {
			continue
		}
		entryKey := key
		sh.updateEntry(entryKey, func(current *entry) {
			current.state.ServingState = state
		}, markDirty)
	}
}

func (sh *shard) suppressScopeEntries(workspaceID, scopeID string, state ServingState) {
	for key := range sh.entries {
		if key.workspaceID != workspaceID || key.scopeID != scopeID {
			continue
		}
		sh.suppressEntry(key, state)
	}
}

func (sh *shard) suppressEntry(key entryKey, state ServingState) {
	if sh.entries[key] == nil {
		return
	}
	sh.updateEntry(key, func(current *entry) {
		current.state.ServingState = state
	}, false)
}

func (sh *shard) suppressEntryWithDelta(key entryKey, state ServingState) (UnreadDelta, bool) {
	if sh.isScopePermanentlyInvalidated(key.workspaceID, key.scopeID) {
		return UnreadDelta{}, false
	}
	return sh.suppressEntryWithDeltaUnchecked(key, state)
}

func (sh *shard) suppressEntryWithDeltaUnchecked(key entryKey, state ServingState) (UnreadDelta, bool) {
	current := sh.entries[key]
	if current == nil {
		return UnreadDelta{}, false
	}
	sh.updateEntryUnchecked(key, func(current *entry) {
		current.state.ServingState = state
	}, false)
	current = sh.entries[key]
	if current == nil || current.userID == "" {
		return UnreadDelta{}, false
	}
	return UnreadDelta{
		WorkspaceID:                      key.workspaceID,
		UserID:                           current.userID,
		MemberID:                         key.memberID,
		ScopeID:                          key.scopeID,
		ScopeType:                        current.state.ScopeType,
		Count:                            0,
		ManualUnread:                     false,
		HasUnread:                        false,
		WorkspaceHasUnread:               sh.aggregates[memberKey{workspaceID: key.workspaceID, memberID: key.memberID}] > 0,
		WorkspaceUnreadConversationCount: int64(sh.aggregates[memberKey{workspaceID: key.workspaceID, memberID: key.memberID}]),
		ServingState:                     state,
		StateVersion:                     current.state.StateVersion,
	}, true
}

// clearEntryForNotificationMute zeroes one member/scope entry's loaded unread,
// suppresses it, and remembers the pre-mute roster baseline so messages posted
// while the scope stays muted remain skipped until the roster is refreshed on
// re-activation. The entry is marked dirty so the cleared zero state is
// persisted durably instead of being revived after a restart or a later
// re-activation message.
func (sh *shard) clearEntryForNotificationMute(key entryKey) (UnreadDelta, bool) {
	if sh.isScopePermanentlyInvalidated(key.workspaceID, key.scopeID) {
		return UnreadDelta{}, false
	}
	// Always remember the pre-mute roster baseline (from the entry when present,
	// otherwise from the loaded roster row) so a member who mutes before any
	// entry exists still has mute-period messages skipped.
	sh.rememberSuppressedRosterMember(key, 0, 0, false)
	if sh.entries[key] == nil {
		return UnreadDelta{}, false
	}
	sh.updateEntry(key, func(current *entry) {
		current.state.MessageUnreadCount = 0
		current.state.ActivityUnreadCount = 0
		current.state.ManualUnread = false
		clearThrough := maxInt64(
			maxInt64(current.lastMessageReadSeq, current.latestMessageAccountedSeq),
			maxInt64(current.lastActivityReadSeq, current.latestActivityAccountedSeq),
		)
		current.clearMessageUnreadThrough(clearThrough)
		current.clearActivityUnreadThrough(clearThrough)
		current.state.ServingState = ServingStateStale
	}, true)
	current := sh.entries[key]
	if current == nil || current.userID == "" {
		return UnreadDelta{}, false
	}
	return UnreadDelta{
		WorkspaceID:                      key.workspaceID,
		UserID:                           current.userID,
		MemberID:                         key.memberID,
		ScopeID:                          key.scopeID,
		ScopeType:                        current.state.ScopeType,
		Count:                            0,
		ManualUnread:                     false,
		HasUnread:                        false,
		WorkspaceHasUnread:               sh.aggregates[memberKey{workspaceID: key.workspaceID, memberID: key.memberID}] > 0,
		WorkspaceUnreadConversationCount: int64(sh.aggregates[memberKey{workspaceID: key.workspaceID, memberID: key.memberID}]),
		ServingState:                     ServingStateStale,
		StateVersion:                     current.state.StateVersion,
	}, true
}

func (sh *shard) rememberSuppressedRosterMember(
	key entryKey,
	suppressedMessageThroughSeq int64,
	suppressedActivityThroughSeq int64,
	sequenceFrontiersKnown bool,
) {
	if sh.suppressedRoster == nil {
		sh.suppressedRoster = make(map[entryKey]suppressedRosterMember)
	}
	if current := sh.entries[key]; current != nil {
		suppressed := suppressedRosterMemberFromEntry(current)
		suppressed.suppressedMessageThroughSeq = maxInt64(suppressedMessageThroughSeq, 0)
		suppressed.suppressedActivityThroughSeq = maxInt64(suppressedActivityThroughSeq, 0)
		suppressed.sequenceFrontiersKnown = sequenceFrontiersKnown
		if frontier, ok := sh.pending[scopeKey{workspaceID: key.workspaceID, scopeID: key.scopeID}]; ok {
			suppressed.effectiveFromFloor = maxInt64(
				suppressed.effectiveFromFloor,
				maxInt64(frontier.MessageSeq, frontier.ActivitySeq)+1,
			)
		}
		suppressed.effectiveFromFloor = maxInt64(
			suppressed.effectiveFromFloor,
			maxInt64(current.latestMessageAccountedSeq, current.latestActivityAccountedSeq)+1,
		)
		sh.suppressedRoster[key] = suppressed
		return
	}
	scope := sh.scopes[scopeKey{workspaceID: key.workspaceID, scopeID: key.scopeID}]
	if scope == nil || scope.members == nil {
		return
	}
	member, ok := scope.members[key.memberID]
	if !ok {
		return
	}
	suppressed := suppressedRosterMemberFromRoster(member)
	suppressed.suppressedMessageThroughSeq = maxInt64(suppressedMessageThroughSeq, 0)
	suppressed.suppressedActivityThroughSeq = maxInt64(suppressedActivityThroughSeq, 0)
	suppressed.sequenceFrontiersKnown = sequenceFrontiersKnown
	if frontier, ok := sh.pending[scopeKey{workspaceID: key.workspaceID, scopeID: key.scopeID}]; ok {
		suppressed.effectiveFromFloor = maxInt64(
			suppressed.effectiveFromFloor,
			maxInt64(frontier.MessageSeq, frontier.ActivitySeq)+1,
		)
	}
	sh.suppressedRoster[key] = suppressed
}

// applyTaskHideFrontiers clears the exact committed Task message/Activity
// prefixes. During the additive hidden-through-Activity rollout, only a
// causally later message can supersede the hide. Activity-driven reopening
// remains disabled until every legacy backend writer has drained and the
// Activity frontier has been backfilled again; otherwise an old writer that
// only advances hidden_through_seq could make a new reader spuriously unhide a
// Task during a rolling deploy.
func (sh *shard) applyTaskHideFrontiers(key entryKey, event ScopeInvalidated) bool {
	if !event.FrontiersKnown {
		return false
	}
	isTask := false
	if current := sh.entries[key]; current != nil {
		isTask = current.state.isTaskScope()
	}
	if !isTask {
		if scope := sh.scopes[scopeKey{workspaceID: key.workspaceID, scopeID: key.scopeID}]; scope != nil {
			isTask = scope.scopeType == ScopeTypeTask
		}
	}
	if !isTask {
		return false
	}
	superseded := false
	sh.updateEntry(key, func(current *entry) {
		current.state.ScopeType = ScopeTypeTask
		superseded = current.latestMessageAccountedSeq > event.Frontier
		current.clearMessageUnreadThrough(event.Frontier)
		current.clearActivityUnreadThrough(event.ActivityFrontier)
		current.lastMessageReadSeq = maxInt64(current.lastMessageReadSeq, event.Frontier)
		current.lastActivityReadSeq = maxInt64(current.lastActivityReadSeq, event.ActivityFrontier)
		current.state.ManualUnread = false
		current.manualUnreadMutationSet = true
		if superseded {
			current.state.ServingState = ServingStateReady
		}
	}, false)
	return superseded
}

func (sh *shard) skipSuppressedRosterMember(key entryKey, member RosterMember, eventSeq int64, activity bool) bool {
	if len(sh.suppressedRoster) == 0 {
		return false
	}
	suppressed, ok := sh.suppressedRoster[key]
	if !ok {
		return false
	}
	if suppressed.sequenceFrontiersKnown {
		frontier := suppressed.suppressedMessageThroughSeq
		if activity {
			// Phase 1 keeps legacy message-only hide/reopen semantics. A Task
			// Activity can still produce unread and wake while the Task is
			// visible, but it cannot reveal a hidden Inbox row until the
			// post-drain contract release activates the additive Activity
			// frontier on every reader.
			return true
		}
		if eventSeq <= frontier {
			return true
		}
		// A Task message after its hide frontier retains the legacy reopen
		// contract. The entry's advanced read cursors continue rejecting delayed
		// callbacks from the hidden prefix.
		delete(sh.suppressedRoster, key)
		return false
	}
	if suppressed.matchesRosterMember(member) {
		return true
	}
	if suppressed.effectiveFromFloor > 0 && member.EffectiveFromSeq <= suppressed.effectiveFromFloor {
		return true
	}
	if !suppressed.matchesRosterMember(member) {
		delete(sh.suppressedRoster, key)
	}
	return false
}

func (sh *shard) coalescePendingMessage(event MessageStored) {
	key := scopeKey{workspaceID: event.WorkspaceID, scopeID: event.ScopeID}
	if sh.isScopePermanentlyInvalidatedKey(key) {
		return
	}
	frontier := sh.pending[key]
	frontier.WorkspaceID = event.WorkspaceID
	frontier.ScopeID = event.ScopeID
	if frontier.ScopeType == "" {
		frontier.ScopeType = event.ScopeType
	}
	frontier.MessageSeq = maxInt64(frontier.MessageSeq, event.Seq)
	frontier.MessageGeneration = maxInt64(frontier.MessageGeneration, event.MessageGeneration)
	sh.pending[key] = frontier
}

func (sh *shard) queuePendingMessage(event MessageStored, limit int) {
	key := scopeKey{workspaceID: event.WorkspaceID, scopeID: event.ScopeID}
	if sh.isScopePermanentlyInvalidatedKey(key) {
		return
	}
	pending := sh.pendingMessages[key]
	if limit > 0 && len(pending) >= limit {
		sh.pendingOverflow[key] = struct{}{}
		return
	}
	for index, existing := range pending {
		if existing.Seq == event.Seq {
			pending[index] = mergePendingMessage(existing, event)
			sh.pendingMessages[key] = pending
			return
		}
	}
	sh.pendingMessages[key] = append(pending, event)
}

func (sh *shard) coalescePendingActivity(event NonMessageActivityStored) {
	key := scopeKey{workspaceID: event.WorkspaceID, scopeID: event.ScopeID}
	if sh.isScopePermanentlyInvalidatedKey(key) {
		return
	}
	frontier := sh.pending[key]
	frontier.WorkspaceID = event.WorkspaceID
	frontier.ScopeID = event.ScopeID
	if frontier.ScopeType == "" {
		frontier.ScopeType = event.ScopeType
	}
	frontier.ActivitySeq = maxInt64(frontier.ActivitySeq, event.Seq)
	frontier.ActivityGeneration = maxInt64(frontier.ActivityGeneration, event.ActivityGeneration)
	sh.pending[key] = frontier
}

func (sh *shard) queuePendingActivity(event NonMessageActivityStored, limit int) {
	key := scopeKey{workspaceID: event.WorkspaceID, scopeID: event.ScopeID}
	if sh.isScopePermanentlyInvalidatedKey(key) {
		return
	}
	pending := sh.pendingActivities[key]
	if limit > 0 && len(pending) >= limit {
		sh.pendingOverflow[key] = struct{}{}
		return
	}
	for index, existing := range pending {
		if existing.Seq == event.Seq {
			pending[index] = mergePendingActivity(existing, event)
			sh.pendingActivities[key] = pending
			return
		}
	}
	sh.pendingActivities[key] = append(pending, event)
}

func (r *Runtime) drainPendingScopeEvents(workspaceID, scopeID string) ([]MessageStored, []NonMessageActivityStored, bool) {
	sh := r.shardForWorkspace(workspaceID)
	if sh == nil {
		return nil, nil, false
	}
	key := scopeKey{workspaceID: workspaceID, scopeID: scopeID}
	sh.mu.Lock()
	defer sh.mu.Unlock()
	if sh.isScopePermanentlyInvalidatedKey(key) {
		delete(sh.pending, key)
		delete(sh.pendingMessages, key)
		delete(sh.pendingActivities, key)
		delete(sh.pendingOverflow, key)
		return nil, nil, true
	}

	messages := append([]MessageStored(nil), sh.pendingMessages[key]...)
	activities := append([]NonMessageActivityStored(nil), sh.pendingActivities[key]...)
	if len(messages) == 0 && len(activities) == 0 {
		return nil, nil, false
	}
	delete(sh.pendingMessages, key)
	delete(sh.pendingActivities, key)
	if _, overflow := sh.pendingOverflow[key]; !overflow {
		delete(sh.pending, key)
	}
	delete(sh.pendingOverflow, key)
	return messages, activities, false
}

func mergePendingMessage(existing MessageStored, next MessageStored) MessageStored {
	if existing.MessageGeneration < next.MessageGeneration {
		existing.MessageGeneration = next.MessageGeneration
	}
	if existing.AuthorMemberID == "" {
		existing.AuthorMemberID = next.AuthorMemberID
	}
	if existing.NotifyPrincipalMemberID == "" {
		existing.NotifyPrincipalMemberID = next.NotifyPrincipalMemberID
	}
	existing.AllowAuthorUnread = existing.AllowAuthorUnread || next.AllowAuthorUnread
	if existing.ScopeType == "" {
		existing.ScopeType = next.ScopeType
	}
	existing.RecipientMemberIDs = normalizeRecipientMemberIDs(append(append([]string(nil), existing.RecipientMemberIDs...), next.RecipientMemberIDs...))
	return existing
}

func mergePendingActivity(existing NonMessageActivityStored, next NonMessageActivityStored) NonMessageActivityStored {
	if existing.ActivityGeneration < next.ActivityGeneration {
		existing.ActivityGeneration = next.ActivityGeneration
	}
	if existing.ActorMemberID == "" {
		existing.ActorMemberID = next.ActorMemberID
	}
	if existing.ScopeType == "" {
		existing.ScopeType = next.ScopeType
	}
	existing.RecipientMemberIDs = normalizeRecipientMemberIDs(append(append([]string(nil), existing.RecipientMemberIDs...), next.RecipientMemberIDs...))
	return existing
}

func applyRosterMember(current *entry, member RosterMember) {
	if member.UserID != "" {
		current.userID = member.UserID
	}
	current.lastMessageReadSeq = maxInt64(current.lastMessageReadSeq, member.LastMessageReadSeq)
	current.lastActivityReadSeq = maxInt64(current.lastActivityReadSeq, member.LastActivityReadSeq)
	current.effectiveFromSeq = maxInt64(current.effectiveFromSeq, member.EffectiveFromSeq)
	current.notificationState = cleanState(member.NotificationState)
	current.membershipVisibility = cleanState(member.Visibility)
}

func suppressedRosterMemberFromEntry(current *entry) suppressedRosterMember {
	if current == nil {
		return suppressedRosterMember{}
	}
	hasRosterBaseline :=
		current.lastMessageReadSeq > 0 ||
			current.lastActivityReadSeq > 0 ||
			current.effectiveFromSeq > 0 ||
			current.notificationState != "" ||
			current.membershipVisibility != ""
	return suppressedRosterMember{
		lastMessageReadSeq:  current.lastMessageReadSeq,
		lastActivityReadSeq: current.lastActivityReadSeq,
		effectiveFromSeq:    current.effectiveFromSeq,
		hasRosterBaseline:   hasRosterBaseline,
		notificationState:   current.notificationState,
		visibility:          current.membershipVisibility,
	}
}

func suppressedRosterMemberFromRoster(member RosterMember) suppressedRosterMember {
	member, _ = normalizeRosterMember(member)
	return suppressedRosterMember{
		lastMessageReadSeq:  member.LastMessageReadSeq,
		lastActivityReadSeq: member.LastActivityReadSeq,
		effectiveFromSeq:    member.EffectiveFromSeq,
		hasRosterBaseline:   true,
		notificationState:   cleanState(member.NotificationState),
		visibility:          cleanState(member.Visibility),
	}
}

func (current suppressedRosterMember) matchesRosterMember(member RosterMember) bool {
	member, ok := normalizeRosterMember(member)
	if !ok {
		return false
	}
	if !current.hasRosterBaseline {
		return false
	}
	return current.lastMessageReadSeq == member.LastMessageReadSeq &&
		current.lastActivityReadSeq == member.LastActivityReadSeq &&
		current.effectiveFromSeq == member.EffectiveFromSeq &&
		current.notificationState == cleanState(member.NotificationState) &&
		current.visibility == cleanState(member.Visibility)
}

func (sh *shard) deltaForEntry(key entryKey) (UnreadDelta, bool) {
	current := sh.entries[key]
	if current == nil {
		return UnreadDelta{}, false
	}
	state := current.state
	return UnreadDelta{
		WorkspaceID:                      state.WorkspaceID,
		UserID:                           current.userID,
		MemberID:                         state.MemberID,
		ScopeID:                          state.ScopeID,
		ScopeType:                        state.ScopeType,
		Count:                            state.Count(),
		ManualUnread:                     state.ManualUnread,
		HasUnread:                        state.HasUnread(),
		WorkspaceHasUnread:               sh.aggregates[memberKey{workspaceID: key.workspaceID, memberID: key.memberID}] > 0,
		WorkspaceUnreadConversationCount: int64(sh.aggregates[memberKey{workspaceID: key.workspaceID, memberID: key.memberID}]),
		ServingState:                     state.ServingState,
		StateVersion:                     state.StateVersion,
	}, true
}

func eligibleForMessage(member RosterMember, event MessageStored) bool {
	if !eligibleForEventAudience(member, event.ScopeType, event.RecipientMemberIDs) {
		return false
	}
	if !event.AllowAuthorUnread && member.MemberID == event.NotifyPrincipalMemberID {
		return false
	}
	return event.Seq > member.LastMessageReadSeq && event.Seq >= member.EffectiveFromSeq
}

func effectiveRewoundMessageNotifyPrincipal(message RewoundMessage) string {
	if principal := cleanID(message.NotifyPrincipalMemberID); principal != "" {
		return principal
	}
	return cleanID(message.AuthorMemberID)
}

func eligibleForActivity(member RosterMember, event NonMessageActivityStored) bool {
	if !eligibleForEventAudience(member, event.ScopeType, event.RecipientMemberIDs) {
		return false
	}
	if member.MemberID == event.ActorMemberID {
		return false
	}
	return event.Seq > member.LastActivityReadSeq && event.Seq >= member.EffectiveFromSeq
}

func eligibleForEventAudience(member RosterMember, scopeType ScopeType, recipientMemberIDs []string) bool {
	if scopeType == ScopeTypeTask {
		if !recipientMemberIDContains(recipientMemberIDs, member.MemberID) {
			return false
		}
		return eligibleTaskAttentionMember(member)
	}
	if len(recipientMemberIDs) > 0 && !recipientMemberIDContains(recipientMemberIDs, member.MemberID) {
		return false
	}
	return eligibleForUnread(member)
}

func eligibleTaskAttentionMember(member RosterMember) bool {
	if member.Ineligible {
		return false
	}
	switch cleanState(member.Visibility) {
	case "", "public", "dm":
		return true
	default:
		return false
	}
}

func normalizeRecipientMemberIDs(values []string) []string {
	if len(values) == 0 {
		return nil
	}
	normalized := make([]string, 0, len(values))
	seen := make(map[string]struct{}, len(values))
	for _, value := range values {
		value = cleanID(value)
		if value == "" {
			continue
		}
		if _, ok := seen[value]; ok {
			continue
		}
		seen[value] = struct{}{}
		normalized = append(normalized, value)
	}
	return normalized
}

func recipientMemberIDContains(values []string, memberID string) bool {
	memberID = cleanID(memberID)
	for _, value := range values {
		if cleanID(value) == memberID {
			return true
		}
	}
	return false
}

func eligibleForUnread(member RosterMember) bool {
	if member.Ineligible {
		return false
	}
	switch cleanState(member.NotificationState) {
	case "", "active", "quiet":
	default:
		return false
	}
	switch cleanState(member.Visibility) {
	case "", "public", "dm":
		return true
	default:
		return false
	}
}

func servesExternally(state ServingState) bool {
	return state == ServingStateReady || state == ServingStateSnapshotServing
}

func scopeTypeForSnapshotScopeID(scopeID string) ScopeType {
	scopeID = cleanID(scopeID)
	switch {
	case strings.HasPrefix(scopeID, "tsk_"):
		return ScopeTypeTask
	case strings.HasPrefix(scopeID, "dm_"):
		return ScopeTypeDM
	case strings.HasPrefix(scopeID, "thr_"):
		return ScopeTypeThread
	case strings.HasPrefix(scopeID, "sasc_"):
		return ScopeTypeServiceAccount
	default:
		return ""
	}
}

func batchesFor(total, size int) int {
	if total <= 0 {
		return 0
	}
	if size <= 0 {
		return 1
	}
	return (total + size - 1) / size
}

func (current *entry) snapshotRow(now time.Time) Snapshot {
	return Snapshot{
		WorkspaceID:                current.state.WorkspaceID,
		MemberID:                   current.state.MemberID,
		ScopeID:                    current.state.ScopeID,
		ScopeType:                  current.state.ScopeType,
		MessageUnreadCount:         current.state.MessageUnreadCount,
		ActivityUnreadCount:        current.state.ActivityUnreadCount,
		ManualUnread:               current.state.ManualUnread,
		LastMessageReadSeq:         current.lastMessageReadSeq,
		LastActivityReadSeq:        current.lastActivityReadSeq,
		EffectiveFromSeq:           current.effectiveFromSeq,
		LatestMessageAccountedSeq:  current.latestMessageAccountedSeq,
		LatestActivityAccountedSeq: current.latestActivityAccountedSeq,
		NotificationState:          current.notificationState,
		MembershipVisibility:       current.membershipVisibility,
		ScopeMessageGeneration:     current.scopeMessageGeneration,
		ScopeActivityGeneration:    current.scopeActivityGeneration,
		MembershipGeneration:       current.membershipGeneration,
		NotificationGeneration:     current.notificationGeneration,
		ManualUnreadGeneration:     current.manualUnreadGeneration,
		RewindGeneration:           current.rewindGeneration,
		UpdatedAt:                  now,
		ServingState:               current.state.ServingState,
	}
}

const maxTrackedTaskUnreadSeqs = 256

func (current *entry) appendTaskMessageUnreadSeq(seq int64) bool {
	if seq <= 0 || (current.messageUnreadBaselineSeq > 0 && seq <= current.messageUnreadBaselineSeq) {
		return false
	}
	index := sort.Search(len(current.messageUnreadSeqs), func(index int) bool {
		return current.messageUnreadSeqs[index] >= seq
	})
	if index < len(current.messageUnreadSeqs) && current.messageUnreadSeqs[index] == seq {
		return false
	}
	current.messageUnreadSeqs = append(current.messageUnreadSeqs, 0)
	copy(current.messageUnreadSeqs[index+1:], current.messageUnreadSeqs[index:])
	current.messageUnreadSeqs[index] = seq
	if overflow := len(current.messageUnreadSeqs) - maxTrackedTaskUnreadSeqs; len(current.messageUnreadSeqs) > 2*maxTrackedTaskUnreadSeqs {
		current.messageUnreadBaselineCount += int64(overflow)
		current.messageUnreadBaselineSeq = current.messageUnreadSeqs[overflow-1]
		current.messageUnreadSeqs = append([]int64(nil), current.messageUnreadSeqs[overflow:]...)
	}
	current.syncMessageUnreadCount()
	return true
}

func (current *entry) appendTaskActivityUnreadSeq(seq int64) bool {
	if seq <= 0 || (current.activityUnreadBaselineSeq > 0 && seq <= current.activityUnreadBaselineSeq) {
		return false
	}
	index := sort.Search(len(current.activityUnreadSeqs), func(index int) bool {
		return current.activityUnreadSeqs[index] >= seq
	})
	if index < len(current.activityUnreadSeqs) && current.activityUnreadSeqs[index] == seq {
		return false
	}
	current.activityUnreadSeqs = append(current.activityUnreadSeqs, 0)
	copy(current.activityUnreadSeqs[index+1:], current.activityUnreadSeqs[index:])
	current.activityUnreadSeqs[index] = seq
	if overflow := len(current.activityUnreadSeqs) - maxTrackedTaskUnreadSeqs; len(current.activityUnreadSeqs) > 2*maxTrackedTaskUnreadSeqs {
		current.activityUnreadBaselineCount += int64(overflow)
		current.activityUnreadBaselineSeq = current.activityUnreadSeqs[overflow-1]
		current.activityUnreadSeqs = append([]int64(nil), current.activityUnreadSeqs[overflow:]...)
	}
	current.syncActivityUnreadCount()
	return true
}

func (current *entry) syncMessageUnreadCount() {
	current.state.MessageUnreadCount = maxInt64(current.messageUnreadBaselineCount, 0) + int64(len(current.messageUnreadSeqs))
}

func (current *entry) syncActivityUnreadCount() {
	current.state.ActivityUnreadCount = maxInt64(current.activityUnreadBaselineCount, 0) + int64(len(current.activityUnreadSeqs))
}

func (current *entry) clearMessageUnreadThrough(readSeq int64) {
	if current.messageUnreadBaselineCount > 0 &&
		current.messageUnreadBaselineSeq > 0 &&
		readSeq >= current.messageUnreadBaselineSeq {
		current.messageUnreadBaselineCount = 0
		current.messageUnreadBaselineSeq = 0
	}
	current.messageUnreadSeqs = retainSeqsAfter(current.messageUnreadSeqs, readSeq)
	current.syncMessageUnreadCount()
}

func (current *entry) clearActivityUnreadThrough(readSeq int64) {
	if current.activityUnreadBaselineCount > 0 &&
		current.activityUnreadBaselineSeq > 0 &&
		readSeq >= current.activityUnreadBaselineSeq {
		current.activityUnreadBaselineCount = 0
		current.activityUnreadBaselineSeq = 0
	}
	current.activityUnreadSeqs = retainSeqsAfter(current.activityUnreadSeqs, readSeq)
	current.syncActivityUnreadCount()
}

func retainSeqsAfter(seqs []int64, frontier int64) []int64 {
	if len(seqs) == 0 {
		return nil
	}
	retained := seqs[:0]
	for _, seq := range seqs {
		if seq > frontier {
			retained = append(retained, seq)
		}
	}
	if len(retained) == 0 {
		return nil
	}
	return retained
}

func maxInt64(left, right int64) int64 {
	if left > right {
		return left
	}
	return right
}