"""`sim/stations.py` — the station state machines the phase-03 verb ladder spends on. The matrix is the spec: `ACCEPTS` (station kind x item kind) and `TRANSITIONS` (`(station kind, item form) -> (output form, seconds)`) are data, not `if` ladders. This file pins: * the full acceptance matrix, cell by cell, against an explicit expectation table — a new `StationKind` or `ItemKind` without an explicit cell here fails the build (PLAN §5: the matrix is the spec); * one test per `TRANSITIONS` row, with the timing boundary at tick granularity (substep `duration - 1` is still `WORKING`, substep `duration` is `READY`); * the stove's raw duration refined per kind (`STOVE_RAW_SECONDS`); * the hold semantics: a release freezes the progress, a resume continues it, and it never rolls back; * the occupancy: the single `worker_id` slot, the `RELEASE_TICKS`-substep grace on `cooldown`, the `HOLD_WORK` re-claim; * the dirty gating: the `STATION_DIRTY` refusal, the in-place wash that clears the dirt and pays the prop's stock back, the pick's dirty side effect (a prep's or stove's output leaves the fixture dirty); * the stock counters: `plates_clean` / `pots_clean` decremented when the work that needs the prop starts, incremented by the wash, the missing counter's `DEFAULT_STOCK` read, and the zero-stock `NOTHING_HERE` + `why` refusals; * the `BLOCKED` clear, the `READY`-empty fallback to `IDLE`, the idle no-op (hash neutrality), and the `WORKER_REACH` coupling to the action button's reach. """ from __future__ import annotations from dataclasses import replace as _replace from itertools import pairwise from typing import Any import pytest from undercooked.sim.grid import Grid from undercooked.sim.interact import ( FORGIVE, REACH, ItemTarget, StationTarget, Verb, ) from undercooked.sim.model import ( EventKind, Facing, Hazard, HazardKind, Item, ItemForm, ItemKind, KitchenState, Player, PlayerState, Station, StationKind, StationState, TileKind, Vec2, ) from undercooked.sim.recipe import RecipeBook from undercooked.sim.stations import ( DEFAULT_STOCK, RELEASE_TICKS, STOVE_RAW_SECONDS, TRANSITIONS, WORKER_REACH, _begin_work, _first_item, _station_by_id, accepts_item, accepts_kind, apply_station_verb, step_stations, transition_for, ) from undercooked.sim.step import STEP_DT, PlayerInput, step pytestmark = pytest.mark.unit WIDTH, HEIGHT = 13, 9 COOK_POS = Vec2(5.5, 5.5) # in reach of a fixture at (5, 4), whose centre is (5.5, 4.5) # --- fixtures --------------------------------------------------------------------------- def make_state( stations: tuple[Station, ...] = (), players: tuple[Player, ...] = (), hazards: tuple[Hazard, ...] = (), counters: dict[str, int] | None = None, tick: int = 0, ) -> KitchenState: return KitchenState( level_id="04_stations", seed=11, grid=Grid(width=WIDTH, height=HEIGHT, tiles=(TileKind.FLOOR,) * (WIDTH * HEIGHT)), tick=tick, stations=stations, players=players, hazards=hazards, counters=dict(counters) if counters is not None else {}, target_score=40, time_left=90.0, ) def cook( player_id: str = "p0", *, pos: Vec2 = COOK_POS, facing: Facing = Facing.UP, carrying: Item | None = None, state: PlayerState = PlayerState.IDLE, seat: int = 0, held_ticks: int = 0, ) -> Player: return Player( id=player_id, pos=pos, facing=facing, carrying=carrying, state=state, seat=seat, action_held_ticks=held_ticks, ) def station( kind: StationKind = StationKind.STOVE, *, x: int = 5, y: int = 4, state: StationState = StationState.IDLE, contents: tuple[Item, ...] = (), worker_id: str | None = None, dirty: bool = False, progress: float = 0.0, duration: float = 0.0, cooldown: float = 0.0, ) -> Station: return Station( id=f"{kind.value}-{x}-{y}", kind=kind, x=x, y=y, state=state, contents=contents, worker_id=worker_id, dirty=dirty, progress=progress, duration=duration, cooldown=cooldown, ) def item(kind: ItemKind, form: ItemForm = ItemForm.RAW) -> Item: return Item(kind=kind, form=form) def working_station( kind: StationKind, contents: tuple[Item, ...], duration: float, *, worker_id: str = "p0", progress: float = 0.0, ) -> Station: return station( kind, state=StationState.WORKING, contents=contents, worker_id=worker_id, progress=progress, duration=duration, ) def run(state: KitchenState, substeps: int) -> tuple[KitchenState, list[Any]]: """Drive `substeps` substeps through `step_stations`; collect the events.""" events: list[Any] = [] for _ in range(substeps): state, produced = step_stations(state, STEP_DT) events.extend(produced) return state, events def ticks(duration: float) -> int: """The substeps a `duration`-second work needs (the kernel's 0.05 s grid).""" return round(duration / STEP_DT) # --- the acceptance matrix: total, and a new enum member must have a cell ------------- #: The spec, cell by cell — written over the *named* members so a new #: `StationKind` / `ItemKind` without an explicit cell here fails the build. #: #: * the spawn shelf takes a set-down of anything (stock; the kind comes from #: the level's legend, phase 05); #: * the chop board works every raw ingredient except rice (PLAN §5: "raw rice #: needs no board") and never a plate; #: * the stove works every ingredient but not a plate; #: * the prep plates every ingredient (the plate is part of what it operates on); #: * the serve window consumes a plated dish only; #: * the sink washes a dirty prop — the plate, until task 04 adds the pot's item; #: * the trash consumes anything. ACCEPTS_EXPECTED: dict[StationKind, dict[ItemKind, bool]] = { StationKind.SPAWN_SHELF: { ItemKind.TOMATO: True, ItemKind.ONION: True, ItemKind.LETTUCE: True, ItemKind.DOUGH: True, ItemKind.PATTY: True, ItemKind.RICE: True, ItemKind.FISH: True, ItemKind.PLATE: True, }, StationKind.CHOP_BOARD: { ItemKind.TOMATO: True, ItemKind.ONION: True, ItemKind.LETTUCE: True, ItemKind.DOUGH: True, ItemKind.PATTY: True, ItemKind.RICE: False, ItemKind.FISH: True, ItemKind.PLATE: False, }, StationKind.STOVE: { ItemKind.TOMATO: True, ItemKind.ONION: True, ItemKind.LETTUCE: True, ItemKind.DOUGH: True, ItemKind.PATTY: True, ItemKind.RICE: True, ItemKind.FISH: True, ItemKind.PLATE: False, }, StationKind.PREP: { ItemKind.TOMATO: True, ItemKind.ONION: True, ItemKind.LETTUCE: True, ItemKind.DOUGH: True, ItemKind.PATTY: True, ItemKind.RICE: True, ItemKind.FISH: True, ItemKind.PLATE: True, }, StationKind.SERVE: { ItemKind.TOMATO: False, ItemKind.ONION: False, ItemKind.LETTUCE: False, ItemKind.DOUGH: False, ItemKind.PATTY: False, ItemKind.RICE: False, ItemKind.FISH: False, ItemKind.PLATE: True, }, StationKind.SINK: { ItemKind.TOMATO: False, ItemKind.ONION: False, ItemKind.LETTUCE: False, ItemKind.DOUGH: False, ItemKind.PATTY: False, ItemKind.RICE: False, ItemKind.FISH: False, ItemKind.PLATE: True, }, StationKind.TRASH: { ItemKind.TOMATO: True, ItemKind.ONION: True, ItemKind.LETTUCE: True, ItemKind.DOUGH: True, ItemKind.PATTY: True, ItemKind.RICE: True, ItemKind.FISH: True, ItemKind.PLATE: True, }, } def test_the_acceptance_matrix_is_total_and_matches_the_spec() -> None: """Every `StationKind x ItemKind` cell has an explicit expectation, and the module's data agrees with it — so a new enum member without a cell fails.""" assert set(ACCEPTS_EXPECTED) == set(StationKind), "a station kind has no expectation row" for kind in StationKind: assert set(ACCEPTS_EXPECTED[kind]) == set(ItemKind), ( f"a station kind's row is missing a cell: {kind}" ) for kind in StationKind: for entry in ItemKind: assert accepts_kind(kind, entry) is ACCEPTS_EXPECTED[kind][entry], ( f"matrix cell {kind} x {entry} disagrees with the spec" ) def test_accepts_item_is_the_transitions_lookup() -> None: """`accepts_item` (form-level) is `transition_for` returning non-`None`.""" for kind in StationKind: for entry in ItemKind: for form in ( ItemForm.RAW, ItemForm.CHOPPED, ItemForm.COOKED, ItemForm.PLATED, ItemForm.DIRTY, ): assert accepts_item(kind, item(entry, form)) is ( transition_for(kind, item(entry, form)) is not None ) # --- one test per transition row, with the timing boundary ------------------------------ def test_chop_board_works_raw_to_chopped_in_point_nine() -> None: state = make_state( stations=(working_station(StationKind.CHOP_BOARD, (item(ItemKind.TOMATO),), 0.9),), players=(cook("p0", held_ticks=1),), ) state, _events = run(state, ticks(0.9) - 1) assert state.stations[0].state is StationState.WORKING # the boundary: one tick short assert state.stations[0].progress == pytest.approx((ticks(0.9) - 1) * STEP_DT / 0.9) state, _events = run(state, 1) assert state.stations[0].state is StationState.READY assert state.stations[0].contents[0].form is ItemForm.CHOPPED assert state.stations[0].worker_id is None def test_stove_works_chopped_to_cooked_in_six() -> None: state = make_state( stations=( working_station( StationKind.STOVE, (item(ItemKind.TOMATO, ItemForm.CHOPPED),), 6.0 ), ), players=(cook("p0", held_ticks=1),), ) state, _events = run(state, ticks(6.0) - 1) assert state.stations[0].state is StationState.WORKING state, _events = run(state, 1) assert state.stations[0].state is StationState.READY assert state.stations[0].contents[0].form is ItemForm.COOKED def test_stove_works_raw_to_cooked_in_eight_by_default() -> None: state = make_state( stations=( working_station(StationKind.STOVE, (item(ItemKind.ONION),), 8.0), ), players=(cook("p0", held_ticks=1),), ) state, _events = run(state, ticks(8.0) - 1) assert state.stations[0].state is StationState.WORKING state, _events = run(state, 1) assert state.stations[0].state is StationState.READY assert state.stations[0].contents[0].form is ItemForm.COOKED @pytest.mark.parametrize( "kind,seconds", [ (ItemKind.RICE, 8.0), (ItemKind.PATTY, 5.0), (ItemKind.FISH, 5.0), ], ids=["rice", "patty", "fish"], ) def test_stove_raw_durations_are_refined_per_kind(kind: ItemKind, seconds: float) -> None: refined = transition_for(StationKind.STOVE, item(kind)) assert refined is not None assert refined[0] is ItemForm.COOKED assert refined[1] == STOVE_RAW_SECONDS[kind] == seconds state = make_state( stations=(working_station(StationKind.STOVE, (item(kind),), seconds),), players=(cook("p0", held_ticks=1),), ) state, _events = run(state, ticks(seconds) - 1) assert state.stations[0].state is StationState.WORKING state, _events = run(state, 1) assert state.stations[0].state is StationState.READY def test_a_stove_kind_absent_from_the_refinement_keeps_the_default() -> None: """A kind with no `STOVE_RAW_SECONDS` entry keeps the table's 8 s duration.""" refined = transition_for(StationKind.STOVE, item(ItemKind.LETTUCE)) assert refined == (ItemForm.COOKED, TRANSITIONS[(StationKind.STOVE, ItemForm.RAW)][1]) assert ItemKind.LETTUCE not in STOVE_RAW_SECONDS def test_prep_works_cooked_to_plated_in_point_six() -> None: state = make_state( stations=( working_station( StationKind.PREP, (item(ItemKind.TOMATO, ItemForm.COOKED),), 0.6 ), ), players=(cook("p0", held_ticks=1),), ) state, _events = run(state, ticks(0.6) - 1) assert state.stations[0].state is StationState.WORKING state, _events = run(state, 1) assert state.stations[0].state is StationState.READY assert state.stations[0].contents[0].form is ItemForm.PLATED def test_prep_works_chopped_to_plated_in_point_six() -> None: state = make_state( stations=( working_station( StationKind.PREP, (item(ItemKind.LETTUCE, ItemForm.CHOPPED),), 0.6 ), ), players=(cook("p0", held_ticks=1),), ) state, _events = run(state, ticks(0.6) - 1) assert state.stations[0].state is StationState.WORKING state, _events = run(state, 1) assert state.stations[0].state is StationState.READY assert state.stations[0].contents[0].form is ItemForm.PLATED def test_sink_washes_dirty_to_clean_in_point_two() -> None: state = make_state( stations=( working_station( StationKind.SINK, (item(ItemKind.PLATE, ItemForm.DIRTY),), 1.2, ), ), players=(cook("p0", held_ticks=1),), ) state, _events = run(state, ticks(1.2) - 1) assert state.stations[0].state is StationState.WORKING state, _events = run(state, 1) assert state.stations[0].state is StationState.READY # the dirt goes, the prop comes back clean assert state.stations[0].contents[0].form is ItemForm.RAW # --- the hold semantics: freeze, resume, never roll back -------------------------------- def test_a_released_hold_freezes_the_progress() -> None: state = make_state( stations=(working_station(StationKind.CHOP_BOARD, (item(ItemKind.TOMATO),), 0.9),), players=(cook("p0", held_ticks=1),), ) state, _events = run(state, 9) frozen = state.stations[0].progress assert state.stations[0].state is StationState.WORKING # the button comes up: the gate's job zeroes `action_held_ticks` (the stage-4 # driver reads the cook the state holds — here the cook is swapped directly). released = replace_player(state, "p0", held_ticks=0) for _ in range(6): # several substeps of "released" released, _events = step_stations(released, STEP_DT) assert released.stations[0].state is StationState.WORKING assert released.stations[0].progress == pytest.approx(frozen) # frozen, not rolled back def test_a_resumed_hold_continues_and_never_rolls_back() -> None: state = make_state( stations=(working_station(StationKind.CHOP_BOARD, (item(ItemKind.TOMATO),), 0.9),), players=(cook("p0", held_ticks=1),), ) state, _events = run(state, 9) state = replace_player(state, "p0", held_ticks=0) for _ in range(4): state, _events = step_stations(state, STEP_DT) state = replace_player(state, "p0", held_ticks=1) seen: list[float] = [] for _ in range(ticks(0.9)): state, _events = step_stations(state, STEP_DT) seen.append(state.stations[0].progress) assert all(b >= a for a, b in pairwise(seen)) # never rolls back assert state.stations[0].state is StationState.READY assert state.stations[0].progress == 1.0 def test_a_worker_out_of_reach_freezes_the_progress() -> None: state = make_state( stations=(working_station(StationKind.CHOP_BOARD, (item(ItemKind.TOMATO),), 0.9),), players=(cook("p0", held_ticks=1),), ) state, _events = run(state, 4) frozen = state.stations[0].progress # the cook walks out of `WORKER_REACH` (3.0 tiles from the fixture's centre) state = replace_player(state, "p0", pos=Vec2(8.5, 7.5), held_ticks=1) for _ in range(3): state, _events = step_stations(state, STEP_DT) assert state.stations[0].progress == pytest.approx(frozen) assert state.stations[0].state is StationState.WORKING # --- the occupancy: the single worker slot, the grace, the re-claim --------------------- def test_a_second_cook_is_busy_at_a_working_station() -> None: state = make_state( stations=(working_station(StationKind.CHOP_BOARD, (item(ItemKind.TOMATO),), 0.9),), players=(cook("p0", held_ticks=1), cook("p1", pos=Vec2(6.5, 5.5), seat=1)), ) target = StationTarget(station=state.stations[0]) _new_state, events = apply_station_verb(state, state.players[1], _verb("start_work"), target) assert len(events) == 1 assert dict(events[0].payload)["reason"] == "station_busy" def test_the_workers_own_press_is_a_held_no_op() -> None: state = make_state( stations=(working_station(StationKind.CHOP_BOARD, (item(ItemKind.TOMATO),), 0.9),), players=(cook("p0", held_ticks=1),), ) target = StationTarget(station=state.stations[0]) new_state, events = apply_station_verb(state, state.players[0], _verb("hold_work"), target) assert new_state is state # the progress is stage 4's, not the press's assert events == () def test_the_release_grace_runs_on_the_cooldown_and_expires_on_the_fourth_substep() -> None: assert RELEASE_TICKS == 4 state = make_state( stations=( working_station(StationKind.CHOP_BOARD, (item(ItemKind.TOMATO),), 0.9, progress=0.5), ), players=(cook("p0", held_ticks=1),), ) # the cook leaves the reach: substep 1 starts the grace on `cooldown` state = replace_player(state, "p0", pos=Vec2(9.5, 8.5), held_ticks=1) for _ in range(1, RELEASE_TICKS): state, _events = step_stations(state, STEP_DT) assert state.stations[0].worker_id == "p0" # the slot is held through the grace assert state.stations[0].state is StationState.WORKING assert state.stations[0].progress == pytest.approx(0.5) # frozen the whole time state, _events = step_stations(state, STEP_DT) # the RELEASE_TICKS-th substep assert state.stations[0].worker_id is None # released assert state.stations[0].state is StationState.WORKING # the work stays, frozen assert state.stations[0].cooldown == 0.0 def test_a_released_station_stays_frozen_without_a_new_worker() -> None: state = make_state( stations=( working_station(StationKind.CHOP_BOARD, (item(ItemKind.TOMATO),), 0.9, progress=0.5), ), players=(cook("p0", pos=Vec2(9.5, 8.5), held_ticks=1),), ) for _ in range(RELEASE_TICKS): # the grace expires state, _events = step_stations(state, STEP_DT) assert state.stations[0].worker_id is None for _ in range(3): # and stays released (no oscillation of the grace) state, _events = step_stations(state, STEP_DT) assert state.stations[0].worker_id is None assert state.stations[0].progress == pytest.approx(0.5) def test_a_new_cooks_hold_work_press_reclaims_a_released_slot() -> None: state = make_state( stations=( working_station(StationKind.CHOP_BOARD, (item(ItemKind.TOMATO),), 0.9, progress=0.5), ), players=( cook("p0", pos=Vec2(9.5, 8.5), held_ticks=1), cook("p1", pos=Vec2(6.5, 5.5), seat=1), ), ) for _ in range(RELEASE_TICKS): # p0's grace expires state, _events = step_stations(state, STEP_DT) assert state.stations[0].worker_id is None # p1 (in reach, holding) presses: the ladder gives `HOLD_WORK`, the press re-claims state, _events = step_stations(state, STEP_DT) # still released: nothing advances it assert state.stations[0].worker_id is None # a hold does not claim; the press does target = StationTarget(station=state.stations[0]) state, _events = apply_station_verb(state, state.players[1], _verb("hold_work"), target) assert state.stations[0].worker_id == "p1" state = replace_player(state, "p1", held_ticks=1) state, _events = step_stations(state, STEP_DT) # the new worker's hold advances the work assert state.stations[0].progress > 0.5 def test_the_hold_advances_progress_only_while_held_and_in_reach() -> None: state = make_state( stations=( working_station(StationKind.STOVE, (item(ItemKind.TOMATO, ItemForm.CHOPPED),), 6.0), ), players=(cook("p0", held_ticks=1),), ) state, _events = run(state, 10) assert state.stations[0].progress > 0.0 # the button comes up, the cook stays: frozen (no release — they are still in reach) state = replace_player(state, "p0", held_ticks=0) state, _events = step_stations(state, STEP_DT) assert state.stations[0].worker_id == "p0" # no grace: the worker never left the reach assert state.stations[0].cooldown == 0.0 # --- the dirty gating: the refusal, the in-place wash, the pick's side effect ---------- def test_a_dirty_station_refuses_the_use_verbs_at_the_pipeline() -> None: state = make_state( stations=(station(StationKind.PREP, dirty=True, state=StationState.DIRTY),), players=(cook("p0", carrying=item(ItemKind.TOMATO, ItemForm.COOKED)),), ) target = StationTarget(station=state.stations[0]) _new_state, events = apply_station_verb(state, state.players[0], _verb("plate_dish"), target) assert len(events) == 1 assert dict(events[0].payload)["reason"] == "station_dirty" def test_the_in_place_wash_clears_the_dirt_and_pays_the_stock_back() -> None: state = make_state( stations=(station(StationKind.PREP, dirty=True, state=StationState.DIRTY),), players=(cook("p0"),), counters={"plates_clean": 3}, ) target = StationTarget(station=state.stations[0]) state, _events = apply_station_verb(state, state.players[0], _verb("wash"), target) assert state.stations[0].state is StationState.WORKING assert state.stations[0].duration == 1.2 state = replace_player(state, "p0", held_ticks=1) # the gate's increment, simulated state, _events = run(state, ticks(1.2)) assert state.stations[0].state is StationState.READY assert not state.stations[0].dirty assert state.counters["plates_clean"] == 4 # the prop is back in stock kinds = [event.kind for event in _events] assert kinds.count(EventKind.WASHED) == 1 state, _events = run(state, 1) # a READY with empty contents is idle again assert state.stations[0].state is StationState.IDLE def test_the_in_place_wash_of_a_stove_returns_the_pot_to_stock() -> None: state = make_state( stations=(station(StationKind.STOVE, dirty=True, state=StationState.DIRTY),), players=(cook("p0"),), counters={"pots_clean": 1}, ) target = StationTarget(station=state.stations[0]) state, _events = apply_station_verb(state, state.players[0], _verb("wash"), target) state = replace_player(state, "p0", held_ticks=1) # the gate's increment, simulated state, _events = run(state, ticks(1.2)) assert state.counters["pots_clean"] == 2 def test_a_dirty_station_refuses_a_second_wash_while_working() -> None: state = make_state( stations=( station( StationKind.PREP, state=StationState.WORKING, dirty=True, worker_id="p0", progress=0.2, duration=1.2, ), ), players=(cook("p0", held_ticks=1), cook("p1", pos=Vec2(6.5, 5.5), seat=1)), ) target = StationTarget(station=state.stations[0]) _new_state, events = apply_station_verb(state, state.players[1], _verb("wash"), target) assert len(events) == 1 assert dict(events[0].payload)["reason"] == "station_busy" def test_picking_a_preps_output_leaves_the_prep_dirty() -> None: state = make_state( stations=( station( StationKind.PREP, state=StationState.READY, contents=(item(ItemKind.TOMATO, ItemForm.PLATED),), ), ), players=(cook("p0"),), ) target = ItemTarget(item=state.stations[0].contents[0], on_station_id=state.stations[0].id) new_state, events = apply_station_verb(state, state.players[0], _verb("pick_item"), target) assert new_state.stations[0].state is StationState.DIRTY assert new_state.stations[0].dirty assert new_state.stations[0].worker_id is None assert len(events) == 1 and events[0].kind is EventKind.PICKED def test_picking_a_stoves_output_leaves_the_stove_dirty() -> None: state = make_state( stations=( station( StationKind.STOVE, state=StationState.READY, contents=(item(ItemKind.RICE, ItemForm.COOKED),), ), ), players=(cook("p0"),), ) target = ItemTarget(item=state.stations[0].contents[0], on_station_id=state.stations[0].id) new_state, _events = apply_station_verb(state, state.players[0], _verb("pick_item"), target) assert new_state.stations[0].state is StationState.DIRTY assert new_state.stations[0].dirty def test_picking_a_raw_output_does_not_dirty_the_station() -> None: """Only a prep's plated dish or a stove's cooked item leaves the fixture dirty.""" state = make_state( stations=( station( StationKind.STOVE, state=StationState.READY, contents=(item(ItemKind.TOMATO, ItemForm.CHOPPED),), ), ), players=(cook("p0"),), ) target = ItemTarget(item=state.stations[0].contents[0], on_station_id=state.stations[0].id) new_state, _events = apply_station_verb(state, state.players[0], _verb("pick_item"), target) assert not new_state.stations[0].dirty assert new_state.stations[0].state is StationState.READY # --- the stock: lazy default, the decrement, the zero-stock refusal --------------------- def test_a_missing_stock_counter_reads_as_the_default() -> None: state = make_state( stations=(station(StationKind.PREP),), players=(cook("p0", carrying=item(ItemKind.TOMATO, ItemForm.COOKED)),), ) target = StationTarget(station=state.stations[0]) new_state, events = apply_station_verb(state, state.players[0], _verb("plate_dish"), target) assert events == () # the default 99 allows the work assert new_state.counters["plates_clean"] == DEFAULT_STOCK - 1 def test_a_zero_plate_stock_refuses_a_plate_dish_with_the_why() -> None: state = make_state( stations=(station(StationKind.PREP),), players=(cook("p0", carrying=item(ItemKind.TOMATO, ItemForm.COOKED)),), counters={"plates_clean": 0}, ) target = StationTarget(station=state.stations[0]) _new_state, events = apply_station_verb(state, state.players[0], _verb("plate_dish"), target) assert len(events) == 1 assert dict(events[0].payload) == { "player_id": "p0", "reason": "nothing_here", "verb": "plate_dish", "why": "no_plate", } def test_a_zero_pot_stock_refuses_a_stove_start_with_the_why() -> None: state = make_state( stations=(station(StationKind.STOVE, contents=(item(ItemKind.TOMATO, ItemForm.CHOPPED),)),), players=(cook("p0"),), counters={"pots_clean": 0}, ) target = StationTarget(station=state.stations[0]) _new_state, events = apply_station_verb(state, state.players[0], _verb("start_work"), target) assert len(events) == 1 assert dict(events[0].payload)["reason"] == "nothing_here" assert dict(events[0].payload)["why"] == "no_pot" def test_a_set_down_on_a_stove_takes_a_pot_and_starts_the_work() -> None: state = make_state( stations=(station(StationKind.STOVE),), players=(cook("p0", carrying=item(ItemKind.TOMATO, ItemForm.CHOPPED)),), counters={"pots_clean": 2}, ) target = StationTarget(station=state.stations[0]) new_state, events = apply_station_verb( state, state.players[0], _verb("place_on_station"), target ) assert len(events) == 1 and events[0].kind is EventKind.DROPPED assert new_state.stations[0].state is StationState.WORKING assert new_state.stations[0].duration == 6.0 assert new_state.counters["pots_clean"] == 1 assert new_state.players[0].carrying is None def test_a_non_workable_set_down_stays_a_marked_object() -> None: """A set-down the station does not work (a raw item at the prep) is not a refusal: the set-down happens and stays a grabbable object (the hand-off's memory).""" state = make_state( stations=(station(StationKind.PREP),), players=(cook("p0", carrying=item(ItemKind.LETTUCE)),), ) target = StationTarget(station=state.stations[0]) new_state, events = apply_station_verb(state, state.players[0], _verb("plate_dish"), target) assert len(events) == 1 and events[0].kind is EventKind.DROPPED assert new_state.stations[0].state is StationState.IDLE # no work started assert new_state.item_marks[state.stations[0].id] == (("p0", 0),) # the mark stays def test_a_start_work_on_a_surface_that_works_nothing_is_wrong_item() -> None: state = make_state( stations=(station(StationKind.PREP, contents=(item(ItemKind.LETTUCE),)),), players=(cook("p0"),), ) target = StationTarget(station=state.stations[0]) _new_state, events = apply_station_verb(state, state.players[0], _verb("start_work"), target) assert len(events) == 1 assert dict(events[0].payload)["reason"] == "wrong_item" def test_a_start_work_by_a_full_hand_is_hands_full() -> None: state = make_state( stations=(station(StationKind.CHOP_BOARD, contents=(item(ItemKind.TOMATO),)),), players=(cook("p0", carrying=item(ItemKind.ONION)),), ) target = StationTarget(station=state.stations[0]) _new_state, events = apply_station_verb(state, state.players[0], _verb("start_work"), target) assert len(events) == 1 assert dict(events[0].payload)["reason"] == "hands_full" # --- the serve window, the trash, the shelf no-op -------------------------------------- def test_a_bookless_serve_press_is_the_named_caller_bug() -> None: """The `SERVE` press is the scoring's (`sim/scoring.py`, task 03): the kernel threads the book in. A bookless press that would validate a dish is the named `ValueError` — a caller bug, not a silent rejection.""" state = make_state( stations=(station(StationKind.SERVE),), players=(cook("p0", carrying=item(ItemKind.PLATE, ItemForm.PLATED)),), ) target = StationTarget(station=state.stations[0]) with pytest.raises(ValueError, match="recipe book"): apply_station_verb(state, state.players[0], _verb("serve"), target) def test_a_serve_press_with_the_book_is_the_scoring_rejection_when_the_rail_is_empty() -> None: """With the book, the `SERVE` press routes to `scoring.resolve_serve`: an empty rail is the named `DENIED` (`no_orders`) and the state is the same object back — the dish stays in the hand.""" state = make_state( stations=(station(StationKind.SERVE),), players=(cook("p0", carrying=item(ItemKind.PLATE, ItemForm.PLATED)),), ) target = StationTarget(station=state.stations[0]) new_state, events = apply_station_verb( state, state.players[0], _verb("serve"), target, book() ) assert new_state is state assert len(events) == 1 assert events[0].kind is EventKind.DENIED assert dict(events[0].payload)["reason"] == "no_orders" assert state.players[0].carrying is not None def test_a_trash_press_is_the_deliberate_no_op_the_gate_spends() -> None: state = make_state( stations=(station(StationKind.TRASH),), players=(cook("p0", carrying=item(ItemKind.TOMATO)),), ) target = StationTarget(station=state.stations[0]) new_state, events = apply_station_verb(state, state.players[0], _verb("trash"), target) assert new_state is state assert events == () def test_an_empty_shelf_pick_is_the_documented_no_op() -> None: """The shelf's stock is the level's legend (phase 05's room seeds it); a pick from an empty shelf is a no-op, not a refusal.""" state = make_state( stations=(station(StationKind.SPAWN_SHELF),), players=(cook("p0"),), ) target = StationTarget(station=state.stations[0]) new_state, events = apply_station_verb( state, state.players[0], _verb("pick_from_station"), target ) assert new_state is state assert events == () def test_the_wash_verb_at_a_clean_station_is_invalid_state() -> None: state = make_state( stations=(station(StationKind.CHOP_BOARD),), players=(cook("p0"),), ) target = StationTarget(station=state.stations[0]) _new_state, events = apply_station_verb(state, state.players[0], _verb("wash"), target) assert len(events) == 1 assert dict(events[0].payload)["reason"] == "invalid_state" # --- BLOCKED, READY-empty, the idle no-op, the reach coupling -------------------------- def test_a_blocked_station_stays_blocked_while_a_hazard_names_it() -> None: state = make_state( stations=(station(StationKind.STOVE, state=StationState.BLOCKED),), players=(cook("p0"),), hazards=(Hazard(id="h1", kind=HazardKind.FIRE, station_id="stove-5-4"),), ) state, _events = run(state, 3) assert state.stations[0].state is StationState.BLOCKED def test_a_blocked_station_clears_when_its_cause_goes() -> None: state = make_state( stations=(station(StationKind.STOVE, state=StationState.BLOCKED),), players=(cook("p0"),), hazards=(Hazard(id="h1", kind=HazardKind.FIRE, station_id="stove-5-4"),), ) state = make_state( # the hazard is gone (phase 10's removal) stations=state.stations, players=state.players, ) state, _events = run(state, 1) assert state.stations[0].state is StationState.IDLE def test_a_ready_station_with_empty_contents_falls_back_to_idle() -> None: state = make_state( stations=(station(StationKind.PREP, state=StationState.READY, progress=1.0),), players=(cook("p0"),), ) state, _events = run(state, 1) assert state.stations[0].state is StationState.IDLE assert state.stations[0].progress == 0.0 def test_a_ready_station_with_contents_stays_ready() -> None: state = make_state( stations=( station( StationKind.PREP, state=StationState.READY, contents=(item(ItemKind.TOMATO, ItemForm.PLATED),), ), ), players=(cook("p0"),), ) state, _events = run(state, 3) assert state.stations[0].state is StationState.READY def test_an_idle_kitchen_is_an_unmodified_no_op() -> None: """Hash neutrality: a kitchen with no advancing station is the same state back, untouched — the committed replay goldens rest on it.""" fresh = make_state( stations=( station(StationKind.SPAWN_SHELF, contents=(item(ItemKind.TOMATO),)), station(StationKind.PREP, x=7), station(StationKind.STOVE, x=9, dirty=True), ), players=(cook("p0"),), ) before = fresh.state_hash() for _ in range(5): state, events = step_stations(fresh, STEP_DT) assert state is fresh # the same state object back, every substep assert events == () assert state.state_hash() == before def test_worker_reach_couples_to_the_action_buttons_reach() -> None: """A worker stands exactly where a press reaches: `WORKER_REACH` is `REACH + FORGIVE`.""" assert WORKER_REACH == REACH + FORGIVE # --- the verb guards, the helpers' error paths, the edge completions ---------------- def test_a_sink_wash_via_the_verb_starts_and_completes_and_pays_the_stock_back() -> None: state = make_state( stations=(station(StationKind.SINK),), players=(cook("p0", carrying=item(ItemKind.PLATE, ItemForm.DIRTY)),), counters={"plates_clean": 2}, ) target = StationTarget(station=state.stations[0]) state, _events = apply_station_verb(state, state.players[0], _verb("wash"), target) assert state.stations[0].state is StationState.WORKING assert state.stations[0].duration == 1.2 # the sink's DIRTY -> RAW row assert state.players[0].carrying is None # the prop left the hand state = replace_player(state, "p0", held_ticks=1) # the gate's increment, simulated state, events = run(state, ticks(1.2)) assert state.stations[0].state is StationState.READY assert state.stations[0].contents[0].form is ItemForm.RAW assert state.counters["plates_clean"] == 3 # the washed plate is back in stock assert [event.kind for event in events].count(EventKind.WASHED) == 1 def test_the_verb_guards_name_their_refusals() -> None: items = { "tomato": item(ItemKind.TOMATO), "cooked": item(ItemKind.TOMATO, ItemForm.COOKED), "raw": item(ItemKind.LETTUCE), } cases: list[tuple[str, str, dict[str, Any], dict[str, Any], str]] = [ # (name, verb, station kwargs, player kwargs, expected reason) ( "a_set_down_with_empty_hands_is_hands_empty", "place_on_station", {"kind": StationKind.CHOP_BOARD}, {}, "hands_empty", ), ( "a_set_down_at_a_working_station_is_station_busy", "place_on_station", {"kind": StationKind.CHOP_BOARD, "state": StationState.WORKING}, {"carrying": items["tomato"]}, "station_busy", ), ( "a_plate_dish_at_a_non_prep_is_invalid_state", "plate_dish", {"kind": StationKind.CHOP_BOARD}, {"carrying": items["cooked"]}, "invalid_state", ), ( "a_plate_dish_with_empty_hands_is_hands_empty", "plate_dish", {"kind": StationKind.PREP}, {}, "hands_empty", ), ( "a_plate_dish_at_a_loaded_prep_is_station_busy", "plate_dish", {"kind": StationKind.PREP, "contents": (items["raw"],)}, {"carrying": items["cooked"]}, "station_busy", ), ( "a_start_work_at_a_non_work_surface_is_invalid_state", "start_work", { "kind": StationKind.SINK, "contents": (item(ItemKind.PLATE, ItemForm.DIRTY),), }, {}, "invalid_state", ), ( "an_in_place_wash_with_a_full_hand_is_hands_full", "wash", {"kind": StationKind.PREP, "dirty": True, "state": StationState.DIRTY}, {"carrying": items["tomato"]}, "hands_full", ), ( "a_busy_station_refuses_with_station_busy", "wash", {"kind": StationKind.PREP, "state": StationState.BUSY}, {}, "station_busy", ), ] for name, verb_name, station_kwargs, player_kwargs, expected in cases: state = make_state( stations=(station(**station_kwargs),), players=(cook("p0", **player_kwargs),), ) target = StationTarget(station=state.stations[0]) _new_state, events = apply_station_verb( state, state.players[0], _verb(verb_name), target ) assert len(events) == 1, name assert dict(events[0].payload)["reason"] == expected, name def test_a_start_work_on_a_loaded_prep_takes_a_plate_and_starts_the_work() -> None: state = make_state( stations=( station(StationKind.PREP, contents=(item(ItemKind.TOMATO, ItemForm.COOKED),)), ), players=(cook("p0"),), counters={"plates_clean": 4}, ) target = StationTarget(station=state.stations[0]) new_state, _events = apply_station_verb(state, state.players[0], _verb("start_work"), target) assert new_state.stations[0].state is StationState.WORKING assert new_state.stations[0].duration == 0.6 assert new_state.counters["plates_clean"] == 3 def test_a_none_target_and_a_ghost_holder_are_deliberate_no_ops() -> None: state = make_state( stations=(station(StationKind.CHOP_BOARD, contents=(item(ItemKind.TOMATO),)),), players=(cook("p0"),), ) new_state, events = apply_station_verb(state, state.players[0], _verb("start_work"), None) assert new_state is state and events == () ghost = ItemTarget(item=item(ItemKind.TOMATO), on_station_id="ghost-1-1") new_state, events = apply_station_verb(state, state.players[0], _verb("pick_item"), ghost) assert new_state is state and events == () def test_the_helpers_raise_on_caller_bugs() -> None: state = make_state( stations=(station(StationKind.CHOP_BOARD, contents=(item(ItemKind.TOMATO),)),), players=(cook("p0"),), ) with pytest.raises(ValueError, match="not in the state"): _station_by_id(state, "ghost-1-1") empty_prep = station(StationKind.PREP) with pytest.raises(ValueError, match="no content to work"): _first_item(empty_prep) with pytest.raises(ValueError, match="no work to start"): _begin_work( state, state.players[0], station(StationKind.PREP, contents=(item(ItemKind.LETTUCE),)) ) def test_a_start_work_on_a_marked_board_keeps_the_other_marks() -> None: state = make_state( stations=( station( StationKind.CHOP_BOARD, contents=(item(ItemKind.TOMATO), item(ItemKind.ONION)), ), ), players=(cook("p0"),), ) state = _replace( state, item_marks={state.stations[0].id: (("p9", 0), ("p9", 1))} ) target = StationTarget(station=state.stations[0]) new_state, _events = apply_station_verb(state, state.players[0], _verb("start_work"), target) # the mark of the worked entry goes; the other entry's mark stays assert new_state.item_marks[state.stations[0].id] == (("p9", 1),) def test_an_empty_working_station_completes_to_ready_and_back_to_idle() -> None: state = make_state( stations=( station( StationKind.CHOP_BOARD, state=StationState.WORKING, worker_id="p0", duration=0.6, ), ), players=(cook("p0", held_ticks=1),), ) state, _events = run(state, ticks(0.6)) assert state.stations[0].state is StationState.READY assert state.stations[0].contents == () state, _events = run(state, 1) assert state.stations[0].state is StationState.IDLE # --- the kernel wiring (integration) ---------------------------------------------------- def test_the_kernel_drives_the_pipeline_from_a_press() -> None: """End to end: the press starts the work (stage 2b) and stage 4 advances it in the same substep — the 18th `step()` is the `READY` flip.""" state = make_state( stations=(station(StationKind.CHOP_BOARD, contents=(item(ItemKind.TOMATO),)),), players=(cook("p0"),), ) for substep in range(1, ticks(0.9)): state, _events = step( state, STEP_DT, (PlayerInput(player_id="p0", seq=substep, action=True, received_at_tick=state.tick),), ) assert state.stations[0].state is StationState.WORKING assert state.players[0].action_held_ticks == substep state, _events = step( state, STEP_DT, (PlayerInput(player_id="p0", seq=ticks(0.9), action=True, received_at_tick=state.tick),), ) assert state.stations[0].state is StationState.READY assert state.stations[0].contents[0].form is ItemForm.CHOPPED def test_the_kernel_freezes_a_released_hold_and_resumes_it() -> None: """Through the kernel: a release freezes the progress (the gate zeroes the hold), a resume continues it, and it never rolls back.""" state = make_state( stations=(working_station(StationKind.CHOP_BOARD, (item(ItemKind.TOMATO),), 0.9),), players=(cook("p0"),), ) for substep in range(9): state, _events = step( state, STEP_DT, (PlayerInput(player_id="p0", seq=substep, action=True, received_at_tick=state.tick),), ) assert state.stations[0].state is StationState.WORKING frozen = state.stations[0].progress for substep in range(9, 13): # the button comes up state, _events = step( state, STEP_DT, (PlayerInput(player_id="p0", seq=substep, action=False, received_at_tick=state.tick),), ) assert state.players[0].action_held_ticks == 0 assert state.stations[0].progress == pytest.approx(frozen) seen: list[float] = [] for substep in range(13, 13 + ticks(0.9)): # the button comes back state, _events = step( state, STEP_DT, (PlayerInput(player_id="p0", seq=substep, action=True, received_at_tick=state.tick),), ) seen.append(state.stations[0].progress) assert all(b >= a for a, b in pairwise(seen)) assert state.stations[0].state is StationState.READY # --- helpers --------------------------------------------------------------------------- def _verb(name: str) -> Verb: """The `Verb` member by value (the tests import the enum through `interact`).""" return Verb(name) def book() -> RecipeBook: """A one-recipe book: `soup` from a tomato, worth 40 (the tests' value).""" return RecipeBook.model_validate( { "recipes": { "soup": { "id": "soup", "name": "Soup", "base_score": 40, "ingredients": [{"kind": "tomato", "form": "raw"}], "steps": [{"kind": "serve", "station": "serve"}], } } } ) def replace_player( state: KitchenState, player_id: str, **changes: Any ) -> KitchenState: """`state` with one cook swapped for a copy with `changes` applied (pos/hold).""" if "held_ticks" in changes: changes["action_held_ticks"] = changes.pop("held_ticks") return _replace( state, players=tuple( _replace(player, **changes) if player.id == player_id else player for player in state.players ), )