diff --git a/Dockerfile b/Dockerfile
index 0d9d715..4587714 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -24,4 +24,4 @@ ENV PORT 8880
 EXPOSE 8880
 
 # Define the command to run your app using CMD which defines your runtime
-CMD ["sh", "-c", "python rose/main.py --listen 0.0.0.0 --track ${TRACK} --port ${PORT}"]
+CMD ["sh", "-c", "python main.py --listen 0.0.0.0 --track ${TRACK} --port ${PORT}"]
diff --git a/README.md b/README.md
index 7e3ba42..9f7517e 100644
--- a/README.md
+++ b/README.md
@@ -129,3 +129,41 @@ The response in `JSON` format should include the car name and the recommended ac
   }
 }
 ```
+
+## Batch simulation (headless automation)
+
+`simulate.py` runs many full games back-to-back between two drivers, without a
+websocket/UI and without the live game's rate throttle, then writes aggregated
+win/loss/tie stats to a JSON file. This is the same HTTP driver contract used by
+the live engine (see `Testing your driver` above), so it works against any real
+`rose-game-ai` driver process, community driver, or your own `mydriver.py`.
+
+```bash
+# Start two drivers first, e.g.:
+#   (cd ../rose-game-ai && python main.py --driver mydriver.py --port 8081)
+#   (cd ../rose-game-ai && python main.py --driver examples/driver.py --port 8082)
+
+python simulate.py \
+  --drivers http://127.0.0.1:8081 http://127.0.0.1:8082 \
+  --games 50 \
+  --track random \
+  --output batch_stats.json
+```
+
+`batch_stats.json` contains per-driver `wins`/`losses`/`ties`/`avg_score`, plus a
+`per_game` breakdown:
+
+```json
+{
+  "games": 50,
+  "track_type": "random",
+  "drivers": ["http://127.0.0.1:8081", "http://127.0.0.1:8082"],
+  "results": {
+    "DriverA": {"wins": 27, "losses": 21, "ties": 2, "avg_score": 612.4},
+    "DriverB": {"wins": 21, "losses": 27, "ties": 2, "avg_score": 588.9}
+  },
+  "per_game": [{"scores": {"DriverA": 620, "DriverB": 590}, "winner": "DriverA"}]
+}
+```
+
+Run `python simulate.py --help` for all options.
diff --git a/rose/engine/logic.py b/rose/engine/logic.py
index 9f92aba..296a8ee 100644
--- a/rose/engine/logic.py
+++ b/rose/engine/logic.py
@@ -1,6 +1,7 @@
 import asyncio
 import logging
 import random
+import time
 import aiohttp
 
 from rose.engine import config
@@ -14,10 +15,17 @@
 
 
 async def initialize_game(state):
-    """Reset game settings and return re-initialized track and players."""
+    """Reset game settings and return re-initialized track and players.
+
+    Does NOT touch state["running"] — the game loop only polls the reset
+    flag periodically, so clobbering running here would silently drop a
+    Run request issued shortly after Reset, before this had a chance to run.
+    Callers that want reset to also stop the game must set running=0
+    themselves, atomically with the reset request.
+    """
     state["reset"] = None
-    state["running"] = 0
     state["timeleft"] = config.game_duration
+    state["game_started_at"] = time.monotonic()
     track = initialize_track(state["track_type"] != "same")
     players = await initialize_players(state["drivers"])
     return track, players
@@ -74,13 +82,23 @@ async def initialize_players(drivers):
     return players
 
 
-async def game_loop(state, active_websockets):
+def determine_winner(players):
+    """Return the name of the highest-scoring player, or None if tied/empty."""
+    if not players:
+        return None
+    best_score = max(player.score for player in players)
+    leaders = [player.name for player in players if player.score == best_score]
+    return leaders[0] if len(leaders) == 1 else None
+
+
+async def game_loop(state, active_websockets, telemetry=None):
     """
     Asynchronously execute the game loop, using provided state and active websockets.
 
     Args:
         state (dict): Dictionary containing game state data (rate, running status, time left, etc.).
         active_websockets (set): A set of active websocket connections for communication.
+        telemetry (TelemetryObserver, optional): observer notified on each tick. Defaults to None.
 
     Returns:
         None
@@ -103,7 +121,7 @@ async def game_loop(state, active_websockets):
         if state["running"] == 1:
             # Start executing a step in the game
             task = asyncio.create_task(
-                game_step(state, players, track, active_websockets)
+                game_step(state, players, track, active_websockets, telemetry)
             )
 
             # Pause the game loop for a specified duration, based on the rate defined in the state
@@ -121,7 +139,33 @@ async def game_loop(state, active_websockets):
             await asyncio.sleep(1)
 
 
-async def game_step(state, players, track, active_websockets):
+async def play_tick(players, track, telemetry=None, step_index=None):
+    """
+    Execute a single tick of game logic: fetch drivers' actions, advance the
+    track, and score the resulting player actions. Shared by the live
+    websocket-driven game loop and headless batch simulation.
+
+    Args:
+        players (list): List of Player objects.
+        track (Track): the game track.
+        telemetry (TelemetryObserver, optional): observer notified after scoring.
+        step_index (int, optional): tick counter passed through to the observer.
+    """
+
+    # Fetch players actions using an asynchronous HTTP session
+    await net.fetch_drivers_actions(players, track.matrix())
+
+    # Update track
+    track.update()
+
+    # Process the actions of the players
+    score.process(players, track)
+
+    if telemetry is not None:
+        telemetry.on_step(step_index, players, track)
+
+
+async def game_step(state, players, track, active_websockets, telemetry=None):
     """
     Execute a game step: Update the track, fetch drivers' actions, process actions, and update websockets.
 
@@ -130,17 +174,13 @@ async def game_step(state, players, track, active_websockets):
         players (list): List of Player objects.
         track (Track): the game track.
         active_websockets (Any): Active websockets for communication (assuming a suitable data structure).
+        telemetry (TelemetryObserver, optional): observer notified after scoring.
     """
 
     try:
-        # Fetch players actions using an asynchronous HTTP session
-        await net.fetch_drivers_actions(players, track.matrix())
-
-        # Update track
-        track.update()
-
-        # Process the actions of the players
-        score.process(players, track)
+        await play_tick(
+            players, track, telemetry, config.game_duration - state["timeleft"]
+        )
 
         # Send data to all WebSocket connections
         await net.update_websockets(True, state, players, track, active_websockets)
@@ -148,6 +188,17 @@ async def game_step(state, players, track, active_websockets):
         # Progress the game's timer
         state["timeleft"] -= 1
 
+        if telemetry is not None and state["timeleft"] < 1:
+            result = {
+                "scores": {player.name: player.score for player in players},
+                "winner": determine_winner(players),
+                "players": {player.name: player.state() for player in players},
+                "duration_seconds": round(
+                    time.monotonic() - state["game_started_at"], 2
+                ),
+            }
+            telemetry.on_game_end(players, result)
+
     except asyncio.CancelledError:
         log.info("Game step was canceled!")
         raise
diff --git a/rose/engine/player.py b/rose/engine/player.py
index 208dc0e..3ccae82 100644
--- a/rose/engine/player.py
+++ b/rose/engine/player.py
@@ -36,6 +36,9 @@ def __init__(self, name, car, lane):
         self.pickups = None
         self.misses = None
         self.hits = None
+        self.wall_hits = None
+        self.water_hits = None
+        self.crack_hits = None
         self.breaks = None
         self.jumps = None
         self.collisions = None
@@ -50,6 +53,9 @@ def reset(self):
         self.pickups = 0
         self.misses = 0
         self.hits = 0
+        self.wall_hits = 0
+        self.water_hits = 0
+        self.crack_hits = 0
         self.breaks = 0
         self.collisions = 0
         self.jumps = 0
@@ -81,6 +87,9 @@ def state(self):
             "pickups": self.pickups,
             "misses": self.misses,
             "hits": self.hits,
+            "wall_hits": self.wall_hits,
+            "water_hits": self.water_hits,
+            "crack_hits": self.crack_hits,
             "breaks": self.breaks,
             "jumps": self.jumps,
             "collisions": self.collisions,
diff --git a/rose/engine/score.py b/rose/engine/score.py
index 1936432..10889e5 100644
--- a/rose/engine/score.py
+++ b/rose/engine/score.py
@@ -70,6 +70,7 @@ def process(players, track):
             player.y += 1
             player.score += config.score_move_backward
             player.hits += 1
+            player.wall_hits += 1
 
             log.debug(
                 "player %s hit %s: lost %d points, moved back to %d,%d",
@@ -99,6 +100,7 @@ def process(players, track):
                 player.y += 1
                 player.score += config.score_move_backward
                 player.hits += 1
+                player.crack_hits += 1
 
                 log.debug(
                     "player %s hit %s: lost %d points, moved back to %d,%d",
@@ -128,6 +130,7 @@ def process(players, track):
                 player.y += 1
                 player.score += config.score_move_backward
                 player.hits += 1
+                player.water_hits += 1
 
                 log.debug(
                     "player %s hit %s: lost %d points, moved back to %d,%d",
@@ -155,6 +158,7 @@ def process(players, track):
             else:
                 # Move forward leaving the obstacle on the track
                 player.score += config.score_move_forward
+                player.misses += 1
 
                 log.debug("player %s missed %s", player.name, obstacle)
 
diff --git a/rose/engine/server.py b/rose/engine/server.py
index cb9f09b..4c68181 100644
--- a/rose/engine/server.py
+++ b/rose/engine/server.py
@@ -1,10 +1,17 @@
+import asyncio
 import json
+import logging
+import uuid
 
 import aiohttp
 from aiohttp import web
 
 from rose.engine import config
 from rose.engine import logic
+from rose.engine import simulate
+from rose.telemetry.sinks import LiveSink
+
+log = logging.getLogger("server")
 
 # Global active_websockets
 # IMPORTANT - shared with game loop in game.py
@@ -14,6 +21,13 @@
 # IMPORTANT - shared with game loop in game.py
 state = {"rate": None, "running": None, "reset": None, "drivers": [], "timeleft": None}
 
+# Global telemetry sink, fed by the game loop, read by the /telemetry route.
+telemetry = LiveSink()
+
+# Global batch-simulation jobs, keyed by job id.
+# {"status": "running" | "done" | "error", "result": dict or None, "error": str or None}
+simulation_jobs = {}
+
 
 async def admin_handler(request):
     """
@@ -25,8 +39,6 @@ async def admin_handler(request):
     Returns:
         aiohttp.web.Response: A response indicating the new game rate or an error message.
     """
-    global state
-
     rate = request.rel_url.query.get("rate")
     if rate:
         try:
@@ -47,6 +59,11 @@ async def admin_handler(request):
             state["reset"] = int(reset)
         except ValueError:
             return web.Response(text="Invalid reset provided", status=400)
+        # Stop the game on reset, unless this same request also asked to run
+        # (formerly done as a side effect inside initialize_game(), which
+        # raced with a Run request issued shortly after Reset).
+        if running is None:
+            state["running"] = 0
 
     # This expects the drivers to be passed as a comma-separated list in the query param
     # e.g., ?drivers=http://localhost:8081/drv2,http://driver.com:8090/
@@ -60,6 +77,117 @@ async def admin_handler(request):
     return web.Response(text=json.dumps(state))
 
 
+def _telemetry_snapshot():
+    snapshot = telemetry.snapshot()
+    snapshot["running"] = bool(state.get("running"))
+    return snapshot
+
+
+async def telemetry_handler(request):
+    """Return recent per-tick history and recently finished game results."""
+    return web.json_response(_telemetry_snapshot())
+
+
+async def telemetry_clear_handler(request):
+    """Clear the displayed tick log and match history."""
+    telemetry.clear()
+    return web.json_response(_telemetry_snapshot())
+
+
+async def simulate_start_handler(request):
+    """
+    Start a batch simulation as a background job. Each round is driven
+    through the same shared game state/websockets as manual play, so it
+    plays out live on the main game screen, same as a human clicking
+    Reset then Run would see, one round after another.
+
+    Expected POST JSON body:
+        {"drivers": [url1, url2], "games": <int>, "track": "same"|"random"}
+
+    Returns:
+        {"job_id": <str>} with status 202, so the caller can poll
+        GET /simulate/{job_id} for progress/results.
+    """
+    try:
+        payload = await request.json()
+    except json.JSONDecodeError:
+        return web.Response(text="Invalid JSON body", status=400)
+
+    drivers = payload.get("drivers")
+    if not isinstance(drivers, list) or len(drivers) != 2:
+        return web.Response(text="Exactly 2 driver URLs are required", status=400)
+
+    try:
+        games = int(payload.get("games", 10))
+    except (TypeError, ValueError):
+        return web.Response(text="Invalid games value", status=400)
+
+    track_type = payload.get("track", "random")
+    if track_type not in ("same", "random"):
+        return web.Response(text="track must be 'same' or 'random'", status=400)
+
+    job_id = uuid.uuid4().hex
+    simulation_jobs[job_id] = {"status": "running", "result": None, "error": None}
+    asyncio.create_task(_run_simulation_job(job_id, drivers, games, track_type))
+
+    return web.json_response({"job_id": job_id}, status=202)
+
+
+ROUND_POLL_INTERVAL_S = 0.2
+INTER_ROUND_PAUSE_S = 2
+
+
+async def _run_live_batch(drivers, games, track_type):
+    """
+    Play `games` rounds back-to-back through the shared live game loop, so
+    each one is visible on the main game screen (same state/websockets a
+    human's Reset+Run drives), and collect their results as they finish.
+
+    Relies on the game loop's own reset/running handling (see
+    logic.game_loop / logic.initialize_game): setting reset=1 and running=1
+    together starts a fresh round immediately, and telemetry.on_game_end
+    fires once it's done, which is what we poll for below.
+    """
+    per_game = []
+    for game_index in range(games):
+        finished_before = telemetry.result_count()
+
+        state["drivers"] = list(drivers)
+        state["track_type"] = track_type
+        state["reset"] = 1
+        state["running"] = 1
+
+        while telemetry.result_count() == finished_before:
+            await asyncio.sleep(ROUND_POLL_INTERVAL_S)
+
+        result = telemetry.latest_result()
+        per_game.append(result)
+        log.info("live batch game %d/%d: %s", game_index + 1, games, result)
+
+        if game_index < games - 1:
+            await asyncio.sleep(INTER_ROUND_PAUSE_S)
+
+    return simulate._aggregate(drivers, games, track_type, per_game)
+
+
+async def _run_simulation_job(job_id, drivers, games, track_type):
+    """Run a batch simulation and store its outcome for simulate_status_handler."""
+    try:
+        result = await _run_live_batch(drivers, games, track_type)
+        simulation_jobs[job_id] = {"status": "done", "result": result, "error": None}
+    except Exception as e:
+        simulation_jobs[job_id] = {"status": "error", "result": None, "error": str(e)}
+
+
+async def simulate_status_handler(request):
+    """Return the status/result of a batch-simulation job started via POST /simulate."""
+    job = simulation_jobs.get(request.match_info["job_id"])
+    if job is None:
+        return web.Response(text="Unknown job id", status=404)
+
+    return web.json_response(job)
+
+
 async def websocket_handler(request):
     """
     Handle WebSocket connections, echoing received messages with a prefix.
@@ -110,8 +238,6 @@ async def run(
         theme (str): Path to the static them resources directory.
         track_type (str): Type of track can be "random" or "same".
     """
-    global state
-
     state["rate"] = initial_rate
     state["running"] = 1 if initial_running else 0
     state["drivers"] = initial_drivers
@@ -123,6 +249,10 @@ async def run(
     # Add application routes
     app.router.add_get("/ws", websocket_handler)
     app.router.add_post("/admin", admin_handler)
+    app.router.add_get("/telemetry", telemetry_handler)
+    app.router.add_post("/telemetry/clear", telemetry_clear_handler)
+    app.router.add_post("/simulate", simulate_start_handler)
+    app.router.add_get("/simulate/{job_id}", simulate_status_handler)
 
     runner = aiohttp.web.AppRunner(app)
     await runner.setup()
@@ -138,4 +268,4 @@ async def run(
 
     # Start game loop
     # IMPORTANT: state and active_websockets are references, changes in this file will affect the game loop.
-    await logic.game_loop(state, active_websockets)
+    await logic.game_loop(state, active_websockets, telemetry)
diff --git a/rose/engine/simulate.py b/rose/engine/simulate.py
new file mode 100644
index 0000000..97543b5
--- /dev/null
+++ b/rose/engine/simulate.py
@@ -0,0 +1,114 @@
+"""Headless batch simulation.
+
+Runs full games against real rose-game-ai driver HTTP servers (the same
+GET/POST contract used by the live engine, see rose.engine.net), but without
+websockets and without the wall-clock throttle used by the live game loop, so
+games run back-to-back as fast as the drivers respond. Useful for automated
+evaluation of drivers (e.g. CI, tournaments) and for feeding a telemetry sink.
+"""
+
+import logging
+import time
+
+from rose.engine import config
+from rose.engine import logic
+from rose.engine.track import Track
+
+log = logging.getLogger("simulate")
+
+
+async def run_single_game(drivers, track_type, telemetry=None):
+    """
+    Play one full headless game between the given drivers.
+
+    Args:
+        drivers (list of str): driver URLs, one per player.
+        track_type (str): "same" or "random", see Track.
+        telemetry (TelemetryObserver, optional): observer notified per tick.
+
+    Returns:
+        dict: {"scores": {driver_name: score, ...}, "winner": name_or_None,
+            "players": {driver_name: Player.state(), ...},
+            "duration_seconds": float}.
+
+    Raises:
+        RuntimeError: if any driver failed to respond during initialization.
+    """
+    track = Track(track_type != "same")
+    track.reset()
+
+    players = await logic.initialize_players(drivers)
+    if len(players) != len(drivers):
+        raise RuntimeError(
+            f"Only {len(players)}/{len(drivers)} drivers responded to initialization"
+        )
+
+    if telemetry is not None:
+        telemetry.on_game_start(track, players, {"track_type": track_type})
+
+    started_at = time.monotonic()
+    for step_index in range(config.game_duration):
+        await logic.play_tick(players, track, telemetry, step_index)
+
+    result = {
+        "scores": {player.name: player.score for player in players},
+        "winner": logic.determine_winner(players),
+        "players": {player.name: player.state() for player in players},
+        "duration_seconds": round(time.monotonic() - started_at, 2),
+    }
+
+    if telemetry is not None:
+        telemetry.on_game_end(players, result)
+
+    return result
+
+
+async def run_batch(drivers, games, track_type, telemetry=None):
+    """
+    Play `games` consecutive headless games between the given drivers and
+    aggregate win/loss/tie stats per driver.
+
+    Games run sequentially: rose-game-ai's reference server is a blocking
+    socketserver.TCPServer, so concurrent games would just queue on it anyway.
+
+    Returns:
+        dict: aggregated stats, see rose.engine.simulate module docstring
+            for shape, or the README for an example.
+    """
+    per_game = []
+    for game_index in range(games):
+        result = await run_single_game(drivers, track_type, telemetry)
+        per_game.append(result)
+        log.info("game %d/%d: %s", game_index + 1, games, result)
+
+    return _aggregate(drivers, games, track_type, per_game)
+
+
+def _aggregate(drivers, games, track_type, per_game):
+    names = sorted({name for result in per_game for name in result["scores"]})
+    results = {
+        name: {"wins": 0, "losses": 0, "ties": 0, "avg_score": 0.0} for name in names
+    }
+
+    for result in per_game:
+        winner = result["winner"]
+        for name, score in result["scores"].items():
+            results[name]["avg_score"] += score
+            if winner is None:
+                results[name]["ties"] += 1
+            elif name == winner:
+                results[name]["wins"] += 1
+            else:
+                results[name]["losses"] += 1
+
+    for name in names:
+        if games:
+            results[name]["avg_score"] = round(results[name]["avg_score"] / games, 2)
+
+    return {
+        "games": games,
+        "track_type": track_type,
+        "drivers": list(drivers),
+        "results": results,
+        "per_game": per_game,
+    }
diff --git a/rose/engine/test_player.py b/rose/engine/test_player.py
index 3ddac3b..e93efc8 100644
--- a/rose/engine/test_player.py
+++ b/rose/engine/test_player.py
@@ -82,6 +82,9 @@ def test_player_state():
         "pickups": 0,
         "misses": 0,
         "hits": 0,
+        "wall_hits": 0,
+        "water_hits": 0,
+        "crack_hits": 0,
         "breaks": 0,
         "jumps": 0,
         "collisions": 0,
diff --git a/rose/engine/test_score.py b/rose/engine/test_score.py
index 680b7e3..048d44e 100644
--- a/rose/engine/test_score.py
+++ b/rose/engine/test_score.py
@@ -119,6 +119,7 @@ def test_other(self, action):
         self.process()
         self.assert_score(self.score)
         self.assert_keep_obstacle()
+        assert self.player.misses == 1
 
 
 class MagicActionTest(SinglePlayerTest):
@@ -132,6 +133,7 @@ class MagicActionTest(SinglePlayerTest):
     # Must be defined in subclass
     action = None
     magic_score = None
+    hit_field = None
 
     @pytest.mark.parametrize("action", FORWARD_ACTIONS)
     def test_forward(self, action):
@@ -143,6 +145,7 @@ def test_forward(self, action):
         else:
             self.assert_move_back()
             self.assert_remove_obstacle()
+            assert getattr(self.player, self.hit_field) == 1
 
     def test_right(self):
         self.player.action = actions.RIGHT
@@ -161,12 +164,14 @@ class TestCrack(MagicActionTest):
     magic_score = config.score_jump
     obstacle = obstacles.CRACK
     action = actions.JUMP
+    hit_field = "crack_hits"
 
 
 class TestWater(MagicActionTest):
     magic_score = config.score_brake
     obstacle = obstacles.WATER
     action = actions.BRAKE
+    hit_field = "water_hits"
 
 
 class TurnTest(SinglePlayerTest):
@@ -195,6 +200,7 @@ def test_other(self, action):
         # TODO: decrease points on redundant action?
         self.assert_move_back_no_punish()
         self.assert_remove_obstacle()
+        assert self.player.wall_hits == 1
 
 
 class TestTrash(TurnTest):
diff --git a/rose/engine/test_server.py b/rose/engine/test_server.py
new file mode 100644
index 0000000..631fc50
--- /dev/null
+++ b/rose/engine/test_server.py
@@ -0,0 +1,156 @@
+import asyncio
+import contextlib
+
+from aiohttp import web
+from aiohttp.test_utils import TestClient
+from aiohttp.test_utils import TestServer
+
+from rose.common import actions
+from rose.engine import config
+from rose.engine import logic
+from rose.engine import server
+from rose.telemetry.sinks import LiveSink
+
+
+class FakePlayer:
+    def __init__(self, name, score):
+        self.name = name
+        self.score = score
+
+    def state(self):
+        return {"name": self.name, "score": self.score}
+
+
+def build_app():
+    app = web.Application()
+    app.router.add_get("/telemetry", server.telemetry_handler)
+    app.router.add_post("/simulate", server.simulate_start_handler)
+    app.router.add_get("/simulate/{job_id}", server.simulate_status_handler)
+    return app
+
+
+def make_driver_app(name, action_fn):
+    async def get_handler(request):
+        return web.json_response({"info": {"name": name}})
+
+    async def post_handler(request):
+        payload = await request.json()
+        return web.json_response({"info": {"name": name, "action": action_fn(payload)}})
+
+    driver_app = web.Application()
+    driver_app.router.add_get("/", get_handler)
+    driver_app.router.add_post("/", post_handler)
+    return driver_app
+
+
+class FakeDriver:
+    def __init__(self, name, action_fn):
+        self._server = TestServer(make_driver_app(name, action_fn))
+
+    async def __aenter__(self):
+        await self._server.start_server()
+        return str(self._server.make_url("/"))
+
+    async def __aexit__(self, *exc_info):
+        await self._server.close()
+
+
+def test_telemetry_handler_returns_recent_history_and_results(monkeypatch):
+    sink = LiveSink()
+    sink.on_step(0, [FakePlayer("A", 10), FakePlayer("B", 5)], track=None)
+    sink.on_game_end(
+        players=None, result={"scores": {"A": 100, "B": 90}, "winner": "A"}
+    )
+    monkeypatch.setattr(server, "telemetry", sink)
+
+    async def scenario():
+        async with TestClient(TestServer(build_app())) as client:
+            resp = await client.get("/telemetry")
+            return resp.status, await resp.json()
+
+    status, body = asyncio.run(scenario())
+
+    assert status == 200
+    assert len(body["history"]) == 1
+    assert body["history"][0]["players"] == [
+        {"name": "A", "score": 10},
+        {"name": "B", "score": 5},
+    ]
+    assert body["recent_results"] == [{"scores": {"A": 100, "B": 90}, "winner": "A"}]
+
+
+def test_simulate_endpoint_rejects_bad_payloads():
+    async def scenario():
+        async with TestClient(TestServer(build_app())) as client:
+            resp = await client.post("/simulate", json={"drivers": ["only-one"]})
+            return resp.status
+
+    assert asyncio.run(scenario()) == 400
+
+
+def test_simulate_status_unknown_job_is_404():
+    async def scenario():
+        async with TestClient(TestServer(build_app())) as client:
+            resp = await client.get("/simulate/does-not-exist")
+            return resp.status
+
+    assert asyncio.run(scenario()) == 404
+
+
+def test_simulate_job_runs_in_background_and_completes(monkeypatch):
+    # _run_live_batch (behind /simulate) drives games through the shared
+    # game_loop, same as manual play would -- it doesn't call simulate.py's
+    # headless runner. So this test must actually run that loop in the
+    # background, same as server.run() does in production, or state/reset
+    # signals never get picked up and the job hangs forever.
+    monkeypatch.setattr(config, "game_duration", 4)
+    server.state.update(
+        {
+            "rate": 50,
+            "running": 0,
+            "reset": None,
+            "drivers": [],
+            "timeleft": None,
+            "track_type": "same",
+        }
+    )
+
+    async def scenario():
+        loop_task = asyncio.create_task(
+            logic.game_loop(server.state, server.active_websockets, server.telemetry)
+        )
+        try:
+            async with FakeDriver(
+                "DriverA", lambda payload: actions.NONE
+            ) as url_a, FakeDriver(
+                "DriverB", lambda payload: actions.NONE
+            ) as url_b, TestClient(
+                TestServer(build_app())
+            ) as client:
+                start_resp = await client.post(
+                    "/simulate",
+                    json={"drivers": [url_a, url_b], "games": 2, "track": "same"},
+                )
+                assert start_resp.status == 202
+                job_id = (await start_resp.json())["job_id"]
+
+                # Budget comfortably above the mandatory 2s INTER_ROUND_PAUSE_S
+                # between the 2 games plus tick overhead.
+                for _ in range(400):
+                    status_resp = await client.get(f"/simulate/{job_id}")
+                    body = await status_resp.json()
+                    if body["status"] != "running":
+                        return body
+                    await asyncio.sleep(0.02)
+
+                raise AssertionError("job never finished")
+        finally:
+            loop_task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await loop_task
+
+    job = asyncio.run(scenario())
+
+    assert job["status"] == "done"
+    assert job["result"]["games"] == 2
+    assert set(job["result"]["results"].keys()) == {"DriverA", "DriverB"}
diff --git a/rose/engine/test_simulate.py b/rose/engine/test_simulate.py
new file mode 100644
index 0000000..6c83bfd
--- /dev/null
+++ b/rose/engine/test_simulate.py
@@ -0,0 +1,115 @@
+import asyncio
+import random
+
+import pytest
+from aiohttp import web
+from aiohttp.test_utils import TestServer
+
+from rose.common import actions
+from rose.common import obstacles
+from rose.engine import config
+from rose.engine import simulate
+
+
+def make_driver_app(name, action_fn):
+    """A minimal stand-in for a rose-game-ai driver HTTP server."""
+
+    async def get_handler(request):
+        return web.json_response({"info": {"name": name}})
+
+    async def post_handler(request):
+        payload = await request.json()
+        return web.json_response({"info": {"name": name, "action": action_fn(payload)}})
+
+    app = web.Application()
+    app.router.add_get("/", get_handler)
+    app.router.add_post("/", post_handler)
+    return app
+
+
+class FakeDriver:
+    """Runs a driver app on an ephemeral local port for the duration of a `with` block."""
+
+    def __init__(self, name, action_fn):
+        self._server = TestServer(make_driver_app(name, action_fn))
+
+    async def __aenter__(self):
+        await self._server.start_server()
+        return str(self._server.make_url("/"))
+
+    async def __aexit__(self, *exc_info):
+        await self._server.close()
+
+
+def test_run_single_game_ties_when_drivers_behave_identically(monkeypatch):
+    monkeypatch.setattr(config, "game_duration", 6)
+
+    async def scenario():
+        async with FakeDriver(
+            "DriverA", lambda payload: actions.NONE
+        ) as url_a, FakeDriver("DriverB", lambda payload: actions.NONE) as url_b:
+            return await simulate.run_single_game([url_a, url_b], track_type="same")
+
+    result = asyncio.run(scenario())
+
+    assert result["winner"] is None
+    assert result["scores"]["DriverA"] == result["scores"]["DriverB"]
+
+
+def test_run_single_game_determines_a_winner(monkeypatch):
+    # Pin the obstacle stream so the outcome doesn't depend on real randomness:
+    # every generated row is a CRACK placed exactly where both (stationary)
+    # players sit in their lane.
+    monkeypatch.setattr(config, "game_duration", 12)
+    monkeypatch.setattr(obstacles, "get_random_obstacle", lambda: obstacles.CRACK)
+    monkeypatch.setattr(random, "choice", lambda seq: 1)
+
+    async def scenario():
+        async with FakeDriver(
+            "DriverA", lambda payload: actions.JUMP
+        ) as url_a, FakeDriver("DriverB", lambda payload: actions.NONE) as url_b:
+            return await simulate.run_single_game([url_a, url_b], track_type="same")
+
+    result = asyncio.run(scenario())
+
+    assert result["winner"] == "DriverA"
+    assert result["scores"]["DriverA"] > result["scores"]["DriverB"]
+
+
+def test_run_single_game_raises_if_a_driver_does_not_respond():
+    async def scenario():
+        async with FakeDriver("DriverA", lambda payload: actions.NONE) as url_a:
+            return await simulate.run_single_game(
+                [url_a, "http://127.0.0.1:1/unreachable"], track_type="same"
+            )
+
+    with pytest.raises(RuntimeError):
+        asyncio.run(scenario())
+
+
+def test_run_batch_aggregates_win_loss_tie_stats(monkeypatch):
+    monkeypatch.setattr(config, "game_duration", 5)
+
+    async def scenario():
+        async with FakeDriver(
+            "DriverA", lambda payload: actions.NONE
+        ) as url_a, FakeDriver("DriverB", lambda payload: actions.NONE) as url_b:
+            stats = await simulate.run_batch([url_a, url_b], games=3, track_type="same")
+            return stats, url_a, url_b
+
+    stats, url_a, url_b = asyncio.run(scenario())
+
+    assert stats["games"] == 3
+    assert stats["track_type"] == "same"
+    assert stats["drivers"] == [url_a, url_b]
+    assert len(stats["per_game"]) == 3
+
+    for name in ("DriverA", "DriverB"):
+        assert stats["results"][name]["ties"] == 3
+        assert stats["results"][name]["wins"] == 0
+        assert stats["results"][name]["losses"] == 0
+
+    assert (
+        stats["results"]["DriverA"]["avg_score"]
+        == stats["results"]["DriverB"]["avg_score"]
+    )
diff --git a/rose/telemetry/__init__.py b/rose/telemetry/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/rose/telemetry/observer.py b/rose/telemetry/observer.py
new file mode 100644
index 0000000..69cb058
--- /dev/null
+++ b/rose/telemetry/observer.py
@@ -0,0 +1,60 @@
+"""Telemetry observer interface for the game engine.
+
+An observer is notified at well-defined points in the game lifecycle
+(game start, each tick, game end). It is purely additive: nothing in
+the engine requires an observer to be present, and the default
+NullObserver is a no-op so existing call sites are unaffected.
+"""
+
+
+class TelemetryObserver(object):
+    """Base class for telemetry observers. Override any subset of hooks."""
+
+    def on_game_start(self, track, players, meta=None):
+        """Called once, right after players/track are initialized.
+
+        Args:
+            track (Track): the initialized game track.
+            players (list of Player): the initialized players.
+            meta (dict, optional): extra context (e.g. track_type).
+        """
+
+    def on_step(self, step_index, players, track):
+        """Called once per tick, right after score.process() has run.
+
+        Args:
+            step_index (int): 0-based tick counter for this game.
+            players (list of Player): players with up-to-date state.
+            track (Track): track with up-to-date state.
+        """
+
+    def on_game_end(self, players, result):
+        """Called once, after the last tick of a game.
+
+        Args:
+            players (list of Player): players in their final state.
+            result (dict): {"scores": {name: score}, "winner": name_or_None}.
+        """
+
+
+class NullObserver(TelemetryObserver):
+    """No-op observer, used as the implicit default everywhere."""
+
+
+class CompositeObserver(TelemetryObserver):
+    """Fans out every hook to a list of child observers."""
+
+    def __init__(self, observers):
+        self._observers = list(observers)
+
+    def on_game_start(self, track, players, meta=None):
+        for observer in self._observers:
+            observer.on_game_start(track, players, meta)
+
+    def on_step(self, step_index, players, track):
+        for observer in self._observers:
+            observer.on_step(step_index, players, track)
+
+    def on_game_end(self, players, result):
+        for observer in self._observers:
+            observer.on_game_end(players, result)
diff --git a/rose/telemetry/sinks.py b/rose/telemetry/sinks.py
new file mode 100644
index 0000000..312dd2b
--- /dev/null
+++ b/rose/telemetry/sinks.py
@@ -0,0 +1,136 @@
+"""Concrete TelemetryObserver sinks."""
+
+import collections
+import json
+
+from rose.telemetry.observer import TelemetryObserver
+
+
+class InMemorySink(TelemetryObserver):
+    """Keeps per-step snapshots and the final result in memory.
+
+    Handy for batch simulation, where the run is over long before anyone
+    could tail a file, and for tests.
+    """
+
+    def __init__(self):
+        self.games = []
+        self._current_game = None
+
+    def on_game_start(self, track, players, meta=None):
+        self._current_game = {
+            "meta": meta or {},
+            "steps": [],
+            "result": None,
+        }
+        self.games.append(self._current_game)
+
+    def on_step(self, step_index, players, track):
+        self._current_game["steps"].append(
+            {
+                "step": step_index,
+                "players": [player.state() for player in players],
+                "track": track.state(),
+            }
+        )
+
+    def on_game_end(self, players, result):
+        self._current_game["result"] = result
+
+
+class LiveSink(TelemetryObserver):
+    """Keeps a bounded in-memory history for the live, running game server.
+
+    Unlike InMemorySink, this isn't grouped per-game: it's a flat rolling log
+    of recent ticks plus a rolling log of recently finished games, meant to be
+    polled over HTTP (see rose.engine.server's `/telemetry` route) by a
+    dashboard that just wants "what's been happening lately".
+    """
+
+    def __init__(self, max_ticks=500, max_results=20):
+        self._history = collections.deque(maxlen=max_ticks)
+        self._recent_results = collections.deque(maxlen=max_results)
+        self._total_finished = 0
+        # Snapshot of _total_finished as of the last clear(), subtracted out
+        # for display so round numbering restarts at 1 after a clear, without
+        # touching _total_finished itself (see result_count()/clear() below).
+        self._round_offset = 0
+
+    def on_step(self, step_index, players, track):
+        self._history.append(
+            {
+                "step": step_index,
+                "players": [player.state() for player in players],
+            }
+        )
+
+    def on_game_end(self, players, result):
+        self._recent_results.append(result)
+        self._total_finished += 1
+
+    def result_count(self):
+        """Total number of finished games recorded so far (monotonic).
+
+        Deliberately backed by a counter, not len(_recent_results) — that
+        deque is capped at max_results, so its length would plateau and
+        break callers polling for "has another game finished" past that cap.
+        """
+        return self._total_finished
+
+    def latest_result(self):
+        """Most recently finished game's result, or None if none yet."""
+        return self._recent_results[-1] if self._recent_results else None
+
+    def clear(self):
+        """Clear the displayed tick log, match history, and round numbering.
+
+        Deliberately leaves _total_finished itself untouched: it's used by
+        server._run_live_batch to detect "has another round finished", and
+        resetting it here would break any batch job in flight. Round
+        numbering for display is reset instead via _round_offset (see
+        snapshot()), so the next finished game shows as round 1 again
+        without disturbing that internal counter.
+        """
+        self._history.clear()
+        self._recent_results.clear()
+        self._round_offset = self._total_finished
+
+    def snapshot(self):
+        """Return a JSON-serializable view of recent ticks and results."""
+        return {
+            "history": list(self._history),
+            "recent_results": list(self._recent_results),
+            "total_finished": self._total_finished - self._round_offset,
+        }
+
+
+class JSONLSink(TelemetryObserver):
+    """Appends one JSON object per event to a file, one event per line.
+
+    This is a natural feed for a future dashboard to tail (`tail -f` or a
+    simple polling reader), without the dashboard needing to understand
+    engine internals beyond the JSON shape written here.
+    """
+
+    def __init__(self, path):
+        self._path = path
+
+    def on_game_start(self, track, players, meta=None):
+        self._append({"event": "game_start", "meta": meta or {}})
+
+    def on_step(self, step_index, players, track):
+        self._append(
+            {
+                "event": "step",
+                "step": step_index,
+                "players": [player.state() for player in players],
+                "track": track.state(),
+            }
+        )
+
+    def on_game_end(self, players, result):
+        self._append({"event": "game_end", "result": result})
+
+    def _append(self, record):
+        with open(self._path, "a") as f:
+            f.write(json.dumps(record) + "\n")
diff --git a/rose/telemetry/test_observer.py b/rose/telemetry/test_observer.py
new file mode 100644
index 0000000..ad90ab9
--- /dev/null
+++ b/rose/telemetry/test_observer.py
@@ -0,0 +1,105 @@
+from rose.telemetry.observer import CompositeObserver
+from rose.telemetry.observer import NullObserver
+from rose.telemetry.observer import TelemetryObserver
+from rose.telemetry.sinks import InMemorySink
+from rose.telemetry.sinks import LiveSink
+
+
+def test_null_observer_is_a_noop():
+    observer = NullObserver()
+
+    # None of these should raise, regardless of arguments.
+    observer.on_game_start(track=None, players=None)
+    observer.on_step(step_index=0, players=None, track=None)
+    observer.on_game_end(players=None, result=None)
+
+
+def test_in_memory_sink_records_game_lifecycle():
+    sink = InMemorySink()
+
+    class FakePlayer:
+        def __init__(self, name, score):
+            self.name = name
+            self.score = score
+
+        def state(self):
+            return {"name": self.name, "score": self.score}
+
+    class FakeTrack:
+        def state(self):
+            return []
+
+    players = [FakePlayer("A", 0), FakePlayer("B", 0)]
+    track = FakeTrack()
+
+    sink.on_game_start(track, players, {"track_type": "same"})
+    sink.on_step(0, players, track)
+    sink.on_step(1, players, track)
+    result = {"scores": {"A": 10, "B": 5}, "winner": "A"}
+    sink.on_game_end(players, result)
+
+    assert len(sink.games) == 1
+    game = sink.games[0]
+    assert game["meta"] == {"track_type": "same"}
+    assert len(game["steps"]) == 2
+    assert game["steps"][0]["step"] == 0
+    assert game["result"] == result
+
+
+def test_in_memory_sink_tracks_multiple_games_independently():
+    sink = InMemorySink()
+
+    sink.on_game_start(track=None, players=[])
+    sink.on_game_end(players=[], result={"scores": {}, "winner": None})
+
+    sink.on_game_start(track=None, players=[])
+    sink.on_game_end(players=[], result={"scores": {}, "winner": None})
+
+    assert len(sink.games) == 2
+
+
+def test_live_sink_clear_resets_displayed_round_numbering():
+    sink = LiveSink()
+
+    sink.on_game_end(players=None, result={"scores": {"A": 10}, "winner": "A"})
+    sink.on_game_end(players=None, result={"scores": {"A": 10}, "winner": "A"})
+    assert sink.snapshot()["total_finished"] == 2
+
+    sink.clear()
+    assert sink.snapshot()["total_finished"] == 0
+    assert sink.snapshot()["recent_results"] == []
+
+    sink.on_game_end(players=None, result={"scores": {"A": 10}, "winner": "A"})
+    assert sink.snapshot()["total_finished"] == 1
+
+
+def test_live_sink_clear_does_not_disturb_result_count():
+    # result_count() backs server._run_live_batch's "did another round
+    # finish" polling and must stay monotonic across a clear(), even though
+    # the displayed total_finished resets for round numbering.
+    sink = LiveSink()
+
+    sink.on_game_end(players=None, result={"scores": {"A": 10}, "winner": "A"})
+    assert sink.result_count() == 1
+
+    sink.clear()
+    assert sink.result_count() == 1
+
+    sink.on_game_end(players=None, result={"scores": {"A": 10}, "winner": "A"})
+    assert sink.result_count() == 2
+
+
+def test_composite_observer_fans_out_to_children():
+    calls = []
+
+    class RecordingObserver(TelemetryObserver):
+        def __init__(self, tag):
+            self.tag = tag
+
+        def on_step(self, step_index, players, track):
+            calls.append((self.tag, step_index))
+
+    composite = CompositeObserver([RecordingObserver("a"), RecordingObserver("b")])
+    composite.on_step(3, players=None, track=None)
+
+    assert calls == [("a", 3), ("b", 3)]
diff --git a/simulate.py b/simulate.py
new file mode 100644
index 0000000..91c18db
--- /dev/null
+++ b/simulate.py
@@ -0,0 +1,55 @@
+import argparse
+import asyncio
+import json
+import logging
+
+from rose.engine import simulate
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Run headless batch simulations of the engine against rose-game-ai drivers."
+    )
+    parser.add_argument(
+        "-d",
+        "--drivers",
+        nargs=2,
+        required=True,
+        metavar=("DRIVER1_URL", "DRIVER2_URL"),
+        help="Exactly two driver URLs to simulate (matches the game's 2-player design).",
+    )
+    parser.add_argument(
+        "-g", "--games", type=int, default=10, help="Number of games to simulate."
+    )
+    parser.add_argument(
+        "-t",
+        "--track",
+        choices=["same", "random"],
+        default="random",
+        help="Choose the track type. Can be 'same' or 'random'.",
+    )
+    parser.add_argument(
+        "-o",
+        "--output",
+        default="batch_stats.json",
+        help="Path to write the aggregated JSON stats to.",
+    )
+    parser.add_argument(
+        "--log", default="WARNING", help="Set the logging level. E.g. --log DEBUG"
+    )
+
+    args = parser.parse_args()
+
+    logging.basicConfig(level=getattr(logging, args.log.upper()))
+
+    stats = asyncio.run(simulate.run_batch(args.drivers, args.games, args.track))
+
+    with open(args.output, "w") as f:
+        json.dump(stats, f, indent=2)
+
+    print(f"Simulated {args.games} games -> {args.output}")
+    print(json.dumps(stats["results"], indent=2))
+
+
+if __name__ == "__main__":
+    main()