README.md

@@ -17,6 +17,7 @@ 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.
+- **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.
 
 ## 🛠 Setup
 
@@ -81,4 +82,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)

docs/ARCHITECTURE.md

@@ -16,6 +16,11 @@ 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.
+- **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,232 @@
+# 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). 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

@@ -69,6 +69,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,6 +13,7 @@ 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_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. |
 

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,17 +471,17 @@ 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
         self.log = logging.getLogger(f"OTTDA-{admin_name}")
-        
+
         # State
         self.encryption_enabled = False
         self.joined = asyncio.Event()
         self.shutdown_event = asyncio.Event()
-        
+
         # Internal crypto
         self._admin_password = ""
         self._session_key_send = None
@@ -488,17 +489,24 @@ class OpenTTDAdminClient:
         self._encryption_nonce = None
         self._send_aead = None
         self._recv_aead = None
-        
+
         # Callbacks
         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):
@@ -750,6 +762,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 +989,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,40 @@ class NetworkAuthenticationMethod(IntEnum):
     X25519_PAKE = 1
     X25519_AuthorizedKey = 2
 
+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"
@@ -85,7 +85,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_e2e.py

@@ -13,6 +13,7 @@ from openttd import OpenTTDAdminClient, OpenTTDClient
 from openttd.protocol import (
     AdminUpdateFrequency,
     AdminUpdateType,
+    GameEventType,
     ModifyTimetableFlags,
     OpenTTDAdminProtocol,
     OpenTTDProtocol,
@@ -27,6 +28,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 +310,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)
@@ -716,6 +723,92 @@ async def test_e2e_admin_get_station_cargo_invalid_cargo(connected_admin):
         await connected_admin.get_station_cargo(stations[0]["id"], 250, timeout=10.0)
 
 
+# --- 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 ---
 
 @pytest.mark.e2e

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 == []