Add briscola chiamata

Five-player briscola with a descending calling auction and a secret
partner. Starting with the player after the dealer, each player bids a
lower rank (A > 3 > K > H > J > 7 > 6 > 5 > 4 > 2) or passes; once the
standing bid is the two the auction continues on points (62, 64, ...).
The last player standing must bid, and the winner names any card of the
won rank: its holder is the secret partner, or the caller plays alone
when it is one of their own cards, and its suit becomes the trump. The
calling side must reach 61 (or the raised target); a 60-60 deal with the
plain target is a patta and is replayed. A match is a single deal, so
the caller and partner (or the lone caller) against the other players
map onto the platform's two-team result, Elo and match history.

The rules live in tavolo.briscola.chiamata with the
BriscolaChiamataEngine adapter registered alongside the other briscola
variants; the partner's role and the side totals stay hidden until the
called card is played. The web client gets a five-seat table, the
auction panel with legal bids and point raises, the call picker, the
reveal bands and a chiamata-aware final summary.

server/tests/test_briscola_chiamata.py

@@ -0,0 +1,343 @@
+"""Briscola chiamata integration tests: lobby, auction, websocket play, timeouts.
+
+The per-package suite covers the rules and the engine contract; these
+tests wire the game through the full platform stack (REST, WebSocket,
+deadline scheduler).
+"""
+from __future__ import annotations
+
+import asyncio
+import unittest
+from datetime import datetime, timedelta, timezone
+
+from httpx import ASGITransport, AsyncClient
+from httpx_ws import aconnect_ws
+from httpx_ws.transport import ASGIWebSocketTransport
+
+from tavolo.app import app, game_store, platform
+from tavolo.briscola.chiamata import (
+    PHASE_AUCTION,
+    PHASE_CALLING,
+    PHASE_PLAYING,
+    Auction,
+    Bid,
+)
+from tavolo.platform import GameSession, Seat
+from tests.helpers import async_test, make_user, oidc_user, ws_users
+
+PLAYERS = ("alice", "bob", "carol", "dave", "erin")
+
+
+async def _create_and_join(client: AsyncClient) -> dict:
+    """Create a chiamata game and seat all five players; return the view."""
+    with oidc_user(PLAYERS[0]):
+        created = await client.post("/api/games", json={"game_type": "briscola_chiamata"})
+    assert created.status_code == 201, created.text
+    code = created.json()["join_code"]
+    body = created.json()
+    for player in PLAYERS[1:]:
+        with oidc_user(player):
+            joined = await client.post("/api/games/join", json={"code": code})
+        assert joined.status_code == 200, joined.text
+        body = joined.json()
+    return body
+
+
+class ChiamataRoutesTest(unittest.TestCase):
+    @async_test
+    async def test_create_and_seat_five_players(self) -> None:
+        transport = ASGITransport(app=app)
+        async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
+            with oidc_user(PLAYERS[0]):
+                created = await client.post(
+                    "/api/games", json={"game_type": "briscola_chiamata"}
+                )
+            self.assertEqual(201, created.status_code)
+            body = created.json()
+            self.assertEqual("briscola_chiamata", body["game_type"])
+            self.assertEqual("lobby", body["phase"])
+            self.assertEqual(4, body["seats_open"])
+            self.assertIsNone(body["players"][0]["team"])
+            code = body["join_code"]
+
+            for player in PLAYERS[1:4]:
+                with oidc_user(player):
+                    joined = await client.post("/api/games/join", json={"code": code})
+                self.assertEqual("lobby", joined.json()["phase"])
+                self.assertEqual(
+                    4 - (PLAYERS.index(player)), joined.json()["seats_open"]
+                )
+
+            with oidc_user(PLAYERS[4]):
+                started = await client.post("/api/games/join", json={"code": code})
+            state = started.json()
+            self.assertEqual(PHASE_AUCTION, state["phase"])
+            self.assertEqual(0, state["seats_open"])
+            self.assertEqual(5, len(state["players"]))
+            for seat in state["players"]:
+                # Roles are hidden until the called card is played.
+                self.assertEqual("", seat["team"])
+                self.assertEqual(8, seat["cards_left"])
+            # The auction opens with the player after the dealer (seat 0).
+            self.assertEqual(1, state["auction"]["turn"])
+            self.assertEqual([0, 1, 2, 3], state["auction"]["active"][0:4])
+            self.assertIsNone(state["auction"]["current"])
+            self.assertIsNotNone(state["auction_deadline"])
+            # The last joiner's personalized view carries their own hand.
+            self.assertIn("hand", state["players"][4])
+            self.assertEqual(8, len(state["players"][4]["hand"]))
+            self.assertNotIn("hand", state["players"][0])
+            # There is no stock in this variant.
+            self.assertNotIn("stock_count", state)
+
+            # A non-player watches without hands or the join code.
+            with oidc_user("mallory"):
+                watched = await client.get(f"/api/games/{state['id']}")
+            self.assertEqual(200, watched.status_code)
+            spectator = watched.json()
+            self.assertTrue(spectator["spectator"])
+            self.assertNotIn("join_code", spectator)
+            for seat in spectator["players"]:
+                self.assertNotIn("hand", seat)
+
+    @async_test
+    async def test_preview_lists_chiamata(self) -> None:
+        transport = ASGITransport(app=app)
+        async with AsyncClient(transport=transport, base_url="http://127.0.0.1") as client:
+            with oidc_user("alice"):
+                created = await client.post(
+                    "/api/games", json={"game_type": "briscola_chiamata"}
+                )
+            code = created.json()["join_code"]
+            with oidc_user("bob"):
+                preview = await client.get(f"/api/games/preview?code={code}")
+            self.assertEqual(200, preview.status_code)
+            body = preview.json()
+            self.assertEqual("briscola_chiamata", body["game_type"])
+            self.assertEqual("Briscola chiamata", body["game_name"])
+            self.assertTrue(body["game_description"])
+            self.assertEqual("lobby", body["phase"])
+            self.assertEqual(4, body["seats_open"])
+
+
+class ChiamataWebSocketTest(unittest.TestCase):
+    @async_test
+    async def test_auction_call_and_play_over_the_socket(self) -> None:
+        api_transport = ASGITransport(app=app)
+        async with AsyncClient(transport=api_transport, base_url="http://127.0.0.1") as client:
+            state = await _create_and_join(client)
+            game_id = state["id"]
+
+            ws_transport = ASGIWebSocketTransport(app=app)
+            async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client:
+                with ws_users([make_user("bob"), make_user("alice")]):
+                    async with aconnect_ws(f"/ws/games/{game_id}", ws_client) as bob_ws:
+                        first = await bob_ws.receive_json()
+                        self.assertEqual("state", first["type"])
+                        self.assertEqual(PHASE_AUCTION, first["game"]["phase"])
+                        self.assertTrue(first["game"].get("your_turn"))
+                        self.assertIn("legal_bids", first["game"])
+
+                        async with aconnect_ws(f"/ws/games/{game_id}", ws_client) as alice_ws:
+                            await alice_ws.receive_json()
+                            # Bob (seat 1) opens the bidding with the two.
+                            await bob_ws.send_json({"action": "bid", "rank": 2})
+                            update = await bob_ws.receive_json()
+                            self.assertEqual("state", update["type"])
+                            current = update["game"]["auction"]["current"]
+                            self.assertEqual(
+                                {"seat": 1, "rank": 2, "points": 61}, current
+                            )
+                            self.assertEqual(2, update["game"]["auction"]["turn"])
+                            # Alice receives the broadcast of Bob's bid.
+                            broadcast = await alice_ws.receive_json()
+                            self.assertEqual("state", broadcast["type"])
+                            # Alice (seat 0) is not on turn: her bid is an error.
+                            await alice_ws.send_json({"action": "bid", "rank": 3})
+                            error = await alice_ws.receive_json()
+                            self.assertEqual("error", error["type"])
+
+            # The remaining players pass; Bob wins the auction and is
+            # asked to name the card.
+            engine = platform.registry.require("briscola_chiamata")
+            session = await game_store.load(game_id)
+            assert session is not None
+            for sub in ("carol", "dave", "erin", "alice"):
+                engine.handle_action(session, sub, "pass", {})
+            await game_store.save(session)
+            self.assertEqual(PHASE_CALLING, session.state.phase)
+            self.assertEqual(1, session.state.caller)
+
+            async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client:
+                with ws_users([make_user("bob")]):
+                    async with aconnect_ws(f"/ws/games/{game_id}", ws_client) as ws:
+                        first = await ws.receive_json()
+                        self.assertEqual(PHASE_CALLING, first["game"]["phase"])
+                        self.assertTrue(first["game"].get("your_turn"))
+                        self.assertEqual(
+                            ["02D", "02C", "02S", "02B"],
+                            first["game"]["callable_cards"],
+                        )
+                        await ws.send_json(
+                            {"action": "call", "card": "02D"}
+                        )
+                        update = await ws.receive_json()
+                        self.assertEqual(PHASE_PLAYING, update["game"]["phase"])
+                        self.assertEqual("D", update["game"]["briscola_suit"])
+                        self.assertEqual("02D", update["game"]["called_card"])
+                        self.assertEqual(1, update["game"]["turn"])
+
+                        # Bob is first to play (player after the dealer).
+                        bob = next(
+                            p for p in update["game"]["players"] if p["seat"] == 1
+                        )
+                        card = bob["hand"][0]
+                        await ws.send_json({"action": "play", "card": card})
+                        update = await ws.receive_json()
+                        self.assertEqual(
+                            [{"seat": 1, "card": card}], update["game"]["trick"]
+                        )
+
+    @async_test
+    async def test_spectator_cannot_act(self) -> None:
+        api_transport = ASGITransport(app=app)
+        async with AsyncClient(transport=api_transport, base_url="http://127.0.0.1") as client:
+            state = await _create_and_join(client)
+            game_id = state["id"]
+
+        ws_transport = ASGIWebSocketTransport(app=app)
+        async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client:
+            with ws_users([make_user("mallory")]):
+                async with aconnect_ws(f"/ws/games/{game_id}", ws_client) as ws:
+                    first = await ws.receive_json()
+                    self.assertTrue(first["game"]["spectator"])
+                    self.assertNotIn("join_code", first["game"])
+                    for seat in first["game"]["players"]:
+                        self.assertNotIn("hand", seat)
+                    await ws.send_json({"action": "bid", "rank": 1})
+                    error = await ws.receive_json()
+                    self.assertEqual("forbidden", error["code"])
+
+
+class ChiamataTimeoutTest(unittest.TestCase):
+    async def _seed(self, game_id: str, join_code: str, state_mutator) -> str:
+        from tavolo.briscola import chiamata as rules
+
+        session = GameSession(
+            id=game_id,
+            game_type="briscola_chiamata",
+            join_code=join_code,
+            creator_sub=PLAYERS[0],
+            players=[
+                Seat(user_sub=name, display_name=name.capitalize())
+                for name in PLAYERS
+            ],
+        )
+        session.state = rules.create_game(PLAYERS[0], "Alice", turn_timeout=1)
+        for name in PLAYERS[1:]:
+            rules.join_game(session.state, name, name.capitalize())
+        state_mutator(session.state)
+        await game_store.save(session)
+        return session.id
+
+    @async_test
+    async def test_auction_timeout_auto_bids(self) -> None:
+        game_id = await self._seed(
+            "chiamata-auction-timeout-1", "CH0001", lambda state: None
+        )
+        ws_transport = ASGIWebSocketTransport(app=app)
+        async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client:
+            with ws_users([make_user("bob")]):
+                async with aconnect_ws(f"/ws/games/{game_id}", ws_client) as ws:
+                    first = await ws.receive_json()
+                    self.assertEqual(PHASE_AUCTION, first["game"]["phase"])
+                    self.assertIsNotNone(first["game"]["auction_deadline"])
+
+                    update = None
+                    for _ in range(20):
+                        try:
+                            update = await asyncio.wait_for(
+                                ws.receive_json(), timeout=2
+                            )
+                        except asyncio.TimeoutError:
+                            break
+                        if (
+                            update.get("type") == "state"
+                            and update["game"]["auction"]["current"] is not None
+                        ):
+                            break
+                    self.assertIsNotNone(update)
+                    assert update is not None
+                    # The absent seat 1 was auto-bid for.
+                    current = update["game"]["auction"]["current"]
+                    self.assertEqual(1, current["seat"])
+                    self.assertIn(current["rank"], (1, 3, 10, 9, 8, 7, 6, 5, 4, 2))
+
+    @async_test
+    async def test_call_and_turn_timeouts(self) -> None:
+        def seed(state) -> None:
+            state.phase = PHASE_CALLING
+            state.auction = Auction(
+                turn=1,
+                active=[1],
+                passed=[0, 2, 3, 4],
+                current=Bid(seat=1, rank=2, points=61),
+            )
+            state.caller = 1
+            state.turn = 1
+            state.auction_deadline = None
+            state.call_deadline = (
+                datetime.now(timezone.utc) + timedelta(seconds=1)
+            ).isoformat()
+
+        game_id = await self._seed(
+            "chiamata-call-timeout-1", "CH0002", seed
+        )
+        ws_transport = ASGIWebSocketTransport(app=app)
+        async with AsyncClient(transport=ws_transport, base_url="http://testserver") as ws_client:
+            with ws_users([make_user("bob")]):
+                async with aconnect_ws(f"/ws/games/{game_id}", ws_client) as ws:
+                    first = await ws.receive_json()
+                    self.assertEqual(PHASE_CALLING, first["game"]["phase"])
+
+                    called = None
+                    for _ in range(20):
+                        try:
+                            called = await asyncio.wait_for(
+                                ws.receive_json(), timeout=2
+                            )
+                        except asyncio.TimeoutError:
+                            break
+                        if (
+                            called.get("type") == "state"
+                            and called["game"]["phase"] == PHASE_PLAYING
+                        ):
+                            break
+                    self.assertIsNotNone(called)
+                    assert called is not None
+                    self.assertEqual(PHASE_PLAYING, called["game"]["phase"])
+                    card = called["game"]["called_card"]
+                    self.assertIsNotNone(card)
+                    self.assertTrue(card.startswith("02"))
+
+                    played = None
+                    for _ in range(20):
+                        try:
+                            played = await asyncio.wait_for(
+                                ws.receive_json(), timeout=2
+                            )
+                        except asyncio.TimeoutError:
+                            break
+                        if (
+                            played.get("type") == "state"
+                            and len(played["game"]["trick"]) == 1
+                        ):
+                            break
+                    self.assertIsNotNone(played)
+                    assert played is not None
+                    # Seat 1 was on turn and was auto-played for.
+                    self.assertEqual(1, played["game"]["trick"][0]["seat"])
+
+
+if __name__ == "__main__":
+    unittest.main()

