undercooked/tests/unit/test_ghost_parking.py

"""`sim/ghost.py` — the ghost's state and hands (phase 06 task 05, US-09).

The wall-clock grace is the room's (measured off `seat.last_seen`); this file
pins the sim's half, pure and deterministic:

* **`mark_ghost`** — the sim-side half of a drop: `connected=False`, the cook
  in `DISCONNECTED`; idempotent (a second call changes nothing);
* **`restore_player`** — the sim-side half of the rejoin inside the grace:
  `connected=True`, the cook in `IDLE`; a still-carried item is kept;
  idempotent;
* **`release_ghost_carry`** — the park's release: the carried item goes onto
  the station on the cook's tile (a marked set-down, one `DROPPED`) or into
  `counters["lost_items"]` (one `GHOST_PARK`) on a bare tile; an item is never
  deleted silently; idempotent (a double call cannot double-release); a
  not-yet-marked (connected) cook is switched first — the forced release on
  a `SeatAbandoned` when the grace ran out before the park deadline;
* **the ghost parks in the movement stage** — a disconnected cook's (stale)
  input does not move it, and a stun that runs out on a disconnected cook
  revives it in `DISCONNECTED`, not `IDLE`;
* **the events encode** — `ghost_on`/`ghost_off`/`ghost_park` ride the `k`
  names with the short payload keys (`i`/`it`/`fm`), and the unlisted keys
  never leave the server.
"""

from __future__ import annotations

import pytest

from undercooked.server.encoding import encode_events
from undercooked.sim.carry import first_loose_entry
from undercooked.sim.ghost import (
    PARK_AFTER_S,
    mark_ghost,
    release_ghost_carry,
    restore_player,
    station_at,
)
from undercooked.sim.grid import Grid
from undercooked.sim.model import (
    Event,
    EventKind,
    Facing,
    Item,
    ItemForm,
    ItemKind,
    KitchenState,
    Player,
    PlayerState,
    Station,
    StationKind,
    TileKind,
    Vec2,
)
from undercooked.sim.movement import apply_stun, resolve_moves
from undercooked.sim.step import PlayerInput

pytestmark = pytest.mark.unit

WIDTH, HEIGHT = 7, 5
#: The cook's cell — the station test's fixture sits on it.
CELL_X, CELL_Y = 3, 2


def make_state(
    stations: tuple[Station, ...] = (),
    players: tuple[Player, ...] = (),
) -> KitchenState:
    return KitchenState(
        level_id="03_ghost",
        seed=7,
        grid=Grid(width=WIDTH, height=HEIGHT, tiles=(TileKind.FLOOR,) * (WIDTH * HEIGHT)),
        stations=stations,
        players=players,
        target_score=40,
        time_left=90.0,
    )


def cook(
    player_id: str = "p0",
    *,
    pos: Vec2 | None = None,
    carrying: Item | None = None,
    connected: bool = True,
    state: PlayerState = PlayerState.IDLE,
    stun: float = 0.0,
    seat: int = 0,
) -> Player:
    return Player(
        id=player_id,
        pos=pos if pos is not None else Vec2(CELL_X + 0.5, CELL_Y + 0.5),
        facing=Facing.DOWN,
        carrying=carrying,
        connected=connected,
        state=state,
        stun=stun,
        seat=seat,
    )


def tomato(form: ItemForm = ItemForm.RAW) -> Item:
    return Item(kind=ItemKind.TOMATO, form=form)


def fixture(station_id: str, kind: StationKind = StationKind.PREP) -> Station:
    return Station(id=station_id, kind=kind, x=float(CELL_X), y=float(CELL_Y))


# --- the constant ---------------------------------------------------------------


def test_the_park_deadline_is_three_seconds() -> None:
    assert PARK_AFTER_S == 3.0


# --- mark / restore -------------------------------------------------------------


def test_mark_ghost_switches_the_cook_and_is_idempotent() -> None:
    state = make_state(players=(cook(),))
    once = mark_ghost(state, state.players[0])
    assert once.players[0].connected is False
    assert once.players[0].state is PlayerState.DISCONNECTED
    # Idempotent: a second call changes nothing (the event is the room's,
    # so the sim must not double-apply the switch either).
    twice = mark_ghost(once, once.players[0])
    assert twice is once
    assert twice.players[0].connected is False


