--[[
    Cerebro - V2 RPC Dispatch & Broadcast
    Extracted from CommunicationV2.lua (Phase 2a decomposition)

    RPC message handling, all broadcast functions, convergence tracking,
    repair cycle, and guild roster update handler.

    @module Sync.RPCDispatch
    @version 0.12.0
    @invariant All RPC handlers require IsOfficer(sender) + epoch check + quarantine guard
    @invariant Heartbeat gated to Sync Leader during active raids; event-driven ungated
]]--

local AddonName = "Cerebro"
local Cerebro = LibStub("AceAddon-3.0"):GetAddon(AddonName)
local CommunicationV2 = Cerebro:GetModule("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

-- Repair cycle constants
local MAX_REPAIR_ATTEMPTS = 3
local REPAIR_VERIFY_TIMEOUT = 5  -- seconds to wait for REPAIR_ACK

-- Batch size for BroadcastAllTransactions
local BATCH_SIZE = 25

--[[=============================================
	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)
	self:HandleBossKill(data, sender)
end

RPC_DISPATCH["SESSION_START"] = function(self, sender, data) -- luacheck: ignore self
	-- 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

	self:HandleSessionEnd(data, sender)

	-- 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) -- luacheck: ignore self
	-- 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)
	self:HandleBidSubmit(data, sender)
end

RPC_DISPATCH["BID_LOCK"] = function(self, sender, data)
	self:HandleBidLock(data, sender)
end

RPC_DISPATCH["BID_UNLOCK"] = function(self, sender, data)
	self:HandleBidUnlock(data, sender)
end

RPC_DISPATCH["ITEM_DROP"] = function(self, sender, data)
	self:HandleItemDrop(data, sender)
end

RPC_DISPATCH["ITEM_AWARD"] = function(self, sender, data)
	self:HandleItemAward(data, sender)
end

RPC_DISPATCH["DECAY"] = function(self, sender, data)
	self:HandleDecay(data, sender)
end

RPC_DISPATCH["FACTORY_RESET"] = function(self, sender, data)
	self:HandleReset(data, sender)
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)
	self:HandleVersionCheck(data, sender)
end

RPC_DISPATCH["ANNOUNCE"] = function(self, sender, data)
	self:HandleAnnounceSelf(data, sender)
end

RPC_DISPATCH["DISCOVERY"] = function(self, sender, data)
	self:HandleDiscoveryRequest(data, sender)
end

RPC_DISPATCH["EPGP_UPDATE"] = function(self, sender, data)
	self:HandleEPGPUpdate(data, sender)
end

RPC_DISPATCH["EPGP_FORWARD"] = function(self, sender, data)
	self:HandleEPGPForward(data, sender)
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: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
		-- 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,
		}
		self:HandleEPGPUpdate(epgpData, sender)
		appliedCount = appliedCount + 1
	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)
	self:HandleSyncReceipts(data, sender)
end

RPC_DISPATCH["RECEIPT_REQUEST"] = function(self, sender, data)
	self:HandleReceiptRequest(data, sender)
end

RPC_DISPATCH["SYNC_AUDIT_REQ"] = function(self, sender, data)
	self:HandleSyncAuditRequest(data, sender)
end

RPC_DISPATCH["SYNC_AUDIT_RESP"] = function(self, sender, data)
	self:HandleSyncAuditResponse(data, sender)
end

RPC_DISPATCH["PLAYER_UPDATE"] = function(self, sender, data)
	self:HandlePlayerUpdate(data, sender)
end

RPC_DISPATCH["PREP_REQUEST"] = function(self, sender, data) -- luacheck: ignore self
	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) -- luacheck: ignore self
	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)
	self:HandleBroadcastTransaction(data, sender)
end

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

RPC_DISPATCH["BROADCAST_AUDIT"] = function(self, sender, data)
	self:HandleBroadcastAuditLog(data, sender)
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