server/tests/test_routes_games.py

@@ -300,7 +300,12 @@ class GameTypesRouteTest(unittest.TestCase):
         self.assertEqual(200, response.status_code)
         results = response.json()["results"]
         self.assertEqual(
-            ["scopone_scientifico", "briscola", "briscola_a_quattro"],
+            [
+                "scopone_scientifico",
+                "briscola",
+                "briscola_a_quattro",
+                "briscola_chiamata",
+            ],
             [g["id"] for g in results],
         )
         scopone = results[0]
@@ -325,6 +330,12 @@ class GameTypesRouteTest(unittest.TestCase):
         self.assertEqual(4, quattro["min_players"])
         self.assertEqual(4, quattro["max_players"])
         self.assertIn("target_score", quattro["options_schema"]["properties"])
+        chiamata = results[3]
+        self.assertEqual("Briscola chiamata", chiamata["name"])
+        self.assertTrue(chiamata["description"])
+        self.assertEqual(5, chiamata["min_players"])
+        self.assertEqual(5, chiamata["max_players"])
+        self.assertEqual({}, chiamata["options_schema"])
 
     @async_test
     async def test_create_defaults_game_type(self) -> None:

server/tests/test_stats.py

@@ -12,8 +12,11 @@ import unittest
 from httpx import ASGITransport, AsyncClient
 
 from tavolo.app import app, platform, tortoise_mixin
