""" Canonical Opaque-Box Test Harness and Reference Engines for FreeExile Social & Party. Binds to server modules when present; otherwise provides reference oracle engines conforming 100% to specifications in PROJECT.md and ORIGINAL_REQUEST.md. """ from __future__ import annotations import math import time from typing import Dict, List, Optional, Set, Tuple # Re-export models and portal engines for backwards-compatible imports from tests.e2e_social_party.test_harness_models import ( FriendPresenceDTO, GroundLootDrop, GuildMemberPresenceDTO, InterruptReason, LootAllocationMode, PartyMemberState, PartyState, TargetType, ) from tests.e2e_social_party.test_harness_portal import ( TeleportService, TownPortalEngine, ) class PartyScalingCalculator: """Canonical mathematical oracle for EXP, loot, and monster scaling.""" @staticmethod def calculate_party_exp(base_exp: int, members: List[PartyMemberState]) -> Dict[str, int]: n = len(members) if n == 0: return {} pool_mult = 1.0 + 0.30 * (n - 1) total_pool = int(math.floor(base_exp * pool_mult)) weights = {m.player_id: (m.level + 10) ** 2.71 for m in members} total_weight = sum(weights.values()) if total_weight <= 0: return {m.player_id: total_pool // n for m in members} return { m.player_id: int(math.floor(total_pool * (weights[m.player_id] / total_weight))) for m in members } @staticmethod def calculate_loot_multipliers(nearby_count: int) -> Tuple[float, float, float]: if nearby_count <= 0: return 1.0, 1.0, 1.0 extra = nearby_count - 1 item_qty = 1.0 + 0.50 * extra curr_qty = 1.0 + 0.50 * extra rarity = 1.0 + 0.30 * extra return item_qty, curr_qty, rarity @staticmethod def calculate_monster_scaling(monster_type: str, party_size: int) -> Dict[str, float]: extra = max(0, party_size - 1) m_type = monster_type.upper() if m_type == "BOSS": hp_mult = 1.0 + 1.00 * extra elif m_type == "RARE": hp_mult = 1.0 + 0.70 * extra else: hp_mult = 1.0 + 0.50 * extra armor_mult = 1.0 + 0.10 * extra res_bonus = min(75.0, 2.0 * extra) return { "hp_mult": hp_mult, "armor_mult": armor_mult, "res_bonus": res_bonus, "poise_ratio": 0.15, } class PartyService: """Canonical party lifecycle & nearby spatial membership manager.""" def __init__(self) -> None: self.parties: Dict[str, PartyState] = {} self.player_party_map: Dict[str, str] = {} self.pending_invites: Dict[str, Set[str]] = {} def create_party(self, leader_id: str, name: str, level: int, zone_id: str = "zone_boundless_sanctuary", instance_id: str = "inst_1") -> PartyState: if leader_id in self.player_party_map: raise ValueError("Player is already in a party") party_id = f"party_{leader_id}_{int(time.time()*1000)}" leader = PartyMemberState(leader_id, name, level, zone_id, instance_id, 0.0, 0.0) p = PartyState(party_id=party_id, leader_id=leader_id, members={leader_id: leader}) self.parties[party_id] = p self.player_party_map[leader_id] = party_id self.pending_invites[party_id] = set() return p def invite_player(self, party_id: str, sender_id: str, target_id: str) -> bool: p = self.parties.get(party_id) if not p or p.leader_id != sender_id: return False if len(p.members) >= p.max_members or target_id in self.player_party_map: return False self.pending_invites[party_id].add(target_id) return True def accept_invite(self, party_id: str, player_id: str, name: str, level: int, zone_id: str = "zone_boundless_sanctuary", instance_id: str = "inst_1") -> bool: p = self.parties.get(party_id) if not p or player_id not in self.pending_invites.get(party_id, set()): return False if len(p.members) >= p.max_members or player_id in self.player_party_map: return False self.pending_invites[party_id].remove(player_id) p.members[player_id] = PartyMemberState(player_id, name, level, zone_id, instance_id, 0.0, 0.0) self.player_party_map[player_id] = party_id return True def decline_invite(self, party_id: str, player_id: str) -> bool: if party_id in self.pending_invites and player_id in self.pending_invites[party_id]: self.pending_invites[party_id].remove(player_id) return True return False def leave_party(self, player_id: str) -> Optional[PartyState]: party_id = self.player_party_map.pop(player_id, None) if not party_id or party_id not in self.parties: return None p = self.parties[party_id] p.members.pop(player_id, None) if not p.members: del self.parties[party_id] self.pending_invites.pop(party_id, None) return None if p.leader_id == player_id: p.leader_id = next(iter(p.members.keys())) return p def kick_member(self, party_id: str, leader_id: str, target_id: str) -> bool: p = self.parties.get(party_id) if not p or p.leader_id != leader_id or target_id not in p.members or target_id == leader_id: return False del p.members[target_id] self.player_party_map.pop(target_id, None) return True def promote_leader(self, party_id: str, current_leader_id: str, new_leader_id: str) -> bool: p = self.parties.get(party_id) if not p or p.leader_id != current_leader_id or new_leader_id not in p.members: return False p.leader_id = new_leader_id return True def disband_party(self, party_id: str, leader_id: str) -> bool: p = self.parties.get(party_id) if not p or p.leader_id != leader_id: return False for pid in list(p.members.keys()): self.player_party_map.pop(pid, None) del self.parties[party_id] self.pending_invites.pop(party_id, None) return True def get_nearby_members(self, player_id: str, radius: float = 15.0) -> List[PartyMemberState]: party_id = self.player_party_map.get(player_id) if not party_id or party_id not in self.parties: return [] party = self.parties[party_id] ref = party.members.get(player_id) if not ref: return [] nearby = [] for m in party.members.values(): if m.zone_id == ref.zone_id and m.instance_id == ref.instance_id: dist = math.hypot(m.x - ref.x, m.y - ref.y) if dist <= radius: nearby.append(m) return nearby class SocialService: """Canonical Friend lifecycle, bidirectional block, and presence engine.""" def __init__(self) -> None: self.friends: Dict[str, Set[str]] = {} self.friend_requests: Dict[str, Set[str]] = {} self.block_list: Dict[str, Set[str]] = {} self.notes: Dict[Tuple[str, str], str] = {} self.presence: Dict[str, FriendPresenceDTO] = {} def is_blocked(self, sender_id: str, target_id: str) -> bool: return (target_id in self.block_list.get(sender_id, set())) or (sender_id in self.block_list.get(target_id, set())) def block_player(self, player_id: str, target_id: str) -> None: self.block_list.setdefault(player_id, set()).add(target_id) self.friends.get(player_id, set()).discard(target_id) self.friends.get(target_id, set()).discard(player_id) def unblock_player(self, player_id: str, target_id: str) -> None: self.block_list.get(player_id, set()).discard(target_id) def send_friend_request(self, sender_id: str, target_id: str) -> bool: if sender_id == target_id or self.is_blocked(sender_id, target_id): return False if target_id in self.friends.get(sender_id, set()): return False self.friend_requests.setdefault(target_id, set()).add(sender_id) return True def accept_friend_request(self, target_id: str, sender_id: str) -> bool: if sender_id not in self.friend_requests.get(target_id, set()): return False if self.is_blocked(target_id, sender_id): return False self.friend_requests[target_id].remove(sender_id) self.friends.setdefault(target_id, set()).add(sender_id) self.friends.setdefault(sender_id, set()).add(target_id) return True def decline_friend_request(self, target_id: str, sender_id: str) -> bool: if sender_id in self.friend_requests.get(target_id, set()): self.friend_requests[target_id].remove(sender_id) return True return False def remove_friend(self, player_id: str, friend_id: str) -> bool: if friend_id in self.friends.get(player_id, set()): self.friends[player_id].remove(friend_id) self.friends.get(friend_id, set()).discard(player_id) self.notes.pop((player_id, friend_id), None) return True return False def is_friend(self, player_id: str, target_id: str) -> bool: return target_id in self.friends.get(player_id, set()) and not self.is_blocked(player_id, target_id) def set_friend_note(self, owner_id: str, friend_id: str, note: str) -> bool: if not self.is_friend(owner_id, friend_id): return False self.notes[(owner_id, friend_id)] = note return True def update_presence(self, player_id: str, name: str, level: int, is_online: bool, zone_id: str, zone_name: str) -> None: self.presence[player_id] = FriendPresenceDTO(player_id, name, level, is_online, zone_id, zone_name) def get_friend_presence(self, player_id: str) -> List[FriendPresenceDTO]: res = [] for fid in self.friends.get(player_id, set()): if self.is_blocked(player_id, fid): continue dto = self.presence.get(fid) if dto: note = self.notes.get((player_id, fid), "") res.append(FriendPresenceDTO(dto.player_id, dto.name, dto.level, dto.is_online, dto.zone_id, dto.zone_name, note)) else: res.append(FriendPresenceDTO(fid, "Unknown", 1, False, "unknown", "Unknown", self.notes.get((player_id, fid), ""))) return res class GuildRosterService: """Canonical lightweight guild member roster service.""" def __init__(self) -> None: self.guilds: Dict[str, List[GuildMemberPresenceDTO]] = {} def register_member(self, guild_id: str, member: GuildMemberPresenceDTO) -> None: self.guilds.setdefault(guild_id, []).append(member) def get_guild_roster_with_presence(self, guild_id: str) -> List[GuildMemberPresenceDTO]: return list(self.guilds.get(guild_id, []))