fix round tie loop + improve tie ranking

src/warchron/model/campaign.py

@@ -187,6 +187,7 @@ class Campaign:
         mission: str | None,
         description: str | None,
     ) -> None:
+        # TODO raise error if sector used in a closed round (potential tokens)
         if self.is_over:
             raise ForbiddenOperation("Can't update sector in a closed campaign.")
         sect = self.get_sector(sector_id)

src/warchron/model/closure_service.py

@@ -27,7 +27,8 @@ class ClosureService:
     @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}"
+            isinstance(e, InfluenceGained)
+            and e.context_id == f"battle:{battle.sector_id}"
             for e in war.events
         )
         if already_granted:
@@ -49,7 +50,8 @@ class ClosureService:
                 InfluenceGained(
                     participant_id=effective_winner,
                     amount=1,
-                    source=f"battle:{battle.sector_id}",
+                    context_type=ContextType.BATTLE,
+                    context_id=f"battle:{battle.sector_id}",
                 )
             )
 

src/warchron/model/tie_manager.py

@@ -16,6 +16,14 @@ class TieContext:
     participants: List[str]  # war_participant_ids
 
 
+@dataclass
+class TieState:
+    winner: str | None
+    tied_players: List[str]
+    eliminated: List[str]
+    # is_final: bool
+
+
 class TieResolver:
 
     @staticmethod
@@ -26,20 +34,20 @@ class TieResolver:
         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 resolved:
+            if TieResolver.is_tie_resolved(war, ContextType.BATTLE, battle.sector_id):
                 continue
+
             if campaign is None:
                 raise RuntimeError("No campaign for this battle tie")
             if battle.player_1_id is None or battle.player_2_id is None:
                 raise RuntimeError("Missing player(s) in this battle context.")
             p1 = campaign.participants[battle.player_1_id].war_participant_id
             p2 = campaign.participants[battle.player_2_id].war_participant_id
+            if not TieResolver.can_tie_be_resolved(war, [p1, p2]):
+                war.events.append(
+                    TieResolved(None, ContextType.BATTLE, battle.sector_id)
+                )
+                continue
             ties.append(
                 TieContext(
                     context_type=ContextType.BATTLE,
@@ -51,13 +59,7 @@ class TieResolver:
 
     @staticmethod
     def find_campaign_ties(war: War, campaign_id: str) -> List[TieContext]:
-        resolved = any(
-            isinstance(e, TieResolved)
-            and e.context_type == ContextType.CAMPAIGN
-            and e.context_id == campaign_id
-            for e in war.events
-        )
-        if resolved:
+        if TieResolver.is_tie_resolved(war, ContextType.CAMPAIGN, campaign_id):
             return []
         scores = ScoreService.compute_scores(war, ContextType.CAMPAIGN, campaign_id)
         buckets: DefaultDict[int, List[str]] = defaultdict(list)
@@ -96,46 +98,54 @@ class TieResolver:
                     participant_id=war_part_id,
                     amount=1,
                     context_type=context_type,
+                    context_id=context_id,
                 )
             )
 
     @staticmethod
-    def try_tie_break(
+    def rank_by_tokens(
         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
+        participants: List[str],
+    ) -> List[List[str]]:
+        spent = {pid: 0 for pid in participants}
+        for ev in war.events:
+            if (
+                isinstance(ev, InfluenceSpent)
+                and ev.context_type == context_type
+                and ev.context_id == context_id
+                and ev.participant_id in spent
+            ):
+                spent[ev.participant_id] += ev.amount
+        sorted_items = sorted(spent.items(), key=lambda x: x[1], reverse=True)
+        groups: List[List[str]] = []
+        current_score = None
+        for pid, score in sorted_items:
+            if score != current_score:
+                groups.append([])
+                current_score = score
+            groups[-1].append(pid)
+        return groups
+
+    @staticmethod
+    def resolve_tie_state(
+        war: War,
+        context_type: ContextType,
+        context_id: str,
+        participants: List[str],
+        bids: dict[str, bool] | None = None,
+    ) -> None:
+        # confirmed draw if bids are 0
+        if bids is not None and not any(bids.values()):
+            war.events.append(TieResolved(None, context_type, context_id))
+            return
+        # else rank_by_tokens
+        groups = TieResolver.rank_by_tokens(war, context_type, context_id, participants)
+        if len(groups[0]) == 1:
+            war.events.append(TieResolved(groups[0][0], context_type, context_id))
+            return
+        # if tie persists, do nothing, workflow will call again
 
     @staticmethod
     def can_tie_be_resolved(war: War, participants: List[str]) -> bool:
@@ -155,3 +165,12 @@ class TieResolver:
             ):
                 return ev.participant_id is not None
         return False