def test_mark_ghost_keeps_the_carry() -> None:
    state = make_state(players=(cook(carrying=tomato()),))
    marked = mark_ghost(state, state.players[0])
    assert marked.players[0].carrying is not None  # the item waits for the park
    assert marked.players[0].state is PlayerState.DISCONNECTED


def test_restore_player_brings_the_cook_back_and_keeps_the_carry() -> None:
    ghost = mark_ghost(make_state(players=(cook(carrying=tomato()),)), cook(carrying=tomato()))
    player = ghost.players[0]
    restored = restore_player(ghost, player)
    assert restored.players[0].connected is True
    assert restored.players[0].state is PlayerState.IDLE
    assert restored.players[0].carrying is not None  # the rejoin keeps the carry
    # Idempotent: a second restore changes nothing.
    again = restore_player(restored, restored.players[0])
    assert again is restored


# --- the release (the park / the forced release) --------------------------------


def test_station_at_finds_the_station_on_the_tile_and_none_on_bare_floor() -> None:
    shelf = fixture("prep_0", StationKind.PREP)
    state = make_state(stations=(shelf,), players=(cook(),))
    assert station_at(state, state.players[0].pos) is shelf
    bare = make_state(players=(cook(pos=Vec2(1.5, 1.5)),))
    assert station_at(bare, Vec2(1.5, 1.5)) is None


def test_release_parks_onto_the_station_with_a_marked_set_down() -> None:
    shelf = fixture("prep_0", StationKind.PREP)
    state = make_state(stations=(shelf,), players=(cook(carrying=tomato()),))
    state = mark_ghost(state, state.players[0])
    released, events = release_ghost_carry(state, state.players[0])
    assert released.players[0].carrying is None
    station = released.stations[0]
    assert station.contents == (tomato(),)
    # A marked set-down: grabbable at once (the hand-off window applies — a
    # released item can be passed).
    entry = first_loose_entry(released, station)
    assert entry is not None and entry[0] == tomato()
    assert [e.kind for e in events] == [EventKind.DROPPED]


def test_release_parks_into_lost_items_on_a_bare_tile() -> None:
    state = make_state(players=(cook(pos=Vec2(1.5, 1.5), carrying=tomato(ItemForm.CHOPPED)),))
    state = mark_ghost(state, state.players[0])
    released, events = release_ghost_carry(state, state.players[0])
    assert released.players[0].carrying is None
    assert released.counters["lost_items"] == 1
    assert len(events) == 1
    event = events[0]
    assert event.kind is EventKind.GHOST_PARK
    assert event.payload == {
        "player_id": "p0",
        "item": "tomato",
        "form": "chopped",
    }


def test_release_is_idempotent_a_double_call_cannot_double_release() -> None:
    shelf = fixture("prep_0", StationKind.PREP)
    state = make_state(stations=(shelf,), players=(cook(carrying=tomato()),))
    state = mark_ghost(state, state.players[0])
    released, first = release_ghost_carry(state, state.players[0])
    assert len(first) == 1
    # Second call: the hand is empty, nothing to release — no events, no
    # counter bump, the set-down untouched.
    again, second = release_ghost_carry(released, released.players[0])
    assert second == ()
    assert released.counters.get("lost_items", 0) == 0
    assert again.stations[0].contents == (tomato(),)
    # Same for the counter path: the tombstone is counted exactly once.
    bare = make_state(players=(cook(pos=Vec2(1.5, 1.5), carrying=tomato()),))
    bare = mark_ghost(bare, bare.players[0])
    parked, _ = release_ghost_carry(bare, bare.players[0])
    again_bare, second_bare = release_ghost_carry(parked, parked.players[0])
    assert second_bare == ()
    assert again_bare.counters["lost_items"] == 1


