Add game-state rollback (Phase 1)

Full-state JSON snapshots let a scene be rolled back to an earlier
action. All game state lives in game_id-scoped tables, so a checkpoint
is just a serialized dump of the gameplay rows and rollback restores
it -- sidestepping deterministic replay (the deck shuffle is
materialized into state and captured verbatim).

- Checkpoint table + GameEvent.checkpoint_id (tri-state) +
  Game.rollback_timeline_version (models.py, Alembic migration).
- crud_rollback.py: serialize/apply/capture/seal-purge/rollback.
- Capture is driven by the broadcast middleware: snapshot per action
  while in the `scene` phase, seal+purge otherwise. Confined to the
  current scene; older scenes' checkpoints are purged at scene end.
- POST /game/{gid}/player/{pid}/rollback (routes_rollback.py),
  server-enforced for Admins and Deep players.
- EventLog.svelte: per-event rollback buttons + timeline reconciliation.
- Remove the orphaned /scene/rollback per-card-play undo (dead code
  since d7f8483, never wired up; superseded by full-state rollback).

Phase 1 truncates the future immediately; the greyed/undoable redo is
deferred to Phase 2.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

tests/test_game.py

@@ -1,9 +1,9 @@
 import pytest
 import json
-from sqlmodel import SQLModel, create_engine, Session
+from sqlmodel import SQLModel, create_engine, Session, select
 from pirats import cards
 from pirats import crud
-from pirats.models import Obstacle
+from pirats.models import Obstacle, Player, Challenge, GameEvent, Checkpoint
 
 # In-memory database for testing
 @pytest.fixture(name="session")
@@ -2037,3 +2037,224 @@ def test_leave_endpoint_and_last_admin_guard():
         assert crud.get_player(session, creator.id) is None
     finally:
         app.dependency_overrides.clear()
