Add game event support to the admin client

Everything on the admin GameScript channel so far has been request/reply.
This adds the other direction: subscribe_events() opens a push stream so a
bot can react to the game instead of polling it, consumed either by awaiting
wait_for_event() or via an on_event callback. Both see every event; an event
goes to at most one waiter, and unclaimed ones sit in a bounded buffer.

Sixteen kinds, from two sources. The engine raises no GameScript event for a
vehicle reaching a stop or cargo arriving, so vehicle_arrive, vehicle_depart
and cargo_waiting are synthesised by the bridge sampling state every
`interval` ticks and diffing against the previous sample -- which means a
stop shorter than the interval is never reported, and the first sample only
establishes a baseline. The rest (crashes, industries, towns, companies,
subsidies) are engine events forwarded verbatim. vehicle_lost,
vehicle_waiting_in_depot and vehicle_unprofitable are deliberately absent:
the engine raises those only for AI companies, so a GameScript can never
observe them.

The server-side half lives in the AdminBridge GameScript, which is not in
this repo -- docker/config is gitignored -- so it has to be updated
separately for any of this to work.

Also repoints the scheduled-dispatch E2E test at a dedicated vehicle
(DISPATCH_VEHICLE_ID). It had been silently skipping because vehicle 7
carries a hand-built annual dispatch schedule, which left eight dispatch
methods unverified end to end while check_public_calls.py reported them
green off static analysis of the call sites.

Co-Authored-By: Claude <[email protected]>

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