detect and resolve battle tie with influence_token

src/warchron/model/battle.py

@@ -19,17 +19,17 @@ class Battle:
         self.victory_condition: str | None = None
         self.comment: str | None = None
 
-    def set_id(self, new_id: str) -> None:
-        self.sector_id = new_id
+    def set_sector(self, new_sector_id: str) -> None:
+        self.sector_id = new_sector_id
 
-    def set_player_1(self, new_player_id: str | None) -> None:
-        self.player_1_id = new_player_id
+    def set_player_1(self, new_camp_part_id: str | None) -> None:
+        self.player_1_id = new_camp_part_id
 
-    def set_player_2(self, new_player_id: str | None) -> None:
-        self.player_2_id = new_player_id
+    def set_player_2(self, new_camp_part_id: str | None) -> None:
+        self.player_2_id = new_camp_part_id
 
-    def set_winner(self, new_player_id: str | None) -> None:
-        self.winner_id = new_player_id
+    def set_winner(self, new_camp_part_id: str | None) -> None:
+        self.winner_id = new_camp_part_id
 
     def set_score(self, new_score: str | None) -> None:
         self.score = new_score

src/warchron/model/campaign_participant.py

@@ -15,8 +15,8 @@ class CampaignParticipant:
     def set_id(self, new_id: str) -> None:
         self.id = new_id
 
-    def set_war_participant(self, new_participant: str) -> None:
-        self.war_participant_id = new_participant
+    def set_war_participant(self, new_war_part_id: str) -> None:
+        self.war_participant_id = new_war_part_id
 
     def set_leader(self, new_faction: str) -> None:
         self.leader = new_faction

src/warchron/model/choice.py

@@ -18,8 +18,8 @@ class Choice:
         )
         self.comment: str | None = None
 
-    def set_id(self, new_id: str) -> None:
-        self.participant_id = new_id
+    def set_participant(self, new_camp_part_id: str) -> None:
+        self.participant_id = new_camp_part_id
 
     def set_priority(self, new_priority_id: str | None) -> None:
         self.priority_sector_id = new_priority_id

src/warchron/model/closure_service.py

@@ -2,44 +2,69 @@ from __future__ import annotations
 from typing import List
 
 from warchron.constants import ContextType
-from warchron.model.tie_manager import ResolutionContext
+from warchron.model.exception import ForbiddenOperation
+from warchron.model.tie_manager import TieResolver
+from warchron.model.war_event import InfluenceGained
 from warchron.model.war import War
 from warchron.model.campaign import Campaign
 from warchron.model.round import Round
+from warchron.model.battle import Battle
 
 
 class ClosureService:
 
-    @staticmethod
-    def close_round(round: Round) -> List[ResolutionContext]:
-        if not round.all_battles_finished():
-            raise RuntimeError("All battles must be finished to close their round")
-        ties = []
-        for battle in round.battles.values():
-            if battle.is_draw():
-                participants: list[str] = []
-                if battle.player_1_id is not None:
-                    participants.append(battle.player_1_id)
-                if battle.player_2_id is not None:
-                    participants.append(battle.player_2_id)
-                ties.append(
-                    ResolutionContext(
-                        context_type=ContextType.BATTLE,
-                        context_id=battle.sector_id,
-                        # TODO ref to War.participants at some point
-                        participant_ids=participants,
-                    )
-                )
-        if ties:
-            return ties
-        round.is_over = True
-        return []
+    # Round methods
 
     @staticmethod