+
+# --- Rollback ---
+
+def _checkpoints(session, game):
+    return session.exec(
+        select(Checkpoint).where(Checkpoint.game_id == game.id).order_by(Checkpoint.id)
+    ).all()
+
+def _events(session, game):
+    return session.exec(select(GameEvent).where(GameEvent.game_id == game.id)).all()
+
+def test_rollback_capture_tags_only_new_events(session):
+    game, deep, (p2,) = make_scene_game(session)
+    # Simulate the middleware having sealed the lobby/setup events (it does so on
+    # every non-scene mutation); now only fresh scene actions get tagged.
+    crud.seal_and_purge(session, game)
+    assert all(e.checkpoint_id == 0 for e in _events(session, game))
+
+    crud.add_game_event(session, game.id, "Action A", kind="info")
+    cp = crud.capture_checkpoint(session, game)
+
+    tagged = [e for e in _events(session, game) if e.message == "Action A"]
+    assert tagged and all(e.checkpoint_id == cp.id for e in tagged)
+    # The previously-sealed events are untouched.
+    assert all(e.checkpoint_id == 0 for e in _events(session, game) if e.message != "Action A")
+
+def test_rollback_round_trip_restores_state(session):
+    game, deep, (p2,) = make_scene_game(session)
+    crud.seal_and_purge(session, game)
+
+    obs = game.obstacles[0]
+    p2.hand_cards = json.dumps(["10D", "2H", "JS", "3C"])
+    obs.current_value = 5
+    session.add_all([p2, obs])
+    session.commit()
+    saved_hand = json.loads(p2.hand_cards)
+    saved_deck = crud.get_game_deck(game)
+    crud.add_game_event(session, game.id, "Known state", kind="info")
+    cp_known = crud.capture_checkpoint(session, game)
+
+    # A destructive action: open a challenge and play a card against the obstacle.
+    crud.create_challenge(session, game.id, deep.id, p2.id, [obs.id], stakes="x")
+    crud.capture_checkpoint(session, game)
+    crud.play_challenge_card(session, p2.id, obs.id, "10D")
+    crud.capture_checkpoint(session, game)
+    session.refresh(obs)
+    assert obs.current_value == 10  # changed
+    assert len(_checkpoints(session, game)) == 3
+
+    ok, msg = crud.rollback_to_checkpoint(session, game, deep, cp_known.id)
+    assert ok, msg
+
+    # State restored exactly; the challenge (created after cp_known) is gone.
+    obs_after = session.get(Obstacle, obs.id)
+    p2_after = session.get(Player, p2.id)
+    session.refresh(game)
+    assert obs_after.current_value == 5
+    assert crud.get_player_hand(p2_after) == saved_hand
+    assert crud.get_game_deck(game) == saved_deck
+    assert session.exec(select(Challenge).where(Challenge.game_id == game.id)).all() == []
+
+def test_rollback_truncates_future_and_bumps_version(session):
+    game, deep, (p2,) = make_scene_game(session)
+    crud.seal_and_purge(session, game)
+
+    crud.add_game_event(session, game.id, "First", kind="info")
+    cp1 = crud.capture_checkpoint(session, game)
+    v0 = game.rollback_timeline_version
+
+    crud.add_game_event(session, game.id, "Second", kind="info")
+    cp2 = crud.capture_checkpoint(session, game)
+    assert {c.id for c in _checkpoints(session, game)} == {cp1.id, cp2.id}
+
+    ok, msg = crud.rollback_to_checkpoint(session, game, deep, cp1.id)
+    assert ok, msg
+    session.refresh(game)
+
+    # The future checkpoint and its event are gone; cp1 and its event survive.
+    assert [c.id for c in _checkpoints(session, game)] == [cp1.id]
+    msgs = {e.message for e in _events(session, game)}
+    assert "First" in msgs and "Second" not in msgs
+    assert game.rollback_timeline_version == v0 + 1
+
+def test_rollback_permissions(session):
+    game, deep, (p2, p3) = make_scene_game(session, num_pirats=2)
+    crud.seal_and_purge(session, game)
+    crud.add_game_event(session, game.id, "Floor", kind="info")
+    cp = crud.capture_checkpoint(session, game)
+
+    # A plain Pi-Rat (not admin, not Deep) cannot roll back.
+    assert not p2.is_admin and p2.role == "pirat"
+    ok, msg = crud.rollback_to_checkpoint(session, game, p2, cp.id)
+    assert not ok
+
+    # A Deep player who is not an admin can.
+    p3.role = "deep"
+    p3.is_admin = False
+    session.add(p3)
+    session.commit()
+    ok, msg = crud.rollback_to_checkpoint(session, game, p3, cp.id)
+    assert ok, msg
+
+    # Not during a scene: rejected.
+    game.phase = "between_scenes"
+    session.add(game)
+    session.commit()
+    ok, msg = crud.rollback_to_checkpoint(session, game, deep, cp.id)
+    assert not ok
+
+def test_scene_end_seals_and_purges(session):
+    game, deep, (p2,) = make_scene_game(session)
+    crud.seal_and_purge(session, game)
+    crud.add_game_event(session, game.id, "Mid-scene", kind="info")
+    cp = crud.capture_checkpoint(session, game)
+    assert len(_checkpoints(session, game)) == 1
+
+    # Leaving the scene runs the non-scene branch: history is purged and sealed.
+    game.phase = "between_scenes"
+    session.add(game)
+    session.commit()
+    crud.maintain_rollback_history(session, game)
+
+    assert _checkpoints(session, game) == []
+    assert all(e.checkpoint_id == 0 for e in _events(session, game))
+    # The now-stale checkpoint id is rejected (back in a scene, even).
+    game.phase = "scene"
+    session.add(game)
+    session.commit()
+    ok, msg = crud.rollback_to_checkpoint(session, game, deep, cp.id)
+    assert not ok
+
+def test_rollback_middleware_capture_and_route_end_to_end():
+    """Exercises the real middleware capture path: a normal scene mutation over HTTP
+    must create a checkpoint, and the rollback endpoint must restore the prior state.
+    The middleware uses pirats.main.engine directly, so point it at the test engine."""
+    from fastapi.testclient import TestClient
+    from sqlalchemy.pool import StaticPool
+    import pirats.main as main_module
+    from pirats.database import get_session
+
+    engine = create_engine(
+        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool,
+    )
+    SQLModel.metadata.create_all(engine)
+    session = Session(engine)
+
+    game, deep, (p2,) = make_scene_game(session)
+    obs = game.obstacles[0]
+    p2.hand_cards = json.dumps(["10D", "2H", "JS", "3C"])
+    obs.current_value = 5
+    session.add_all([p2, obs])
+    session.commit()
+    crud.create_challenge(session, game.id, deep.id, p2.id, [obs.id], stakes="x")
+
+    # A floor checkpoint representing the pre-play state.
+    floor = crud.capture_checkpoint(session, game)
+
+    def get_session_override():
+        yield session
+
+    app = main_module.app
+    app.dependency_overrides[get_session] = get_session_override
+    saved_engine = main_module.engine
+    main_module.engine = engine  # so the capture middleware writes to the test DB
+    client = TestClient(app)
+    try:
+        # Normal scene mutation over HTTP -> the middleware should capture a checkpoint.
+        r = client.post(f"/api/game/{game.id}/player/{p2.id}/play-card",
+                        data={"obstacle_id": obs.id, "card_code": "10D"})
+        assert r.status_code == 200
+        session.expire_all()
+        assert session.get(Obstacle, obs.id).current_value == 10
+        assert len(_checkpoints(session, game)) == 2  # floor + the one the middleware took
+
+        # Roll back to the floor over HTTP.
+        r = client.post(f"/api/game/{game.id}/player/{deep.id}/rollback",
+                        data={"checkpoint_id": floor.id})
+        assert r.status_code == 200
+        session.expire_all()
+        assert session.get(Obstacle, obs.id).current_value == 5
+        assert [c.id for c in _checkpoints(session, game)] == [floor.id]
+    finally:
+        main_module.engine = saved_engine
+        app.dependency_overrides.clear()
+
+def test_rollback_route_enforces_permission():
+    from fastapi.testclient import TestClient
+    from sqlalchemy.pool import StaticPool
+    from pirats.main import app
+    from pirats.database import get_session
+
+    engine = create_engine(
+        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool,
+    )
+    SQLModel.metadata.create_all(engine)
+    session = Session(engine)
+
+    game, deep, pirats = make_scene_game(session, num_pirats=2)
+    p2 = pirats[0]
+    crud.seal_and_purge(session, game)
+    crud.add_game_event(session, game.id, "Floor", kind="info")
+    cp = crud.capture_checkpoint(session, game)
+
+    def get_session_override():
+        yield session
+
+    app.dependency_overrides[get_session] = get_session_override
+    client = TestClient(app)
+    try:
+        # A plain Pi-Rat is rejected by the endpoint.
+        r = client.post(f"/api/game/{game.id}/player/{p2.id}/rollback",
+                        data={"checkpoint_id": cp.id})
+        assert r.status_code == 400
+
+        # The Deep/admin succeeds.
+        r = client.post(f"/api/game/{game.id}/player/{deep.id}/rollback",
+                        data={"checkpoint_id": cp.id})
+        assert r.status_code == 200
+        assert r.json()["status"] == "ok"
+    finally:
+        app.dependency_overrides.clear()