+
+    @staticmethod
+    def is_tie_resolved(war: War, context_type: ContextType, context_id: str) -> bool:
+        return any(
+            isinstance(ev, TieResolved)
+            and ev.context_type == context_type
+            and ev.context_id == context_id
+            for ev in war.events
+        )

src/warchron/model/war_event.py

@@ -15,9 +15,11 @@ def register_event(cls: Type[T]) -> Type[T]:
 class WarEvent:
     TYPE = "WarEvent"
 
-    def __init__(self, participant_id: str | None = None):
+    def __init__(self, participant_id: str | None, context_type: str, context_id: str):
         self.id: str = str(uuid4())
         self.participant_id: str | None = participant_id
+        self.context_type = context_type  # battle, round, campaign, war
+        self.context_id = context_id
         self.timestamp: datetime = datetime.now()
 
     def set_id(self, new_id: str) -> None:
@@ -34,6 +36,8 @@ class WarEvent:
             "type": self.TYPE,
             "id": self.id,
             "participant_id": self.participant_id,
+            "context_type": self.context_type,
+            "context_id": self.context_id,
             "timestamp": self.timestamp.isoformat(),
         }
 
@@ -41,6 +45,8 @@ class WarEvent:
     def _base_fromDict(cls: Type[T], ev: T, data: Dict[str, Any]) -> T:
         ev.id = data["id"]
         ev.participant_id = data["participant_id"]
+        ev.context_type = data["context_type"]
+        ev.context_id = data["context_id"]
         ev.timestamp = datetime.fromisoformat(data["timestamp"])
         return ev
 
@@ -58,21 +64,10 @@ class TieResolved(WarEvent):
     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
+        super().__init__(participant_id, context_type, 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
@@ -89,17 +84,17 @@ class TieResolved(WarEvent):
 class InfluenceGained(WarEvent):
     TYPE = "InfluenceGained"
 
-    def __init__(self, participant_id: str, amount: int, source: str):
-        super().__init__(participant_id)
+    def __init__(
+        self, participant_id: str, amount: int, context_type: str, context_id: str
+    ):
+        super().__init__(participant_id, context_type, context_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
@@ -108,8 +103,9 @@ class InfluenceGained(WarEvent):
     def fromDict(cls, data: Dict[str, Any]) -> InfluenceGained:
         ev = cls(
             data["participant_id"],
-            data["amount"],
-            data["source"],
+            int(data["amount"]),
+            data["context_type"],
+            data["context_id"],
         )
         return cls._base_fromDict(ev, data)
 
@@ -118,17 +114,17 @@ class InfluenceGained(WarEvent):
 class InfluenceSpent(WarEvent):
     TYPE = "InfluenceSpent"
 
-    def __init__(self, participant_id: str, amount: int, context_type: str):
-        super().__init__(participant_id)
+    def __init__(
+        self, participant_id: str, amount: int, context_type: str, context_id: str
+    ):
+        super().__init__(participant_id, context_type, context_id)
         self.amount = amount
-        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
@@ -137,7 +133,8 @@ class InfluenceSpent(WarEvent):
     def fromDict(cls, data: Dict[str, Any]) -> InfluenceSpent:
         ev = cls(
             data["participant_id"],
-            data["amount"],
+            int(data["amount"]),
             data["context_type"],
+            data["context_id"],
         )
         return cls._base_fromDict(ev, data)