-    def close_campaign(campaign: Campaign) -> List[ResolutionContext]:
+    def check_round_closable(round: Round) -> None:
+        if round.is_over:
+            raise ForbiddenOperation("Round already closed")
+        if not round.all_battles_finished():
+            raise ForbiddenOperation(
+                "All battles must be finished to close their round"
+            )
+
+    @staticmethod
+    def apply_battle_outcomes(war: War, campaign: Campaign, battle: Battle) -> None:
+        already_granted = any(
+            isinstance(e, InfluenceGained) and e.source == f"battle:{battle.sector_id}"
+            for e in war.events
+        )
+        if already_granted:
+            return
+        if battle.winner_id is not None:
+            base_winner = campaign.participants[battle.winner_id].war_participant_id
+        else:
+            base_winner = None
+        effective_winner = TieResolver.get_effective_winner_id(
+            war,
+            ContextType.BATTLE,
+            battle.sector_id,
+            base_winner,
+        )
+        if effective_winner is None:
+            return
+        sector = campaign.sectors[battle.sector_id]
+        if sector.influence_objective_id and war.influence_token:
+            war.events.append(
+                InfluenceGained(
+                    participant_id=effective_winner,
+                    amount=1,
+                    source=f"battle:{battle.sector_id}",
+                )
+            )
+
+    @staticmethod
+    def finalize_round(round: Round) -> None:
+        round.is_over = True
+
+    # Campaign methods
+
+    @staticmethod
+    def close_campaign(campaign: Campaign) -> List[str]:
         if not campaign.all_rounds_finished():
             raise RuntimeError("All rounds must be finished to close their campaign")
-        ties: List[ResolutionContext] = []
+        ties: List[str] = []
         # for round in campaign.rounds:
         #     # compute score
         #     # if participants have same score
@@ -59,11 +84,13 @@ class ClosureService:
         campaign.is_over = True
         return []
 
+    # War methods
+
     @staticmethod
-    def close_war(war: War) -> List[ResolutionContext]:
+    def close_war(war: War) -> List[str]:
         if not war.all_campaigns_finished():
             raise RuntimeError("All campaigns must be finished to close their war")
-        ties: List[ResolutionContext] = []
+        ties: List[str] = []
         # for campaign in war.campaigns:
         #     # compute score
         #     # if participants have same score

src/warchron/model/closure_workflow.py

@@ -1,19 +0,0 @@
-from typing import TYPE_CHECKING
-
-if TYPE_CHECKING:
-    from warchron.model.war import War
-    from warchron.model.war import War
-    from warchron.model.war import War
-    from warchron.model.closure_service import ClosureService
-
-
-class RoundClosureWorkflow:
-
-    def close_round(self, round_id):
-        rnd = repo.get_round(round_id)
-
-        ties = ClosureService.close_round(rnd)
-
-        repo.save()
-
-        return ties

src/warchron/model/influence_service.py

@@ -1,23 +0,0 @@
-from warchron.model.war import War
-from warchron.model.campaign import Campaign
-from warchron.model.battle import Battle
-from warchron.model.war_event import InfluenceGained
-
-
-class InfluenceService:
-
-    @staticmethod
-    def apply_battle_result(war: War, campaign: Campaign, battle: Battle) -> None:
-        if battle.winner_id is None:
-            return
-        sector = campaign.sectors[battle.sector_id]
-        # if sector grants influence
-        if sector.influence_objective_id and war.influence_token:
-            participant = war.participants[battle.winner_id]
-            participant.events.append(
-                InfluenceGained(
-                    participant_id=participant.id,
-                    amount=1,
-                    source=f"battle:{battle.sector_id}",
-                )
-            )

src/warchron/model/score_service.py

@@ -1,4 +1,4 @@
-from typing import TYPE_CHECKING
+from typing import Dict, TYPE_CHECKING
 
 if TYPE_CHECKING:
     from warchron.model.war import War
@@ -23,8 +23,8 @@ class ScoreService:
     @staticmethod
     def compute_narrative_points_for_participant(
         war: "War", participant_id: str
-    ) -> dict[str, int]:
-        totals: dict[str, int] = {}
+    ) -> Dict[str, int]:
+        totals: Dict[str, int] = {}
         for obj_id in war.objectives:
             totals[obj_id] = 0
         for campaign in war.campaigns:

src/warchron/model/tie_manager.py

@@ -1,46 +1,108 @@
+from typing import List, Dict
+
+from warchron.constants import ContextType
+from warchron.model.exception import ForbiddenOperation
 from warchron.model.war import War
-from warchron.model.war_event import InfluenceSpent
-
-
-class ResolutionContext:
-    def __init__(self, context_type: str, context_id: str, participant_ids: list[str]):
-        self.context_type = context_type
-        self.context_id = context_id
-        self.participant_ids = participant_ids
-
-        self.current_bids: dict[str, int] = {}
-        self.round_index: int = 0
-        self.is_resolved: bool = False
+from warchron.model.round import Round
+from warchron.model.battle import Battle
+from warchron.model.war_event import InfluenceSpent, TieResolved
 
 
 class TieResolver:
 
     @staticmethod
