diff --git a/.github/workflows/docker-build.yml b/.github/workflows/docker-build.yml index 241f39485..01ca48e9e 100644 --- a/.github/workflows/docker-build.yml +++ b/.github/workflows/docker-build.yml @@ -86,7 +86,9 @@ jobs: context: envs/git_env - name: connect4_env dockerfile: envs/connect4_env/server/Dockerfile - context: envs/connect4_env + - name: minesweeper-env + dockerfile: envs/minesweeper_env/server/Dockerfile + context: envs/minesweeper_env - name: chess-env dockerfile: envs/chess_env/server/Dockerfile context: envs/chess_env diff --git a/envs/minesweeper_env/README.md b/envs/minesweeper_env/README.md new file mode 100644 index 000000000..7b7a6749d --- /dev/null +++ b/envs/minesweeper_env/README.md @@ -0,0 +1,143 @@ +# Minesweeper Environment + +A Minesweeper game environment for reinforcement learning agents. The environment consists of a grid with hidden mines where the agent must reveal all non-mine cells without triggering any mines. + +## Overview + +The agent can perform two types of actions: +- Reveal cells to uncover numbers indicating adjacent mines +- Place or remove flags on suspected mine locations + +The game ends when all non-mine cells are revealed (win) or a mine is revealed (loss). + +## Quick Start + +```python +from envs.minesweeper_env import MinesweeperAction, MinesweeperEnv + +# Create environment from Docker image +minesweeper_env = MinesweeperEnv.from_docker_image("minesweeper-env:latest") + +try: + # Reset the environment + result = minesweeper_env.reset() + print(f"Board size: {result.observation.board_height}x{result.observation.board_width}") + print(f"Number of mines: {result.observation.num_mines}") + + # Reveal a cell + result = minesweeper_env.step(MinesweeperAction(row=2, col=2, action_type="reveal")) + print(f"Cells revealed: {result.observation.cells_revealed}") + print(f"Reward: {result.observation.reward}") + + # Place a flag + result = minesweeper_env.step(MinesweeperAction(row=1, col=1, action_type="flag")) + print(f"Flags placed: {result.observation.flags_placed}") + +finally: + minesweeper_env.close() +``` + +## Building the Docker Image + +Build the Docker image from the project root: + +```bash +docker build -t minesweeper-env:latest -f envs/minesweeper_env/server/Dockerfile envs/minesweeper_env +``` + +Or use the build script: + +```bash +cd envs/minesweeper_env/server +./build_docker.sh latest +``` + +## Environment Details + +### Action + +**MinesweeperAction**: Specifies the cell and action type +- `row` (int) - Row index (0-indexed) +- `col` (int) - Column index (0-indexed) +- `action_type` (str) - Either "reveal" or "flag" + +### Observation + +**MinesweeperObservation**: Current board state and game information +- `board` (list[list]) - 2D grid showing the current state of each cell: + - `-1`: Unrevealed cell + - `0-8`: Number of adjacent mines (revealed cell) + - `'F'`: Flagged cell + - `'*'`: Mine (only shown when game is lost) +- `num_mines` (int) - Total number of mines on the board +- `flags_placed` (int) - Number of flags currently placed +- `cells_revealed` (int) - Number of cells that have been revealed +- `game_status` (GameStatus) - Current game status (ONGOING, WON, or LOST) +- `done` (bool) - Whether the game has ended +- `reward` (float) - Reward from the last action +- `metadata` (dict) - Additional information + +### Rewards + +- Revealing a safe cell: +1.0 +- Placing a flag on a mine: +0.5 +- Revealing a mine (game over): -10.0 +- Revealing an already revealed cell: -0.05 +- Invalid action: -0.1 + +### Game Status + +- `GameStatus.ONGOING`: Game is still in progress +- `GameStatus.WON`: All non-mine cells have been revealed +- `GameStatus.LOST`: A mine was revealed + +## Configuration + +The default configuration is: +- Board height: 5 +- Board width: 5 +- Number of mines: 5 + +These can be configured when initializing the environment server. + +## Connecting to an Existing Server + +If you have a server already running: + +```python +from envs.minesweeper_env import MinesweeperEnv + +# Connect to existing server +minesweeper_env = MinesweeperEnv(base_url="http://localhost:8000") + +# Use as normal +result = minesweeper_env.reset() +``` + +Note: When connecting to an existing server, `close()` will not stop the server. + +## Running Tests + +Run the test suite: + +```bash +python tests/envs/test_minesweeper_env.py +``` + +## Project Structure + +``` +minesweeper_env/ +├── __init__.py # Module exports +├── README.md # This file +├── client.py # MinesweeperEnv client implementation +├── models.py # Action, Observation, and State models +├── openenv.yaml # Environment configuration +├── pyproject.toml # Package dependencies +└── server/ + ├── __init__.py # Server module exports + ├── minesweeper_environment.py # Core game logic + ├── app.py # FastAPI application + ├── Dockerfile # Container image definition + └── build_docker.sh # Build script +``` diff --git a/envs/minesweeper_env/__init__.py b/envs/minesweeper_env/__init__.py new file mode 100644 index 000000000..73633caf2 --- /dev/null +++ b/envs/minesweeper_env/__init__.py @@ -0,0 +1,17 @@ +# Copyright (c) Meta Platforms, Inc. and affiliates. +# All rights reserved. +# +# This source code is licensed under the BSD-style license found in the +# LICENSE file in the root directory of this source tree. + +"""Minesweeper Environment - a grid-based puzzle game for OpenEnv.""" + +from .client import MinesweeperEnv +from .models import GameStatus, MinesweeperAction, MinesweeperObservation + +__all__ = [ + "GameStatus", + "MinesweeperAction", + "MinesweeperEnv", + "MinesweeperObservation", +] diff --git a/envs/minesweeper_env/client.py b/envs/minesweeper_env/client.py new file mode 100644 index 000000000..3bc7d0fb8 --- /dev/null +++ b/envs/minesweeper_env/client.py @@ -0,0 +1,115 @@ +# Copyright (c) Meta Platforms, Inc. and affiliates. +# All rights reserved. +# +# This source code is licensed under the BSD-style license found in the +# LICENSE file in the root directory of this source tree. + +""" +Minesweeper Environment Client. + +This module provides the client for connecting to a Minesweeper Environment server +via WebSocket for persistent sessions. +""" + +from typing import Dict + +from openenv.core.client_types import StepResult +from openenv.core.env_client import EnvClient +from openenv.core.env_server.types import State + +try: + # In-repo imports (when running from OpenEnv repository) + from .models import MinesweeperAction, MinesweeperObservation +except ImportError: + # Standalone imports (when this module is imported as a top-level package) + from models import MinesweeperAction, MinesweeperObservation + + +class MinesweeperEnv(EnvClient[MinesweeperAction, MinesweeperObservation, State]): + """ + Client for the Minesweeper Environment. + + This client maintains a persistent WebSocket connection to the environment + server, enabling efficient multi-step interactions with lower latency. + Each client instance has its own dedicated environment session on the server. + + Example: + >>> # Connect to a running server + >>> with MinesweeperEnv(base_url="http://localhost:8000") as client: + ... result = client.reset() + ... print(result.observation.board) + ... print(result.observation.game_status) + ... + ... # Reveal a cell + ... result = client.step(MinesweeperAction(row=0, col=0, action_type="reveal")) + ... print(result.observation.board) + ... print(result.reward) + + Example with Docker: + >>> # Automatically start container and connect + >>> client = MinesweeperEnv.from_docker_image("minesweeper-env:latest") + >>> try: + ... result = client.reset() + ... result = client.step(MinesweeperAction(row=2, col=3, action_type="reveal")) + ... finally: + ... client.close() + """ + + def _step_payload(self, action: MinesweeperAction) -> Dict: + """ + Convert MinesweeperAction to JSON payload for step request. + + Args: + action: MinesweeperAction instance + + Returns: + Dictionary representation suitable for JSON encoding + """ + return { + "row": action.row, + "col": action.col, + "action_type": action.action_type, + } + + def _parse_result(self, payload: Dict) -> StepResult[MinesweeperObservation]: + """ + Parse server response into StepResult[MinesweeperObservation]. + + Args: + payload: JSON response from server + + Returns: + StepResult with MinesweeperObservation + """ + obs_data = payload.get("observation", {}) + observation = MinesweeperObservation( + board=obs_data.get("board", []), + num_mines=obs_data.get("num_mines", 0), + flags_placed=obs_data.get("flags_placed", 0), + cells_revealed=obs_data.get("cells_revealed", 0), + game_status=obs_data.get("game_status", "ongoing"), + done=payload.get("done", False), + reward=payload.get("reward"), + metadata=obs_data.get("metadata", {}), + ) + + return StepResult( + observation=observation, + reward=payload.get("reward"), + done=payload.get("done", False), + ) + + def _parse_state(self, payload: Dict) -> State: + """ + Parse server response into State object. + + Args: + payload: JSON response from /state endpoint + + Returns: + State object with episode_id and step_count + """ + return State( + episode_id=payload.get("episode_id"), + step_count=payload.get("step_count", 0), + ) diff --git a/envs/minesweeper_env/models.py b/envs/minesweeper_env/models.py new file mode 100644 index 000000000..b22e1d0f0 --- /dev/null +++ b/envs/minesweeper_env/models.py @@ -0,0 +1,144 @@ +# Copyright (c) Meta Platforms, Inc. and affiliates. +# All rights reserved. +# +# This source code is licensed under the BSD-style license found in the +# LICENSE file in the root directory of this source tree. + +""" +Data models for the Minesweeper Environment. + +The minesweeper_env environment is a Minesweeper game where agents reveal cells and place flags +to identify mines on a grid board. +""" + +from enum import Enum +from typing import Any, List, Set, Tuple + +from openenv.core.env_server.types import Action, Observation +from pydantic import BaseModel, Field + + +class GameStatus(Enum): + """Status of the Minesweeper game.""" + + ONGOING = "ongoing" + WON = "won" + LOST = "lost" + + +class MinesweeperAction(Action): + """ + Action for the Minesweeper environment. + + Attributes: + row: Row index of the cell to act on (0-indexed). + col: Column index of the cell to act on (0-indexed). + action_type: Type of action - 'reveal' to uncover a cell, 'flag' to place/remove a flag. + """ + + row: int = Field(..., ge=0, description="Row index of the cell") + col: int = Field(..., ge=0, description="Column index of the cell") + action_type: str = Field( + ..., pattern="^(reveal|flag)$", description="Type of action: 'reveal' or 'flag'" + ) + + +class MinesweeperObservation(Observation): + """ + Observation from the Minesweeper environment. + + This represents what the agent can see - a partial view of the board with hidden mine locations (unless revealed). + + Attributes: + board: 2D list representing the current state of the board. Each cell can be: + - -1: unrevealed + - 0-8: number of adjacent mines (if revealed) + - 'F': flagged cell + - '*': mine (only revealed if game is lost) + num_mines: Total number of mines on the board. + flags_placed: Number of flags currently placed by the agent. + cells_revealed: Number of cells that have been revealed so far. + game_status: Current status of the game - ongoing, won, or lost. + """ + + board: List[List[Any]] = Field(default_factory=list, description="2D board state") + num_mines: int = Field(..., ge=0, description="Total number of mines") + flags_placed: int = Field(..., ge=0, description="Number of flags placed") + cells_revealed: int = Field(..., ge=0, description="Number of cells revealed") + game_status: GameStatus = Field(..., description="Current game status") + + @property + def board_height(self) -> int: + """Height of the board (number of rows).""" + return len(self.board) + + @property + def board_width(self) -> int: + """Width of the board (number of columns).""" + return len(self.board[0]) if self.board else 0 + + +class MinesweeperState(BaseModel): + """ + Internal state of the Minesweeper environment. + + This represents the full internal state of the environment, including hidden information. + + Attributes: + episode_id: Unique identifier for the current episode. + step_count: Number of steps taken in the current episode. + board_height: Height of the board (number of rows). + board_width: Width of the board (number of columns). + mine_locations: Set of (row, col) tuples indicating where mines are located. + revealed_cells: Set of (row, col) tuples indicating which cells have been revealed. + flags: Set of (row, col) tuples indicating where flags have been placed. + mine_counts: 2D list with counts of adjacent mines for each cell. + game_status: Current status of the game - ongoing, won, or lost. + """ + + episode_id: str + step_count: int + board_height: int + board_width: int + mine_locations: Set[Tuple[int, int]] + revealed_cells: Set[Tuple[int, int]] + flags: Set[Tuple[int, int]] + mine_counts: List[List[int]] + game_status: GameStatus + + def to_observation(self) -> MinesweeperObservation: + """ + Convert the full state to a partial observation for the agent. + + Returns: + MinesweeperObservation representing the agent's view of the board. + """ + board = [] + for r in range(self.board_height): + row = [] + for c in range(self.board_width): + if (r, c) in self.revealed_cells: + if (r, c) in self.mine_locations: + cell_value = "*" # Revealed mine + else: + cell_value = self.mine_counts[r][c] # Number of adjacent mines + elif (r, c) in self.flags: + cell_value = "F" # Flagged cell + else: + cell_value = -1 # Unrevealed cell + row.append(cell_value) + board.append(row) + + return MinesweeperObservation( + board=board, + num_mines=len(self.mine_locations), + flags_placed=len(self.flags), + cells_revealed=len(self.revealed_cells), + game_status=self.game_status, + done=self.game_status != GameStatus.ONGOING, + reward=0.0, + metadata={ + "episode_id": self.episode_id, + "step_count": self.step_count, + }, + ) diff --git a/envs/minesweeper_env/openenv.yaml b/envs/minesweeper_env/openenv.yaml new file mode 100644 index 000000000..1ca9ce903 --- /dev/null +++ b/envs/minesweeper_env/openenv.yaml @@ -0,0 +1,6 @@ +spec_version: 1 +name: minesweeper +type: space +runtime: fastapi +app: server.app:app +port: 8000 diff --git a/envs/minesweeper_env/pyproject.toml b/envs/minesweeper_env/pyproject.toml new file mode 100644 index 000000000..aeaafb585 --- /dev/null +++ b/envs/minesweeper_env/pyproject.toml @@ -0,0 +1,36 @@ +# Copyright (c) Meta Platforms, Inc. and affiliates. +# All rights reserved. +# +# This source code is licensed under the BSD-style license found in the +# LICENSE file in the root directory of this source tree. + +[build-system] +requires = ["setuptools>=45", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "openenv-minesweeper-env" +version = "0.1.0" +description = "Minesweeper Environment for OpenEnv - grid-based puzzle game" +requires-python = ">=3.10" +dependencies = [ + "openenv-core[core]>=0.2.2", + "fastapi>=0.115.0", + "pydantic>=2.0.0", + "uvicorn>=0.24.0", + "requests>=2.31.0", +] + +[project.optional-dependencies] +dev = [ + "pytest>=9.0.3", + "pytest-cov>=4.0.0", +] + +[project.scripts] +server = "minesweeper_env.server.app:main" + +[tool.setuptools] +include-package-data = true +packages = ["minesweeper_env", "minesweeper_env.server"] +package-dir = { "minesweeper_env" = ".", "minesweeper_env.server" = "server" } diff --git a/envs/minesweeper_env/server/Dockerfile b/envs/minesweeper_env/server/Dockerfile new file mode 100644 index 000000000..b744245b2 --- /dev/null +++ b/envs/minesweeper_env/server/Dockerfile @@ -0,0 +1,54 @@ +# Copyright (c) Meta Platforms, Inc. and affiliates. +# All rights reserved. +# +# This source code is licensed under the BSD-style license found in the +# LICENSE file in the root directory of this source tree. + +ARG BASE_IMAGE=ghcr.io/meta-pytorch/openenv-base:latest +FROM ${BASE_IMAGE} AS builder + +WORKDIR /app + +COPY . /app/env + +WORKDIR /app/env + +RUN if ! command -v uv >/dev/null 2>&1; then \ + curl -LsSf https://astral.sh/uv/install.sh | sh && \ + mv /root/.local/bin/uv /usr/local/bin/uv && \ + mv /root/.local/bin/uvx /usr/local/bin/uvx; \ + fi + +RUN apt-get update && apt-get install -y --no-install-recommends \ + git \ + && rm -rf /var/lib/apt/lists/* + +RUN --mount=type=cache,target=/root/.cache/uv \ + if [ -f uv.lock ]; then \ + uv sync --frozen --no-install-project --no-editable; \ + else \ + uv sync --no-install-project --no-editable; \ + fi + +RUN --mount=type=cache,target=/root/.cache/uv \ + if [ -f uv.lock ]; then \ + uv sync --frozen --no-editable; \ + else \ + uv sync --no-editable; \ + fi + +# Final runtime stage +FROM ${BASE_IMAGE} + +WORKDIR /app + +COPY --from=builder /app/env/.venv /app/.venv +COPY --from=builder /app/env /app/env + +ENV PATH="/app/.venv/bin:$PATH" +ENV PYTHONPATH="/app/env:$PYTHONPATH" + +HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \ + CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')" || exit 1 + +CMD ["sh", "-c", "cd /app/env && uvicorn server.app:app --host 0.0.0.0 --port 8000"] diff --git a/envs/minesweeper_env/server/__init__.py b/envs/minesweeper_env/server/__init__.py new file mode 100644 index 000000000..2fde76979 --- /dev/null +++ b/envs/minesweeper_env/server/__init__.py @@ -0,0 +1,11 @@ +# Copyright (c) Meta Platforms, Inc. and affiliates. +# All rights reserved. +# +# This source code is licensed under the BSD-style license found in the +# LICENSE file in the root directory of this source tree. + +"""Minesweeper environment server components.""" + +from .minesweeper_environment import MinesweeperEnvironment + +__all__ = ["MinesweeperEnvironment"] diff --git a/envs/minesweeper_env/server/app.py b/envs/minesweeper_env/server/app.py new file mode 100644 index 000000000..939724dde --- /dev/null +++ b/envs/minesweeper_env/server/app.py @@ -0,0 +1,26 @@ +# Copyright (c) Meta Platforms, Inc. and affiliates. +# All rights reserved. +# +# This source code is licensed under the BSD-style license found in the +# LICENSE file in the root directory of this source tree. + +"""FastAPI application for the Minesweeper Environment.""" + +from openenv.core.env_server import create_app + +from ..models import MinesweeperAction, MinesweeperObservation +from .minesweeper_environment import MinesweeperEnvironment + +# Create the FastAPI app +# Pass the class (factory) instead of an instance for WebSocket session support +app = create_app( + MinesweeperEnvironment, + MinesweeperAction, + MinesweeperObservation, + env_name="minesweeper_env", +) + +if __name__ == "__main__": + import uvicorn + + uvicorn.run(app, host="0.0.0.0", port=8000) diff --git a/envs/minesweeper_env/server/build_docker.sh b/envs/minesweeper_env/server/build_docker.sh new file mode 100755 index 000000000..4397c7f1b --- /dev/null +++ b/envs/minesweeper_env/server/build_docker.sh @@ -0,0 +1,48 @@ +#!/bin/bash +# Copyright (c) Meta Platforms, Inc. and affiliates. +# All rights reserved. +# +# This source code is licensed under the BSD-style license found in the +# LICENSE file in the root directory of this source tree. + +# Script to build the Minesweeper environment Docker image +# Usage: ./build_docker.sh [tag] + +set -e + +TAG="${1:-latest}" +IMAGE_NAME="minesweeper-env:${TAG}" + +echo "🐳 Building Minesweeper Environment Docker Image" +echo "================================================" +echo "Image: $IMAGE_NAME" +echo "" + +# Get script directory +SCRIPT_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )" + +# Navigate to OpenEnv root (4 levels up from server/) +OPENENV_ROOT="$(cd "$SCRIPT_DIR/../../../.." && pwd)" + +echo "📁 OpenEnv root: $OPENENV_ROOT" +echo "" + +# Build Minesweeper environment image +echo "⏳ Building..." +docker build \ + -f "$SCRIPT_DIR/Dockerfile" \ + -t "$IMAGE_NAME" \ + "$OPENENV_ROOT" + +if [ $? -eq 0 ]; then + echo "" + echo "✅ Build successful!" + echo "" + echo "🚀 Run with:" + echo " docker run -p 8000:8000 $IMAGE_NAME" + echo "" +else + echo "" + echo "❌ Build failed!" + exit 1 +fi diff --git a/envs/minesweeper_env/server/minesweeper_environment.py b/envs/minesweeper_env/server/minesweeper_environment.py new file mode 100644 index 000000000..8677c00e7 --- /dev/null +++ b/envs/minesweeper_env/server/minesweeper_environment.py @@ -0,0 +1,356 @@ +# Copyright (c) Meta Platforms, Inc. and affiliates. +# All rights reserved. +# +# This source code is licensed under the BSD-style license found in the +# LICENSE file in the root directory of this source tree. + +""" +Minesweeper Environment Implementation. + +A Minesweeper game environment where agents must reveal cells and place flags +to identify mines on a grid board without triggering any mines. +""" + +import random +from typing import Any, Dict, List, Optional, Set, Tuple +from uuid import uuid4 + +from openenv.core.env_server.interfaces import Environment +from openenv.core.env_server.types import State + +from ..models import ( + GameStatus, + MinesweeperAction, + MinesweeperObservation, + MinesweeperState, +) + + +class MinesweeperEnvironment( + Environment[MinesweeperAction, MinesweeperObservation, State] +): + """ + Minesweeper game environment implementation for Reinforcement Learning. + The environment consists of a grid with hidden mines. The agent can reveal cells or place flags. + The goal is to reveal all non-mine cells without triggering a mine. + The agent must: + - Reveal cells to uncover numbers indicating adjacent mines. + - Place flags on suspected mine locations. + The game ends when all non-mine cells are revealed (win) or a mine is revealed (loss). + + Observation encoding: + -1: unrevealed + 0-8: number of adjacent mines (if revealed) + 'F': flagged cell + '*': mine (only revealed if game is lost) + + Example: + >>> env = MinesweeperEnvironment(height=5, width=5, num_mines=5) + >>> obs = env.reset() + >>> action = MinesweeperAction(row=2, col=3, action_type='reveal') + """ + + def __init__(self, height: int = 5, width: int = 5, num_mines: int = 5): + """Initialize the minesweeper_env environment. + Args: + height: Height of the minesweeper board. + width: Width of the minesweeper board. + num_mines: Number of mines to place on the board. + """ + super().__init__() + self.height = height + self.width = width + self.num_mines = num_mines + + self._state = State(episode_id=str(uuid4()), step_count=0) + self._reset_count = 0 + + # Internal game state + self._mine_positions: Set[Tuple[int, int]] = set() + self._revealed_cells: Set[Tuple[int, int]] = set() + self._flags_placed: Set[Tuple[int, int]] = set() + self._mine_counts: List[List[int]] = [ + [0 for _ in range(width)] for _ in range(height) + ] + self._game_status = GameStatus.ONGOING + + def reset(self) -> MinesweeperObservation: + """ + Reset the environment and starts a new game. + + Returns: + MinesweeperObservation with initial board state + """ + self._state = State(episode_id=str(uuid4()), step_count=0) + self._reset_count += 1 + + # Reset internal game state + self._revealed_cells.clear() + self._flags_placed.clear() + self._game_status = GameStatus.ONGOING + + # Place mines randomly + self._place_mines() + + # Compute mine counts for each cell + self._compute_mine_counts() + + return self._create_observation( + done=False, + reward=0.0, + ) + + def step(self, action: MinesweeperAction) -> MinesweeperObservation: # type: ignore[override] + """ + Execute a step in the environment by performing the given action. + + Args: + action: MinesweeperAction specifying row, col and action_type + + Returns: + MinesweeperObservation with updated board state and reward + """ + self._state.step_count += 1 + + row, col = action.row, action.col + + # Validate action + if not self._is_valid_position(row, col): + # Invalid action or game already over + return self._create_observation( + done=self._game_status != GameStatus.ONGOING, + reward=-0.1, + metadata={"error": "Invalid action"}, + ) + + # If game already over, no further actions allowed + if self._game_status != GameStatus.ONGOING: + return self._create_observation( + done=True, + reward=0.0, + metadata={"info": "Game already over"}, + ) + + reward = 0.0 + error: Optional[str] = None + + if action.action_type == "reveal": + reward, error = self._reveal_cell(row, col) + elif action.action_type == "flag": + reward, error = self._toggle_flag(row, col) + else: + reward = -0.1 + error = f"Unknown action_type: {action.action_type!r}" + + self._check_win_condition() + + metadata = {"error": error} if error else None + return self._create_observation( + done=self._game_status != GameStatus.ONGOING, + reward=reward, + metadata=metadata, + ) + + def _place_mines(self) -> None: + """Randomly place mines on the board.""" + self._mine_positions.clear() + while len(self._mine_positions) < self.num_mines: + r = random.randint(0, self.height - 1) + c = random.randint(0, self.width - 1) + self._mine_positions.add((r, c)) + + def _compute_mine_counts(self) -> None: + """Compute the number of adjacent mines for each cell.""" + self._mine_counts = [[0 for _ in range(self.width)] for _ in range(self.height)] + for row in range(self.height): + for col in range(self.width): + if (row, col) not in self._mine_positions: + count = self._count_adjacent_mines(row, col) + self._mine_counts[row][col] = count + + def _count_adjacent_mines(self, row: int, col: int) -> int: + """Count the number of mines adjacent to the given cell.""" + count = 0 + for dr in [-1, 0, 1]: + for dc in [-1, 0, 1]: + if dr == 0 and dc == 0: + continue + r, c = row + dr, col + dc + if self._is_valid_position(r, c) and (r, c) in self._mine_positions: + count += 1 + return count + + def _reveal_cell(self, row: int, col: int) -> Tuple[float, Optional[str]]: + """Reveal the cell at (row, col). + + Returns: + (reward, error). ``error`` is a human-readable reason the action was + a no-op (e.g. cell already revealed / flagged) or ``None``. + """ + if (row, col) in self._revealed_cells: + return -0.05, f"Cell ({row}, {col}) is already revealed" + if (row, col) in self._flags_placed: + return -0.05, f"Cell ({row}, {col}) is flagged; remove the flag first" + + if (row, col) in self._mine_positions: + self._game_status = GameStatus.LOST + self._revealed_cells.add((row, col)) + return -10.0, None + + self._reveal_recursive(row, col) + return 1.0, None + + def _reveal_recursive(self, row: int, col: int) -> None: + """Recursively reveal cells with 0 adjacent mines.""" + if not self._is_valid_position(row, col): + return + if (row, col) in self._revealed_cells or (row, col) in self._flags_placed: + return + + if (row, col) in self._mine_positions: + return + + self._revealed_cells.add((row, col)) + + if self._mine_counts[row][col] == 0: + for dr in [-1, 0, 1]: + for dc in [-1, 0, 1]: + if dr == 0 and dc == 0: + continue + self._reveal_recursive(row + dr, col + dc) + + def _toggle_flag(self, row: int, col: int) -> Tuple[float, Optional[str]]: + """Toggle a flag on the cell at (row, col). + + Returns: + (reward, error). ``error`` is set when the action is rejected + (e.g. trying to flag a revealed cell). + """ + if (row, col) in self._revealed_cells: + return -0.05, f"Cannot flag revealed cell ({row}, {col})" + + if (row, col) in self._flags_placed: + self._flags_placed.remove((row, col)) + return 0.0, None + + self._flags_placed.add((row, col)) + if (row, col) in self._mine_positions: + return 0.5, None + return 0.0, None + + def _check_win_condition(self) -> None: + """Check if the game has been won. + + Only counts non-mine cells: hitting a mine adds it to ``_revealed_cells`` + but must never satisfy the win threshold (which would otherwise overwrite + a freshly-set LOST status on a near-complete board). + """ + if self._game_status != GameStatus.ONGOING: + return + safe_revealed = len(self._revealed_cells - self._mine_positions) + if safe_revealed == self.height * self.width - self.num_mines: + self._game_status = GameStatus.WON + + def _is_valid_position(self, row: int, col: int) -> bool: + """Check if the given (row, col) is within board bounds.""" + return 0 <= row < self.height and 0 <= col < self.width + + def _create_observation( + self, + done: bool, + reward: Optional[float] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> MinesweeperObservation: + """Create the current observation of the board. + Args: + done: Whether the episode is done. + reward: Reward obtained from the last action. + metadata: Additional metadata to include. + Returns: + MinesweeperObservation representing the current board state. + """ + board = [] + for r in range(self.height): + row = [] + for c in range(self.width): + if (r, c) in self._revealed_cells: + if (r, c) in self._mine_positions: + row.append("*") + else: + row.append(self._mine_counts[r][c]) + elif (r, c) in self._flags_placed: + row.append("F") + else: + row.append(-1) + board.append(row) + + return MinesweeperObservation( + board=board, + num_mines=self.num_mines, + flags_placed=len(self._flags_placed), + cells_revealed=len(self._revealed_cells), + game_status=self._game_status, + done=done, + reward=reward, + metadata=metadata or {}, + ) + + @property + def state(self) -> State: + """ + Get the current environment state. + + Returns: + Current State with episode_id and step_count + """ + return self._state + + def _get_full_state(self) -> MinesweeperState: + """ + Get the full internal state of the environment (server-side only). + + Exposes the mine layout and is intended for testing / introspection. + Never call this from a code path reachable by an agent. + + Returns: + MinesweeperState representing the full internal state + """ + return MinesweeperState( + episode_id=self._state.episode_id or "", + step_count=self._state.step_count, + board_height=self.height, + board_width=self.width, + mine_locations=self._mine_positions, + revealed_cells=self._revealed_cells, + flags=self._flags_placed, + mine_counts=self._mine_counts, + game_status=self._game_status, + ) + + def get_legal_actions(self) -> List[MinesweeperAction]: + """ + Get the list of legal actions available in the current state. + + Returns: + List of MinesweeperAction instances representing legal actions + """ + legal_actions = [] + + # If game is over, no legal actions + if self._game_status != GameStatus.ONGOING: + return legal_actions + + for r in range(self.height): + for c in range(self.width): + if (r, c) not in self._revealed_cells and ( + r, + c, + ) not in self._flags_placed: + legal_actions.append( + MinesweeperAction(row=r, col=c, action_type="reveal") + ) + if (r, c) not in self._revealed_cells: + legal_actions.append( + MinesweeperAction(row=r, col=c, action_type="flag") + ) + return legal_actions diff --git a/examples/minesweeper_agent.py b/examples/minesweeper_agent.py new file mode 100644 index 000000000..3a74a31bf --- /dev/null +++ b/examples/minesweeper_agent.py @@ -0,0 +1,299 @@ +"""Minesweeper Agent Loop Example. + +This script demonstrates how to create an agent that interacts with the Minesweeper +environment. The agent uses a simple strategy to play the game. + +Prerequisites: +- Minesweeper Docker container must be running +- Default URL: http://localhost:8000 + +Usage: + python examples/minesweeper_agent.py +""" + +import random +import sys +from pathlib import Path +from typing import List, Set, Tuple + +# Add src to path for imports +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +from envs.minesweeper_env import MinesweeperEnv, MinesweeperAction, GameStatus + +# Configuration +MAX_EPISODES = 5 +MAX_STEPS_PER_EPISODE = 100 +BASE_URL = "http://localhost:8000" + + +class SimpleMinesweeperAgent: + """ + A simple agent that plays Minesweeper using a basic strategy: + 1. Start by revealing a corner cell (safer statistically) + 2. Reveal cells with 0 adjacent mines first + 3. Flag cells that are definitely mines based on revealed numbers + 4. Make random safe moves when no obvious safe cells exist + """ + + def __init__(self): + self.revealed_positions: Set[Tuple[int, int]] = set() + self.flagged_positions: Set[Tuple[int, int]] = set() + self.board_height = 0 + self.board_width = 0 + + def reset(self): + """Reset agent state for a new episode.""" + self.revealed_positions.clear() + self.flagged_positions.clear() + self.board_height = 0 + self.board_width = 0 + + def choose_action(self, observation) -> MinesweeperAction: + """ + Choose the next action based on the current board state. + + Args: + observation: MinesweeperObservation from the environment + + Returns: + MinesweeperAction to take + """ + board = observation.board + self.board_height = len(board) + self.board_width = len(board[0]) if board else 0 + + # Update what we know + self._update_state(board) + + # Strategy 1: If this is the first move, reveal a corner (statistically safer) + if len(self.revealed_positions) == 0: + return MinesweeperAction(row=0, col=0, action_type="reveal") + + # Strategy 2: Look for cells that are definitely safe + safe_cells = self._find_safe_cells(board) + if safe_cells: + row, col = random.choice(list(safe_cells)) + return MinesweeperAction(row=row, col=col, action_type="reveal") + + # Strategy 3: Look for cells that are definitely mines and flag them + mine_cells = self._find_definite_mines(board) + if mine_cells: + row, col = random.choice(list(mine_cells)) + return MinesweeperAction(row=row, col=col, action_type="flag") + + # Strategy 4: Make a random move on unrevealed cells (risky but necessary) + unrevealed = self._get_unrevealed_unflagged_cells(board) + if unrevealed: + row, col = random.choice(list(unrevealed)) + return MinesweeperAction(row=row, col=col, action_type="reveal") + + # Fallback: no-op (shouldn't reach here) + return MinesweeperAction(row=0, col=0, action_type="reveal") + + def _update_state(self, board): + """Update internal state based on current board.""" + for r in range(len(board)): + for c in range(len(board[0])): + cell = board[r][c] + if cell != -1 and cell != 'F': + self.revealed_positions.add((r, c)) + if cell == 'F': + self.flagged_positions.add((r, c)) + + def _get_unrevealed_unflagged_cells(self, board) -> Set[Tuple[int, int]]: + """Get all cells that are unrevealed and not flagged.""" + unrevealed = set() + for r in range(len(board)): + for c in range(len(board[0])): + if board[r][c] == -1: + unrevealed.add((r, c)) + return unrevealed + + def _find_safe_cells(self, board) -> Set[Tuple[int, int]]: + """ + Find cells that are definitely safe based on revealed numbers. + + A cell is safe if it's adjacent to a revealed cell whose mine count + equals the number of flags around it. + """ + safe_cells = set() + + for r in range(len(board)): + for c in range(len(board[0])): + cell = board[r][c] + + # Only check revealed numbered cells + if isinstance(cell, int) and 0 <= cell <= 8: + neighbors = self._get_neighbors(r, c) + unrevealed_neighbors = [ + (nr, nc) for nr, nc in neighbors + if board[nr][nc] == -1 + ] + flagged_neighbors = [ + (nr, nc) for nr, nc in neighbors + if board[nr][nc] == 'F' + ] + + # If all mines are flagged, remaining unrevealed cells are safe + if len(flagged_neighbors) == cell: + safe_cells.update(unrevealed_neighbors) + + return safe_cells + + def _find_definite_mines(self, board) -> Set[Tuple[int, int]]: + """ + Find cells that are definitely mines based on revealed numbers. + + A cell is definitely a mine if it's adjacent to a revealed cell whose + mine count equals the number of unrevealed + flagged neighbors. + """ + mine_cells = set() + + for r in range(len(board)): + for c in range(len(board[0])): + cell = board[r][c] + + # Only check revealed numbered cells with value > 0 + if isinstance(cell, int) and 1 <= cell <= 8: + neighbors = self._get_neighbors(r, c) + unrevealed_neighbors = [ + (nr, nc) for nr, nc in neighbors + if board[nr][nc] == -1 + ] + flagged_neighbors = [ + (nr, nc) for nr, nc in neighbors + if board[nr][nc] == 'F' + ] + + # If unrevealed + flagged equals mine count, + # all unrevealed are mines + if len(unrevealed_neighbors) + len(flagged_neighbors) == cell: + mine_cells.update(unrevealed_neighbors) + + return mine_cells + + def _get_neighbors(self, row: int, col: int) -> List[Tuple[int, int]]: + """Get all valid neighboring cells.""" + neighbors = [] + for dr in [-1, 0, 1]: + for dc in [-1, 0, 1]: + if dr == 0 and dc == 0: + continue + nr, nc = row + dr, col + dc + if 0 <= nr < self.board_height and 0 <= nc < self.board_width: + neighbors.append((nr, nc)) + return neighbors + + +def print_board(board): + """Pretty print the Minesweeper board.""" + print("\n ", end="") + if board: + for c in range(len(board[0])): + print(f"{c:3}", end="") + print() + + for r, row in enumerate(board): + print(f"{r:2} ", end="") + for cell in row: + if cell == -1: + print(" . ", end="") + elif cell == 'F': + print(" F ", end="") + elif cell == '*': + print(" * ", end="") + else: + print(f" {cell} ", end="") + print() + print() + + +def main(): + """Run the agent loop.""" + print("=" * 60) + print("Minesweeper Agent Loop") + print("=" * 60) + + # Connect to the environment + print(f"Connecting to Minesweeper server at {BASE_URL}...") + + # Option 1: Connect to existing server + env = MinesweeperEnv(base_url=BASE_URL) + + # Option 2: Start from Docker image (uncomment to use) + # env = MinesweeperEnv.from_docker_image("minesweeper_env-env:latest") + + agent = SimpleMinesweeperAgent() + + total_wins = 0 + total_losses = 0 + + try: + for episode in range(1, MAX_EPISODES + 1): + print(f"\n{'=' * 60}") + print(f"Episode {episode}/{MAX_EPISODES}") + print(f"{'=' * 60}") + + agent.reset() + result = env.reset() + observation = result.observation + + print(f"Board size: {observation.board_height}x{observation.board_width}") + print(f"Number of mines: {observation.num_mines}") + print_board(observation.board) + + episode_reward = 0.0 + + for step in range(1, MAX_STEPS_PER_EPISODE + 1): + if result.done: + break + + # Agent chooses action + action = agent.choose_action(observation) + + # Execute action + result = env.step(action) + observation = result.observation + reward = result.reward or 0.0 + episode_reward += reward + + print(f"Step {step}: {action.action_type} ({action.row}, {action.col}) -> reward: {reward:+.2f}") + print_board(observation.board) + + # Handle game_status being either an Enum or an int + if isinstance(observation.game_status, GameStatus): + status_name = observation.game_status.name + elif isinstance(observation.game_status, int): + status_name = GameStatus(observation.game_status).name + else: + status_name = str(observation.game_status) + + print(f"Status: {status_name} | Revealed: {observation.cells_revealed} | Flags: {observation.flags_placed}") + + if result.done: + if status_name == "WON": + print(f"\n🎉 Episode {episode}: WON! Total reward: {episode_reward:.2f}") + total_wins += 1 + elif status_name == "LOST": + print(f"\n💥 Episode {episode}: LOST! Total reward: {episode_reward:.2f}") + total_losses += 1 + break + else: + print(f"\n⏰ Episode {episode}: Reached max steps ({MAX_STEPS_PER_EPISODE})") + + # Summary + print(f"\n{'=' * 60}") + print("Summary") + print(f"{'=' * 60}") + print(f"Episodes played: {MAX_EPISODES}") + print(f"Wins: {total_wins} ({100 * total_wins / MAX_EPISODES:.1f}%)") + print(f"Losses: {total_losses} ({100 * total_losses / MAX_EPISODES:.1f}%)") + + finally: + env.close() + print("\nEnvironment closed.") + + +if __name__ == "__main__": + main() diff --git a/tests/envs/test_minesweeper_env.py b/tests/envs/test_minesweeper_env.py new file mode 100644 index 000000000..76368a057 --- /dev/null +++ b/tests/envs/test_minesweeper_env.py @@ -0,0 +1,384 @@ +# Copyright (c) Meta Platforms, Inc. and affiliates. +# All rights reserved. +# +# This source code is licensed under the BSD-style license found in the +# LICENSE file in the root directory of this source tree. + +"""Test suite for Minesweeper Environment.""" + +import os +import signal +import subprocess +import sys +import time +import unittest + +import requests +from envs.minesweeper_env import ( + GameStatus, + MinesweeperAction, + MinesweeperEnv, + MinesweeperObservation, +) +from envs.minesweeper_env.server.minesweeper_environment import MinesweeperEnvironment + + +class TestMinesweeperEnv(unittest.IsolatedAsyncioTestCase): + """Test cases for the Minesweeper environment.""" + + server_process = None + + @classmethod + def setUpClass(cls): + """Start the server once for all tests.""" + cls.server_process = subprocess.Popen( + [sys.executable, "-m", "envs.minesweeper_env.server.app"], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + time.sleep(3) # Give server time to start + + # Verify server is running + try: + response = requests.get("http://127.0.0.1:8000/health") + if response.status_code != 200: + raise RuntimeError("Server health check failed") + except requests.ConnectionError: + raise RuntimeError("Server did not start or is unreachable") + + @classmethod + def tearDownClass(cls): + """Clean up server after all tests.""" + if cls.server_process: + cls.server_process.terminate() + try: + cls.server_process.wait(timeout=5) + except subprocess.TimeoutExpired: + os.kill(cls.server_process.pid, signal.SIGKILL) + + for stream in [ + cls.server_process.stdin, + cls.server_process.stdout, + cls.server_process.stderr, + ]: + if stream and not stream.closed: + stream.close() + + async def test_minesweeper_env_client(self): + """Test Minesweeper environment client initialization.""" + async with MinesweeperEnv(base_url="http://127.0.0.1:8000") as client: + assert isinstance(client, MinesweeperEnv) + + async def test_minesweeper_initial_state(self): + """Test the initial state after reset.""" + async with MinesweeperEnv(base_url="http://127.0.0.1:8000") as client: + result = await client.reset() + observation = result.observation + + # Check observation type and attributes + assert isinstance(observation, MinesweeperObservation) + assert isinstance(observation.board, list) + assert isinstance(observation.done, bool) + assert isinstance(observation.reward, float) + assert isinstance(observation.num_mines, int) + assert isinstance(observation.flags_placed, int) + assert isinstance(observation.cells_revealed, int) + # game_status may be int or GameStatus enum due to HTTP serialization + assert isinstance(observation.game_status, (GameStatus, int)) + + # Check initial state values + assert observation.done is False + assert observation.reward == 0.0 + assert observation.flags_placed == 0 + assert observation.cells_revealed == 0 + # Compare with enum value (handles both int and enum) + assert ( + observation.game_status == GameStatus.ONGOING.value + or observation.game_status == GameStatus.ONGOING + ) + + # Check board structure (default 5x5) + assert len(observation.board) == 5 # 5 rows + assert all(len(row) == 5 for row in observation.board) # 5 columns + + # Check all cells are initially unrevealed + assert all(cell == -1 for row in observation.board for cell in row), ( + "All cells should be unrevealed (-1) at start" + ) + + async def test_reveal_action(self): + """Test revealing a cell.""" + async with MinesweeperEnv(base_url="http://127.0.0.1:8000") as client: + await client.reset() + + # Try revealing a cell + action = MinesweeperAction(row=0, col=0, action_type="reveal") + result = await client.step(action) + observation = result.observation + + assert isinstance(observation, MinesweeperObservation) + assert observation.cells_revealed > 0, ( + "At least one cell should be revealed" + ) + + async def test_flag_action(self): + """Test placing a flag.""" + async with MinesweeperEnv(base_url="http://127.0.0.1:8000") as client: + await client.reset() + + # Place a flag + action = MinesweeperAction(row=1, col=1, action_type="flag") + result = await client.step(action) + observation = result.observation + + assert isinstance(observation, MinesweeperObservation) + assert observation.flags_placed == 1, "One flag should be placed" + assert observation.board[1][1] == "F", "Cell should show flag marker" + + async def test_toggle_flag(self): + """Test toggling a flag on and off.""" + async with MinesweeperEnv(base_url="http://127.0.0.1:8000") as client: + await client.reset() + + # Place a flag + action = MinesweeperAction(row=2, col=2, action_type="flag") + result = await client.step(action) + observation = result.observation + assert observation.flags_placed == 1 + + # Remove the flag + action = MinesweeperAction(row=2, col=2, action_type="flag") + result = await client.step(action) + observation = result.observation + assert observation.flags_placed == 0, "Flag should be removed" + + async def test_invalid_position(self): + """Test action with invalid position.""" + async with MinesweeperEnv(base_url="http://127.0.0.1:8000") as client: + await client.reset() + + # Try invalid row + action = MinesweeperAction(row=10, col=0, action_type="reveal") + result = await client.step(action) + observation = result.observation + + assert observation.reward < 0, ( + "Should receive negative reward for invalid action" + ) + + # Try invalid column + action = MinesweeperAction(row=0, col=10, action_type="reveal") + result = await client.step(action) + observation = result.observation + + assert observation.reward < 0, ( + "Should receive negative reward for invalid action" + ) + + async def test_reveal_already_revealed(self): + """Test revealing an already revealed cell.""" + async with MinesweeperEnv(base_url="http://127.0.0.1:8000") as client: + await client.reset() + + # Find a safe revealed cell to re-target. Mines end the game, so + # we keep trying until we land on a non-mine revealed cell. + candidates = [(2, 2), (1, 1), (0, 0), (4, 4), (3, 3)] + test_row, test_col = None, None + for r, c in candidates: + action = MinesweeperAction(row=r, col=c, action_type="reveal") + result = await client.step(action) + ongoing = ( + result.observation.game_status == GameStatus.ONGOING.value + or result.observation.game_status == GameStatus.ONGOING + ) + if ongoing and result.observation.board[r][c] != -1: + test_row, test_col = r, c + break + if not ongoing: + await client.reset() + + assert test_row is not None, "Could not reveal a safe cell across resets" + + # Re-revealing the same cell should be penalised. + action = MinesweeperAction(row=test_row, col=test_col, action_type="reveal") + result = await client.step(action) + assert result.observation.reward < 0, ( + f"Should receive penalty for revealing already revealed cell, " + f"got {result.observation.reward}" + ) + + async def test_game_status_ongoing(self): + """Test that game status remains ONGOING during normal play.""" + async with MinesweeperEnv(base_url="http://127.0.0.1:8000") as client: + await client.reset() + + # Make a few safe moves + action = MinesweeperAction(row=0, col=0, action_type="reveal") + result = await client.step(action) + + # Game should still be ongoing if we didn't hit a mine or win + # Handle both int and enum types for game_status + if ( + result.observation.game_status == GameStatus.ONGOING.value + or result.observation.game_status == GameStatus.ONGOING + ): + assert result.observation.done is False + + async def test_board_cell_values(self): + """Test that board cells contain valid values.""" + async with MinesweeperEnv(base_url="http://127.0.0.1:8000") as client: + await client.reset() + + # Reveal a cell + action = MinesweeperAction(row=2, col=2, action_type="reveal") + result = await client.step(action) + observation = result.observation + + # Check that revealed cells have valid values (0-8 or '*') + for row in observation.board: + for cell in row: + assert ( + cell == -1 # Unrevealed + or cell == "F" # Flagged + or cell == "*" # Mine (if revealed) + or ( + isinstance(cell, int) and 0 <= cell <= 8 + ) # Number of adjacent mines + ), f"Invalid cell value: {cell}" + + async def test_metadata_in_observation(self): + """Test that observations contain metadata.""" + async with MinesweeperEnv(base_url="http://127.0.0.1:8000") as client: + result = await client.reset() + observation = result.observation + + assert hasattr(observation, "metadata"), "Observation should have metadata" + assert isinstance(observation.metadata, dict), ( + "Metadata should be a dictionary" + ) + + async def test_multiple_steps(self): + """Test taking multiple steps in the environment.""" + async with MinesweeperEnv(base_url="http://127.0.0.1:8000") as client: + await client.reset() + + # Take several actions + actions = [ + MinesweeperAction(row=0, col=0, action_type="reveal"), + MinesweeperAction(row=0, col=1, action_type="flag"), + MinesweeperAction(row=1, col=0, action_type="reveal"), + ] + + for action in actions: + result = await client.step(action) + assert isinstance(result.observation, MinesweeperObservation) + + async def test_reset_clears_state(self): + """Test that reset properly clears the game state.""" + async with MinesweeperEnv(base_url="http://127.0.0.1:8000") as client: + # First game + await client.reset() + action = MinesweeperAction(row=0, col=0, action_type="flag") + await client.step(action) + + # Reset and check state is cleared + result = await client.reset() + observation = result.observation + + assert observation.flags_placed == 0, "Flags should be cleared after reset" + assert observation.cells_revealed == 0, ( + "Revealed cells should be cleared after reset" + ) + # Compare with enum value (handles both int and enum) + assert ( + observation.game_status == GameStatus.ONGOING.value + or observation.game_status == GameStatus.ONGOING + ), "Game should be ongoing after reset" + + +class TestMinesweeperEnvironmentLogic(unittest.TestCase): + """Direct unit tests against MinesweeperEnvironment (no HTTP server).""" + + def test_hitting_last_mine_does_not_overwrite_lost_with_won(self): + """Revealing a mine must not be reclassified as a win, even if the + revealed-cell count happens to equal total_cells - num_mines.""" + # 2x2 with 2 mines: safe = 2, mines = 2. + # Pre-reveal 1 safe cell. Then reveal a mine — _revealed_cells grows + # to 2, which equals total(4) - mines(2). The buggy check used to + # count the mine and overwrite LOST with WON. + env = MinesweeperEnvironment(height=2, width=2, num_mines=2) + env.reset() + + env._mine_positions = {(0, 1), (1, 1)} + env._compute_mine_counts() + env._revealed_cells = {(0, 0)} + env._flags_placed = set() + env._game_status = GameStatus.ONGOING + + obs = env.step(MinesweeperAction(row=0, col=1, action_type="reveal")) + + assert env._game_status == GameStatus.LOST, ( + f"Game must be LOST after revealing a mine, got {env._game_status}" + ) + assert obs.game_status == GameStatus.LOST + assert obs.done is True + assert obs.reward == -10.0 + + def test_revealing_last_safe_cell_wins(self): + """The normal win path still works after the LOST guard.""" + env = MinesweeperEnvironment(height=2, width=2, num_mines=1) + env.reset() + + env._mine_positions = {(1, 1)} + env._compute_mine_counts() + env._revealed_cells = {(0, 0), (0, 1)} + env._flags_placed = set() + env._game_status = GameStatus.ONGOING + + obs = env.step(MinesweeperAction(row=1, col=0, action_type="reveal")) + + assert env._game_status == GameStatus.WON + assert obs.game_status == GameStatus.WON + assert obs.done is True + + def test_revealing_flagged_cell_returns_error_metadata(self): + """A reject reason for a no-op action is surfaced in metadata.error.""" + env = MinesweeperEnvironment(height=3, width=3, num_mines=1) + env.reset() + + env._mine_positions = {(2, 2)} + env._compute_mine_counts() + env._revealed_cells = set() + env._flags_placed = {(0, 0)} + env._game_status = GameStatus.ONGOING + + obs = env.step(MinesweeperAction(row=0, col=0, action_type="reveal")) + + assert obs.reward == -0.05 + assert obs.metadata.get("error"), ( + "Expected metadata.error explaining the rejection" + ) + assert "flag" in obs.metadata["error"].lower() + + def test_flagging_revealed_cell_returns_error_metadata(self): + """Flagging an already-revealed cell is rejected with an error message.""" + env = MinesweeperEnvironment(height=3, width=3, num_mines=1) + env.reset() + + env._mine_positions = {(2, 2)} + env._compute_mine_counts() + env._revealed_cells = {(0, 0)} + env._flags_placed = set() + env._game_status = GameStatus.ONGOING + + obs = env.step(MinesweeperAction(row=0, col=0, action_type="flag")) + + assert obs.reward == -0.05 + assert obs.metadata.get("error"), ( + "Expected metadata.error explaining the rejection" + ) + + +if __name__ == "__main__": + unittest.main()