def test_release_of_a_connected_cook_switches_and_releases() -> None:
    """The forced release on a `SeatAbandoned`: the seat is emptied before
    the 3 s park deadline, and the item must release at once (the sim
    switches the ghost first, then releases)."""
    shelf = fixture("prep_0", StationKind.PREP)
    state = make_state(stations=(shelf,), players=(cook(carrying=tomato()),))
    released, events = release_ghost_carry(state, state.players[0])
    assert released.players[0].connected is False
    assert released.players[0].state is PlayerState.DISCONNECTED
    assert released.players[0].carrying is None
    assert released.stations[0].contents == (tomato(),)
    assert [e.kind for e in events] == [EventKind.DROPPED]


def test_release_of_an_empty_hand_is_a_silent_noop() -> None:
    state = make_state(players=(cook(pos=Vec2(1.5, 1.5)),))
    state = mark_ghost(state, state.players[0])
    released, events = release_ghost_carry(state, state.players[0])
    assert events == ()
    assert "lost_items" not in released.counters


# --- the ghost parks in the movement stage ---------------------------------------


def test_a_stun_that_runs_out_on_a_disconnected_cook_revives_it_disconnected() -> None:
    players, _ = apply_stun((cook(connected=False, state=PlayerState.STUNNED, stun=0.05),), 0.05)
    assert players[0].state is PlayerState.DISCONNECTED
    assert players[0].connected is False
    # A stun still running keeps the cook pinned (and stunned).
    running, _ = apply_stun((cook(connected=False, state=PlayerState.STUNNED, stun=0.3),), 0.05)
    assert running[0].state is PlayerState.STUNNED
    assert running[0].stun == pytest.approx(0.25)
    # A stun-zero STUNNED cook revives to `DISCONNECTED` as well, not `IDLE`.
    zero, _ = apply_stun((cook(connected=False, state=PlayerState.STUNNED, stun=0.0),), 0.05)
    assert zero[0].state is PlayerState.DISCONNECTED
    # A connected cook still revives to `IDLE`.
    alive, _ = apply_stun((cook(state=PlayerState.STUNNED, stun=0.05),), 0.05)
    assert alive[0].state is PlayerState.IDLE


def test_a_disconnected_cook_parks_stale_input_does_not_move_it() -> None:
    state = make_state(players=(cook(connected=False, state=PlayerState.DISCONNECTED),))
    before = state.players[0].pos
    after, events = resolve_moves(
        state, 0.05, {"p0": PlayerInput(player_id="p0", move=Vec2(1.0, 0.0))}
    )
    assert after[0].pos == before  # the ghost does not walk
    assert after[0].vel.length == 0.0
    assert after[0].state is PlayerState.DISCONNECTED
    assert events == ()
    # A connected cook on the same frame still moves.
    state2 = make_state(players=(cook(state=PlayerState.IDLE),))
    after2, _ = resolve_moves(
        state2, 0.05, {"p0": PlayerInput(player_id="p0", move=Vec2(1.0, 0.0))}
    )
    assert after2[0].pos.x > state2.players[0].pos.x
    assert after2[0].state is PlayerState.MOVING


# --- the events encode -----------------------------------------------------------


def test_the_ghost_events_encode_with_their_k_names_and_short_payloads() -> None:
    encoded = encode_events(
        (
            Event(
                kind=EventKind.GHOST_ON,
                x=3.5,
                y=2.5,
                payload={"player_id": "p0"},
            ),
            Event(kind=EventKind.GHOST_OFF, x=3.5, y=2.5, payload={"player_id": "p0"}),
            Event(
                kind=EventKind.GHOST_PARK,
                x=1.5,
                y=1.5,
                payload={"player_id": "p0", "item": "tomato", "form": "chopped"},
            ),
        )
    )
    assert [entry["k"] for entry in encoded] == ["ghost_on", "ghost_off", "ghost_park"]
    assert encoded[0] == {"k": "ghost_on", "x": 3.5, "y": 2.5, "i": "p0"}
    assert encoded[2] == {
        "k": "ghost_park",
        "x": 1.5,
        "y": 1.5,
        "i": "p0",
        "it": "tomato",
        "fm": "chopped",
    }