-    def resolve(
+    def find_round_ties(round: Round, war: War) -> List[Battle]:
+        ties = []
+        for battle in round.battles.values():
+            if not battle.is_draw():
+                continue
+            resolved = any(
+                isinstance(e, TieResolved)
+                and e.context_type == ContextType.BATTLE
+                and e.context_id == battle.sector_id
+                for e in war.events
+            )
+            if not resolved:
+                ties.append(battle)
+        return ties
+
+    @staticmethod
+    def apply_bids(
         war: War,
-        context: ResolutionContext,
-        bids: dict[str, int],
-    ) -> str | None:
-        # verify available token for each player
-        for pid, amount in bids.items():
-            participant = war.participants[pid]
-            if participant.influence_tokens() < amount:
-                raise ValueError("Not enough influence tokens")
-        # apply spending
-        for pid, amount in bids.items():
-            if amount > 0:
-                war.participants[pid].events.append(
-                    InfluenceSpent(
-                        participant_id=pid,
-                        amount=amount,
-                        context=context.context_type,
-                    )
+        context_type: ContextType,
+        context_id: str,
+        bids: Dict[str, bool],  # war_participant_id -> spend?
+    ) -> None:
+        for war_part_id, spend in bids.items():
+            if not spend:
+                continue
+            if war.get_influence_tokens(war_part_id) < 1:
+                raise ForbiddenOperation("Not enough tokens")
+            war.events.append(
+                InfluenceSpent(
+                    participant_id=war_part_id,
+                    amount=1,
+                    context_type=context_type,
                 )
-        # determine winner
-        max_bid = max(bids.values())
-        winners = [pid for pid, b in bids.items() if b == max_bid]
-        if len(winners) == 1:
-            context.is_resolved = True
-            return winners[0]
-        # persisting tie → None
+            )
+
+    @staticmethod
+    def try_tie_break(
+        war: War,
+        context_type: ContextType,
+        context_id: str,
+        participants: List[str],  # war_participant_ids
+    ) -> bool:
+        spent: Dict[str, int] = {}
+        for war_part_id in participants:
+            spent[war_part_id] = sum(
+                e.amount
+                for e in war.events
+                if isinstance(e, InfluenceSpent)
+                and e.participant_id == war_part_id
+                and e.context_type == context_type
+            )
+        values = set(spent.values())
+        if values == {0}:  # no bid = confirmed draw
+            war.events.append(
+                TieResolved(
+                    participant_id=None,
+                    context_type=context_type,
+                    context_id=context_id,
+                )
+            )
+            return True
+        if len(values) == 1:  # tie again, continue
+            return False
+        winner = max(spent.items(), key=lambda item: item[1])[0]
+        war.events.append(
+            TieResolved(
+                participant_id=winner,
+                context_type=context_type,
+                context_id=context_id,
+            )
+        )
+        return True
+
+    @staticmethod
+    def can_tie_be_resolved(war: War, participants: List[str]) -> bool:
+        return any(war.get_influence_tokens(pid) > 0 for pid in participants)
+
+    @staticmethod
+    def get_effective_winner_id(
+        war: War,
+        context_type: ContextType,
+        context_id: str,
+        base_winner_id: str | None,
+    ) -> str | None:
+        if base_winner_id is not None:
+            return base_winner_id
+        for ev in reversed(war.events):
+            if (
+                isinstance(ev, TieResolved)
+                and ev.context_type == context_type
+                and ev.context_id == context_id
+            ):
+                return ev.participant_id  # None if confirmed draw
+
         return None

src/warchron/model/war.py

@@ -3,6 +3,7 @@ from uuid import uuid4
 from datetime import datetime
 from typing import Any, Dict, List
 
+from warchron.model.war_event import WarEvent, InfluenceGained, InfluenceSpent
 from warchron.model.exception import ForbiddenOperation
 from warchron.model.war_participant import WarParticipant
 from warchron.model.objective import Objective
@@ -25,6 +26,7 @@ class War:
         self.participants: Dict[str, WarParticipant] = {}
         self.objectives: Dict[str, Objective] = {}
         self.campaigns: List[Campaign] = []
+        self.events: List[WarEvent] = []
         self.is_over: bool = False
 
     def set_id(self, new_id: str) -> None:
@@ -53,6 +55,9 @@ class War:
     def set_influence_token(self, new_state: bool) -> None:
         if self.is_over:
             raise ForbiddenOperation("Can't set influence token of a closed war.")
+        # TODO raise RequiresConfirmation
+        #   * disable: cleanup if any token has already been gained/spent
+        #   * enable: retrigger battle_outcomes and resolve tie again if any draw
         self.influence_token = new_state
 
     def set_state(self, new_state: bool) -> None:
@@ -69,6 +74,7 @@ class War:
             "participants": [part.toDict() for part in self.participants.values()],
             "objectives": [obj.toDict() for obj in self.objectives.values()],
             "campaigns": [camp.toDict() for camp in self.campaigns],
+            "events": [ev.toDict() for ev in self.events],
             "is_over": self.is_over,
         }
 