-- Remote Diagnostic Support (read-only, officer-gated)
RPC_DISPATCH["SUPPORT_REQUEST"] = function(_self, sender, data)
	local diag = Cerebro:GetModule("Diagnostics", true)
	if diag and diag.HandleSupportRequest then
		diag:HandleSupportRequest(sender, data)
	end
end

RPC_DISPATCH["SUPPORT_RESPONSE"] = function(_self, sender, data)
	local diag = Cerebro:GetModule("Diagnostics", true)
	if diag and diag.HandleSupportResponse then
		diag:HandleSupportResponse(sender, data)
	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 Cerebro: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.IsPlayerReachable then
			if not Cerebro: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

	-- @invariant SPEC §8 SA-5: Support API observers run AFTER production
	-- dispatch completes. SA-6: errors in observer callbacks are isolated
	-- via pcall inside NotifyObservers and cannot affect production handling.
	if self.modules.SupportAPI then
		self.modules.SupportAPI:NotifyObservers(msgType, sender, data)
	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.
		-- Session-Aware Sync: sender identity for debounced SYNC_ACK handshake
		senderName = Cerebro:GetPlayerFullName(),
	}

	-- 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.
-- InvalidateHashCache moved to Modules/Sync/HashEngine.lua

--[[=============================================
	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: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()
	-- Send VERSION_CHECK via V2 RPC to GUILD channel
	self:SendRPC("VERSION_CHECK", { request = true, version = Cerebro.version }, "GUILD")
	Cerebro:Print("|cff00ff00[Version Check]|r Broadcasting via V2...")
	Cerebro:Print("Collecting responses for 5 seconds...")
	-- HandleVersionCheck in V2Handlers.lua will process responses
	self._versionCache = {}
	self._versionCheckStart = time()
	Cerebro:ScheduleTimer(function()
		-- Report collected versions
		local count = 0
		for _ in pairs(self._versionCache or {}) do count = count + 1 end
		if count == 0 then
			Cerebro:Print("|cffff8800No responses received.|r Other officers may be offline.")
		else
			Cerebro:Print(string.format("|cff00ff00[Version Check]|r %d response(s):", count))
			for name, ver in pairs(self._versionCache) do
				Cerebro:Print("  " .. name .. ": " .. tostring(ver))
			end
		end
	end, 5)
end

--- Request sync audit from online officers via V2 RPC.
function CommunicationV2:RequestSyncAudit()
	if not Cerebro:IsOfficerSafe() then
		Cerebro:Print("Sync audit requires officer permissions.")
		return
	end
	self:SendRPC("SYNC_AUDIT_REQ", { request = true }, "GUILD")
	Cerebro:Print("|cff00ff00[Sync Audit]|r Request sent via V2...")
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.
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

--[[=============================================
	Session-Aware Sync: Repair-Verify-Retry (RVR) Protocol
	Guarantees convergence via verified hash ACK after each
	repair attempt, with escalation after max retries.
===============================================]]

--- Unified entry point for all repair scenarios.
-- Called by: OnSessionStartConvergence, OnEncounterStartConvergence,
-- ReportConvergence (batch stragglers), PostCombatFlush.
function CommunicationV2:StartRepairCycle(stragglers, expectedHash, reason)
	self._repairState = self._repairState or {}

	for _, name in ipairs(stragglers) do
		-- Initialize per-client repair tracker
		self._repairState[name] = {
			attempt = 0,
			expectedHash = expectedHash,
			reason = reason,
		}
		-- Fire first repair attempt
		self:SendRepairAttempt(name)
	end
end

function CommunicationV2:SendRepairAttempt(name)
	local state = self._repairState and self._repairState[name]
	if not state then return end

	state.attempt = state.attempt + 1

	if state.attempt > MAX_REPAIR_ATTEMPTS then
		-- ESCALATION: flag as unrecoverable
		local shortName = Cerebro:GetShortPlayerName(name) or name
		Cerebro:Print(string.format(
			"|cffff0000[SYNC ALERT]|r %s failed to converge after %d attempts (%s)",
			shortName, MAX_REPAIR_ATTEMPTS, state.reason))

		-- Broadcast alert to officers
		self:SendRPC("CONVERGENCE_ALERT", {
			straggler = name,
			attempts = MAX_REPAIR_ATTEMPTS,
			reason = state.reason,
			expectedHash = state.expectedHash,
		}, IsInRaid() and "RAID" or "GUILD")

		-- Audit log
		if Cerebro.Ledger and Cerebro.Ledger.AddAuditLog then
			Cerebro.Ledger:AddAuditLog("SYNC_FAILURE",
				string.format("%s: %s failed to converge after %d repair attempts",
					state.reason, shortName, MAX_REPAIR_ATTEMPTS))
		end

		self._repairState[name] = nil
		return
	end

	-- Send repair snapshot
	local shortName = Cerebro:GetShortPlayerName(name)
	local comm = Cerebro.modules.Communication
	if shortName and comm and Cerebro:IsPlayerReachable(name) then
		local snapshot = self:BuildSnapshot(true, false)
		snapshot.repairAttempt = state.attempt  -- tag for REPAIR_ACK
		self:SendV2Message(MSG_V2.STATE_SNAPSHOT, snapshot, "WHISPER", shortName)
		Cerebro:Debug(string.format("RVR: repair attempt %d/%d to %s (%s)",
			state.attempt, MAX_REPAIR_ATTEMPTS, shortName, state.reason))
	end

	-- Schedule verification timeout
	local timerKey = "_repairTimer_" .. name
	if self[timerKey] then
		Cerebro:CancelTimer(self[timerKey])
	end
	self[timerKey] = Cerebro:ScheduleTimer(function()
		self[timerKey] = nil
		-- Timeout: no REPAIR_ACK received — retry with backoff
		local backoff = 2 * (state.attempt or 1)
		Cerebro:Debug(string.format("RVR: no ACK from %s, retrying in %ds",
			shortName or name, backoff))
		Cerebro:ScheduleTimer(function()
			self:SendRepairAttempt(name)
		end, backoff)
	end, REPAIR_VERIFY_TIMEOUT)
end

--[[=============================================
	Session-Aware Sync: New RPC Handlers
===============================================]]

--- SYNC_ACK: Debounced hash acknowledgment from receivers after EP/GP updates.
-- Pattern: receivers coalesce rapid item awards into a single ACK with
-- their current roster hash, sent 3s after the last update.
RPC_DISPATCH["SYNC_ACK"] = function(self, sender, data)
	-- Only officers need to process convergence tracking
	if not Cerebro:IsOfficerSafe() then return end
	if not data then return end

	-- Feed into convergence tracker if active
	if self._convergenceTracker then
		self:OnBatchAck(sender, data)
	end

	-- Update peer hash cache (so encounter gate has fresh data)
	local s = Cerebro.db.global.syncStateV2 or {}
	s.peerHashes = s.peerHashes or {}
	if s.peerHashes[sender] then
		s.peerHashes[sender].rosterHash = data.rosterHash
	end
	Cerebro.db.global.syncStateV2 = s
end

--- REPAIR_ACK: Sent by repaired clients after applying a repair snapshot.
-- Validates convergence via hash comparison and drives the RVR retry loop.
RPC_DISPATCH["REPAIR_ACK"] = function(self, sender, data)
	-- Only the sync leader (officer) processes repair state
	if not Cerebro:IsOfficerSafe() then return end
	if not data then return end

	local state = self._repairState and self._repairState[sender]
	if not state then return end

	-- Cancel the verification timeout
	local timerKey = "_repairTimer_" .. sender
	if self[timerKey] then
		Cerebro:CancelTimer(self[timerKey])
		self[timerKey] = nil
	end

	-- Verify hash convergence
	if data.rosterHash == state.expectedHash then
		-- SUCCESS: client converged
		Cerebro:Debug(string.format("RVR: %s CONVERGED on attempt %d (%s)",
			Cerebro:GetShortPlayerName(sender) or sender, state.attempt, state.reason))
		self._repairState[sender] = nil
	else
		-- STILL DIVERGENT: retry with backoff
		Cerebro:Debug(string.format(
			"RVR: %s still divergent after attempt %d (got %s, expected %s)",
			Cerebro:GetShortPlayerName(sender) or sender, state.attempt,
			tostring(data.rosterHash), tostring(state.expectedHash)))
		local backoff = 2 * state.attempt
		Cerebro:ScheduleTimer(function()
			self:SendRepairAttempt(sender)
		end, backoff)
	end

	-- Update peer hash cache
	local s = Cerebro.db.global.syncStateV2 or {}
	s.peerHashes = s.peerHashes or {}
	if s.peerHashes[sender] then
		s.peerHashes[sender].rosterHash = data.rosterHash
	end
	Cerebro.db.global.syncStateV2 = s
end

--- CONVERGENCE_ALERT: Notification broadcast to officers when RVR
-- exhausts all retry attempts for a straggler client.
RPC_DISPATCH["CONVERGENCE_ALERT"] = function(_self, sender, data)
	if not Cerebro:IsOfficerSafe() then return end
	if not data then return end

	Cerebro:Print(string.format(
		"|cffff0000[SYNC ALERT]|r %s reports: %s failed to converge after %d attempts (%s)",
		Cerebro:GetShortPlayerName(sender) or sender,
		Cerebro:GetShortPlayerName(data.straggler) or data.straggler or "?",
		data.attempts or 0,
		data.reason or "unknown"))
end

--[[=============================================
	Session-Aware Sync: Late Joiner Detection
	Fires during active sessions when GROUP_ROSTER_UPDATE
	detects new raid members. Debounced to prevent snapshot
	storm during raid formation.
===============================================]]

function CommunicationV2:GROUP_ROSTER_UPDATE()
	local syncSched = Cerebro.modules.SyncScheduler
	if not syncSched or not syncSched._sessionActive then return end
	if not syncSched:IsSyncLeader() then return end

	-- Compare current raid roster against _knownRaidMembers
	self._knownRaidMembers = self._knownRaidMembers or {}
	self._pendingJoiners = self._pendingJoiners or {}

	local current = {}
	for i = 1, GetNumGroupMembers() do
		local name = GetRaidRosterInfo(i)
		if name then
			local full = Cerebro:GetFullPlayerName(name)
			current[full] = true
			if not self._knownRaidMembers[full] and full ~= Cerebro:GetPlayerFullName() then
				-- Collect new member into pending set (don't push yet)
				self._pendingJoiners[full] = true
			end
		end
	end
	self._knownRaidMembers = current

	-- Debounce: reset the 10s timer on every roster change.
	-- During raid formation (20 people joining in 30s), the timer keeps
	-- resetting. Only fires ONCE after joins stop, pushing to all at once.
	if next(self._pendingJoiners) then
		if self._joinerPushTimer then
			Cerebro:CancelTimer(self._joinerPushTimer)
		end
		self._joinerPushTimer = Cerebro:ScheduleTimer(function()
			self._joinerPushTimer = nil
			local snapshot = self:BuildSnapshot(true, false)
			local count = 0
			for full in pairs(self._pendingJoiners) do
				local shortName = Cerebro:GetShortPlayerName(full)
				local comm = Cerebro.modules.Communication
				if shortName and comm and Cerebro:IsPlayerReachable(full) then
					self:SendV2Message(MSG_V2.STATE_SNAPSHOT, snapshot, "WHISPER", shortName)
					count = count + 1
				end
			end
			if count > 0 then
				Cerebro:Debug(string.format("Late joiner push: %d new members synced", count))
			end
			self._pendingJoiners = {}
		end, 10)
	end
end
