--[[
	Cerebro - Communication V2 Module
	Hash-based state synchronization with 4 message types:
	  STATE_HEARTBEAT  — Periodic hash broadcast for divergence detection
	  STATE_FETCH      — Request full state snapshot from officer
	  STATE_SNAPSHOT   — Full state payload (players + config + purge ledger)
	  RPC_CONTROL      — Non-persistent action commands (bids, session, etc.)

	Design reference: docs/proposals/communication2.0/PLAN.md
	                   docs/proposals/communication2.0/ROADMAP.md

	IMPORTANT: This module runs in PARALLEL with Communication v1.
	It registers a separate addon prefix (CerebroV2) so v1 and v2 clients
	coexist safely via prefix isolation. v1 is NOT modified.
]]--

local AddonName = "Cerebro"
local Cerebro = LibStub("AceAddon-3.0"):GetAddon(AddonName)

local CommunicationV2 = Cerebro:NewModule("CommunicationV2", "AceEvent-3.0", "AceTimer-3.0")
Cerebro.modules.CommunicationV2 = CommunicationV2

-- V2 Addon Prefix (separate from v1 "Cerebro")
local V2_PREFIX = "CerebroV2"

-- V2 Message Types (4 total)
local MSG_V2 = {
	STATE_HEARTBEAT = "HB",
	STATE_FETCH     = "FETCH",
	STATE_SNAPSHOT  = "SNAP",
	RPC_CONTROL     = "RPC",
}

-- Heartbeat interval (seconds)
local HEARTBEAT_INTERVAL = 15  -- Build B: reduced from 30s

-- Sync barrier convergence window (seconds)
local SYNC_BARRIER_WINDOW = 10


--[[=============================================
	Module Lifecycle
===============================================]]

function CommunicationV2:OnEnable()
	-- Register v2 addon prefix
	Cerebro:RegisterComm(V2_PREFIX, "OnCommReceivedV2")

	-- Initialize v2 sync state
	self:InitSyncState()

	-- Heartbeat timer: REMOVED (v1.0.3) — SyncScheduler:OnHeartbeat() is the
	-- single owner of the 30s heartbeat timer. It includes combat-gating,
	-- quarantine guard, configurable interval, and post-combat flush logic.
	-- Keeping a second timer here caused doubled heartbeat messages.

	-- Fix 1: Listen for combat end so we can flush deferred addon-state UI refreshes.
	self:RegisterEvent("PLAYER_REGEN_ENABLED")

	-- Announce presence after a short delay
	self:ScheduleTimer("OnStartup", 5)

	Cerebro:Debug("CommunicationV2 enabled (prefix: " .. V2_PREFIX .. ")")
end

--- Fix 1: Post-combat flush for deferred addon-state UI refresh.
-- If a heartbeat arrived during a boss encounter and caused an addon state
-- change, we set _pendingAddonStateRefresh = true to avoid hitching mid-pull.
-- This handler fires after combat lockdown clears and delivers the refresh.
function CommunicationV2:PLAYER_REGEN_ENABLED()
	if self._pendingAddonStateRefresh then
		self._pendingAddonStateRefresh = nil
		-- Small delay to ensure lockdown is fully cleared (mirrors SyncScheduler pattern)
		self:ScheduleTimer(function()
			if not InCombatLockdown() then
				Cerebro:RefreshAllUI()
				Cerebro:Debug("V2 post-combat addon state refresh flushed")
			end
		end, 1)
	end
end

function CommunicationV2:OnDisable()
	Cerebro:Debug("CommunicationV2 disabled")
end

function CommunicationV2:InitSyncState()
	-- Initialize v2-specific sync state (separate from v1 syncState)
	local db = Cerebro.db.global
	db.syncStateV2 = db.syncStateV2 or {}

	local s = db.syncStateV2
	s.peerHashes = s.peerHashes or {}    -- { [playerName] = { rosterHash, configHash, lastSeen } }
	s.isStale = s.isStale or false       -- True when we know we're behind
	s.lastHeartbeatSent = s.lastHeartbeatSent or 0
	s.lastSnapshotReceived = s.lastSnapshotReceived or 0
	s.addonStatus = s.addonStatus or {}  -- { [playerName] = { lastSeenAt, lastKnownVersion } }

	-- Fix 10: Prune addonStatus entries older than 7 days to prevent unbounded SavedVariables growth.
	-- Runs once per login/reload (InitSyncState is called from OnEnable).
	local PRUNE_THRESHOLD = 7 * 86400  -- 7 days in seconds
	local now = time()
	for name, info in pairs(s.addonStatus) do
		if info.lastSeenAt and (now - info.lastSeenAt) > PRUNE_THRESHOLD then
			s.addonStatus[name] = nil
		end
	end

	-- Reset transient addon state cache on login/reload (in-memory only)
	self._lastKnownAddonState = {}
	-- Fix 11: Invalidate max-version cache on login/reload (peers may have updated)
	self._cachedMaxVersion = nil
	-- Fix 7: Invalidate online-member cache on login/reload
	self._onlineGuildCache = nil
	self._onlineGuildCacheAt = nil
end

--- Fix 6: Compare two semver strings numerically.
-- Returns: -1 if a < b, 0 if equal, 1 if a > b.
-- Falls back to 0 (treat as equal) if either string is nil or unparseable.
function CommunicationV2:CompareVersions(a, b)
	if not a or not b then return 0 end
	local aMaj, aMin, aPat = a:match("^(%d+)%.(%d+)%.(%d+)")
	local bMaj, bMin, bPat = b:match("^(%d+)%.(%d+)%.(%d+)")
	if not aMaj or not bMaj then return 0 end
	aMaj, aMin, aPat = tonumber(aMaj), tonumber(aMin), tonumber(aPat)
	bMaj, bMin, bPat = tonumber(bMaj), tonumber(bMin), tonumber(bPat)
	if aMaj ~= bMaj then return aMaj < bMaj and -1 or 1 end
	if aMin ~= bMin then return aMin < bMin and -1 or 1 end
	if aPat ~= bPat then return aPat < bPat and -1 or 1 end
	return 0
end

--- Fix 7: Build or return cached online guild member lookup (10s TTL).
-- Used by GetAddonState() to avoid O(N) guild scan per standings row.
function CommunicationV2:GetOnlineGuildMembers()
	local nowT = time()
	if self._onlineGuildCache and (nowT - (self._onlineGuildCacheAt or 0)) < 10 then
		return self._onlineGuildCache
	end
	local lookup = {}
	for i = 1, GetNumGuildMembers() do
		local name, _, _, _, _, _, _, _, online = GetGuildRosterInfo(i)
		if name and online then
			lookup[Cerebro:GetFullPlayerName(name)] = true
		end
	end
	self._onlineGuildCache = lookup
	self._onlineGuildCacheAt = nowT
	return lookup
end

--- Fix 11: Return the maximum non-dev addon version seen across all peers.
-- Cached; invalidated when addonStatus changes in HandleHeartbeat.
function CommunicationV2:GetMaxAddonVersion()
	if self._cachedMaxVersion then return self._cachedMaxVersion end
	local maxVersion = nil
	local s = Cerebro.db.global.syncStateV2 or {}
	for _, info in pairs(s.addonStatus or {}) do
		local v = info.lastKnownVersion
		-- Fix 12: anchor _dev$ to end of string
		if v and not v:find("_dev$") then
			if not maxVersion or self:CompareVersions(v, maxVersion) > 0 then
				maxVersion = v
			end
		end
	end
	-- Also floor max at our own version
	if Cerebro.version and not Cerebro.version:find("_dev$") then
		if not maxVersion or self:CompareVersions(Cerebro.version, maxVersion) > 0 then
			maxVersion = Cerebro.version
		end
	end
	self._cachedMaxVersion = maxVersion
	return maxVersion
end

--- Returns the current addon compliance state for a player.
-- States (in priority order):
--   "synced"   - seen within 90s AND version matches guild max
--   "outdated" - seen within 90s AND version < guild max
--   "inactive" - was seen before but >90s ago (offline/AFK)
--   "no_addon" - in guild, online, but NEVER seen in addonStatus
--   "unknown"  - guild member, never seen, cannot determine online status
-- _dev versions are excluded from max computation so dev builds
-- don't push the effective max above what live players are running.
function CommunicationV2:GetAddonState(fullName)
	-- Self-check: the local player never gets an addonStatus entry because
	-- HandleHeartbeat excludes our own messages. We know we have the addon
	-- running right now, so short-circuit to 'synced' immediately.
	if fullName == Cerebro:GetPlayerFullName() then
		return "synced"
	end

	local s = Cerebro.db.global.syncStateV2 or {}
	local addonStatus = s.addonStatus or {}
	-- Fix 4 (INACTIVE_THRESHOLD): defined once at module level via GetMaxAddonVersion.
	-- Use the module-level constant so StandingsFrame can read CommunicationV2.INACTIVE_THRESHOLD.
	local INACTIVE_THRESHOLD = CommunicationV2.INACTIVE_THRESHOLD

	-- Fix 11: Use cached max-version helper (invalidated on each HB update)
	local maxVersion = self:GetMaxAddonVersion()

	local info = addonStatus[fullName]
	if not info then
		-- Fix 7: Never-seen — use cached online lookup instead of full scan per call
		local onlineMembers = self:GetOnlineGuildMembers()
		if onlineMembers[fullName] then
			return "no_addon"  -- Online but no HB ever received
		end
		return "unknown"  -- Offline or not in guild
	end

	local now = time()
	local age = now - (info.lastSeenAt or 0)
	if age > INACTIVE_THRESHOLD then
		return "inactive"
	end

	local theirVersion = info.lastKnownVersion
	-- Fix 12: anchor _dev$ + Fix 6: use CompareVersions for semver comparison
	if theirVersion and not theirVersion:find("_dev$") and maxVersion then
		if self:CompareVersions(theirVersion, maxVersion) < 0 then
			return "outdated"
		end
	end

	return "synced"
end

-- Fix 4: Module-level INACTIVE_THRESHOLD constant.
-- Single source of truth: StandingsFrame reads CommunicationV2.INACTIVE_THRESHOLD.
CommunicationV2.INACTIVE_THRESHOLD = 90  -- seconds (15s heartbeat × 6 misses)

--[[=============================================
	Canonical Hash Functions
	
	These produce deterministic hashes from game state.
	PR and Attendance% are EXCLUDED (derived fields — §1.6).
	Class, lastSeen, lifetimeEarned/Spent are EXCLUDED (display-only).
===============================================]]