@@ -87,6 +93,8 @@ class War:
             war.objectives[obj.id] = obj
         for camp_data in data.get("campaigns", []):
             war.campaigns.append(Campaign.fromDict(camp_data))
+        for ev_data in data.get("events", []):
+            war.events.append(WarEvent.fromDict(ev_data))
         war.set_state(data.get("is_over", False))
         return war
 
@@ -439,3 +447,18 @@ class War:
         camp = self.get_campaign_by_round(round_id)
         if camp is not None:
             camp.remove_battle(round_id, sector_id)
+
+    # Event methods
+
+    def get_influence_tokens(self, participant_id: str) -> int:
+        gained = sum(
+            e.amount
+            for e in self.events
+            if isinstance(e, InfluenceGained) and e.participant_id == participant_id
+        )
+        spent = sum(
+            e.amount
+            for e in self.events
+            if isinstance(e, InfluenceSpent) and e.participant_id == participant_id
+        )
+        return gained - spent

src/warchron/model/war_event.py

@@ -1,30 +1,143 @@
+from __future__ import annotations
+from typing import Dict, Any, TypeVar, Type, cast
 from datetime import datetime
 from uuid import uuid4
 
+T = TypeVar("T", bound="WarEvent")
+EVENT_REGISTRY: Dict[str, Type["WarEvent"]] = {}
+
+
+def register_event(cls: Type[T]) -> Type[T]:
+    EVENT_REGISTRY[cls.TYPE] = cls
+    return cls
+
 
 class WarEvent:
-    def __init__(self, participant_id: str):
+    TYPE = "WarEvent"
+
+    def __init__(self, participant_id: str | None = None):
         self.id: str = str(uuid4())
-        self.participant_id: str = participant_id
+        self.participant_id: str | None = participant_id
         self.timestamp: datetime = datetime.now()
 
+    def set_id(self, new_id: str) -> None:
+        self.id = new_id
 
+    def set_participant(self, new_war_part_id: str | None) -> None:
+        self.participant_id = new_war_part_id
+
+    def set_timestamp(self, new_timestamp: datetime) -> None:
+        self.timestamp = new_timestamp
+
+    def toDict(self) -> Dict[str, Any]:
+        return {
+            "type": self.TYPE,
+            "id": self.id,
+            "participant_id": self.participant_id,
+            "timestamp": self.timestamp.isoformat(),
+        }
+
+    @classmethod
+    def _base_fromDict(cls: Type[T], ev: T, data: Dict[str, Any]) -> T:
+        ev.id = data["id"]
+        ev.participant_id = data["participant_id"]
+        ev.timestamp = datetime.fromisoformat(data["timestamp"])
+        return ev
+
+    @staticmethod
+    def fromDict(data: Dict[str, Any]) -> "WarEvent":
+        ev_type = data["type"]
+        cls = cast(Type[WarEvent], EVENT_REGISTRY.get(ev_type))
+        if cls is None:
+            raise ValueError(f"Unknown event type: {ev_type}")
+        return cls.fromDict(data)
+
+
+@register_event
 class TieResolved(WarEvent):
