Add KantBench HF Space code to spaces/kant/

Pulls the live HF Space (jtowarek/KantBench) into the monorepo so it's
tracked in version control. The space is a self-contained standalone
deployment of the KantBench game-theory environment using openenv-core.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

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=KantBench
+
+# 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

@@ -0,0 +1,255 @@
+---
+title: Kantbench Environment Server
+emoji: 🕹️
+colorFrom: green
+colorTo: yellow
+sdk: docker
+pinned: false
+app_port: 8000
+base_path: /web
+tags:
+  - openenv
+---
+
+# Kantbench 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 Kantbench environment is through the `KantbenchEnv` class:
+
+```python
+from KantBench import KantbenchAction, KantbenchEnv
+
+try:
+    # Create environment from Docker image
+    KantBenchenv = KantbenchEnv.from_docker_image("KantBench-env:latest")
+
+    # Reset
+    result = KantBenchenv.reset()
+    print(f"Reset: {result.observation.echoed_message}")
+
+    # Send multiple messages
+    messages = ["Hello, World!", "Testing echo", "Final message"]
+
+    for msg in messages:
+        result = KantBenchenv.step(KantbenchAction(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
+    KantBenchenv.close()
+```
+
+That's it! The `KantbenchEnv.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 KantBench-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
+**KantbenchAction**: Contains a single field
+- `message` (str) - The message to echo back
+
+### Observation
+**KantbenchObservation**: 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 Kantbench environment server running, you can connect directly:
+
+```python
+from KantBench import KantbenchEnv
+
+# Connect to existing server
+KantBenchenv = KantbenchEnv(base_url="<ENV_HTTP_URL_HERE>")
+
+# Use as normal
+result = KantBenchenv.reset()
+result = KantBenchenv.step(KantbenchAction(message="Hello!"))
+```
+
+Note: When connecting to an existing server, `KantBenchenv.close()` will NOT stop the server.
+
+### Using the Context Manager
+
+The client supports context manager usage for automatic connection management:
+
+```python
+from KantBench import KantbenchAction, KantbenchEnv
+
+# Connect with context manager (auto-connects and closes)
+with KantbenchEnv(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(KantbenchAction(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(
+    KantbenchEnvironment,  # Pass class, not instance
+    KantbenchAction,
+    KantbenchObservation,
+    max_concurrent_envs=4,  # Allow 4 concurrent sessions
+)
+```
+
+Then multiple clients can connect simultaneously:
+
+```python
+from KantBench import KantbenchAction, KantbenchEnv
+from concurrent.futures import ThreadPoolExecutor
+
+def run_episode(client_id: int):
+    with KantbenchEnv(base_url="http://localhost:8000") as env:
+        result = env.reset()
+        for i in range(10):
+            result = env.step(KantbenchAction(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/KantBench_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
+
+```
+KantBench/
+├── .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              # KantbenchEnv client
+├── models.py              # Action and Observation models
+└── server/
+    ├── __init__.py        # Server module exports
+    ├── KantBench_environment.py  # Core environment logic
+    ├── app.py             # FastAPI application (HTTP + WebSocket endpoints)
+    └── Dockerfile         # Container image definition
+```

__init__.py

@@ -0,0 +1,16 @@
+# 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.
+
+"""Kantbench Environment."""
+
+from .client import KantbenchEnv
+from .models import KantbenchAction, KantbenchObservation
+
+__all__ = [
+    "KantbenchAction",
+    "KantbenchObservation",
+    "KantbenchEnv",
+]

client.py

@@ -0,0 +1,99 @@
+# 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.
+
+"""Kantbench Environment Client."""
+
+from typing import Dict
+
+from openenv.core.client_types import StepResult
+from openenv.core.env_server.types import State
+from openenv.core import EnvClient
+
+from .models import KantbenchAction, KantbenchObservation
+
+
+class KantbenchEnv(
+    EnvClient[KantbenchAction, KantbenchObservation]
+):
+    """
+    Client for the Kantbench 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 KantbenchEnv(base_url="http://localhost:8000") as client:
+        ...     result = client.reset()
+        ...     print(result.observation.echoed_message)
+        ...
+        ...     result = client.step(KantbenchAction(message="Hello!"))
+        ...     print(result.observation.echoed_message)
+
+    Example with Docker:
+        >>> # Automatically start container and connect
+        >>> client = KantbenchEnv.from_docker_image("KantBench-env:latest")
+        >>> try:
+        ...     result = client.reset()
+        ...     result = client.step(KantbenchAction(message="Test"))
+        ... finally:
+        ...     client.close()
+    """
+
+    def _step_payload(self, action: KantbenchAction) -> Dict:
+        """
+        Convert KantbenchAction to JSON payload for step message.
+
+        Args:
+            action: KantbenchAction instance
+
+        Returns:
+            Dictionary representation suitable for JSON encoding
+        """
+        return {
+            "message": action.message,
+        }
+
+    def _parse_result(self, payload: Dict) -> StepResult[KantbenchObservation]:
+        """
+        Parse server response into StepResult[KantbenchObservation].
+
+        Args:
+            payload: JSON response data from server
+
+        Returns:
+            StepResult with KantbenchObservation
+        """
+        obs_data = payload.get("observation", {})
+        observation = KantbenchObservation(
+            echoed_message=obs_data.get("echoed_message", ""),
+            message_length=obs_data.get("message_length", 0),
+            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 request
+
+        Returns:
+            State object with episode_id and step_count
+        """
+        return State(
+            episode_id=payload.get("episode_id"),
+            step_count=payload.get("step_count", 0),
+        )

models.py

@@ -0,0 +1,29 @@
+"""Data models for the KantBench game theory environment."""
+
+from typing import Any
+from pydantic import Field
+from openenv.core.env_server.types import Action, Observation
+
+
+class KantBenchAction(Action):
+    """Action for the KantBench environment — a move in a 2-player game."""
+
+    move: str = Field(..., description="Your move (e.g. 'cooperate', 'defect', 'hawk', 'dove')")
+
+
+class KantBenchObservation(Observation):
+    """Observation from the KantBench environment after one round."""
+
+    game_name: str = Field(default="", description="Name of the current game")
+    game_description: str = Field(default="", description="Description of the game")
+    available_moves: list[str] = Field(default_factory=list, description="Valid moves for this game")
+    your_move: str = Field(default="", description="Your move this round")
+    opponent_move: str = Field(default="", description="Opponent's move this round")
+    your_payoff: float = Field(default=0.0, description="Your payoff this round")
+    opponent_payoff: float = Field(default=0.0, description="Opponent's payoff this round")
+    cumulative_score: float = Field(default=0.0, description="Your total score so far")
+    round_number: int = Field(default=0, description="Current round number")
+    max_rounds: int = Field(default=10, description="Total rounds in this episode")
+    opponent_strategy: str = Field(default="", description="Opponent's strategy name")
+    history: list[dict[str, Any]] = Field(default_factory=list, description="Round history")
+    message: str = Field(default="", description="Status message")

openenv.yaml

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

pyproject.toml

@@ -0,0 +1,45 @@
+# 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-KantBench"
+version = "0.1.0"
+description = "Kantbench environment for OpenEnv"
+requires-python = ">=3.10"
+dependencies = [
+    # Core OpenEnv runtime (provides FastAPI server + HTTP client types)
+    # install from github
+    # "openenv-core[core] @ git+https://github.com/meta-pytorch/OpenEnv.git",
+    "openenv-core[core]>=0.2.0",
+    # Environment-specific dependencies
+    # Add all dependencies needed for your environment here
+    # Examples:
+    # "numpy>=1.19.0",
+    # "torch>=2.0.0",
+    # "gymnasium>=0.29.0",
+    # "openspiel>=1.0.0",
+    # "smolagents>=1.22.0,<2",
+]
+
+[project.optional-dependencies]
+dev = [
+    "pytest>=8.0.0",
+    "pytest-cov>=4.0.0",
+]
+
+[project.scripts]
+# Server entry point - enables running via: uv run --project . server
+# or: python -m KantBench.server.app
+server = "KantBench.server.app:main"
+
+[tool.setuptools]
+include-package-data = true
+packages = ["KantBench", "KantBench.server"]
+package-dir = { "KantBench" = ".", "KantBench.server" = "server" }

server/KantBench_environment.py

@@ -0,0 +1,289 @@
+"""KantBench: a game theory RL environment for OpenEnv.
+
+Each episode is one repeated game (e.g. Prisoner's Dilemma) against a
+fixed strategy opponent.  The agent chooses a move each round; the
+environment computes payoffs and returns a structured observation.
+
+Supported games: Prisoner's Dilemma, Stag Hunt, Hawk-Dove,
+                 Battle of Sexes, Chicken, Matching Pennies,
+                 Rock-Paper-Scissors.
+Opponent strategies: random, always_first, always_last, tit_for_tat,
+                     grim_trigger, pavlov.
+"""
+
+from __future__ import annotations
+
+import random
+from typing import Any
+from uuid import uuid4
+
+from openenv.core.env_server.interfaces import Environment
+from openenv.core.env_server.types import State
+
+from models import KantBenchAction, KantBenchObservation
+
+# ---------------------------------------------------------------------------
+# Game definitions (self-contained payoff matrices)
+# ---------------------------------------------------------------------------
+
+def _matrix(m: dict[tuple[str, str], tuple[float, float]]):
+    """Return a payoff function from a matrix dict."""
+    def fn(a: str, b: str) -> tuple[float, float]:
+        return m[(a, b)]
+    return fn
+
+
+GAMES: dict[str, dict[str, Any]] = {
+    "prisoners_dilemma": {
+        "name": "Prisoner's Dilemma",
+        "description": (
+            "Two players choose to cooperate or defect simultaneously. "
+            "Mutual cooperation is best collectively; defection is individually tempting."
+        ),
+        "actions": ["cooperate", "defect"],
+        "rounds": 10,
+        "payoff_fn": _matrix({
+            ("cooperate", "cooperate"): (3.0, 3.0),
+            ("cooperate", "defect"):    (0.0, 5.0),
+            ("defect",    "cooperate"): (5.0, 0.0),
+            ("defect",    "defect"):    (1.0, 1.0),
+        }),
+    },
+    "stag_hunt": {
+        "name": "Stag Hunt",
+        "description": (
+            "Two hunters choose to hunt stag (requires coordination) or hare "
+            "(safe alone). Mutual cooperation yields the best outcome."
+        ),
+        "actions": ["stag", "hare"],
+        "rounds": 10,
+        "payoff_fn": _matrix({
+            ("stag", "stag"): (4.0, 4.0),
+            ("stag", "hare"): (0.0, 2.0),
+            ("hare", "stag"): (2.0, 0.0),
+            ("hare", "hare"): (2.0, 2.0),
+        }),
+    },
+    "hawk_dove": {
+        "name": "Hawk-Dove",
+        "description": (
+            "Two players compete over a resource. Hawk is aggressive; Dove is passive. "
+            "Two hawks fight and both lose; two doves share."
+        ),
+        "actions": ["hawk", "dove"],
+        "rounds": 10,
+        "payoff_fn": _matrix({
+            ("hawk", "hawk"): (-1.0, -1.0),
+            ("hawk", "dove"): (4.0,   0.0),
+            ("dove", "hawk"): (0.0,   4.0),
+            ("dove", "dove"): (2.0,   2.0),
+        }),
+    },
+    "battle_of_sexes": {
+        "name": "Battle of the Sexes",
+        "description": (
+            "Two players want to coordinate but prefer different options. "
+            "Player 1 prefers opera; Player 2 prefers football. "
+            "Both prefer to be together over going alone."
+        ),
+        "actions": ["opera", "football"],
+        "rounds": 10,
+        "payoff_fn": _matrix({
+            ("opera",    "opera"):    (3.0, 1.0),
+            ("opera",    "football"): (0.0, 0.0),
+            ("football", "opera"):    (0.0, 0.0),
+            ("football", "football"): (1.0, 3.0),
+        }),
+    },
+    "chicken": {
+        "name": "Chicken (Snowdrift)",
+        "description": (
+            "Two drivers head toward each other. Swerving is safe but cowardly; "
+            "going straight is bold but catastrophic if both do it."
+        ),
+        "actions": ["straight", "swerve"],
+        "rounds": 10,
+        "payoff_fn": _matrix({
+            ("straight", "straight"): (-10.0, -10.0),
+            ("straight", "swerve"):   (5.0,   -1.0),
+            ("swerve",   "straight"): (-1.0,   5.0),
+            ("swerve",   "swerve"):   (0.0,    0.0),
+        }),
+    },
+    "matching_pennies": {
+        "name": "Matching Pennies",
+        "description": (
+            "Player 1 wins if both show the same side; Player 2 wins if they differ. "
+            "Pure zero-sum game with no stable pure-strategy Nash equilibrium."
+        ),
+        "actions": ["heads", "tails"],
+        "rounds": 20,
+        "payoff_fn": _matrix({
+            ("heads", "heads"): (1.0,  -1.0),
+            ("heads", "tails"): (-1.0,  1.0),
+            ("tails", "heads"): (-1.0,  1.0),
+            ("tails", "tails"): (1.0,  -1.0),
+        }),
+    },
+    "rock_paper_scissors": {
+        "name": "Rock-Paper-Scissors",
+        "description": (
+            "Classic zero-sum game. Rock beats Scissors, Scissors beats Paper, "
+            "Paper beats Rock. Ties yield 0."
+        ),
+        "actions": ["rock", "paper", "scissors"],
+        "rounds": 20,
+        "payoff_fn": _matrix({
+            ("rock",     "rock"):     (0.0,  0.0),
+            ("rock",     "paper"):    (-1.0, 1.0),
+            ("rock",     "scissors"): (1.0, -1.0),
+            ("paper",    "rock"):     (1.0, -1.0),
+            ("paper",    "paper"):    (0.0,  0.0),
+            ("paper",    "scissors"): (-1.0, 1.0),
+            ("scissors", "rock"):     (-1.0, 1.0),
+            ("scissors", "paper"):    (1.0, -1.0),
+            ("scissors", "scissors"): (0.0,  0.0),
+        }),
+    },
+}
+
+STRATEGIES = ["random", "always_first", "always_last", "tit_for_tat", "grim_trigger", "pavlov"]
+
+
+def _opponent_move(strategy: str, actions: list[str], history: list[dict]) -> str:
+    """Compute opponent's move given strategy and history."""
+    if strategy == "random":
+        return random.choice(actions)
+    if strategy == "always_first":
+        return actions[0]
+    if strategy == "always_last":
+        return actions[-1]
+    if not history:
+        return actions[0]  # default opening for reactive strategies
+    last_agent_move = history[-1]["your_move"]
+    last_opp_move   = history[-1]["opponent_move"]
+    if strategy == "tit_for_tat":
+        return last_agent_move if last_agent_move in actions else actions[0]
+    if strategy == "grim_trigger":
+        # Defect forever once agent defects; cooperate otherwise
+        ever_defected = any(r["your_move"] == actions[-1] for r in history)
+        return actions[-1] if ever_defected else actions[0]
+    if strategy == "pavlov":
+        # Repeat own last move if it paid well (i.e. opponent cooperated), else switch
+        if last_opp_move == actions[0]:
+            return last_opp_move  # mirror cooperation
+        return actions[0] if last_opp_move != actions[0] else actions[-1]
+    return random.choice(actions)
+
+
+# ---------------------------------------------------------------------------
+# Environment
+# ---------------------------------------------------------------------------
+
+class KantbenchEnvironment(Environment):
+    """Game theory environment for benchmarking LLM strategic reasoning.
+
+    Each episode is a repeated 2-player game against one of six opponent
+    strategies.  The agent submits a move each round and receives the payoff
+    result as a structured observation.
+
+    Example::
+
+        env = KantBenchEnvironment()
+        obs = env.reset()
+        # obs.game_name == "Prisoner's Dilemma"
+        # obs.available_moves == ["cooperate", "defect"]
+
+        obs = env.step(KantBenchAction(move="cooperate"))
+        # obs.your_payoff, obs.opponent_move, obs.cumulative_score, ...
+    """
+
+    SUPPORTS_CONCURRENT_SESSIONS: bool = True
+
+    def __init__(self):
+        self._state = State(episode_id=str(uuid4()), step_count=0)
+        self._game_key: str = "prisoners_dilemma"
+        self._strategy: str = "random"
+        self._history: list[dict] = []
+        self._cumulative_score: float = 0.0
+
+    def reset(self, **kwargs) -> KantBenchObservation:
+        """Start a new episode with a randomly chosen game and opponent strategy."""
+        self._game_key = random.choice(list(GAMES.keys()))
+        self._strategy = random.choice(STRATEGIES)
+        self._history = []
+        self._cumulative_score = 0.0
+        self._state = State(episode_id=str(uuid4()), step_count=0)
+
+        game = GAMES[self._game_key]
+        return KantBenchObservation(
+            game_name=game["name"],
+            game_description=game["description"],
+            available_moves=game["actions"],
+            your_move="",
+            opponent_move="",
+            your_payoff=0.0,
+            opponent_payoff=0.0,
+            cumulative_score=0.0,
+            round_number=0,
+            max_rounds=game["rounds"],
+            opponent_strategy=self._strategy,
+            history=[],
+            done=False,
+            reward=0.0,
+            message=(
+                f"New episode: {game['name']} vs {self._strategy}. "
+                f"Choose one of: {game['actions']}"
+            ),
+        )
+
+    def step(self, action: KantBenchAction, **kwargs) -> KantBenchObservation:  # type: ignore[override]
+        """Play one round of the current game."""
+        game = GAMES[self._game_key]
+        actions = game["actions"]
+        max_rounds = game["rounds"]
+
+        # Validate move
+        move = action.move.lower().strip()
+        if move not in actions:
+            closest = actions[0]
+            move = closest
+
+        opp_move = _opponent_move(self._strategy, actions, self._history)
+        your_pay, opp_pay = game["payoff_fn"](move, opp_move)
+
+        self._state.step_count += 1
+        self._cumulative_score += your_pay
+
+        round_record = {
+            "round": self._state.step_count,
+            "your_move": move,
+            "opponent_move": opp_move,
+            "your_payoff": your_pay,
+            "opponent_payoff": opp_pay,
+        }
+        self._history.append(round_record)
+
+        done = self._state.step_count >= max_rounds
+
+        return KantBenchObservation(
+            game_name=game["name"],
+            game_description=game["description"],
+            available_moves=actions,
+            your_move=move,
+            opponent_move=opp_move,
+            your_payoff=your_pay,
+            opponent_payoff=opp_pay,
+            cumulative_score=self._cumulative_score,
+            round_number=self._state.step_count,
+            max_rounds=max_rounds,
+            opponent_strategy=self._strategy,
+            history=self._history,
+            done=done,
+            reward=your_pay,
+            message="Game over — call reset() to start a new episode." if done else "",
+        )
+
+    @property
+    def state(self) -> State:
+        return self._state

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.
+
+"""Kantbench environment server components."""
+
+from .KantBench_environment import KantbenchEnvironment
+
+__all__ = ["KantbenchEnvironment"]

server/app.py

@@ -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.
+
+"""
+FastAPI application for the Kantbench Environment.
+
+This module creates an HTTP server that exposes the KantbenchEnvironment
+over HTTP and WebSocket endpoints, compatible with EnvClient.
+
+Endpoints:
+    - POST /reset: Reset the environment
+    - POST /step: Execute an action
+    - GET /state: Get current environment state
+    - GET /schema: Get action/observation schemas
+    - WS /ws: WebSocket endpoint for persistent sessions
+
+Usage:
+    # Development (with auto-reload):
+    uvicorn server.app:app --reload --host 0.0.0.0 --port 8000
+
+    # Production:
+    uvicorn server.app:app --host 0.0.0.0 --port 8000 --workers 4
+
+    # Or run directly:
+    python -m server.app
+"""
+
+try:
+    from openenv.core.env_server.http_server import create_app
+except Exception as e:  # pragma: no cover
+    raise ImportError(
+        "openenv is required for the web interface. Install dependencies with '\n    uv sync\n'"
+    ) from e
+
+# Import from local models.py (PYTHONPATH includes /app/env in Docker)
+from models import KantBenchAction, KantBenchObservation
+from .KantBench_environment import KantbenchEnvironment
+
+
+# Create the app with web interface and README integration
+app = create_app(
+    KantbenchEnvironment,
+    KantBenchAction,
+    KantBenchObservation,
+    env_name="KantBench",
+    max_concurrent_envs=1,  # increase this number to allow more concurrent WebSocket sessions
+)
+
+
+def main(host: str = "0.0.0.0", port: int = 8000):
+    """
+    Entry point for direct execution via uv run or python -m.
+
+    This function enables running the server without Docker:
+        uv run --project . server
+        uv run --project . server --port 8001
+        python -m KantBench.server.app
+
+    Args:
+        host: Host address to bind to (default: "0.0.0.0")
+        port: Port number to listen on (default: 8000)
+
+    For production deployments, consider using uvicorn directly with
+    multiple workers:
+        uvicorn KantBench.server.app:app --workers 4
+    """
+    import uvicorn
+
+    uvicorn.run(app, host=host, port=port)
+
+
+if __name__ == "__main__":
+    import argparse
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--port", type=int, default=8000)
+    args = parser.parse_args()
+    main(port=args.port)

server/requirements.txt

@@ -0,0 +1,6 @@
+openenv[core]>=0.2.0
+fastapi>=0.115.0
+uvicorn>=0.24.0
+
+
+