Add optional asso pigliatutto rule with a lobby toggle

An ace played when no equal rank is on the table sweeps the whole
table; the sweep never scores a scopa, and an ace already on the table
still forces the mandatory equal-rank capture. The rule is off by
default, selectable per match via the schema-driven creation form, and
announced in the game view.

server/packages/tavolo-scopone/README.md

@@ -11,15 +11,17 @@ implementation.
   (de)serialization. No session envelope, no transport, no I/O.
 - `engine.py` — the pure rules engine: deck, legal captures, plays,
   auto-play, hand scoring (carte, denara, settebello, primiera, scope,
-  napola), hand-end acknowledgements. Deterministic and I/O-free apart
-  from logging, so the whole rule set is unit-testable.
+  napola), the optional asso pigliatutto sweep, hand-end
+  acknowledgements. Deterministic and I/O-free apart from logging, so the
+  whole rule set is unit-testable.
 - `errors.py` — scopone-specific errors (`CardNotInHand`); every other
   failure mode is a shared `tavolo.platform.errors` subclass.
 - `plugin.py` — `ScoponeEngine(GameEngine)`: the platform-facing adapter.
   It translates create/join/websocket actions/deadlines into rules-engine
-  calls and back, validates the `target_score`/`napola` creation options,
-  and extracts the `MatchResult` (teams, winner, per-player scores and the
-  match summary persisted as the match's JSON `result`).
+  calls and back, validates the `target_score`/`napola`/`asso_pigliatutto`
+  creation options, and extracts the `MatchResult` (teams, winner,
+  per-player scores and the match summary persisted as the match's JSON
+  `result`).
 
 Timeouts are constructor arguments (`turn_timeout_seconds`,
 `hand_ack_timeout_seconds`), wired from the environment by the

server/packages/tavolo-scopone/src/tavolo/scopone/engine.py

@@ -28,6 +28,10 @@ Rules implemented
   consecutive denari starting from the ace scores one point per card when
   it reaches at least three cards (A-2-3 = 3, A-2-3-4 = 4, ...). A team
   capturing the whole denari suit (ace to king) wins the match instantly.
+* Optional ``asso_pigliatutto`` rule (disabled by default): an ace played
+  when no equal rank is on the table sweeps the whole table. The sweep
+  never scores a scopa, even when cards remain in hand. When another ace
+  is on the table the mandatory equal-rank capture still applies.
 * The match ends when a team reaches the target score with a clear lead; a
   tie at or above the target is broken by playing another hand.
 """
@@ -110,17 +114,22 @@ def shuffled_deck(rng: Optional[random.Random] = None) -> List[Card]:
     return deck
 
 
-def legal_captures(table: Sequence[Card], card: Card) -> List[List[Card]]:
+def legal_captures(
+    table: Sequence[Card], card: Card, *, asso_pigliatutto: bool = False
+) -> List[List[Card]]:
     """Return every legal capture (a list of card sets) for ``card``.
 
     If an equal-ranked card is on the table, only those single-card
     captures are returned (the rule forbids taking a combination instead).
-    Otherwise every subset of the table whose ranks sum to ``card.rank`` is
-    returned.
+    Otherwise, when ``asso_pigliatutto`` is enabled and ``card`` is an ace,
+    the only capture is the whole table; without the option every subset of
+    the table whose ranks sum to ``card.rank`` is returned.
     """
     equal = [c for c in table if c.rank == card.rank]
     if equal:
         return [[c] for c in equal]
+    if asso_pigliatutto and card.rank == 1:
+        return [list(table)] if table else []
 
     candidates = [c for c in table if c.rank <= card.rank]
     captures: List[List[Card]] = []
@@ -140,6 +149,7 @@ def create_game(
     hand_ack_timeout: int = DEFAULT_HAND_ACK_TIMEOUT_SECONDS,
     turn_timeout: int = DEFAULT_TURN_TIMEOUT_SECONDS,
     napola: bool = True,
+    asso_pigliatutto: bool = False,
 ) -> ScoponeState:
     """Create a lobby game with the creator seated first."""
     if target_score < 1 or target_score > 100:
@@ -147,6 +157,7 @@ def create_game(
     return ScoponeState(
         target_score=target_score,
         napola=napola,
+        asso_pigliatutto=asso_pigliatutto,
         phase=PHASE_LOBBY,
         players=[PlayerState(sub=creator_sub, name=creator_name, seat=0)],
         hand_ack_timeout=hand_ack_timeout,
@@ -240,7 +251,18 @@ def play(
     player.hand.remove(card)
 
     requested = [parse_card(c) for c in (capture_codes or [])]
-    options = legal_captures(state.table, card)
+    table_before = list(state.table)
+    options = legal_captures(
+        table_before, card, asso_pigliatutto=state.asso_pigliatutto
+    )
+    # An asso pigliatutto sweep: the ace takes the whole table because no
+    # equal rank was on it. Such a sweep never scores a scopa.
+    pigliatutto = (
+        state.asso_pigliatutto
+        and card.rank == 1
+        and bool(table_before)
+        and all(c.rank != card.rank for c in table_before)
+    )
 
     taken: List[Card] = []
     scopa = False
@@ -260,7 +282,7 @@ def play(
         state.last_taker = player.seat
         # A scopa scores only if cards remain to be played this hand.
         hands_empty = all(not p.hand for p in state.players)
-        if not state.table and not hands_empty:
+        if not state.table and not hands_empty and not pigliatutto:
             player.scope += 1
             scopa = True
 
@@ -270,6 +292,7 @@ def play(
         card=card.code,
         captured=[c.code for c in taken],
         scopa=scopa,
+        pigliatutto=pigliatutto,
     )
 
     if all(not p.hand for p in state.players):
@@ -308,7 +331,9 @@ def auto_play(state: ScoponeState, rng: Optional[random.Random] = None) -> None:
         raise IllegalMove("the player on turn has no cards")
     chooser = rng or _rng
     card = chooser.choice(player.hand)
-    options = legal_captures(state.table, card)
+    options = legal_captures(
+        state.table, card, asso_pigliatutto=state.asso_pigliatutto
+    )
     capture = chooser.choice(options) if options else None
     play(
         state,
@@ -526,6 +551,7 @@ def state_for_player(state: ScoponeState, sub: str) -> Dict[str, object]:
         "phase": state.phase,
         "target_score": state.target_score,
         "napola": state.napola,
+        "asso_pigliatutto": state.asso_pigliatutto,
         "hand_number": state.hand_number,
         "dealer": state.dealer,
         "turn": state.turn,
@@ -545,7 +571,9 @@ def state_for_player(state: ScoponeState, sub: str) -> Dict[str, object]:
         # rule logic stays server-side.
         legal_moves: Dict[str, List[List[str]]] = {}
         for hand_card in viewer.hand:
-            options = legal_captures(state.table, hand_card)
+            options = legal_captures(
+                state.table, hand_card, asso_pigliatutto=state.asso_pigliatutto
+            )
             if options:
                 legal_moves[hand_card.code] = [
                     [c.code for c in option] for option in options

server/packages/tavolo-scopone/src/tavolo/scopone/plugin.py

@@ -75,6 +75,12 @@ class ScoponeEngine(GameEngine):
                 "description": "Score the napola rule; a full denari "
                                "sweep wins the match",
             },
+            "asso_pigliatutto": {
+                "type": "boolean",
+                "default": False,
+                "description": "Score the asso pigliatutto rule; an ace "
+                               "sweeps the whole table (no scopa)",
+            },
         },
     }
 
@@ -95,6 +101,9 @@ class ScoponeEngine(GameEngine):
         napola = options.get("napola", True)
         if not isinstance(napola, bool):
             raise IllegalMove("napola must be a boolean")
+        asso_pigliatutto = options.get("asso_pigliatutto", False)
+        if not isinstance(asso_pigliatutto, bool):
+            raise IllegalMove("asso_pigliatutto must be a boolean")
         creator = session.players[0]
         session.state = engine.create_game(
             creator_sub=creator.user_sub,
@@ -103,6 +112,7 @@ class ScoponeEngine(GameEngine):
             hand_ack_timeout=self._hand_ack_timeout,
             turn_timeout=self._turn_timeout,
             napola=napola,
+            asso_pigliatutto=asso_pigliatutto,
         )
         # The creator takes seat 0, i.e. team A.
         session.players[0] = Seat(
@@ -133,6 +143,7 @@ class ScoponeEngine(GameEngine):
             "phase": state.phase,
             "target_score": state.target_score,
             "napola": state.napola,
+            "asso_pigliatutto": state.asso_pigliatutto,
         }
 
     # -- play -------------------------------------------------------------

server/packages/tavolo-scopone/src/tavolo/scopone/state.py

@@ -92,6 +92,8 @@ class Move:
     card: str
     captured: List[str] = field(default_factory=list)
     scopa: bool = False
+    # The play was an asso pigliatutto sweep (it took the whole table).
+    pigliatutto: bool = False
 
     def to_json(self) -> Dict[str, Any]:
         return {
@@ -100,6 +102,7 @@ class Move:
             "card": self.card,
             "captured": list(self.captured),
             "scopa": self.scopa,
+            "pigliatutto": self.pigliatutto,
         }
 
     @staticmethod
@@ -110,6 +113,7 @@ class Move:
             card=str(data["card"]),
             captured=[str(c) for c in data.get("captured", [])],
             scopa=bool(data.get("scopa", False)),
+            pigliatutto=bool(data.get("pigliatutto", False)),
         )
 
 
@@ -154,6 +158,10 @@ class ScoponeState:
     # Whether the napola rule is scored (denari run from the ace; a full
     # suit wins the match instantly). Default on.
     napola: bool = True
+    # Whether the asso pigliatutto rule is scored (an ace sweeps the whole
+    # table when no equal rank is on it; the sweep never scores a scopa).
+    # Default off, also for states serialized before the option existed.
+    asso_pigliatutto: bool = False
     phase: str = PHASE_LOBBY
     players: List[PlayerState] = field(default_factory=list)
     table: List[Card] = field(default_factory=list)
@@ -184,6 +192,7 @@ class ScoponeState:
         return {
             "target_score": self.target_score,
             "napola": self.napola,
+            "asso_pigliatutto": self.asso_pigliatutto,
             "phase": self.phase,
             "players": [p.to_json() for p in self.players],
             "table": [c.to_json() for c in self.table],
@@ -207,6 +216,7 @@ class ScoponeState:
         return ScoponeState(
             target_score=int(data.get("target_score", DEFAULT_TARGET_SCORE)),
             napola=bool(data.get("napola", True)),
+            asso_pigliatutto=bool(data.get("asso_pigliatutto", False)),
             phase=str(data.get("phase", PHASE_LOBBY)),
             players=[PlayerState.from_json(p) for p in data.get("players", [])],
             table=[Card.from_json(c) for c in data.get("table", [])],

server/packages/tavolo-scopone/tests/test_engine.py

@@ -325,6 +325,138 @@ class NapolaTest(unittest.TestCase):
         )
 
 
+class AssoPigliatuttoTest(unittest.TestCase):
+    def test_ace_sweeps_the_whole_table_without_scopa(self) -> None:
+        # The other players still hold cards, so emptying the table would
+        # normally score a scopa: the sweep must suppress it.
+        state = make_state(
+            [["01D", "09C"], ["04D"], ["05D"], ["06D"]],
+            table=["05C", "02D", "03S"],
+        )
+        state.asso_pigliatutto = True
+        engine.play(state, "p0", "01D", ["05C", "02D", "03S"])
+        self.assertEqual([], state.table)
+        self.assertEqual(
+            {"05C", "02D", "03S", "01D"},
+            {c.code for c in state.players[0].captured},
+        )
+        self.assertEqual(0, state.players[0].scope)
+        assert state.last_move is not None
+        self.assertTrue(state.last_move.pigliatutto)
+        self.assertFalse(state.last_move.scopa)
+
+    def test_legal_captures_offer_only_the_whole_table(self) -> None:
+        table = [card("05C"), card("02D"), card("03S")]
+        options = engine.legal_captures(
+            table, card("01D"), asso_pigliatutto=True
+        )
+        self.assertEqual(
+            [["05C", "02D", "03S"]], [[c.code for c in o] for o in options]
+        )
+        # Non-ace cards keep the normal rules.
+        self.assertEqual(
+            [],
+            engine.legal_captures(table, card("04B"), asso_pigliatutto=True),
+        )
+
+    def test_equal_rank_ace_blocks_the_sweep(self) -> None:
+        table = ["01C", "02D"]
+        state = make_state([["01D"], ["04D"], ["05D"], ["06D"]], table=table)
+        state.asso_pigliatutto = True
+        options = engine.legal_captures(
+            state.table, card("01D"), asso_pigliatutto=True
+        )
+        self.assertEqual([["01C"]], [[c.code for c in o] for o in options])
+        with self.assertRaises(IllegalMove):
+            engine.play(state, "p0", "01D", ["01C", "02D"])
+        # A rejected play consumes the card, so retry on a fresh state.
+        state = make_state([["01D"], ["04D"], ["05D"], ["06D"]], table=table)
+        state.asso_pigliatutto = True
+        engine.play(state, "p0", "01D", ["01C"])
+        self.assertEqual(["02D"], [c.code for c in state.table])
+        assert state.last_move is not None
+        self.assertFalse(state.last_move.pigliatutto)
+
+    def test_disabled_by_default(self) -> None:
+        state = make_state(
+            [["01D", "09C"], ["04D"], ["05D"], ["06D"]],
+            table=["05C", "02D"],
+        )
+        engine.play(state, "p0", "01D")
+        # Without the rule an ace captures nothing here and stays on the
+        # table.
+        self.assertEqual({"05C", "02D", "01D"}, {c.code for c in state.table})
+        assert state.last_move is not None
+        self.assertFalse(state.last_move.pigliatutto)
+        self.assertEqual(
+            [],
+            engine.legal_captures(
+                [card("05C")], card("01D"), asso_pigliatutto=False
+            ),
+        )
+
+    def test_ace_on_empty_table_is_a_normal_play(self) -> None:
+        state = make_state(
+            [["01D", "09C"], ["04D"], ["05D"], ["06D"]], table=[]
+        )
+        state.asso_pigliatutto = True
+        engine.play(state, "p0", "01D")
+        self.assertEqual(["01D"], [c.code for c in state.table])
+        assert state.last_move is not None
+        self.assertFalse(state.last_move.pigliatutto)
+
+    def test_move_pigliatutto_serialization(self) -> None:
+        move = engine.Move(
+            seat=0, name="p0", card="01D", captured=["05C"], pigliatutto=True
+        )
+        restored = engine.Move.from_json(move.to_json())
+        self.assertTrue(restored.pigliatutto)
+        legacy = move.to_json()
+        del legacy["pigliatutto"]
+        self.assertFalse(engine.Move.from_json(legacy).pigliatutto)
+
+    def test_state_serialization_roundtrip(self) -> None:
+        state = make_state([["02D"], [], [], []], table=[])
+        self.assertFalse(state.asso_pigliatutto)
+        state.asso_pigliatutto = True
+        self.assertTrue(
+            ScoponeState.from_json(state.to_json()).asso_pigliatutto
+        )
+        # States serialized before the option existed default to disabled.
+        data = state.to_json()
+        del data["asso_pigliatutto"]
+        self.assertFalse(ScoponeState.from_json(data).asso_pigliatutto)
+
+    def test_create_game_default_and_override(self) -> None:
+        self.assertFalse(engine.create_game("p0", "p0").asso_pigliatutto)
+        self.assertTrue(
+            engine.create_game("p0", "p0", asso_pigliatutto=True).asso_pigliatutto
+        )
+
+    def test_state_for_player_exposes_rule_and_legal_moves(self) -> None:
+        state = make_state(
+            [["01D"], ["04D"], ["05D"], ["06D"]],
+            table=["05C", "02D", "03S"],
+        )
+        state.asso_pigliatutto = True
+        view = engine.state_for_player(state, "p0")
+        self.assertTrue(view["asso_pigliatutto"])
+        self.assertEqual(
+            {"01D": [["05C", "02D", "03S"]]}, view["legal_moves"]
+        )
+
+    def test_auto_play_sweeps_with_the_rule(self) -> None:
+        state = make_state(
+            [["01D"], ["04D"], ["05D"], ["06D"]],
+            table=["05C", "02D"],
+        )
+        state.asso_pigliatutto = True
+        engine.auto_play(state)
+        assert state.last_move is not None
+        self.assertTrue(state.last_move.pigliatutto)
+        self.assertEqual(["05C", "02D"], state.last_move.captured)
+
+
 class MatchFlowTest(unittest.TestCase):
     def test_join_starts_when_full(self) -> None:
         state = engine.create_game("p0", "p0", target_score=11)

server/packages/tavolo-scopone/tests/test_plugin.py

@@ -43,6 +43,12 @@ class CreateTest(unittest.TestCase):
         session = _session(engine, target_score=16, napola=False)
         self.assertEqual(16, session.state.target_score)
         self.assertFalse(session.state.napola)
+        self.assertFalse(session.state.asso_pigliatutto)
+
+    def test_create_asso_pigliatutto_option(self) -> None:
+        engine = ScoponeEngine()
+        session = _session(engine, asso_pigliatutto=True)
+        self.assertTrue(session.state.asso_pigliatutto)
 
     def test_create_rejects_bad_options(self) -> None:
         engine = ScoponeEngine()
@@ -52,6 +58,8 @@ class CreateTest(unittest.TestCase):
             _session(engine, target_score="eleven")
         with self.assertRaises(IllegalMove):
             _session(engine, napola="yes")
+        with self.assertRaises(IllegalMove):
+            _session(engine, asso_pigliatutto="yes")
 
     def test_timeouts_come_from_the_engine(self) -> None:
         engine = ScoponeEngine(turn_timeout_seconds=7, hand_ack_timeout_seconds=9)
@@ -159,6 +167,7 @@ class ViewTest(unittest.TestCase):
         self.assertEqual("lobby", lobby["phase"])
         self.assertEqual(16, lobby["target_score"])
         self.assertTrue(lobby["napola"])
+        self.assertFalse(lobby["asso_pigliatutto"])
 
 
 class SerializationTest(unittest.TestCase):
@@ -283,6 +292,7 @@ class ResultTest(unittest.TestCase):
         self.assertEqual(4, engine.max_players)
         self.assertIn("target_score", engine.options_schema["properties"])
         self.assertIn("napola", engine.options_schema["properties"])
+        self.assertIn("asso_pigliatutto", engine.options_schema["properties"])
 
 
 if __name__ == "__main__":