Upload folder using huggingface_hub

Dockerfile

@@ -0,0 +1,81 @@
+# 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.
+
+# Multi-stage build using openenv-base
+# This Dockerfile is flexible and works for both:
+# - In-repo environments (with local OpenEnv sources)
+# - Standalone environments (with openenv from PyPI/Git)
+# The build script (openenv build) handles context detection and sets appropriate build args.
+
+ARG BASE_IMAGE=ghcr.io/meta-pytorch/openenv-base:latest
+FROM ${BASE_IMAGE} AS builder
+
+WORKDIR /app
+
+# Ensure git is available (required for installing dependencies from VCS)
+RUN apt-get update && \
+    apt-get install -y --no-install-recommends git && \
+    rm -rf /var/lib/apt/lists/*
+
+# Build argument to control whether we're building standalone or in-repo
+ARG BUILD_MODE=in-repo
+ARG ENV_NAME=MiniGridEnv
+
+# Copy environment code (always at root of build context)
+COPY . /app/env
+
+# For in-repo builds, openenv is already vendored in the build context
+# For standalone builds, openenv will be installed via pyproject.toml
+WORKDIR /app/env
+
+# Ensure uv is available (for local builds where base image lacks it)
+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
+    
+# Install dependencies using uv sync
+# If uv.lock exists, use it; otherwise resolve on the fly
+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 the virtual environment from builder
+COPY --from=builder /app/env/.venv /app/.venv
+
+# Copy the environment code
+COPY --from=builder /app/env /app/env
+
+# Set PATH to use the virtual environment
+ENV PATH="/app/.venv/bin:$PATH"
+
+# Set PYTHONPATH so imports work correctly
+ENV PYTHONPATH="/app/env:$PYTHONPATH"
+
+# Health check
+HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
+    CMD curl -f http://localhost:8000/health || exit 1
+
+# Run the FastAPI server
+# The module path is constructed to work with the /app/env structure
+ENV ENABLE_WEB_INTERFACE=true
+CMD ["sh", "-c", "cd /app/env && uvicorn server.app:app --host 0.0.0.0 --port 8000"]

README.md

@@ -1,10 +1,255 @@
 ---
-title: MiniGridEnv
-emoji: 🐢
-colorFrom: pink
-colorTo: red
+title: Minigridenv Environment Server
+emoji: 🔉
+colorFrom: purple
+colorTo: indigo
 sdk: docker
 pinned: false
+app_port: 8000
+base_path: /web
+tags:
+  - openenv
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Minigridenv Environment
+
+A simple test environment that echoes back messages. Perfect for testing the env APIs as well as demonstrating environment usage patterns.
+
+## Quick Start
+
+The simplest way to use the Minigridenv environment is through the `MinigridenvEnv` class:
+
+```python
+from MiniGridEnv import MinigridenvAction, MinigridenvEnv
+
+try:
+    # Create environment from Docker image
+    MiniGridEnvenv = MinigridenvEnv.from_docker_image("MiniGridEnv-env:latest")
+
+    # Reset
+    result = MiniGridEnvenv.reset()
+    print(f"Reset: {result.observation.echoed_message}")
+
+    # Send multiple messages
+    messages = ["Hello, World!", "Testing echo", "Final message"]
+
+    for msg in messages:
+        result = MiniGridEnvenv.step(MinigridenvAction(message=msg))
+        print(f"Sent: '{msg}'")
+        print(f"  → Echoed: '{result.observation.echoed_message}'")
+        print(f"  → Length: {result.observation.message_length}")
+        print(f"  → Reward: {result.reward}")
+
+finally:
+    # Always clean up
+    MiniGridEnvenv.close()
+```
+
+That's it! The `MinigridenvEnv.from_docker_image()` method handles:
+- Starting the Docker container
+- Waiting for the server to be ready
+- Connecting to the environment
+- Container cleanup when you call `close()`
+
+## Building the Docker Image
+
+Before using the environment, you need to build the Docker image:
+
+```bash
+# From project root
+docker build -t MiniGridEnv-env:latest -f server/Dockerfile .
+```
+
+## Deploying to Hugging Face Spaces
+
+You can easily deploy your OpenEnv environment to Hugging Face Spaces using the `openenv push` command:
+
+```bash
+# From the environment directory (where openenv.yaml is located)
+openenv push
+
+# Or specify options
+openenv push --namespace my-org --private
+```
+
+The `openenv push` command will:
+1. Validate that the directory is an OpenEnv environment (checks for `openenv.yaml`)
+2. Prepare a custom build for Hugging Face Docker space (enables web interface)
+3. Upload to Hugging Face (ensuring you're logged in)
+
+### Prerequisites
+
+- Authenticate with Hugging Face: The command will prompt for login if not already authenticated
+
+### Options
+
+- `--directory`, `-d`: Directory containing the OpenEnv environment (defaults to current directory)
+- `--repo-id`, `-r`: Repository ID in format 'username/repo-name' (defaults to 'username/env-name' from openenv.yaml)
+- `--base-image`, `-b`: Base Docker image to use (overrides Dockerfile FROM)
+- `--private`: Deploy the space as private (default: public)
+
+### Examples
+
+```bash
+# Push to your personal namespace (defaults to username/env-name from openenv.yaml)
+openenv push
+
+# Push to a specific repository
+openenv push --repo-id my-org/my-env
+
+# Push with a custom base image
+openenv push --base-image ghcr.io/meta-pytorch/openenv-base:latest
+
+# Push as a private space
+openenv push --private
+
+# Combine options
+openenv push --repo-id my-org/my-env --base-image custom-base:latest --private
+```
+
+After deployment, your space will be available at:
+`https://huggingface.co/spaces/<repo-id>`
+
+The deployed space includes:
+- **Web Interface** at `/web` - Interactive UI for exploring the environment
+- **API Documentation** at `/docs` - Full OpenAPI/Swagger interface
+- **Health Check** at `/health` - Container health monitoring
+- **WebSocket** at `/ws` - Persistent session endpoint for low-latency interactions
+
+## Environment Details
+
+### Action
+**MinigridenvAction**: Contains a single field
+- `message` (str) - The message to echo back
+
+### Observation
+**MinigridenvObservation**: Contains the echo response and metadata
+- `echoed_message` (str) - The message echoed back
+- `message_length` (int) - Length of the message
+- `reward` (float) - Reward based on message length (length × 0.1)
+- `done` (bool) - Always False for echo environment
+- `metadata` (dict) - Additional info like step count
+
+### Reward
+The reward is calculated as: `message_length × 0.1`
+- "Hi" → reward: 0.2
+- "Hello, World!" → reward: 1.3
+- Empty message → reward: 0.0
+
+## Advanced Usage
+
+### Connecting to an Existing Server
+
+If you already have a Minigridenv environment server running, you can connect directly:
+
+```python
+from MiniGridEnv import MinigridenvEnv
+
+# Connect to existing server
+MiniGridEnvenv = MinigridenvEnv(base_url="<ENV_HTTP_URL_HERE>")
+
+# Use as normal
+result = MiniGridEnvenv.reset()
+result = MiniGridEnvenv.step(MinigridenvAction(message="Hello!"))
+```
+
+Note: When connecting to an existing server, `MiniGridEnvenv.close()` will NOT stop the server.
+
+### Using the Context Manager
+
+The client supports context manager usage for automatic connection management:
+
+```python
+from MiniGridEnv import MinigridenvAction, MinigridenvEnv
+
+# Connect with context manager (auto-connects and closes)
+with MinigridenvEnv(base_url="http://localhost:8000") as env:
+    result = env.reset()
+    print(f"Reset: {result.observation.echoed_message}")
+    # Multiple steps with low latency
+    for msg in ["Hello", "World", "!"]:
+        result = env.step(MinigridenvAction(message=msg))
+        print(f"Echoed: {result.observation.echoed_message}")
+```
+
+The client uses WebSocket connections for:
+- **Lower latency**: No HTTP connection overhead per request
+- **Persistent session**: Server maintains your environment state
+- **Efficient for episodes**: Better for many sequential steps
+
+### Concurrent WebSocket Sessions
+
+The server supports multiple concurrent WebSocket connections. To enable this,
+modify `server/app.py` to use factory mode:
+
+```python
+# In server/app.py - use factory mode for concurrent sessions
+app = create_app(
+    MinigridenvEnvironment,  # Pass class, not instance
+    MinigridenvAction,
+    MinigridenvObservation,
+    max_concurrent_envs=4,  # Allow 4 concurrent sessions
+)
+```
+
+Then multiple clients can connect simultaneously:
+
+```python
+from MiniGridEnv import MinigridenvAction, MinigridenvEnv
+from concurrent.futures import ThreadPoolExecutor
+
+def run_episode(client_id: int):
+    with MinigridenvEnv(base_url="http://localhost:8000") as env:
+        result = env.reset()
+        for i in range(10):
+            result = env.step(MinigridenvAction(message=f"Client {client_id}, step {i}"))
+        return client_id, result.observation.message_length
+
+# Run 4 episodes concurrently
+with ThreadPoolExecutor(max_workers=4) as executor:
+    results = list(executor.map(run_episode, range(4)))
+```
+
+## Development & Testing
+
+### Direct Environment Testing
+
+Test the environment logic directly without starting the HTTP server:
+
+```bash
+# From the server directory
+python3 server/MiniGridEnv_environment.py
+```
+
+This verifies that:
+- Environment resets correctly
+- Step executes actions properly
+- State tracking works
+- Rewards are calculated correctly
+
+### Running Locally
+
+Run the server locally for development:
+
+```bash
+uvicorn server.app:app --reload
+```
+
+## Project Structure
+
+```
+MiniGridEnv/
+├── .dockerignore         # Docker build exclusions
+├── __init__.py            # Module exports
+├── README.md              # This file
+├── openenv.yaml           # OpenEnv manifest
+├── pyproject.toml         # Project metadata and dependencies
+├── uv.lock                # Locked dependencies (generated)
+├── client.py              # MinigridenvEnv client
+├── models.py              # Action and Observation models
+└── server/
+    ├── __init__.py        # Server module exports
+    ├── MiniGridEnv_environment.py  # Core environment logic
+    ├── app.py             # FastAPI application (HTTP + WebSocket endpoints)
+    └── Dockerfile         # Container image definition
+```

__init__.py

@@ -0,0 +1,6 @@
+"""MiniGridEnv package exports."""
+
+from .client import MiniGridEnvClient
+from .models import MiniGridAction, MiniGridObservation, MiniGridState
+
+__all__ = ["MiniGridAction", "MiniGridObservation", "MiniGridState", "MiniGridEnvClient"]

baselines/__init__.py

@@ -0,0 +1,7 @@
+"""Baseline agent implementations for MiniGridEnv."""
+
+from .heuristic_bot import BabyAIBotBaseline
+from .llm_zeroshot import LLMZeroShotBaseline
+from .random_action import RandomActionBaseline
+
+__all__ = ["RandomActionBaseline", "BabyAIBotBaseline", "LLMZeroShotBaseline"]

baselines/heuristic_bot.py

@@ -0,0 +1,35 @@
+"""BabyAI built-in bot baseline (near-optimal upper bound)."""
+
+from __future__ import annotations
+
+try:
+    from ..env.models import MiniGridAction, MiniGridObservation
+except ImportError:
+    from env.models import MiniGridAction, MiniGridObservation
+
+INT_TO_TEXT = {
+    0: "turn left",
+    1: "turn right",
+    2: "go forward",
+    3: "pickup",
+    4: "drop",
+    5: "toggle",
+    6: "done",
+}
+
+
+class BabyAIBotBaseline:
+    """Adapter for BabyAI's symbolic planner bot."""
+
+    def __init__(self, gym_env):
+        from minigrid.envs.babyai import BotAgent  # type: ignore
+
+        self._bot = BotAgent(gym_env.unwrapped)
+
+    def select_action(
+        self, obs: MiniGridObservation, raw_obs: dict
+    ) -> MiniGridAction:
+        del obs
+        action_int = self._bot.act(raw_obs)
+        command = INT_TO_TEXT.get(int(action_int), "done")
+        return MiniGridAction(command=command)

baselines/llm_zeroshot.py

@@ -0,0 +1,42 @@
+"""Zero-shot LLM baseline for MiniGridEnv."""
+
+from __future__ import annotations
+
+from typing import Callable, Optional
+
+try:
+    from ..env.models import MiniGridAction, MiniGridObservation
+except ImportError:
+    from env.models import MiniGridAction, MiniGridObservation
+
+
+class LLMZeroShotBaseline:
+    """Use an external LLM completion function with no RL fine-tuning."""
+
+    def __init__(
+        self,
+        model: str = "gpt-4o-mini",
+        provider: str = "openai",
+        completion_fn: Optional[Callable[[str], str]] = None,
+    ):
+        self._model = model
+        self._provider = provider
+        self._completion_fn = completion_fn
+
+    def select_action(self, obs: MiniGridObservation) -> MiniGridAction:
+        if self._completion_fn is None:
+            raise RuntimeError(
+                "LLMZeroShotBaseline requires completion_fn(prompt) -> text."
+            )
+        prompt = self._format_prompt(obs)
+        response = self._completion_fn(prompt)
+        return MiniGridAction(command=response.strip(), thought=None)
+
+    def _format_prompt(self, obs: MiniGridObservation) -> str:
+        return (
+            "You are navigating a grid world. Respond with exactly one action.\n\n"
+            "Valid actions: turn left, turn right, go forward, pickup, drop, toggle, done\n\n"
+            f"{obs.text}\n\n"
+            f"Step {obs.step_idx}/{obs.max_steps}. What is your next action?\n"
+            "Action:"
+        )

baselines/random_action.py

@@ -0,0 +1,34 @@
+"""Random-action baseline for MiniGridEnv."""
+
+from __future__ import annotations
+
+import random
+
+try:
+    from ..env.models import MiniGridAction, MiniGridObservation
+except ImportError:
+    from env.models import MiniGridAction, MiniGridObservation
+
+CANONICAL_ACTIONS = [
+    "turn left",
+    "turn right",
+    "go forward",
+    "pickup",
+    "drop",
+    "toggle",
+    "done",
+]
+
+
+class RandomActionBaseline:
+    """Uniform random baseline with movement bias."""
+
+    def select_action(
+        self, obs: MiniGridObservation, rng: random.Random | None = None
+    ) -> MiniGridAction:
+        del obs
+        chooser = rng or random
+        # Slightly prefer movement to avoid stalling.
+        weights = [1.0, 1.0, 3.0, 1.0, 0.5, 1.0, 0.1]
+        command = chooser.choices(CANONICAL_ACTIONS, weights=weights, k=1)[0]
+        return MiniGridAction(command=command)

client.py

@@ -0,0 +1,68 @@
+"""Typed OpenEnv client for MiniGridEnv."""
+
+from __future__ import annotations
+
+from typing import Any, Dict
+
+from openenv.core.client_types import StepResult
+from openenv.core.env_client import EnvClient
+
+try:
+    from .env.models import MiniGridAction, MiniGridObservation, MiniGridState
+except ImportError:
+    from env.models import MiniGridAction, MiniGridObservation, MiniGridState
+
+
+class MiniGridEnvClient(EnvClient[MiniGridAction, MiniGridObservation, MiniGridState]):
+    """WebSocket client for interacting with a MiniGridEnv server."""
+
+    def _step_payload(self, action: MiniGridAction) -> Dict[str, Any]:
+        payload: Dict[str, Any] = {"command": action.command}
+        if action.thought:
+            payload["thought"] = action.thought
+        return payload
+
+    def _parse_result(self, payload: Dict[str, Any]) -> StepResult[MiniGridObservation]:
+        obs_data = payload.get("observation")
+        if not isinstance(obs_data, dict):
+            obs_data = payload if isinstance(payload, dict) else {}
+        done = bool(payload.get("done", obs_data.get("done", False)))
+        reward = payload.get("reward", obs_data.get("reward"))
+        observation = MiniGridObservation(
+            text=obs_data.get("text", ""),
+            mission=obs_data.get("mission", ""),
+            step_idx=obs_data.get("step_idx", 0),
+            steps_remaining=obs_data.get("steps_remaining", 0),
+            max_steps=obs_data.get("max_steps", 1),
+            history=obs_data.get("history", []),
+            level_name=obs_data.get("level_name", ""),
+            last_action=obs_data.get("last_action"),
+            action_success=obs_data.get("action_success"),
+            done=done,
+            reward=reward,
+            metadata=obs_data.get("metadata", {}),
+        )
+        return StepResult(observation=observation, reward=reward, done=done)
+
+    def _parse_state(self, payload: Dict[str, Any]) -> MiniGridState:
+        state_data = payload.get("state")
+        if not isinstance(state_data, dict):
+            state_data = payload if isinstance(payload, dict) else {}
+        return MiniGridState(
+            episode_id=state_data.get("episode_id"),
+            step_count=state_data.get("step_count", 0),
+            level_name=state_data.get("level_name", ""),
+            level_difficulty=state_data.get("level_difficulty", 0),
+            completed=state_data.get("completed", False),
+            truncated=state_data.get("truncated", False),
+            total_reward=state_data.get("total_reward", 0.0),
+            steps_taken=state_data.get("steps_taken", 0),
+            optimal_steps=state_data.get("optimal_steps"),
+            efficiency_ratio=state_data.get("efficiency_ratio"),
+            valid_actions=state_data.get("valid_actions", 0),
+            invalid_actions=state_data.get("invalid_actions", 0),
+            action_distribution=state_data.get("action_distribution", {}),
+        )
+
+
+MiniGridEnv = MiniGridEnvClient

env/__init__.py

@@ -0,0 +1,14 @@
+"""MiniGridEnv core environment package."""
+
+from .config import EnvConfig, RewardConfig
+from .minigrid_env import MiniGridEnvironment
+from .models import MiniGridAction, MiniGridObservation, MiniGridState
+
+__all__ = [
+    "MiniGridAction",
+    "MiniGridObservation",
+    "MiniGridState",
+    "RewardConfig",
+    "EnvConfig",
+    "MiniGridEnvironment",
+]

env/action_parser.py

@@ -0,0 +1,93 @@
+"""Parse free-form model text into MiniGrid discrete actions."""
+
+from __future__ import annotations
+
+import re
+
+CANONICAL_ACTION_TO_INDEX: dict[str, int] = {
+    "turn left": 0,
+    "turn right": 1,
+    "go forward": 2,
+    "pickup": 3,
+    "drop": 4,
+    "toggle": 5,
+    "done": 6,
+}
+
+ACTION_MAP: dict[str, str] = {
+    "turn left": "turn left",
+    "turn right": "turn right",
+    "go forward": "go forward",
+    "move forward": "go forward",
+    "forward": "go forward",
+    "pickup": "pickup",
+    "pick up": "pickup",
+    "grab": "pickup",
+    "drop": "drop",
+    "toggle": "toggle",
+    "open": "toggle",
+    "close": "toggle",
+    "done": "done",
+    "wait": "done",
+    "noop": "done",
+}
+
+ALIASES: dict[str, str] = {
+    "left": "turn left",
+    "right": "turn right",
+    "ahead": "go forward",
+    "step": "go forward",
+    "walk": "go forward",
+    "take": "pickup",
+    "get": "pickup",
+    "release": "drop",
+    "put down": "drop",
+    "unlock": "toggle",
+    "switch": "toggle",
+    "stop": "done",
+}
+
+_ACTION_PATTERN = re.compile(r"action\s*:\s*(.+)", re.IGNORECASE)
+
+
+def _extract_structured_action(text: str) -> str:
+    """Extract action payload from `Action: ...` format when present."""
+    match = _ACTION_PATTERN.search(text)
+    if not match:
+        return text
+    candidate = match.group(1).strip()
+    return candidate.splitlines()[0].strip()
+
+
+def _match_from_map(cleaned: str, mapping: dict[str, str]) -> str | None:
+    if cleaned in mapping:
+        return mapping[cleaned]
+
+    best_key = None
+    best_len = -1
+    for key in mapping:
+        if key in cleaned and len(key) > best_len:
+            best_key = key
+            best_len = len(key)
+    if best_key is None:
+        return None
+    return mapping[best_key]
+
+
+def parse_action(text: str) -> tuple[int, str, bool]:
+    """Parse model output text into `(action_index, canonical_action, is_valid)`."""
+    cleaned = (text or "").strip().lower()
+    if not cleaned:
+        return CANONICAL_ACTION_TO_INDEX["go forward"], "go forward", False
+
+    cleaned = _extract_structured_action(cleaned)
+
+    canonical = _match_from_map(cleaned, ACTION_MAP)
+    if canonical is not None:
+        return CANONICAL_ACTION_TO_INDEX[canonical], canonical, True
+
+    canonical = _match_from_map(cleaned, ALIASES)
+    if canonical is not None:
+        return CANONICAL_ACTION_TO_INDEX[canonical], canonical, True
+
+    return CANONICAL_ACTION_TO_INDEX["go forward"], "go forward", False

env/config.py

@@ -0,0 +1,40 @@
+"""Configuration objects for MiniGridEnv."""
+
+from dataclasses import dataclass, field
+from typing import Optional
+
+
+@dataclass
+class RewardConfig:
+    """Reward-shaping controls for environment step rewards."""
+
+    mode: str = "binary"  # "binary" | "shaped" | "efficiency"
+    completion_reward: float = 1.0
+    failure_reward: float = 0.0
+    step_penalty: float = 0.0
+    invalid_action_penalty: float = 0.0
+    efficiency_bonus_weight: float = 0.5
+
+
+@dataclass
+class EnvConfig:
+    """Episode, observation, and reproducibility settings."""
+
+    # Level selection
+    level_name: str = "GoToRedBall"
+
+    # Episode limits
+    max_steps_override: Optional[int] = None
+
+    # Reward configuration
+    reward: RewardConfig = field(default_factory=RewardConfig)
+
+    # Observation configuration
+    include_history: bool = True
+    max_history_length: int = 20
+    include_raw_grid: bool = False
+
+    # Reproducibility and optional metrics
+    seed: int = 42
+    compute_optimal: bool = False
+    track_carrying: bool = True

env/grid_to_text.py

@@ -0,0 +1,261 @@
+"""Convert MiniGrid/BabyAI observations into rich natural language text."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import numpy as np
+
+OBJECT_TYPES = {
+    0: "unseen",
+    1: "empty",
+    2: "wall",
+    3: "floor",
+    4: "door",
+    5: "key",
+    6: "ball",
+    7: "box",
+    8: "goal",
+    9: "lava",
+    10: "agent",
+}
+
+COLORS = {
+    0: "red",
+    1: "green",
+    2: "blue",
+    3: "purple",
+    4: "yellow",
+    5: "grey",
+}
+
+DOOR_STATES = {0: "open", 1: "closed", 2: "locked"}
+DIRECTION_NAMES = {0: "east", 1: "south", 2: "west", 3: "north"}
+
+_AGENT_ROW = 6
+_AGENT_COL = 3
+
+
+def _format_object_name(obj_type: str, color: str | None, state: str | None = None) -> str:
+    if obj_type == "door":
+        prefix = f"{state} {color}".strip() if color else (state or "door")
+        return f"a {prefix} door".replace("  ", " ").strip()
+    if color:
+        return f"a {color} {obj_type}"
+    return f"a {obj_type}"
+
+
+def _relative_position_phrase(rel_row: int, rel_col: int) -> str:
+    parts: list[str] = []
+
+    if rel_row < 0:
+        steps_ahead = abs(rel_row)
+        parts.append(f"{steps_ahead} step{'s' if steps_ahead != 1 else ''} ahead")
+    elif rel_row > 0:
+        steps_behind = rel_row
+        parts.append(f"{steps_behind} step{'s' if steps_behind != 1 else ''} behind")
+
+    if rel_col < 0:
+        steps_left = abs(rel_col)
+        parts.append(f"{steps_left} to your left")
+    elif rel_col > 0:
+        steps_right = rel_col
+        parts.append(f"{steps_right} to your right")
+
+    if not parts:
+        return "at your position"
+    if len(parts) == 1:
+        return parts[0]
+    return f"{parts[0]} and {parts[1]}"
+
+
+def _describe_cell(grid: np.ndarray, row: int, col: int) -> str:
+    if row < 0 or row >= grid.shape[0] or col < 0 or col >= grid.shape[1]:
+        return "a wall boundary"
+
+    obj_idx = int(grid[row, col, 0])
+    color_idx = int(grid[row, col, 1])
+    state_idx = int(grid[row, col, 2])
+
+    obj_type = OBJECT_TYPES.get(obj_idx, "unknown")
+    color = COLORS.get(color_idx)
+
+    if obj_type in {"empty", "floor"}:
+        return "empty space"
+    if obj_type == "unseen":
+        return "unseen area"
+    if obj_type == "wall":
+        return "a wall"
+    if obj_type == "door":
+        return _format_object_name("door", color, DOOR_STATES.get(state_idx, "closed"))
+    if obj_type == "lava":
+        return "lava"
+    return _format_object_name(obj_type, color)
+
+
+def _scan_objects(grid: np.ndarray) -> list[dict[str, Any]]:
+    """Extract notable interactive objects with relative positions."""
+    objects: list[dict[str, Any]] = []
+
+    for row in range(grid.shape[0]):
+        for col in range(grid.shape[1]):
+            obj_idx = int(grid[row, col, 0])
+            color_idx = int(grid[row, col, 1])
+            state_idx = int(grid[row, col, 2])
+
+            obj_type = OBJECT_TYPES.get(obj_idx, "unknown")
+            if obj_type in {"unseen", "empty", "wall", "floor", "agent"}:
+                continue
+
+            rel_row = row - _AGENT_ROW
+            rel_col = col - _AGENT_COL
+            state = DOOR_STATES.get(state_idx) if obj_type == "door" else None
+            color = COLORS.get(color_idx)
+            objects.append(
+                {
+                    "type": obj_type,
+                    "color": color,
+                    "state": state,
+                    "row": row,
+                    "col": col,
+                    "rel_row": rel_row,
+                    "rel_col": rel_col,
+                    "distance": abs(rel_row) + abs(rel_col),
+                    "direction_desc": _relative_position_phrase(rel_row, rel_col),
+                }
+            )
+
+    objects.sort(key=lambda item: (item["distance"], item["row"], item["col"]))
+    return objects
+
+
+def _describe_immediate_surroundings(grid: np.ndarray) -> str:
+    """Describe the nearest cells around the agent."""
+    ahead = _describe_cell(grid, _AGENT_ROW - 1, _AGENT_COL)
+    left = _describe_cell(grid, _AGENT_ROW, _AGENT_COL - 1)
+    right = _describe_cell(grid, _AGENT_ROW, _AGENT_COL + 1)
+    return (
+        f"Directly ahead: {ahead}.\n"
+        f"To your left: {left}.\n"
+        f"To your right: {right}."
+    )
+
+
+def _describe_path_ahead(grid: np.ndarray) -> str:
+    """Describe what appears in the straight-ahead lane."""
+    segments: list[str] = []
+    empty_run = 0
+
+    for row in range(_AGENT_ROW - 1, -1, -1):
+        cell_desc = _describe_cell(grid, row, _AGENT_COL)
+        if cell_desc == "empty space":
+            empty_run += 1
+            continue
+
+        if empty_run > 0:
+            segments.append(
+                f"empty space for {empty_run} step{'s' if empty_run != 1 else ''}"
+            )
+            empty_run = 0
+        segments.append(cell_desc)
+        if cell_desc in {"a wall", "a wall boundary", "unseen area"}:
+            break
+
+    if empty_run > 0:
+        segments.append(
+            f"empty space for {empty_run} step{'s' if empty_run != 1 else ''}"
+        )
+
+    if not segments:
+        return "Looking ahead: no clear information."
+    if len(segments) == 1:
+        return f"Looking ahead: {segments[0]}."
+    return f"Looking ahead: {', then '.join(segments)}."
+
+
+def _describe_notable_objects(objects: list[dict[str, Any]]) -> str:
+    """List visible interactive objects with positions."""
+    if not objects:
+        return "Notable objects: none visible."
+
+    lines = ["Notable objects:"]
+    for obj in objects:
+        name = _format_object_name(obj["type"], obj.get("color"), obj.get("state"))
+        lines.append(f"- {name} ({obj['direction_desc']}).")
+    return "\n".join(lines)
+
+
+def _describe_carrying_status(carrying: Any) -> str:
+    """Describe what the agent is currently carrying."""
+    if carrying is None:
+        return "You are carrying: nothing."
+
+    if isinstance(carrying, dict):
+        obj_type = carrying.get("type")
+        color = carrying.get("color")
+    else:
+        obj_type = getattr(carrying, "type", None)
+        color = getattr(carrying, "color", None)
+
+    if obj_type is None:
+        return "You are carrying: an object."
+    if color:
+        return f"You are carrying: a {color} {obj_type}."
+    return f"You are carrying: a {obj_type}."
+
+
+def _render_ascii_grid(grid: np.ndarray) -> str:
+    """Render a compact ASCII view for debugging."""
+    glyphs = {
+        "unseen": "?",
+        "empty": ".",
+        "wall": "#",
+        "floor": ".",
+        "door": "D",
+        "key": "K",
+        "ball": "B",
+        "box": "X",
+        "goal": "G",
+        "lava": "L",
+        "agent": "A",
+    }
+    rows: list[str] = []
+    for row in range(grid.shape[0]):
+        chars: list[str] = []
+        for col in range(grid.shape[1]):
+            obj_type = OBJECT_TYPES.get(int(grid[row, col, 0]), "unknown")
+            chars.append(glyphs.get(obj_type, "!"))
+        rows.append("".join(chars))
+    return "\n".join(rows)
+
+
+def grid_to_text(
+    obs: dict[str, Any], carrying: Any = None, include_raw_grid: bool = False
+) -> str:
+    """Convert MiniGrid raw observation dict to a rich language description."""
+    grid = obs.get("image")
+    if grid is None:
+        return "Mission: unknown.\nObservation is missing grid image."
+
+    mission = str(obs.get("mission", "")).strip() or "unknown mission"
+    direction = int(obs.get("direction", 0))
+    direction_name = DIRECTION_NAMES.get(direction, "unknown")
+
+    if not isinstance(grid, np.ndarray):
+        grid = np.asarray(grid)
+
+    objects = _scan_objects(grid)
+    parts = [
+        f"Mission: {mission}",
+        f"You are facing {direction_name}.",
+        "",
+        _describe_immediate_surroundings(grid),
+        _describe_path_ahead(grid),
+        _describe_notable_objects(objects),
+        _describe_carrying_status(carrying),
+    ]
+
+    if include_raw_grid:
+        parts.extend(["", "Raw grid (debug):", _render_ascii_grid(grid)])
+
+    return "\n".join(part for part in parts if part is not None)

env/levels.py

@@ -0,0 +1,175 @@
+"""BabyAI level registry and curriculum definitions for MiniGridEnv."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+
+@dataclass(frozen=True)
+class LevelConfig:
+    """Configuration for a single BabyAI level."""
+
+    gym_id: str
+    name: str
+    description: str
+    difficulty: int
+    max_steps: int
+    expected_optimal_steps: int
+    requires_interaction: bool
+    num_objects: int
+    involves_language_composition: bool
+    fallback_gym_ids: tuple[str, ...] = ()
+
+    @property
+    def candidate_gym_ids(self) -> list[str]:
+        return [self.gym_id, *self.fallback_gym_ids]
+
+
+LEVEL_REGISTRY: list[LevelConfig] = [
+    # Stage 0: simple navigation
+    LevelConfig(
+        gym_id="BabyAI-GoToRedBallGrey-v0",
+        fallback_gym_ids=("BabyAI-GoToRedBall-v0",),
+        name="GoToRedBall",
+        description="Navigate to the red ball in a single room.",
+        difficulty=0,
+        max_steps=64,
+        expected_optimal_steps=10,
+        requires_interaction=False,
+        num_objects=1,
+        involves_language_composition=False,
+    ),
+    LevelConfig(
+        gym_id="BabyAI-GoToObj-v0",
+        name="GoToObj",
+        description="Navigate to a specific colored object.",
+        difficulty=1,
+        max_steps=64,
+        expected_optimal_steps=12,
+        requires_interaction=False,
+        num_objects=2,
+        involves_language_composition=False,
+    ),
+    LevelConfig(
+        gym_id="BabyAI-GoToLocal-v0",
+        name="GoToLocal",
+        description="Navigate to a specific object with distractors.",
+        difficulty=2,
+        max_steps=64,
+        expected_optimal_steps=15,
+        requires_interaction=False,
+        num_objects=4,
+        involves_language_composition=False,
+    ),
+    # Stage 1: object interaction
+    LevelConfig(
+        gym_id="BabyAI-PickupLoc-v0",
+        name="PickupLoc",
+        description="Pick up a specific object in a single room.",
+        difficulty=3,
+        max_steps=64,
+        expected_optimal_steps=14,
+        requires_interaction=True,
+        num_objects=4,
+        involves_language_composition=False,
+    ),
+    LevelConfig(
+        gym_id="BabyAI-OpenDoor-v0",
+        name="OpenDoor",
+        description="Open a door of a specified color.",
+        difficulty=3,
+        max_steps=64,
+        expected_optimal_steps=12,
+        requires_interaction=True,
+        num_objects=1,
+        involves_language_composition=False,
+    ),
+    LevelConfig(
+        gym_id="BabyAI-UnlockLocal-v0",
+        name="UnlockLocal",
+        description="Unlock a local door with the matching key.",
+        difficulty=4,
+        max_steps=128,
+        expected_optimal_steps=25,
+        requires_interaction=True,
+        num_objects=3,
+        involves_language_composition=False,
+    ),
+    # Stage 2: multi-room and compositional
+    LevelConfig(
+        gym_id="BabyAI-GoTo-v0",
+        name="GoTo",
+        description="Navigate to a specified object across rooms.",
+        difficulty=5,
+        max_steps=128,
+        expected_optimal_steps=30,
+        requires_interaction=True,
+        num_objects=6,
+        involves_language_composition=False,
+    ),
+    LevelConfig(
+        gym_id="BabyAI-PutNextLocal-v0",
+        name="PutNextLocal",
+        description="Place one object next to another local object.",
+        difficulty=6,
+        max_steps=128,
+        expected_optimal_steps=20,
+        requires_interaction=True,
+        num_objects=4,
+        involves_language_composition=True,
+    ),
+    # Stage 3: hardest compositional tasks
+    LevelConfig(
+        gym_id="BabyAI-Synth-v0",
+        name="Synth",
+        description="Random compositional instructions.",
+        difficulty=7,
+        max_steps=256,
+        expected_optimal_steps=40,
+        requires_interaction=True,
+        num_objects=8,
+        involves_language_composition=True,
+    ),
+    LevelConfig(
+        gym_id="BabyAI-BossLevel-v0",
+        name="BossLevel",
+        description="Hardest compositional BabyAI level.",
+        difficulty=8,
+        max_steps=512,
+        expected_optimal_steps=80,
+        requires_interaction=True,
+        num_objects=10,
+        involves_language_composition=True,
+    ),
+]
+
+
+def get_level(name: str) -> LevelConfig:
+    """Return a level by short name (case-insensitive)."""
+    for level in LEVEL_REGISTRY:
+        if level.name.lower() == name.lower():
+            return level
+    available = ", ".join(level.name for level in LEVEL_REGISTRY)
+    raise ValueError(f"Unknown level '{name}'. Available levels: {available}")
+
+
+def get_levels_by_difficulty(max_difficulty: int) -> list[LevelConfig]:
+    """Return levels with difficulty <= max_difficulty."""
+    return [level for level in LEVEL_REGISTRY if level.difficulty <= max_difficulty]
+
+
+@dataclass
+class CurriculumConfig:
+    """Curriculum stages and advancement settings."""
+
+    stages: list[list[str]] = field(
+        default_factory=lambda: [
+            ["GoToRedBall"],
+            ["GoToObj", "GoToLocal"],
+            ["PickupLoc", "OpenDoor", "UnlockLocal"],
+            ["GoTo", "PutNextLocal"],
+            ["Synth", "BossLevel"],
+        ]
+    )
+    advance_threshold: float = 0.8
+    episodes_per_eval: int = 100

env/minigrid_env.py

@@ -0,0 +1,319 @@
+"""Core OpenEnv environment for text-based MiniGrid/BabyAI interaction."""
+
+from __future__ import annotations
+
+import uuid
+from typing import Any, Optional
+
+import gymnasium as gym
+import numpy as np
+
+try:
+    from .action_parser import parse_action
+    from .config import EnvConfig
+    from .grid_to_text import grid_to_text
+    from .levels import LevelConfig, get_level
+    from .models import MiniGridAction, MiniGridObservation, MiniGridState
+    from .reward import compute_step_reward
+except ImportError:
+    from env.action_parser import parse_action
+    from env.config import EnvConfig
+    from env.grid_to_text import grid_to_text
+    from env.levels import LevelConfig, get_level
+    from env.models import MiniGridAction, MiniGridObservation, MiniGridState
+    from env.reward import compute_step_reward
+
+try:
+    from openenv.core.env_server.interfaces import Environment
+except ImportError:
+    from abc import ABC, abstractmethod
+    from typing import Generic, TypeVar
+
+    ActT = TypeVar("ActT")
+    ObsT = TypeVar("ObsT")
+    StateT = TypeVar("StateT")
+
+    class Environment(ABC, Generic[ActT, ObsT, StateT]):
+        @abstractmethod
+        def reset(self, seed=None, episode_id=None, **kwargs): ...
+
+        @abstractmethod
+        def step(self, action, timeout_s=None, **kwargs): ...
+
+        @property
+        @abstractmethod
+        def state(self): ...
+
+
+class MiniGridEnvironment(Environment[MiniGridAction, MiniGridObservation, MiniGridState]):
+    """OpenEnv wrapper around BabyAI levels in MiniGrid."""
+
+    SUPPORTS_CONCURRENT_SESSIONS = True
+
+    def __init__(self, config: Optional[EnvConfig] = None, **kwargs: Any):
+        super().__init__(**kwargs)
+        self._config = config or EnvConfig()
+        self._level: LevelConfig = get_level(self._config.level_name)
+        self._max_steps = self._config.max_steps_override or self._level.max_steps
+
+        self._gym_env: Optional[gym.Env] = None
+        self._episode_id: str = ""
+        self._active_seed: int = self._config.seed
+        self._step_idx: int = 0
+        self._done: bool = False
+        self._history: list[dict[str, Any]] = []
+        self._total_reward: float = 0.0
+        self._completed: bool = False
+        self._truncated: bool = False
+        self._valid_actions: int = 0
+        self._invalid_actions: int = 0
+        self._action_counts: dict[str, int] = {}
+        self._last_obs: Optional[dict[str, Any]] = None
+        self._carrying: Any = None
+        self._optimal_steps: Optional[int] = None
+
+    def _make_gym_env(self) -> gym.Env:
+        # Import registers BabyAI env IDs in gymnasium.
+        import minigrid  # noqa: F401
+
+        last_error: Optional[Exception] = None
+        for gym_id in self._level.candidate_gym_ids:
+            try:
+                return gym.make(gym_id)
+            except Exception as exc:  # pragma: no cover - depends on installed minigrid
+                last_error = exc
+        raise RuntimeError(
+            f"Failed to create level '{self._level.name}' with ids {self._level.candidate_gym_ids}"
+        ) from last_error
+
+    def reset(
+        self,
+        seed: Optional[int] = None,
+        episode_id: Optional[str] = None,
+        **kwargs: Any,
+    ) -> MiniGridObservation:
+        del kwargs
+        if self._gym_env is not None:
+            try:
+                self._gym_env.close()
+            except Exception:
+                pass
+
+        self._episode_id = episode_id or str(uuid.uuid4())
+        self._active_seed = self._config.seed if seed is None else seed
+        self._gym_env = self._make_gym_env()
+        raw_obs, _info = self._gym_env.reset(seed=self._active_seed)
+
+        self._step_idx = 0
+        self._done = False
+        self._history = []
+        self._total_reward = 0.0
+        self._completed = False
+        self._truncated = False
+        self._valid_actions = 0
+        self._invalid_actions = 0
+        self._action_counts = {}
+        self._last_obs = raw_obs
+        self._carrying = getattr(self._gym_env.unwrapped, "carrying", None)
+
+        self._optimal_steps = self._compute_optimal_steps() if self._config.compute_optimal else None
+
+        text = grid_to_text(
+            raw_obs,
+            carrying=self._carrying,
+            include_raw_grid=self._config.include_raw_grid,
+        )
+        return MiniGridObservation(
+            text=text,
+            mission=raw_obs.get("mission", ""),
+            step_idx=0,
+            steps_remaining=self._max_steps,
+            max_steps=self._max_steps,
+            history=[],
+            level_name=self._level.name,
+            last_action=None,
+            action_success=None,
+            done=False,
+            reward=None,
+        )
+
+    def step(
+        self,
+        action: MiniGridAction,
+        timeout_s: Optional[float] = None,
+        **kwargs: Any,
+    ) -> MiniGridObservation:
+        del timeout_s, kwargs
+        if self._done:
+            raise RuntimeError("Episode is finished. Call reset() before stepping again.")
+        if self._gym_env is None:
+            raise RuntimeError("Environment is not initialized. Call reset() first.")
+
+        action_idx, canonical_action, was_valid = parse_action(action.command)
+        if was_valid:
+            self._valid_actions += 1
+        else:
+            self._invalid_actions += 1
+        self._action_counts[canonical_action] = self._action_counts.get(canonical_action, 0) + 1
+
+        prev_obs = self._last_obs
+        prev_carrying = self._carrying
+        raw_obs, _gym_reward, terminated, truncated, _info = self._gym_env.step(action_idx)
+
+        self._step_idx += 1
+        if self._step_idx >= self._max_steps and not terminated:
+            truncated = True
+
+        done = bool(terminated or truncated)
+        self._carrying = getattr(self._gym_env.unwrapped, "carrying", None)
+
+        reward, _reward_breakdown = compute_step_reward(
+            terminated=bool(terminated),
+            truncated=bool(truncated),
+            action_valid=was_valid,
+            step_idx=self._step_idx,
+            max_steps=self._max_steps,
+            optimal_steps=self._optimal_steps,
+            config=self._config.reward,
+        )
+        self._total_reward += reward
+
+        self._history.append(
+            {
+                "step": self._step_idx - 1,
+                "action": canonical_action,
+                "action_raw": action.command,
+                "action_valid": was_valid,
+                "reward": reward,
+                **({"thought": action.thought} if action.thought else {}),
+            }
+        )
+
+        visible_history: list[dict[str, Any]] = []
+        if self._config.include_history:
+            if self._config.max_history_length > 0:
+                visible_history = self._history[-self._config.max_history_length :]
+            else:
+                visible_history = []
+
+        self._done = done
+        self._completed = bool(terminated)
+        self._truncated = bool(truncated)
+        self._last_obs = raw_obs
+
+        text = grid_to_text(
+            raw_obs,
+            carrying=self._carrying,
+            include_raw_grid=self._config.include_raw_grid,
+        )
+        action_success = self._detect_action_success(
+            canonical_action=canonical_action,
+            prev_obs=prev_obs,
+            next_obs=raw_obs,
+            prev_carrying=prev_carrying,
+            terminated=bool(terminated),
+        )
+
+        return MiniGridObservation(
+            text=text,
+            mission=raw_obs.get("mission", ""),
+            step_idx=self._step_idx,
+            steps_remaining=max(0, self._max_steps - self._step_idx),
+            max_steps=self._max_steps,
+            history=visible_history,
+            level_name=self._level.name,
+            last_action=canonical_action,
+            action_success=action_success,
+            done=done,
+            reward=reward,
+        )
+
+    def _compute_optimal_steps(self) -> Optional[int]:
+        """Attempt to compute near-optimal length with BabyAI's built-in bot."""
+        try:
+            from minigrid.envs.babyai import BotAgent  # type: ignore
+        except Exception:
+            return None
+
+        eval_env: Optional[gym.Env] = None
+        try:
+            eval_env = self._make_gym_env()
+            obs, _info = eval_env.reset(seed=self._active_seed)
+            bot = BotAgent(eval_env.unwrapped)
+
+            steps = 0
+            done = False
+            while not done and steps < self._max_steps:
+                action = bot.act(obs)
+                obs, _rew, terminated, truncated, _info = eval_env.step(action)
+                done = bool(terminated or truncated)
+                steps += 1
+            return steps if done else None
+        except Exception:
+            return None
+        finally:
+            if eval_env is not None:
+                try:
+                    eval_env.close()
+                except Exception:
+                    pass
+
+    @staticmethod
+    def _images_equal(obs_a: Optional[dict[str, Any]], obs_b: Optional[dict[str, Any]]) -> bool:
+        if not obs_a or not obs_b:
+            return False
+        img_a = obs_a.get("image")
+        img_b = obs_b.get("image")
+        if img_a is None or img_b is None:
+            return False
+        return bool(np.array_equal(np.asarray(img_a), np.asarray(img_b)))
+
+    def _detect_action_success(
+        self,
+        *,
+        canonical_action: str,
+        prev_obs: Optional[dict[str, Any]],
+        next_obs: Optional[dict[str, Any]],
+        prev_carrying: Any,
+        terminated: bool,
+    ) -> Optional[bool]:
+        if terminated:
+            return True
+        if canonical_action == "pickup":
+            return prev_carrying is None and self._carrying is not None
+        if canonical_action == "drop":
+            return prev_carrying is not None and self._carrying is None
+        if canonical_action in {"turn left", "turn right"}:
+            if not prev_obs or not next_obs:
+                return None
+            return prev_obs.get("direction") != next_obs.get("direction")
+        if canonical_action == "go forward":
+            return not self._images_equal(prev_obs, next_obs)
+        if canonical_action == "toggle":
+            return not self._images_equal(prev_obs, next_obs)
+        return None
+
+    @property
+    def state(self) -> MiniGridState:
+        efficiency = None
+        if self._optimal_steps and self._step_idx > 0:
+            efficiency = float(self._optimal_steps) / float(self._step_idx)
+
+        return MiniGridState(
+            episode_id=self._episode_id,
+            step_count=self._step_idx,
+            level_name=self._level.name,
+            level_difficulty=self._level.difficulty,
+            completed=self._completed,
+            truncated=self._truncated,
+            total_reward=self._total_reward,
+            steps_taken=self._step_idx,
+            optimal_steps=self._optimal_steps,
+            efficiency_ratio=efficiency,
+            valid_actions=self._valid_actions,
+            invalid_actions=self._invalid_actions,
+            action_distribution=dict(self._action_counts),
+        )
+
+
+MiniGridEnv = MiniGridEnvironment

env/models.py

@@ -0,0 +1,88 @@
+"""OpenEnv Pydantic models for MiniGridEnv."""
+
+from typing import Any, Optional
+
+from pydantic import BaseModel, ConfigDict, Field
+
+try:
+    from openenv.core.env_server.types import Action as _ActionBase
+    from openenv.core.env_server.types import Observation as _ObservationBase
+    from openenv.core.env_server.types import State as _StateBase
+except ImportError:
+    _ActionBase = BaseModel
+    _ObservationBase = BaseModel
+    _StateBase = BaseModel
+
+
+class MiniGridAction(_ActionBase):
+    """Agent action represented as a natural language command."""
+
+    if _ActionBase is BaseModel:
+        model_config = ConfigDict(extra="forbid")
+        metadata: dict[str, Any] = Field(default_factory=dict)
+
+    command: str = Field(
+        ...,
+        description="Natural language command (for example: 'go forward', 'turn left', 'pickup').",
+    )
+    thought: Optional[str] = Field(
+        default=None,
+        description="Optional reasoning trace for logging/analysis.",
+    )
+
+
+class MiniGridObservation(_ObservationBase):
+    """Text observation derived from MiniGrid's egocentric 7x7 view."""
+
+    if _ObservationBase is BaseModel:
+        model_config = ConfigDict(extra="forbid")
+        done: bool = Field(default=False)
+        reward: Optional[float] = Field(default=None)
+        metadata: dict[str, Any] = Field(default_factory=dict)
+
+    text: str = Field(..., description="Natural-language environment description.")
+    mission: str = Field(..., description="Current mission instruction.")
+    step_idx: int = Field(..., ge=0, description="Current 0-indexed step number.")
+    steps_remaining: int = Field(..., ge=0, description="Steps left before truncation.")
+    max_steps: int = Field(..., ge=1, description="Maximum number of steps in this episode.")
+    history: list[dict[str, Any]] = Field(
+        default_factory=list,
+        description="Recent step history entries for prompt reconstruction.",
+    )
+    level_name: str = Field(default="", description="Selected BabyAI level short name.")
+    last_action: Optional[str] = Field(default=None, description="Canonical last action string.")
+    action_success: Optional[bool] = Field(
+        default=None,
+        description="Whether the last action had an observable effect.",
+    )
+
+
+class MiniGridState(_StateBase):
+    """Episode-level state metrics for logging and debugging."""
+
+    if _StateBase is BaseModel:
+        model_config = ConfigDict(extra="allow")
+        episode_id: Optional[str] = Field(default=None)
+        step_count: int = Field(default=0, ge=0)
+
+    level_name: str = Field(default="", description="Selected level name.")
+    level_difficulty: int = Field(default=0, ge=0, description="Difficulty stage index.")
+    completed: bool = Field(default=False, description="True when mission succeeds.")
+    truncated: bool = Field(default=False, description="True when max step budget is exhausted.")
+    total_reward: float = Field(default=0.0, description="Cumulative reward this episode.")
+    steps_taken: int = Field(default=0, ge=0, description="Total steps executed.")
+    optimal_steps: Optional[int] = Field(
+        default=None,
+        ge=0,
+        description="Optional best-path length from the BabyAI bot.",
+    )
+    efficiency_ratio: Optional[float] = Field(
+        default=None,
+        description="optimal_steps / steps_taken when available.",
+    )
+    valid_actions: int = Field(default=0, ge=0, description="Count of valid parsed actions.")
+    invalid_actions: int = Field(default=0, ge=0, description="Count of invalid parsed actions.")
+    action_distribution: dict[str, int] = Field(
+        default_factory=dict,
+        description="Histogram of canonical actions used so far.",
+    )

env/reward.py

@@ -0,0 +1,51 @@
+"""Reward helpers for MiniGridEnv episodes."""
+
+from __future__ import annotations
+
+try:
+    from .config import RewardConfig
+except ImportError:
+    from env.config import RewardConfig
+
+
+def compute_step_reward(
+    *,
+    terminated: bool,
+    truncated: bool,
+    action_valid: bool,
+    step_idx: int,
+    max_steps: int,
+    optimal_steps: int | None,
+    config: RewardConfig,
+) -> tuple[float, dict[str, float]]:
+    """Compute reward and breakdown for a single environment step."""
+    del optimal_steps  # Reserved for future ratio-based shaping.
+
+    mode = config.mode.lower().strip()
+    if mode not in {"binary", "shaped", "efficiency"}:
+        mode = "binary"
+
+    breakdown: dict[str, float] = {}
+    total = 0.0
+
+    if terminated:
+        total += config.completion_reward
+        breakdown["completion"] = config.completion_reward
+        if mode == "efficiency":
+            efficiency = max(0.0, 1.0 - (float(step_idx) / float(max_steps)))
+            bonus = config.efficiency_bonus_weight * efficiency
+            total += bonus
+            breakdown["efficiency_bonus"] = bonus
+    elif truncated:
+        total += config.failure_reward
+        breakdown["timeout"] = config.failure_reward
+
+    if mode in {"shaped", "efficiency"}:
+        total += config.step_penalty
+        breakdown["step_penalty"] = config.step_penalty
+        if not action_valid:
+            total += config.invalid_action_penalty
+            breakdown["invalid_action"] = config.invalid_action_penalty
+
+    breakdown["total"] = total
+    return total, breakdown

eval/__init__.py

@@ -0,0 +1 @@
+"""Evaluation helpers for MiniGridEnv."""

eval/evaluate.py

@@ -0,0 +1,140 @@
+"""Baseline evaluation harness for MiniGridEnv."""
+
+from __future__ import annotations
+
+import inspect
+import math
+from collections import Counter
+from statistics import mean
+from typing import Any, Callable
+
+try:
+    from ..env.config import EnvConfig
+    from ..env.minigrid_env import MiniGridEnvironment
+except ImportError:
+    from env.config import EnvConfig
+    from env.minigrid_env import MiniGridEnvironment
+
+
+def _sem(values: list[float]) -> float:
+    if len(values) < 2:
+        return 0.0
+    avg = mean(values)
+    variance = sum((value - avg) ** 2 for value in values) / (len(values) - 1)
+    return math.sqrt(variance / len(values))
+
+
+def _resolve_episode_baseline(
+    baseline: Any, env: MiniGridEnvironment
+) -> Any:
+    if hasattr(baseline, "select_action"):
+        return baseline
+    if callable(baseline):
+        try:
+            return baseline(env._gym_env)  # type: ignore[attr-defined]
+        except TypeError:
+            return baseline()
+    raise TypeError("baseline must be an agent object or a callable factory")
+
+
+def _select_action(agent: Any, obs: Any, raw_obs: dict | None):
+    select_action = getattr(agent, "select_action")
+    sig = inspect.signature(select_action)
+    if len(sig.parameters) >= 2:
+        return select_action(obs, raw_obs)
+    return select_action(obs)
+
+
+def evaluate_baseline(
+    baseline: Any,
+    level_name: str,
+    n_episodes: int = 100,
+    seed: int = 42,
+) -> dict[str, Any]:
+    """Evaluate one baseline on one level and return aggregate metrics."""
+    completed_flags: list[float] = []
+    completed_steps: list[float] = []
+    rewards: list[float] = []
+    efficiencies: list[float] = []
+    action_counter: Counter[str] = Counter()
+    total_valid = 0
+    total_invalid = 0
+
+    for episode_offset in range(n_episodes):
+        env = MiniGridEnvironment(
+            config=EnvConfig(level_name=level_name, seed=seed + episode_offset)
+        )
+        obs = env.reset(seed=seed + episode_offset)
+        agent = _resolve_episode_baseline(baseline, env)
+
+        while not obs.done:
+            raw_obs = env._last_obs  # type: ignore[attr-defined]
+            action = _select_action(agent, obs, raw_obs)
+            obs = env.step(action)
+
+        state = env.state
+        completed = 1.0 if state.completed else 0.0
+        completed_flags.append(completed)
+        rewards.append(float(state.total_reward))
+        total_valid += int(state.valid_actions)
+        total_invalid += int(state.invalid_actions)
+        action_counter.update(state.action_distribution)
+        if state.completed:
+            completed_steps.append(float(state.steps_taken))
+            if state.efficiency_ratio is not None:
+                efficiencies.append(float(state.efficiency_ratio))
+
+    total_actions = total_valid + total_invalid
+    return {
+        "level": level_name,
+        "episodes": n_episodes,
+        "completion_rate": mean(completed_flags) if completed_flags else 0.0,
+        "completion_rate_sem": _sem(completed_flags),
+        "mean_steps_completed": mean(completed_steps) if completed_steps else 0.0,
+        "mean_steps_completed_sem": _sem(completed_steps),
+        "mean_reward": mean(rewards) if rewards else 0.0,
+        "mean_reward_sem": _sem(rewards),
+        "efficiency": mean(efficiencies) if efficiencies else 0.0,
+        "efficiency_sem": _sem(efficiencies),
+        "action_parse_rate": (float(total_valid) / float(total_actions)) if total_actions else 0.0,
+        "action_distribution": dict(action_counter),
+    }
+
+
+def evaluate_across_levels(
+    baseline: Any,
+    level_names: list[str],
+    n_episodes_per_level: int = 100,
+    seed: int = 42,
+) -> dict[str, dict[str, Any]]:
+    """Run evaluate_baseline for all level names."""
+    return {
+        level_name: evaluate_baseline(
+            baseline=baseline,
+            level_name=level_name,
+            n_episodes=n_episodes_per_level,
+            seed=seed,
+        )
+        for level_name in level_names
+    }
+
+
+def print_comparison_table(results: dict[str, dict[str, dict[str, Any]]]) -> None:
+    """Pretty-print baseline comparison by completion rate."""
+    all_levels: set[str] = set()
+    for baseline_results in results.values():
+        all_levels.update(baseline_results.keys())
+    ordered_levels = sorted(all_levels)
+    baseline_names = list(results.keys())
+
+    header = ["Level", *baseline_names]
+    row_sep = "| " + " | ".join(["---"] * len(header)) + " |"
+    print("| " + " | ".join(header) + " |")
+    print(row_sep)
+    for level in ordered_levels:
+        row = [level]
+        for baseline_name in baseline_names:
+            metrics = results.get(baseline_name, {}).get(level, {})
+            value = float(metrics.get("completion_rate", 0.0)) * 100.0
+            row.append(f"{value:.1f}%")
+        print("| " + " | ".join(row) + " |")

models.py

@@ -0,0 +1,8 @@
+"""OpenEnv models at package root for CLI/schema compatibility."""
+
+try:
+    from .env.models import MiniGridAction, MiniGridObservation, MiniGridState
+except ImportError:
+    from env.models import MiniGridAction, MiniGridObservation, MiniGridState
+
+__all__ = ["MiniGridAction", "MiniGridObservation", "MiniGridState"]

openenv.yaml

@@ -0,0 +1,7 @@
+spec_version: 1
+name: MiniGridEnv
+type: space
+runtime: fastapi
+app: server.app:app
+port: 8000
+

openenv_MiniGridEnv.egg-info/PKG-INFO

@@ -0,0 +1,15 @@
+Metadata-Version: 2.4
+Name: openenv-MiniGridEnv
+Version: 0.1.0
+Summary: MiniGrid/BabyAI text-grounded environment for OpenEnv
+Requires-Python: >=3.10
+Requires-Dist: openenv-core[core]>=0.2.0
+Requires-Dist: minigrid>=2.3.0
+Requires-Dist: gymnasium>=0.29.0
+Requires-Dist: numpy>=1.24.0
+Requires-Dist: pydantic>=2.0.0
+Requires-Dist: fastapi>=0.100.0
+Requires-Dist: uvicorn>=0.23.0
+Provides-Extra: dev
+Requires-Dist: pytest>=8.0.0; extra == "dev"
+Requires-Dist: pytest-cov>=4.0.0; extra == "dev"

openenv_MiniGridEnv.egg-info/SOURCES.txt

@@ -0,0 +1,33 @@
+README.md
+pyproject.toml
+./__init__.py
+./client.py
+./models.py
+baselines/__init__.py
+baselines/heuristic_bot.py
+baselines/llm_zeroshot.py
+baselines/random_action.py
+env/__init__.py
+env/action_parser.py
+env/config.py
+env/grid_to_text.py
+env/levels.py
+env/minigrid_env.py
+env/models.py
+env/reward.py
+eval/__init__.py
+eval/evaluate.py
+openenv_MiniGridEnv.egg-info/PKG-INFO
+openenv_MiniGridEnv.egg-info/SOURCES.txt
+openenv_MiniGridEnv.egg-info/dependency_links.txt
+openenv_MiniGridEnv.egg-info/entry_points.txt
+openenv_MiniGridEnv.egg-info/requires.txt
+openenv_MiniGridEnv.egg-info/top_level.txt
+server/__init__.py
+server/app.py
+tests/__init__.py
+tests/test_action_parser.py
+tests/test_contract.py
+tests/test_env.py
+tests/test_grid_to_text.py
+tests/test_reward.py

openenv_MiniGridEnv.egg-info/dependency_links.txt

@@ -0,0 +1 @@
+

openenv_MiniGridEnv.egg-info/entry_points.txt

@@ -0,0 +1,2 @@
+[console_scripts]
+server = MiniGridEnv.server.app:main

openenv_MiniGridEnv.egg-info/requires.txt

@@ -0,0 +1,11 @@
+openenv-core[core]>=0.2.0
+minigrid>=2.3.0
+gymnasium>=0.29.0
+numpy>=1.24.0
+pydantic>=2.0.0
+fastapi>=0.100.0
+uvicorn>=0.23.0
+
+[dev]
+pytest>=8.0.0
+pytest-cov>=4.0.0

openenv_MiniGridEnv.egg-info/top_level.txt

@@ -0,0 +1 @@
+MiniGridEnv

pyproject.toml

@@ -0,0 +1,52 @@
+# 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-MiniGridEnv"
+version = "0.1.0"
+description = "MiniGrid/BabyAI text-grounded environment for OpenEnv"
+requires-python = ">=3.10"
+dependencies = [
+    "openenv-core[core]>=0.2.0",
+    "minigrid>=2.3.0",
+    "gymnasium>=0.29.0",
+    "numpy>=1.24.0",
+    "pydantic>=2.0.0",
+    "fastapi>=0.100.0",
+    "uvicorn>=0.23.0",
+]
+
+[project.optional-dependencies]
+dev = [
+    "pytest>=8.0.0",
+    "pytest-cov>=4.0.0",
+]
+
+[project.scripts]
+server = "MiniGridEnv.server.app:main"
+
+[tool.setuptools]
+include-package-data = true
+packages = [
+    "MiniGridEnv",
+    "MiniGridEnv.server",
+    "MiniGridEnv.env",
+    "MiniGridEnv.baselines",
+    "MiniGridEnv.eval",
+    "MiniGridEnv.tests",
+]
+
+[tool.setuptools.package-dir]
+"MiniGridEnv" = "."
+"MiniGridEnv.server" = "server"
+"MiniGridEnv.env" = "env"
+"MiniGridEnv.baselines" = "baselines"
+"MiniGridEnv.eval" = "eval"
+"MiniGridEnv.tests" = "tests"

requirements.txt

@@ -0,0 +1,7 @@
+openenv-core[core]>=0.2.0
+minigrid>=2.3.0
+gymnasium>=0.29.0
+numpy>=1.24.0
+pydantic>=2.0.0
+fastapi>=0.100.0
+uvicorn>=0.23.0

server/__init__.py

@@ -0,0 +1,8 @@
+"""MiniGridEnv server package."""
+
+try:
+    from ..env.minigrid_env import MiniGridEnvironment
+except ImportError:
+    from env.minigrid_env import MiniGridEnvironment
+
+__all__ = ["MiniGridEnvironment"]

server/app.py

@@ -0,0 +1,46 @@
+"""FastAPI app exposing MiniGridEnv through OpenEnv."""
+
+try:
+    from ..env.config import EnvConfig
+    from ..env.minigrid_env import MiniGridEnvironment
+    from ..env.models import MiniGridAction, MiniGridObservation
+except ImportError:
+    from env.config import EnvConfig
+    from env.minigrid_env import MiniGridEnvironment
+    from env.models import MiniGridAction, MiniGridObservation
+
+try:
+    from openenv.core.env_server import create_app
+except ImportError:
+    create_app = None  # type: ignore
+
+
+def _env_factory() -> MiniGridEnvironment:
+    """Create a fresh env instance per client session."""
+    return MiniGridEnvironment(config=EnvConfig())
+
+
+if create_app is not None:
+    app = create_app(
+        _env_factory,
+        MiniGridAction,
+        MiniGridObservation,
+        env_name="MiniGridEnv",
+        max_concurrent_envs=256,
+    )
+else:
+    from fastapi import FastAPI
+
+    app = FastAPI(title="MiniGridEnv")
+    app.get("/health")(lambda: {"status": "ok"})
+
+
+def main() -> None:
+    """Entry point for `uv run server` or `python -m`."""
+    import uvicorn
+
+    uvicorn.run(app, host="0.0.0.0", port=8000)
+
+
+if __name__ == "__main__":
+    main()

server/requirements.txt

@@ -0,0 +1,7 @@
+openenv-core[core]>=0.2.0
+minigrid>=2.3.0
+gymnasium>=0.29.0
+numpy>=1.24.0
+pydantic>=2.0.0
+fastapi>=0.100.0
+uvicorn>=0.23.0

tests/__init__.py

@@ -0,0 +1 @@
+"""Test package for MiniGridEnv."""

tests/test_action_parser.py

@@ -0,0 +1,47 @@
+"""Unit tests for text-to-action parsing."""
+
+from MiniGridEnv.env.action_parser import parse_action
+
+
+def test_exact_match():
+    assert parse_action("go forward") == (2, "go forward", True)
+
+
+def test_case_insensitive():
+    assert parse_action("Turn Left") == (0, "turn left", True)
+
+
+def test_structured_format():
+    action_idx, canonical, valid = parse_action("Thought: I see a key.\nAction: pickup")
+    assert (action_idx, canonical, valid) == (3, "pickup", True)
+
+
+def test_alias():
+    assert parse_action("left") == (0, "turn left", True)
+
+
+def test_substring():
+    assert parse_action("I should go forward now") == (2, "go forward", True)
+
+
+def test_invalid_fallback():
+    action_idx, canonical, valid = parse_action("fly to the moon")
+    assert (action_idx, canonical, valid) == (2, "go forward", False)
+
+
+def test_all_canonical_actions():
+    expected = [
+        ("turn left", 0),
+        ("turn right", 1),
+        ("go forward", 2),
+        ("pickup", 3),
+        ("drop", 4),
+        ("toggle", 5),
+        ("done", 6),
+    ]
+    for command, index in expected:
+        assert parse_action(command)[0] == index
+
+
+def test_whitespace_handling():
+    assert parse_action("  go forward  \n") == (2, "go forward", True)

tests/test_contract.py

@@ -0,0 +1,155 @@
+"""Contract-style tests for MiniGridEnv reset/step/state behavior."""
+
+from __future__ import annotations
+
+import pytest
+
+minigrid = pytest.importorskip("minigrid")
+assert minigrid is not None
+
+from MiniGridEnv.env.config import EnvConfig
+from MiniGridEnv.env.minigrid_env import MiniGridEnvironment
+from MiniGridEnv.env.models import MiniGridAction, MiniGridObservation
+
+
+def test_reset_returns_valid_observation():
+    env = MiniGridEnvironment(config=EnvConfig(level_name="GoToRedBall"))
+    obs = env.reset(seed=123)
+    assert isinstance(obs, MiniGridObservation)
+    assert obs.step_idx == 0
+    assert obs.history == []
+    assert obs.done is False
+    assert isinstance(obs.text, str) and len(obs.text.strip()) > 0
+
+
+def test_observation_text_is_natural_language():
+    env = MiniGridEnvironment(config=EnvConfig(level_name="GoToRedBall"))
+    obs = env.reset(seed=123)
+    assert "Mission:" in obs.text
+    assert "You are facing" in obs.text
+
+
+def test_mission_is_populated():
+    env = MiniGridEnvironment(config=EnvConfig(level_name="GoToRedBall"))
+    obs = env.reset(seed=123)
+    assert isinstance(obs.mission, str) and len(obs.mission.strip()) > 0
+
+
+def test_step_accepts_valid_command():
+    env = MiniGridEnvironment(config=EnvConfig(level_name="GoToRedBall"))
+    env.reset(seed=123)
+    obs = env.step(MiniGridAction(command="go forward"))
+    assert isinstance(obs, MiniGridObservation)
+    assert obs.step_idx == 1
+
+
+def test_step_handles_invalid_command():
+    env = MiniGridEnvironment(config=EnvConfig(level_name="GoToRedBall"))
+    env.reset(seed=123)
+    obs = env.step(MiniGridAction(command="fly away"))
+    assert isinstance(obs, MiniGridObservation)
+    assert obs.last_action == "go forward"
+    assert env.state.invalid_actions == 1
+
+
+def test_episode_terminates_on_success():
+    try:
+        from minigrid.envs.babyai import BotAgent  # type: ignore
+    except Exception:
+        pytest.skip("BotAgent unavailable for success-path test")
+
+    int_to_text = {
+        0: "turn left",
+        1: "turn right",
+        2: "go forward",
+        3: "pickup",
+        4: "drop",
+        5: "toggle",
+        6: "done",
+    }
+
+    env = MiniGridEnvironment(config=EnvConfig(level_name="GoToRedBall"))
+    obs = env.reset(seed=123)
+    bot = BotAgent(env._gym_env.unwrapped)  # type: ignore[attr-defined]
+
+    while not obs.done and obs.step_idx < obs.max_steps:
+        raw_obs = env._last_obs  # type: ignore[attr-defined]
+        action = bot.act(raw_obs)
+        obs = env.step(MiniGridAction(command=int_to_text.get(int(action), "done")))
+
+    assert obs.done is True
+    assert env.state.completed is True
+    assert env.state.total_reward > 0.0
+
+
+def test_episode_truncates_on_max_steps():
+    env = MiniGridEnvironment(
+        config=EnvConfig(level_name="GoToRedBall", max_steps_override=1)
+    )
+    env.reset(seed=123)
+    obs = env.step(MiniGridAction(command="go forward"))
+    assert obs.done is True
+    assert env.state.truncated is True
+
+
+def test_deterministic_with_seed():
+    actions = ["turn left", "turn right", "go forward", "go forward"]
+
+    env_a = MiniGridEnvironment(config=EnvConfig(level_name="GoToRedBall"))
+    env_b = MiniGridEnvironment(config=EnvConfig(level_name="GoToRedBall"))
+    obs_a = env_a.reset(seed=777)
+    obs_b = env_b.reset(seed=777)
+
+    assert obs_a.text == obs_b.text
+    for command in actions:
+        if obs_a.done or obs_b.done:
+            assert obs_a.done == obs_b.done
+            break
+        obs_a = env_a.step(MiniGridAction(command=command))
+        obs_b = env_b.step(MiniGridAction(command=command))
+        assert obs_a.text == obs_b.text
+        assert obs_a.done == obs_b.done
+
+
+def test_state_tracks_metrics():
+    env = MiniGridEnvironment(config=EnvConfig(level_name="GoToRedBall"))
+    env.reset(seed=123)
+    env.step(MiniGridAction(command="go forward"))
+    state = env.state
+    assert state.steps_taken == 1
+    assert state.valid_actions == 1
+    assert state.action_distribution.get("go forward", 0) == 1
+
+
+def test_all_seven_actions_accepted():
+    commands = [
+        "turn left",
+        "turn right",
+        "go forward",
+        "pickup",
+        "drop",
+        "toggle",
+        "done",
+    ]
+    env = MiniGridEnvironment(config=EnvConfig(level_name="GoToRedBall"))
+    for idx, command in enumerate(commands):
+        env.reset(seed=100 + idx)
+        obs = env.step(MiniGridAction(command=command))
+        assert isinstance(obs, MiniGridObservation)
+
+
+def test_history_grows_each_step():
+    env = MiniGridEnvironment(config=EnvConfig(level_name="GoToRedBall"))
+    env.reset(seed=123)
+    obs = env.step(MiniGridAction(command="go forward"))
+    assert len(obs.history) == 1
+    obs = env.step(MiniGridAction(command="turn left"))
+    assert len(obs.history) == 2
+
+
+def test_different_levels_produce_different_missions():
+    env_a = MiniGridEnvironment(config=EnvConfig(level_name="GoToRedBall"))
+    env_b = MiniGridEnvironment(config=EnvConfig(level_name="GoToObj"))
+    obs_a = env_a.reset(seed=123)
+    obs_b = env_b.reset(seed=123)
+    assert obs_a.mission != obs_b.mission

tests/test_env.py

@@ -0,0 +1,52 @@
+"""Additional environment smoke tests."""
+
+from __future__ import annotations
+
+import pytest
+
+minigrid = pytest.importorskip("minigrid")
+assert minigrid is not None
+
+from MiniGridEnv.env.config import EnvConfig
+from MiniGridEnv.env.minigrid_env import MiniGridEnvironment
+from MiniGridEnv.env.models import MiniGridAction
+
+
+def test_step_before_reset_raises():
+    env = MiniGridEnvironment(config=EnvConfig(level_name="GoToRedBall"))
+    with pytest.raises(RuntimeError):
+        env.step(MiniGridAction(command="go forward"))
+
+
+def test_step_after_done_raises():
+    env = MiniGridEnvironment(
+        config=EnvConfig(level_name="GoToRedBall", max_steps_override=1)
+    )
+    env.reset(seed=1)
+    env.step(MiniGridAction(command="go forward"))
+    with pytest.raises(RuntimeError):
+        env.step(MiniGridAction(command="go forward"))
+
+
+def test_reset_creates_new_episode_id():
+    env = MiniGridEnvironment(config=EnvConfig(level_name="GoToRedBall"))
+    env.reset(seed=1)
+    episode_one = env.state.episode_id
+    env.reset(seed=2)
+    episode_two = env.state.episode_id
+    assert episode_one != episode_two
+
+
+def test_reset_honors_explicit_episode_id():
+    env = MiniGridEnvironment(config=EnvConfig(level_name="GoToRedBall"))
+    env.reset(seed=1, episode_id="episode-123")
+    assert env.state.episode_id == "episode-123"
+
+
+def test_history_can_be_disabled():
+    env = MiniGridEnvironment(
+        config=EnvConfig(level_name="GoToRedBall", include_history=False)
+    )
+    env.reset(seed=5)
+    obs = env.step(MiniGridAction(command="go forward"))
+    assert obs.history == []

tests/test_grid_to_text.py

@@ -0,0 +1,81 @@
+"""Tests for MiniGrid observation text rendering."""
+
+from types import SimpleNamespace
+
+import numpy as np
+
+from MiniGridEnv.env.grid_to_text import grid_to_text
+
+
+def _empty_obs():
+    grid = np.ones((7, 7, 3), dtype=np.int64)  # object type: empty
+    grid[:, :, 1] = 0  # default color red (unused for empties)
+    grid[:, :, 2] = 0
+    return {"image": grid, "direction": 0, "mission": "go to the red ball"}
+
+
+def test_empty_room_description():
+    text = grid_to_text(_empty_obs())
+    assert "Mission: go to the red ball" in text
+    assert "You are facing east." in text
+    assert "Notable objects: none visible." in text
+
+
+def test_object_detection():
+    obs = _empty_obs()
+    obs["image"][4, 2, 0] = 6  # ball
+    obs["image"][4, 2, 1] = 0  # red
+    text = grid_to_text(obs)
+    assert "red ball" in text
+
+
+def test_door_states():
+    obs = _empty_obs()
+    obs["image"][5, 3, 0] = 4  # door ahead
+    obs["image"][5, 3, 1] = 2  # blue
+    obs["image"][5, 3, 2] = 1  # closed
+    text = grid_to_text(obs)
+    assert "closed blue door" in text
+
+
+def test_wall_boundaries():
+    obs = _empty_obs()
+    obs["image"][5, 3, 0] = 2  # wall
+    text = grid_to_text(obs)
+    assert "Directly ahead: a wall." in text
+
+
+def test_carrying_object():
+    carrying = SimpleNamespace(type="key", color="blue")
+    text = grid_to_text(_empty_obs(), carrying=carrying)
+    assert "You are carrying: a blue key." in text
+
+
+def test_egocentric_directions():
+    obs = _empty_obs()
+    obs["image"][1, 3, 0] = 5  # key
+    obs["image"][1, 3, 1] = 1  # green
+    text = grid_to_text(obs)
+    assert "5 steps ahead" in text
+
+
+def test_multiple_objects():
+    obs = _empty_obs()
+    obs["image"][5, 2, 0] = 5  # key
+    obs["image"][5, 2, 1] = 2  # blue
+    obs["image"][4, 4, 0] = 6  # ball
+    obs["image"][4, 4, 1] = 0  # red
+    obs["image"][5, 3, 0] = 4  # door
+    obs["image"][5, 3, 1] = 4  # yellow
+    obs["image"][5, 3, 2] = 0  # open
+    text = grid_to_text(obs)
+    assert "blue key" in text
+    assert "red ball" in text
+    assert "open yellow door" in text
+
+
+def test_unseen_cells_ignored():
+    obs = _empty_obs()
+    obs["image"][1, 1, 0] = 0  # unseen
+    text = grid_to_text(obs)
+    assert "Notable objects: none visible." in text

tests/test_reward.py

@@ -0,0 +1,59 @@
+"""Reward computation tests."""
+
+from MiniGridEnv.env.config import RewardConfig
+from MiniGridEnv.env.reward import compute_step_reward
+
+
+def test_binary_completion():
+    reward, _ = compute_step_reward(
+        terminated=True,
+        truncated=False,
+        action_valid=True,
+        step_idx=5,
+        max_steps=64,
+        optimal_steps=None,
+        config=RewardConfig(mode="binary"),
+    )
+    assert reward == 1.0
+
+
+def test_binary_truncation():
+    reward, _ = compute_step_reward(
+        terminated=False,
+        truncated=True,
+        action_valid=True,
+        step_idx=64,
+        max_steps=64,
+        optimal_steps=None,
+        config=RewardConfig(mode="binary"),
+    )
+    assert reward == 0.0
+
+
+def test_shaped_penalties():
+    reward, breakdown = compute_step_reward(
+        terminated=False,
+        truncated=False,
+        action_valid=False,
+        step_idx=3,
+        max_steps=64,
+        optimal_steps=None,
+        config=RewardConfig(mode="shaped", step_penalty=-0.01, invalid_action_penalty=-0.05),
+    )
+    assert abs(reward - (-0.06)) < 1e-9
+    assert breakdown["step_penalty"] == -0.01
+    assert breakdown["invalid_action"] == -0.05
+
+
+def test_efficiency_bonus_on_success():
+    reward, breakdown = compute_step_reward(
+        terminated=True,
+        truncated=False,
+        action_valid=True,
+        step_idx=5,
+        max_steps=64,
+        optimal_steps=10,
+        config=RewardConfig(mode="efficiency", efficiency_bonus_weight=0.5),
+    )
+    assert reward > 1.0
+    assert breakdown["efficiency_bonus"] > 0.0