Merge pull request 'Update debian:trixie-slim Docker digest to a99cfc5' (#32) from renovate/debian-trixie-slim into main

Reviewed-on: #32

.gitignore

@@ -1,3 +1,5 @@
 venv
 __pycache__
 docker/config
+.coverage
+.pytest_cache

README.md

@@ -17,6 +17,9 @@ A high-performance, Object-Oriented Python client for OpenTTD servers, specifica
 - **Station Listing:** Enumerate stations via the Admin GameScript channel with `list_stations()`.
 - **Station Cargo Snapshots:** `OpenTTDAdminClient.get_station()` returns a station's live per-cargo state from the running game — both the **real-time** amount waiting and the **planned** flow through the cargodist link graph — over the AdminBridge GS (stock GameScript API, no server patch needed).
 - **Cargo Flow Breakdown:** `OpenTTDAdminClient.get_station_cargo()` breaks one cargo type down by **source station** and **next hop** (routing destination) for both waiting (real-time) and planned amounts, with optional `from_station`/`via_station` filters.
+- **Cargo Table:** `OpenTTDAdminClient.list_cargo()` names the bare `cargo_id`s the station queries and cargo events return — every cargo type in the running game with its label (`PASS`, `COAL`, …) and freight flag, read live so NewGRF-specific ids resolve correctly.
+- **Game Events:** react to the game instead of polling it — `subscribe_events()` streams events over the AdminBridge GS as they happen: a vehicle reaching or leaving a stop (`vehicle_arrive`/`vehicle_depart`, with dwell time and cargo aboard), a station's waiting cargo changing (`cargo_waiting`), plus crashes, industries opening/closing, towns, companies and subsidies. Consume them with `await wait_for_event()` or an `on_event` callback; filter by kind, company, vehicle, station or cargo.
+- **Version-checked Server Bridge:** the server-side AdminBridge GameScript that answers all of the above lives in [`gamescript/AdminBridge/`](gamescript/AdminBridge/README.md) and is mounted into the Docker server automatically. `get_bridge_version()` verifies at startup that the running bridge is new enough, so an outdated one fails by name instead of hanging every query.
 
 ## 🛠 Setup
 
@@ -67,6 +70,8 @@ await client.joined.wait()
 
 - `main.py`: Main entry point and usage example.
 - `lib/openttd/`: Core package containing the protocol and client logic.
+- `gamescript/AdminBridge/`: The server-side GameScript the admin client's GameScript channel talks to.
+- `docker/`: Dedicated JGRPP server (Dockerfile, compose, and the local server patches in `docker/patches/`).
 - `docs/`: Extensive documentation on architecture, protocol, and contributing.
 - `tests/`: Comprehensive test suite (Logic, Protocol, E2E).
 
@@ -81,4 +86,5 @@ For detailed instructions on E2E testing and coverage reports, see the [Testing
 - [Architecture & Design](docs/ARCHITECTURE.md)
 - [Protocol Internals (PAKE/Encryption)](docs/PROTOCOL.md)
 - [Vehicle Timetables Usage Guide](docs/TIMETABLES.md)
+- [Game Events Usage Guide](docs/EVENTS.md)
 - [Contributor Guide](docs/CONTRIBUTING.md)

docker/Dockerfile

@@ -1,5 +1,5 @@
 # Build stage
-FROM debian:13@sha256:34cd9e9fd437c0a095ec39cb2e73422c9f30821b0d0848ed74fd0d43bae4d958 AS builder
+FROM debian:13@sha256:9cc080028c43b27d2074d63a5f9caf7166d731494965616c1a6d2827a004585c AS builder
 
 RUN apt-get update && apt-get install -y \
     build-essential \
@@ -29,7 +29,7 @@ RUN cmake .. \
 # Must track the builder's Debian release: the builder (debian:13/trixie) links
 # against glibc 2.38+, so an older runtime (e.g. bookworm, glibc 2.36) cannot run
 # the resulting binary. Package names use the trixie t64 spelling.
-FROM debian:trixie-slim@sha256:3a39a0592364683e6bab97937b72cad5a8fa6dcbbee90edb3bb48c7f8e94f258
+FROM debian:trixie-slim@sha256:a99cfc517144bc59b1978475ec53b46ecabec7e43635402ee5b77cc54cd1b20a
 
 RUN apt-get update && apt-get install -y \
     libcurl3t64-gnutls \

docker/README.md

@@ -24,6 +24,19 @@ This setup builds OpenTTD with the JGR Patch Pack (JGRPP) from source and runs i
 4.  **Save Games:**
     Save games are stored in `config/save/`.
 
+## AdminBridge GameScript
+`config/` is gitignored (it holds savegames, downloaded content and generated config), so the
+AdminBridge GameScript — the server-side half of the Python client's admin GameScript channel —
+is kept in [gamescript/AdminBridge/](../gamescript/AdminBridge/README.md) instead and bind-mounted
+read-only over the container's `game/AdminBridge` by `docker-compose.yml`. Nothing to install:
+`docker-compose up` serves the tracked copy. Edit it there, not under `config/game/`, which the
+mount shadows. A GameScript is loaded at game start, so restart the server to pick up a change.
+
+Running the server outside this compose file? Copy the directory in by hand instead:
+```bash
+cp -r ../gamescript/AdminBridge ~/.local/share/openttd/game/
+```
+
 ## JGRPP Source
 The source code is cloned from the `jgrpp` branch of `https://github.com/JGRennison/OpenTTD-patches`
 (currently at tag `jgrpp-0.71.1`) and carries local patches from `patches/` — see

docker/docker-compose.yml

@@ -9,6 +9,10 @@ services:
       - "3977:3977/tcp"
     volumes:
       - ./config:/home/openttd/.local/share/openttd
+      # The AdminBridge GameScript is served from its tracked source rather than from the
+      # (gitignored) config directory, so the server always runs the reviewed copy. Nested
+      # inside the mount above, and read-only: edit gamescript/AdminBridge/, not the container.
+      - ../gamescript/AdminBridge:/home/openttd/.local/share/openttd/game/AdminBridge:ro
     environment:
       - PUID=1000
       - PGID=1000

docs/ARCHITECTURE.md

@@ -16,6 +16,12 @@ The primary API for developers.
 - **Event-Driven:** Uses `asyncio.Event` (like `self.joined`) to signal state changes to the calling code.
 - **Callback System:** Provides hooks like `on_chat` to allow users to react to game events without modifying the core library.
 
+### 3. `OpenTTDAdminClient` (Admin Network)
+Talks to the Admin port (TCP 3977) and, through the AdminBridge GameScript's JSON channel, to the running game itself.
+- **Correlated Queries:** `get_timetable()`, `get_station()`, `get_dispatch()` and friends all funnel through `_gs_query()`, which tags each request with a monotonic `request_id`, parks a future, and lets `receive_ServerGamescript` resolve it when the matching reply arrives.
+- **Two Halves:** the queries above only work because a matching GameScript is running on the server. That script is part of this project, in [`gamescript/AdminBridge/`](../gamescript/AdminBridge/README.md), and the client's `get_bridge_version()` checks the running copy is new enough for what it is about to send.
+- **Push Events:** `subscribe_events()` opens the one stream that flows the other way. Event batches carry no `request_id`, so `receive_ServerGamescript` routes them to the event consumers instead: each event goes to the longest-waiting matching `wait_for_event()` caller, or into a bounded buffer if nobody is waiting, and to the `on_event` observer either way. See [EVENTS.md](EVENTS.md).
+
 ## Handshake Flow
 
 1.  **Connection:** TCP connection established to Port 3979.

docs/EVENTS.md

@@ -0,0 +1,234 @@
+# Game Events: Usage Guide
+
+Everything else in this library is a **request**: you ask the server something and it answers.
+Events are the other direction — the server tells you when something happens, so a bot can react
+to the game instead of polling it in a loop. This guide covers `OpenTTDAdminClient`'s
+`subscribe_events()`, `unsubscribe_events()`, `wait_for_event()` and the `on_event` callback.
+For the JSON that goes over the wire, see [PROTOCOL.md](PROTOCOL.md#game-events).
+
+## Quick start
+
+```python
+import asyncio
+from openttd import OpenTTDAdminClient
+from openttd.protocol import GameEventType
+
+async def main():
+    admin = OpenTTDAdminClient("127.0.0.1", admin_name="EventWatcher")
+    await admin.connect(admin_password="asd", secure=True)
+    await admin.joined.wait()
+
+    await admin.subscribe_events(
+        events=[GameEventType.VehicleArrive, GameEventType.VehicleDepart],
+    )
+
+    for _ in range(10):
+        event = await admin.wait_for_event(timeout=60.0)
+        print(f"vehicle {event['vehicle_id']} {event['event']} "
+              f"at station {event['station_id']}")
+
+    await admin.unsubscribe_events()
+    await admin.quit()
+
+asyncio.run(main())
+```
+
+## Two ways to consume events
+
+Both see **every** event, so you can use either or mix them.
+
+**Pull — `await admin.wait_for_event(kind=None, timeout=5.0)`.** Returns the next matching event.
+`kind` is a `GameEventType` (or plain string), an iterable of them, or `None` for "anything".
+Events that arrive while you are not waiting are buffered, so nothing is lost between two calls,
+and each event goes to at most one waiter. Raises `asyncio.TimeoutError` if nothing matching
+arrives in time, `ConnectionError` if the admin connection drops mid-wait.
+
+```python
+# The next arrival or departure, whichever comes first.
+event = await admin.wait_for_event(
+    {GameEventType.VehicleArrive, GameEventType.VehicleDepart}, timeout=30.0)
+```
+
+**Push — `admin.on_event = handler`.** A plain callback (like `on_chat` / `on_console`), invoked
+with each event dict as it arrives. Use it for a long-running loop that reacts to everything:
+
+```python
+def handle(event):
+    if event["event"] == "cargo_waiting" and event["delta"] > 0:
+        print(f"{event['delta']} units of cargo {event['cargo_id']} "
+              f"arrived at station {event['station_id']}")
+
+admin.on_event = handle
+await admin.subscribe_events(events=[GameEventType.CargoWaiting])
+await asyncio.sleep(600)   # events arrive on the connection's own task
+```
+
+The buffer behind `wait_for_event()` holds the last 256 unclaimed events by default
+(`OpenTTDAdminClient(..., event_buffer_size=N)`); past that the oldest are dropped, so a
+subscription nobody drains cannot grow without bound.
+
+## Event kinds
+
+Import them from `openttd.protocol` as `GameEventType`, or just use the string. They come in two
+flavours, which behave differently in ways worth knowing.
+
+### Polled events — sampled, not instantaneous
+
+The engine raises **no** event when a vehicle reaches a stop or when cargo shows up, so the
+AdminBridge GameScript synthesises these by sampling game state every `interval` ticks and
+reporting what changed since the previous sample.
+
+| Kind | Fires when | Fields |
+|---|---|---|
+| `vehicle_arrive` | a vehicle starts loading/unloading at a station | `vehicle_id`, `station_id`, `owner`, `vehicle_type`, `order_position`, `cargo` |
+| `vehicle_depart` | a vehicle stops loading/unloading at a station | the same, plus `dwell` (ticks it stayed) |
+| `cargo_waiting` | a station's waiting amount of a cargo changes | `station_id`, `cargo_id`, `waiting` (new total), `delta` (signed change) |
+
+`cargo` is a list of `{"cargo_id": n, "load": units}` for the cargo actually aboard (omit it with
+`include_cargo=False`). `vehicle_type` is `0` rail, `1` road, `2` water, `3` air.
+`order_position` is the index of the vehicle's current order, or `-1` if it has none.
+`delta` is negative when cargo was picked up rather than delivered.
+
+Three consequences of sampling that you should design around:
+
+- **Short stops can be missed entirely.** If a vehicle arrives and leaves between two samples,
+  neither event exists. Lower `interval` to narrow the window; you cannot close it.
+- **The first sample only establishes a baseline.** A vehicle already sitting at a station when
+  you subscribe did not just arrive and gets no `vehicle_arrive`. Its eventual `dwell` is counted
+  from that first sample, not from its real arrival.
+- **`delta` is measured against the previous sample, not the previous event.** With a
+  `min_cargo_delta` filter, suppressed changes still move the baseline, so the next reported
+  `delta` covers everything since the last *sample*.
+
+Two more edge cases in how "at a stop" is defined: it means *loading or unloading* at a station,
+so a vehicle passing through a station non-stop never registers, and a vehicle that is stopped by
+hand or breaks down at a platform reads as having left (and as arriving again when it resumes).
+
+### Forwarded engine events — exact, as they happen
+
+These the engine does raise, so they are reported the moment they occur, with no sampling
+involved and no `interval` dependence.
+
+| Kind | Fields |
+|---|---|
+| `vehicle_crashed` | `vehicle_id`, `tile`, `reason`, `victims`, `owner` |
+| `station_first_vehicle` | `station_id`, `vehicle_id` |
+| `industry_open`, `industry_close` | `industry_id` |
+| `town_founded` | `town_id` |
+| `company_new`, `company_in_trouble`, `company_bankrupt` | `company_id` |
+| `subsidy_offer`, `subsidy_offer_expired`, `subsidy_awarded`, `subsidy_expired` | `subsidy_id` |
+
+Every event of either flavour also carries `event` (its kind) and `tick` (the game tick it was
+observed at).
+
+There is deliberately no `vehicle_lost`, `vehicle_waiting_in_depot` or `vehicle_unprofitable`:
+the engine raises those only for AI companies, never for a GameScript, so no bridge could
+forward them. Watch `vehicle_arrive`/`vehicle_depart` or poll `get_timetable()` instead.
+
+### `events_dropped`
+
+Not something you subscribe to — the bridge emits it when a single poll produced more events than
+its per-poll cap (200) and discarded `count` of them. Treat it as "your view has a hole in it":
+if you see it, your subscription is too broad or your interval too coarse for what the map is
+doing. Re-read state with `get_station()` / `get_timetable()` rather than trusting your
+event-derived view.
+
+## `subscribe_events(...)`
+
+Awaitable; returns the bridge's confirmation `{"events": [accepted kinds], "interval": N}`. Every
+argument narrows what you get, and all are optional.
+
+| Parameter | Type | Meaning |
+|---|---|---|
+| `events` | iterable | Which kinds to receive. Default: all of them. |
+| `interval` | `int` | Ticks between state samples for the polled kinds. Default `10`. This is their resolution, not a delay. |
+| `company_id` | `int` | Only report vehicles and stations owned by this company. |
+| `vehicles` | iterable | Only sample these vehicle ids instead of every vehicle. |
+| `stations` | iterable | Only sample these station ids instead of every station. |
+| `cargo` | iterable | Only inspect these cargo types (for `cargo_waiting`, and for the load on vehicle events). |
+| `min_cargo_delta` | `int` | Suppress `cargo_waiting` events smaller than this many units. Default `1` (report everything). |
+| `include_cargo` | `bool` | `False` leaves the per-cargo load off vehicle events. Default `True`. |
+| `timeout` | `float` | Seconds to wait for the confirmation. Default `5.0`. |
+
+Subscribing **replaces** any previous subscription and resets the baseline. `unsubscribe_events()`
+stops the stream and drops the bridge's sampling state.
+
+Errors: `ValueError` when the GameScript rejects the request (`unknown_event`,
+`invalid_interval`, `invalid_min_cargo_delta`, `invalid_cargo`), `asyncio.TimeoutError` when it
+does not answer (GameScripts don't run while the game is **paused**, so a paused server always
+times out), `ConnectionError` if the connection drops. The Gamescript update-frequency
+subscription the stream needs is set up automatically.
+
+## Cost: subscribe narrowly
+
+The bridge samples every watched vehicle and every watched station × cargo pair on each interval,
+inside a GameScript's limited per-tick opcode budget. The defaults — every kind, every vehicle,
+every station, every cargo, every 10 ticks — are fine on a small map and wasteful on a large busy
+one, where the poll can take longer than the interval it is trying to keep.
+
+Narrow it to what you actually need:
+
+```python
+# Watch two specific vehicles closely.
+await admin.subscribe_events(
+    events=[GameEventType.VehicleArrive, GameEventType.VehicleDepart],
+    vehicles=[7, 9], interval=2)
+
+# Watch coal piling up at one station, ignoring changes under 10 units.
+await admin.subscribe_events(
+    events=[GameEventType.CargoWaiting],
+    stations=[3], cargo=[0], min_cargo_delta=10, interval=30)
+
+# Only one company's traffic, without the per-vehicle cargo detail.
+await admin.subscribe_events(company_id=0, include_cargo=False)
+```
+
+## Putting it together
+
+```python
+import asyncio
+from openttd import OpenTTDAdminClient
+from openttd.protocol import GameEventType
+
+async def watch_route(admin, vehicle_id):
+    """Report how long a vehicle spends at each stop on its route."""
+    await admin.subscribe_events(
+        events=[GameEventType.VehicleArrive, GameEventType.VehicleDepart],
+        vehicles=[vehicle_id], interval=2)
+    try:
+        while True:
+            event = await admin.wait_for_event(timeout=300.0)
+            if event["event"] == "vehicle_arrive":
+                print(f"arrived at station {event['station_id']} (order {event['order_position']})")
+            else:
+                loaded = sum(c["load"] for c in event["cargo"])
+                print(f"left station {event['station_id']} after {event['dwell']} ticks "
+                      f"carrying {loaded} units")
+    except asyncio.TimeoutError:
+        print("vehicle went quiet")
+    finally:
+        await admin.unsubscribe_events()
+
+async def main():
+    admin = OpenTTDAdminClient("127.0.0.1", admin_name="RouteWatcher")
+    await admin.connect(admin_password="asd", secure=True)
+    await admin.joined.wait()
+    await watch_route(admin, vehicle_id=7)
+    await admin.quit()
+
+asyncio.run(main())
+```
+
+## Requirements
+
+The server must run the bundled AdminBridge GameScript (version 3 or newer) from
+[`gamescript/AdminBridge/`](../gamescript/AdminBridge/README.md) — `get_bridge_version()` checks
+that it is new enough, and it is worth calling once at startup. Events need **no**
+server patch — unlike `get_timetable()` and `get_dispatch()`, every getter involved is part of the
+stock GameScript API. Since GameScripts do not tick while the game is paused, no events are
+produced on a paused server.
+
+## See also
+- [PROTOCOL.md — Game Events](PROTOCOL.md#game-events) for the request/reply JSON.
+- [TIMETABLES.md](TIMETABLES.md) for reading and editing what the events tell you changed.
+- [ARCHITECTURE.md](ARCHITECTURE.md) for how the admin client fits together.

docs/PROTOCOL.md

@@ -26,6 +26,36 @@ Similar to the Game Port, the Admin Network uses X25519 PAKE for secure authenti
 ### Update Frequencies
 Admins can subscribe to various updates (Date, Client Info, Company Info, etc.) at different frequencies (Poll, Daily, Weekly, Monthly, Quarterly, Annually, Automatic).
 
+### The AdminBridge GameScript
+Everything in this section rides on a companion GameScript running on the server — the other half
+of the protocol, kept in [`gamescript/AdminBridge/`](../gamescript/AdminBridge/README.md). The
+Admin Network itself offers no way to ask the game about individual vehicles, stations or orders;
+what it does offer is an opaque JSON channel to whatever GameScript is loaded (`AdminGamescript`
+out, `ServerGamescript` back), and this bridge is what gives that channel meaning.
+
+Note what the channel does **not** give you: a bridge that does not recognise a command drops it
+silently. There is no "unknown command" reply, so a client talking to a bridge older than itself
+sees nothing but timeouts.
+
+### Bridge Version
+`get_bridge_version()` exists to turn that silence into an error. The bridge reports the version
+of the JSON protocol it implements, and the client compares it against `GS_BRIDGE_VERSION` (the
+version it was written against, in `protocol.py`).
+
+- **Request:** `{"command": "get_version", "request_id": X}`.
+- **Reply:** `{"command": "get_version", "request_id": X, "version": N, "commands": [...], "events": [...]}` —
+  `commands` is every command name the bridge answers (sorted) and `events` every event kind it
+  can push, so a client can feature-detect a single command instead of comparing version numbers.
+- **No reply at all** is the answer from a bridge older than version 4, which is when
+  `get_version` was added; `get_bridge_version()` surfaces that as `asyncio.TimeoutError`, the
+  same as a paused game or a server with no bridge loaded.
+
+The version covers the shape of the JSON protocol, not the implementation, and is declared in
+three places that [`tests/test_gamescript.py`](../tests/test_gamescript.py) keeps in step:
+`info.nut`'s `GetVersion()`, `main.nut`'s `BRIDGE_VERSION` and `protocol.py`'s
+`GS_BRIDGE_VERSION`. The same test checks the bridge's event catalogue against `GameEventType`
+and its command table against the commands the client actually sends.
+
 ### Vehicle Listing
 The Admin Network has no native packet or `AdminUpdateType` for listing individual vehicles — `ServerCompanyStats` only reports aggregate per-company vehicle counts (trains/lorries/buses/planes/ships). To retrieve an actual vehicle list, this client sends a `list_vehicles` command over the GameScript JSON channel (`AdminGamescript`/`ServerGamescript`) via `list_vehicles()`. This requires a companion GameScript running server-side that understands the `list_vehicles` command and replies with vehicle data through `ServerGamescript`.
 
@@ -62,6 +92,15 @@ Correlation, the `update_frequency` subscription requirement (auto-subscribed on
 - **Filters:** `via_station` restricts the query (and the `*_by_from` breakdowns) to cargo whose next hop is that station; `from_station` restricts it (and the `*_by_via` breakdowns) to cargo from that source; supplying both makes `waiting`/`planned` the exact source-and-next-hop amount. Pass `65535` to target `STATION_INVALID`.
 - **Reply (error):** same envelope with an `"error"` field: `"invalid_station"`, `"invalid_cargo"`, `"invalid_from_station"`/`"invalid_via_station"` (a filter that is neither a valid station nor `STATION_INVALID`), or `"response_too_large"`. `get_station_cargo()` raises `ValueError` for these.
 
+### Cargo Listing
+Cargo appears in the replies above (and in the `cargo_waiting` / vehicle events) as a bare numeric `cargo_id`. Those ids index the cargo table the loaded NewGRFs build for the running game, so they are **not stable across games** — the same id can be coal in one save and grain in another. `list_cargo()` resolves them, sending a `list_cargo` command over the same GameScript JSON channel and awaiting the correlated reply. The GS reads the table with the stock `GSCargoList`, `GSCargo.GetCargoLabel` and `GSCargo.IsFreight`, so this needs no server patch.
+
+- **Request:** `{"command": "list_cargo", "request_id": X}`.
+- **Reply (success):** `{"command": "list_cargo", "request_id": X, "cargo": [{"cargo_id": N, "label": "COAL", "freight": 0|1}, ...]}` — one entry per cargo type in the game, in `GSCargoList` order rather than by id, so index the list by `cargo_id`. `label` is the four-character NewGRF cargo label (`GetCargoLabel`, underscore-padded: `OIL_`), or `""` if the GS could not read it; `freight` is 1 for freight cargo and 0 for the rest (passengers, mail, …).
+- **Reply (error):** the GS defines no error for this command, so `list_cargo()` never raises `ValueError` — only `asyncio.TimeoutError` (paused game, GS not loaded) and `ConnectionError`.
+
+Correlation and the `update_frequency` subscription requirement (auto-subscribed on first use) are as described for the Timetable Query above. Unlike `list_vehicles()`/`list_stations()` — which are fire-and-forget despite the similar name — this one is awaitable and its reply does **not** reach the `on_gamescript` callback.
+
 ### Dispatch Query
 `get_dispatch()` fetches an authoritative snapshot of a vehicle's **scheduled dispatch** state over the GameScript JSON channel, the vehicle analogue of `get_timetable()` for JGRPP's scheduled dispatch feature. Like the timetable getters, the dispatch getters it relies on are added by a **server patch** (`docker/patches/0002-*`, adding `GSOrder.GetScheduledDispatch*` / `IsScheduledDispatchEnabled`), so it needs the patched JGRPP build. Correlation, the auto-subscribe, and the paused-game timeout behave exactly as for the Timetable Query.
 
@@ -69,6 +108,16 @@ Correlation, the `update_frequency` subscription requirement (auto-subscribed on
 - **Reply (success):** `{"command": "get_dispatch", "vehicle_id": N, "request_id": X, "enabled": 0|1, "schedules": [{"index", "duration", "start_tick", "delay", "reuse_slots", "slots": [{"offset", "flags"}, ...]}, ...]}`. `enabled` is whether scheduled dispatch is turned on for the vehicle; each schedule reports its `duration` (ticks), `start_tick`, `delay` (max allowed delay), `reuse_slots` (0/1) and its `slots` (each a departure `offset` within the duration plus a 16-bit `flags` word). These are the same schedules and slots edited by the game-port dispatch methods.
 - **Reply (error):** same envelope with an `"error"` field: `"invalid_vehicle"` or `"response_too_large"`. `get_dispatch()` raises `ValueError` for these.
 
+### Game Events
+Every other GameScript command above is request/reply. `subscribe_events()` instead opens a **push** stream: the AdminBridge GameScript sends event batches over `ServerGamescript` as things happen, unsolicited and without a `request_id`. For the calling API and the semantics of each kind, see the [Game Events Usage Guide](EVENTS.md).
+
+- **Subscribe request:** `{"command": "subscribe_events", "request_id": X}` plus any of the optional narrowing fields `"events"` (array of kind strings), `"interval"` (ticks between state samples, default 10), `"company_id"`, `"vehicles"`, `"stations"`, `"cargo"` (arrays of ids), `"min_cargo_delta"` (default 1) and `"include_cargo"` (bool, default true).
+- **Subscribe reply:** `{"command": "subscribe_events", "request_id": X, "events": [accepted kinds in catalogue order], "interval": N}`, or the same envelope with an `"error"` field: `"unknown_event"` (plus the offending `"event"`), `"invalid_interval"`, `"invalid_min_cargo_delta"`, or `"invalid_cargo"` (plus the offending `"cargo_id"`). `subscribe_events()` raises `ValueError` for these. Subscribing replaces any previous subscription and resets the bridge's sampling baseline.
+- **Unsubscribe:** `{"command": "unsubscribe_events", "request_id": X}` → `{"command": "unsubscribe_events", "request_id": X, "events": []}`. This also drops the sampling state.
+- **Event batch (unsolicited):** `{"command": "events", "events": [{"event": <kind>, "tick": T, ...}, ...]}`. Batches carry at most 24 events per packet, so one poll can produce several; a poll that generated more than 200 events is truncated and followed by a single `{"event": "events_dropped", "count": N}` entry. Because these batches carry no `request_id`, `receive_ServerGamescript` routes them to the event consumers (`on_event` / `wait_for_event()`) instead of the generic `on_gamescript` callback; every other GameScript payload reaches `on_gamescript` unchanged. The usual `update_frequency(Gamescript, Automatic)` subscription applies and is set up automatically by `subscribe_events()`.
+
+Event kinds come from two sources. `vehicle_arrive`, `vehicle_depart` and `cargo_waiting` are **synthesised** by the bridge, because the engine raises no GameScript event for them: every `interval` ticks it samples `GSVehicle.GetState`/`GetLocation` for the watched vehicles and `GSStation.GetCargoWaiting` for the watched station-cargo pairs, and emits an event per change against the previous sample (so a stop shorter than the interval is never reported, and the first sample after subscribing only sets a baseline). All the remaining kinds — `vehicle_crashed`, `station_first_vehicle`, `industry_open`/`industry_close`, `town_founded`, `company_new`/`company_in_trouble`/`company_bankrupt` and the four `subsidy_*` kinds — are engine events forwarded verbatim from `GSEventController`. Only those the engine actually raises for a **deity** script are available: `ET_VEHICLE_LOST`, `ET_VEHICLE_WAITING_IN_DEPOT` and `ET_VEHICLE_UNPROFITABLE` are `@api ai` only and can never reach a GameScript, so the bridge does not offer them. Unlike the timetable and dispatch queries, none of this needs a server patch — every getter used is stock GameScript API.
+
 ## Vehicle Orders & Timetables (Game Port DoCommands)
 Unlike vehicle listing, a vehicle's order list, timetables and scheduled dispatch have no writable GameScript API surface (this project adds read-only timetable and dispatch getters via server patches — see "Timetable Query" and "Dispatch Query" above). Reading and modifying them requires real engine commands (`DoCommand`s) sent over the **game port** (TCP 3979) via `ClientCommand`/`ServerCommand` packets, not the Admin Network. This section covers the wire format; for how to call the methods and what each parameter means, see the [Vehicle Timetables Usage Guide](TIMETABLES.md).
 

docs/TESTING.md

@@ -13,7 +13,9 @@ The tests are located in the `tests/` directory:
 | [`test_admin.py`](file:///home/kovagoadi/openttd-client/tests/test_admin.py) | `OpenTTDAdminClient` | Tests admin client initialization, admin packet types, and basic protocol constants. |
 | [`test_protocol.py`](file:///home/kovagoadi/openttd-client/tests/test_protocol.py) | `OpenTTDProtocol` | Tests binary serialization, custom parsers, and stream encryption/decryption (XChaCha20-Poly1305). |
 | [`test_logic.py`](file:///home/kovagoadi/openttd-client/tests/test_logic.py) | `OpenTTDClient` | Tests client connection lifecycle, company joining flow, authentication, and state management. |
+| [`test_events.py`](file:///home/kovagoadi/openttd-client/tests/test_events.py) | Game Events | Tests event subscription encoding, the push/pull consumption paths (`on_event`, `wait_for_event()`), buffering and waiter lifecycle. |
 | [`test_coverage.py`](file:///home/kovagoadi/openttd-client/tests/test_coverage.py) | Coverage Helpers | Auxiliary unit tests targeting connection errors, fallback packet handlers, and missing passwords to ensure high test coverage. |
+| [`test_gamescript.py`](file:///home/kovagoadi/openttd-client/tests/test_gamescript.py) | AdminBridge GameScript | Reads `gamescript/AdminBridge/*.nut` and checks it against the client: protocol version, event catalogue, command table, and that the docker setup serves the tracked copy. No Squirrel toolchain needed — see [`gamescript/AdminBridge/README.md`](file:///home/kovagoadi/openttd-client/gamescript/AdminBridge/README.md). |
 | [`test_e2e.py`](file:///home/kovagoadi/openttd-client/tests/test_e2e.py) | Integration / E2E | Connects to a running local OpenTTD server (e.g., in Docker) to verify full socket interactions, stream cryptography, and keep-alive frames. |
 
 ---

docs/TIMETABLES.md

@@ -65,10 +65,11 @@ vehicle id; `response_too_large` if a very long order list overflows the admin p
 don't run while the game is **paused**, so a paused server always times out), and
 `ConnectionError` if the admin connection drops mid-query.
 
-Requirements: the server must run the bundled AdminBridge GameScript **and** the patched JGRPP
+Requirements: the server must run the bundled AdminBridge GameScript (from
-build with the `GSOrder` timetable getters (both included in this repo's `docker/` setup — see
+`gamescript/AdminBridge/`, mounted into the container by this repo's `docker/` setup — see
-`docker/patches/README.md`). The Gamescript update-frequency subscription it needs is set up
+[`gamescript/AdminBridge/README.md`](../gamescript/AdminBridge/README.md)) **and** the patched
-automatically on first call.
+JGRPP build with the `GSOrder` timetable getters (`docker/patches/README.md`). The Gamescript
+update-frequency subscription it needs is set up automatically on first call.
 
 ## Writing (and the legacy observer): things you must know
 

gamescript/AdminBridge/README.md

@@ -0,0 +1,88 @@
+# AdminBridge GameScript
+
+The server-side half of this project. Every `OpenTTDAdminClient` feature that goes over the
+GameScript JSON channel — `list_vehicles`, `list_stations`, `list_cargo`, `get_timetable`,
+`get_station`, `get_station_cargo`, `get_dispatch` and the `subscribe_events` push stream — is a
+command this script answers. The Python client is only half the protocol; this is the other half,
+and the two are documented together in [docs/PROTOCOL.md](../../docs/PROTOCOL.md).
+
+It is a **deity** GameScript: it joins no company, builds nothing, and only reads game state and
+replies on the admin port via `GSAdmin.Send()`.
+
+- `info.nut` — the manifest OpenTTD's script scanner reads (name, version, API version).
+- `main.nut` — the bridge itself: a command dispatch table, one handler per command, and the
+  event poller.
+
+## Why it lives here
+
+`docker/config/` is gitignored — it also holds savegames, downloaded content and generated
+config — so a GameScript kept there is invisible to review and to CI, and a fresh clone cannot
+reproduce the server side at all. This directory is the source of truth, the same arrangement
+[docker/patches/](../../docker/patches/README.md) uses for the local JGRPP patches.
+
+`docker-compose.yml` bind-mounts this directory read-only over the container's
+`game/AdminBridge`, so the server runs the tracked copy and nothing else:
+
+```yaml
+- ../gamescript/AdminBridge:/home/openttd/.local/share/openttd/game/AdminBridge:ro
+```
+
+Editing the copy under `docker/config/game/AdminBridge/` therefore has no effect; that path is
+shadowed by the mount. On a server not started through this compose file, copy the directory
+into the OpenTTD data dir instead:
+
+```bash
+cp -r gamescript/AdminBridge ~/.local/share/openttd/game/
+```
+
+A GameScript is loaded when the game starts, so a change needs the server restarted (or the
+game reloaded) before it takes effect — the running instance keeps the old code.
+
+## Protocol version
+
+`info.nut`'s `GetVersion()` is the version of the **JSON protocol**, not of the implementation:
+bump it when a command, field or event kind changes shape, not for a refactor. Three places
+carry it and all three must move together:
+
+| Where | What |
+| :--- | :--- |
+| `info.nut` → `GetVersion()` | what OpenTTD records in the savegame |
+| `main.nut` → `BRIDGE_VERSION` | what the `get_version` command reports |
+| `lib/openttd/protocol.py` → `GS_BRIDGE_VERSION` | what the client requires |
+
+A GameScript cannot read its own `info.nut` at runtime — `GSController.GetVersion()` returns the
+*OpenTTD* version — hence the duplication. [`tests/test_gamescript.py`](../../tests/test_gamescript.py)
+pins the three together, along with the event catalogue and the command list, so a half-finished
+bump fails in CI rather than against a live server.
+
+`OpenTTDAdminClient.get_bridge_version()` is the client end of this: it asks the running bridge
+and raises if it is too old, which is worth doing once at startup. Without it a stale bridge
+gives no error at all — it does not recognise the command, so the client just waits out its
+timeout. A bridge older than version 4 predates `get_version` itself and can only fail that way.
+
+## Requirements
+
+- **JGRPP**, patched with `docker/patches/` — `get_timetable` and `get_dispatch` call `GSOrder`
+  getters those patches add. Every other command uses stock GameScript API, so an unpatched
+  server still answers them.
+- The admin client must subscribe with `update_frequency(Gamescript, Automatic)` or the server
+  drops every reply. The client does this automatically on first use.
+- GameScripts do not tick while the game is **paused**, so a paused server answers nothing and
+  every query times out.
+
+## Which GameScript actually runs
+
+OpenTTD runs exactly **one** GameScript, and a savegame remembers the one it was played with by
+name: loading it overrides `[game_scripts]` in `openttd.cfg`. So a savegame that pins some other
+script leaves this bridge unloaded, and every command times out no matter what is mounted where.
+The symptom to recognise is commands timing out *uniformly* — as opposed to a stale bridge, where
+the older commands still answer and only newer ones hang.
+
+## Verifying a change
+
+`docker/config` being untracked used to mean this file could only be tested against a running
+server. It still needs one for the real check — `pytest -m e2e` exercises every command end to
+end, and [docs/TESTING.md](../../docs/TESTING.md) covers starting the server. Before that,
+`pytest -m "not e2e" tests/test_gamescript.py` catches the drift a server would only reveal as a
+timeout. Note that a syntax error anywhere in `main.nut` stops the whole bridge from loading, so
+every command fails at once — check the server log (`docker compose logs`) for the compile error.

gamescript/AdminBridge/info.nut

@@ -0,0 +1,19 @@
+class AdminBridge extends GSInfo {
+	function GetAuthor()      { return "Gemini"; }
+	function GetName()        { return "AdminBridge"; }
+	function GetShortName()   { return "ADMB"; }
+	function GetDescription() { return "Answers JSON commands from the admin port on behalf of an admin client."; }
+	/* Bridge protocol version. Must equal BRIDGE_VERSION in main.nut, which is what the
+	 * get_version command reports; tests/test_gamescript.py pins the two together and to
+	 * openttd.protocol.GS_BRIDGE_VERSION. Bump it whenever the JSON protocol changes. */
+	function GetVersion()     { return 4; }
+	/* Any version of this script can take over from any older one: the bridge keeps no
+	 * savegame state of its own (it has no Save()), so there is nothing to migrate. Without
+	 * this, the default is GetVersion(), and loading a savegame that pinned an older version
+	 * makes the engine fall back with a "no longer available" warning. */
+	function MinVersionToLoad() { return 1; }
+	function GetAPIVersion()  { return "1.10"; }
+	function GetDate()        { return "2026-08-31"; }
+	function CreateInstance() { return "AdminBridge"; }
+}
+RegisterGS(AdminBridge());

gamescript/AdminBridge/main.nut

@@ -0,0 +1,722 @@
+class AdminBridge extends GSController {
+	/* --- Event subscription state (see HandleSubscribeEvents) --- */
+	event_kinds = null;        // table of subscribed kind -> true; null when not subscribed
+	event_order = null;        // canonical kind ordering, so subscribe replies are stable
+	event_catalog = null;      // table of every supported kind -> true, for validation
+	interval = 10;             // ticks between state polls
+	sleep_ticks = 10;          // ticks slept per main-loop iteration
+	last_poll = 0;             // tick of the last state poll
+	seeded = false;            // false until the first poll has recorded a state baseline
+	company_id = null;         // optional owner filter for the polled vehicle/station events
+	watch_vehicles = null;     // explicit vehicle ids to poll; null = every vehicle
+	watch_stations = null;     // explicit station ids to poll; null = every station
+	cargo_ids = null;          // cargo ids to inspect; set at subscribe time
+	min_cargo_delta = 1;       // smallest waiting-amount change worth a cargo_waiting event
+	include_cargo = true;      // attach each vehicle's load to arrive/depart events
+	vehicle_at = null;         // vehicle id -> { station = id or -1, since = tick }
+	cargo_prev = null;         // station id -> table of cargo id -> last seen waiting amount
+
+	/* Cap on how many events one poll may emit. A first poll over a busy map can otherwise
+	 * produce thousands at once; past the cap the rest are dropped and reported as such. */
+	MAX_EVENTS_PER_POLL = 200;
+	/* Events per admin packet. Kept well under the packet size limit so a batch always fits. */
+	EVENT_BATCH_SIZE = 24;
+
+	/* Version of the JSON protocol spoken over the admin port, reported by get_version.
+	 * A GameScript cannot read its own info.nut at runtime (GSController.GetVersion() returns
+	 * the *OpenTTD* version), so this duplicates info.nut's GetVersion() and the two are pinned
+	 * together — along with the client's openttd.protocol.GS_BRIDGE_VERSION — by
+	 * tests/test_gamescript.py. Bump all three whenever the protocol changes. */
+	BRIDGE_VERSION = 4;
+
+	/* Every command the bridge answers: name -> handler method plus the request fields that
+	 * handler needs. HandleCommand dispatches through this and get_version reports it, so the
+	 * catalogue a client feature-detects against cannot drift from what is implemented. */
+	COMMANDS = {
+		get_version        = { handler = "HandleGetVersion",        requires = [] },
+		list_vehicles      = { handler = "HandleListVehicles",      requires = [] },
+		list_stations      = { handler = "HandleListStations",      requires = [] },
+		list_cargo         = { handler = "HandleListCargo",         requires = [] },
+		get_timetable      = { handler = "HandleGetTimetable",      requires = ["vehicle_id"] },
+		get_station        = { handler = "HandleGetStation",        requires = ["station_id"] },
+		get_station_cargo  = { handler = "HandleGetStationCargo",   requires = ["station_id", "cargo_id"] },
+		get_dispatch       = { handler = "HandleGetDispatch",       requires = ["vehicle_id"] },
+		subscribe_events   = { handler = "HandleSubscribeEvents",   requires = [] },
+		unsubscribe_events = { handler = "HandleUnsubscribeEvents", requires = [] },
+	};
+
+	function InitEvents() {
+		this.ResetEventState();
+		this.event_order = [
+			/* Synthesised by polling game state (PollState). */
+			"vehicle_arrive", "vehicle_depart", "cargo_waiting",
+			/* Native GameScript events, forwarded as they arrive (HandleNativeEvent). */
+			"vehicle_crashed", "station_first_vehicle",
+			"industry_open", "industry_close", "town_founded",
+			"company_new", "company_in_trouble", "company_bankrupt",
+			"subsidy_offer", "subsidy_offer_expired", "subsidy_awarded", "subsidy_expired"
+		];
+		this.event_catalog = {};
+		foreach (kind in this.event_order) this.event_catalog[kind] <- true;
+	}
+
+	function Start() {
+		GSLog.Info("AdminBridge started.");
+		this.InitEvents();
+		while (true) {
+			this.Sleep(this.sleep_ticks);
+			local pending = [];
+			local event = GSEventController.GetNextEvent();
+			while (event != null) {
+				if (event.GetEventType() == GSEvent.ET_ADMIN_PORT) {
+					local admin_event = GSEventAdminPort.Convert(event);
+					local request = admin_event.GetObject();
+					if (request != null && "command" in request) this.HandleCommand(request);
+				} else if (this.event_kinds != null) {
+					this.HandleNativeEvent(event, pending);
+				}
+				event = GSEventController.GetNextEvent();
+			}
+			if (this.event_kinds != null) this.PollState(pending);
+			this.SendEventBatch(pending);
+		}
+	}
+
+	function HandleCommand(request) {
+		if (!(request.command in this.COMMANDS)) return;
+		local spec = this.COMMANDS[request.command];
+		/* A command missing the fields its handler needs is dropped rather than answered:
+		 * there is nothing to answer *about*, and no request_id is guaranteed either. */
+		foreach (field in spec.requires) {
+			if (!(field in request)) return;
+		}
+		this[spec.handler](request);
+	}
+
+	/* Version handshake. A client expecting a newer bridge than the server runs would otherwise
+	 * just time out on the first command this bridge does not know; asking here turns that into
+	 * an answer. The two catalogues let a client feature-detect, which survives a bridge that
+	 * gained commands out of version order better than comparing numbers does. */
+	function HandleGetVersion(request) {
+		local commands = [];
+		foreach (name, _ in this.COMMANDS) commands.append(name);
+		commands.sort();   // table iteration order is arbitrary; keep the reply stable
+		local events = [];
+		foreach (kind in this.event_order) events.append(kind);
+		local reply = { command = "get_version", version = this.BRIDGE_VERSION,
+			commands = commands, events = events };
+		if ("request_id" in request) reply.request_id <- request.request_id;
+		GSAdmin.Send(reply);
+	}
+
+	/* Cargo id -> label ("PASS", "COAL", ...) so callers can name what a vehicle carries.
+	 * Labels come from the NewGRF cargo table, so ids are not stable across games. */
+	function HandleListCargo(request) {
+		local cargo = [];
+		foreach (c, _ in GSCargoList()) {
+			local label = "";
+			try { label = GSCargo.GetCargoLabel(c); } catch (e) { label = ""; }
+			local freight = 0;
+			try { freight = GSCargo.IsFreight(c) ? 1 : 0; } catch (e) { freight = 0; }
+			cargo.append({ cargo_id = c, label = label, freight = freight });
+		}
+		local reply = { command = "list_cargo", cargo = cargo };
+		if ("request_id" in request) reply.request_id <- request.request_id;
+		GSAdmin.Send(reply);
+	}
+
+	function HandleListVehicles(request) {
+		local vehicle_list = GSVehicleList();
+		local vehicles = [];
+		foreach (v, _ in vehicle_list) {
+			if (!GSVehicle.IsValidVehicle(v)) continue;
+			local next_stop = "None";
+			if (GSOrder.GetOrderCount(v) > 0) {
+				local dest_tile = GSOrder.GetOrderDestination(v, GSOrder.ORDER_CURRENT);
+				local station_id = GSStation.GetStationID(dest_tile);
+				if (GSStation.IsValidStation(station_id)) next_stop = GSStation.GetName(station_id);
+			}
+			/* Which cargoes this vehicle can actually carry. A vehicle (an articulated
+			 * train especially) can have capacity for several cargo types, so report
+			 * every one with a non-zero capacity rather than a single "cargo type". */
+			local cargo = [];
+			foreach (c, _ in GSCargoList()) {
+				local cap = GSVehicle.GetCapacity(v, c);
+				if (cap > 0) cargo.append({ cargo_id = c, capacity = cap });
+			}
+			vehicles.append({
+				id = v,
+				age = GSVehicle.GetAge(v),
+				max_age = GSVehicle.GetMaxAge(v),
+				next_stop = next_stop,
+				type = GSVehicle.GetVehicleType(v),
+				owner = GSVehicle.GetOwner(v),
+				order_count = GSOrder.GetOrderCount(v),
+				cargo = cargo
+			});
+		}
+		local reply = { vehicles = vehicles };
+		if ("request_id" in request) reply.request_id <- request.request_id;
+		GSAdmin.Send(reply);
+	}
+
+	function HandleGetTimetable(request) {
+		local v = request.vehicle_id;
+		local reply = { command = "get_timetable", vehicle_id = v };
+		if ("request_id" in request) reply.request_id <- request.request_id;
+		if (!GSVehicle.IsValidVehicle(v)) {
+			reply.error <- "invalid_vehicle";
+			GSAdmin.Send(reply);
+			return;
+		}
+		reply.lateness <- GSOrder.GetTimetableLateness(v);
+		reply.start_tick <- GSOrder.GetTimetableStartTick(v);
+		reply.current_order_time <- GSOrder.GetCurrentOrderTime(v);
+		reply.total_duration <- GSOrder.GetTimetableTotalDuration(v);
+		local orders = [];
+		local count = GSOrder.GetOrderCount(v);
+		for (local i = 0; i < count; i++) {
+			/* Resolve the order's destination station so callers can build a station
+			 * graph (vertices = stations, edges = segments). Non-station orders
+			 * (depots, waypoints, conditional) have no station: report -1. */
+			local station_id = -1;
+			if (GSOrder.IsValidVehicleOrder(v, i) && GSOrder.IsGotoStationOrder(v, i)) {
+				local sid = GSStation.GetStationID(GSOrder.GetOrderDestination(v, i));
+				if (GSStation.IsValidStation(sid)) station_id = sid;
+			}
+			orders.append({
+				position = i,
+				station_id = station_id,
+				wait_time = GSOrder.GetTimetableWaitTime(v, i),
+				travel_time = GSOrder.GetTimetableTravelTime(v, i),
+				wait_timetabled = GSOrder.IsWaitTimetabled(v, i) ? 1 : 0,
+				travel_timetabled = GSOrder.IsTravelTimetabled(v, i) ? 1 : 0,
+				wait_fixed = GSOrder.IsWaitFixed(v, i) ? 1 : 0,
+				travel_fixed = GSOrder.IsTravelFixed(v, i) ? 1 : 0,
+				leave_type = GSOrder.GetLeaveType(v, i),
+				max_speed = GSOrder.GetTimetableMaxSpeed(v, i)
+			});
+		}
+		reply.orders <- orders;
+		if (!GSAdmin.Send(reply)) {
+			/* Response exceeded the admin packet size limit; send a small error
+			 * instead so the waiting client fails fast rather than timing out. */
+			local fallback = { command = "get_timetable", vehicle_id = v, error = "response_too_large" };
+			if ("request_id" in request) fallback.request_id <- request.request_id;
+			GSAdmin.Send(fallback);
+		}
+	}
+
+	function HandleListStations(request) {
+		/* GameScripts run as a deity, so GSStationList lists every company's stations;
+		 * an optional company_id filters to a single owner (the deity list ignores it). */
+		local filter_owner = ("company_id" in request) ? request.company_id : null;
+		local station_list = GSStationList(GSStation.STATION_ANY);
+		local stations = [];
+		foreach (s, _ in station_list) {
+			if (!GSStation.IsValidStation(s)) continue;
+			local owner = GSStation.GetOwner(s);
+			if (filter_owner != null && owner != filter_owner) continue;
+			stations.append({
+				id = s,
+				name = GSStation.GetName(s),
+				location = GSStation.GetLocation(s),
+				owner = owner
+			});
+		}
+		local reply = { command = "list_stations", stations = stations };
+		if ("request_id" in request) reply.request_id <- request.request_id;
+		GSAdmin.Send(reply);
+	}
+
+	function HandleGetStation(request) {
+		local sid = request.station_id;
+		local reply = { command = "get_station", station_id = sid };
+		if ("request_id" in request) reply.request_id <- request.request_id;
+		if (!GSStation.IsValidStation(sid)) {
+			reply.error <- "invalid_station";
+			GSAdmin.Send(reply);
+			return;
+		}
+		reply.name <- GSStation.GetName(sid);
+		reply.location <- GSStation.GetLocation(sid);
+		reply.owner <- GSStation.GetOwner(sid);
+		local cargo = [];
+		local cargo_list = GSCargoList();
+		foreach (c, _ in cargo_list) {
+			local waiting = GSStation.GetCargoWaiting(sid, c);
+			local planned = GSStation.GetCargoPlanned(sid, c);
+			local has_rating = GSStation.HasCargoRating(sid, c);
+			/* Skip cargo the station has never handled to keep the reply compact. */
+			if (waiting <= 0 && planned <= 0 && !has_rating) continue;
+			cargo.append({
+				cargo_id = c,
+				waiting = waiting,      // real-time: units currently waiting
+				planned = planned,      // planned: cargodist link-graph flow
+				rating = has_rating ? GSStation.GetCargoRating(sid, c) : null
+			});
+		}
+		reply.cargo <- cargo;
+		if (!GSAdmin.Send(reply)) {
+			/* Response exceeded the admin packet size limit; send a small error
+			 * instead so the waiting client fails fast rather than timing out. */
+			local fallback = { command = "get_station", station_id = sid, error = "response_too_large" };
+			if ("request_id" in request) fallback.request_id <- request.request_id;
+			GSAdmin.Send(fallback);
+		}
+	}
+
+	/* Convert a GSList of station_id -> amount into an array of {station, amount},
+	 * dropping zero entries to keep the reply compact. */
+	function CargoListToPairs(list) {
+		local pairs = [];
+		foreach (station, _ in list) {
+			local amount = list.GetValue(station);
+			if (amount <= 0) continue;
+			pairs.append({ station = station, amount = amount });
+		}
+		return pairs;
+	}
+
+	function HandleGetStationCargo(request) {
+		local sid = request.station_id;
+		local cargo = request.cargo_id;
+		local reply = { command = "get_station_cargo", station_id = sid, cargo_id = cargo };
+		if ("request_id" in request) reply.request_id <- request.request_id;
+		if (!GSStation.IsValidStation(sid)) {
+			reply.error <- "invalid_station";
+			GSAdmin.Send(reply);
+			return;
+		}
+		if (!GSCargo.IsValidCargo(cargo)) {
+			reply.error <- "invalid_cargo";
+			GSAdmin.Send(reply);
+			return;
+		}
+
+		/* Optional source (from) and next-hop (via) filters. STATION_INVALID is a legal value
+		 * (deleted source / "via any"); any other non-station value is rejected. */
+		local from = null;
+		local via = null;
+		if ("from_station" in request) {
+			from = request.from_station;
+			if (from != GSStation.STATION_INVALID && !GSStation.IsValidStation(from)) {
+				reply.error <- "invalid_from_station";
+				GSAdmin.Send(reply);
+				return;
+			}
+			reply.from_station <- from;
+		}
+		if ("via_station" in request) {
+			via = request.via_station;
+			if (via != GSStation.STATION_INVALID && !GSStation.IsValidStation(via)) {
+				reply.error <- "invalid_via_station";
+				GSAdmin.Send(reply);
+				return;
+			}
+			reply.via_station <- via;
+		}
+
+		/* Totals, honouring whichever filters were supplied. */
+		if (from != null && via != null) {
+			reply.waiting <- GSStation.GetCargoWaitingFromVia(sid, from, via, cargo);
+			reply.planned <- GSStation.GetCargoPlannedFromVia(sid, from, via, cargo);
+		} else if (from != null) {
+			reply.waiting <- GSStation.GetCargoWaitingFrom(sid, from, cargo);
+			reply.planned <- GSStation.GetCargoPlannedFrom(sid, from, cargo);
+		} else if (via != null) {
+			reply.waiting <- GSStation.GetCargoWaitingVia(sid, via, cargo);
+			reply.planned <- GSStation.GetCargoPlannedVia(sid, via, cargo);
+		} else {
+			reply.waiting <- GSStation.GetCargoWaiting(sid, cargo);
+			reply.planned <- GSStation.GetCargoPlanned(sid, cargo);
+		}
+
+		/* Breakdown grouped by source station (restricted to the via filter if given). */
+		local w_by_from = (via != null)
+			? GSStationList_CargoWaitingViaByFrom(sid, cargo, via)
+			: GSStationList_CargoWaitingByFrom(sid, cargo);
+		local p_by_from = (via != null)
+			? GSStationList_CargoPlannedViaByFrom(sid, cargo, via)
+			: GSStationList_CargoPlannedByFrom(sid, cargo);
+		reply.waiting_by_from <- this.CargoListToPairs(w_by_from);
+		reply.planned_by_from <- this.CargoListToPairs(p_by_from);
+
+		/* Breakdown grouped by next hop (restricted to the from filter if given). */
+		local w_by_via = (from != null)
+			? GSStationList_CargoWaitingFromByVia(sid, cargo, from)
+			: GSStationList_CargoWaitingByVia(sid, cargo);
+		local p_by_via = (from != null)
+			? GSStationList_CargoPlannedFromByVia(sid, cargo, from)
+			: GSStationList_CargoPlannedByVia(sid, cargo);
+		reply.waiting_by_via <- this.CargoListToPairs(w_by_via);
+		reply.planned_by_via <- this.CargoListToPairs(p_by_via);
+
+		if (!GSAdmin.Send(reply)) {
+			/* Response exceeded the admin packet size limit; send a small error
+			 * instead so the waiting client fails fast rather than timing out. */
+			local fallback = { command = "get_station_cargo", station_id = sid, cargo_id = cargo, error = "response_too_large" };
+			if ("request_id" in request) fallback.request_id <- request.request_id;
+			GSAdmin.Send(fallback);
+		}
+	}
+
+	function HandleGetDispatch(request) {
+		local v = request.vehicle_id;
+		local reply = { command = "get_dispatch", vehicle_id = v };
+		if ("request_id" in request) reply.request_id <- request.request_id;
+
+		local count = GSOrder.GetScheduledDispatchScheduleCount(v);
+		if (count < 0) {
+			reply.error <- "invalid_vehicle";
+			GSAdmin.Send(reply);
+			return;
+		}
+		reply.enabled <- GSOrder.IsScheduledDispatchEnabled(v);
+		local schedules = [];
+		for (local s = 0; s < count; s++) {
+			local slots = [];
+			local slot_count = GSOrder.GetScheduledDispatchSlotCount(v, s);
+			for (local k = 0; k < slot_count; k++) {
+				slots.append({
+					offset = GSOrder.GetScheduledDispatchSlotOffset(v, s, k),
+					flags = GSOrder.GetScheduledDispatchSlotFlags(v, s, k)
+				});
+			}
+			schedules.append({
+				index = s,
+				duration = GSOrder.GetScheduledDispatchDuration(v, s),
+				start_tick = GSOrder.GetScheduledDispatchStartTick(v, s),
+				delay = GSOrder.GetScheduledDispatchDelay(v, s),
+				reuse_slots = GSOrder.GetScheduledDispatchReuseSlots(v, s),
+				slots = slots
+			});
+		}
+		reply.schedules <- schedules;
+		if (!GSAdmin.Send(reply)) {
+			/* Response exceeded the admin packet size limit; send a small error
+			 * instead so the waiting client fails fast rather than timing out. */
+			local fallback = { command = "get_dispatch", vehicle_id = v, error = "response_too_large" };
+			if ("request_id" in request) fallback.request_id <- request.request_id;
+			GSAdmin.Send(fallback);
+		}
+	}
+
+	/* --- Events ---
+	 *
+	 * The engine has no GameScript event for "a vehicle reached a stop" or "cargo arrived",
+	 * so those are synthesised here: every `interval` ticks the bridge samples the watched
+	 * vehicles and stations and emits an event for each change against the previous sample.
+	 * The handful of events the engine *does* raise for a deity script (crashes, industries,
+	 * companies, ...) are forwarded straight through. Everything is pushed to the admin port
+	 * unsolicited, batched as { command = "events", events = [...] }. */
+
+	function ResetEventState() {
+		this.event_kinds = null;
+		this.interval = 10;
+		this.sleep_ticks = 10;
+		this.last_poll = 0;
+		this.seeded = false;
+		this.company_id = null;
+		this.watch_vehicles = null;
+		this.watch_stations = null;
+		this.cargo_ids = null;
+		this.min_cargo_delta = 1;
+		this.include_cargo = true;
+		this.vehicle_at = {};
+		this.cargo_prev = {};
+	}
+
+	/* Copy a request field that must be an array of integers, or null when absent. */
+	function ReadIdList(request, key) {
+		if (!(key in request) || request[key] == null) return null;
+		local out = [];
+		foreach (id in request[key]) out.append(id);
+		return out;
+	}
+
+	function HandleSubscribeEvents(request) {
+		local reply = { command = "subscribe_events" };
+		if ("request_id" in request) reply.request_id <- request.request_id;
+
+		/* Validate everything before touching the live subscription, so a rejected request
+		 * leaves whatever was subscribed before running untouched. */
+		local kinds = {};
+		if ("events" in request && request.events != null) {
+			foreach (kind in request.events) {
+				if (!(kind in this.event_catalog)) {
+					reply.error <- "unknown_event";
+					reply.event <- kind;
+					GSAdmin.Send(reply);
+					return;
+				}
+				kinds[kind] <- true;
+			}
+		} else {
+			foreach (kind in this.event_order) kinds[kind] <- true;
+		}
+
+		local new_interval = ("interval" in request) ? request.interval : 10;
+		if (typeof new_interval != "integer" || new_interval < 1) {
+			reply.error <- "invalid_interval";
+			GSAdmin.Send(reply);
+			return;
+		}
+		local new_min_delta = ("min_cargo_delta" in request) ? request.min_cargo_delta : 1;
+		if (typeof new_min_delta != "integer" || new_min_delta < 1) {
+			reply.error <- "invalid_min_cargo_delta";
+			GSAdmin.Send(reply);
+			return;
+		}
+
+		/* Cargo ids are validated up front because the amount getters below take them as a
+		 * precondition; station and vehicle ids are not, since they can vanish mid-subscription
+		 * anyway and are re-checked on every poll. */
+		local new_cargo = this.ReadIdList(request, "cargo");
+		if (new_cargo != null) {
+			foreach (c in new_cargo) {
+				if (!GSCargo.IsValidCargo(c)) {
+					reply.error <- "invalid_cargo";
+					reply.cargo_id <- c;
+					GSAdmin.Send(reply);
+					return;
+				}
+			}
+		} else {
+			new_cargo = [];
+			foreach (c, _ in GSCargoList()) new_cargo.append(c);
+		}
+
+		this.ResetEventState();
+		this.event_kinds = kinds;
+		this.interval = new_interval;
+		this.sleep_ticks = new_interval < 10 ? new_interval : 10;
+		this.min_cargo_delta = new_min_delta;
+		this.cargo_ids = new_cargo;
+		this.watch_vehicles = this.ReadIdList(request, "vehicles");
+		this.watch_stations = this.ReadIdList(request, "stations");
+		if ("company_id" in request && request.company_id != null) this.company_id = request.company_id;
+		if ("include_cargo" in request) this.include_cargo = request.include_cargo ? true : false;
+
+		local accepted = [];
+		foreach (kind in this.event_order) {
+			if (kind in this.event_kinds) accepted.append(kind);
+		}
+		reply.events <- accepted;
+		reply.interval <- this.interval;
+		GSAdmin.Send(reply);
+	}
+
+	function HandleUnsubscribeEvents(request) {
+		this.ResetEventState();
+		local reply = { command = "unsubscribe_events", events = [] };
+		if ("request_id" in request) reply.request_id <- request.request_id;
+		GSAdmin.Send(reply);
+	}
+
+	function Wants(kind) {
+		return this.event_kinds != null && (kind in this.event_kinds);
+	}
+
+	/* Forward the events the engine raises for a deity GameScript. Vehicle "lost", "waiting in
+	 * depot" and "unprofitable" are deliberately absent: the engine only ever raises those for
+	 * AI companies, so a GameScript can never observe them. */
+	function HandleNativeEvent(event, out) {
+		local type = event.GetEventType();
+		local tick = this.GetTick();
+		if (type == GSEvent.ET_VEHICLE_CRASHED) {
+			if (!this.Wants("vehicle_crashed")) return;
+			local e = GSEventVehicleCrashed.Convert(event);
+			out.append({ event = "vehicle_crashed", tick = tick, vehicle_id = e.GetVehicleID(),
+				tile = e.GetCrashSite(), reason = e.GetCrashReason(),
+				victims = e.GetVictims(), owner = e.GetVehicleOwner() });
+		} else if (type == GSEvent.ET_STATION_FIRST_VEHICLE) {
+			if (!this.Wants("station_first_vehicle")) return;
+			local e = GSEventStationFirstVehicle.Convert(event);
+			out.append({ event = "station_first_vehicle", tick = tick,
+				station_id = e.GetStationID(), vehicle_id = e.GetVehicleID() });
+		} else if (type == GSEvent.ET_INDUSTRY_OPEN) {
+			if (!this.Wants("industry_open")) return;
+			out.append({ event = "industry_open", tick = tick,
+				industry_id = GSEventIndustryOpen.Convert(event).GetIndustryID() });
+		} else if (type == GSEvent.ET_INDUSTRY_CLOSE) {
+			if (!this.Wants("industry_close")) return;
+			out.append({ event = "industry_close", tick = tick,
+				industry_id = GSEventIndustryClose.Convert(event).GetIndustryID() });
+		} else if (type == GSEvent.ET_TOWN_FOUNDED) {
+			if (!this.Wants("town_founded")) return;
+			out.append({ event = "town_founded", tick = tick,
+				town_id = GSEventTownFounded.Convert(event).GetTownID() });
+		} else if (type == GSEvent.ET_COMPANY_NEW) {
+			if (!this.Wants("company_new")) return;
+			out.append({ event = "company_new", tick = tick,
+				company_id = GSEventCompanyNew.Convert(event).GetCompanyID() });
+		} else if (type == GSEvent.ET_COMPANY_IN_TROUBLE) {
+			if (!this.Wants("company_in_trouble")) return;
+			out.append({ event = "company_in_trouble", tick = tick,
+				company_id = GSEventCompanyInTrouble.Convert(event).GetCompanyID() });
+		} else if (type == GSEvent.ET_COMPANY_BANKRUPT) {
+			if (!this.Wants("company_bankrupt")) return;
+			out.append({ event = "company_bankrupt", tick = tick,
+				company_id = GSEventCompanyBankrupt.Convert(event).GetCompanyID() });
+		} else if (type == GSEvent.ET_SUBSIDY_OFFER) {
+			if (!this.Wants("subsidy_offer")) return;
+			out.append({ event = "subsidy_offer", tick = tick,
+				subsidy_id = GSEventSubsidyOffer.Convert(event).GetSubsidyID() });
+		} else if (type == GSEvent.ET_SUBSIDY_OFFER_EXPIRED) {
+			if (!this.Wants("subsidy_offer_expired")) return;
+			out.append({ event = "subsidy_offer_expired", tick = tick,
+				subsidy_id = GSEventSubsidyOfferExpired.Convert(event).GetSubsidyID() });
+		} else if (type == GSEvent.ET_SUBSIDY_AWARDED) {
+			if (!this.Wants("subsidy_awarded")) return;
+			out.append({ event = "subsidy_awarded", tick = tick,
+				subsidy_id = GSEventSubsidyAwarded.Convert(event).GetSubsidyID() });
+		} else if (type == GSEvent.ET_SUBSIDY_EXPIRED) {
+			if (!this.Wants("subsidy_expired")) return;
+			out.append({ event = "subsidy_expired", tick = tick,
+				subsidy_id = GSEventSubsidyExpired.Convert(event).GetSubsidyID() });
+		}
+	}
+
+	function PollState(out) {
+		local now = this.GetTick();
+		if (this.seeded && now - this.last_poll < this.interval) return;
+		this.last_poll = now;
+		this.PollVehicles(now, out);
+		this.PollCargo(now, out);
+		/* The first poll only records where everything already is: a vehicle that was sitting
+		 * at a station when the subscription started did not just arrive. */
+		this.seeded = true;
+	}
+
+	function WatchedVehicles() {
+		if (this.watch_vehicles != null) return this.watch_vehicles;
+		local out = [];
+		foreach (v, _ in GSVehicleList()) out.append(v);
+		return out;
+	}
+
+	function WatchedStations() {
+		if (this.watch_stations != null) return this.watch_stations;
+		local out = [];
+		foreach (s, _ in GSStationList(GSStation.STATION_ANY)) out.append(s);
+		return out;
+	}
+
+	/* Which station a vehicle is stopped at, or -1 when it is not loading at one. Note that a
+	 * vehicle stopped by hand or broken down at a platform reports its own state instead, so it
+	 * reads here as having left the station. */
+	function VehicleStation(v) {
+		if (GSVehicle.GetState(v) != GSVehicle.VS_AT_STATION) return -1;
+		local sid = GSStation.GetStationID(GSVehicle.GetLocation(v));
+		return GSStation.IsValidStation(sid) ? sid : -1;
+	}
+
+	function PollVehicles(now, out) {
+		local want_arrive = this.Wants("vehicle_arrive");
+		local want_depart = this.Wants("vehicle_depart");
+		if (!want_arrive && !want_depart) return;
+		local live = {};
+		foreach (v in this.WatchedVehicles()) {
+			if (!GSVehicle.IsValidVehicle(v)) continue;
+			if (this.company_id != null && GSVehicle.GetOwner(v) != this.company_id) continue;
+			live[v] <- true;
+			local at = this.VehicleStation(v);
+			local known = (v in this.vehicle_at) ? this.vehicle_at[v] : null;
+			local was = (known == null) ? -1 : known.station;
+			if (was == at) continue;
+			if (this.seeded) {
+				/* A vehicle that moves from one station straight to another in a single
+				 * sampling window yields both a depart and an arrive, in that order. */
+				if (was != -1 && want_depart) out.append(this.VehicleEvent("vehicle_depart", now, v, was, now - known.since));
+				if (at != -1 && want_arrive) out.append(this.VehicleEvent("vehicle_arrive", now, v, at, 0));
+			}
+			this.vehicle_at[v] <- { station = at, since = now };
+		}
+		/* Forget vehicles that were sold or fell out of the filter, so the table cannot grow
+		 * without bound over a long subscription. */
+		local stale = [];
+		foreach (v, _ in this.vehicle_at) {
+			if (!(v in live)) stale.append(v);
+		}
+		foreach (v in stale) delete this.vehicle_at[v];
+	}
+
+	function VehicleEvent(kind, tick, v, sid, dwell) {
+		local ev = {
+			event = kind,
+			tick = tick,
+			vehicle_id = v,
+			station_id = sid,
+			owner = GSVehicle.GetOwner(v),
+			vehicle_type = GSVehicle.GetVehicleType(v),
+			order_position = GSOrder.ResolveOrderPosition(v, GSOrder.ORDER_CURRENT)
+		};
+		/* How long the vehicle had been loading, in ticks. For a vehicle that was already at a
+		 * station when the subscription started this counts from the first poll, not from the
+		 * real arrival. */
+		if (kind == "vehicle_depart") ev.dwell <- dwell;
+		if (this.include_cargo) ev.cargo <- this.VehicleCargo(v);
+		return ev;
+	}
+
+	function VehicleCargo(v) {
+		local out = [];
+		foreach (c in this.cargo_ids) {
+			local load = GSVehicle.GetCargoLoad(v, c);
+			if (load > 0) out.append({ cargo_id = c, load = load });
+		}
+		return out;
+	}
+
+	function PollCargo(now, out) {
+		if (!this.Wants("cargo_waiting")) return;
+		local live = {};
+		foreach (sid in this.WatchedStations()) {
+			if (!GSStation.IsValidStation(sid)) continue;
+			if (this.company_id != null && GSStation.GetOwner(sid) != this.company_id) continue;
+			live[sid] <- true;
+			if (!(sid in this.cargo_prev)) this.cargo_prev[sid] <- {};
+			local prev = this.cargo_prev[sid];
+			foreach (c in this.cargo_ids) {
+				local waiting = GSStation.GetCargoWaiting(sid, c);
+				local before = (c in prev) ? prev[c] : 0;
+				if (waiting == before) continue;
+				prev[c] <- waiting;
+				if (!this.seeded) continue;
+				local delta = waiting - before;
+				local magnitude = delta < 0 ? -delta : delta;
+				if (magnitude < this.min_cargo_delta) continue;
+				out.append({ event = "cargo_waiting", tick = now, station_id = sid,
+					cargo_id = c, waiting = waiting, delta = delta });
+			}
+		}
+		local stale = [];
+		foreach (sid, _ in this.cargo_prev) {
+			if (!(sid in live)) stale.append(sid);
+		}
+		foreach (sid in stale) delete this.cargo_prev[sid];
+	}
+
+	function SendEventBatch(events) {
+		if (events.len() == 0) return;
+		local dropped = 0;
+		if (events.len() > this.MAX_EVENTS_PER_POLL) {
+			dropped = events.len() - this.MAX_EVENTS_PER_POLL;
+			events = events.slice(0, this.MAX_EVENTS_PER_POLL);
+		}
+		local sent = 0;
+		while (sent < events.len()) {
+			local end = sent + this.EVENT_BATCH_SIZE;
+			if (end > events.len()) end = events.len();
+			GSAdmin.Send({ command = "events", events = events.slice(sent, end) });
+			sent = end;
+		}
+		/* Tell the client its view has a hole in it rather than letting it silently miss
+		 * transitions it is counting on. */
+		if (dropped > 0) {
+			GSAdmin.Send({ command = "events",
+				events = [{ event = "events_dropped", tick = this.GetTick(), count = dropped }] });
+		}
+	}
+}

lib/openttd/client.py

@@ -3,6 +3,7 @@ import hashlib
 import logging
 import os
 import uuid
+from collections import deque
 from typing import ClassVar
 
 import monocypher
@@ -470,7 +471,7 @@ class OpenTTDClient:
 
 class OpenTTDAdminClient:
     """High-level OpenTTD Admin client."""
-    def __init__(self, host, port=3977, admin_name="GeminiAdmin"):
+    def __init__(self, host, port=3977, admin_name="GeminiAdmin", event_buffer_size=256):
         self.host = host
         self.port = port
         self.admin_name = admin_name
@@ -493,12 +494,19 @@ class OpenTTDAdminClient:
         self.on_chat = None
         self.on_console = None
         self.on_gamescript = None
+        self.on_event = None
 
         # GameScript request/response correlation
         self._gs_request_id = 0
         self._gs_futures = {}
         self._gs_subscribed = False
 
+        # Game events pushed by the AdminBridge GameScript. Events nobody is waiting for are
+        # kept here so a wait_for_event() call can still pick up something that arrived just
+        # before it; the deque bounds the memory a subscription nobody drains can cost.
+        self._event_buffer = deque(maxlen=event_buffer_size)
+        self._event_waiters = []
+
     async def connect(self, admin_password="", secure=False):
         """Connect to the admin port and initiate handshake."""
         self._admin_password = admin_password
@@ -532,6 +540,10 @@ class OpenTTDAdminClient:
             if not fut.done():
                 fut.set_exception(ConnectionError("admin disconnected"))
         self._gs_futures.clear()
+        for _, fut in self._event_waiters:
+            if not fut.done():
+                fut.set_exception(ConnectionError("admin disconnected"))
+        self._event_waiters.clear()
         self.shutdown_event.set()
 
     async def quit(self):
@@ -645,6 +657,37 @@ class OpenTTDAdminClient:
             raise ValueError(f"{context}: {data['error']}")
         return data
 
+    async def get_bridge_version(self, minimum=None, timeout=5.0):
+        """Ask the AdminBridge GameScript which protocol version it speaks, and check it is new enough.
+
+        Every other GameScript method here needs a bridge that understands the command it sends.
+        Against an older bridge those commands are simply ignored and the caller waits out its
+        timeout with no explanation, so call this once after connecting to turn that into an
+        immediate, named failure.
+
+        Returns a dict with "version" (the bridge's protocol version), "commands" (the command
+        names it answers) and "events" (the event kinds it can push) — the two catalogues allow
+        feature-detecting a single command instead of comparing version numbers.
+
+        `minimum` defaults to GS_BRIDGE_VERSION, the version this client is written against;
+        pass 0 to read the version without requiring anything of it.
+
+        Raises ValueError if the bridge is older than `minimum`, and asyncio.TimeoutError if it
+        does not answer at all — note that a bridge predating get_version itself (version 3 and
+        earlier) can only fail that second way, as can a paused game or a server with no bridge
+        loaded. ConnectionError if the admin connection drops while waiting.
+        """
+        from .protocol import GS_BRIDGE_VERSION
+        if minimum is None:
+            minimum = GS_BRIDGE_VERSION
+        data = await self._gs_query({"command": "get_version"}, timeout, "get_version")
+        version = data.get("version", 0)
+        if version < minimum:
+            raise ValueError(
+                f"AdminBridge GameScript is version {version}, but at least {minimum} is "
+                f"required; update the server's copy from gamescript/AdminBridge/")
+        return data
+
     async def get_timetable(self, vehicle_id, timeout=5.0):
         """Fetch an authoritative timetable snapshot for a vehicle via the AdminBridge GameScript.
 
@@ -726,6 +769,30 @@ class OpenTTDAdminClient:
             payload["via_station"] = via_station
         return await self._gs_query(payload, timeout, f"get_station_cargo({station_id}, {cargo_id})")
 
+    async def list_cargo(self, timeout=5.0):
+        """Fetch the running game's cargo table via the AdminBridge GameScript, id to label.
+
+        Every other reply names a cargo by its bare numeric id — get_station()'s and
+        get_station_cargo()'s cargo_id, the cargo_waiting events, the per-cargo load on vehicle
+        events. Those ids index the cargo table the loaded NewGRFs build, so they mean different
+        things in different games and are not worth hardcoding; resolve them against this list
+        instead. Auto-subscribes to Gamescript updates on first use; if you manage update
+        frequencies yourself, ensure update_frequency(Gamescript, Automatic) is active before
+        calling.
+
+        Returns a dict with a "cargo" list holding every cargo type in the game, in no particular
+        order (index it by "cargo_id"; a cargo's position in the list is not its id). Each entry:
+          - "cargo_id": the id used by all the replies above
+          - "label": the cargo label ("PASS", "COAL", ...), or "" if the GameScript could not
+            read it
+          - "freight": 1 for freight cargo, 0 for the rest (passengers, mail, ...)
+
+        Raises asyncio.TimeoutError if no reply arrives (e.g. game paused, GS not loaded) and
+        ConnectionError if the admin connection drops while waiting. The GameScript reports no
+        error for this command, so unlike get_station() it never raises ValueError.
+        """
+        return await self._gs_query({"command": "list_cargo"}, timeout, "list_cargo")
+
     async def get_dispatch(self, vehicle_id, timeout=5.0):
         """Fetch an authoritative snapshot of a vehicle's scheduled dispatch state via the AdminBridge GS.
 
@@ -750,6 +817,131 @@ class OpenTTDAdminClient:
             {"command": "get_dispatch", "vehicle_id": vehicle_id}, timeout,
             f"get_dispatch({vehicle_id})")
 
+    # --- Game events ---
+    #
+    # Everything above is a request the caller makes; this is the other direction. The
+    # AdminBridge GameScript pushes events as they happen, so a bot can react to the game
+    # instead of polling it. Two kinds of thing arrive on the same channel: transitions the
+    # bridge synthesises by sampling game state on an interval (a vehicle reaching or leaving
+    # a stop, a station's waiting cargo changing), and the events the engine itself raises for
+    # a GameScript (crashes, industries opening, companies going bankrupt, ...). See
+    # GameEventType for the full catalogue.
+    #
+    # Consume them either by setting on_event (push) or by awaiting wait_for_event() (pull);
+    # both see every event, so the two can be mixed.
+
+    async def subscribe_events(self, events=None, interval=None, company_id=None, vehicles=None,
+                               stations=None, cargo=None, min_cargo_delta=None,
+                               include_cargo=None, timeout=5.0):
+        """Ask the AdminBridge GameScript to start pushing game events, and wait for it to confirm.
+
+        Every argument narrows what gets sent; the defaults subscribe to every event kind for
+        every vehicle, station and cargo, which is the right starting point on a small map and
+        the wrong one on a large busy map (see the cost note below).
+
+          - events: which GameEventType kinds to receive (default: all of them).
+          - interval: ticks between state samples for the polled kinds (default 10). This is
+            the resolution of those events, not a delay: a stop shorter than `interval` can
+            begin and end between two samples and is then never reported at all.
+          - company_id: only report vehicles and stations owned by this company.
+          - vehicles / stations: only sample these ids, instead of every vehicle / station.
+          - cargo: only inspect these cargo types (for cargo_waiting and for the load reported
+            on vehicle events).
+          - min_cargo_delta: suppress cargo_waiting events whose amount moved by less than this
+            many units since the previous sample (default 1, i.e. report every change).
+          - include_cargo: set False to leave the per-cargo load off vehicle events.
+
+        Subscribing replaces any previous subscription and resets the bridge's baseline, so the
+        first sample after this call only records where everything already is — a vehicle that
+        was sitting at a station when you subscribed did not just arrive, and gets no event.
+
+        Cost: the bridge samples every watched vehicle and every watched station-cargo pair on
+        each interval, inside a GameScript's limited per-tick budget. On a large map prefer a
+        coarser interval and explicit vehicles/stations/cargo lists over the defaults.
+
+        Returns the confirmation dict: {"events": [accepted kinds], "interval": N}. Raises
+        asyncio.TimeoutError if the GameScript does not answer (e.g. game paused, GS not
+        loaded), ValueError on a rejected request (unknown_event, invalid_interval,
+        invalid_min_cargo_delta, invalid_cargo), and ConnectionError if the admin connection
+        drops while waiting.
+        """
+        payload = {"command": "subscribe_events"}
+        if events is not None:
+            payload["events"] = [str(event) for event in events]
+        if interval is not None:
+            payload["interval"] = interval
+        if company_id is not None:
+            payload["company_id"] = company_id
+        if vehicles is not None:
+            payload["vehicles"] = list(vehicles)
+        if stations is not None:
+            payload["stations"] = list(stations)
+        if cargo is not None:
+            payload["cargo"] = list(cargo)
+        if min_cargo_delta is not None:
+            payload["min_cargo_delta"] = min_cargo_delta
+        if include_cargo is not None:
+            payload["include_cargo"] = bool(include_cargo)
+        return await self._gs_query(payload, timeout, "subscribe_events")
+
+    async def unsubscribe_events(self, timeout=5.0):
+        """Stop the event stream and wait for the GameScript to confirm.
+
+        This also drops the bridge's sampling state, so a later subscribe_events() starts from
+        a fresh baseline. Events already delivered stay in this client's buffer; drain or ignore
+        them as you like.
+        """
+        return await self._gs_query({"command": "unsubscribe_events"}, timeout,
+                                    "unsubscribe_events")
+
+    async def wait_for_event(self, kind=None, timeout=5.0):
+        """Await the next game event, optionally of a specific kind (or any of several kinds).
+
+        `kind` is a GameEventType (or plain string), an iterable of them, or None for "any
+        event". Events that arrived earlier and were not taken by another waiter are buffered,
+        so this returns immediately when a matching one is already in hand; the oldest matching
+        event wins. An event is handed to at most one waiter, but the on_event callback (if set)
+        still sees every event regardless.
+
+        Returns the event dict, which always carries "event" (its GameEventType) and "tick"
+        (the game tick it was observed at) plus per-kind fields — see docs/EVENTS.md. Raises
+        asyncio.TimeoutError if nothing matching arrives in time (note that a subscription is
+        needed first — see subscribe_events()) and ConnectionError if the admin connection drops
+        while waiting.
+        """
+        kinds = None
+        if kind is not None:
+            kinds = {str(kind)} if isinstance(kind, str) else {str(k) for k in kind}
+        for buffered in list(self._event_buffer):
+            if self._event_matches(buffered, kinds):
+                self._event_buffer.remove(buffered)
+                return buffered
+        fut = asyncio.get_running_loop().create_future()
+        waiter = (kinds, fut)
+        self._event_waiters.append(waiter)
+        try:
+            return await asyncio.wait_for(fut, timeout)
+        finally:
+            if waiter in self._event_waiters:
+                self._event_waiters.remove(waiter)
+
+    @staticmethod
+    def _event_matches(event, kinds):
+        return kinds is None or (isinstance(event, dict) and event.get("event") in kinds)
+
+    def _dispatch_event(self, event):
+        """Hand one event to the longest-waiting matching waiter, else buffer it; then observe."""
+        for waiter in self._event_waiters:
+            kinds, fut = waiter
+            if not fut.done() and self._event_matches(event, kinds):
+                fut.set_result(event)
+                self._event_waiters.remove(waiter)
+                break
+        else:
+            self._event_buffer.append(event)
+        if self.on_event:
+            self.on_event(event)
+
     async def send_gamescript(self, json_data):
         """Send a JSON string to the GameScript."""
         import json
@@ -852,6 +1044,12 @@ class OpenTTDAdminClient:
                 if not fut.done():
                     fut.set_result(data)
                 return
+            # Unsolicited event batch from the AdminBridge GameScript: fan it out to the
+            # event consumers rather than the generic GameScript callback.
+            if data.get('command') == 'events' and isinstance(data.get('events'), list):
+                for event in data['events']:
+                    self._dispatch_event(event)
+                return
         if self.on_gamescript:
             self.on_gamescript(data)
         else:

lib/openttd/protocol.py

@@ -1,5 +1,5 @@
 import struct
-from enum import IntEnum
+from enum import IntEnum, StrEnum
 
 import monocypher
 from openttd_protocol.wire.exceptions import SocketClosed
@@ -220,6 +220,47 @@ class NetworkAuthenticationMethod(IntEnum):
     X25519_PAKE = 1
     X25519_AuthorizedKey = 2
 
+# Version of the AdminBridge GameScript's JSON protocol this client is written against. The
+# bridge reports its own version via get_version (OpenTTDAdminClient.get_bridge_version()), and
+# an older one will not understand everything sent here. The bridge source lives in
+# gamescript/AdminBridge/; tests/test_gamescript.py keeps this in step with the version declared
+# there, so bumping one without the other fails in CI rather than at runtime.
+GS_BRIDGE_VERSION = 4
+
+class GameEventType(StrEnum):
+    """Event kinds the AdminBridge GameScript can push over the Admin Network.
+
+    These are the values of the "event" field of each event dict, and what
+    OpenTTDAdminClient.subscribe_events() and wait_for_event() take. They are plain strings,
+    so a bare "vehicle_arrive" works everywhere a member does.
+
+    The first three are synthesised by the GameScript sampling game state on an interval,
+    because the engine raises no event for them; the rest are engine events forwarded as they
+    happen. VehicleLost, VehicleWaitingInDepot and VehicleUnprofitable have deliberately no
+    entry here: the engine only ever raises those for AI companies, never for a GameScript.
+    """
+
+    # Polled: derived by diffing successive samples of the game state.
+    VehicleArrive = "vehicle_arrive"
+    VehicleDepart = "vehicle_depart"
+    CargoWaiting = "cargo_waiting"
+    # Forwarded straight from the engine's own GameScript events.
+    VehicleCrashed = "vehicle_crashed"
+    StationFirstVehicle = "station_first_vehicle"
+    IndustryOpen = "industry_open"
+    IndustryClose = "industry_close"
+    TownFounded = "town_founded"
+    CompanyNew = "company_new"
+    CompanyInTrouble = "company_in_trouble"
+    CompanyBankrupt = "company_bankrupt"
+    SubsidyOffer = "subsidy_offer"
+    SubsidyOfferExpired = "subsidy_offer_expired"
+    SubsidyAwarded = "subsidy_awarded"
+    SubsidyExpired = "subsidy_expired"
+    # Emitted by the bridge itself, never subscribed to: one poll produced more events than
+    # fit in the per-poll cap and `count` of them were discarded.
+    EventsDropped = "events_dropped"
+
 class OpenTTDProtocol(TCPProtocol):
     """Low-level OpenTTD TCP protocol handler with encryption support."""
     PacketType = PacketGameType

main_admin.py

@@ -7,7 +7,7 @@ import sys
 sys.path.append(os.path.join(os.path.dirname(__file__), 'lib'))
 
 from openttd import OpenTTDAdminClient
-from openttd.protocol import AdminUpdateFrequency, AdminUpdateType
+from openttd.protocol import AdminUpdateFrequency, AdminUpdateType, GameEventType
 
 # Configuration
 SERVER_HOST = "127.0.0.1"
@@ -67,17 +67,23 @@ async def run_admin():
         if stations and stations[-1]:
             sid = stations[-1][0]["id"]
             try:
+                # Cargo ids come from the loaded NewGRFs, so resolve them to labels to print.
+                labels = {c["cargo_id"]: c["label"]
+                          for c in (await admin.list_cargo(timeout=10.0))["cargo"]}
+
                 data = await admin.get_station(sid, timeout=10.0)
                 print(f"--- Station {sid} ({data.get('name')}) cargo: real-time waiting vs planned ---")
                 for cargo in data.get("cargo", []):
-                    print(f"    cargo {cargo['cargo_id']}: waiting={cargo['waiting']} "
+                    print(f"    {labels.get(cargo['cargo_id'], cargo['cargo_id'])}: "
+                          f"waiting={cargo['waiting']} "
                           f"planned={cargo['planned']} rating={cargo['rating']}")
 
                 # Break the first cargo down by source station and by next hop (routing destination).
                 if data.get("cargo"):
                     cid = data["cargo"][0]["cargo_id"]
                     flow = await admin.get_station_cargo(sid, cid, timeout=10.0)
-                    print(f"--- Station {sid} cargo {cid} flow breakdown (station 65535 = none/deleted) ---")
+                    print(f"--- Station {sid} cargo {labels.get(cid, cid)} flow breakdown "
+                          f"(station 65535 = none/deleted) ---")
                     print(f"    waiting by source: {flow['waiting_by_from']}")
                     print(f"    waiting by next hop: {flow['waiting_by_via']}")
                     print(f"    planned by source: {flow['planned_by_from']}")
@@ -85,7 +91,26 @@ async def run_admin():
             except Exception as e:  # noqa: BLE001 - demo script: one failed station query should not abort the walk
                 print(f"!!! station query failed: {e}")
 
-        await asyncio.sleep(5)
+        # Watch the game live: vehicles reaching/leaving stops and cargo piling up at stations.
+        print("--- Subscribing to game events ---")
+        try:
+            accepted = await admin.subscribe_events(
+                events=[GameEventType.VehicleArrive, GameEventType.VehicleDepart,
+                        GameEventType.CargoWaiting],
+                interval=5, timeout=10.0)
+            print(f"    subscribed to {accepted['events']} every {accepted['interval']} ticks")
+
+            for _ in range(5):
+                try:
+                    event = await admin.wait_for_event(timeout=15.0)
+                except asyncio.TimeoutError:
+                    print("    (nothing happened -- is the server paused or idle?)")
+                    break
+                print(f">>> [EVENT] {event}")
+            await admin.unsubscribe_events()
+        except Exception as e:  # noqa: BLE001 - demo script: report and carry on to a clean quit
+            print(f"!!! event subscription failed: {e}")
+
         print("--- Quitting ---")
         await admin.quit()
 

tests/test_admin.py

@@ -5,7 +5,12 @@ import os
 import monocypher
 import pytest
 from openttd import OpenTTDAdminClient
-from openttd.protocol import AdminUpdateFrequency, AdminUpdateType, PacketAdminType
+from openttd.protocol import (
+    GS_BRIDGE_VERSION,
+    AdminUpdateFrequency,
+    AdminUpdateType,
+    PacketAdminType,
+)
 
 
 class FakeNetworkError(OSError):
@@ -388,6 +393,40 @@ async def test_admin_get_station_cargo_error_response():
         await task
     assert client._gs_futures == {}
 
+@pytest.mark.asyncio
+async def test_admin_list_cargo_request_and_response():
+    client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin")
+    proto = MockProtocol()
+    client._protocol = proto
+    client._transport = MockTransport()
+
+    task = asyncio.ensure_future(client.list_cargo())
+    await asyncio.sleep(0)  # let the task send the request
+
+    # First use auto-subscribes to Gamescript updates, then sends the query.
+    assert len(proto.sent) == 2
+    assert proto.sent[0][2] == PacketAdminType.AdminUpdateFrequency
+    assert decode_gamescript_payload(proto.sent[1]) == {
+        "command": "list_cargo", "request_id": 1,
+    }
+
+    response = {"command": "list_cargo", "request_id": 1,
+                "cargo": [{"cargo_id": 0, "label": "PASS", "freight": 0},
+                          {"cargo_id": 1, "label": "COAL", "freight": 1}]}
+    await client.receive_ServerGamescript(None, data=response)
+    assert await task == response
+    assert client._gs_futures == {}
+
+@pytest.mark.asyncio
+async def test_admin_list_cargo_timeout():
+    client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin")
+    client._protocol = MockProtocol()
+    client._transport = MockTransport()
+
+    with pytest.raises(asyncio.TimeoutError):
+        await client.list_cargo(timeout=0.05)
+    assert client._gs_futures == {}
+
 @pytest.mark.asyncio
 async def test_admin_get_dispatch_request_and_response():
     client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin")
@@ -429,6 +468,58 @@ async def test_admin_get_dispatch_error_response():
         await task
     assert client._gs_futures == {}
 
+@pytest.mark.asyncio
+async def test_admin_get_bridge_version_request_and_response():
+    client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin")
+    proto = MockProtocol()
+    client._protocol = proto
+    client._transport = MockTransport()
+
+    task = asyncio.ensure_future(client.get_bridge_version())
+    await asyncio.sleep(0)  # let the task send the request
+
+    # First use auto-subscribes to Gamescript updates, then sends the query.
+    assert len(proto.sent) == 2
+    assert proto.sent[0][2] == PacketAdminType.AdminUpdateFrequency
+    assert decode_gamescript_payload(proto.sent[1]) == {
+        "command": "get_version", "request_id": 1,
+    }
+
+    response = {"command": "get_version", "request_id": 1, "version": GS_BRIDGE_VERSION,
+                "commands": ["get_timetable", "get_version"],
+                "events": ["vehicle_arrive", "vehicle_depart"]}
+    await client.receive_ServerGamescript(None, data=response)
+    assert await task == response
+    assert client._gs_futures == {}
+
+@pytest.mark.asyncio
+async def test_admin_get_bridge_version_rejects_older_bridge():
+    """A bridge that answers but is too old fails by name, not by timeout."""
+    client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin")
+    client._protocol = MockProtocol()
+    client._transport = MockTransport()
+
+    task = asyncio.ensure_future(client.get_bridge_version(minimum=GS_BRIDGE_VERSION + 1))
+    await asyncio.sleep(0)
+    await client.receive_ServerGamescript(
+        None, data={"command": "get_version", "request_id": 1, "version": GS_BRIDGE_VERSION})
+    with pytest.raises(ValueError, match=f"version {GS_BRIDGE_VERSION}"):
+        await task
+    assert client._gs_futures == {}
+
+@pytest.mark.asyncio
+async def test_admin_get_bridge_version_minimum_zero_accepts_anything():
+    """minimum=0 reads the version without requiring anything, even if the reply omits it."""
+    client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin")
+    client._protocol = MockProtocol()
+    client._transport = MockTransport()
+
+    task = asyncio.ensure_future(client.get_bridge_version(minimum=0))
+    await asyncio.sleep(0)
+    await client.receive_ServerGamescript(
+        None, data={"command": "get_version", "request_id": 1, "commands": []})
+    assert await task == {"command": "get_version", "request_id": 1, "commands": []}
+
 @pytest.mark.asyncio
 async def test_admin_gamescript_passthrough_unmatched():
     client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin")

tests/test_e2e.py

@@ -11,8 +11,10 @@ sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'lib'))
 
 from openttd import OpenTTDAdminClient, OpenTTDClient
 from openttd.protocol import (
+    GS_BRIDGE_VERSION,
     AdminUpdateFrequency,
     AdminUpdateType,
+    GameEventType,
     ModifyTimetableFlags,
     OpenTTDAdminProtocol,
     OpenTTDProtocol,
@@ -27,6 +29,12 @@ TIMETABLE_VEHICLE_ID = 7
 TIMETABLE_ORDER_POSITION = 0
 # A station TIMETABLE_VEHICLE_ID can legally serve, used for add_order/remove_order tests.
 ORDER_STATION_ID = 6
+# A second vehicle of the same company, dedicated to the scheduled-dispatch test, which owns and
+# overwrites this vehicle's dispatch state. It must have its own order list -- dispatch schedules
+# live on the order list, so pointing this at a vehicle that *shares* orders with another would
+# silently rewrite that other vehicle's schedules too. (Cloning a vehicle without sharing orders
+# gives an independent list, but copies the source's schedules along with it.)
+DISPATCH_VEHICLE_ID = 14
 
 
 # --- Pytest Fixtures ---
@@ -303,17 +311,17 @@ async def test_e2e_client_scheduled_dispatch_edit_and_view(connected_owner_clien
     #   add_dispatch_slot(), remove_dispatch_slot(), clear_dispatch_schedule(), set_dispatch_duration(),
     #   set_dispatch_start_date(), and get_dispatch(). Edits go over the game port and are read back
     #   authoritatively via the admin get_dispatch(). The test leaves the vehicle with no schedules.
-    veh = TIMETABLE_VEHICLE_ID
+    veh = DISPATCH_VEHICLE_ID
     owner = connected_owner_client
     await connected_admin.update_frequency(AdminUpdateType.Gamescript, AdminUpdateFrequency.Automatic)
 
     async def dispatch():
         return await connected_admin.get_dispatch(veh, timeout=10.0)
 
+    # The assertions below address schedules by absolute index, so DISPATCH_VEHICLE_ID must start
+    # with none of its own; the test restores that state on the way out.
     start = await dispatch()  # get_dispatch input 1: a valid vehicle
     assert "schedules" in start and isinstance(start["schedules"], list)
-    if start["schedules"]:
-        pytest.skip("Test vehicle already has dispatch schedules; expected a clean vehicle.")
 
     # add_dispatch_schedule: two schedules (indices 0 and 1) with different start ticks/durations.
     await owner.add_dispatch_schedule(veh, 0, 3000)
@@ -511,6 +519,38 @@ async def test_e2e_admin_send_gamescript_multiple_inputs(connected_admin):
     await connected_admin.send_gamescript({"command": "ping", "sequence": 1})
     await asyncio.sleep(0.5)
 
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_e2e_admin_get_bridge_version_supported(connected_admin):
+    # Public function: get_bridge_version()
+    # Input 1: the default minimum, i.e. the version this client is written against.
+    #
+    # Every GameScript test below this one fails as an unexplained timeout if the server runs a
+    # stale AdminBridge (or none at all), because a bridge that does not know a command simply
+    # drops it. This test is the one that says so out loud, so run it first when the GS tests
+    # start hanging.
+    try:
+        data = await connected_admin.get_bridge_version(timeout=10.0)
+    except asyncio.TimeoutError:
+        pytest.fail(
+            "The server's AdminBridge GameScript did not answer get_version. It is older than "
+            f"version {GS_BRIDGE_VERSION} (which introduced the command), not loaded at all, or "
+            "the game is paused. See gamescript/AdminBridge/README.md.")
+
+    assert data["version"] >= GS_BRIDGE_VERSION
+    # The catalogues let a client feature-detect one command rather than compare versions.
+    assert "get_version" in data["commands"]
+    assert set(data["events"]) == {e.value for e in GameEventType} - {GameEventType.EventsDropped}
+
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_e2e_admin_get_bridge_version_minimum_zero(connected_admin):
+    # Public function: get_bridge_version()
+    # Input 2: minimum=0 -> read the version without requiring anything of it.
+    data = await connected_admin.get_bridge_version(minimum=0, timeout=10.0)
+    assert isinstance(data["version"], int)
+    assert "get_timetable" in data["commands"]
+
 @pytest.mark.e2e
 @pytest.mark.asyncio
 async def test_e2e_admin_list_vehicles_all_companies(connected_admin):
@@ -715,6 +755,130 @@ async def test_e2e_admin_get_station_cargo_invalid_cargo(connected_admin):
     with pytest.raises(ValueError, match="invalid_cargo"):
         await connected_admin.get_station_cargo(stations[0]["id"], 250, timeout=10.0)
 
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_e2e_admin_list_cargo_table(connected_admin):
+    # Public function: list_cargo()
+    # Input 1: an explicit timeout. Every game has a cargo table, so an empty list would be a bug.
+    data = await connected_admin.list_cargo(timeout=10.0)
+    assert isinstance(data["cargo"], list) and data["cargo"]
+    for cargo in data["cargo"]:
+        for key in ("cargo_id", "label", "freight"):
+            assert key in cargo
+        assert cargo["freight"] in (0, 1)
+    ids = [cargo["cargo_id"] for cargo in data["cargo"]]
+    assert len(ids) == len(set(ids))
+    # Any cargo set carries passengers as well as freight, so both kinds must show up.
+    assert any(cargo["freight"] == 0 for cargo in data["cargo"])
+    assert any(cargo["freight"] == 1 for cargo in data["cargo"])
+
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_e2e_admin_list_cargo_resolves_station_cargo_ids(connected_admin):
+    # Public function: list_cargo()
+    # Input 2: the default timeout. The point of the call: naming the bare ids get_station() returns.
+    responses = []
+    connected_admin.on_gamescript = lambda data: responses.append(data)
+    await connected_admin.update_frequency(AdminUpdateType.Gamescript, AdminUpdateFrequency.Automatic)
+    await connected_admin.list_stations()
+    await asyncio.sleep(0.5)
+
+    assert responses and "stations" in responses[-1]
+    stations = responses[-1]["stations"]
+    if not stations:
+        pytest.skip("No stations on the test server to query.")
+
+    labels = {cargo["cargo_id"]: cargo["label"] for cargo in (await connected_admin.list_cargo())["cargo"]}
+    detail = await connected_admin.get_station(stations[0]["id"], timeout=10.0)
+    for cargo in detail["cargo"]:
+        assert cargo["cargo_id"] in labels
+
+
+# --- Game Events ---
+
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_e2e_admin_subscribe_events_all_kinds(connected_admin):
+    # Public functions: subscribe_events(), unsubscribe_events()
+    # Input 1: no arguments -> every event kind, default interval
+    data = await connected_admin.subscribe_events(timeout=10.0)
+    assert isinstance(data["events"], list)
+    assert "vehicle_arrive" in data["events"] and "cargo_waiting" in data["events"]
+    assert data["interval"] == 10
+
+    stopped = await connected_admin.unsubscribe_events()
+    assert stopped["events"] == []
+
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_e2e_admin_subscribe_events_filtered(connected_admin):
+    # Public functions: subscribe_events(), unsubscribe_events()
+    # Input 2: a narrowed subscription -> only the requested kinds come back
+    data = await connected_admin.subscribe_events(
+        events=[GameEventType.VehicleArrive, GameEventType.CargoWaiting],
+        interval=5, company_id=0, min_cargo_delta=2, include_cargo=False, timeout=10.0)
+    assert sorted(data["events"]) == ["cargo_waiting", "vehicle_arrive"]
+    assert data["interval"] == 5
+
+    await connected_admin.unsubscribe_events(timeout=10.0)
+
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_e2e_admin_subscribe_events_unknown_kind(connected_admin):
+    # Public function: subscribe_events()
+    # A kind the GameScript does not know -> it reports unknown_event
+    with pytest.raises(ValueError, match="unknown_event"):
+        await connected_admin.subscribe_events(events=["definitely_not_an_event"], timeout=10.0)
+
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_e2e_admin_wait_for_event_vehicle_reaches_a_stop(connected_admin):
+    # Public function: wait_for_event()
+    # Input 1: filtered by kind. Needs traffic on the server, so a quiet map skips.
+    await connected_admin.subscribe_events(
+        events=[GameEventType.VehicleArrive, GameEventType.VehicleDepart],
+        interval=2, timeout=10.0)
+    try:
+        event = await connected_admin.wait_for_event(
+            {GameEventType.VehicleArrive, GameEventType.VehicleDepart}, timeout=60.0)
+    except asyncio.TimeoutError:
+        pytest.skip("No vehicle reached or left a stop on the test server within the timeout.")
+    finally:
+        await connected_admin.unsubscribe_events()
+
+    assert event["event"] in ("vehicle_arrive", "vehicle_depart")
+    for key in ("tick", "vehicle_id", "station_id", "owner", "vehicle_type", "order_position"):
+        assert key in event
+    if event["event"] == "vehicle_depart":
+        assert event["dwell"] >= 0
+
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_e2e_admin_wait_for_event_any_kind(connected_admin):
+    # Public function: wait_for_event()
+    # Input 2: no kind filter -> whatever the game produces first
+    await connected_admin.subscribe_events(interval=2, timeout=10.0)
+    try:
+        event = await connected_admin.wait_for_event(timeout=60.0)
+    except asyncio.TimeoutError:
+        pytest.skip("Nothing happened on the test server within the timeout.")
+    finally:
+        await connected_admin.unsubscribe_events()
+
+    assert "event" in event and "tick" in event
+
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_e2e_admin_unsubscribe_events_stops_the_stream(connected_admin):
+    # Public function: unsubscribe_events()
+    # After unsubscribing the bridge must go quiet, so a fresh wait times out.
+    await connected_admin.subscribe_events(interval=2, timeout=10.0)
+    await connected_admin.unsubscribe_events(timeout=10.0)
+
+    connected_admin._event_buffer.clear()  # drop anything delivered before we unsubscribed
+    with pytest.raises(asyncio.TimeoutError):
+        await connected_admin.wait_for_event(timeout=5.0)
+
 
 # --- Protocol Public Functions ---
 

tests/test_events.py

@@ -0,0 +1,251 @@
+import asyncio
+import json
+
+import pytest
+from openttd import OpenTTDAdminClient
+from openttd.protocol import GameEventType, PacketAdminType
+
+
+class MockTransport:
+    def is_closing(self):
+        return False
+    def close(self):
+        pass
+    def write(self, data):
+        return len(data)
+
+class MockProtocol:
+    def __init__(self):
+        self.sent = []
+    async def send_packet(self, data):
+        self.sent.append(data)
+        return len(data)
+
+def decode_gamescript_payload(packet):
+    """Decode the JSON payload of an AdminGamescript packet (2-byte length + 1-byte type + string)."""
+    return json.loads(packet[3:].split(b"\x00")[0])
+
+def make_admin(subscribed=True, **kwargs):
+    """An admin client wired to mock transports, with the Gamescript auto-subscribe already done
+    unless a test wants to observe it."""
+    client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin", **kwargs)
+    client._protocol = MockProtocol()
+    client._transport = MockTransport()
+    client._gs_subscribed = subscribed
+    return client
+
+def events_packet(*events):
+    """The envelope the AdminBridge GameScript pushes event batches in."""
+    return {"command": "events", "events": list(events)}
+
+def arrival(vehicle_id=7, station_id=3, tick=100):
+    return {"event": "vehicle_arrive", "tick": tick, "vehicle_id": vehicle_id,
+            "station_id": station_id, "owner": 0, "vehicle_type": 0, "order_position": 0,
+            "cargo": [{"cargo_id": 0, "load": 12}]}
+
+
+# --- Event kinds ---
+
+def test_event_type_values_are_the_wire_strings():
+    assert GameEventType.VehicleArrive == "vehicle_arrive"
+    assert GameEventType.CargoWaiting == "cargo_waiting"
+    # Members are plain strings, so they can be used interchangeably with literals.
+    assert str(GameEventType.VehicleDepart) == "vehicle_depart"
+
+
+# --- Subscribing ---
+
+@pytest.mark.asyncio
+async def test_subscribe_events_defaults_send_only_the_command():
+    client = make_admin(subscribed=False)
+
+    task = asyncio.ensure_future(client.subscribe_events())
+    await asyncio.sleep(0)
+
+    # First use auto-subscribes to Gamescript updates, then sends the request.
+    assert len(client._protocol.sent) == 2
+    assert client._protocol.sent[0][2] == PacketAdminType.AdminUpdateFrequency
+    assert decode_gamescript_payload(client._protocol.sent[1]) == {
+        "command": "subscribe_events", "request_id": 1,
+    }
+
+    reply = {"command": "subscribe_events", "request_id": 1,
+             "events": ["vehicle_arrive", "vehicle_depart"], "interval": 10}
+    await client.receive_ServerGamescript(None, data=reply)
+    assert await task == reply
+    assert client._gs_futures == {}
+
+@pytest.mark.asyncio
+async def test_subscribe_events_encodes_every_filter():
+    client = make_admin()
+
+    task = asyncio.ensure_future(client.subscribe_events(
+        events=[GameEventType.VehicleArrive, "cargo_waiting"], interval=25, company_id=0,
+        vehicles=(7, 9), stations=(3,), cargo=[0, 1], min_cargo_delta=5, include_cargo=False))
+    await asyncio.sleep(0)
+
+    assert decode_gamescript_payload(client._protocol.sent[0]) == {
+        "command": "subscribe_events", "request_id": 1,
+        "events": ["vehicle_arrive", "cargo_waiting"], "interval": 25, "company_id": 0,
+        "vehicles": [7, 9], "stations": [3], "cargo": [0, 1],
+        "min_cargo_delta": 5, "include_cargo": False,
+    }
+    await client.receive_ServerGamescript(
+        None, data={"request_id": 1, "events": ["vehicle_arrive", "cargo_waiting"]})
+    assert (await task)["events"] == ["vehicle_arrive", "cargo_waiting"]
+
+@pytest.mark.asyncio
+async def test_subscribe_events_rejected_request_raises():
+    client = make_admin()
+
+    task = asyncio.ensure_future(client.subscribe_events(events=["not_an_event"]))
+    await asyncio.sleep(0)
+    await client.receive_ServerGamescript(
+        None, data={"command": "subscribe_events", "request_id": 1,
+                    "error": "unknown_event", "event": "not_an_event"})
+    with pytest.raises(ValueError, match="unknown_event"):
+        await task
+    assert client._gs_futures == {}
+
+@pytest.mark.asyncio
+async def test_subscribe_events_timeout():
+    client = make_admin()
+    with pytest.raises(asyncio.TimeoutError):
+        await client.subscribe_events(timeout=0.05)
+    assert client._gs_futures == {}
+
+@pytest.mark.asyncio
+async def test_unsubscribe_events():
+    client = make_admin()
+
+    task = asyncio.ensure_future(client.unsubscribe_events())
+    await asyncio.sleep(0)
+    assert decode_gamescript_payload(client._protocol.sent[0]) == {
+        "command": "unsubscribe_events", "request_id": 1,
+    }
+
+    reply = {"command": "unsubscribe_events", "request_id": 1, "events": []}
+    await client.receive_ServerGamescript(None, data=reply)
+    assert await task == reply
+
+
+# --- Receiving events ---
+
+@pytest.mark.asyncio
+async def test_event_batch_reaches_on_event_and_bypasses_on_gamescript():
+    client = make_admin()
+    seen, other = [], []
+    client.on_event = seen.append
+    client.on_gamescript = other.append
+
+    depart = {"event": "vehicle_depart", "tick": 120, "vehicle_id": 7,
+              "station_id": 3, "dwell": 74}
+    await client.receive_ServerGamescript(None, data=events_packet(arrival(), depart))
+
+    assert [event["event"] for event in seen] == ["vehicle_arrive", "vehicle_depart"]
+    assert other == []
+
+@pytest.mark.asyncio
+async def test_non_event_gamescript_payloads_still_reach_on_gamescript():
+    client = make_admin()
+    seen, other = [], []
+    client.on_event = seen.append
+    client.on_gamescript = other.append
+
+    # A reply that merely mentions events, and a malformed batch, are not event batches.
+    await client.receive_ServerGamescript(None, data={"vehicles": []})
+    await client.receive_ServerGamescript(None, data={"command": "events", "events": "nope"})
+
+    assert seen == []
+    assert other == [{"vehicles": []}, {"command": "events", "events": "nope"}]
+
+@pytest.mark.asyncio
+async def test_events_are_buffered_until_awaited():
+    client = make_admin()
+    await client.receive_ServerGamescript(None, data=events_packet(arrival(station_id=3),
+                                                                   arrival(station_id=4)))
+
+    # The oldest matching event is handed out first, and only once.
+    first = await client.wait_for_event("vehicle_arrive")
+    second = await client.wait_for_event(GameEventType.VehicleArrive)
+    assert (first["station_id"], second["station_id"]) == (3, 4)
+    assert len(client._event_buffer) == 0
+
+@pytest.mark.asyncio
+async def test_wait_for_event_resolves_on_arrival():
+    client = make_admin()
+    task = asyncio.ensure_future(client.wait_for_event("cargo_waiting", timeout=5.0))
+    await asyncio.sleep(0)
+
+    cargo = {"event": "cargo_waiting", "tick": 300, "station_id": 3,
+             "cargo_id": 0, "waiting": 25, "delta": 15}
+    await client.receive_ServerGamescript(None, data=events_packet(arrival(), cargo))
+
+    assert await task == cargo
+    # The event that did not match the waiter is still buffered for the next caller.
+    assert await client.wait_for_event() == arrival()
+    assert client._event_waiters == []
+
+@pytest.mark.asyncio
+async def test_wait_for_event_accepts_several_kinds():
+    client = make_admin()
+    task = asyncio.ensure_future(
+        client.wait_for_event({GameEventType.VehicleArrive, GameEventType.VehicleDepart}))
+    await asyncio.sleep(0)
+
+    await client.receive_ServerGamescript(
+        None, data=events_packet({"event": "town_founded", "tick": 5, "town_id": 2}, arrival()))
+    assert (await task)["event"] == "vehicle_arrive"
+
+@pytest.mark.asyncio
+async def test_wait_for_event_ignores_non_dict_events_when_filtering():
+    client = make_admin()
+    task = asyncio.ensure_future(client.wait_for_event("vehicle_arrive", timeout=0.05))
+    await asyncio.sleep(0)
+
+    # A malformed entry inside a batch must not satisfy a filtered waiter.
+    await client.receive_ServerGamescript(None, data=events_packet("garbage"))
+    with pytest.raises(asyncio.TimeoutError):
+        await task
+    assert list(client._event_buffer) == ["garbage"]
+
+@pytest.mark.asyncio
+async def test_wait_for_event_timeout_deregisters_the_waiter():
+    client = make_admin()
+    with pytest.raises(asyncio.TimeoutError):
+        await client.wait_for_event("vehicle_arrive", timeout=0.05)
+    assert client._event_waiters == []
+
+@pytest.mark.asyncio
+async def test_already_resolved_waiters_are_skipped():
+    client = make_admin()
+    # A waiter whose future completed but that has not been cleaned up yet must not swallow
+    # the event; it goes to the next live waiter instead.
+    stale = asyncio.get_running_loop().create_future()
+    stale.set_result("already done")
+    live = asyncio.get_running_loop().create_future()
+    client._event_waiters = [(None, stale), (None, live)]
+
+    await client.receive_ServerGamescript(None, data=events_packet(arrival()))
+    assert live.result() == arrival()
+    assert len(client._event_buffer) == 0
+
+@pytest.mark.asyncio
+async def test_event_buffer_is_bounded():
+    client = make_admin(event_buffer_size=2)
+    await client.receive_ServerGamescript(None, data=events_packet(
+        arrival(station_id=1), arrival(station_id=2), arrival(station_id=3)))
+
+    # The oldest event is dropped rather than growing the buffer without bound.
+    assert [event["station_id"] for event in client._event_buffer] == [2, 3]
+
+@pytest.mark.asyncio
+async def test_disconnect_fails_pending_event_waiters():
+    client = make_admin()
+    task = asyncio.ensure_future(client.wait_for_event())
+    await asyncio.sleep(0)
+
+    client.disconnect(None)
+    with pytest.raises(ConnectionError):
+        await task
+    assert client._event_waiters == []

tests/test_gamescript.py

@@ -0,0 +1,89 @@
+"""Checks the checked-in AdminBridge GameScript against the client that talks to it.
+
+The bridge is Squirrel, and CI has no Squirrel toolchain, so these are not a substitute for
+running it (see gamescript/AdminBridge/README.md). They cover the one class of breakage that
+is invisible until a server is in front of you: the two halves of the protocol drifting apart
+-- a version bumped on one side only, an event kind or a command the client uses that the
+bridge does not implement, or the docker setup no longer serving the tracked copy.
+"""
+
+import re
+from pathlib import Path
+
+import pytest
+from openttd.protocol import GS_BRIDGE_VERSION, GameEventType
+
+ROOT = Path(__file__).resolve().parents[1]
+GS_DIR = ROOT / "gamescript" / "AdminBridge"
+INFO_NUT = (GS_DIR / "info.nut").read_text()
+MAIN_NUT = (GS_DIR / "main.nut").read_text()
+
+
+def _block(source, opening, closing):
+    """Return the text between `opening` and the next `closing`, e.g. an array or table body."""
+    start = source.index(opening) + len(opening)
+    return source[start:source.index(closing, start)]
+
+
+@pytest.mark.unit
+def test_gamescript_version_matches_client():
+    """info.nut, main.nut and the client must all name the same protocol version.
+
+    A GameScript cannot read its own info.nut at runtime, so main.nut duplicates the version;
+    this is what keeps the copy honest, and what makes a bump that misses a file fail here.
+    """
+    info_version = int(re.search(r"function GetVersion\(\)\s*{\s*return (\d+);", INFO_NUT).group(1))
+    main_version = int(re.search(r"BRIDGE_VERSION = (\d+);", MAIN_NUT).group(1))
+    assert info_version == main_version == GS_BRIDGE_VERSION
+
+
+@pytest.mark.unit
+def test_gamescript_can_load_older_savegames():
+    """The bridge must stay loadable by savegames that pinned an older version of it.
+
+    OpenTTD defaults MinVersionToLoad() to GetVersion(), so without an explicit override every
+    version bump orphans existing savegames: the engine finds no compatible script and falls
+    back with a warning. The bridge keeps no savegame state, so any version can take over.
+    """
+    min_version = int(re.search(r"function MinVersionToLoad\(\)\s*{\s*return (\d+);", INFO_NUT).group(1))
+    assert min_version <= GS_BRIDGE_VERSION
+
+
+@pytest.mark.unit
+def test_gamescript_event_catalogue_matches_client_enum():
+    """Every kind the bridge can push has a GameEventType, and vice versa.
+
+    GameEventType is what subscribe_events() validates against, so a kind in one list and not
+    the other is either an event nobody can subscribe to or a subscription the bridge rejects.
+    """
+    catalogue = set(re.findall(r'"(\w+)"', _block(MAIN_NUT, "this.event_order = [", "];")))
+    # EventsDropped is emitted by the bridge itself and never subscribed to, so it is
+    # deliberately absent from the subscribable catalogue.
+    assert catalogue == {e.value for e in GameEventType} - {GameEventType.EventsDropped}
+
+
+@pytest.mark.unit
+def test_gamescript_implements_every_command_the_client_sends():
+    """Each command in a client payload must have a handler in the bridge's dispatch table.
+
+    A command the bridge does not know is silently dropped, so the caller only sees a timeout.
+    Checked one way round only: the bridge is allowed to implement more than the client wraps,
+    which is how list_cargo sat there answering nobody until a method was written for it.
+    """
+    commands = set(re.findall(r"^\s*(\w+)\s*=\s*{ handler",
+                              _block(MAIN_NUT, "COMMANDS = {", "\n\t};"), re.MULTILINE))
+    client_source = (ROOT / "lib" / "openttd" / "client.py").read_text()
+    sent = set(re.findall(r'{"command": "(\w+)"', client_source))
+    assert sent, "no GameScript commands found in the client -- has the payload spelling changed?"
+    assert sent <= commands
+
+
+@pytest.mark.unit
+def test_docker_serves_the_tracked_gamescript():
+    """The container must run the copy in git, not an untracked one under docker/config/.
+
+    That mount is the whole reason the tracked copy stays honest: without it the server reads a
+    file nobody reviews, which is how the bridge went unversioned in the first place.
+    """
+    compose = (ROOT / "docker" / "docker-compose.yml").read_text()
+    assert "../gamescript/AdminBridge:" in compose