--[[
	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.

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

	Cerebro:Debug("CommunicationV2 enabled (prefix: " .. V2_PREFIX .. ")")
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
end

--[[=============================================
	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()
	-- 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:%d:%d:%s:%d",
			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, "|")
	return self:SimpleHash(canonical), #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()
	local config = Cerebro.db.global.config or {}

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

	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
			keys[#keys + 1] = tostring(k)
		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, "|")
	return self:SimpleHash(canonical)
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=%d gp=%d status=%s lastUpdate=%d",
			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(),
	}
	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 — broadcast snapshot to RAID/GUILD (O(1) instead of O(N)).
			-- One broadcast covers all raiders simultaneously, eliminating the per-raider WHISPER fan-out.
			-- Uses _lastRaidBroadcastAt (scalar, independent of _lastSnapshotTo officer↔officer table).
			-- SyncToRaid() post-combat also broadcasts; if it fired recently the cooldown skips this
			-- push — which is correct behavior (raiders already received a fresh snapshot).
			now = time()
			self._lastRaidBroadcastAt = self._lastRaidBroadcastAt or 0
			if (now - self._lastRaidBroadcastAt) >= 30 then
				self._lastRaidBroadcastAt = 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()
				local channel = IsInRaid(LE_PARTY_CATEGORY_HOME) and "RAID" or "GUILD"
				self:SendV2Message(MSG_V2.STATE_SNAPSHOT, snapshot, channel)
				Cerebro:Debug(string.format("V2 PUSH broadcast via %s (triggered by mismatch with %s)",
					channel, Cerebro:GetShortPlayerName(sender)))
			else
				Cerebro:Debug(string.format("V2 PUSH broadcast skipped (cooldown, %ds ago)",
					now - self._lastRaidBroadcastAt))
			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 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 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)
		if data.customEP then
			self:MergeSelectionData(data.customEP, sender)
		end
	end

	-- Merge config data
	if data.includes and data.includes.config and data.config then
		self:MergeConfigData(data.config, sender)
	end

	-- Merge signupAwards (union merge: keep latest per eventKey)
	if data.signupAwards then
		self:MergeSignupAwards(data.signupAwards, 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()

	-- 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
	if data.configHash then
		local postMergeConfigHash = self:ComputeConfigHash()
		if postMergeConfigHash ~= data.configHash then
			Cerebro:Debug(string.format("V2 SNAPSHOT integrity FAILED: config expected=%s got=%s (from %s)",
				data.configHash, postMergeConfigHash, sender))
			integrityOk = false
		end
	end
	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.
	local incomingTimes
	if type(incomingConfig.configLastUpdate) == "table" then
		incomingTimes = incomingConfig.configLastUpdate
	elseif type(incomingConfig.configLastUpdate) == "number" then
		-- Legacy: spread single timestamp across all sections
		local ts = 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
	elseif type(localConfig.configLastUpdate) == "number" then
		local ts = 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=%d, mine=%d)", inEpgpTime, myEpgpTime))
	else
		Cerebro:Debug(string.format("V2 config merge: kept local epgp (mine=%d >= theirs=%d)", 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=%d, mine=%d)", inPrepTime, myPrepTime))
	else
		Cerebro:Debug(string.format("V2 config merge: kept local prep (mine=%d >= theirs=%d)", 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=%d > mine=%d)",
				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=%d >= theirs=%d)",
				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=%d, mine=%d)", 
			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)
		end
	else
		Cerebro:Debug(string.format("V2 selection merge from %s: REJECTED (theirs=%d, mine=%d)", 
			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

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

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.
-- Not a thundering herd risk: fires once per officer award action, not periodic.
function CommunicationV2:BroadcastEPGPUpdate(player, newBalance, amount, reason, currency)
	local data = {
		player = player,
		newBalance = newBalance,
		amount = amount,
		reason = reason,
		currency = currency,
		timestamp = time(),
	}

	-- 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

--- 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

--- Drop-in replacement for Communication:BroadcastTransaction().
-- Uses V2 RPC BROADCAST_TRANSACTION to send ledger entries to officers.
function CommunicationV2:BroadcastTransaction(entry)
	self:SendRPC("BROADCAST_TRANSACTION", {
		entry = entry,
	}, "GUILD")
	Cerebro:Debug("V2 BroadcastTransaction sent")
end

--- Drop-in replacement for Communication:BroadcastAuditLog().
-- Uses V2 RPC BROADCAST_AUDIT to send system log entries to officers.
function CommunicationV2:BroadcastAuditLog(entry)
	self:SendRPC("BROADCAST_AUDIT", {
		entry = entry,
	}, "GUILD")
	Cerebro:Debug("V2 BroadcastAuditLog sent")
end