+from tavolo.briscola.chiamata import PHASE_AUCTION, PHASE_CALLING
+from tavolo.briscola.state import Card
 from tavolo.platform import GameSession, Seat
 from tavolo.platform.elo import INITIAL_RATING
+from tavolo.platform.errors import IllegalMove
 from tavolo.platform.models import Match, MatchPlayer, PlayerRating
 from tavolo.platform.stats import save_match_result
 from tavolo.scopone import engine as rules
@@ -222,5 +225,98 @@ class BriscolaStatsTest(unittest.TestCase):
             self.assertEqual(2, len(results[0]["players"]))
 
 
+async def _finished_chiamata_session() -> GameSession:
+    engine = platform.registry.require("briscola_chiamata")
+    session = GameSession(
+        id="stats-chiamata-1",
+        game_type=engine.id,
+        join_code="BC0001",
+        creator_sub="alice",
+        players=[
+            Seat(user_sub=name, display_name=name.capitalize())
+            for name in ("alice", "bob", "carol", "dave", "erin")
+        ],
+    )
+    engine.create(session, {})
+    for name in ("bob", "carol", "dave", "erin"):
+        engine.join(session, name, name.capitalize())
+    moves = 0
+    while not engine.is_finished(session) and moves < 200000:
+        state = session.state
+        if state.phase == PHASE_AUCTION:
+            player = next(p for p in state.players if p.seat == state.turn)
+            try:
+                engine.handle_action(session, player.sub, "pass", {})
+            except IllegalMove:
+                engine.handle_action(session, player.sub, "bid", {"rank": 2})
+        elif state.phase == PHASE_CALLING:
+            assert state.auction is not None and state.auction.current is not None
+            caller = next(p for p in state.players if p.seat == state.caller)
+            card = Card(rank=state.auction.current.rank, suit="D")
+            engine.handle_action(session, caller.sub, "call", {"card": card.code})
+        else:
+            player = next(p for p in state.players if p.seat == state.turn)
+            engine.handle_action(
+                session, player.sub, "play", {"card": player.hand[0].code}
+            )
+        moves += 1
+    assert engine.is_finished(session)
+    return session
+
+
+class ChiamataStatsTest(unittest.TestCase):
+    @async_test
+    async def test_finished_chiamata_match_persisted_with_summary(self) -> None:
+        await tortoise_mixin._bind()
+        ctx = tortoise_mixin._ctx
+        assert ctx is not None
+        engine = platform.registry.require("briscola_chiamata")
+        session = await _finished_chiamata_session()
+        with ctx:
+            await save_match_result(session, engine)
+            self.assertEqual(1, await Match.all().count())
+            self.assertEqual(5, await MatchPlayer.all().count())
+
+            match = await Match.all().first()
+            assert match is not None
+            self.assertEqual("briscola_chiamata", match.game_type)
+            summary = match.result
+            for key in (
+                "team_a_score",
+                "team_b_score",
+                "winner_team",
+                "declared_target",
+                "called_card",
+                "caller",
+                "solo",
+                "deals_played",
+                "deal_scores",
+            ):
+                self.assertIn(key, summary)
+            self.assertIn(summary["winner_team"], ("Chiamante", "Difensori"))
+
+            players = await MatchPlayer.all()
+            roles = {p.team for p in players}
+            self.assertTrue(roles.issubset({"Chiamante", "Socio", "Difensori"}))
+            # Roles match the sides: exactly one caller, the rest defenders
+            # plus at most one partner.
+            self.assertEqual(1, sum(1 for p in players if p.team == "Chiamante"))
+            winner_team = summary["winner_team"]
+            for row in players:
+                on_caller_side = row.team in ("Chiamante", "Socio")
+                self.assertEqual(row.won, on_caller_side == (winner_team == "Chiamante"))
+            # Mean-team Elo: 2v3 or 1v4 moves every winner by +16.
+            ratings = {r.user_sub: r.rating for r in await PlayerRating.all()}
+            self.assertEqual(5, len(ratings))
+            winners = {p.user_sub for p in players if p.won}
+            self.assertEqual(
+                {INITIAL_RATING + 16}, {ratings[sub] for sub in winners}
+            )
+            self.assertEqual(
+                {INITIAL_RATING - 16},
+                {ratings[sub] for sub in ratings if sub not in winners},
+            )
+
+
 if __name__ == "__main__":
     unittest.main()