--- Compute a deterministic hash of all player data.
-- Iterates players sorted by name. Each player contributes a canonical line.
-- Returns a hex string (CRC32-style via simple polynomial hash).
function CommunicationV2:ComputeRosterHash()
	-- Frame-level cache: avoid recomputing within same heartbeat cycle
	local now = GetTime()
	if self._rosterHashCache and self._rosterHashCacheAt == now then
		return self._rosterHashCache, self._rosterHashPlayerCount
	end

	-- CRITICAL: Use same filtered player set as BuildSnapshot/GetFilteredPlayersForSync.
	-- This ensures the hash matches the payload (excludes synthetics).
	local players = Cerebro:GetFilteredPlayersForSync()
	local baseGP = (Cerebro.db.global.config.epgp and Cerebro.db.global.config.epgp.baseGP) or 200

	-- Collect and sort participant names for deterministic order.
	-- PARTICIPANT FILTER: Only players with real EPGP data drive the hash.
	-- Ghost Pending stubs (EP=0, GP=baseGP, status=Pending) are excluded so
	-- different clients holding different ghost records cannot produce a permanent
	-- hash mismatch. Non-participants still travel in snapshot payloads for UI display.
	local names = {}
	for name, pData in pairs(players) do
		if Cerebro:IsParticipant(pData, baseGP) then
			names[#names + 1] = name
		end
	end
	table.sort(names)

	-- Build canonical string
	local parts = {}
	for _, name in ipairs(names) do
		local p = players[name]
		-- Canonical line: deterministic field order, coerced to safe defaults
		-- Fields included: ep, gp, status, lastUpdate
		-- Fields EXCLUDED: pr (derived), class (display), lastSeen (display),
		--                  lifetimeEarned/Spent (display), role (display),
		--                  bossKillsPresent (display), attendance (display),
		--                  eligibleBossKills (derived)
		-- NOTE: Attendance fields excluded because eligibleBossKills diverges across
		--       clients after resets, causing permanent hash mismatches +
		--       infinite sync push loops. They still sync via snapshot payloads.
		parts[#parts + 1] = string.format("%s:%.0f:%.0f:%s:%.0f",
			name,
			p.ep or 0,
			p.gp or 0,
			p.status or "Pending",
			p.lastUpdate or 0
		)
	end

	-- Include active session ID in hash so session start/end triggers divergence detection
	local sessionID = Cerebro.db.global.currentSession and Cerebro.db.global.currentSession.sessionID or "NONE"
	parts[#parts + 1] = "session:" .. sessionID

	-- NOTE: selection state (customEP.lastUpdate) is intentionally excluded from the roster hash.
	-- Selection has its own LWW merge path (MergeSelectionData) and its timestamp legitimately
	-- differs between officers (each officer may have a different active selection).
	-- Including it here caused permanent officer↔officer hash divergence and infinite FETCH loops.

	local canonical = table.concat(parts, "|")
	local hash = self:SimpleHash(canonical)

	-- Cache result for this frame
	self._rosterHashCache = hash
	self._rosterHashPlayerCount = #names
	self._rosterHashCacheAt = now

	return hash, #names
end

--- Compute a deterministic hash of synced config fields.
-- Uses dynamic recursive serialization so ALL config fields are covered
-- automatically. No maintenance needed when adding new settings.
--
-- EXCLUDED from hash (local-only, not synced):
--   debug, debugLevel, ui, configVersion, lastSeenConfigVersion,
--   configLastUpdate (merge-only timestamp, not content)
--
-- NOTE: If features like signupStatuses or epSignUp are re-implemented
-- in a future version, they will be automatically included via the
-- dynamic serializer. No code change needed.
function CommunicationV2:ComputeConfigHash()
	-- Frame-level cache: avoid recomputing within same heartbeat cycle
	local now = GetTime()
	if self._configHashCache and self._configHashCacheAt == now then
		return self._configHashCache
	end

	local config = Cerebro.db.global.config or {}

	-- Fields that are local-only and must NOT be hashed (top-level sections)
	local EXCLUDED_KEYS = {
		debug = true, debugLevel = true, ui = true,
		configVersion = true, lastSeenConfigVersion = true,
		configLastUpdate = true,  -- merge-only LWW timestamp
	}

	-- Fields excluded at ANY nesting depth (merge-metadata, not content).
	-- `lastUpdate` inside config.sync (or any future section) is a LWW timestamp
	-- used only by MergeConfigData — it must NOT contribute to the content hash.
	-- Without this exclusion: two officers with identical settings but different
	-- save timestamps produce different hashes → permanent config ping-pong.
	local EXCLUDED_NESTED_KEYS = {
		lastUpdate = true,
	}

	local parts = {}

	-- Recursively serialize a table with sorted keys for deterministic output.
	-- Coerces all keys to strings and sorts them to ensure identical ordering
	-- regardless of Lua table iteration order.
	local function serialize(tbl, prefix)
		local keys = {}
		for k in pairs(tbl) do
			if not EXCLUDED_NESTED_KEYS[k] then
				keys[#keys + 1] = tostring(k)
			end
		end
		table.sort(keys)
		for _, k in ipairs(keys) do
			-- Look up with both string and number key (AceComm may deliver
			-- table keys as numbers even when stored as strings)
			local v = tbl[k]
			if v == nil then v = tbl[tonumber(k)] end
			local fullKey = prefix .. "." .. k
			if type(v) == "table" then
				serialize(v, fullKey)
			elseif type(v) == "number" then
				-- Use %.6g to handle both integers and decimals deterministically
				parts[#parts + 1] = string.format("%s=%.6g", fullKey, v)
			else
				parts[#parts + 1] = fullKey .. "=" .. tostring(v)
			end
		end
	end

	-- Hash only synced config sections (exclude local-only: debug, ui, etc.)
	-- IMPORTANT: Must sort section names for deterministic output across clients.
	-- pairs() iteration order is non-deterministic in Lua.
	local sections = {}
	for section, value in pairs(config) do
		if type(value) == "table" and not EXCLUDED_KEYS[section] then
			sections[#sections + 1] = section
		end
	end
	table.sort(sections)
	for _, section in ipairs(sections) do
		serialize(config[section], section)
	end

	local canonical = table.concat(parts, "|")
	local hash = self:SimpleHash(canonical)

	-- Cache result for this frame
	self._configHashCache = hash
	self._configHashCacheAt = now

	return hash
end

--- Simple polynomial hash function (DJB2 variant).
-- Returns a hex string. Deterministic across clients.
-- Not cryptographic — purely for state comparison.
function CommunicationV2:SimpleHash(str)
	local hash = 5381
	for i = 1, #str do
		hash = ((hash * 33) + string.byte(str, i)) % 4294967296  -- mod 2^32
	end
	return string.format("%08x", hash)
end

--[[=============================================
	Startup & Heartbeat
===============================================]]

function CommunicationV2:OnStartup()
	-- On login/reload, if officer, send an initial heartbeat
	if Cerebro:IsOfficerSafe() then
		self:SendHeartbeat()
	else
		-- Non-officers: listen passively. If no heartbeat received after 15s,
		-- request sync via STATE_FETCH to guild
		self:ScheduleTimer("MaybeRequestInitialSync", 15)
	end
end

function CommunicationV2:MaybeRequestInitialSync()
	local s = Cerebro.db.global.syncStateV2 or {}
	if s.lastSnapshotReceived and s.lastSnapshotReceived > 0 then
		return  -- Already received data from an officer
	end

	-- No officer heartbeat received — try to fetch
	if IsInGuild() then
		Cerebro:Debug("V2: No heartbeat received, broadcasting STATE_FETCH to guild")
		self:SendFetch(nil, "GUILD")
	end
end

-- RETIRED (v1.0.3): OnHeartbeatTick removed — SyncScheduler:OnHeartbeat() is the
-- single heartbeat timer owner with combat-gating, quarantine guard, and post-combat flush.
-- function CommunicationV2:OnHeartbeatTick()
-- 	if not Cerebro:IsOfficerSafe() then return end
-- 	if Cerebro.modules.SyncScheduler and not Cerebro.modules.SyncScheduler:ShouldSync() then
-- 		return
-- 	end
-- 	if Cerebro.modules.Communication and Cerebro.modules.Communication:IsQuarantined() then
-- 		return
-- 	end
-- 	self:SendHeartbeat()
-- end

--[[=============================================
	STATE_HEARTBEAT — Hash broadcast
	
	Officers broadcast their state hashes periodically.
	All clients compare and trigger STATE_FETCH if divergent.
===============================================]]

function CommunicationV2:SendHeartbeat()
	local rosterHash, playerCount = self:ComputeRosterHash()
	local configHash = self:ComputeConfigHash()

	local payload = {
		epoch = Cerebro.db.global.resetEpoch or 0,
		rosterHash = rosterHash,
		configHash = configHash,
		playerCount = playerCount,
		totalBossKills = Cerebro.db.global.totalBossKills or 0,
		version = Cerebro.version,
		timestamp = GetServerTime and GetServerTime() or time(),
	}

	-- Send to appropriate channel
	-- LE_PARTY_CATEGORY_HOME excludes LFR/instance groups where the RAID addon channel is blocked
	if IsInRaid(LE_PARTY_CATEGORY_HOME) then
		self:SendV2Message(MSG_V2.STATE_HEARTBEAT, payload, "RAID")
	elseif IsInGuild() then
		self:SendV2Message(MSG_V2.STATE_HEARTBEAT, payload, "GUILD")
	end

	local s = Cerebro.db.global.syncStateV2 or {}
	s.lastHeartbeatSent = time()
	Cerebro.db.global.syncStateV2 = s

	Cerebro:Debug(string.format("V2 HB sent: roster=%s config=%s players=%d",
		rosterHash, configHash, playerCount))
end

--- Build C: Fire an immediate heartbeat after an officer state mutation.
-- Debounced (2s) to coalesce rapid sequential changes (e.g., batch EP award)
-- into a single heartbeat. Much cheaper than SyncToRaid (full snapshot).
-- Call this after: admit, purge, block, EP/GP award, session start/end, decay, reset.
function CommunicationV2:NotifyStateChange(reason)
	if not Cerebro:IsOfficerSafe() then return end

	-- Debounce: cancel any pending notification and schedule a new one in 2s
	if self._notifyTimer then
		self:CancelTimer(self._notifyTimer)
	end
	self._notifyTimer = self:ScheduleTimer(function()
		self._notifyTimer = nil
		self:SendHeartbeat()
		Cerebro:Debug("V2 immediate HB (state change: " .. (reason or "?") .. ")")
	end, 2)
end

--- Print a detailed hash breakdown for cross-client debugging.
-- Strictly read-only diagnostic — no mutations, no side effects.
function CommunicationV2:DebugHash()
	-- Roster hash breakdown
	local rosterHash, playerCount = self:ComputeRosterHash()
	local configHash = self:ComputeConfigHash()
	local sessionID = Cerebro.db.global.currentSession
		and Cerebro.db.global.currentSession.sessionID or "NONE"

	Cerebro:Print("=== V2 Hash Diagnostic ===")
	Cerebro:Print(string.format("  |cff00ccffRoster Hash:|r %s  |cff00ccffConfig Hash:|r %s", rosterHash, configHash))
	Cerebro:Print(string.format("  |cff00ccffPlayers:|r %d  |cff00ccffSession:|r %s", playerCount, sessionID))
	Cerebro:Print(string.format("  |cff00ccffEpoch:|r %d  |cff00ccffBossKills:|r %d",
		Cerebro.db.global.resetEpoch or 0, Cerebro.db.global.totalBossKills or 0))

	-- Player canonical lines (first 5 + last 5 for comparison)
	local players = Cerebro:GetFilteredPlayersForSync()
	local names = {}
	for name in pairs(players) do
		names[#names + 1] = name
	end
	table.sort(names)

	local function formatLine(idx)
		local name = names[idx]
		local p = players[name]
		return string.format("  [%d] %s: ep=%.0f gp=%.0f status=%s lastUpdate=%.0f",
			idx, Cerebro:GetShortPlayerName(name),
			p.ep or 0, p.gp or 0, p.status or "Pending", p.lastUpdate or 0)
	end

	local shown = math.min(5, #names)
	Cerebro:Print(string.format("  --- First %d players ---", shown))
	for i = 1, shown do
		Cerebro:Print(formatLine(i))
	end

	if #names > 10 then
		Cerebro:Print(string.format("  ... (%d more) ...", #names - 10))
	end

	if #names > 5 then
		local lastStart = math.max(shown + 1, #names - 4)
		Cerebro:Print(string.format("  --- Last %d players ---", #names - lastStart + 1))
		for i = lastStart, #names do
			Cerebro:Print(formatLine(i))
		end
	end

	-- Peer hash comparison (from last heartbeat exchange)
	local s = Cerebro.db.global.syncStateV2 or {}
	local peerHashes = s.peerHashes or {}
	local peerCount = 0
	for _ in pairs(peerHashes) do peerCount = peerCount + 1 end

	if peerCount > 0 then
		Cerebro:Print(string.format("  --- Known Peers (%d) ---", peerCount))
		for peer, info in pairs(peerHashes) do
			local rMatch = (info.rosterHash == rosterHash) and "|cff00ff00MATCH|r" or "|cffff4444DIFFER|r"
			local cMatch = (info.configHash == configHash) and "|cff00ff00MATCH|r" or "|cffff4444DIFFER|r"
			Cerebro:Print(string.format("  %s: roster=%s (%s) config=%s (%s) players=%s",
				Cerebro:GetShortPlayerName(peer),
				info.rosterHash or "?", rMatch,
				info.configHash or "?", cMatch,
				tostring(info.playerCount or "?")))
		end
	end

	Cerebro:Print("=========================")
end

--[[=============================================
	EPOCH MISMATCH HANDLER
	
	Shared helper for HandleHeartbeat, HandleSnapshot, HandleFetch.
	Mirrors V1 HandleSyncData epoch logic:
	  - payload < local: reject (stale)
	  - payload > local + officer: backup, wipe, accept
	  - payload > local + non-officer: quarantine
	  - payload == local: proceed
===============================================]]

function CommunicationV2:HandleEpochMismatch(sender, payloadEpoch, context)
	local localEpoch = Cerebro.db.global.resetEpoch or 0
	if payloadEpoch == localEpoch then return true end

	if payloadEpoch < localEpoch then
		Cerebro:Debug(string.format("V2 %s: stale epoch from %s (theirs=%d < mine=%d)",
			context, sender, payloadEpoch, localEpoch))
		return false
	end

	-- payloadEpoch > localEpoch: epoch upgrade needed
	local senderIsOfficer = Cerebro:IsOfficer(sender)
	if senderIsOfficer then
		-- Auto-backup before wipe (same pattern as V1 HandleSyncData)
		Cerebro.db.global.backup = {
			timestamp = time(),
			version = Cerebro.version,
			players = Cerebro:CopyTable(Cerebro.db.global.players),
			transactions = Cerebro:CopyTable(Cerebro.db.global.transactions),
			resetEpoch = localEpoch,
		}

		-- Accept epoch upgrade + wipe local data
		Cerebro.db.global.players = {}
		Cerebro.db.global.transactions = {}
		Cerebro.db.global.awards = {}
		Cerebro.db.global.history = {}
		Cerebro.db.global.seasonBossKills = 0
		Cerebro.db.global.totalBossKills = 0
		Cerebro.db.global.currentSession = nil
		Cerebro.db.global.resetEpoch = payloadEpoch
		Cerebro.db.global.epochQuarantine = nil

		-- Reset receipt cursor state
		Cerebro:ResetReceiptCursorState("V2 epoch upgrade")

		if Cerebro.Ledger then
			Cerebro.Ledger:AddAuditLog("EPOCH_UPGRADE",
				string.format("V2 %s from %s: epoch %d -> %d",
					context, sender, localEpoch, payloadEpoch))
		end

		Cerebro:Print(string.format(
			"|cffff8800[Epoch Upgrade]|r Accepted epoch %d from %s (was %d). Data reset.",
			payloadEpoch, sender, localEpoch))

		-- Immediately request full data from the officer who triggered the upgrade.
		-- Without this, the client has empty player tables and shows "Pending" 
		-- until the next heartbeat cycle detects hash mismatch (up to 3+ minutes).
		Cerebro:ScheduleTimer(function()
			if self.SendFetch then
				local whisperTarget = Cerebro:GetShortPlayerName(sender)
				self:SendFetch(whisperTarget, "WHISPER")
				Cerebro:Debug("V2 epoch upgrade: auto-fetch sent to " .. whisperTarget)
			end
		end, 1)  -- 1s delay: let the officer's reset finish before requesting

		-- Refresh UI to reflect the epoch wipe. This may trigger GetPlayerData()
		-- auto-creation with status="Pending" for the local player, but the
		-- post-merge self-correction in HandleSnapshot (line ~720) fixes that
		-- when the officer's snapshot arrives with the correct status.
		Cerebro:RefreshAllUI()
		return true
	else
		-- Non-officer: enter quarantine until officer confirms
		Cerebro.db.global.epochQuarantine = {
			at = time(),
			expectedEpoch = payloadEpoch,
			triggeredBy = sender,
		}
		Cerebro:Debug(string.format("V2 %s: epoch quarantine (raider %s has epoch %d, mine=%d)",
			context, sender, payloadEpoch, localEpoch))
		return false
	end
end

function CommunicationV2:HandleHeartbeat(sender, data)
	-- Receive-side dedup: skip if we processed a HB from this sender in the last 5 seconds.
	-- Multiple SendHeartbeat() callers (timer, SyncToRaid, session-start, force broadcast)
	-- can fire close together, creating duplicate messages with identical hash data.
	local HB_DEDUP_WINDOW = 5
	self._lastHBReceived = self._lastHBReceived or {}
	local dedupNow = time()
	local lastSeen = self._lastHBReceived[sender] or 0
	if (dedupNow - lastSeen) < HB_DEDUP_WINDOW then
		return -- Already processed recently, hash data is identical
	end
	self._lastHBReceived[sender] = dedupNow
	if not data then return end

	-- Self-exclusion
	local myName = Cerebro:GetPlayerFullName()
	if sender == myName or Cerebro:GetShortPlayerName(sender) == Cerebro:GetShortPlayerName(myName) then
		return
	end

	-- Epoch check: upgrade or reject via shared handler
	if data.epoch then
		if not self:HandleEpochMismatch(sender, data.epoch, "HB") then return end
	end

	-- Track peer hash (Fix 4: also store epoch so CheckSeasonChange can detect
	-- if a peer officer already performed the season reset before we fire ours)
	local s = Cerebro.db.global.syncStateV2 or {}
	s.peerHashes = s.peerHashes or {}
	s.peerHashes[sender] = {
		rosterHash = data.rosterHash,
		configHash = data.configHash,
		playerCount = data.playerCount,
		epoch = data.epoch or 0,
		lastSeen = time(),
	}

	-- Addon status tracking: record last seen timestamp and version per peer.
	-- Stored in syncStateV2.addonStatus (NOT in db.global.players) so it
	-- cannot affect the roster/config hash by design.
	-- IMPORTANT: AceComm passes `sender` as short name on same-realm (e.g. "Mysterthunt").
	-- Normalize to full Name-Realm format so addonStatus keys match db.global.players keys.
	local fullSender = Cerebro:GetFullPlayerName(sender) or sender
	s.addonStatus = s.addonStatus or {}
	s.addonStatus[fullSender] = {
		lastSeenAt = time(),
		lastKnownVersion = data.version,
	}
	-- Fix 11: Invalidate max-version cache whenever addonStatus changes
	self._cachedMaxVersion = nil
	-- Fix 7: Invalidate online-member cache on any heartbeat (peer may have come online)
	self._onlineGuildCache = nil
	self._onlineGuildCacheAt = nil

	-- Fix 1: Real-time reactivity with combat guard.
	-- Detect addon state transitions and refresh UI, but NEVER during combat lockdown
	-- (heartbeats arrive mid-encounter; RefreshAllUI() causes hitching on boss pulls).
	if Cerebro:IsOfficerSafe() then
		self._lastKnownAddonState = self._lastKnownAddonState or {}
		local newState = self:GetAddonState(fullSender)
		if self._lastKnownAddonState[fullSender] ~= newState then
			self._lastKnownAddonState[fullSender] = newState
			-- Debounce: coalesce rapid multi-peer HB bursts into a single redraw
			if self._addonStateRefreshTimer then
				self:CancelTimer(self._addonStateRefreshTimer)
			end
			self._addonStateRefreshTimer = self:ScheduleTimer(function()
				self._addonStateRefreshTimer = nil
				-- Fix 1: If in combat, defer refresh until post-combat flush
				if InCombatLockdown() then
					self._pendingAddonStateRefresh = true
					Cerebro:Debug("V2 addon state change for " .. fullSender .. " deferred (combat lockdown)")
					return
				end
				Cerebro:RefreshAllUI()
				Cerebro:Debug("V2 addon state change for " .. fullSender .. ": " .. newState .. " → UI refresh")
			end, 1)
		end
	end

	Cerebro.db.global.syncStateV2 = s

	-- Echo our hashes back to the sender (so synccheck and GetSyncStatus work)
	-- Only echo if this is NOT already an echo (prevent infinite loop)
	-- Rate-limited: max once per peer per 90 seconds to avoid O(N²) message storms
	local ECHO_COOLDOWN = 90
	if not data.isEcho then
		self._lastEchoSent = self._lastEchoSent or {}
		local now = time()
		local lastEcho = self._lastEchoSent[sender] or 0
		if (now - lastEcho) >= ECHO_COOLDOWN then
			self._lastEchoSent[sender] = now
			local myRosterHash, myPlayerCount = self:ComputeRosterHash()
			local myConfigHash = self:ComputeConfigHash()
			local echoPayload = {
				epoch = Cerebro.db.global.resetEpoch or 0,
				rosterHash = myRosterHash,
				configHash = myConfigHash,
				playerCount = myPlayerCount,
				version = Cerebro.version,
				timestamp = GetServerTime and GetServerTime() or time(),
				isEcho = true,
			}
			-- CRITICAL FIX: Retain full name for SendV2Message target (so IsPlayerReachable works)
			-- SendV2Message will strip it to short name right before handing off to AceComm WHISPER
			local printTarget = Cerebro:GetShortPlayerName(sender)
			self:SendV2Message(MSG_V2.STATE_HEARTBEAT, echoPayload, "WHISPER", sender)
			-- Diagnostic trace: user-visible with /cb debug simple
			Cerebro:DebugSimple(string.format("V2 echo → %s", printTarget))
			Cerebro:Debug(string.format("V2 HB echo sent to %s (roster=%s config=%s)",
				printTarget, myRosterHash, myConfigHash))
		end
	else
		-- Diagnostic trace: received an echo from a peer
		Cerebro:DebugSimple(string.format("V2 echo ← %s", Cerebro:GetShortPlayerName(sender)))
	end

	-- Compare hashes
	local myRosterHash = self:ComputeRosterHash()
	local myConfigHash = self:ComputeConfigHash()

	local rosterMatch = (data.rosterHash == myRosterHash)
	local configMatch = (data.configHash == myConfigHash)

	if rosterMatch and configMatch then
		-- Hashes match — we're in sync
		s.isStale = false
		-- Reset backoff counter on convergence
		self._fetchRetries = self._fetchRetries or {}
		self._fetchRetries[sender] = nil
		Cerebro:Debug(string.format("V2 HB from %s: MATCH (roster=%s config=%s)",
			sender, data.rosterHash, data.configHash))
	else
		-- Hashes differ — request full snapshot (with backoff)
		s.isStale = true

		-- Backoff: track consecutive mismatches per peer
		-- Prevent rapid retries from echo storms (e.g., getting 3 mismatched HBs in 5 seconds)
		-- by silently dropping mismatches if we already sent a fetch/push within 15 seconds.
		local now = time()
		local lastFetch = self._lastFetchTo and self._lastFetchTo[sender] or 0
		local lastPush = self._lastSnapshotTo and self._lastSnapshotTo[sender] or 0
		if (now - lastFetch < 15) or (now - lastPush < 15) then
			-- Diagnostic trace (optional): Cerebro:Debug("V2 HB MISMATCH: retries paused for cooldown")
			return
		end

		self._fetchRetries = self._fetchRetries or {}
		self._fetchRetryHashes = self._fetchRetryHashes or {}
		local retries = self._fetchRetries[sender] or 0
		local prevHash = self._fetchRetryHashes[sender]

		-- Reset retries if peer's hash changed (they got a partial update)
		if prevHash and prevHash ~= data.rosterHash then
			retries = 0
		end
		self._fetchRetryHashes[sender] = data.rosterHash

		-- After 3 consecutive failures, back off — but auto-reset after 5 minutes
		local MAX_RETRIES_BEFORE_BACKOFF = 3
		local BACKOFF_RESET_SECONDS = 300 -- 5 minutes
		if retries >= MAX_RETRIES_BEFORE_BACKOFF then
			-- Check if backoff TTL has expired → auto-reset
			self._fetchRetryBackoffAt = self._fetchRetryBackoffAt or {}
			if not self._fetchRetryBackoffAt[sender] then
				self._fetchRetryBackoffAt[sender] = now
			end
			if (now - self._fetchRetryBackoffAt[sender]) >= BACKOFF_RESET_SECONDS then
				-- TTL expired — reset and allow re-attempt
				self._fetchRetries[sender] = 0
				self._fetchRetryBackoffAt[sender] = nil
				self._fetchRetryHashes[sender] = nil
				Cerebro:Debug(string.format("V2 HB from %s: backoff reset (TTL expired after %ds)",
					sender, BACKOFF_RESET_SECONDS))
				-- Fall through to normal mismatch handling below
				retries = 0
			else
				Cerebro:Debug(string.format("V2 HB from %s: MISMATCH (suppressed, %d retries — retry in %ds)",
					sender, retries, BACKOFF_RESET_SECONDS - (now - self._fetchRetryBackoffAt[sender])))
				return
			end
		end

		self._fetchRetries[sender] = retries + 1
		Cerebro:Debug(string.format("V2 HB from %s: MISMATCH (roster:%s/%s config:%s/%s) → %s (attempt %d/%d)",
			sender,
			data.rosterHash, myRosterHash,
			data.configHash, myConfigHash,
			Cerebro:IsOfficer(sender) and "FETCH" or "PUSH",
			retries + 1, MAX_RETRIES_BEFORE_BACKOFF))
		Cerebro:DebugSimple(string.format("Sync: data mismatch with %s, %s...",
			Cerebro:GetShortPlayerName(sender) or sender,
			Cerebro:IsOfficer(sender) and "requesting update" or "pushing data"))

		if Cerebro:IsOfficer(sender) then
			-- Officer-to-officer: bidirectional exchange
			-- FETCH their data (they may have updates we lack)
			-- Option B: Jitter 0-8s to stagger simultaneous fetches from multiple officers
			local jitterB1 = math.random(0, 8)
			Cerebro:ScheduleTimer(function()
				self:SendFetch(sender, "WHISPER")
			end, jitterB1)
			-- PUSH our data (we may have updates they lack, e.g. season reset)
			now = time()
			self._lastSnapshotTo = self._lastSnapshotTo or {}
			if not self._lastSnapshotTo[sender] or (now - self._lastSnapshotTo[sender]) >= 30 then
				self._lastSnapshotTo[sender] = now
				local snapshot = self:BuildSnapshot(true, true)
				snapshot.epoch = Cerebro.db.global.resetEpoch or 0
				snapshot.version = Cerebro.version
				snapshot.sender = Cerebro:GetPlayerFullName()
				snapshot.timestamp = GetServerTime and GetServerTime() or time()
				self:SendV2Message(MSG_V2.STATE_SNAPSHOT, snapshot, "WHISPER", sender)
				Cerebro:Debug(string.format("V2 officer exchange: PUSH snapshot to %s",
					Cerebro:GetShortPlayerName(sender)))
			end
		elseif Cerebro:IsOfficerSafe() then
			-- OPTION A: Officer-to-raider — whisper only the divergent peer.
			-- Previously this was a GUILD/RAID broadcast, which disrupted all other
			-- already-synced officers every time any single stale raider logged in.
			-- A targeted WHISPER is sufficient: the raider gets the same data,
			-- and synced peers are left completely undisturbed.
			now = time()
			self._lastSnapshotTo = self._lastSnapshotTo or {}
			if not self._lastSnapshotTo[sender] or (now - self._lastSnapshotTo[sender]) >= 30 then
				self._lastSnapshotTo[sender] = now
				local snapshot = self:BuildSnapshot(true, true)
				snapshot.epoch = Cerebro.db.global.resetEpoch or 0
				snapshot.version = Cerebro.version
				snapshot.sender = Cerebro:GetPlayerFullName()
				snapshot.timestamp = GetServerTime and GetServerTime() or time()
				self:SendV2Message(MSG_V2.STATE_SNAPSHOT, snapshot, "WHISPER", sender)
				Cerebro:Debug(string.format("V2 PUSH whisper to %s (triggered by mismatch, targeted)",
					Cerebro:GetShortPlayerName(sender)))
			else
				Cerebro:Debug(string.format("V2 PUSH to %s skipped (cooldown, %ds ago)",
					Cerebro:GetShortPlayerName(sender), now - self._lastSnapshotTo[sender]))
			end
		else
			-- Non-officer receiving mismatch from non-officer: request from officers.
			-- Option B: Jitter 0-8s to stagger the thundering-herd of simultaneous FETCHes
			-- from 20 raiders all detecting the same hash mismatch at the same moment.
			local jitterB2 = math.random(0, 8)
			Cerebro:ScheduleTimer(function()
				self:SendFetch(sender, "WHISPER")
			end, jitterB2)
		end
	end
end

--[[=============================================
	STATE_FETCH — Request full state from officer
===============================================]]

function CommunicationV2:SendFetch(target, channel)
	-- Per-peer cooldown: avoid sending multiple FETCHes to the same peer within 30s
	-- Only applies to targeted (WHISPER) fetches, not broadcast fetches (target=nil)
	if target then
		local now = time()
		self._lastFetchTo = self._lastFetchTo or {}
		if self._lastFetchTo[target] and (now - self._lastFetchTo[target]) < 30 then
			Cerebro:Debug(string.format("V2 FETCH to %s skipped (cooldown, %ds ago)",
				target, now - self._lastFetchTo[target]))
			return
		end
		self._lastFetchTo[target] = now
	end

	local myRosterHash = self:ComputeRosterHash()
	local myConfigHash = self:ComputeConfigHash()

	local payload = {
		epoch = Cerebro.db.global.resetEpoch or 0,
		rosterHash = myRosterHash,
		configHash = myConfigHash,
		version = Cerebro.version,
	}

	channel = channel or "WHISPER"
	self:SendV2Message(MSG_V2.STATE_FETCH, payload, channel, target)

	Cerebro:Debug(string.format("V2 FETCH sent to %s via %s",
		target or "all", channel))
end

function CommunicationV2:HandleFetch(sender, data)
	if not data then return end

	-- Only officers respond to fetch requests
	if not Cerebro:IsOfficerSafe() then return end

	-- OPTION E: Sync Leader gate — only the elected leader responds when inside a raid.
	-- Other officers passively receive and merge the resulting snapshot.
	-- Outside of a raid (guild-only, solo officer), IsSyncLeader() always returns true,
	-- so all officers respond normally with no change in behavior.
	if IsInRaid() and Cerebro.modules.SyncScheduler
		and not Cerebro.modules.SyncScheduler:IsSyncLeader() then
		Cerebro:Debug("V2 FETCH from " .. sender .. ": deferring to Sync Leader")
		return
	end

	-- Self-exclusion
	local myName = Cerebro:GetPlayerFullName()
	if sender == myName or Cerebro:GetShortPlayerName(sender) == Cerebro:GetShortPlayerName(myName) then return end

	-- Fresh-install suppression: don't send authoritative snapshots with empty roster
	-- This prevents a fresh-install officer from wiping raider data
	local myPlayerCount = select(2, self:ComputeRosterHash())
	if myPlayerCount == 0 then
		Cerebro:Debug("V2 FETCH: skipping response (empty roster, likely fresh install)")
		return
	end

	-- Per-peer cooldown: avoid sending duplicate SNAPSHOTs to the same peer within 30s
	local now = time()
	self._lastSnapshotTo = self._lastSnapshotTo or {}
	if self._lastSnapshotTo[sender] and (now - self._lastSnapshotTo[sender]) < 30 then
		Cerebro:Debug(string.format("V2 SNAPSHOT to %s skipped (cooldown, %ds ago)",
			sender, now - self._lastSnapshotTo[sender]))
		return
	end

	-- Epoch check: upgrade or reject via shared handler
	if data.epoch then
		if not self:HandleEpochMismatch(sender, data.epoch, "FETCH") then return end
	end

	-- Quarantine guard
	if Cerebro.modules.Communication and Cerebro.modules.Communication:IsQuarantined() then
		Cerebro:Debug("V2 FETCH blocked: quarantine active")
		return
	end

	-- Determine what to send (both, roster only, or config only)
	local myRosterHash = self:ComputeRosterHash()
	local myConfigHash = self:ComputeConfigHash()

	local includeRoster = (data.rosterHash ~= myRosterHash)
	-- Always include config when sending roster to ensure receiver has matching
	-- configuration for consistent behavior.
	local includeConfig = (data.configHash ~= myConfigHash) or includeRoster

	-- If both match, send nothing (peer is already in sync)
	if not includeRoster and not includeConfig then
		Cerebro:Debug("V2 FETCH from " .. sender .. ": already in sync, no snapshot needed")
		return
	end

	-- Record cooldown timestamp
	self._lastSnapshotTo[sender] = now

	-- Build and send snapshot
	local snapshot = self:BuildSnapshot(includeRoster, includeConfig)
	self:SendV2Message(MSG_V2.STATE_SNAPSHOT, snapshot, "WHISPER", sender)

	Cerebro:Debug(string.format("V2 SNAPSHOT sent to %s (roster=%s config=%s)",
		sender, tostring(includeRoster), tostring(includeConfig)))
end

--[[=============================================
	STATE_SNAPSHOT — Full state payload
===============================================]]

function CommunicationV2:BuildSnapshot(includeRoster, includeConfig)
	local snapshot = {
		version = Cerebro.version,
		epoch = Cerebro.db.global.resetEpoch or 0,
		scope = Cerebro.db.global.scope,
		totalBossKills = Cerebro.db.global.totalBossKills or 0,
		timestamp = GetServerTime and GetServerTime() or time(),
		-- Fix 2: carry the sender's lastKnownSeasonID so the receiver can update
		-- db.global.lastKnownSeasonID before CheckSeasonChange() fires, preventing
		-- a duplicate season reset when the 60s timer eventually fires.
		lastKnownSeasonID = Cerebro.db.global.lastKnownSeasonID,
		includes = {
			roster = includeRoster or false,
			config = includeConfig or false,
		},
	}

	if includeRoster then
		-- Filter out synthetic test players
		snapshot.players = Cerebro:GetFilteredPlayersForSync()
		snapshot.purgedPlayers = Cerebro.db.global.purgedPlayers or {}
		snapshot.blockedPlayers = Cerebro.db.global.blockedPlayers or {}
		snapshot.currentSession = Cerebro.db.global.currentSession
		snapshot.customEP = Cerebro.db.global.customEP
		snapshot.signupAwards = Cerebro:CopyTable(Cerebro.db.global.signupAwards or {})

		-- Include hash for verification on receive side
		snapshot.rosterHash = self:ComputeRosterHash()
	end

	if includeConfig then
		snapshot.config = Cerebro:CopyTable(Cerebro.db.global.config)
		snapshot.configHash = self:ComputeConfigHash()
	end

	return snapshot
end

function CommunicationV2:HandleSnapshot(sender, data)
	if not data then return end

	-- Self-exclusion
	local myName = Cerebro:GetPlayerFullName()
	if sender == myName or Cerebro:GetShortPlayerName(sender) == Cerebro:GetShortPlayerName(myName) then return end

	-- Epoch check: upgrade or reject via shared handler
	if data.epoch then
		if not self:HandleEpochMismatch(sender, data.epoch, "SNAP") then return end
	end

	-- Quarantine guard
	if Cerebro.modules.Communication and Cerebro.modules.Communication:IsQuarantined() then
		Cerebro:Debug("V2 SNAPSHOT blocked: quarantine active")
		return
	end

	Cerebro:Debug(string.format("V2 SNAPSHOT received from %s (roster=%s config=%s)",
		sender,
		tostring(data.includes and data.includes.roster),
		tostring(data.includes and data.includes.config)))

	-- CRITICAL ORDER: Purge/block ledger merges FIRST (zombie prevention)
	-- SECURITY: Only accept purge/block data from officers. Non-officers should
	-- never send snapshots (all send paths are officer-guarded), but this is
	-- defense-in-depth to prevent stale purge data from non-officer clients.
	local senderIsOfficer = Cerebro:IsOfficer(sender)

	if senderIsOfficer and data.purgedPlayers then
		Cerebro.db.global.purgedPlayers = Cerebro.db.global.purgedPlayers or {}
		for name, purgeTime in pairs(data.purgedPlayers) do
			-- Normalize realm name (strip spaces to match API format)
			local normalized = Cerebro:GetFullPlayerName(name) or name
			local localTime = Cerebro.db.global.purgedPlayers[normalized] or 0
			if purgeTime > localTime then
				Cerebro.db.global.purgedPlayers[normalized] = purgeTime
			end
		end
	elseif data.purgedPlayers then
		Cerebro:Debug("V2 SNAPSHOT: ignoring purge data from non-officer " .. tostring(sender))
	end

	-- Merge blocked players (union — keep latest timestamp per entry)
	if senderIsOfficer and data.blockedPlayers then
		Cerebro.db.global.blockedPlayers = Cerebro.db.global.blockedPlayers or {}
		for name, blockTime in pairs(data.blockedPlayers) do
			-- Normalize realm name (strip spaces to match API format)
			local normalized = Cerebro:GetFullPlayerName(name) or name
			local localTime = Cerebro.db.global.blockedPlayers[normalized] or 0
			if blockTime > localTime then
				Cerebro.db.global.blockedPlayers[normalized] = blockTime
			end
		end
	elseif data.blockedPlayers then
		Cerebro:Debug("V2 SNAPSHOT: ignoring block data from non-officer " .. tostring(sender))
	end

	-- Zombie cleanup: delete local player records now covered by purge/block directives.
	-- Without this, a purge directive arrives but the existing local record persists,
	-- causing permanent participant count divergence between officers.
	for name in pairs(Cerebro.db.global.purgedPlayers or {}) do
		if Cerebro.db.global.players[name] then
			Cerebro.db.global.players[name] = nil
			Cerebro:Debug("Zombie cleanup: removed purged player " .. name)
		end
	end
	for name in pairs(Cerebro.db.global.blockedPlayers or {}) do
		if Cerebro.db.global.players[name] then
			Cerebro.db.global.players[name] = nil
			Cerebro:Debug("Zombie cleanup: removed blocked player " .. name)
		end
	end

	-- Merge roster data (LWW on lastUpdate)
	local rosterMerged = 0
	if data.includes and data.includes.roster and data.players then
		rosterMerged = self:MergeRosterData(data.players, sender)
		-- Also merge selection state (LWW on customEP.lastUpdate) — officer only
		if senderIsOfficer and data.customEP then
			self:MergeSelectionData(data.customEP, sender)
		elseif data.customEP then
			Cerebro:Debug("V2 SNAPSHOT: ignoring selection data from non-officer " .. tostring(sender))
		end
	end

	-- Merge config data — officer only
	if senderIsOfficer and data.includes and data.includes.config and data.config then
		self:MergeConfigData(data.config, sender)
	elseif data.includes and data.includes.config and data.config then
		Cerebro:Debug("V2 SNAPSHOT: ignoring config data from non-officer " .. tostring(sender))
	end

	-- Merge signupAwards (union merge: keep latest per eventKey) — officer only
	if senderIsOfficer and data.signupAwards then
		self:MergeSignupAwards(data.signupAwards, sender)
	elseif data.signupAwards then
		Cerebro:Debug("V2 SNAPSHOT: ignoring signupAwards from non-officer " .. tostring(sender))
	end

	-- Update sync metadata
	if data.totalBossKills ~= nil then
		local localKills = Cerebro.db.global.totalBossKills or 0
		-- Officers are authoritative: always accept their totalBossKills
		-- Non-officers: only accept higher values (monotonic)
		if Cerebro:IsOfficer(sender) then
			Cerebro.db.global.totalBossKills = data.totalBossKills
		elseif data.totalBossKills > localKills then
			Cerebro.db.global.totalBossKills = data.totalBossKills
		end
	end

	-- Update session state from snapshot
	-- Sessions are SHARED STATE between officers — all officers should see the same
	-- active session. When sessions disagree, use LWW (startedAt) to pick the newer one.
	-- RULES:
	--   0. Non-officers ignore session data entirely.
	--   1. Only accept session state from officers.
	--   2. Don't re-inject a session we just ended (stale snapshot race, 60s window).
	--   3. When sessions differ, newer startedAt wins (LWW).
	--   4. If peer has nil session and we have one, clear ours (session was ended).
	if data.includes and data.includes.roster and Cerebro:IsOfficerSafe() then
		local incomingSession = data.currentSession
		local localSession = Cerebro.db.global.currentSession
		local sessionMod = Cerebro:GetModule("Session", true)
		local justEnded = sessionMod and sessionMod._lastEndedAt
			and (time() - sessionMod._lastEndedAt) < 60

		if not Cerebro:IsOfficer(sender) then
			Cerebro:Debug("V2 SNAPSHOT: ignoring session from non-officer " .. tostring(sender))
		elseif justEnded and incomingSession
			and sessionMod._lastEndedSessionID == incomingSession.sessionID then
			-- Stale snapshot: peer still had the session we just ended. Ignore.
			Cerebro:Debug("V2 SNAPSHOT: ignoring stale session re-injection ("
				.. tostring(incomingSession.sessionID) .. ") - ended " .. (time() - sessionMod._lastEndedAt) .. "s ago")
		elseif incomingSession == nil and localSession then
			-- Officer peer has no session but we do — session was ended.
			-- Guard: don't clear if WE just started (our session is newer than peer's snapshot)
			local localStart = localSession.startedAt or 0
			local snapshotTime = data.timestamp or 0
			if snapshotTime > localStart then
				Cerebro:Debug("V2 SNAPSHOT: clearing local session (peer " .. tostring(sender) .. " has no session)")
				Cerebro.db.global.currentSession = nil
				Cerebro:RefreshAllUI()
			else
				Cerebro:Debug("V2 SNAPSHOT: keeping local session (started after peer's snapshot)")
			end
		elseif incomingSession and incomingSession.sessionID then
			if not localSession or not localSession.sessionID then
				-- No local session — adopt incoming
				Cerebro.db.global.currentSession = incomingSession
				Cerebro:Debug("V2 SNAPSHOT: adopted session from officer " .. tostring(sender))
				Cerebro:RefreshAllUI()
			elseif incomingSession.sessionID ~= localSession.sessionID then
				-- Different sessions — LWW by startedAt
				local incomingStart = incomingSession.startedAt or 0
				local localStart = localSession.startedAt or 0
				if incomingStart > localStart then
					Cerebro.db.global.currentSession = incomingSession
					Cerebro:Debug("V2 SNAPSHOT: replaced session (incoming newer: " .. incomingStart .. " > " .. localStart .. ")")
					Cerebro:RefreshAllUI()
				else
					Cerebro:Debug("V2 SNAPSHOT: keeping local session (local newer: " .. localStart .. " >= " .. incomingStart .. ")")
				end
			-- else: same sessionID — already in sync, do nothing
			end
		end
	end

	-- Scope propagation (bootstrapping)
	if data.scope and not Cerebro.db.global.scope then
		Cerebro.db.global.scope = data.scope
		Cerebro:Debug("V2: Scope received from " .. sender)
	end

	-- Fix 2: Adopt lastKnownSeasonID from officer snapshot (monotonic merge).
	-- This ensures that when CheckSeasonChange() fires (either via early cancel
	-- or the 60s fallback), the currentID == storedID guard fires correctly and
	-- suppresses a duplicate season reset.
	if senderIsOfficer and data.lastKnownSeasonID and
	   data.lastKnownSeasonID > (Cerebro.db.global.lastKnownSeasonID or 0) then
		Cerebro.db.global.lastKnownSeasonID = data.lastKnownSeasonID
		Cerebro:Debug(string.format("V2 SNAPSHOT: adopted lastKnownSeasonID %d from %s",
			data.lastKnownSeasonID, sender))
	end

	-- Fix 1 cancel: snapshot arrived — cancel the 60s season check timer and
	-- fire CheckSeasonChange() immediately. Since lastKnownSeasonID is now
	-- current (adopted above), the change-detection guard returns early safely.
	if Cerebro._seasonCheckTimer then
		Cerebro:CancelTimer(Cerebro._seasonCheckTimer, true)
		Cerebro._seasonCheckTimer = nil
		Cerebro:Debug("V2 SNAPSHOT: cancelled season check timer, firing immediately")
		Cerebro:CheckSeasonChange()
	end

	-- Recalculate derived fields
	Cerebro:RecalculateAllPR()

	-- Invalidate hash cache: merge mutated db.global.players and config,
	-- so any hashes cached earlier in this frame (e.g. from a heartbeat
	-- processed moments before) are stale. Without this, the integrity
	-- check below and SNAPSHOT_ACK may report wrong hashes when a heartbeat
	-- and snapshot are deserialized in the same WoW frame.
	self:InvalidateHashCache()

	-- Clear stale flag
	local s = Cerebro.db.global.syncStateV2 or {}
	s.isStale = false
	s.lastSnapshotReceived = time()
	Cerebro.db.global.syncStateV2 = s

	-- Update v1 sync status for /cb status compatibility
	Cerebro.db.global.syncStatus = Cerebro.db.global.syncStatus or {}
	Cerebro.db.global.syncStatus.lastOfficerSync = time()

	-- Post-merge self-correction: if our record was in the incoming snapshot
	-- with a non-Pending status, trust the officer's status over any locally
	-- auto-created Pending record (race condition during epoch upgrades).
	local me = Cerebro:GetPlayerFullName()
	local myData = Cerebro.db.global.players[me]
	if myData and myData.status == "Pending" and data.players and data.players[me] then
		local officerStatus = data.players[me].status
		if officerStatus and officerStatus ~= "Pending" then
			myData.status = officerStatus
			myData.lastUpdate = data.players[me].lastUpdate or myData.lastUpdate
			Cerebro:Debug("V2 post-merge self-correction: " .. me .. " status -> " .. officerStatus)
		end
	end

	-- Refresh UI
	Cerebro:RefreshAllUI()

	-- Post-merge integrity check: verify local state matches snapshot's expected hashes
	-- If the merge produced different state (LWW race, partial delivery, etc.),
	-- immediately schedule a re-fetch from the sender to converge.
	local integrityOk = true
	if data.rosterHash then
		local postMergeRosterHash = self:ComputeRosterHash()
		if postMergeRosterHash ~= data.rosterHash then
			Cerebro:Debug(string.format("V2 SNAPSHOT integrity FAILED: roster expected=%s got=%s (from %s)",
				data.rosterHash, postMergeRosterHash, sender))
			integrityOk = false
		end
	end
	-- NOTE: Config hash intentionally excluded from integrity check.
	-- MergeConfigData uses per-section LWW — the receiver legitimately keeps
	-- its own newer values, making the sender's hash an unreliable baseline.
	-- Config convergence is detected by the heartbeat system instead.
	if not integrityOk then
		-- Only re-fetch if LWW rejected everything (merged=0, skipped>0).
		-- This means the receiver already has newer data for all players AND
		-- may be missing data the sender has in other fields — pulling is useful.
		--
		-- When data was accepted (merged>0), the hash mismatch is structural:
		-- version difference, participant filter, or session state — none of which
		-- re-fetching can resolve. The heartbeat will detect and repair persistent
		-- divergence on its next cycle.
		local incomingPlayerCount = data.players and Cerebro:TableCount(data.players) or 0
		if rosterMerged == 0 and incomingPlayerCount > 0 then
			local fetchSender = sender  -- capture for closure
			Cerebro:ScheduleTimer(function()
				self:SendFetch(fetchSender, "WHISPER")
			end, 1)
		else
			Cerebro:Debug(string.format("V2 integrity FAILED but %d merged — heartbeat will reconcile", rosterMerged))
		end
	end

	-- SNAPSHOT_ACK: notify sender of our post-merge state so they can detect delivery failure
	local postRosterHash = self:ComputeRosterHash()
	local postConfigHash = self:ComputeConfigHash()
	-- CRITICAL FIX: Pass full sender name (not short name) for cross-realm compatibility.
	-- SendV2Message strips to short name AFTER the IsPlayerReachable check.
	self:SendRPC("SNAPSHOT_ACK", {
		rosterHash = postRosterHash,
		configHash = postConfigHash,
		converged = integrityOk,
	}, "WHISPER", sender)

	Cerebro:Debug("V2 SNAPSHOT merged from " .. sender .. " (integrity: " .. tostring(integrityOk) .. ")")
	Cerebro:DebugSimple(string.format("Sync complete from %s",
		Cerebro:GetShortPlayerName(sender) or sender))
end

--[[=============================================
	Roster & Config Merge (LWW)
===============================================]]

--- Valid player data fields for sync merge whitelist.
-- IMPORTANT: If you add a new field to player data, you MUST add it here
-- or it will be silently dropped during sync merge.
local VALID_PLAYER_FIELDS = {
	ep = true, gp = true, pr = true, class = true, status = true,
	lastUpdate = true, lastSeen = true, coreAdmittedAt = true,
	bossKillsPresent = true, eligibleBossKills = true,
	attendance = true, fieldRev = true, rev = true, createdEpoch = true,
}

--- Merge incoming player data using Last-Write-Wins on lastUpdate.
function CommunicationV2:MergeRosterData(incomingPlayers, sender)
	local localPlayers = Cerebro.db.global.players
	local purged = Cerebro.db.global.purgedPlayers or {}
	local blocked = Cerebro.db.global.blockedPlayers or {}
	local mergedCount = 0
	local skippedCount = 0
	local senderIsOfficer = Cerebro:IsOfficer(sender)
	local localIsOfficer = Cerebro:IsOfficerSafe()
	local baseGP = (Cerebro.db.global.config.epgp and Cerebro.db.global.config.epgp.baseGP) or 200

	for rawName, incoming in pairs(incomingPlayers) do
		-- Normalize realm name (strip spaces to match API format)
		local name = Cerebro:GetFullPlayerName(rawName) or rawName
		-- Skip non-table entries (data corruption guard)
		if type(incoming) ~= "table" then
			Cerebro:Debug("MergeRosterData: skipping corrupt entry for " .. tostring(name) .. " (type=" .. type(incoming) .. ")")
		-- Skip entries that aren't valid Name-Realm format (fake entries from corruption)
		elseif not name:find("-", 1, true) then
			Cerebro:Debug("MergeRosterData: skipping non-player key: " .. tostring(name))
		-- Skip purged players (zombie prevention)
		elseif purged[name] then
			skippedCount = skippedCount + 1
		-- Skip blocked players (corruption prevention)
		elseif blocked[name] then
			skippedCount = skippedCount + 1
		else
			local local_ = localPlayers[name]
			if not local_ then
				-- New player — accept from officers always.
				-- Non-officers can only inject Core players (not Pending stubs).
				-- This lets raiders propagate real Core data while preventing
				-- Pending stub pollution from random Cerebro clients.
				local incomingStatus = incoming.status or "Pending"
				if not senderIsOfficer and incomingStatus ~= "Core" then
					skippedCount = skippedCount + 1
				else
					local clean = {}
					for k, v in pairs(incoming) do
						if VALID_PLAYER_FIELDS[k] then clean[k] = v end
					end
					-- Ensure class is populated (try guild roster if missing)
					if not clean.class then
						pcall(function()
							for gi = 1, GetNumGuildMembers() do
								local gn, _, _, _, _, _, _, _, _, _, classFileName = GetGuildRosterInfo(gi)
								if gn and Cerebro:GetFullPlayerName(gn) == name then
									clean.class = classFileName
									break
								end
							end
						end)
					end
					localPlayers[name] = clean
					mergedCount = mergedCount + 1
				end
			else
				-- Existing player — LWW on lastUpdate
				-- Officer-aware tie-breaking: officers win timestamp ties (>=)
				-- because some fields (e.g. eligibleBossKills) are updated
				-- without bumping lastUpdate, creating permanent LWW ties.
				-- Non-officer senders use strict (>) to prevent raider→raider loops.
				local incomingTime = incoming.lastUpdate or 0
				local localTime = local_.lastUpdate or 0
				local accepted = false

				if incomingTime > localTime or (senderIsOfficer and incomingTime >= localTime) then
					accepted = true
				elseif senderIsOfficer and not localIsOfficer then
					-- RAIDER STUB OVERRIDE: if the local record was auto-created
					-- by GetPlayerData() (Pending with default EP/GP), it has a
					-- newer lastUpdate than the officer's real data. Accept the
					-- officer's version regardless of timestamp to fix the stale stub.
					-- Records with real EP/GP or non-Pending status are preserved
					-- under normal LWW to protect multi-officer scenarios.
					local isStub = (local_.status == "Pending" and (local_.ep or 0) == 0
						and (local_.gp or baseGP) == baseGP)
					if isStub then
						accepted = true
						Cerebro:Debug(string.format(
							"V2 merge: overriding stub for %s (local=%d officer=%d)",
							name, localTime, incomingTime))
					end
				end

				if accepted then
					for k, v in pairs(incoming) do
						if VALID_PLAYER_FIELDS[k] then
							-- PROTECT: never downgrade Core→Pending via sync.
							-- Status downgrades should only happen via explicit
							-- officer action (purge/block), not stale sync data.
							if k == "status" and local_[k] == "Core" and v == "Pending" then
								Cerebro:Debug("V2 merge: blocked Core→Pending downgrade for " .. name)
							-- PROTECT: preserve coreAdmittedAt if local has it
							elseif k == "coreAdmittedAt" and local_[k] and not v then
								Cerebro:Debug("V2 merge: preserved local coreAdmittedAt for " .. name)
							else
								local_[k] = v
							end
						end
					end
					mergedCount = mergedCount + 1
				else
					skippedCount = skippedCount + 1
				end
			end
		end
	end

	Cerebro:Debug(string.format("V2 roster merge from %s: %d merged, %d skipped",
		sender, mergedCount, skippedCount))

	-- Reconciliation: remove local players not in the officer's snapshot.
	-- Two paths based on local client role:
	--   RAIDER receiving from OFFICER: officer is authoritative — adopt their
	--     player set exactly (remove any local extra, except self).
	--   OFFICER receiving from OFFICER: both sides authoritative — only remove
	--     players NOT in snapshot AND NOT in guild roster (safety net).
	--
	-- DISABLED (v0.11.34): Reconciliation was deleting local Core players when
	-- the incoming officer's snapshot didn't include them. In multi-officer setups
	-- with different auto-discovery results, this caused catastrophic data loss.
	-- Players should only be removed via explicit purge/block, never via sync.
	-- Sync is now purely additive: merge updates, never delete.
	--
	-- Original behavior:
	--   RAIDER from OFFICER: remove participants not in snapshot
	--   OFFICER from OFFICER: remove only if also not in guild roster
	-- Both paths caused legitimate Core players to vanish.
	if false and Cerebro:IsOfficer(sender) then
		local myName = Cerebro:GetPlayerFullName()
		local removedCount = 0
		-- baseGP already declared at function scope (line 965) — reused here.

		if not localIsOfficer then
			-- RAIDER PATH: officer's snapshot is authoritative.
			-- Remove participants absent from snapshot; leave non-participants alone.
			for name, pData in pairs(localPlayers) do
				if not incomingPlayers[name] and name ~= myName
					and Cerebro:IsParticipant(pData, baseGP)
				then
					localPlayers[name] = nil
					removedCount = removedCount + 1
					Cerebro:Debug("V2 reconcile (raider): removed participant " .. name)
				end
			end
		else
			-- OFFICER PATH: guild roster safety net (both sides authoritative).
			-- Only run when guild data is available AND only delete participants.
			if GetNumGuildMembers() > 0 then
				local guildNames = {}
				for i = 1, GetNumGuildMembers() do
					local gn = GetGuildRosterInfo(i)
					if gn then guildNames[Cerebro:GetFullPlayerName(gn)] = true end
				end

				for name, pData in pairs(localPlayers) do
					if not incomingPlayers[name]
						and not guildNames[name]
						and name ~= myName
						and not Cerebro:IsPlayerInRaid(name)
						and Cerebro:IsParticipant(pData, baseGP)
					then
						localPlayers[name] = nil
						removedCount = removedCount + 1
						Cerebro:Debug("V2 reconcile (officer): removed phantom participant " .. name)
					end
				end
			end
		end

		if removedCount > 0 then
			Cerebro:Debug(string.format(
				"V2 reconcile: removed %d players after merge from %s (%s path)",
				removedCount, sender, localIsOfficer and "officer" or "raider"))
		end
	end

	return mergedCount, skippedCount
end

--- Merge incoming config data.
-- Config uses a simple "officer wins" model — always accept from officer.
function CommunicationV2:MergeConfigData(incomingConfig, sender)
	if not incomingConfig then return end

	-- Officers are authoritative for config
	-- Simply overwrite local config with incoming
	local localConfig = Cerebro.db.global.config

	-- Preserve local-only settings (debug, debugLevel, ui preferences)
	local preserveDebug = localConfig.debug
	local preserveDebugLevel = localConfig.debugLevel
	local preserveUI = localConfig.ui and Cerebro:CopyTable(localConfig.ui)

	-- COMPAT: v0.10.23 sent configLastUpdate as a single number (Unix timestamp).
	-- v0.10.24+ uses a table {epgp=ts, prep=ts, sync=ts, admin=ts} for per-section LWW.
	-- Convert legacy number to per-section table so config from upgrading officers propagates.
	-- Sanitize timestamps: detect millisecond-precision values (> 10 billion)
	-- and convert back to seconds. Prevents integer overflow in string.format
	-- and stops corrupted ms-timestamps from propagating via sync.
	local function sanitizeTimestamp(ts)
		if type(ts) ~= "number" then return ts end
		if ts > 10000000000 then  -- > year 2286 in seconds, but normal range for ms
			return math.floor(ts / 1000)
		end
		return ts
	end

	local incomingTimes
	if type(incomingConfig.configLastUpdate) == "table" then
		incomingTimes = incomingConfig.configLastUpdate
		-- Sanitize each section timestamp
		for k, v in pairs(incomingTimes) do
			incomingTimes[k] = sanitizeTimestamp(v)
		end
	elseif type(incomingConfig.configLastUpdate) == "number" then
		-- Legacy: spread single timestamp across all sections
		local ts = sanitizeTimestamp(incomingConfig.configLastUpdate)
		incomingTimes = { epgp = ts, prep = ts, sync = ts, admin = ts }
	else
		incomingTimes = {}
	end
	local localTimes
	if type(localConfig.configLastUpdate) == "table" then
		localTimes = localConfig.configLastUpdate
		for k, v in pairs(localTimes) do
			localTimes[k] = sanitizeTimestamp(v)
		end
	elseif type(localConfig.configLastUpdate) == "number" then
		local ts = sanitizeTimestamp(localConfig.configLastUpdate)
		localTimes = { epgp = ts, prep = ts, sync = ts, admin = ts }
	else
		localTimes = {}
	end

	-- Merge EPGP section
	local inEpgpTime = incomingTimes.epgp or 0
	local myEpgpTime = localTimes.epgp or 0
	if inEpgpTime > myEpgpTime or (inEpgpTime == 0 and myEpgpTime == 0) then
		if incomingConfig.epgp then localConfig.epgp = incomingConfig.epgp end
		localTimes.epgp = inEpgpTime
		Cerebro:Debug(string.format("V2 config merge: accepted epgp (theirs=%.0f, mine=%.0f)", inEpgpTime, myEpgpTime))
	else
		Cerebro:Debug(string.format("V2 config merge: kept local epgp (mine=%.0f >= theirs=%.0f)", myEpgpTime, inEpgpTime))
	end

	-- Merge Prep section
	local inPrepTime = incomingTimes.prep or 0
	local myPrepTime = localTimes.prep or 0
	if inPrepTime > myPrepTime or (inPrepTime == 0 and myPrepTime == 0) then
		if incomingConfig.prep then localConfig.prep = incomingConfig.prep end
		localTimes.prep = inPrepTime
		Cerebro:Debug(string.format("V2 config merge: accepted prep (theirs=%.0f, mine=%.0f)", inPrepTime, myPrepTime))
	else
		Cerebro:Debug(string.format("V2 config merge: kept local prep (mine=%.0f >= theirs=%.0f)", myPrepTime, inPrepTime))
	end

	localConfig.configLastUpdate = localTimes

	-- FIX A (merge portion): Merge sync settings using LWW on config.sync.lastUpdate.
	-- The previous "preserve if empty" guard caused permanent oscillation when
	-- two officers had different visibleRanks — neither would accept the other's.
	-- Now the most recently modified sync config wins unconditionally.
	if incomingConfig.sync then
		local incomingTime = incomingConfig.sync.lastUpdate or 0
		local localTime = (localConfig.sync and localConfig.sync.lastUpdate) or 0

		if incomingTime > localTime then
			-- Incoming is newer — accept wholesale (includes visibleRanks, syncLeaderRanks)
			localConfig.sync = incomingConfig.sync
			Cerebro:Debug(string.format("V2 config merge: accepted incoming sync config (theirs=%.0f > mine=%.0f)",
				incomingTime, localTime))
		elseif incomingTime == 0 and localTime == 0 then
			-- Legacy fallback: neither side has a timestamp (pre-LWW clients).
			-- Accept incoming (officer-authoritative) to bootstrap convergence.
			localConfig.sync = incomingConfig.sync
			Cerebro:Debug("V2 config merge: legacy sync config accepted (no timestamps)")
		else
			-- Local is newer or equal — keep our sync config
			Cerebro:Debug(string.format("V2 config merge: kept local sync config (mine=%.0f >= theirs=%.0f)",
				localTime, incomingTime))
		end
	end

	-- Merge admin settings
	if incomingConfig.admin then
		localConfig.admin = incomingConfig.admin
	end

	-- Restore local-only settings
	localConfig.debug = preserveDebug
	localConfig.debugLevel = preserveDebugLevel
	if preserveUI then
		localConfig.ui = preserveUI
	end

	Cerebro:Debug("V2 config merged from " .. sender)
end

--- Merge incoming selection state (LWW on lastUpdate).
function CommunicationV2:MergeSelectionData(incoming, sender)
	if not incoming or not incoming.lastUpdate then return end

	Cerebro.db.global.customEP = Cerebro.db.global.customEP or { selectedPlayers = {}, lastUpdate = 0 }
	local local_ = Cerebro.db.global.customEP
	local incomingTime = incoming.lastUpdate or 0
	local localTime = local_.lastUpdate or 0

	-- LWW: Officer wins timestamp ties
	local senderIsOfficer = Cerebro:IsOfficer(sender)
	if incomingTime > localTime or (senderIsOfficer and incomingTime >= localTime) then
		-- Only merge selection state — preserve local templates
		local localTemplates = local_.templates
		local_.selectedPlayers = Cerebro:CopyTable(incoming.selectedPlayers or {})
		local_.lastUpdate = incoming.lastUpdate
		local_.templates = localTemplates  -- Keep our templates intact
		Cerebro:Debug(string.format("V2 selection merge from %s: ACCEPTED (theirs=%.0f, mine=%.0f)", 
			sender, incomingTime, localTime))
		
		-- Notification: Ensure StandingsFrame updates its in-memory 'selected' map
		local sf = Cerebro:GetModule("StandingsFrame", true)
		if sf and sf.SetSelection then
			local names = {}
			for nm in pairs(incoming.selectedPlayers or {}) do
				names[#names + 1] = nm
			end
			sf:SetSelection(names)
			-- FIX: SetSelection() calls SaveSelection() which bumps lastUpdate = time(),
			-- ratcheting our timestamp forward on every accepted merge and permanently
			-- blocking future peers from being accepted (theirs < mine forever).
			-- Restore the correct incoming timestamp that was already written to local_.
			Cerebro.db.global.customEP.lastUpdate = incoming.lastUpdate
		end
	else
		Cerebro:Debug(string.format("V2 selection merge from %s: REJECTED (theirs=%.0f, mine=%.0f)", 
			sender, incomingTime, localTime))
	end
end


--- Merge signupAwards deduplication state (union merge).
-- Each key is an event identifier (e.g. "2026-03-10|RaidName").
-- Value is a table { awardedAt = timestamp, awardedBy = fullName }.
-- We keep the entry with the latest awardedAt timestamp per key.
function CommunicationV2:MergeSignupAwards(incoming, sender)
	if not incoming or type(incoming) ~= "table" then return end

	Cerebro.db.global.signupAwards = Cerebro.db.global.signupAwards or {}
	local local_ = Cerebro.db.global.signupAwards
	local mergeCount = 0

	for eventKey, award in pairs(incoming) do
		if type(award) == "table" then
			local localAward = local_[eventKey]
			local incomingAt = award.awardedAt or 0
			local localAt = (localAward and localAward.awardedAt) or 0

			if incomingAt > localAt then
				local_[eventKey] = Cerebro:CopyTable(award)
				mergeCount = mergeCount + 1
			end
		end
	end

	if mergeCount > 0 then
		Cerebro:Debug(string.format("V2 signupAwards merge from %s: %d entries updated", sender, mergeCount))
	end
end

--[[=============================================
	RPC_CONTROL — Non-persistent action dispatch
	
	RPC methods carry action commands that don't affect
	the canonical state hash. Dispatched via table lookup.
===============================================]]

-- Forward declaration of dispatch table
local RPC_DISPATCH = {}

function CommunicationV2:SendRPC(method, data, channel, target)
	local payload = {
		method = method,
		data = data or {},
		sender = Cerebro:GetPlayerFullName(),
		epoch = Cerebro.db.global.resetEpoch or 0,
		timestamp = GetServerTime and GetServerTime() or time(),
	}

	channel = channel or "RAID"
	self:SendV2Message(MSG_V2.RPC_CONTROL, payload, channel, target)

	Cerebro:Debug(string.format("V2 RPC sent: %s via %s", method, channel))
end

function CommunicationV2:HandleRPC(sender, payload)
	if not payload or not payload.method then
		Cerebro:Debug("V2 RPC: missing method from " .. sender)
		return
	end

	-- Self-exclusion
	local myName = Cerebro:GetPlayerFullName()
	if sender == myName or Cerebro:GetShortPlayerName(sender) == Cerebro:GetShortPlayerName(myName) then
		return
	end

	-- Epoch guard: reject RPCs from senders with stale epoch (prevents zombie data after reset)
	if payload.epoch then
		local localEpoch = Cerebro.db.global.resetEpoch or 0
		if payload.epoch < localEpoch then
			Cerebro:Debug(string.format("V2 RPC [%s] rejected: stale epoch %d < %d from %s",
				payload.method or "?", payload.epoch, localEpoch, sender))
			return
		end
	end

	local handler = RPC_DISPATCH[payload.method]
	if handler then
		-- pcall for safety — one bad RPC shouldn't crash the module
		local ok, err = pcall(handler, self, sender, payload.data)
		if not ok then
			Cerebro:Debug(string.format("V2 RPC error [%s]: %s", payload.method, tostring(err)))
		end
	else
		Cerebro:Debug("V2 RPC: unknown method '" .. payload.method .. "' from " .. sender)
	end
end

--[[=============================================
	RPC Method Handlers
	
	Each handler migrates logic from the corresponding v1
	message handler in Communication.lua. These are
	non-persistent actions — they don't change the canonical hash.
===============================================]]

RPC_DISPATCH["BOSS_KILL"] = function(self, sender, data)
	-- Delegate to v1 handler for now (gradual migration)
	if Cerebro.modules.Communication and Cerebro.modules.Communication.HandleBossKill then
		Cerebro.modules.Communication:HandleBossKill(data, sender)
	end
end

RPC_DISPATCH["SESSION_START"] = function(self, sender, data)
	-- Inline session adoption: create local session state directly from RPC data
	-- instead of delegating to V1 (which has its own epoch/format conflicts).
	if not data then return end

	-- Non-officers ignore session data entirely
	if not Cerebro:IsOfficerSafe() then return end

	-- Guard: only accept session commands from our guild's officers
	if not Cerebro:IsOfficer(sender) then
		Cerebro:Debug("SESSION_START from " .. sender .. " rejected (not a guild officer)")
		return
	end

	-- Replace existing session — a new SESSION_START from an officer supersedes
	-- any previous session. Without this, officers get stuck with stale sessions.
	local existing = Cerebro.db.global.currentSession
	if existing and existing.sessionID then
		Cerebro:Debug("SESSION_START from " .. tostring(sender) .. " replacing existing session: " .. tostring(existing.sessionID))
	end

	-- Build session record matching Session:StartSession() schema
	-- Critical: must include roster/bosses tables or EndSession will nil-crash
	Cerebro.db.global.currentSession = {
		sessionID = data.sessionID or ("remote_" .. time()),
		raidName = data.raidName or data.instanceName or "Remote Session",
		startTime = data.startTime or data.startedAt or time(),
		startedBy = data.startedBy or sender,
		endTime = nil,
		roster = {},
		bosses = {},
	}

	Cerebro:Debug("SESSION_START: adopted session " .. tostring(Cerebro.db.global.currentSession.sessionID) .. " from " .. tostring(sender))

	-- Log to History tab (local entry — transactions are not synced)
	local sessionRecord = Cerebro.db.global.currentSession
	if Cerebro.Ledger and Cerebro.Ledger.AddTransaction then
		Cerebro.Ledger:AddTransaction(
			"SESSION", 0, "EP", "SESSION_START",
			{ sessionInfo = string.format("[%s_%s] Session Started.",
				sessionRecord.raidName or "?", sessionRecord.sessionID) },
			false, { silent = true })
	end
	if Cerebro.Ledger and Cerebro.Ledger.AddAuditLog then
		Cerebro.Ledger:AddAuditLog("SESSION_START",
			string.format("[%s_%s] Session Started (by %s, adopted via V2 RPC).",
				sessionRecord.raidName or "?", sessionRecord.sessionID,
				Cerebro:GetShortPlayerName(sender)))
	end

	Cerebro:Print(string.format("|cff00ff00[Session]|r %s started a session.",
		Cerebro:GetShortPlayerName(sender)))
	Cerebro:RefreshAllUI()
end

RPC_DISPATCH["SESSION_END"] = function(self, sender, data)
	-- Non-officers ignore session data entirely
	if not Cerebro:IsOfficerSafe() then return end
	-- Guard: only accept session commands from our guild's officers
	if not Cerebro:IsOfficer(sender) then
		Cerebro:Debug("SESSION_END from " .. sender .. " rejected (not a guild officer)")
		return
	end

	-- Capture session info before HandleSessionEnd clears it
	local localSession = Cerebro.db.global.currentSession
	local sessionInfo = localSession and {
		raidName = localSession.raidName or "?",
		sessionID = localSession.sessionID or "?",
		startTime = localSession.startTime or 0,
	} or nil

	if Cerebro.modules.Communication and Cerebro.modules.Communication.HandleSessionEnd then
		Cerebro.modules.Communication:HandleSessionEnd(data, sender)
	end

	-- Log to History tab (local entry — transactions are not synced)
	if sessionInfo and Cerebro.Ledger and Cerebro.Ledger.AddTransaction then
		local duration = math.floor((time() - sessionInfo.startTime) / 60)
		Cerebro.Ledger:AddTransaction(
			"SESSION", 0, "EP", "SESSION_END",
			{ sessionInfo = string.format("[%s_%s] Session Ended. (Duration: %dm)",
				sessionInfo.raidName, sessionInfo.sessionID, duration) },
			false, { silent = true })
	end
	if sessionInfo and Cerebro.Ledger and Cerebro.Ledger.AddAuditLog then
		Cerebro.Ledger:AddAuditLog("SESSION_END",
			string.format("[%s_%s] Session Ended (by %s, received via V2 RPC).",
				sessionInfo.raidName, sessionInfo.sessionID,
				Cerebro:GetShortPlayerName(sender)))
	end
end

RPC_DISPATCH["PURGE"] = function(self, sender, data)
	-- RETIRED V1 DELEGATION (Build A) — handler now lives entirely in V2.
	-- Only accept from officers
	if not Cerebro:IsOfficer(sender) then
		Cerebro:Debug("V2 PURGE from non-officer " .. tostring(sender) .. " — ignored")
		return
	end

	local playerName = data and data.player
	if not playerName then return end

	-- Update purge ledger
	Cerebro.db.global.purgedPlayers = Cerebro.db.global.purgedPlayers or {}
	Cerebro.db.global.purgedPlayers[playerName] = data.timestamp or time()

	-- Delete from players if exists
	if Cerebro.db.global.players[playerName] then
		Cerebro.db.global.players[playerName] = nil
		Cerebro:Debug("V2 PURGE: removed " .. playerName .. " (from " .. tostring(sender) .. ")")
	end

	-- Refresh UI
	Cerebro:RefreshAllUI()
end

RPC_DISPATCH["BID_SUB"] = function(self, sender, data)
	if Cerebro.modules.Communication and Cerebro.modules.Communication.HandleBidSubmit then
		Cerebro.modules.Communication:HandleBidSubmit(data, sender)
	end
end

RPC_DISPATCH["BID_LOCK"] = function(self, sender, data)
	if Cerebro.modules.Communication and Cerebro.modules.Communication.HandleBidLock then
		Cerebro.modules.Communication:HandleBidLock(sender)
	end
end

RPC_DISPATCH["BID_UNLOCK"] = function(self, sender, data)
	if Cerebro.modules.Communication and Cerebro.modules.Communication.HandleBidUnlock then
		Cerebro.modules.Communication:HandleBidUnlock(sender)
	end
end

RPC_DISPATCH["ITEM_DROP"] = function(self, sender, data)
	if Cerebro.modules.Communication and Cerebro.modules.Communication.HandleItemDrop then
		Cerebro.modules.Communication:HandleItemDrop(data, sender)
	end
end

RPC_DISPATCH["ITEM_AWARD"] = function(self, sender, data)
	if Cerebro.modules.Communication and Cerebro.modules.Communication.HandleItemAward then
		Cerebro.modules.Communication:HandleItemAward(data, sender)
	end
end

RPC_DISPATCH["DECAY"] = function(self, sender, data)
	if Cerebro.modules.Communication and Cerebro.modules.Communication.HandleDecay then
		Cerebro.modules.Communication:HandleDecay(data, sender)
	end
end

RPC_DISPATCH["FACTORY_RESET"] = function(self, sender, data)
	if Cerebro.modules.Communication and Cerebro.modules.Communication.HandleReset then
		Cerebro.modules.Communication:HandleReset(data, sender)
	end
end

-- BroadcastReset() sends under key "RESET" — ensure both keys route to the same handler
RPC_DISPATCH["RESET"] = RPC_DISPATCH["FACTORY_RESET"]

RPC_DISPATCH["VERSION_CHECK"] = function(self, sender, data)
	if Cerebro.modules.Communication and Cerebro.modules.Communication.HandleVersionCheck then
		Cerebro.modules.Communication:HandleVersionCheck(data, sender)
	end
end

RPC_DISPATCH["ANNOUNCE"] = function(self, sender, data)
	if Cerebro.modules.Communication and Cerebro.modules.Communication.HandleAnnounceSelf then
		Cerebro.modules.Communication:HandleAnnounceSelf(data, sender)
	end
end

RPC_DISPATCH["DISCOVERY"] = function(self, sender, data)
	if Cerebro.modules.Communication and Cerebro.modules.Communication.HandleDiscoveryRequest then
		Cerebro.modules.Communication:HandleDiscoveryRequest(sender)
	end
end

RPC_DISPATCH["EPGP_UPDATE"] = function(self, sender, data)
	if Cerebro.modules.Communication and Cerebro.modules.Communication.HandleEPGPUpdate then
		Cerebro.modules.Communication:HandleEPGPUpdate(data, sender)
	end
end

RPC_DISPATCH["EPGP_FORWARD"] = function(self, sender, data)
	if Cerebro.modules.Communication and Cerebro.modules.Communication.HandleEPGPForward then
		Cerebro.modules.Communication:HandleEPGPForward(data, sender)
	end
end

-- Batch EPGP update: applies all awards in a single pass, sends BATCH_ACK back
RPC_DISPATCH["BATCH_EPGP_UPDATE"] = function(self, sender, data)
	if not Cerebro:IsOfficer(sender) then return end
	if not data or not data.awards then return end

	-- C1 fix: Quarantine guard — block ingestion during factory reset window
	if Cerebro.modules.Communication and Cerebro.modules.Communication:IsQuarantined() then
		Cerebro:Debug("BATCH_EPGP_UPDATE rejected: quarantine active from " .. sender)
		return
	end

	-- C3 fix: Dual-channel dedup — RAID+GUILD sends cause double delivery; skip if already processed
	local batchKey = (data.senderName or sender) .. ":" .. (data.timestamp or 0)
	if self._lastProcessedBatchKey == batchKey then
		Cerebro:Debug("BATCH_EPGP_UPDATE dedup: already processed batch from " .. sender)
		return
	end
	self._lastProcessedBatchKey = batchKey

	-- Epoch check: reject stale batch from pre-reset epoch
	if data.epoch and data.epoch < (Cerebro.db.global.resetEpoch or 0) then
		Cerebro:Debug("BATCH_EPGP_UPDATE rejected: stale epoch from " .. sender)
		return
	end

	-- Apply each award using existing handler logic (reuses security guards)
	local appliedCount = 0
	for _, award in ipairs(data.awards) do
		if Cerebro.modules.Communication and Cerebro.modules.Communication.HandleEPGPUpdate then
			-- Build data format expected by HandleEPGPUpdate
			local epgpData = {
				player = award.player,
				newBalance = award.newBalance,
				amount = award.amount,
				reason = award.reason,
				currency = award.currency,
				txId = award.txId,
				timestamp = data.timestamp,
				epoch = data.epoch,
			}
			Cerebro.modules.Communication:HandleEPGPUpdate(epgpData, sender)
			appliedCount = appliedCount + 1
		end
	end

	Cerebro:Debug(string.format("V2 BATCH_EPGP_UPDATE: applied %d/%d awards from %s",
		appliedCount, #data.awards, sender))

	-- Send BATCH_ACK back to originator via WHISPER
	if data.senderName then
		-- A1 fix: Invalidate hash cache so ACK carries post-award hash (not stale cached value)
		self:InvalidateHashCache()
		local rosterHash = self:ComputeRosterHash()
		self:SendRPC("BATCH_ACK", {
			rosterHash = rosterHash,
			appliedCount = appliedCount,
		}, "WHISPER", Cerebro:GetShortPlayerName(data.senderName) or data.senderName)
	end
end

-- BATCH_ACK: routed to convergence tracker on the Sync Leader
RPC_DISPATCH["BATCH_ACK"] = function(self, sender, data)
	-- C2 fix: Only officers (Sync Leader) need convergence tracking
	if not Cerebro:IsOfficerSafe() then return end
	if not data then return end
	self:OnBatchAck(sender, data)
end

-- AWARD_CONVERGENCE: display convergence report to officers
RPC_DISPATCH["AWARD_CONVERGENCE"] = function(_self, sender, data)
	if not Cerebro:IsOfficerSafe() then return end
	if not data then return end

	-- Don't print our own report (already printed locally)
	if sender == Cerebro:GetPlayerFullName() then return end
	if sender == (Cerebro:GetShortPlayerName(Cerebro:GetPlayerFullName())) then return end

	local color = (data.acked == data.total) and "|cff00ff00" or "|cffff8800"
	local status = (data.acked == data.total) and "[CONVERGED]" or "[PARTIAL]"
	local shortSender = Cerebro:GetShortPlayerName(sender) or sender
	Cerebro:Print(string.format("%s%s|r %d/%d synced in %.1fs (%d awards) — via %s",
		color, status, data.acked or 0, data.total or 0,
		data.elapsed or 0, data.awardCount or 0, shortSender))
end

RPC_DISPATCH["MPLUS_PUSH"] = function(_self, sender, data)
	-- Suppress M+ processing while in raid
	if IsInRaid() then return end
	-- Officer-only: store incoming M+ run in guild history
	if not Cerebro:IsOfficer() then return end
	if not data or not data.runKey then
		Cerebro:Debug("V2 MPLUS_PUSH: missing runKey from " .. sender)
		return
	end
	Cerebro.db.global.guildMplusRuns = Cerebro.db.global.guildMplusRuns or {}
	local runs = Cerebro.db.global.guildMplusRuns
	for _, r in ipairs(runs) do
		if r.runKey == data.runKey then return end  -- Dedupe
	end
	data.uploadedBy = sender
	data.receivedAt = time()
	table.insert(runs, data)
	Cerebro:Debug("V2 MPLUS_PUSH: stored run from " .. sender .. " (total: " .. #runs .. ")")
end

RPC_DISPATCH["MPLUS_INJECT"] = function(_self, sender, data)
	-- Officer-only: inject test M+ run into local history
	if not Cerebro:IsOfficer(sender) then return end
	if not data or not data.run then return end
	Cerebro.db.global.mplusRuns = Cerebro.db.global.mplusRuns or {}
	table.insert(Cerebro.db.global.mplusRuns, data.run)
	Cerebro:Debug("V2 MPLUS_INJECT: inserted run from " .. sender)
end

RPC_DISPATCH["RECEIPT_SYNC"] = function(self, sender, data)
	if Cerebro.modules.Communication and Cerebro.modules.Communication.HandleSyncReceipts then
		Cerebro.modules.Communication:HandleSyncReceipts(data, sender)
	end
end

RPC_DISPATCH["RECEIPT_REQUEST"] = function(self, sender, data)
	if Cerebro.modules.Communication and Cerebro.modules.Communication.HandleReceiptRequest then
		Cerebro.modules.Communication:HandleReceiptRequest(data, sender)
	end
end

RPC_DISPATCH["SYNC_AUDIT_REQ"] = function(self, sender, data)
	if Cerebro.modules.Communication and Cerebro.modules.Communication.HandleSyncAuditRequest then
		Cerebro.modules.Communication:HandleSyncAuditRequest(data, sender)
	end
end

RPC_DISPATCH["SYNC_AUDIT_RESP"] = function(self, sender, data)
	if Cerebro.modules.Communication and Cerebro.modules.Communication.HandleSyncAuditResponse then
		Cerebro.modules.Communication:HandleSyncAuditResponse(data, sender)
	end
end

RPC_DISPATCH["PLAYER_UPDATE"] = function(self, sender, data)
	if Cerebro.modules.Communication and Cerebro.modules.Communication.HandlePlayerUpdate then
		Cerebro.modules.Communication:HandlePlayerUpdate(data, sender)
	end
end

RPC_DISPATCH["PREP_REQUEST"] = function(self, sender, data)
	local prep = Cerebro:GetModule("Preparation", true)
	if prep and prep.HandlePrepRequest then
		prep:HandlePrepRequest(sender, data)
	end
end

RPC_DISPATCH["PREP_RESPONSE"] = function(self, sender, data)
	local prep = Cerebro:GetModule("Preparation", true)
	if prep and prep.HandlePrepResponse then
		prep:HandlePrepResponse(sender, data)
	end
end

RPC_DISPATCH["BROADCAST_TX"] = function(self, sender, data)
	if Cerebro.modules.Communication and Cerebro.modules.Communication.HandleBroadcastTransaction then
		Cerebro.modules.Communication:HandleBroadcastTransaction(data, sender)
	end
end

-- Batch version of BROADCAST_TX for retroactive /cb synchistory force-push
RPC_DISPATCH["BROADCAST_TX_BATCH"] = function(self, sender, data)
	if Cerebro.modules.Communication and Cerebro.modules.Communication.HandleBroadcastTransactionBatch then
		Cerebro.modules.Communication:HandleBroadcastTransactionBatch(data, sender)
	end
end

RPC_DISPATCH["BROADCAST_AUDIT"] = function(self, sender, data)
	if Cerebro.modules.Communication and Cerebro.modules.Communication.HandleBroadcastAuditLog then
		Cerebro.modules.Communication:HandleBroadcastAuditLog(data, sender)
	end
end

RPC_DISPATCH["SNAPSHOT_ACK"] = function(_self, sender, data)
	if not data then return end
	if data.converged then
		Cerebro:Debug(string.format("V2 SNAPSHOT_ACK from %s: CONVERGED", sender))
	else
		local myRosterHash = CommunicationV2:ComputeRosterHash()
		local myConfigHash = CommunicationV2:ComputeConfigHash()
		local rosterMatch = (data.rosterHash == myRosterHash)
		local configMatch = (data.configHash == myConfigHash)
		Cerebro:Debug(string.format("V2 SNAPSHOT_ACK from %s: DIVERGENT (roster:%s config:%s)",
			sender, tostring(rosterMatch), tostring(configMatch)))
		-- Next heartbeat cycle (30s) will detect the mismatch and re-push.
		-- No immediate action needed — the receiver already scheduled a re-fetch.
	end
end

--[[=============================================
	Session-Start Sync Barrier (Layer 1)
	
	On /cb startsession, officers broadcast a heartbeat
	and wait for convergence before proceeding.
===============================================]]

function CommunicationV2:OnSessionStart()
	if not Cerebro:IsOfficerSafe() then return end

	Cerebro:Debug("V2: Session-start sync barrier activated")

	-- Broadcast heartbeat immediately to detect divergence
	self:SendHeartbeat()

	-- Start convergence timer
	self._syncBarrierActive = true
	self._syncBarrierStart = time()
	self:ScheduleTimer("CheckSyncBarrier", SYNC_BARRIER_WINDOW)
end

function CommunicationV2:CheckSyncBarrier()
	self._syncBarrierActive = false

	-- Check if any peers are still stale
	local stalePeers = self:GetStalePeers()
	if #stalePeers == 0 then
		Cerebro:Debug("V2: Sync barrier complete — all peers converged")
		return
	end

	-- Some peers are stale — send targeted snapshots
	Cerebro:Debug(string.format("V2: Sync barrier expired — %d stale peers, sending targeted snapshots",
		#stalePeers))

	for _, peerName in ipairs(stalePeers) do
		local snapshot = self:BuildSnapshot(true, true)
		self:SendV2Message(MSG_V2.STATE_SNAPSHOT, snapshot, "WHISPER", peerName)
	end
end

--- Get list of peers whose hashes don't match ours.
function CommunicationV2:GetStalePeers()
	local stale = {}
	local s = Cerebro.db.global.syncStateV2 or {}
	local myRosterHash = self:ComputeRosterHash()
	local myConfigHash = self:ComputeConfigHash()
	local comm = Cerebro:GetModule("Communication", true)

	for name, peer in pairs(s.peerHashes or {}) do
		-- Only consider recently seen peers (within 2 heartbeat cycles)
		if peer.lastSeen and (time() - peer.lastSeen) < (HEARTBEAT_INTERVAL * 2) then
			if peer.rosterHash ~= myRosterHash or peer.configHash ~= myConfigHash then
				-- Skip offline peers (prevents whisper spam)
				if comm and comm.IsPlayerReachable and not comm:IsPlayerReachable(name) then
					Cerebro:Debug("GetStalePeers: skipping offline peer " .. name)
				else
					stale[#stale + 1] = name
				end
			end
		end
	end

	return stale
end

--[[=============================================
	V2 Message Transport
	
	Uses AceSerializer + AceComm on prefix "CerebroV2".
	Same pattern as v1 but with a different prefix.
===============================================]]

function CommunicationV2:SendV2Message(msgType, data, distribution, target)
	distribution = distribution or "RAID"

	-- Smart WHISPER guard: check reachability with FULL name, then normalize to short name
	if distribution == "WHISPER" and target then
		if Cerebro.modules.Communication and Cerebro.modules.Communication.IsPlayerReachable then
			if not Cerebro.modules.Communication:IsPlayerReachable(target) then
				-- Diagnostic trace: whisper blocked by reachability check
				local shortTarget = Cerebro:GetShortPlayerName(target) or target
				Cerebro:DebugSimple(string.format("V2 whisper blocked (offline): %s to %s", msgType, shortTarget))
				Cerebro:Debug("V2 SendMessage blocked: " .. msgType .. " to unreachable " .. target)
				return
			end
		end
		-- Normalize to short name for cross-realm compatibility (matches V1 pattern) AFTER checking
		target = Cerebro:GetShortPlayerName(target) or target
	end

	-- Serialize data
	local payload = Cerebro:Serialize(msgType, data)

	-- Send via AceComm on V2 prefix
	Cerebro:SendCommMessage(V2_PREFIX, payload, distribution, target, "NORMAL")

	Cerebro:Debug("V2 Sent " .. msgType .. " to " .. distribution)
end

--[[=============================================
	V2 Message Receive Handler
	
	Registered via Cerebro:RegisterComm("CerebroV2", "OnCommReceivedV2")
	This is called by AceComm when a CerebroV2-prefixed message arrives.
===============================================]]

function Cerebro:OnCommReceivedV2(prefix, message, distribution, sender)
	if prefix ~= V2_PREFIX then return end

	-- Deserialize
	local success, msgType, data = self:Deserialize(message)
	if not success then
		self:Debug("V2: Failed to deserialize from " .. sender)
		return
	end

	-- Route to handlers
	local v2 = self.modules.CommunicationV2
	if not v2 then return end

	if msgType == MSG_V2.STATE_HEARTBEAT then
		v2:HandleHeartbeat(sender, data)
	elseif msgType == MSG_V2.STATE_FETCH then
		v2:HandleFetch(sender, data)
	elseif msgType == MSG_V2.STATE_SNAPSHOT then
		v2:HandleSnapshot(sender, data)
	elseif msgType == MSG_V2.RPC_CONTROL then
		v2:HandleRPC(sender, data)
	else
		self:Debug("V2: Unknown message type: " .. tostring(msgType))
	end
end

--[[=============================================
	Public API for Other Modules
===============================================]]

--- Check if we're currently stale (know we're behind an officer).
function CommunicationV2:IsStale()
	local s = Cerebro.db.global.syncStateV2 or {}
	return s.isStale or false
end

--- Get the current sync status for UI display.
-- Returns: state, details
--   state: "synced" | "syncing" | "no_peers"
--   details: { peerCount = N, divergedCount = N }
function CommunicationV2:GetSyncStatus()
	local s = Cerebro.db.global.syncStateV2 or {}
	local now = time()
	local timeout = 120  -- Fixed 2-minute peer timeout (independent of heartbeat interval)

	-- Count recent peers and check hash matches
	local peerCount = 0
	local divergedCount = 0
	local myRosterHash = self:ComputeRosterHash()
	local myConfigHash = self:ComputeConfigHash()

	for _, peer in pairs(s.peerHashes or {}) do
		if peer.lastSeen and (now - peer.lastSeen) < timeout then
			peerCount = peerCount + 1
			if peer.rosterHash ~= myRosterHash or peer.configHash ~= myConfigHash then
				divergedCount = divergedCount + 1
			end
		end
	end

	local details = { peerCount = peerCount, divergedCount = divergedCount }

	-- No recent peers
	if peerCount == 0 then
		return "no_peers", details
	end

	-- Currently stale (fetch in progress)
	if s.isStale then
		return "syncing", details
	end

	-- Divergent peers detected
	if divergedCount > 0 then
		return "diverged", details
	end

	return "synced", details
end

--- Force a full sync broadcast (officer only).
function CommunicationV2:ForceBroadcast()
	if not Cerebro:IsOfficerSafe() then return false end

	self:SendHeartbeat()
	return true
end


--[[=============================================
	V1-Compatible Shim Functions
	
	Drop-in replacements for Communication v1 methods.
	Allows callers to migrate from v1 to v2 with minimal
	code changes. These shims map v1 semantics to v2 protocol.
===============================================]]

--- Drop-in replacement for Communication:SyncToRaid().
-- Builds and broadcasts a full STATE_SNAPSHOT to RAID or GUILD.
-- Officers only. Returns true if sent, false if blocked.
function CommunicationV2:SyncToRaid()
	if not Cerebro:IsOfficerSafe() then return false end

	-- Combat gating
	if Cerebro.modules.SyncScheduler and not Cerebro.modules.SyncScheduler:ShouldSync() then
		return false
	end

	Cerebro:RecalculateAllPR()

	local snapshot = self:BuildSnapshot(true, true)
	local channel = IsInRaid(LE_PARTY_CATEGORY_HOME) and "RAID" or "GUILD"
	self:SendV2Message(MSG_V2.STATE_SNAPSHOT, snapshot, channel)

	-- NOTE: No heartbeat sent here — snapshot.rosterHash already carries the current hash.
	-- Sending a heartbeat immediately after a full snapshot is redundant: any FETCH it
	-- triggers would instantly hit the 30s _lastRaidBroadcastAt cooldown, and receivers
	-- already know the hash from the snapshot they just processed.
	-- The next scheduled heartbeat (~30s) handles any remaining divergence.

	Cerebro:Debug("V2 SyncToRaid: full snapshot broadcast via " .. channel)
	return true
end

--- Drop-in replacement for Communication:RequestSync().
-- Non-officers request data from any online officer via STATE_FETCH.
function CommunicationV2:RequestSync()
	self:SendFetch(nil, IsInRaid(LE_PARTY_CATEGORY_HOME) and "RAID" or "GUILD")
	Cerebro:Debug("V2 RequestSync: STATE_FETCH broadcast")
end

--- Drop-in replacement for Communication:BroadcastPlayerUpdate().
-- V2 no-op: hash-based sync detects player data changes automatically.
-- Next heartbeat will show the hash mismatch and trigger sync.
function CommunicationV2:BroadcastPlayerUpdate(_playerName, _updateData)
	-- V2: No explicit player update broadcast needed.
	-- Build C: Fire immediate heartbeat so peers detect hash change in ~2s
	self:NotifyStateChange("player_update")
	Cerebro:Debug("V2: Player update will propagate via immediate heartbeat")
end

--- Drop-in replacement for Communication:AnnounceSelf().
-- Uses V2 RPC ANNOUNCE method for player presence broadcast.
function CommunicationV2:AnnounceSelf()
	-- Cooldown guard (same as v1: 30s minimum)
	local now = time()
	local cooldown = 30
	if self._lastAnnounceSelfAt and (now - self._lastAnnounceSelfAt) < cooldown then
		Cerebro:Debug("V2 AnnounceSelf: skipped (cooldown)")
		return
	end

	-- Need guild info
	if not IsInGuild() then return end

	local playerName = Cerebro:GetPlayerFullName()
	if not playerName then return end

	local _, className = UnitClass("player")
	local guildName, _, rankIndex = GetGuildInfo("player")
	if not guildName then
		-- Guild info not ready, retry
		self:ScheduleTimer("AnnounceSelf", 5)
		return
	end

	self._lastAnnounceSelfAt = now

	self:SendRPC("ANNOUNCE", {
		player = playerName,
		version = Cerebro.version,
		class = className,
		rank = rankIndex,
	}, "GUILD")

	Cerebro:Debug("V2 AnnounceSelf: RPC ANNOUNCE sent")
end

--- Drop-in replacement for Communication:BroadcastDiscovery().
-- Uses V2 RPC DISCOVERY method to request guild presence scan.
function CommunicationV2:BroadcastDiscovery()
	if not Cerebro:IsOfficerSafe() then return end

	self:SendRPC("DISCOVERY", {
		player = Cerebro:GetPlayerFullName(),
		version = Cerebro.version,
	}, "GUILD")

	-- Also announce ourselves
	self:AnnounceSelf()

	Cerebro:Debug("V2 BroadcastDiscovery: RPC DISCOVERY sent")
end

--- Drop-in replacement for Communication:BroadcastEPGPUpdate().
-- Uses V2 RPC EPGP_UPDATE to immediately notify raid/guild of EP/GP changes.
-- Now carries txId + details so receivers can write local History entries.
-- Not a thundering herd risk: fires once per officer award action, not periodic.
function CommunicationV2:BroadcastEPGPUpdate(player, newBalance, amount, reason, currency, txId, details)
	local data = {
		player = player,
		newBalance = newBalance,
		amount = amount,
		reason = reason,
		currency = currency,
		timestamp = time(),
		epoch = Cerebro.db.global.resetEpoch or 0,
		-- History sync fields (v0.10.30+): allows receivers to write local transactions
		txId = txId,      -- UUID from Ledger:AddTransaction() for dedup
		details = details, -- item link, slot, preGP/postGP, etc.
	}

	-- Match V1 behavior: raid/party if in one, otherwise guild
	local channel = (IsInRaid() or IsInGroup()) and "RAID" or "GUILD"
	self:SendRPC("EPGP_UPDATE", data, channel)

	-- Deferred retry during combat: WoW 12.0 throttles addon comms during encounters,
	-- and heartbeat hash convergence is suppressed (RAID_COMBAT mode). A single retry
	-- after 5s gives other clients a second chance to receive the update.
	if Cerebro.modules.SyncScheduler
	   and Cerebro.modules.SyncScheduler:GetSyncMode() == "RAID_COMBAT" then
		Cerebro:ScheduleTimer(function()
			self:SendRPC("EPGP_UPDATE", data, channel)
			Cerebro:Debug("V2 BroadcastEPGPUpdate retry (combat deferred): " .. currency .. " for " .. player)
		end, 5)
	end

	Cerebro:Debug("V2 BroadcastEPGPUpdate: " .. currency .. " " .. tostring(amount) .. " for " .. player)
	-- Build C: immediate heartbeat for EP/GP changes
	self:NotifyStateChange("epgp_update")
end

--- Invalidate hash cache after state mutations.
-- Call after: EP/GP award, player admit/purge, config change, snapshot merge.
function CommunicationV2:InvalidateHashCache()
	self._rosterHashCache = nil
	self._rosterHashCacheAt = nil
	self._configHashCache = nil
	self._configHashCacheAt = nil
end

--[[=============================================
	Batch EPGP Broadcast (Sync Efficiency)

	Sends all EP/GP awards in a single RPC message
	instead of N individual messages. Dramatically
	reduces ChatThrottleLib saturation during boss kills.
	Also starts the convergence tracker and triggers
	an immediate snapshot push.
===============================================]]

--- Broadcast a batch of EP/GP awards as a single RPC.
-- Called by Session:AwardEPToSelected() after collecting all silent awards.
-- awards: array of { player, newBalance, amount, reason, currency, txId }
function CommunicationV2:BroadcastBatchEPGPUpdate(awards)
	if not awards or #awards == 0 then return end

	local data = {
		awards = awards,
		timestamp = time(),
		epoch = Cerebro.db.global.resetEpoch or 0,
		senderName = Cerebro:GetPlayerFullName(),
	}

	-- Single-channel: RAID when in raid (benched players sync via PushAfterAward snapshot),
	-- GUILD otherwise (solo officer adjustments)
	if IsInRaid(LE_PARTY_CATEGORY_HOME) or IsInGroup() then
		self:SendRPC("BATCH_EPGP_UPDATE", data, "RAID")
	elseif IsInGuild() then
		self:SendRPC("BATCH_EPGP_UPDATE", data, "GUILD")
	end

	Cerebro:Debug(string.format("V2 BroadcastBatchEPGPUpdate: %d awards sent", #awards))

	-- Invalidate hash cache (state just changed)
	self:InvalidateHashCache()

	-- Immediate heartbeat for hash convergence
	self:NotifyStateChange("batch_epgp")

	-- Immediate snapshot push (1s debounce)
	self:PushAfterAward()

	-- Start convergence tracker
	self:StartConvergenceTracker(awards)
end

--- Immediate post-award snapshot push.
-- Fires a debounced (1s) SyncToRaid after mass EP/GP operations.
-- Bypasses the heartbeat-based convergence for critical scenarios.
function CommunicationV2:PushAfterAward()
	if not Cerebro:IsOfficerSafe() then return end

	-- Cancel any pending push timer
	if self._pushAfterAwardTimer then
		self:CancelTimer(self._pushAfterAwardTimer)
	end

	self._pushAfterAwardTimer = self:ScheduleTimer(function()
		self._pushAfterAwardTimer = nil
		-- Invalidate hash cache before snapshot so it carries the latest hash
		self:InvalidateHashCache()
		self:SyncToRaid()
		Cerebro:Debug("V2 PushAfterAward: full snapshot broadcast")
	end, 1)
end

--[[=============================================
	Convergence Tracker (Sync Efficiency)

	After a batch EP/GP award, tracks ACKs from raid
	members to confirm data propagation. Reports
	convergence status to all officers. Retries
	stragglers via targeted WHISPER snapshot.
===============================================]]

--- Start tracking convergence after a batch award.
-- Creates a 10-second collection window for BATCH_ACK messages.
function CommunicationV2:StartConvergenceTracker(awards)
	if not Cerebro:IsOfficerSafe() then return end

	-- Determine expected member set (selectedPlayers with addon)
	local selectedPlayers = Cerebro.db.global.customEP
		and Cerebro.db.global.customEP.selectedPlayers or {}
	local expectedMembers = {}
	local expectedCount = 0
	local myName = Cerebro:GetPlayerFullName()

	for name in pairs(selectedPlayers) do
		if name ~= myName then
			-- Only expect ACKs from players who have the addon
			local addonState = self:GetAddonState(name)
			if addonState == "synced" or addonState == "outdated" then
				expectedMembers[name] = true
				expectedCount = expectedCount + 1
			end
		end
	end

	-- Initialize tracker state
	self._convergenceTracker = {
		startTime = GetTime(),
		expectedMembers = expectedMembers,
		expectedCount = expectedCount,
		ackedMembers = {},
		ackedCount = 0,
		expectedHash = self:ComputeRosterHash(),
		awardCount = #awards,
	}

	-- Schedule timeout report at 10 seconds
	if self._convergenceTimer then
		self:CancelTimer(self._convergenceTimer)
	end
	self._convergenceTimer = self:ScheduleTimer(function()
		self._convergenceTimer = nil
		self:ReportConvergence()
	end, 10)

	Cerebro:Debug(string.format("V2 Convergence tracker started: expecting %d ACKs", expectedCount))
end

--- Handle incoming BATCH_ACK from a raid member.
function CommunicationV2:OnBatchAck(sender, data)
	local tracker = self._convergenceTracker
	if not tracker then return end

	-- Only count expected members
	if not tracker.expectedMembers[sender] then return end
	-- Dedup
	if tracker.ackedMembers[sender] then return end

	tracker.ackedMembers[sender] = true
	tracker.ackedCount = tracker.ackedCount + 1

	-- Early completion: all expected members ACKed
	if tracker.ackedCount >= tracker.expectedCount then
		if self._convergenceTimer then
			self:CancelTimer(self._convergenceTimer)
			self._convergenceTimer = nil
		end
		self:ReportConvergence()
	end
end

--- Report convergence results and retry stragglers.
function CommunicationV2:ReportConvergence()
	local tracker = self._convergenceTracker
	if not tracker then return end

	local elapsed = GetTime() - tracker.startTime
	local stragglers = {}

	-- Find who didn't ACK
	for name in pairs(tracker.expectedMembers) do
		if not tracker.ackedMembers[name] then
			table.insert(stragglers, name)
		end
	end

	local total = tracker.expectedCount
	local acked = tracker.ackedCount

	-- Build report message
	local report
	if #stragglers == 0 then
		report = string.format("|cff00ff00[CONVERGED]|r %d/%d synced in %.1fs (%d awards)",
			acked, total, elapsed, tracker.awardCount)
	else
		-- Build short straggler names
		local shortNames = {}
		for _, name in ipairs(stragglers) do
			table.insert(shortNames, Cerebro:GetShortPlayerName(name) or name)
		end
		report = string.format("|cffff8800[PARTIAL]|r %d/%d synced (%.1fs). Pushing to stragglers: %s",
			acked, total, elapsed, table.concat(shortNames, ", "))
	end

	-- Print to Sync Leader's chat
	Cerebro:Print(report)

	-- Broadcast convergence summary to all officers via AWARD_CONVERGENCE RPC
	local summaryData = {
		acked = acked,
		total = total,
		elapsed = elapsed,
		stragglerCount = #stragglers,
		awardCount = tracker.awardCount,
	}
	local channel = IsInRaid(LE_PARTY_CATEGORY_HOME) and "RAID" or "GUILD"
	self:SendRPC("AWARD_CONVERGENCE", summaryData, channel)

	-- Retry stragglers via targeted WHISPER snapshot
	-- CRITICAL: Must check IsPlayerReachable() to avoid "No player named..." spam
	for _, name in ipairs(stragglers) do
		local shortName = Cerebro:GetShortPlayerName(name)
		if shortName and Cerebro.modules.Communication
		   and Cerebro.modules.Communication:IsPlayerReachable(name) then
			-- Send targeted snapshot via WHISPER
			local snapshot = self:BuildSnapshot(true, false)
			self:SendV2Message(MSG_V2.STATE_SNAPSHOT, snapshot, "WHISPER", shortName)
			Cerebro:Debug("V2 Convergence: retry WHISPER snapshot to " .. shortName)
		else
			Cerebro:Debug("V2 Convergence: skipping offline straggler " .. (shortName or name))
		end
	end

	-- Clean up tracker
	self._convergenceTracker = nil
end

--- Drop-in replacement for Communication:BroadcastDecay().
-- Uses V2 RPC DECAY to notify raid of decay (chat notification only, no data mutation).
-- Actual data propagation happens via V2 heartbeat hash-mismatch → snapshot resync.
function CommunicationV2:BroadcastDecay(decayPercent)
	self:SendRPC("DECAY", {
		decayPercent = decayPercent,
		timestamp = time(),
	}, IsInRaid() and "RAID" or "GUILD")

	Cerebro:Debug("V2 BroadcastDecay: " .. tostring(decayPercent) .. "% decay notification sent")
	-- Build C: immediate heartbeat for decay
	self:NotifyStateChange("decay")
end

--- Drop-in replacement for Communication:SendMessage(EPGP_FORWARD, ..., "WHISPER", leader).
-- Uses V2 RPC EPGP_FORWARD to whisper award data to sync leader for durability.
-- Critical path: ensures non-leader officer EP/GP awards reach the sync leader immediately.
function CommunicationV2:ForwardToSyncLeader(fwdData, leaderShortName)
	self:SendRPC("EPGP_FORWARD", fwdData, "WHISPER", leaderShortName)

	Cerebro:Debug(string.format("V2 EPGP_FORWARD: %s %+d %s to leader %s",
		fwdData.player, fwdData.amount, fwdData.currency, leaderShortName))
end

--- Drop-in replacement for Communication:BroadcastSessionStart().
-- Uses V2 RPC SESSION_START with single-channel broadcast to avoid duplicate delivery.
-- When in raid: RAID only (all raiders are guild members — GUILD send is redundant).
-- When not in raid: GUILD so out-of-raid guild officers still receive it.
function CommunicationV2:BroadcastSessionStart(sessionID, raidName)
	local data = {
		sessionID = sessionID,
		raidName = raidName,
		timestamp = time(),
	}
	-- Single-channel: prevents every raid member receiving two copies of the same message.
	if IsInRaid() or IsInGroup() then
		self:SendRPC("SESSION_START", data, "RAID")
	elseif IsInGuild() then
		self:SendRPC("SESSION_START", data, "GUILD")
	end
	Cerebro:Debug("V2 BroadcastSessionStart: " .. sessionID)

	-- Instant propagation: push full state snapshot so all officers
	-- receive the new session immediately (bypasses 30s heartbeat)
	if Cerebro:IsOfficer() then
		Cerebro:ScheduleTimer(function()
			if Cerebro.modules.CommunicationV2 then
				Cerebro.modules.CommunicationV2:SyncToRaid()
				Cerebro:Debug("V2 Post-session-start sync push complete")
			end
		end, 1)
	end
end

--- Drop-in replacement for Communication:BroadcastSessionEnd().
-- Uses V2 RPC SESSION_END with single-channel broadcast to avoid duplicate delivery.
-- Non-raiding guild officers detect the session end via the next heartbeat hash mismatch
-- cycle (<=30s) — acceptable since they are observing, not administering the session.
function CommunicationV2:BroadcastSessionEnd(sessionID)
	local data = {
		sessionID = sessionID,
		timestamp = time(),
	}
	-- Single-channel: prevents every raid member receiving two copies of the same message.
	if IsInRaid() or IsInGroup() then
		self:SendRPC("SESSION_END", data, "RAID")
	elseif IsInGuild() then
		self:SendRPC("SESSION_END", data, "GUILD")
	end

	-- Post-session full sync push (replaces V1 Communication:SyncToGuild)
	if Cerebro:IsOfficer() and IsInGuild() then
		Cerebro:ScheduleTimer(function()
			if Cerebro.modules.CommunicationV2 then
				Cerebro.modules.CommunicationV2:SyncToRaid()
				Cerebro:Debug("V2 Post-session sync push complete")
			end
		end, 3)
	end
	Cerebro:Debug("V2 BroadcastSessionEnd: " .. sessionID)
end

--- Drop-in replacement for Communication:BroadcastReset().
-- Uses V2 RPC RESET to notify entire guild.
function CommunicationV2:BroadcastReset()
	self:SendRPC("RESET", {
		timestamp = time(),
		epoch = Cerebro.db.global.resetEpoch or 0,
	}, "GUILD")
	Cerebro:Debug("V2 BroadcastReset: epoch " .. tostring(Cerebro.db.global.resetEpoch or 0))
end

--- RETIRED (v0.11.0): Superseded by Cerebro:PurgePlayer() in Utilities.lua.
-- This function previously sent "PLAYER_PURGE" RPC, but the handler is registered
-- as "PURGE" — so messages were silently dropped. PurgePlayer() handles the full
-- lifecycle: ledger + delete + audit + RPC("PURGE") + SyncToRaid() + UI refresh.
-- Kept as thin redirect for any remaining callers.
function CommunicationV2:BroadcastPlayerPurge(playerName)
	Cerebro:Debug("BroadcastPlayerPurge: redirecting to PurgePlayer()")
	Cerebro:PurgePlayer(playerName)
end

--- Drop-in replacement for Communication:VersionCheck().
-- Delegates to V1 implementation which manages response collection state.
-- The V1 handler (HandleVersionCheck) is already routed via V2 RPC dispatcher.
function CommunicationV2:VersionCheck()
	if Cerebro.modules.Communication then
		Cerebro.modules.Communication:VersionCheck()
	end
end

--- Drop-in replacement for Communication:RequestSyncAudit().
-- Delegates to V1 implementation which manages complex audit state.
-- The V1 handlers (HandleSyncAuditRequest/Response) are already routed via V2 RPC dispatcher.
function CommunicationV2:RequestSyncAudit()
	if Cerebro.modules.Communication then
		Cerebro.modules.Communication:RequestSyncAudit()
	end
end

--- Sends a single ledger transaction to all online officers.
-- Called automatically by Ledger:AddTransaction() (unless silent=true).
-- Uses BROADCAST_TX RPC key (matches the registered dispatch handler).
function CommunicationV2:BroadcastTransaction(entry)
	if not entry or not entry.id then return end
	local channel = (IsInRaid() or IsInGroup()) and "RAID" or "GUILD"
	self:SendRPC("BROADCAST_TX", { entry = entry }, channel)
	Cerebro:Debug("V2 BroadcastTransaction: " .. entry.id)
end

--- Force-pushes the ENTIRE local transaction log to all online officers.
-- Called by /cb synchistory for retroactive catch-up after missed events.
-- Batches into groups of 25 to stay within AceComm message size limits.
local BATCH_SIZE = 25
function CommunicationV2:BroadcastAllTransactions()
	if not Cerebro:IsOfficer() then return end

	local txns = Cerebro.db.global.transactions or {}
	local total = #txns
	if total == 0 then
		Cerebro:Print("No transactions to sync.")
		return
	end

	local channel = (IsInRaid() or IsInGroup()) and "RAID" or "GUILD"
	local batchCount = 0
	local batch = {}

	for i, entry in ipairs(txns) do
		table.insert(batch, entry)
		if #batch >= BATCH_SIZE or i == total then
			self:SendRPC("BROADCAST_TX_BATCH", {
				entries = batch,
				total = total,
				epoch = Cerebro.db.global.resetEpoch or 0,
			}, channel)
			batchCount = batchCount + 1
			batch = {}
		end
	end

	Cerebro:Debug(string.format("V2 BroadcastAllTransactions: %d entries in %d batches", total, batchCount))
end

--- Sends a single audit log entry to all online officers.
-- Called automatically by Ledger:AddAuditLog().
function CommunicationV2:BroadcastAuditLog(entry)
	self:SendRPC("BROADCAST_AUDIT", { entry = entry }, "GUILD")
	Cerebro:Debug("V2 BroadcastAuditLog sent")
end