diff --git a/src/commander_ai_lab/sim/politics/__init__.py b/src/commander_ai_lab/sim/politics/__init__.py new file mode 100644 index 0000000..9ee8e64 --- /dev/null +++ b/src/commander_ai_lab/sim/politics/__init__.py @@ -0,0 +1,19 @@ +""" +Commander AI Lab — Politics Engine (Phase 6) + +Exposes: + PoliticsEngine — deal proposals, threat broadcasts, spite tracking + TargetingMemory — per-seat history of who targeted whom + Deal — data class for a proposed deal + DealResponse — accept / counter / reject +""" +from commander_ai_lab.sim.politics.engine import PoliticsEngine +from commander_ai_lab.sim.politics.memory import TargetingMemory +from commander_ai_lab.sim.politics.deals import Deal, DealResponse, DealType +from commander_ai_lab.sim.politics.comms import ThreatBroadcast, PoliticsCommsChannel + +__all__ = [ + "PoliticsEngine", "TargetingMemory", + "Deal", "DealResponse", "DealType", + "ThreatBroadcast", "PoliticsCommsChannel", +] diff --git a/src/commander_ai_lab/sim/politics/comms.py b/src/commander_ai_lab/sim/politics/comms.py new file mode 100644 index 0000000..c5e521d --- /dev/null +++ b/src/commander_ai_lab/sim/politics/comms.py @@ -0,0 +1,137 @@ +""" +Commander AI Lab — Politics Comms Channel (Phase 6) +==================================================== +Handles the spoken / narrated side of the politics system: + - ThreatBroadcast: a seat declaring its intentions publicly + - PoliticsCommsChannel: message bus that routes broadcasts to + other seats and to the narration panel + +AI personas use this to say things like: + "Timmy: Anyone who attacks me this turn is getting Craterhoof next turn." + "Spike: I'll leave P2 alone if someone deals with that Seedborn Muse." +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum, auto +from typing import Callable, Awaitable, Optional + + +class BroadcastType(Enum): + THREAT = auto() # "If you X, I will Y" + OFFER = auto() # "I will X if you Y" + WARNING = auto() # "I am about to X" + TAUNT = auto() # flavour / personality expression + GRATITUDE = auto() # acknowledging help + ACCUSATION = auto() # calling out a deal-breaker + CALL_TO_ARMS = auto() # urging others to attack a seat + + +@dataclass +class ThreatBroadcast: + """ + A public statement from one seat addressed to one or all seats. + + Attributes: + speaker: Seat index sending the message. + audience: Seat index of intended recipient (-1 = table). + broadcast_type: Flavour category. + text: Natural-language text (generated by AI persona). + turn: Game turn when spoken. + condition_seat: The seat the statement is about (if any). + """ + speaker: int + audience: int # -1 = everyone + broadcast_type: BroadcastType + text: str + turn: int + condition_seat: Optional[int] = None + + +# Callback type: async function that receives a ThreatBroadcast +BroadcastHandler = Callable[[ThreatBroadcast], Awaitable[None]] + + +class PoliticsCommsChannel: + """ + Message bus for politics broadcasts. + + Handlers can be registered per-seat (AI opponents) or globally + (narration panel, WebSocket broadcast). + + Usage:: + + channel = PoliticsCommsChannel() + channel.register_handler(seat=1, handler=ai_timmy.on_broadcast) + channel.register_global_handler(narration_panel.on_broadcast) + + await channel.broadcast(ThreatBroadcast( + speaker=0, audience=-1, + broadcast_type=BroadcastType.CALL_TO_ARMS, + text="Focus Spike! He has a combo!", + turn=4, condition_seat=2, + )) + """ + + def __init__(self) -> None: + self._seat_handlers: dict[int, list[BroadcastHandler]] = {} + self._global_handlers: list[BroadcastHandler] = [] + self._log: list[ThreatBroadcast] = [] + + def register_handler( + self, + seat: int, + handler: BroadcastHandler, + ) -> None: + """Register a handler that receives broadcasts addressed to `seat`.""" + self._seat_handlers.setdefault(seat, []).append(handler) + + def register_global_handler( + self, + handler: BroadcastHandler, + ) -> None: + """Register a handler that receives ALL broadcasts.""" + self._global_handlers.append(handler) + + async def broadcast( + self, + msg: ThreatBroadcast, + ) -> None: + """ + Dispatch a broadcast to all relevant handlers. + Global handlers always fire. Seat handlers fire if audience matches + or if the broadcast is to the whole table (audience == -1). + """ + self._log.append(msg) + + for handler in self._global_handlers: + await handler(msg) + + if msg.audience == -1: + # Broadcast to all seats except speaker + for seat, handlers in self._seat_handlers.items(): + if seat == msg.speaker: + continue + for h in handlers: + await h(msg) + else: + for h in self._seat_handlers.get(msg.audience, []): + await h(msg) + + def recent_broadcasts( + self, + last_n: int = 10, + speaker: Optional[int] = None, + audience: Optional[int] = None, + ) -> list[ThreatBroadcast]: + """Return the last N broadcasts, optionally filtered.""" + results = [ + b for b in self._log + if (speaker is None or b.speaker == speaker) + and (audience is None or b.audience == audience or b.audience == -1) + ] + return results[-last_n:] + + def clear(self) -> None: + self._log.clear() diff --git a/src/commander_ai_lab/sim/politics/deals.py b/src/commander_ai_lab/sim/politics/deals.py new file mode 100644 index 0000000..a371ca2 --- /dev/null +++ b/src/commander_ai_lab/sim/politics/deals.py @@ -0,0 +1,100 @@ +""" +Commander AI Lab — Deal Data Classes (Phase 6) +=============================================== +Deals are time-limited, seat-to-seat agreements that the politics engine +can propose, counter, accept, or reject. They are stored in the +PoliticsEngine and checked at each phase transition. + +Deal types: + NON_AGGRESSION — neither party attacks the other for N turns + ATTACK_TOGETHER — both parties attack a named third seat this turn + SPARE_THIS_TURN — proposer asks to be left alone during one combat + TRADE_RESOURCES — abstract resource exchange (e.g. card draw for removal) + FREE_FOR_ALL — proposer calls for everyone to focus a specific seat +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum, auto +from typing import Optional + + +class DealType(Enum): + NON_AGGRESSION = auto() + ATTACK_TOGETHER = auto() + SPARE_THIS_TURN = auto() + TRADE_RESOURCES = auto() + FREE_FOR_ALL = auto() + + +class DealResponse(Enum): + PENDING = auto() + ACCEPTED = auto() + REJECTED = auto() + COUNTERED = auto() + EXPIRED = auto() + BROKEN = auto() + + +@dataclass +class Deal: + """ + A deal proposed from one seat to another (or broadcast to all). + + Attributes: + deal_id: Unique identifier (auto-assigned by PoliticsEngine). + proposer: Seat index of the proposing player. + target: Seat index of the deal target (-1 = all players). + deal_type: Type of deal. + duration_turns: How many turns the deal stays active once accepted. + subject_seat: For ATTACK_TOGETHER / FREE_FOR_ALL, the seat to focus. + flavor_text: Natural-language proposal text (spoken by AI). + turn_proposed: Game turn when proposed. + turn_expires: Turn after which this deal auto-expires if not resolved. + response: Current deal status. + responder: Seat that responded (may differ from target if broadcast). + counter_deal: A counter-proposal (if COUNTERED). + """ + proposer: int + target: int + deal_type: DealType + flavor_text: str = "" + duration_turns: int = 2 + subject_seat: Optional[int] = None + turn_proposed: int = 0 + turn_expires: int = 2 + response: DealResponse = DealResponse.PENDING + responder: Optional[int] = None + counter_deal: Optional[Deal] = None + deal_id: int = field(default_factory=lambda: Deal._next_id()) + + _id_counter: int = field(default=0, init=False, repr=False, compare=False) + + @staticmethod + def _next_id() -> int: + Deal.__dict__ # trigger class body + if not hasattr(Deal, '_counter'): + Deal._counter = 0 + Deal._counter += 1 + return Deal._counter + + def is_active(self, current_turn: int) -> bool: + """True if deal is accepted and not yet expired.""" + return ( + self.response == DealResponse.ACCEPTED + and current_turn <= self.turn_proposed + self.duration_turns + ) + + def is_pending(self, current_turn: int) -> bool: + return ( + self.response == DealResponse.PENDING + and current_turn <= self.turn_expires + ) + + def expire(self) -> None: + if self.response == DealResponse.PENDING: + self.response = DealResponse.EXPIRED + + def mark_broken(self) -> None: + self.response = DealResponse.BROKEN diff --git a/src/commander_ai_lab/sim/politics/engine.py b/src/commander_ai_lab/sim/politics/engine.py new file mode 100644 index 0000000..6630335 --- /dev/null +++ b/src/commander_ai_lab/sim/politics/engine.py @@ -0,0 +1,468 @@ +""" +Commander AI Lab — Politics Engine (Phase 6) +============================================= +Orchestrates all AI-to-AI (and AI-to-human) political interactions: + + 1. Deal proposal generation + - Each AI evaluates threat scores + targeting memory to decide + whether to propose a deal this turn. + - Proposal templates vary by AI personality. + + 2. Deal evaluation & response + - Receiving AI evaluates whether the deal is in their interest. + - Factors: current threat level from proposer, board state, personality. + + 3. Deal enforcement monitoring + - At each phase transition, PoliticsEngine checks active deals. + - If a participant acts against a deal, it is marked BROKEN and + the breach is recorded in TargetingMemory with a heavy penalty. + + 4. Spite targeting + - After a deal is broken, the wronged seat gets a spite_targets set + that biases combat and spell targeting toward the breaker. + - Spite decays over SPITE_DECAY_TURNS turns. + + 5. Call-to-arms + - When a seat's threat score crosses CALL_TO_ARMS_THRESHOLD, any AI + can broadcast a call-to-arms to the table via PoliticsCommsChannel. +""" + +from __future__ import annotations + +import random +from dataclasses import dataclass, field +from typing import Optional, Callable, Awaitable + +from commander_ai_lab.sim.politics.deals import Deal, DealType, DealResponse +from commander_ai_lab.sim.politics.memory import TargetingMemory, ActionType +from commander_ai_lab.sim.politics.comms import ( + PoliticsCommsChannel, ThreatBroadcast, BroadcastType, +) + +# --------------------------------------------------------------------------- +# Tuning constants +# --------------------------------------------------------------------------- +DEAL_PROPOSAL_COOLDOWN = 2 # turns between proposals per seat +SPITE_DECAY_TURNS = 3 # turns spite bonus persists +CALL_TO_ARMS_THRESHOLD = 0.75 # threat score that triggers a call-to-arms +DEAL_ACCEPT_BASE_PROB = 0.55 # base probability AI accepts a deal +PERSONALITY_DEAL_BIAS: dict[str, float] = { + "aggressive": -0.20, # less likely to accept deals + "control": 0.10, + "combo": 0.15, + "timmy": -0.10, + "spike": 0.05, + "johnny": 0.20, +} + + +# --------------------------------------------------------------------------- +# Proposal templates per deal type and personality +# --------------------------------------------------------------------------- +_PROPOSAL_TEMPLATES: dict[DealType, list[str]] = { + DealType.NON_AGGRESSION: [ + "How about we leave each other alone for a couple of turns?", + "I'll stay off your back if you stay off mine.", + "Non-aggression pact? I have bigger fish to fry.", + ], + DealType.ATTACK_TOGETHER: [ + "Let's both swing at P{target} — they're way ahead.", + "If we team up on P{target} this turn we can knock them out.", + "P{target} is the biggest threat. Join me?", + ], + DealType.SPARE_THIS_TURN: [ + "Leave me out of combat this turn and I won't touch you.", + "I'm tapped out — please spare me until my upkeep.", + "One turn of peace is all I ask.", + ], + DealType.FREE_FOR_ALL: [ + "Table, we need to focus P{target} RIGHT NOW.", + "P{target} is going to win if we don't all attack them.", + "Everyone pile on P{target} — deal?", + ], +} + + +@dataclass +class SpiteEntry: + target_seat: int + intensity: float # 0.0–1.0 targeting bias added + expires_turn: int + + +class PoliticsEngine: + """ + Central coordinator for the Phase 6 politics system. + + Parameters + ---------- + num_players : Number of seats (2–4). + memory : Shared TargetingMemory instance. + comms : Shared PoliticsCommsChannel instance. + personalities : Dict mapping seat index → personality string. + threat_fn : Callable(viewer_seat) → dict{seat: float} — + returns current threat scores from ThreatAssessor. + """ + + def __init__( + self, + num_players: int, + memory: TargetingMemory, + comms: PoliticsCommsChannel, + personalities: dict[int, str], + threat_fn: Optional[Callable[[int], dict[int, float]]] = None, + ) -> None: + self.num_players = num_players + self.memory = memory + self.comms = comms + self.personalities = personalities + self._threat_fn = threat_fn or (lambda s: {}) + + # Active deals: deal_id -> Deal + self._deals: dict[int, Deal] = {} + # Spite registry: seat -> list[SpiteEntry] + self._spite: dict[int, list[SpiteEntry]] = {i: [] for i in range(num_players)} + # Last proposal turn per seat + self._last_proposal_turn: dict[int, int] = {i: -999 for i in range(num_players)} + # Call-to-arms cooldown per seat + self._last_call_turn: dict[int, int] = {i: -999 for i in range(num_players)} + + # ------------------------------------------------------------------ + # Main per-turn hook (called at start of each main phase) + # ------------------------------------------------------------------ + + async def on_turn_start( + self, + active_seat: int, + current_turn: int, + game_state, + ) -> None: + """ + Called at the start of the active seat's main phase. + Runs: deal expiry, spite decay, proposal generation, call-to-arms. + """ + self._expire_deals(current_turn) + self._decay_spite(current_turn) + await self._maybe_propose_deal(active_seat, current_turn, game_state) + await self._maybe_call_to_arms(active_seat, current_turn) + + # ------------------------------------------------------------------ + # Deal lifecycle + # ------------------------------------------------------------------ + + async def propose_deal( + self, + proposer: int, + target: int, + deal_type: DealType, + current_turn: int, + subject_seat: Optional[int] = None, + duration_turns: int = 2, + ) -> Deal: + """ + Create and broadcast a deal proposal. + The target AI evaluates and responds automatically. + """ + flavor = self._generate_flavor(proposer, deal_type, subject_seat) + deal = Deal( + proposer=proposer, + target=target, + deal_type=deal_type, + flavor_text=flavor, + duration_turns=duration_turns, + subject_seat=subject_seat, + turn_proposed=current_turn, + turn_expires=current_turn + 1, + ) + self._deals[deal.deal_id] = deal + self._last_proposal_turn[proposer] = current_turn + + # Broadcast proposal to table + await self.comms.broadcast(ThreatBroadcast( + speaker=proposer, + audience=target, + broadcast_type=BroadcastType.OFFER, + text=flavor, + turn=current_turn, + condition_seat=subject_seat, + )) + + # Auto-evaluate if target is an AI seat + response = self._evaluate_deal(deal, current_turn) + deal.response = response + deal.responder = target + + resp_text = self._response_flavor(target, deal, response) + await self.comms.broadcast(ThreatBroadcast( + speaker=target, + audience=proposer, + broadcast_type=( + BroadcastType.OFFER if response == DealResponse.ACCEPTED + else BroadcastType.THREAT + ), + text=resp_text, + turn=current_turn, + )) + + return deal + + def check_deal_violation( + self, + actor: int, + action_target: int, + action_type: ActionType, + current_turn: int, + ) -> list[Deal]: + """ + Check whether an action by `actor` toward `action_target` breaks + any active deals. Returns list of broken deals. + """ + broken = [] + for deal in self._deals.values(): + if deal.response != DealResponse.ACCEPTED: + continue + if not deal.is_active(current_turn): + continue + + violated = False + if deal.deal_type == DealType.NON_AGGRESSION: + if actor in (deal.proposer, deal.target) and action_target in (deal.proposer, deal.target): + hostile = action_type in ( + ActionType.ATTACKED, ActionType.TARGETED_SPELL, + ActionType.REMOVED_PERMANENT, ActionType.COUNTERED_SPELL, + ) + violated = hostile + + elif deal.deal_type == DealType.SPARE_THIS_TURN: + if actor == deal.target and action_target == deal.proposer: + violated = action_type == ActionType.ATTACKED + + if violated: + deal.mark_broken() + victim = deal.proposer if actor == deal.target else deal.target + self.memory.record_deal_broken(current_turn, actor, victim) + self._add_spite(victim, actor, current_turn) + broken.append(deal) + + return broken + + def get_active_deals( + self, + seat: int, + current_turn: int, + ) -> list[Deal]: + """Return all active deals involving `seat`.""" + return [ + d for d in self._deals.values() + if d.is_active(current_turn) + and seat in (d.proposer, d.target) + ] + + # ------------------------------------------------------------------ + # Spite system + # ------------------------------------------------------------------ + + def get_spite_bias( + self, + viewer: int, + target: int, + current_turn: int, + ) -> float: + """ + Return the spite-targeting bias that `viewer` has toward `target`. + Range: 0.0 (no spite) – 1.0 (maximum spite bias). + """ + total = 0.0 + for entry in self._spite.get(viewer, []): + if entry.target_seat == target and entry.expires_turn >= current_turn: + total += entry.intensity + return min(total, 1.0) + + def adjusted_threat_score( + self, + viewer: int, + target: int, + current_turn: int, + ) -> float: + """ + Return threat score for `target` from `viewer`'s perspective, + blended with spite bias and memory aggression. + """ + base_threats = self._threat_fn(viewer) + base = base_threats.get(target, 0.0) + spite = self.get_spite_bias(viewer, target, current_turn) + mem = min(self.memory.aggression_score(target, viewer, current_turn) / 10.0, 0.3) + return min(base + spite * 0.3 + mem, 1.0) + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + async def _maybe_propose_deal( + self, + active_seat: int, + current_turn: int, + game_state, + ) -> None: + """Decide whether to propose a deal this turn.""" + cooldown = self._last_proposal_turn.get(active_seat, -999) + if current_turn - cooldown < DEAL_PROPOSAL_COOLDOWN: + return + + threats = self._threat_fn(active_seat) + if not threats: + return + + top_threat_seat = max(threats, key=threats.get) + top_score = threats[top_threat_seat] + + # Propose a non-aggression to the second-most-threatening seat + seats_sorted = sorted( + [s for s in threats if s != active_seat], + key=lambda s: threats.get(s, 0), + ) + if len(seats_sorted) < 2: + return + + partner_seat = seats_sorted[0] # least threatening — most likely to agree + p = random.random() + if p > 0.40: # 60% chance to try + return + + if top_score > CALL_TO_ARMS_THRESHOLD: + await self.propose_deal( + proposer=active_seat, + target=partner_seat, + deal_type=DealType.ATTACK_TOGETHER, + current_turn=current_turn, + subject_seat=top_threat_seat, + ) + else: + await self.propose_deal( + proposer=active_seat, + target=partner_seat, + deal_type=DealType.NON_AGGRESSION, + current_turn=current_turn, + ) + + async def _maybe_call_to_arms( + self, + active_seat: int, + current_turn: int, + ) -> None: + """Broadcast a call-to-arms if any threat exceeds threshold.""" + cooldown = self._last_call_turn.get(active_seat, -999) + if current_turn - cooldown < 3: + return + + threats = self._threat_fn(active_seat) + for seat, score in threats.items(): + if seat == active_seat: + continue + if score >= CALL_TO_ARMS_THRESHOLD: + template = random.choice(_PROPOSAL_TEMPLATES[DealType.FREE_FOR_ALL]) + text = template.format(target=seat) + await self.comms.broadcast(ThreatBroadcast( + speaker=active_seat, + audience=-1, + broadcast_type=BroadcastType.CALL_TO_ARMS, + text=text, + turn=current_turn, + condition_seat=seat, + )) + self._last_call_turn[active_seat] = current_turn + break + + def _evaluate_deal( + self, + deal: Deal, + current_turn: int, + ) -> DealResponse: + """AI auto-evaluation of a deal proposal.""" + target = deal.target + if target < 0: # broadcast deal — accept probabilistically + return DealResponse.ACCEPTED if random.random() < 0.5 else DealResponse.REJECTED + + personality = self.personalities.get(target, "spike") + bias = PERSONALITY_DEAL_BIAS.get(personality, 0.0) + prob = DEAL_ACCEPT_BASE_PROB + bias + + # Bonus: accept more if proposer has been targeting you heavily + aggression = self.memory.aggression_score( + deal.proposer, target, current_turn + ) + if aggression > 5.0: + prob -= 0.20 # suspicious of sudden peace offer + elif aggression < 1.0: + prob += 0.10 # no recent hostility — trust higher + + # Bonus: accept attack-together if target is actually threatening + if deal.deal_type == DealType.ATTACK_TOGETHER and deal.subject_seat is not None: + threats = self._threat_fn(target) + subject_score = threats.get(deal.subject_seat, 0.0) + prob += subject_score * 0.3 + + return DealResponse.ACCEPTED if random.random() < prob else DealResponse.REJECTED + + def _expire_deals( + self, + current_turn: int, + ) -> None: + for deal in self._deals.values(): + if deal.is_pending(current_turn) is False and deal.response == DealResponse.PENDING: + deal.expire() + + def _decay_spite( + self, + current_turn: int, + ) -> None: + for seat in self._spite: + self._spite[seat] = [ + e for e in self._spite[seat] + if e.expires_turn >= current_turn + ] + + def _add_spite( + self, + wronged_seat: int, + target_seat: int, + current_turn: int, + intensity: float = 0.6, + ) -> None: + self._spite[wronged_seat].append(SpiteEntry( + target_seat=target_seat, + intensity=intensity, + expires_turn=current_turn + SPITE_DECAY_TURNS, + )) + + @staticmethod + def _generate_flavor( + proposer: int, + deal_type: DealType, + subject_seat: Optional[int], + ) -> str: + templates = _PROPOSAL_TEMPLATES.get(deal_type, ["Let's make a deal."]) + text = random.choice(templates) + if subject_seat is not None: + text = text.format(target=f"P{subject_seat}") + return text + + @staticmethod + def _response_flavor( + responder: int, + deal: Deal, + response: DealResponse, + ) -> str: + if response == DealResponse.ACCEPTED: + options = [ + "Deal. Don't make me regret this.", + "Fine, I'll hold off.", + "Agreed. For now.", + ] + else: + options = [ + "No deal. I'll handle things myself.", + "I don't trust you.", + "Pass. You're next anyway.", + ] + return random.choice(options) diff --git a/src/commander_ai_lab/sim/politics/memory.py b/src/commander_ai_lab/sim/politics/memory.py new file mode 100644 index 0000000..032f39d --- /dev/null +++ b/src/commander_ai_lab/sim/politics/memory.py @@ -0,0 +1,259 @@ +""" +Commander AI Lab — Targeting Memory (Phase 6) +============================================== +Per-game record of every targeting decision: who cast spells/abilities +targeting whom, who attacked whom, who blocked whose creatures, and +who broke deals. + +The memory feeds threat assessments in ThreatAssessor and politics +decisions in PoliticsEngine, letting AI personas hold grudges, +reciprocate kindness, or shift focus as the game progresses. + +Decay model: + Events older than MEMORY_DECAY_TURNS have their weight halved per turn + beyond that threshold, so recent aggression matters more than old actions. +""" + +from __future__ import annotations + +from collections import defaultdict +from dataclasses import dataclass, field +from enum import Enum, auto +from typing import Optional + + +MEMORY_DECAY_TURNS = 4 # turns before weight decay kicks in +DECAY_FACTOR = 0.5 # per-turn weight multiplier after decay threshold + + +class ActionType(Enum): + TARGETED_SPELL = auto() # single-target spell/ability + ATTACKED = auto() # declared attacker against + BLOCKED = auto() # blocked a creature of + DEALT_COMBAT_DAMAGE = auto() # connected combat damage to + REMOVED_PERMANENT = auto() # destroyed / exiled permanent of + COUNTERED_SPELL = auto() # countered a spell of + STOLE_PERMANENT = auto() # gained control of permanent of + HELPED = auto() # beneficial action toward (negative aggression) + BROKE_DEAL = auto() # violated an accepted deal with + HONORED_DEAL = auto() # complied with an accepted deal with + + +# Aggression weights per action type (positive = hostile, negative = friendly) +ACTION_WEIGHTS: dict[ActionType, float] = { + ActionType.TARGETED_SPELL: 1.0, + ActionType.ATTACKED: 1.5, + ActionType.BLOCKED: 0.5, + ActionType.DEALT_COMBAT_DAMAGE: 2.0, + ActionType.REMOVED_PERMANENT: 2.5, + ActionType.COUNTERED_SPELL: 2.0, + ActionType.STOLE_PERMANENT: 3.0, + ActionType.HELPED: -1.5, + ActionType.BROKE_DEAL: 4.0, + ActionType.HONORED_DEAL: -1.0, +} + + +@dataclass +class MemoryEvent: + turn: int + actor: int # seat doing the action + target: int # seat being acted upon + action_type: ActionType + detail: str = "" # e.g. card name + weight: float = 1.0 # base weight (modified by decay at query time) + + +class TargetingMemory: + """ + Tracks all inter-player targeting events for the duration of a game. + + Usage:: + + mem = TargetingMemory(num_players=4) + mem.record(turn=3, actor=1, target=0, + action_type=ActionType.ATTACKED, detail="declared attacker") + score = mem.aggression_score(actor=1, target=0, current_turn=5) + ranking = mem.most_aggressive_toward(viewer=0, current_turn=5) + """ + + def __init__(self, num_players: int = 4) -> None: + self.num_players = num_players + self._events: list[MemoryEvent] = [] + # Pre-computed cache (invalidated on each record()) + self._cache: dict[tuple, float] = {} + self._cache_turn: int = -1 + + # ------------------------------------------------------------------ + # Recording + # ------------------------------------------------------------------ + + def record( + self, + turn: int, + actor: int, + target: int, + action_type: ActionType, + detail: str = "", + weight_multiplier: float = 1.0, + ) -> None: + """ + Record a new targeting event. + + Args: + weight_multiplier: Extra multiplier (e.g. 2.0 for a Wrath effect + that hits everyone). + """ + base_w = ACTION_WEIGHTS.get(action_type, 1.0) * weight_multiplier + self._events.append(MemoryEvent( + turn=turn, + actor=actor, + target=target, + action_type=action_type, + detail=detail, + weight=base_w, + )) + self._cache_turn = -1 # invalidate cache + + def record_attack(self, turn: int, attacker: int, defender: int) -> None: + self.record(turn, attacker, defender, ActionType.ATTACKED) + + def record_damage(self, turn: int, dealer: int, receiver: int, amount: int) -> None: + self.record(turn, dealer, receiver, ActionType.DEALT_COMBAT_DAMAGE, + detail=f"{amount} damage", weight_multiplier=min(amount / 5.0, 3.0)) + + def record_deal_broken(self, turn: int, breaker: int, victim: int) -> None: + self.record(turn, breaker, victim, ActionType.BROKE_DEAL) + + def record_help(self, turn: int, helper: int, beneficiary: int) -> None: + self.record(turn, helper, beneficiary, ActionType.HELPED) + + # ------------------------------------------------------------------ + # Querying + # ------------------------------------------------------------------ + + def aggression_score( + self, + actor: int, + target: int, + current_turn: int, + ) -> float: + """ + Return the total weighted aggression score from actor toward target. + Higher = more hostile. Can be negative if actor has been very helpful. + Applies temporal decay for events older than MEMORY_DECAY_TURNS. + """ + total = 0.0 + for ev in self._events: + if ev.actor != actor or ev.target != target: + continue + age = current_turn - ev.turn + decay = 1.0 if age <= MEMORY_DECAY_TURNS else ( + DECAY_FACTOR ** (age - MEMORY_DECAY_TURNS) + ) + total += ev.weight * decay + return round(total, 3) + + def mutual_aggression( + self, + seat_a: int, + seat_b: int, + current_turn: int, + ) -> float: + """Combined aggression in both directions between two seats.""" + return ( + self.aggression_score(seat_a, seat_b, current_turn) + + self.aggression_score(seat_b, seat_a, current_turn) + ) + + def most_aggressive_toward( + self, + viewer: int, + current_turn: int, + ) -> list[tuple[int, float]]: + """ + Return seats ranked by their aggression toward `viewer`, descending. + Returns list of (seat, score) excluding viewer's own seat. + """ + scores = [ + (s, self.aggression_score(s, viewer, current_turn)) + for s in range(self.num_players) + if s != viewer + ] + return sorted(scores, key=lambda x: x[1], reverse=True) + + def biggest_threat_overall( + self, + viewer: int, + current_turn: int, + ) -> Optional[int]: + """ + Return the seat that has been most aggressive toward `viewer`. + Returns None if no events recorded yet. + """ + ranking = self.most_aggressive_toward(viewer, current_turn) + if not ranking: + return None + seat, score = ranking[0] + return seat if score > 0 else None + + def recent_actions( + self, + last_n_turns: int, + current_turn: int, + actor: Optional[int] = None, + target: Optional[int] = None, + ) -> list[MemoryEvent]: + """ + Return events from the last N turns, optionally filtered. + """ + cutoff = current_turn - last_n_turns + return [ + ev for ev in self._events + if ev.turn >= cutoff + and (actor is None or ev.actor == actor) + and (target is None or ev.target == target) + ] + + def action_counts( + self, + actor: int, + target: int, + ) -> dict[ActionType, int]: + """Count each action type from actor toward target.""" + counts: dict[ActionType, int] = defaultdict(int) + for ev in self._events: + if ev.actor == actor and ev.target == target: + counts[ev.action_type] += 1 + return dict(counts) + + def total_damage_dealt( + self, + dealer: int, + receiver: int, + ) -> float: + """Sum weight-scaled damage events from dealer to receiver.""" + return sum( + ev.weight + for ev in self._events + if ev.actor == dealer + and ev.target == receiver + and ev.action_type == ActionType.DEALT_COMBAT_DAMAGE + ) + + def summary( + self, + current_turn: int, + ) -> dict[tuple[int, int], float]: + """ + Return a full NxN aggression matrix as {(actor, target): score}. + """ + result = {} + for a in range(self.num_players): + for t in range(self.num_players): + if a == t: + continue + s = self.aggression_score(a, t, current_turn) + if s != 0: + result[(a, t)] = s + return result diff --git a/tests/test_phase6.py b/tests/test_phase6.py new file mode 100644 index 0000000..97d4233 --- /dev/null +++ b/tests/test_phase6.py @@ -0,0 +1,330 @@ +""" +Phase 6 Unit Tests — Politics Engine & Targeting Memory +======================================================= +Run with: pytest tests/test_phase6.py -v +""" + +from __future__ import annotations + +import asyncio +import pytest + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _mem(n=4): + from commander_ai_lab.sim.politics.memory import TargetingMemory + return TargetingMemory(num_players=n) + + +def _engine(personalities=None): + from commander_ai_lab.sim.politics.memory import TargetingMemory + from commander_ai_lab.sim.politics.comms import PoliticsCommsChannel + from commander_ai_lab.sim.politics.engine import PoliticsEngine + mem = TargetingMemory(num_players=4) + comms = PoliticsCommsChannel() + perso = personalities or {0: "timmy", 1: "spike", 2: "johnny", 3: "aggressive"} + return PoliticsEngine( + num_players=4, + memory=mem, + comms=comms, + personalities=perso, + threat_fn=lambda s: {i: 0.3 for i in range(4) if i != s}, + ), mem, comms + + +def _run(coro): + return asyncio.get_event_loop().run_until_complete(coro) + + +# --------------------------------------------------------------------------- +# TargetingMemory +# --------------------------------------------------------------------------- + +class TestTargetingMemory: + def test_initial_aggression_zero(self): + mem = _mem() + assert mem.aggression_score(0, 1, current_turn=1) == 0.0 + + def test_record_increases_aggression(self): + from commander_ai_lab.sim.politics.memory import ActionType + mem = _mem() + mem.record(turn=1, actor=1, target=0, action_type=ActionType.ATTACKED) + assert mem.aggression_score(1, 0, current_turn=1) > 0 + + def test_different_directions_independent(self): + from commander_ai_lab.sim.politics.memory import ActionType + mem = _mem() + mem.record(1, 0, 1, ActionType.ATTACKED) + assert mem.aggression_score(1, 0, current_turn=1) == 0.0 + assert mem.aggression_score(0, 1, current_turn=1) > 0 + + def test_decay_reduces_score(self): + from commander_ai_lab.sim.politics.memory import ActionType, MEMORY_DECAY_TURNS + mem = _mem() + mem.record(turn=1, actor=1, target=0, action_type=ActionType.ATTACKED) + score_fresh = mem.aggression_score(1, 0, current_turn=2) + score_old = mem.aggression_score(1, 0, current_turn=2 + MEMORY_DECAY_TURNS + 3) + assert score_old < score_fresh + + def test_help_reduces_score(self): + from commander_ai_lab.sim.politics.memory import ActionType + mem = _mem() + mem.record(1, 1, 0, ActionType.ATTACKED) + mem.record(2, 1, 0, ActionType.HELPED) + score = mem.aggression_score(1, 0, current_turn=2) + score_attack_only = 1.5 # weight of one ATTACKED event + assert score < score_attack_only + + def test_most_aggressive_toward_sorted(self): + from commander_ai_lab.sim.politics.memory import ActionType + mem = _mem() + mem.record(1, 1, 0, ActionType.ATTACKED) + mem.record(1, 1, 0, ActionType.ATTACKED) + mem.record(1, 2, 0, ActionType.TARGETED_SPELL) + ranking = mem.most_aggressive_toward(viewer=0, current_turn=2) + seats = [s for s, _ in ranking] + assert seats[0] == 1 # seat 1 attacked twice + + def test_biggest_threat_none_when_empty(self): + mem = _mem() + assert mem.biggest_threat_overall(0, current_turn=1) is None + + def test_record_damage_scales_weight(self): + from commander_ai_lab.sim.politics.memory import ActionType + mem = _mem() + mem.record_damage(1, 1, 0, amount=10) + score = mem.aggression_score(1, 0, current_turn=1) + assert score > 0 + + def test_deal_broken_heavy_weight(self): + from commander_ai_lab.sim.politics.memory import ActionType, ACTION_WEIGHTS + assert ACTION_WEIGHTS[ActionType.BROKE_DEAL] >= 4.0 + + def test_summary_returns_matrix(self): + from commander_ai_lab.sim.politics.memory import ActionType + mem = _mem(4) + mem.record(1, 0, 1, ActionType.ATTACKED) + summary = mem.summary(current_turn=2) + assert (0, 1) in summary + + def test_action_counts(self): + from commander_ai_lab.sim.politics.memory import ActionType + mem = _mem() + mem.record(1, 0, 2, ActionType.ATTACKED) + mem.record(2, 0, 2, ActionType.ATTACKED) + counts = mem.action_counts(0, 2) + assert counts.get(ActionType.ATTACKED) == 2 + + def test_recent_actions_filter(self): + from commander_ai_lab.sim.politics.memory import ActionType + mem = _mem() + mem.record(1, 0, 1, ActionType.ATTACKED) + mem.record(5, 0, 1, ActionType.TARGETED_SPELL) + recent = mem.recent_actions(last_n_turns=2, current_turn=6) + assert len(recent) == 1 + + +# --------------------------------------------------------------------------- +# Deal data classes +# --------------------------------------------------------------------------- + +class TestDeal: + def test_is_active_accepted(self): + from commander_ai_lab.sim.politics.deals import Deal, DealType, DealResponse + d = Deal(proposer=0, target=1, deal_type=DealType.NON_AGGRESSION, + turn_proposed=3, duration_turns=2) + d.response = DealResponse.ACCEPTED + assert d.is_active(current_turn=4) is True + assert d.is_active(current_turn=6) is False + + def test_is_pending(self): + from commander_ai_lab.sim.politics.deals import Deal, DealType + d = Deal(proposer=0, target=1, deal_type=DealType.NON_AGGRESSION, + turn_proposed=1, turn_expires=2) + assert d.is_pending(current_turn=2) is True + assert d.is_pending(current_turn=3) is False + + def test_expire(self): + from commander_ai_lab.sim.politics.deals import Deal, DealType, DealResponse + d = Deal(proposer=0, target=1, deal_type=DealType.NON_AGGRESSION) + d.expire() + assert d.response == DealResponse.EXPIRED + + def test_mark_broken(self): + from commander_ai_lab.sim.politics.deals import Deal, DealType, DealResponse + d = Deal(proposer=0, target=1, deal_type=DealType.NON_AGGRESSION) + d.response = DealResponse.ACCEPTED + d.mark_broken() + assert d.response == DealResponse.BROKEN + + def test_unique_ids(self): + from commander_ai_lab.sim.politics.deals import Deal, DealType + d1 = Deal(0, 1, DealType.NON_AGGRESSION) + d2 = Deal(1, 2, DealType.SPARE_THIS_TURN) + assert d1.deal_id != d2.deal_id + + +# --------------------------------------------------------------------------- +# PoliticsCommsChannel +# --------------------------------------------------------------------------- + +class TestPoliticsCommsChannel: + def test_register_and_receive(self): + from commander_ai_lab.sim.politics.comms import ( + PoliticsCommsChannel, ThreatBroadcast, BroadcastType, + ) + channel = PoliticsCommsChannel() + received = [] + + async def handler(msg): received.append(msg) + channel.register_handler(seat=1, handler=handler) + + msg = ThreatBroadcast(speaker=0, audience=1, + broadcast_type=BroadcastType.THREAT, + text="Watch out", turn=1) + _run(channel.broadcast(msg)) + assert len(received) == 1 + + def test_global_handler_receives_all(self): + from commander_ai_lab.sim.politics.comms import ( + PoliticsCommsChannel, ThreatBroadcast, BroadcastType, + ) + channel = PoliticsCommsChannel() + received = [] + + async def handler(msg): received.append(msg) + channel.register_global_handler(handler) + + for audience in [1, 2, -1]: + msg = ThreatBroadcast(speaker=0, audience=audience, + broadcast_type=BroadcastType.TAUNT, + text="Hello", turn=1) + _run(channel.broadcast(msg)) + assert len(received) == 3 + + def test_table_broadcast_skips_speaker(self): + from commander_ai_lab.sim.politics.comms import ( + PoliticsCommsChannel, ThreatBroadcast, BroadcastType, + ) + channel = PoliticsCommsChannel() + received_seats = [] + + for seat in range(4): + async def make_handler(s=seat): + async def h(msg): received_seats.append(s) + return h + import asyncio + h = asyncio.get_event_loop().run_until_complete(make_handler()) + channel.register_handler(seat=seat, handler=h) + + msg = ThreatBroadcast(speaker=0, audience=-1, + broadcast_type=BroadcastType.CALL_TO_ARMS, + text="Focus P2!", turn=2) + _run(channel.broadcast(msg)) + assert 0 not in received_seats # speaker does not receive own broadcast + + def test_recent_broadcasts(self): + from commander_ai_lab.sim.politics.comms import ( + PoliticsCommsChannel, ThreatBroadcast, BroadcastType, + ) + channel = PoliticsCommsChannel() + for i in range(15): + msg = ThreatBroadcast(0, -1, BroadcastType.TAUNT, f"msg{i}", turn=i) + _run(channel.broadcast(msg)) + recent = channel.recent_broadcasts(last_n=5) + assert len(recent) == 5 + + +# --------------------------------------------------------------------------- +# PoliticsEngine +# --------------------------------------------------------------------------- + +class TestPoliticsEngine: + def test_propose_deal_creates_entry(self): + from commander_ai_lab.sim.politics.deals import DealType + engine, mem, comms = _engine() + deal = _run(engine.propose_deal(0, 1, DealType.NON_AGGRESSION, current_turn=2)) + assert deal.deal_id in engine._deals + + def test_deal_gets_response(self): + from commander_ai_lab.sim.politics.deals import DealType, DealResponse + engine, mem, comms = _engine() + deal = _run(engine.propose_deal(0, 1, DealType.NON_AGGRESSION, current_turn=2)) + assert deal.response in (DealResponse.ACCEPTED, DealResponse.REJECTED) + + def test_active_deals_returned(self): + from commander_ai_lab.sim.politics.deals import DealType, DealResponse + engine, mem, comms = _engine() + deal = _run(engine.propose_deal(0, 1, DealType.NON_AGGRESSION, current_turn=2)) + deal.response = DealResponse.ACCEPTED + active = engine.get_active_deals(seat=0, current_turn=2) + assert any(d.deal_id == deal.deal_id for d in active) + + def test_deal_violation_marks_broken(self): + from commander_ai_lab.sim.politics.deals import DealType, DealResponse + from commander_ai_lab.sim.politics.memory import ActionType + engine, mem, comms = _engine() + deal = _run(engine.propose_deal(0, 1, DealType.NON_AGGRESSION, current_turn=2)) + deal.response = DealResponse.ACCEPTED + broken = engine.check_deal_violation( + actor=1, action_target=0, + action_type=ActionType.ATTACKED, + current_turn=3, + ) + assert len(broken) == 1 + assert broken[0].response == DealResponse.BROKEN + + def test_spite_set_after_violation(self): + from commander_ai_lab.sim.politics.deals import DealType, DealResponse + from commander_ai_lab.sim.politics.memory import ActionType + engine, mem, comms = _engine() + deal = _run(engine.propose_deal(0, 1, DealType.NON_AGGRESSION, current_turn=2)) + deal.response = DealResponse.ACCEPTED + engine.check_deal_violation(1, 0, ActionType.ATTACKED, current_turn=3) + spite = engine.get_spite_bias(viewer=0, target=1, current_turn=3) + assert spite > 0 + + def test_spite_decays_after_turns(self): + from commander_ai_lab.sim.politics.deals import DealType, DealResponse + from commander_ai_lab.sim.politics.memory import ActionType + from commander_ai_lab.sim.politics.engine import SPITE_DECAY_TURNS + engine, mem, comms = _engine() + deal = _run(engine.propose_deal(0, 1, DealType.NON_AGGRESSION, current_turn=2)) + deal.response = DealResponse.ACCEPTED + engine.check_deal_violation(1, 0, ActionType.ATTACKED, current_turn=3) + engine._decay_spite(current_turn=3 + SPITE_DECAY_TURNS + 1) + spite = engine.get_spite_bias(viewer=0, target=1, + current_turn=3 + SPITE_DECAY_TURNS + 1) + assert spite == 0.0 + + def test_adjusted_threat_includes_spite(self): + from commander_ai_lab.sim.politics.deals import DealType, DealResponse + from commander_ai_lab.sim.politics.memory import ActionType + engine, mem, comms = _engine() + base_score = engine.adjusted_threat_score(viewer=0, target=1, current_turn=2) + # Add spite + engine._add_spite(wronged_seat=0, target_seat=1, current_turn=2, intensity=0.6) + spite_score = engine.adjusted_threat_score(viewer=0, target=1, current_turn=2) + assert spite_score > base_score + + def test_no_deal_within_cooldown(self): + from commander_ai_lab.sim.politics.engine import DEAL_PROPOSAL_COOLDOWN + engine, mem, comms = _engine() + engine._last_proposal_turn[0] = 5 + count_before = len(engine._deals) + + class _FakeGS: + players = [type('P', (), {'eliminated': False, 'life': 40})() for _ in range(4)] + + _run(engine._maybe_propose_deal(0, current_turn=5 + DEAL_PROPOSAL_COOLDOWN - 1, + game_state=_FakeGS())) + assert len(engine._deals) == count_before + + def test_comms_log_populated_after_proposal(self): + from commander_ai_lab.sim.politics.deals import DealType + engine, mem, comms = _engine() + _run(engine.propose_deal(0, 1, DealType.SPARE_THIS_TURN, current_turn=3)) + assert len(comms._log) >= 2 # proposal + response