-    def __init__(self, participant_id: str, context_type: str, context_id: str):
+    TYPE = "TieResolved"
+
+    def __init__(self, participant_id: str | None, context_type: str, context_id: str):
         super().__init__(participant_id)
+        self.participant_id: str | None = (
+            participant_id  # winner or None (confirmed tie)
+        )
         self.context_type = context_type  # battle, round, campaign, war
         self.context_id = context_id
 
+    def toDict(self) -> Dict[str, Any]:
+        d = super().toDict()
+        d.update(
+            {
+                "context_type": self.context_type,
+                "context_id": self.context_id,
+            }
+        )
+        return d
 
+    @classmethod
+    def fromDict(cls, data: Dict[str, Any]) -> TieResolved:
+        ev = cls(
+            data["participant_id"],
+            data["context_type"],
+            data["context_id"],
+        )
+        return cls._base_fromDict(ev, data)
+
+
+@register_event
 class InfluenceGained(WarEvent):
+    TYPE = "InfluenceGained"
+
     def __init__(self, participant_id: str, amount: int, source: str):
         super().__init__(participant_id)
         self.amount = amount
         self.source = source  # "battle", "tie_resolution", etc.
 
+    def toDict(self) -> Dict[str, Any]:
+        d = super().toDict()
+        d.update(
+            {
+                "amount": self.amount,
+                "source": self.source,
+            }
+        )
+        return d
 
+    @classmethod
+    def fromDict(cls, data: Dict[str, Any]) -> InfluenceGained:
+        ev = cls(
+            data["participant_id"],
+            data["amount"],
+            data["source"],
+        )
+        return cls._base_fromDict(ev, data)
+
+
+@register_event
 class InfluenceSpent(WarEvent):
-    def __init__(self, participant_id: str, amount: int, context: str):
+    TYPE = "InfluenceSpent"
+
+    def __init__(self, participant_id: str, amount: int, context_type: str):
         super().__init__(participant_id)
         self.amount = amount
-        self.context = context  # "battle_tie", "campaign_tie", etc.
+        self.context_type = context_type  # "battle_tie", "campaign_tie", etc.
+
+    def toDict(self) -> Dict[str, Any]:
+        d = super().toDict()
+        d.update(
+            {
+                "amount": self.amount,
+                "context_type": self.context_type,
+            }
+        )
+        return d
+
+    @classmethod
+    def fromDict(cls, data: Dict[str, Any]) -> InfluenceSpent:
+        ev = cls(
+            data["participant_id"],
+            data["amount"],
+            data["context_type"],
+        )
+        return cls._base_fromDict(ev, data)

src/warchron/model/war_participant.py

@@ -1,12 +1,6 @@
 from __future__ import annotations
 from typing import Any, Dict
 from uuid import uuid4
-from typing import TYPE_CHECKING
-
-if TYPE_CHECKING:
-    from warchron.model.war import War
-from warchron.model.war_event import WarEvent, InfluenceSpent, InfluenceGained
-from warchron.model.score_service import ScoreService
 
 
 class WarParticipant:
@@ -14,13 +8,12 @@ class WarParticipant:
         self.id: str = str(uuid4())
         self.player_id: str = player_id  # ref to Model.players
         self.faction: str = faction
-        self.events: list[WarEvent] = []
 
     def set_id(self, new_id: str) -> None:
         self.id = new_id
 
-    def set_player(self, new_player: str) -> None:
-        self.player_id = new_player
+    def set_player(self, new_player_id: str) -> None:
+        self.player_id = new_player_id
 
     def set_faction(self, new_faction: str) -> None:
         self.faction = new_faction
@@ -40,16 +33,3 @@ class WarParticipant:
         )
         part.set_id(data["id"])
         return part
-
-    # Computed properties
-
-    def influence_tokens(self) -> int:
-        gained = sum(e.amount for e in self.events if isinstance(e, InfluenceGained))
-        spent = sum(e.amount for e in self.events if isinstance(e, InfluenceSpent))
-        return gained - spent
-
-    def victory_points(self, war: "War") -> int:
-        return ScoreService.compute_victory_points_for_participant(war, self.id)
-
-    def narrative_points(self, war: "War") -> Dict[str, int]:
-        return ScoreService.compute_narrative_points_for_participant(war, self.id)