diff --git a/CHANGELOG.md b/CHANGELOG.md
index a9548ee..f5f34e0 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,8 +4,72 @@ All notable changes to this project will be documented in this file.
 
 ## [Unreleased]
 
+### Added
+
+- **In-process RCON client** (`api/rcon.py`)
+
+  - Replaces the per-command shell-out to `scripts/rcon-client.sh`, which opened a
+    new TCP connection and re-authenticated for every command. One authenticated
+    connection is now held open and shared, so command batches are practical.
+  - Fixes two defects in the previous Python fallback: a single unframed `recv`
+    truncated or split responses over 4096 bytes, and any reply of four or more
+    bytes was treated as a successful login instead of checking for the `-1`
+    request id that signals auth failure.
+  - Reassembles the multi-packet responses Minecraft sends for output over 4096
+    bytes, reconnects automatically after a server restart, and rejects commands
+    long enough that the server would silently truncate them.
+  - `api/server.py` calls it through a new `run_rcon_command()` helper that falls
+    back to `scripts/rcon-client.sh` when RCON is unconfigured or unreachable, so
+    the `rcon-cli`-inside-the-container path still works.
+
+- **Game event bus** (`api/events.py`) — see [docs/EVENT_BUS.md](docs/EVENT_BUS.md)
+
+  - Parses the server log into typed events: chat, connect, join, leave, death,
+    advancement, command, server ready and server stopping.
+  - Events are appended to `data/events/YYYY-MM-DD.jsonl`. Writes are batched and
+    files older than 30 days are pruned, to limit SD-card wear on the Pi.
+  - Features subscribe with a handler function instead of parsing raw log text.
+    A handler that raises is logged and skipped rather than stopping the bus.
+  - New endpoints `GET /api/events` and `GET /api/events/types`, both requiring
+    `logs.view`, plus a `game_event` WebSocket message alongside the existing raw
+    `logs` stream. `getEvents()` and `getEventTypes()` added to
+    `web/src/services/api.js`.
+
+- **Feature roadmap** ([docs/FAMILY_SERVER_ROADMAP.md](docs/FAMILY_SERVER_ROADMAP.md))
+  covering gameplay and integration features, as distinct from the management
+  product planned in `docs/ROADMAP.md`.
+
+### Fixed
+
+- **The WebSocket `execute_command` handler reached RCON without sanitising its
+  input**, so it bypassed the command allowlist that `POST /api/server/command`
+  enforces. It now runs the same validation and writes the same audit entries.
+- **A non-string `command` value crashed inside the sanitiser** on both the REST
+  and WebSocket paths. The REST endpoint returned a 500 instead of a 400, and the
+  WebSocket raised before its error handler, leaving the client with no response
+  at all. Both now reject the value explicitly.
+
 ### Changed
 
+- **The log follower now runs from API startup rather than from the first browser
+  connection**, and no longer stops when the last client disconnects. Events that
+  occur while the dashboard is closed were previously lost entirely. The follower
+  can be brought down deliberately with `stop_log_reader()`, which kills the
+  `docker logs` process so a quiet server cannot leave the reader parked in a
+  blocking read.
+- **The follower attaches with `--tail 0`.** Because every line it reads is now
+  persisted as an event, replaying a backlog recorded the same events again on
+  every restart and re-attach. Connecting clients still receive scrollback, which
+  is sent separately and does not reach the bus.
+- **RCON retries are limited to commands that provably never reached the
+  server.** Minecraft commands are not idempotent, so a lost response is now
+  reported as an unknown outcome rather than retried or re-run through the shell
+  fallback, which could otherwise apply a `give` or a `kill` twice.
+- **The RCON config is re-read when it changes on disk**, so rotating the
+  password with `scripts/rcon-setup.sh` no longer needs an API restart.
+- **Buffered events are flushed on a timer as well as on publish**, so the last
+  few events on a quiet server are not left in memory indefinitely.
+
 - **Repository and documentation cleanup**
 
   - Consolidated AI assistant configuration on `AGENTS.md` as the single source of
diff --git a/api/events.py b/api/events.py
new file mode 100644
index 0000000..3784b88
--- /dev/null
+++ b/api/events.py
@@ -0,0 +1,502 @@
+#!/usr/bin/env python3
+"""Game event bus: turns Minecraft log lines into typed, persisted events.
+
+The API already follows the container log to stream raw text to the browser.
+That stream is the only real-time signal the server produces, but as raw text it
+can only be displayed, not acted on. This module parses each line into a typed
+event, appends it to a daily JSONL file, and hands it to any registered handler.
+
+Everything reactive builds on this: death messages, join notifications,
+chat-triggered features, statistics that do not depend on re-reading the whole
+log. ``scripts/player-stats-tracker.sh`` re-parses the entire log on every run
+and adds to the previous totals, so its counts inflate; events recorded here are
+recorded exactly once.
+
+Writes are buffered. The Pi runs from an SD card with finite write endurance, so
+a continuous stream of one-line appends is flushed in batches rather than per
+event.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import threading
+from dataclasses import asdict, dataclass, field
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Callable, Iterator, Optional
+
+PROJECT_ROOT = Path(__file__).parent.parent
+EVENTS_DIR = PROJECT_ROOT / "data" / "events"
+
+# Event type constants. Kept as plain strings so they serialise directly.
+EVENT_CHAT = "chat"
+EVENT_CONNECT = "connect"
+EVENT_JOIN = "join"
+EVENT_LEAVE = "leave"
+EVENT_DEATH = "death"
+EVENT_ADVANCEMENT = "advancement"
+EVENT_COMMAND = "command"
+EVENT_SERVER_READY = "server_ready"
+EVENT_SERVER_STOPPING = "server_stopping"
+
+ALL_EVENT_TYPES = (
+    EVENT_CHAT,
+    EVENT_CONNECT,
+    EVENT_JOIN,
+    EVENT_LEAVE,
+    EVENT_DEATH,
+    EVENT_ADVANCEMENT,
+    EVENT_COMMAND,
+    EVENT_SERVER_READY,
+    EVENT_SERVER_STOPPING,
+)
+
+# Flush after this many buffered events, or this many seconds, whichever first.
+FLUSH_EVERY_EVENTS = 20
+FLUSH_EVERY_SECONDS = 30.0
+# Daily files older than this are pruned, again to protect the SD card.
+DEFAULT_RETENTION_DAYS = 30
+
+# Minecraft usernames: 3-16 characters of word characters.
+_NAME = r"[A-Za-z0-9_]{3,16}"
+
+# Strips the log prefix: "[16:04:23] [Server thread/INFO]: message".
+# Paper and Fabric use other thread names, so the thread field is matched loosely.
+_LINE_RE = re.compile(r"^\[(?P<time>\d{2}:\d{2}:\d{2})\]\s*\[(?P<thread>[^\]]+)\]:\s*(?P<message>.*)$")
+
+_CHAT_RE = re.compile(rf"^<(?P<player>{_NAME})>\s?(?P<message>.*)$")
+_JOIN_RE = re.compile(rf"^(?P<player>{_NAME}) joined the game$")
+_LEAVE_RE = re.compile(rf"^(?P<player>{_NAME}) left the game$")
+_COMMAND_RE = re.compile(rf"^(?P<player>{_NAME}) issued server command: (?P<command>.+)$")
+_ADVANCEMENT_RE = re.compile(
+    rf"^(?P<player>{_NAME}) has (?:made the advancement|completed the challenge|reached the goal) "
+    r"\[(?P<advancement>[^\]]+)\]$"
+)
+_LOGIN_RE = re.compile(rf"^(?P<player>{_NAME})\[/(?P<address>[^\]]+)\] logged in with entity id")
+_READY_RE = re.compile(r'^Done \((?P<startup>[^)]+)\)! For help, type "help"$')
+_STOPPING_RE = re.compile(r"^Stopping (?:the )?server$")
+
+# Vanilla death messages have no marker of their own; they are a bare sentence
+# beginning with the player's name. These are the opening phrases of the
+# death.attack.* strings in 1.20.4. Anything not matched here is simply not
+# treated as a death, which is the safe direction to fail.
+_DEATH_PHRASES = (
+    r"was shot by",
+    r"was pummeled by",
+    r"was pricked to death",
+    r"walked into a cactus while trying to escape",
+    r"drowned",
+    r"died from dehydration",
+    r"died",
+    r"experienced kinetic energy",
+    r"blew up",
+    r"was blown up by",
+    r"was killed by",
+    r"hit the ground too hard",
+    r"fell from a high place",
+    r"fell off (?:a ladder|some vines|some weeping vines|some twisting vines|scaffolding)",
+    r"fell while climbing",
+    r"was impaled on a stalagmite",
+    r"was squashed by",
+    r"was skewered by a falling stalactite",
+    r"went up in flames",
+    r"burned to death",
+    r"was burnt to a crisp while fighting",
+    r"went off with a bang",
+    r"tried to swim in lava",
+    r"was struck by lightning",
+    r"discovered the floor was lava",
+    r"walked into the danger zone",
+    r"froze to death",
+    r"was frozen to death by",
+    r"was slain by",
+    r"was fireballed by",
+    r"was stung to death",
+    r"was shot by a skull from",
+    r"starved to death",
+    r"suffocated in a wall",
+    r"was squished too much",
+    r"was poked to death by a sweet berry bush",
+    r"was killed trying to hurt",
+    r"was impaled by",
+    r"fell out of the world",
+    r"didn't want to live in the same world as",
+    r"withered away",
+    r"was roared at by",
+    r"was obliterated by a sonically-charged shriek",
+    r"left the confines of this world",
+    r"got finished off by",
+    r"was doomed to fall",
+    r"was stung to death by",
+)
+_DEATH_RE = re.compile(rf"^(?P<player>{_NAME}) (?P<cause>(?:{'|'.join(_DEATH_PHRASES)})\b.*)$")
+
+
+@dataclass(frozen=True)
+class GameEvent:
+    """One thing that happened on the server.
+
+    ``timestamp`` is when the line was ingested, in UTC. The server's own clock
+    reading is kept in ``data["log_time"]``: log lines carry only ``HH:MM:SS``
+    with no date, so deriving a full timestamp from them breaks across midnight.
+    """
+
+    type: str
+    timestamp: str
+    player: Optional[str] = None
+    data: dict = field(default_factory=dict)
+    raw: str = ""
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+    def to_json(self) -> str:
+        return json.dumps(self.to_dict(), separators=(",", ":"))
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def parse_line(line: str) -> Optional[GameEvent]:
+    """Parse one server log line into a :class:`GameEvent`, or ``None``.
+
+    Returning ``None`` for the overwhelming majority of lines is the expected
+    case; most of a Minecraft log is startup noise and chunk bookkeeping.
+    """
+    if not line:
+        return None
+
+    match = _LINE_RE.match(line.strip())
+    if not match:
+        return None
+
+    log_time = match.group("time")
+    message = match.group("message").strip()
+    if not message:
+        return None
+
+    base = {"log_time": log_time}
+
+    # Chat is checked first and deliberately. A player can type anything,
+    # including a sentence that looks exactly like a death message, so the
+    # "<name> " form has to win before any other pattern is tried.
+    chat = _CHAT_RE.match(message)
+    if chat:
+        return GameEvent(
+            type=EVENT_CHAT,
+            timestamp=_now_iso(),
+            player=chat.group("player"),
+            data={**base, "message": chat.group("message")},
+            raw=line,
+        )
+
+    join = _JOIN_RE.match(message)
+    if join:
+        return GameEvent(EVENT_JOIN, _now_iso(), join.group("player"), dict(base), line)
+
+    leave = _LEAVE_RE.match(message)
+    if leave:
+        return GameEvent(EVENT_LEAVE, _now_iso(), leave.group("player"), dict(base), line)
+
+    advancement = _ADVANCEMENT_RE.match(message)
+    if advancement:
+        return GameEvent(
+            type=EVENT_ADVANCEMENT,
+            timestamp=_now_iso(),
+            player=advancement.group("player"),
+            data={**base, "advancement": advancement.group("advancement")},
+            raw=line,
+        )
+
+    command = _COMMAND_RE.match(message)
+    if command:
+        return GameEvent(
+            type=EVENT_COMMAND,
+            timestamp=_now_iso(),
+            player=command.group("player"),
+            data={**base, "command": command.group("command")},
+            raw=line,
+        )
+
+    login = _LOGIN_RE.match(message)
+    if login:
+        # A player session produces two lines: this one, carrying the network
+        # address, and "joined the game" a moment later. They are separate types
+        # so that counting joins does not double-count every session. The
+        # address is recorded but not surfaced anywhere by default; it is here so
+        # a future "a new device connected" alert can use it.
+        return GameEvent(
+            type=EVENT_CONNECT,
+            timestamp=_now_iso(),
+            player=login.group("player"),
+            data={**base, "address": login.group("address")},
+            raw=line,
+        )
+
+    death = _DEATH_RE.match(message)
+    if death:
+        return GameEvent(
+            type=EVENT_DEATH,
+            timestamp=_now_iso(),
+            player=death.group("player"),
+            data={**base, "cause": death.group("cause"), "message": message},
+            raw=line,
+        )
+
+    ready = _READY_RE.match(message)
+    if ready:
+        return GameEvent(
+            type=EVENT_SERVER_READY,
+            timestamp=_now_iso(),
+            data={**base, "startup_time": ready.group("startup")},
+            raw=line,
+        )
+
+    if _STOPPING_RE.match(message):
+        return GameEvent(EVENT_SERVER_STOPPING, _now_iso(), None, dict(base), line)
+
+    return None
+
+
+class EventBus:
+    """Parses, persists and dispatches game events.
+
+    Handlers are called synchronously in registration order. A handler that
+    raises is logged and skipped rather than being allowed to take down the
+    reader thread that feeds the bus.
+    """
+
+    def __init__(
+        self,
+        events_dir: Optional[Path] = None,
+        retention_days: int = DEFAULT_RETENTION_DAYS,
+        flush_every_events: int = FLUSH_EVERY_EVENTS,
+        flush_every_seconds: float = FLUSH_EVERY_SECONDS,
+    ) -> None:
+        self.events_dir = events_dir if events_dir is not None else EVENTS_DIR
+        self.retention_days = retention_days
+        self.flush_every_events = flush_every_events
+        self.flush_every_seconds = flush_every_seconds
+
+        self._handlers: list[Callable[[GameEvent], None]] = []
+        self._buffer: list[GameEvent] = []
+        self._lock = threading.RLock()
+        # Separate from _lock: the buffer swap is quick, but the file append
+        # must not interleave with another flush, and holding _lock across the
+        # write would block every publish for the duration of the I/O.
+        self._write_lock = threading.Lock()
+        self._last_flush = datetime.now(timezone.utc)
+        self._last_pruned_date: Optional[str] = None
+        self._error_logger: Optional[Callable[[str], None]] = None
+        self._flush_timer: Optional[threading.Timer] = None
+
+    def set_error_logger(self, logger: Callable[[str], None]) -> None:
+        """Route handler errors somewhere visible, normally ``app.logger``."""
+        self._error_logger = logger
+
+    def _log_error(self, message: str) -> None:
+        if self._error_logger is not None:
+            try:
+                self._error_logger(message)
+            except Exception:  # noqa: BLE001 - logging must never raise
+                # The error logger is supplied by the caller and may itself be
+                # broken or closed. Swallowing it here is deliberate: a failure
+                # to report a problem must not become a second problem on the
+                # thread that feeds the bus.
+                pass
+
+    def subscribe(self, handler: Callable[[GameEvent], None]) -> Callable[[GameEvent], None]:
+        """Register a handler. Returns it, so it can be used as a decorator."""
+        with self._lock:
+            if handler not in self._handlers:
+                self._handlers.append(handler)
+        return handler
+
+    def unsubscribe(self, handler: Callable[[GameEvent], None]) -> None:
+        with self._lock:
+            if handler in self._handlers:
+                self._handlers.remove(handler)
+
+    @property
+    def handler_count(self) -> int:
+        with self._lock:
+            return len(self._handlers)
+
+    def handle_line(self, line: str) -> Optional[GameEvent]:
+        """Parse a log line and publish it if it is an event."""
+        event = parse_line(line)
+        if event is not None:
+            self.publish(event)
+        return event
+
+    def publish(self, event: GameEvent) -> None:
+        """Persist an event and hand it to every handler."""
+        with self._lock:
+            self._buffer.append(event)
+            handlers = list(self._handlers)
+            should_flush = len(self._buffer) >= self.flush_every_events or (
+                (datetime.now(timezone.utc) - self._last_flush).total_seconds() >= self.flush_every_seconds
+            )
+
+        if should_flush:
+            self.flush()
+
+        for handler in handlers:
+            try:
+                handler(event)
+            except Exception as exc:  # noqa: BLE001 - one bad handler must not stop the rest
+                self._log_error(f"Event handler {getattr(handler, '__name__', handler)!r} failed: {exc}")
+
+    def flush(self) -> int:
+        """Write buffered events to today's file. Returns the number written."""
+        with self._lock:
+            if not self._buffer:
+                self._last_flush = datetime.now(timezone.utc)
+                return 0
+            pending, self._buffer = self._buffer, []
+            self._last_flush = datetime.now(timezone.utc)
+
+        today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
+        try:
+            with self._write_lock:
+                self.events_dir.mkdir(parents=True, exist_ok=True)
+                target = self.events_dir / f"{today}.jsonl"
+                with open(target, "a", encoding="utf-8") as handle:
+                    for event in pending:
+                        handle.write(event.to_json() + "\n")
+        except OSError as exc:
+            # Losing an event to a full or read-only disk must not stop the
+            # server from running, so this is reported and dropped.
+            self._log_error(f"Could not persist {len(pending)} event(s): {exc}")
+            return 0
+
+        if self._last_pruned_date != today:
+            self._last_pruned_date = today
+            self.prune()
+
+        return len(pending)
+
+    def start_periodic_flush(self) -> None:
+        """Flush on a timer as well as on publish.
+
+        The size and age thresholds are only evaluated when something is
+        published, so on a quiet server the last few events would sit in memory
+        until the next one arrived, and be lost if the process stopped first.
+        A daemon timer bounds that window to ``flush_every_seconds``.
+        """
+        with self._lock:
+            if self._flush_timer is not None:
+                return
+            self._schedule_flush_locked()
+
+    def stop_periodic_flush(self) -> None:
+        """Cancel the flush timer and write out anything still buffered."""
+        with self._lock:
+            timer, self._flush_timer = self._flush_timer, None
+        if timer is not None:
+            timer.cancel()
+        self.flush()
+
+    def _schedule_flush_locked(self) -> None:
+        """Arm the next timer tick. Caller holds ``_lock``."""
+        timer = threading.Timer(self.flush_every_seconds, self._periodic_flush)
+        timer.daemon = True
+        self._flush_timer = timer
+        timer.start()
+
+    def _periodic_flush(self) -> None:
+        try:
+            self.flush()
+        except Exception as exc:  # noqa: BLE001 - a timer thread must not die
+            self._log_error(f"Periodic flush failed: {exc}")
+        finally:
+            with self._lock:
+                if self._flush_timer is not None:
+                    self._schedule_flush_locked()
+
+    def prune(self) -> int:
+        """Delete daily files older than the retention window."""
+        if self.retention_days <= 0 or not self.events_dir.exists():
+            return 0
+
+        cutoff = (datetime.now(timezone.utc) - timedelta(days=self.retention_days)).date()
+        removed = 0
+        for path in self.events_dir.glob("*.jsonl"):
+            try:
+                file_date = datetime.strptime(path.stem, "%Y-%m-%d").date()
+            except ValueError:
+                continue
+            if file_date < cutoff:
+                try:
+                    path.unlink()
+                    removed += 1
+                except OSError:
+                    continue
+        return removed
+
+    def read(self, limit: int = 100, event_type: Optional[str] = None, player: Optional[str] = None) -> list[dict]:
+        """Return the most recent stored events, newest first.
+
+        Buffered events are flushed first so a caller never sees a stale view of
+        what just happened.
+        """
+        self.flush()
+
+        collected: list[dict] = []
+        for path in sorted(self.events_dir.glob("*.jsonl"), reverse=True) if self.events_dir.exists() else []:
+            for record in _read_jsonl_reversed(path):
+                if event_type and record.get("type") != event_type:
+                    continue
+                if player and record.get("player") != player:
+                    continue
+                collected.append(record)
+                if len(collected) >= limit:
+                    return collected
+        return collected
+
+
+def _read_jsonl_reversed(path: Path) -> Iterator[dict]:
+    """Yield the records of a JSONL file newest first.
+
+    Daily event files stay small enough to read whole; a busy day on a family
+    server is measured in thousands of lines, not millions.
+    """
+    try:
+        lines = path.read_text(encoding="utf-8").splitlines()
+    except OSError:
+        return
+    for line in reversed(lines):
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            yield json.loads(line)
+        except json.JSONDecodeError:
+            # A torn final line from an interrupted write; skip it.
+            continue
+
+
+# Shared bus, created on first use.
+_bus: Optional[EventBus] = None
+_bus_lock = threading.Lock()
+
+
+def get_bus() -> EventBus:
+    """Return the process-wide event bus."""
+    global _bus
+    with _bus_lock:
+        if _bus is None:
+            _bus = EventBus()
+        return _bus
+
+
+def reset_bus() -> None:
+    """Drop the shared bus. Used by tests."""
+    global _bus
+    with _bus_lock:
+        if _bus is not None:
+            _bus.stop_periodic_flush()
+        _bus = None
diff --git a/api/openapi.yaml b/api/openapi.yaml
index 9610c52..2f2a9f3 100644
--- a/api/openapi.yaml
+++ b/api/openapi.yaml
@@ -86,6 +86,31 @@ components:
       description: Session cookie from login
 
   schemas:
+    GameEvent:
+      type: object
+      properties:
+        type:
+          type: string
+          description: Event type
+          example: death
+        timestamp:
+          type: string
+          format: date-time
+          description: When the line was ingested, in UTC
+        player:
+          type: string
+          nullable: true
+          description: Player the event concerns, if any
+        data:
+          type: object
+          description: >-
+            Type-specific fields. Always includes log_time, the HH:MM:SS the
+            server itself printed.
+          additionalProperties: true
+        raw:
+          type: string
+          description: The original log line
+
     Error:
       type: object
       properties:
@@ -643,6 +668,85 @@ paths:
                   total_lines:
                     type: integer
 
+  /api/events:
+    get:
+      tags:
+        - Events
+      summary: Get game events
+      description: |
+        Retrieve game events parsed from the server log, newest first.
+        Requires the `logs.view` permission.
+      operationId: getEvents
+      parameters:
+        - name: limit
+          in: query
+          schema:
+            type: integer
+            default: 100
+            minimum: 1
+            maximum: 500
+          description: Maximum number of events to return
+        - name: type
+          in: query
+          schema:
+            type: string
+            enum:
+              - chat
+              - connect
+              - join
+              - leave
+              - death
+              - advancement
+              - command
+              - server_ready
+              - server_stopping
+          description: Only return events of this type
+        - name: player
+          in: query
+          schema:
+            type: string
+          description: Only return events for this player
+      responses:
+        '200':
+          description: Game events, newest first
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  events:
+                    type: array
+                    items:
+                      $ref: '#/components/schemas/GameEvent'
+                  count:
+                    type: integer
+        '400':
+          description: Unknown event type or non-numeric limit
+        '503':
+          description: Event capture is unavailable
+
+  /api/events/types:
+    get:
+      tags:
+        - Events
+      summary: List game event types
+      description: List the event types the bus can produce. Requires `logs.view`.
+      operationId: getEventTypes
+      responses:
+        '200':
+          description: Known event types
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  types:
+                    type: array
+                    items:
+                      type: string
+        '503':
+          description: Event capture is unavailable
+
   /api/metrics:
     get:
       tags:
diff --git a/api/rcon.py b/api/rcon.py
new file mode 100644
index 0000000..fe7fb8e
--- /dev/null
+++ b/api/rcon.py
@@ -0,0 +1,449 @@
+#!/usr/bin/env python3
+"""In-process RCON client for the Minecraft server.
+
+Replaces the per-command ``subprocess`` shell-out to ``scripts/rcon-client.sh``.
+That path opened a fresh TCP connection and re-authenticated for every single
+command, which costs roughly 50-100ms each and makes command batches
+impractical. It also read a single 4096-byte chunk without length framing, so
+long responses were truncated or split across reads, and it treated any reply
+of four or more bytes as a successful login instead of checking for the
+``-1`` request id the protocol uses to signal auth failure.
+
+This module keeps one authenticated connection open, frames packets properly,
+reassembles multi-packet responses, and reconnects transparently when the
+server restarts underneath it.
+
+Protocol reference: the Source RCON protocol as implemented by Minecraft's
+``RconClient``. Responses longer than 4096 bytes are split by the server into
+several packets that all carry the original request id, so a sentinel packet is
+used to detect the end of a response.
+"""
+
+from __future__ import annotations
+
+import socket
+import struct
+import threading
+from pathlib import Path
+from typing import Iterable, Optional
+
+PROJECT_ROOT = Path(__file__).parent.parent
+RCON_CONFIG_FILE = PROJECT_ROOT / "config" / "rcon.conf"
+
+# Packet types from the Source RCON protocol
+PACKET_TYPE_RESPONSE = 0
+PACKET_TYPE_COMMAND = 2
+PACKET_TYPE_LOGIN = 3
+PACKET_TYPE_AUTH_RESPONSE = 2
+
+# A packet is at minimum two int32 fields plus the two terminating nulls.
+MIN_PACKET_LENGTH = 10
+# Generous ceiling so a malformed length field cannot force a huge allocation.
+MAX_PACKET_LENGTH = 64 * 1024
+# Minecraft's inbound RCON buffer is 1460 bytes; longer commands are silently
+# truncated by the server, so reject them here where the caller can see why.
+MAX_COMMAND_LENGTH = 1400
+
+DEFAULT_HOST = "localhost"
+DEFAULT_PORT = 25575
+DEFAULT_TIMEOUT = 5.0
+
+
+class RconError(Exception):
+    """Base class for every RCON failure."""
+
+
+class RconNotConfiguredError(RconError):
+    """Raised when no RCON password is available.
+
+    Callers treat this as "fall back to the shell script" rather than as a hard
+    failure, because the script can also reach ``rcon-cli`` inside the container.
+    """
+
+
+class RconAuthError(RconError):
+    """Raised when the server rejects the RCON password."""
+
+
+class RconConnectionError(RconError):
+    """Raised when the socket cannot be established or is lost mid-command."""
+
+
+class RconNotSentError(RconConnectionError):
+    """Raised when a command failed before any of it reached the server.
+
+    Safe to retry: the server never saw it. This is the ordinary case of a
+    pooled connection that went stale while idle, where the first write fails
+    with a broken pipe or a reset.
+    """
+
+
+class RconUnknownOutcomeError(RconError):
+    """Raised when a command reached the server but its result was not read.
+
+    Must not be retried. The server may well have run it, so repeating a
+    ``give``, ``summon``, ``kill`` or ``stop`` would apply the effect twice.
+    """
+
+
+def load_rcon_config(config_file: Optional[Path] = None) -> dict:
+    """Read ``config/rcon.conf`` into a dict of connection settings.
+
+    The file is written by ``scripts/rcon-setup.sh`` as plain ``KEY=value``
+    shell assignments. Missing files yield defaults with an empty password,
+    which callers detect via :class:`RconNotConfiguredError`.
+    """
+    path = config_file if config_file is not None else RCON_CONFIG_FILE
+    settings = {"host": DEFAULT_HOST, "port": DEFAULT_PORT, "password": ""}
+
+    if not path.exists():
+        return settings
+
+    try:
+        content = path.read_text(encoding="utf-8")
+    except OSError:
+        return settings
+
+    for line in content.splitlines():
+        line = line.strip()
+        if not line or line.startswith("#") or "=" not in line:
+            continue
+        key, value = line.split("=", 1)
+        key = key.strip()
+        # Shell config files may quote values; RCON passwords are generated
+        # without quotes but tolerate both forms.
+        value = value.strip().strip("\"'")
+
+        if key == "RCON_HOST":
+            settings["host"] = value or DEFAULT_HOST
+        elif key == "RCON_PORT":
+            try:
+                settings["port"] = int(value)
+            except ValueError:
+                settings["port"] = DEFAULT_PORT
+        elif key == "RCON_PASSWORD":
+            settings["password"] = value
+
+    return settings
+
+
+def _encode_packet(request_id: int, packet_type: int, body: str) -> bytes:
+    """Build one RCON packet, including its leading length field."""
+    payload = struct.pack("<ii", request_id, packet_type) + body.encode("utf-8") + b"\x00\x00"
+    return struct.pack("<i", len(payload)) + payload
+
+
+def _recv_exactly(sock: socket.socket, count: int) -> bytes:
+    """Read exactly ``count`` bytes, looping until the buffer is full.
+
+    A bare ``recv`` returns whatever happens to have arrived, which is what made
+    the previous implementation lose the tail of long responses.
+    """
+    chunks = []
+    remaining = count
+    while remaining > 0:
+        chunk = sock.recv(remaining)
+        if not chunk:
+            raise RconConnectionError("Connection closed by server")
+        chunks.append(chunk)
+        remaining -= len(chunk)
+    return b"".join(chunks)
+
+
+def _read_packet(sock: socket.socket) -> tuple[int, int, str]:
+    """Read one framed packet and return ``(request_id, packet_type, body)``."""
+    (length,) = struct.unpack("<i", _recv_exactly(sock, 4))
+    if length < MIN_PACKET_LENGTH or length > MAX_PACKET_LENGTH:
+        raise RconConnectionError(f"Malformed RCON packet length: {length}")
+
+    payload = _recv_exactly(sock, length)
+    request_id, packet_type = struct.unpack("<ii", payload[:8])
+    # Drop the two trailing null terminators. Bodies are UTF-8; replace rather
+    # than raise so one odd character cannot break an otherwise good response.
+    body = payload[8:-2].decode("utf-8", errors="replace")
+    return request_id, packet_type, body
+
+
+class RconClient:
+    """A persistent, thread-safe RCON connection.
+
+    One connection is shared by all callers and guarded by a lock, so commands
+    are serialised rather than racing for the socket. If the connection has been
+    dropped, typically because the Minecraft container restarted, the next
+    command reconnects and retries once.
+    """
+
+    def __init__(
+        self,
+        host: str = DEFAULT_HOST,
+        port: int = DEFAULT_PORT,
+        password: str = "",
+        timeout: float = DEFAULT_TIMEOUT,
+    ) -> None:
+        self.host = host
+        self.port = port
+        self.password = password
+        self.timeout = timeout
+        self._sock: Optional[socket.socket] = None
+        self._request_id = 0
+        self._lock = threading.RLock()
+
+    @property
+    def is_connected(self) -> bool:
+        """Whether a socket is currently open and authenticated."""
+        return self._sock is not None
+
+    def _next_request_id(self) -> int:
+        """Return a fresh request id, wrapping well below int32 limits.
+
+        ``-1`` is reserved by the protocol to signal auth failure, so ids stay
+        positive.
+        """
+        self._request_id = (self._request_id % 0x3FFFFFFF) + 1
+        return self._request_id
+
+    def connect(self) -> None:
+        """Open the socket and authenticate. Safe to call when already open."""
+        with self._lock:
+            if self._sock is not None:
+                return
+            if not self.password:
+                raise RconNotConfiguredError("No RCON password configured; run scripts/rcon-setup.sh")
+
+            try:
+                sock = socket.create_connection((self.host, self.port), timeout=self.timeout)
+            except OSError as exc:
+                raise RconConnectionError(f"Cannot connect to RCON at {self.host}:{self.port}: {exc}") from exc
+
+            sock.settimeout(self.timeout)
+            try:
+                self._authenticate(sock)
+            except Exception:
+                _close_quietly(sock)
+                raise
+
+            self._sock = sock
+
+    def _authenticate(self, sock: socket.socket) -> None:
+        """Send the login packet and verify the server accepted it.
+
+        The server may emit an empty RESPONSE_VALUE packet before the auth
+        response, so packets are read until the auth response arrives. A request
+        id of ``-1`` means the password was wrong.
+        """
+        request_id = self._next_request_id()
+        sock.sendall(_encode_packet(request_id, PACKET_TYPE_LOGIN, self.password))
+
+        for _ in range(3):
+            try:
+                response_id, packet_type, _ = _read_packet(sock)
+            except socket.timeout as exc:
+                raise RconConnectionError("Timed out waiting for RCON auth response") from exc
+
+            if response_id == -1:
+                raise RconAuthError("RCON authentication failed: incorrect password")
+            if packet_type == PACKET_TYPE_AUTH_RESPONSE and response_id == request_id:
+                return
+
+        raise RconConnectionError("Did not receive a usable RCON auth response")
+
+    def close(self) -> None:
+        """Close the connection if one is open."""
+        with self._lock:
+            if self._sock is not None:
+                _close_quietly(self._sock)
+                self._sock = None
+
+    def command(self, command: str) -> str:
+        """Run one command and return the server's response text.
+
+        Retries exactly once, and only when the command provably never left
+        this process. Minecraft commands are not idempotent: re-sending a
+        ``give`` or a ``summon`` after a lost response would apply it twice, so
+        a failure that happens after the command reached the server is reported
+        as :class:`RconUnknownOutcomeError` rather than retried.
+
+        The retry therefore covers the one case that is both common and safe: a
+        pooled connection that went stale while idle, where the first write
+        fails before any bytes reach the server.
+        """
+        if not isinstance(command, str):
+            raise RconError("Command must be a string")
+        command = command.strip()
+        if not command:
+            raise RconError("Command must not be empty")
+        if len(command.encode("utf-8")) > MAX_COMMAND_LENGTH:
+            raise RconError(f"Command exceeds {MAX_COMMAND_LENGTH} bytes and would be truncated by the server")
+
+        with self._lock:
+            self.connect()
+            try:
+                return self._send_command(command)
+            except RconNotSentError:
+                # Nothing reached the server, so a fresh connection can safely
+                # carry the same command.
+                self.close()
+                self.connect()
+                return self._send_command(command)
+
+    def commands(self, commands: Iterable[str]) -> list[str]:
+        """Run several commands over the one connection, in order.
+
+        The lock is held for the whole batch so nothing interleaves between
+        commands that are meant to run as a unit.
+        """
+        with self._lock:
+            return [self.command(command) for command in commands]
+
+    def _send_command(self, command: str) -> str:
+        """Send a command and reassemble a possibly multi-packet response.
+
+        Minecraft splits responses longer than 4096 bytes into several packets
+        that all carry the command's request id, and the protocol gives no
+        length up front. A second packet with an unused type is sent straight
+        after the command; the server answers it with a recognisable "Unknown
+        request" reply, which marks the end of the real response.
+        """
+        sock = self._sock
+        if sock is None:
+            raise RconNotSentError("Not connected")
+
+        request_id = self._next_request_id()
+        sentinel_id = self._next_request_id()
+
+        try:
+            sock.sendall(_encode_packet(request_id, PACKET_TYPE_COMMAND, command))
+        except OSError as exc:
+            # A command packet is well under one segment, so a stale socket
+            # fails here with nothing delivered. This is the retryable case.
+            raise RconNotSentError(f"Could not send command: {exc}") from exc
+
+        # Past this point the server has the command. Every failure below leaves
+        # the outcome unknown and must not be retried.
+        try:
+            sock.sendall(_encode_packet(sentinel_id, PACKET_TYPE_RESPONSE, ""))
+        except OSError as exc:
+            raise RconUnknownOutcomeError(f"Command was sent but the connection failed: {exc}") from exc
+
+        parts: list[str] = []
+        while True:
+            try:
+                response_id, _, body = _read_packet(sock)
+            except socket.timeout:
+                # Some server builds ignore the sentinel. Anything already
+                # received is the real response, so return it rather than
+                # failing the command.
+                if parts:
+                    break
+                raise RconUnknownOutcomeError("Command was sent but no response arrived") from None
+            except (OSError, RconConnectionError) as exc:
+                raise RconUnknownOutcomeError(f"Command was sent but the response was lost: {exc}") from exc
+
+            if response_id == sentinel_id:
+                break
+            if response_id == request_id:
+                parts.append(body)
+
+        return "".join(parts)
+
+    def __enter__(self) -> "RconClient":
+        self.connect()
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback) -> None:
+        self.close()
+
+
+def _close_quietly(sock: socket.socket) -> None:
+    """Close a socket, ignoring errors from an already-dead connection."""
+    try:
+        sock.close()
+    except OSError:
+        # The connection is being discarded either way. A socket that is
+        # already closed, reset, or whose fd is gone raises here, and there is
+        # nothing useful left to do about it.
+        pass
+
+
+# Module-level shared client. Built lazily so importing this module never
+# touches the network or the config file.
+_client: Optional[RconClient] = None
+_client_config_mtime: Optional[float] = None
+_client_lock = threading.Lock()
+
+
+def _config_mtime() -> Optional[float]:
+    """Last-modified time of the RCON config, or None if it is absent."""
+    try:
+        return RCON_CONFIG_FILE.stat().st_mtime
+    except OSError:
+        return None
+
+
+def get_client(timeout: float = DEFAULT_TIMEOUT) -> RconClient:
+    """Return the shared client, creating it from config on first use.
+
+    The config file is re-read when it changes on disk, so rotating the RCON
+    password with ``scripts/rcon-setup.sh`` takes effect without restarting the
+    API.
+    """
+    global _client, _client_config_mtime
+    with _client_lock:
+        mtime = _config_mtime()
+        if _client is not None and mtime != _client_config_mtime:
+            _client.close()
+            _client = None
+
+        if _client is None:
+            settings = load_rcon_config()
+            _client = RconClient(
+                host=settings["host"],
+                port=settings["port"],
+                password=settings["password"],
+                timeout=timeout,
+            )
+            _client_config_mtime = mtime
+        return _client
+
+
+def reset_client() -> None:
+    """Drop the shared client so the next call re-reads config.
+
+    Used by tests, and available for forcing a reconnect.
+    """
+    global _client, _client_config_mtime
+    with _client_lock:
+        if _client is not None:
+            _client.close()
+        _client = None
+        _client_config_mtime = None
+
+
+def execute(command: str) -> tuple[Optional[str], Optional[str], int]:
+    """Run a command, returning ``(stdout, stderr, returncode)``.
+
+    Matches the shape of ``run_script`` in ``api/server.py`` so call sites can
+    use either path.
+
+    The code tells the caller whether retrying elsewhere is safe:
+
+    * ``503`` — not configured, nothing was sent. Falling back to
+      ``scripts/rcon-client.sh`` is safe, and worthwhile because that script can
+      also reach ``rcon-cli`` inside the container.
+    * ``502`` — could not reach the server, nothing was sent. Fallback is safe.
+    * ``500`` — the command reached the server but its result was lost. The
+      outcome is unknown, so it must **not** be run again by any other path.
+    * ``401`` / ``400`` — rejected. Retrying repeats the same failure.
+    """
+    try:
+        return get_client().command(command), None, 0
+    except RconNotConfiguredError as exc:
+        return None, str(exc), 503
+    except RconAuthError as exc:
+        return None, str(exc), 401
+    except RconUnknownOutcomeError as exc:
+        return None, str(exc), 500
+    except RconConnectionError as exc:
+        return None, str(exc), 502
+    except RconError as exc:
+        return None, str(exc), 400
diff --git a/api/server.py b/api/server.py
index 16cb4ca..16b6ba3 100644
--- a/api/server.py
+++ b/api/server.py
@@ -87,6 +87,27 @@ def is_rate_limit_exceeded(*args, **kwargs):
         return False
 
 
+# In-process RCON client. Keeps one authenticated connection open instead of
+# paying for a TCP handshake and login per command.
+try:
+    from api import rcon
+
+    RCON_AVAILABLE = True
+except ImportError:
+    RCON_AVAILABLE = False
+    rcon = None
+
+# Game event bus. Parses the server log into typed events that features can
+# subscribe to, instead of every feature re-parsing raw text for itself.
+try:
+    from api import events as game_events
+
+    EVENTS_AVAILABLE = True
+except ImportError:
+    EVENTS_AVAILABLE = False
+    game_events = None
+
+
 app = Flask(__name__)
 # SECRET_KEY is resolved further down, once config/api.conf has been read.
 app.config["MAX_CONTENT_LENGTH"] = 16 * 1024 * 1024  # 16MB max request size
@@ -360,6 +381,28 @@ def run_script(script_name, *args, timeout=DEFAULT_SCRIPT_TIMEOUT):
         return None, str(e), 500
 
 
+def run_rcon_command(command):
+    """Execute a Minecraft command over RCON, returning (stdout, stderr, code).
+
+    Prefers the pooled in-process client in ``api/rcon.py``, which reuses one
+    authenticated connection. Falls back to ``scripts/rcon-client.sh`` only when
+    the command provably never reached the server, because that script can reach
+    ``rcon-cli`` inside the container when the port is not published to the host.
+
+    Minecraft commands are not idempotent. ``502`` and ``503`` mean nothing was
+    sent, so running the command by another route is safe. Every other code is
+    returned as-is: an auth failure or a rejected command would fail the same
+    way again, and ``500`` means the command reached the server but its result
+    was lost, so repeating it could apply a ``give`` or a ``kill`` twice.
+    """
+    if RCON_AVAILABLE:
+        stdout, stderr, code = rcon.execute(command)
+        if code not in (502, 503):
+            return stdout, stderr, code
+
+    return run_script("rcon-client.sh", "command", command)
+
+
 @app.route("/api/health", methods=["GET"])
 def health():
     """Health check endpoint"""
@@ -1791,6 +1834,10 @@ def send_command():
     command = data.get("command")
     if not command:
         return jsonify({"error": "Command required"}), 400
+    if not isinstance(command, str):
+        # The sanitizer calls string methods, so a JSON number or object would
+        # raise inside it and surface as a 500 rather than a bad request.
+        return jsonify({"error": "Command must be a string"}), 400
 
     # Validate and sanitize command to prevent command injection
     if SECURITY_AVAILABLE:
@@ -1810,7 +1857,7 @@ def send_command():
     username = get_username_from_request()
     log_audit_event(username, "server.command", {"command": sanitize_string(command[:100])})
 
-    stdout, stderr, code = run_script("rcon-client.sh", "command", command)
+    stdout, stderr, code = run_rcon_command(command)
 
     if code == 0:
         # Sanitize response before returning
@@ -2174,11 +2221,53 @@ def get_logs():
     return jsonify({"logs": logs.split("\n"), "lines": len(logs.split("\n"))})
 
 
+@app.route("/api/events", methods=["GET"])
+@require_permission("logs.view")
+def get_game_events():
+    """Return recent game events, newest first.
+
+    Query parameters: `limit` (default 100, capped at 500), `type` to filter by
+    event type, and `player` to filter by player name.
+    """
+    if not EVENTS_AVAILABLE:
+        return jsonify({"error": "Event capture is unavailable"}), 503
+
+    try:
+        limit = min(max(int(request.args.get("limit", 100)), 1), 500)
+    except (TypeError, ValueError):
+        return jsonify({"error": "limit must be an integer"}), 400
+
+    event_type = request.args.get("type")
+    if event_type and event_type not in game_events.ALL_EVENT_TYPES:
+        return jsonify({"error": "Unknown event type", "valid_types": list(game_events.ALL_EVENT_TYPES)}), 400
+
+    player = request.args.get("player")
+    if player:
+        player = sanitize_string(player, max_length=16)
+
+    try:
+        records = game_events.get_bus().read(limit=limit, event_type=event_type, player=player)
+    except Exception as e:
+        app.logger.error(f"Error reading events: {e}")
+        return jsonify({"error": "Internal server error"}), 500
+
+    return jsonify({"events": records, "count": len(records)})
+
+
+@app.route("/api/events/types", methods=["GET"])
+@require_permission("logs.view")
+def get_game_event_types():
+    """List the event types the bus can produce."""
+    if not EVENTS_AVAILABLE:
+        return jsonify({"error": "Event capture is unavailable"}), 503
+    return jsonify({"types": list(game_events.ALL_EVENT_TYPES)})
+
+
 @app.route("/api/players", methods=["GET"])
 @require_permission("players.view")
 def get_players():
     """Get list of online players"""
-    stdout, _, _ = run_script("rcon-client.sh", "command", "list")
+    stdout, _, _ = run_rcon_command("list")
 
     # Parse player list from RCON response
     players = []
@@ -3634,8 +3723,10 @@ def save_ddns_config():
     active_log_streams = set()
     _log_streams_lock = threading.Lock()
     # Mutable holder rather than a module-level bool, so the reader and the
-    # starter share one piece of state without `global` declarations.
-    _log_reader_state = {"running": False}
+    # starter share one piece of state without `global` declarations. The
+    # follower no longer stops when the last client disconnects, so `stop` is
+    # what shutdown and the tests use to bring it down deliberately.
+    _log_reader_state = {"running": False, "stop": False, "proc": None}
 
     LOG_BACKLOG_LINES = 200
 
@@ -3654,42 +3745,76 @@ def get_log_tail(lines=100):
         except (subprocess.TimeoutExpired, FileNotFoundError):
             return []
 
+    def _publish_log_line(line):
+        """Feed one log line to the event bus, returning the event or None.
+
+        Persistence failures and handler errors are contained by the bus itself;
+        this wrapper exists so that a bus problem can never stop log streaming.
+        """
+        if not EVENTS_AVAILABLE:
+            return None
+        try:
+            return game_events.get_bus().handle_line(line)
+        except Exception as exc:  # noqa: BLE001 - streaming must outlive the bus
+            app.logger.error(f"Event bus failed on a log line: {exc}")
+            return None
+
     def _log_reader():
-        """Follow the container log once and fan each line out to subscribers.
+        """Follow the container log, fan lines out, and drive the event bus.
 
         The previous implementation gave every client its own thread that ran
         `docker logs --tail 50` every second and diffed the result against the
         last batch, which scaled badly, missed lines that scrolled past between
         polls, and duplicated any line that legitimately repeated. One follower
         process streams the log instead, so lines arrive in order, exactly once.
-        """
-        while True:
-            with _log_streams_lock:
-                if not active_log_streams:
-                    _log_reader_state["running"] = False
-                    return
 
+        It now also runs whether or not anybody is watching. The log is the only
+        real-time signal the Minecraft server produces, so a follower that
+        started on the first browser connection and stopped on the last one
+        meant every death, advancement and chat message that happened with the
+        dashboard closed was lost. Events are parsed and recorded continuously;
+        the WebSocket fanout is just one consumer of them.
+
+        It attaches with `--tail 0` deliberately. Because every line read here
+        is persisted as an event, replaying a backlog would record the same
+        deaths and advancements again on every API restart and every re-attach,
+        and the counts would climb with each one. New clients still get their
+        scrollback: `handle_connect` sends `get_log_tail()` separately, and that
+        path does not touch the bus. The trade is that events occurring while
+        the API is down are not captured, which is far better than recording
+        some of them repeatedly.
+        """
+        while not _log_reader_state["stop"]:
             proc = None
             try:
                 proc = subprocess.Popen(
-                    ["docker", "logs", "-f", "--tail", str(LOG_BACKLOG_LINES), "minecraft-server"],
+                    ["docker", "logs", "-f", "--tail", "0", "minecraft-server"],
                     stdout=subprocess.PIPE,
                     stderr=subprocess.STDOUT,
                     text=True,
                     bufsize=1,
                 )
+                with _log_streams_lock:
+                    _log_reader_state["proc"] = proc
 
                 for line in proc.stdout:
-                    with _log_streams_lock:
-                        subscribers = list(active_log_streams)
-                    if not subscribers:
+                    if _log_reader_state["stop"]:
                         break
 
                     line = line.rstrip("\n")
                     if not line.strip():
                         continue
+
+                    # Parse and persist first, so an event is recorded even if
+                    # no client is connected to receive it.
+                    event = _publish_log_line(line)
+
+                    with _log_streams_lock:
+                        subscribers = list(active_log_streams)
                     for sid in subscribers:
                         socketio.emit("logs", {"logs": [line], "type": "update"}, room=sid)
+                        if event is not None:
+                            socketio.emit("game_event", event.to_dict(), room=sid)
             except FileNotFoundError:
                 # Docker is not installed; there is nothing to stream
                 with _log_streams_lock:
@@ -3705,6 +3830,8 @@ def _log_reader():
                 for sid in subscribers:
                     socketio.emit("error", {"message": f"Log streaming error: {str(e)}"}, room=sid)
             finally:
+                with _log_streams_lock:
+                    _log_reader_state["proc"] = None
                 if proc is not None:
                     try:
                         proc.kill()
@@ -3714,18 +3841,45 @@ def _log_reader():
                         # stop the reader from re-attaching.
                         pass
 
-            # The container may have stopped or restarted. Wait a moment before
-            # re-attaching, as long as somebody is still listening.
+            if _log_reader_state["stop"]:
+                break
+
+            # The container may have stopped or restarted, and `docker logs -f`
+            # exits when it does. Pause briefly, then re-attach, so the server
+            # coming back up resumes the stream without anyone intervening.
             socketio.sleep(2)
 
+        with _log_streams_lock:
+            _log_reader_state["running"] = False
+
     def _ensure_log_reader():
         """Start the single follower thread if it is not already running"""
         with _log_streams_lock:
             if _log_reader_state["running"]:
                 return
             _log_reader_state["running"] = True
+            _log_reader_state["stop"] = False
         socketio.start_background_task(_log_reader)
 
+    def stop_log_reader():
+        """Stop the follower.
+
+        Setting the flag alone is not enough. A quiet server leaves the reader
+        blocked in `for line in proc.stdout`, where it never gets to check the
+        flag, so `docker logs -f` is killed to break the read. The reader then
+        unblocks, sees the flag and exits.
+        """
+        with _log_streams_lock:
+            _log_reader_state["stop"] = True
+            proc = _log_reader_state["proc"]
+
+        if proc is not None:
+            try:
+                proc.kill()
+            except Exception:
+                # Already exited, which is the outcome we wanted anyway.
+                pass
+
     @socketio.on("connect")
     def handle_connect(auth):
         """Handle WebSocket connection"""
@@ -3778,11 +3932,33 @@ def handle_execute_command(data):
         if not command:
             socketio.emit("command_error", {"message": "Command required"}, room=request.sid)
             return
+        if not isinstance(command, str):
+            # Raising inside the sanitizer happens before the try block below,
+            # which would leave the client waiting with no error at all.
+            socketio.emit("command_error", {"message": "Command must be a string"}, room=request.sid)
+            return
+
+        # Sanitise exactly as POST /api/server/command does. This path reached
+        # RCON unvalidated, so the WebSocket was a way around the command
+        # allowlist that the REST endpoint enforces.
+        if SECURITY_AVAILABLE:
+            is_valid, sanitized_command, _ = sanitize_minecraft_command(command)
+            if not is_valid:
+                log_audit_event(
+                    "__api_key__",
+                    "server.command.rejected",
+                    {"original_command": sanitize_string(command[:100]), "source": "websocket"},
+                )
+                socketio.emit("command_error", {"message": "Invalid command format"}, room=request.sid)
+                return
+            command = sanitized_command
+
+        log_audit_event("__api_key__", "server.command", {"command": sanitize_string(command[:100])})
 
         # Check if user has permission (api_key already validated in connect)
         # Execute command via RCON
         try:
-            stdout, stderr, code = run_script("rcon-client.sh", "command", command)
+            stdout, stderr, code = run_rcon_command(command)
             if code == 0:
                 socketio.emit(
                     "command_response", {"command": command, "response": stdout, "success": True}, room=request.sid
@@ -3805,6 +3981,26 @@ def handle_execute_command(data):
     warnings.warn("Flask-SocketIO not available. WebSocket support disabled.", stacklevel=1)
 
 
+def start_event_capture():
+    """Begin following the server log as soon as the API starts.
+
+    Capture must not wait for a browser: events that happen while the dashboard
+    is closed are exactly the ones worth recording. Tests skip this so no
+    background thread or docker call escapes into the suite.
+    """
+    if not (SOCKETIO_AVAILABLE and socketio and EVENTS_AVAILABLE):
+        return False
+
+    bus = game_events.get_bus()
+    bus.set_error_logger(app.logger.error)
+    # Without this, the last few events on a quiet server stay buffered until
+    # something else happens.
+    bus.start_periodic_flush()
+
+    _ensure_log_reader()
+    return True
+
+
 if __name__ == "__main__":
     if not API_ENABLED:
         print("API is disabled in configuration")
@@ -3813,6 +4009,8 @@ def handle_execute_command(data):
     print(f"Starting Minecraft Server API on {API_HOST}:{API_PORT}")
     if SOCKETIO_AVAILABLE and socketio:
         print("WebSocket support enabled")
+        if start_event_capture():
+            print("Game event capture enabled")
         socketio.run(app, host=API_HOST, port=API_PORT, debug=False)
     else:
         print("WebSocket support disabled (Flask-SocketIO not available)")
diff --git a/docs/EVENT_BUS.md b/docs/EVENT_BUS.md
new file mode 100644
index 0000000..2de2732
--- /dev/null
+++ b/docs/EVENT_BUS.md
@@ -0,0 +1,167 @@
+# Game Event Bus
+
+The event bus turns the Minecraft server's log into typed, persisted events that
+features can subscribe to. It is the foundation the gameplay features in
+[FAMILY_SERVER_ROADMAP.md](FAMILY_SERVER_ROADMAP.md) are built on.
+
+## Why it exists
+
+The server log is the only real-time signal Minecraft produces. Before the bus,
+the API followed that log purely to stream raw text to the browser, so anything
+reactive had to re-parse raw lines for itself, and nothing was captured at all
+unless somebody had the dashboard open.
+
+The bus changes three things:
+
+- The follower runs from API startup, not from the first browser connection.
+  Events that happen while nobody is watching are exactly the ones worth keeping.
+- Lines are parsed once into typed events and recorded once, so counts do not
+  drift. `scripts/player-stats-tracker.sh` re-reads the whole log on every run
+  and adds to the previous totals, which inflates its numbers.
+- Anything can subscribe. A new feature is a handler function, not another log
+  parser.
+
+## Event types
+
+| Type | Fired when | Extra fields in `data` |
+| --- | --- | --- |
+| `chat` | A player sends a chat message | `message` |
+| `connect` | A player's client authenticates and connects | `address` |
+| `join` | A player enters the world | |
+| `leave` | A player disconnects | |
+| `death` | A player dies | `cause`, `message` |
+| `advancement` | An advancement, challenge or goal is earned | `advancement` |
+| `command` | A player runs a slash command | `command` |
+| `server_ready` | The server finishes starting | `startup_time` |
+| `server_stopping` | The server begins shutting down | |
+
+Every event also carries `log_time`, the `HH:MM:SS` the server itself printed.
+
+A player session logs two lines, the `connect` carrying the network address and
+the `join` a moment later. They are separate types so that counting joins does
+not double-count every session. Use `join` for "who arrived"; `connect` exists
+so a future "a new device connected" alert has the address it needs.
+
+### Event shape
+
+```json
+{
+  "type": "death",
+  "timestamp": "2026-09-19T21:14:07.418233+00:00",
+  "player": "Silas",
+  "data": {
+    "log_time": "16:14:07",
+    "cause": "was slain by Zombie",
+    "message": "Silas was slain by Zombie"
+  },
+  "raw": "[16:14:07] [Server thread/INFO]: Silas was slain by Zombie"
+}
+```
+
+`timestamp` is when the line was ingested, in UTC. Log lines carry only a time
+of day with no date, so building a full timestamp from them breaks across
+midnight. The server's own reading is kept in `data.log_time` instead.
+
+## Death detection
+
+Vanilla death messages have no marker of their own. They are bare sentences
+beginning with a player name, so they are matched against the opening phrases of
+Minecraft's `death.attack.*` strings, listed in `_DEATH_PHRASES` in
+[`api/events.py`](../api/events.py).
+
+Chat is matched first and deliberately. A player can type a sentence that looks
+exactly like a death message, and the `<name>` form has to win before any other
+pattern is tried. A message the phrase list does not cover is simply not treated
+as a death, which is the safe direction to fail. Add phrases to that tuple as
+new ones appear.
+
+## Writing a handler
+
+```python
+from api import events, rcon
+
+def announce_death(event):
+    if event.type != events.EVENT_DEATH:
+        return
+    rcon.get_client().command(f'tellraw @a {{"text":"RIP {event.player}","color":"red"}}')
+
+events.get_bus().subscribe(announce_death)
+```
+
+Handlers run synchronously, in registration order, on the follower thread.
+Two consequences:
+
+- Keep them fast. Anything slow, such as an API call to a language model, should
+  hand off to a queue or a background task rather than blocking the reader.
+- A handler that raises is logged and skipped. One broken feature cannot silence
+  the bus or stop the others.
+
+## Storage
+
+Events are appended to `data/events/YYYY-MM-DD.jsonl`, one JSON object per line.
+
+Writes are buffered, flushing after 20 events or 30 seconds, whichever comes
+first. The Pi boots from an SD card with finite write endurance, so a continuous
+stream of one-line appends is batched rather than written per event. Files older
+than 30 days are pruned automatically on the first flush of each day.
+
+`data/` is gitignored and lives only on the Pi.
+
+## REST API
+
+Both endpoints require the `logs.view` permission.
+
+```
+GET /api/events?limit=100&type=death&player=Silas
+GET /api/events/types
+```
+
+`limit` defaults to 100 and is capped at 500. `type` must be one of the types
+above; anything else returns 400 with the valid list. Events come back newest
+first.
+
+```bash
+curl -H "X-API-Key: $API_KEY" "http://localhost:8080/api/events?type=death&limit=10"
+```
+
+## WebSocket
+
+Clients already connected to the log stream also receive a `game_event` message
+for every parsed event, carrying the same JSON shape as the REST endpoint.
+
+```javascript
+socket.on('game_event', (event) => {
+  if (event.type === 'death') {
+    console.log(`${event.player}: ${event.data.cause}`);
+  }
+});
+```
+
+Raw `logs` messages continue unchanged, so the existing Logs page is unaffected.
+
+## Operational notes
+
+- **The follower attaches with `--tail 0`, on purpose.** Every line it reads is
+  persisted as an event, so replaying a backlog would record the same deaths and
+  advancements again on every API restart and every re-attach, and the counts
+  would climb with each one. New clients still get their scrollback: the
+  WebSocket sends it separately on connect, and that path does not touch the bus.
+  The trade is that events occurring while the API is down are not captured,
+  which is far better than recording some of them repeatedly.
+- The follower re-attaches by itself. `docker logs -f` exits when the container
+  stops, and the reader waits two seconds and tries again, so a server restart
+  resumes streaming without intervention.
+- If Docker is not installed the reader reports it once and stops, rather than
+  retrying forever.
+- `stop_log_reader()` in [`api/server.py`](../api/server.py) brings the follower
+  down deliberately. It kills the `docker logs` process to break the blocking
+  read, because a quiet server would otherwise leave the reader parked in it.
+- Buffered events are also flushed on a timer, so the last few on a quiet server
+  are not left in memory waiting for the next one.
+
+## Related
+
+- [`api/events.py`](../api/events.py) — the parser, bus and storage
+- [`api/rcon.py`](../api/rcon.py) — how handlers talk back to the server
+- [RCON.md](RCON.md) — RCON setup
+- [FAMILY_SERVER_ROADMAP.md](FAMILY_SERVER_ROADMAP.md) — what this is for
diff --git a/docs/FAMILY_SERVER_ROADMAP.md b/docs/FAMILY_SERVER_ROADMAP.md
new file mode 100644
index 0000000..1e6700a
--- /dev/null
+++ b/docs/FAMILY_SERVER_ROADMAP.md
@@ -0,0 +1,466 @@
+# Family Server Roadmap — The Wow Factor
+
+Audience: Matt, planning what to build next on the Pi 5 server for Jonah and Silas.
+
+Every other roadmap in this repo (`ROADMAP.md`, `TASKS.md`,
+`MINECRAFT_GAMEPLAY_ENHANCEMENTS.md`) plans the *server management product*: more
+endpoints, more admin pages, more ops polish. This one plans the *experience the
+kids actually see*. It assumes the management layer is good enough already and
+asks a different question: what could this server do that no other Minecraft
+server does?
+
+Ratings used below:
+
+- **Green** — works on the Pi 5, on vanilla 1.20.4, with what's already in the repo.
+- **Yellow** — works, but with a real constraint (extra RAM, extra hardware, a
+  server-type change, or the work belongs on the Mac rather than the Pi).
+- **Red** — don't, or not yet.
+
+---
+
+## Phase 0 — Foundations
+
+Four pieces of plumbing. Almost everything in this document depends on at least
+one of them, and several ideas collapse from "hard" to "an afternoon" once these
+exist. Build these first.
+
+### F1. A real RCON client inside the API — Green — **done**
+
+Today every command shells out: `api/server.py` → `subprocess` →
+`scripts/rcon-client.sh` → a brand new TCP connection and auth handshake per
+command. That is roughly 50–100ms of overhead per command, and the pure-Python
+fallback in `scripts/rcon-client.sh` has two defects worth knowing about before
+you build on it: it does a single `sock.recv(4096)` with no length-prefix
+framing, so long responses are truncated or split, and it treats any 4+ byte
+reply as a successful auth rather than checking for a request id of `-1`.
+
+Shipped as [`api/rcon.py`](../api/rcon.py): a persistent, pooled, properly
+framed client with `command()` and a `commands(list)` batch call, multi-packet
+reassembly and automatic reconnection. `api/server.py` reaches it through
+`run_rcon_command()`, which falls back to the shell script when RCON is
+unconfigured or unreachable. `scripts/rcon-client.sh` is unchanged for CLI use.
+
+This is what makes the pixel-art placer, the event responders and the control
+room viable instead of sluggish.
+
+### F2. The Event Bus — Green — **done**
+
+This is the backbone. `_log_reader` at `api/server.py:3657` already follows
+`docker logs -f` in a single well-written thread and fans raw lines out over
+Socket.IO. Two changes turn it into an event bus:
+
+1. **Run it always.** It currently starts only when a browser subscribes and
+   returns as soon as `active_log_streams` is empty. Start it at app boot and keep
+   it running, so events are captured when nobody is watching the dashboard.
+2. **Parse and persist.** Match each line into a typed event and append to
+   `data/events/YYYY-MM-DD.jsonl`, then emit both the raw line (back-compat with
+   the Logs page) and the typed event on a new Socket.IO channel.
+
+Shipped event types: `chat`, `connect`, `join`, `leave`, `death`,
+`advancement`, `command`, `server_ready`, `server_stopping`. Vanilla log lines
+carry all of these.
+
+Shipped as [`api/events.py`](../api/events.py) with the always-on follower in
+`api/server.py`, the `GET /api/events` endpoints and a `game_event` WebSocket
+message. Features subscribe with a handler function; see
+[EVENT_BUS.md](EVENT_BUS.md). An "on death, do X" hook is what makes a dozen
+features below into ten lines of code each.
+
+Still outstanding: the stats rewrite below.
+
+**While you're in there:** `scripts/player-stats-tracker.sh` re-parses the entire
+log file on every invocation and adds to the existing counts, so every run
+inflates the numbers. It also only matches three crude patterns. Vanilla already
+writes authoritative per-player data to `data/<world>/stats/<uuid>.json` and
+`data/<world>/advancements/<uuid>.json`. Read those files instead. They have
+every statistic Minecraft tracks: blocks mined by type, distance walked, damage
+taken, time played, every death cause.
+
+### F3. A datapack pipeline — Green
+
+Datapacks are the cheat code for this whole document. They are plain JSON in
+`data/<world>/datapacks/`, they work on **vanilla**, they hot-reload with
+`/reload`, and they give you custom advancements, recipes, loot tables,
+predicates and functions without a single plugin.
+
+Build `scripts/datapack-manager.sh` (`create`, `list`, `enable`, `disable`,
+`validate`, `reload`) — `docs/MINECRAFT_ENHANCEMENTS.md` already proposes it.
+Keep the family datapack in git so every change is revertible.
+
+### F4. Resource pack hosting — Green
+
+Set `resource-pack=` and `resource-pack-sha1=` in `server.properties` and host
+the zip from Cloudflare R2, which is already wired up for backups in
+`scripts/cloud-backup-r2.sh`. Clients download it automatically on join. This
+unlocks custom sounds, custom music discs, custom item textures and custom
+fonts, again on vanilla.
+
+---
+
+## Tier 1 — Highest wow per hour of work
+
+### W1. The Oracle — a Claude-powered companion in chat — Green
+
+Kids type in chat, something answers. Event bus catches `chat`, the API calls
+the Claude API, the reply goes back via `tellraw` with colored JSON text.
+
+Use `claude-haiku-4-5` for banter (fast and cheap) and `claude-sonnet-5` for
+anything that needs to generate structure, like quests. Round-trip lands around
+2–4 seconds, which reads as "the wizard is thinking" rather than as lag.
+
+Give it a personality and a job: it knows the server's history from the event
+log, it remembers what each kid was building last week, it hands out a daily
+quest, it answers "how do I make a beacon" without either of them alt-tabbing to
+a wiki.
+
+**Guardrails, because this is aimed at children.** Pin a system prompt that
+scopes it to Minecraft and kid-appropriate content, allowlist the player names
+that may invoke it, rate limit per player per minute, cap output tokens so a
+reply always fits a chat line, log every exchange through the existing
+`log_audit_event()`, and put a kill switch on the dashboard.
+
+### W2. The Invention Forge — describe an item, the server creates it — Green/Yellow
+
+This is the one nobody else has. A kid types:
+
+```
+!invent a pickaxe that leaves a trail of flowers
+```
+
+Claude generates the datapack JSON — an advancement, a recipe, a loot table, a
+function — the API validates it against a strict schema, writes it into
+`data/<world>/datapacks/family/`, runs `/reload`, and **the item exists in the
+game seconds later.**
+
+The reason this is Yellow rather than Green is the validator, and it is worth
+doing properly: generated `function` files can contain arbitrary commands, so
+run every generated command through an allowlist before it ever touches the
+disk. `api/security.py` already has `ALLOWED_MINECRAFT_COMMANDS` and
+`BLOCKED_COMMAND_PREFIXES` for exactly this shape of problem — extend that
+pattern. Keep the family datapack in git so a bad generation is one `git revert`
+away, and add a `/undo-invention` that disables the last pack and reloads.
+
+A ten-year-old inventing a working item by describing it is the single most
+"knock your socks off" thing on this list.
+
+### W3. Obituaries and the Hall of Deaths — Green
+
+Death event fires. Claude writes a one-line epitaph in the voice of a Victorian
+newspaper obituary. It gets `tellraw`'d to everyone online, and it lands on a
+Hall of Deaths page on the dashboard with a leaderboard of most creative demises.
+
+Nearly free once F2 and W1 exist, and it will be the thing they quote at dinner.
+
+### W4. The Family Gazette — Green
+
+A weekly newspaper, auto-generated every Sunday morning. Player stats, the
+funniest deaths of the week, whose base grew the most, who mined the most
+diamonds, what's coming up next week, a photo of the map.
+
+Delivered two ways: as an HTML page on the dashboard, and **in-game as an actual
+written book** placed in each player's inventory via `/give` with book NBT.
+Receiving a physical newspaper in your inventory on a Sunday is a ritual.
+
+**Version caution:** 1.20.4 uses the old `written_book{pages:[...]}` NBT syntax.
+1.20.5 replaced item NBT with components and this syntax breaks. Put all book
+construction behind one function so a Minecraft upgrade is a one-file change.
+
+### W5. Bedtime Mode — Green
+
+Scheduled shutdown that doesn't start a fight. Thirty minutes out, a bossbar
+countdown appears. At ten, five and one minute, a title flashes. Then autosave,
+a graceful stop, and a "goodnight" message.
+
+Add a "five more minutes" button on the dashboard, and a Siri Shortcut for it.
+Build on `scripts/command-scheduler.py` (it already does daily schedules and
+conditional execution) and `scripts/announcement-manager.sh` (it already does
+say, title and actionbar).
+
+This is the most useful thing on the list for the adults in the house.
+
+### W6. Siri, Shortcuts and HomeKit — Green
+
+The cheapest big win in the document. The API already exists, already takes an
+`X-API-Key` header, and already has start/stop/status/players endpoints. An iOS
+Shortcut that makes an HTTP request is all that stands between you and:
+
+- "Hey Siri, start Minecraft."
+- "Hey Siri, is Jonah online?"
+- "Hey Siri, Minecraft bedtime." (triggers W5)
+- A Home Screen widget showing who's online.
+
+Zero new server code. Ship a documented Shortcuts bundle and a few convenience
+endpoints. Hours of work, not days. The `homekit-automator` repo can take it
+further and expose the server as a real HomeKit accessory.
+
+---
+
+## Tier 2 — The house and the game talk to each other
+
+This is the category that is genuinely rare, and it is rare only because most
+people don't run their server on a Pi in their own house with a smart-home stack
+next to it. You do.
+
+### H1. Minecraft controls the real house — Green
+
+A lever in the game turns on the real lamp in the bedroom.
+
+Simplest implementation: a command block or sign in-game runs a chat command the
+event bus recognizes, the API fires a webhook at `homehub` or HomeKit, the light
+changes. A slightly cleaner version uses a datapack tick function writing to a
+scoreboard the API polls.
+
+Build a small "control panel" room in the world: one lever per light, a button
+for the kitchen speaker, a redstone lamp that mirrors whether the front door is
+locked. For a kid, this crosses a line they didn't know could be crossed.
+
+### H2. The Control Room — Green
+
+The inverse. An in-game room whose signs and scoreboard sidebar show real-world
+data, refreshed by RCON once a minute: house temperature, tomorrow's forecast
+(from the `weather-app` repo), chores due today (from the `family` repo), the
+server's own TPS and the Pi's CPU temperature (already collected by
+`scripts/analytics-collector.sh`).
+
+### H3. Real-world sync — Green, with a caveat
+
+- Real local weather drives in-game weather.
+- Real sunrise and sunset drive in-game time.
+- Real moon phase matches the in-game moon.
+- Meteor shower tonight (the `sky` and `telescope` repos know) means an automatic
+  firework show at the real peak time.
+
+**Caveat:** hard-syncing time breaks sleeping in beds and scrambles mob-spawn
+expectations, which kids notice fast. Make it a per-world toggle, sync on a slow
+cadence, and consider syncing weather and moon only. The meteor shower fireworks
+are the best part anyway and carry no downside.
+
+### H4. Status hardware on the Pi — Yellow
+
+Physical presence for an invisible thing:
+
+- An LED strip whose color is server health, that pulses when someone joins, and
+  whose brightness follows in-game day and night.
+- A small e-ink panel on the shelf showing who is online right now.
+- A big arcade button that starts the server.
+
+**Two Pi 5 specific gotchas, because both will waste your evening otherwise.**
+The Pi 5 moved GPIO behind the RP1 I/O controller, so `RPi.GPIO` does not work —
+use `gpiozero` with the `lgpio` pin factory. And the classic `rpi_ws281x`
+NeoPixel driver relies on PWM/DMA behavior the Pi 5 doesn't provide, so drive
+addressable pixels over SPI or hang them off a cheap microcontroller over USB.
+E-ink over SPI is fine.
+
+This is also the repo's one completely untouched area: nothing in the project
+drives anything physical today.
+
+---
+
+## Tier 3 — Memory and spectacle
+
+### M1. Build time-lapse — Green, on the Mac
+
+Render the same region from each nightly backup, stitch the frames with ffmpeg.
+"Watch your base grow over a year" in ninety seconds.
+
+Run it on the Mac against `backups/`, not on the Pi. There is no reason to spend
+the Pi's thermal budget on rendering, and the backups are already there.
+
+### M2. A 3D web map — Yellow on the Pi, Green on the Mac
+
+BlueMap gives a genuinely beautiful browsable 3D map. The initial full render of
+a mature world is CPU-hours and will run the Pi hot.
+
+The sharp version: run BlueMap's standalone CLI **on the Mac** against last
+night's backup, publish the static output to Cloudflare Pages or R2, and link it
+from the dashboard. Free hosting, no server load, updates every morning, and it
+works against a vanilla world directory with no plugin required.
+
+### M3. 3D-print your build — Green, and a showstopper
+
+Select a region, export it with Mineways or jmc2obj on the Mac, convert to STL,
+hand it to the `Printer` repo and the Anycubic.
+
+A child holding a physical model of the house they built in Minecraft is an
+all-timer. This is a weekend project with an enormous payoff, and it connects
+two repos you already maintain.
+
+### M4. World rewind and the Museum — Yellow / Green
+
+**Rewind:** boot a backup from any date as a second, read-only server on port
+25566 and walk around the past. Two JVMs on an 8GB Pi is tight and on a 4GB Pi
+it won't fit, so run the rewind instance on the Mac and have the dashboard hand
+out the connection address.
+
+**Museum:** a permanent showcase world where the best builds get copied in with
+structure blocks, with a sign giving the builder and the date. Green, and
+`scripts/world-manager.sh` already has `create-template` and `from-template` to
+build on.
+
+### M5. The Time Capsule — Green
+
+Write a book in-game, seal it in the capsule, and the server delivers it back to
+you on a real date a year from now.
+
+No datapack needed. Store the book's contents as JSON, schedule the `/give`
+through the existing command scheduler, deliver on the day. Technically the
+smallest item in this document, and probably the one that matters most in five
+years.
+
+---
+
+## Tier 4 — Structured play
+
+### P1. Treasure hunts — Green
+
+Claude writes a chain of riddle clues. A script picks real coordinates, places a
+loot chest at each, and delivers the first clue as a book. Each clue's answer is
+the next location. Difficulty tuned per kid.
+
+### P2. A family advancement tree — Green
+
+A custom datapack advancement tab with a custom icon, appearing in the real
+advancements screen alongside the vanilla ones:
+
+- "Sibling Rivalry" — survive three nights with your brother
+- "Neighbors" — build within 50 blocks of Dad's house
+- "First Diamond"
+- "Ten Thousand Blocks"
+- "Night Shift" — play past 9pm (they will find this hilarious)
+
+Vanilla triggers cover most of it; location checks need a tick function. Because
+it shows up in the game's own UI it reads as official rather than bolted on.
+
+### P3. Sibling co-op goals — Green
+
+Shared, server-wide objectives on a scoreboard that only complete if both
+brothers contribute. Completing one unlocks a reward for both. Escalate toward
+real-world prizes: finish the tier, get pizza night. Cooperation with a payout
+beats competition for siblings.
+
+### P4. The Expedition World — Green
+
+Every week a temporary world from a template: skyblock, parkour, hardcore, the
+floor is lava, a Claude-generated challenge. Results go in the Gazette, the best
+build gets copied into the Museum, then the world resets.
+
+`scripts/world-manager.sh` already supports templates, so this is mostly
+scheduling and ceremony.
+
+### P5. Raid Night — Green
+
+Scripted boss encounters via datapack functions: a summoned mob with custom
+attributes, equipment and a name, a bossbar tracking its health, phases that
+trigger at health thresholds, and loot worth the fight. Scheduled for Friday
+evenings and announced in the Gazette all week.
+
+### P6. Birthdays and holidays — Green
+
+Automatic firework shows, cake, custom titles and themed loot on birthdays and
+holidays. `scripts/command-scheduler.py` already handles date-based scheduling,
+so this is content, not engineering.
+
+---
+
+## Tier 5 — Make it theirs
+
+### T1. A resource pack the kids made — Yellow, enormous payoff
+
+Record a sound on an iPad, it becomes a music disc they can play on a jukebox in
+the game. Draw a 16x16 sprite in a web pixel editor on the dashboard, it becomes
+an item texture via CustomModelData. The pack auto-builds, uploads to R2 and
+serves through F4.
+
+Real effort, but the result is their drawings and their voices inside Minecraft.
+
+### T2. Pixel-art uploader — Yellow, Green with one trick
+
+Upload a PNG, get it built in blocks in-game. The naive approach of one
+`setblock` per pixel is far too slow: a 128x128 image is 16,384 commands.
+
+Two ways to fix it. Run-length encode each row into `/fill` commands, which
+typically cuts command count by 10–50x on real images. Or generate an `.nbt`
+structure file directly and place it with a structure block, which is one
+command total. Cap the input at 128x128 either way.
+
+### T3. Mail — Green
+
+Send a letter from the dashboard, or from a phone, and it arrives as a written
+book in the player's inventory. A note from Mom in your inventory when you log
+in. Trivial once F1 and the Gazette's book builder exist.
+
+### T4. NFC portal cards — Yellow
+
+A PN532 reader on the Pi, or the tooling already in the `amiibo` and `flipper`
+repos. Tap a physical card and the game responds: teleport to a saved location,
+switch worlds, grant a kit, trigger an event.
+
+Physical objects that do things in a video game is a category of magic that lands
+extremely well at this age.
+
+---
+
+## Tier 6 — Reach
+
+### R1. Geyser and Floodgate for cross-play — Yellow
+
+Let them join from an iPad, a phone or a Switch. Run Geyser standalone as a
+separate container proxying into the vanilla server.
+
+Cost is an additional JVM, roughly 300–500MB. That's Yellow on an 8GB Pi and Red
+on a 4GB one. Tablets and phones connect easily; a Switch needs DNS redirection
+to work at all.
+
+**If the kids have iPads, this may be the highest-value item in the entire
+document,** because it changes when and where they can play at all. Weigh it
+first against everything above.
+
+### R2. Simple Voice Chat — Yellow, deprioritize
+
+Needs Fabric or Paper server-side and a client mod installed for every player.
+For two brothers in the same house who can simply shout, the value is low. Park
+it.
+
+### R3. Push notifications and a weekly digest — Green
+
+"Silas just got his first diamond." Event bus plus ntfy, Pushover or APNs. Cheap,
+and it keeps the adults connected to what's happening without hovering.
+
+---
+
+## Suggested sequence
+
+| Order | Items | Why here |
+|---|---|---|
+| 1 | F1, F2 | Nothing good happens without the RCON client and the event bus |
+| 2 | W6, W5 | Days of work, immediate daily payoff, no new infrastructure |
+| 3 | W3, W1 | First "whoa" moments; W3 proves the event bus end to end |
+| 4 | F3, P2 | Datapack pipeline plus family advancements, visible in-game |
+| 5 | M2, M1 | Map and time-lapse, rendered on the Mac, zero Pi cost |
+| 6 | W4, T3, M5 | The book pipeline: Gazette, mail, time capsule all share it |
+| 7 | W2 | The Invention Forge, once the datapack validator can be trusted |
+| 8 | H1, H2, H3 | House and game wired together |
+| 9 | R1 | Geyser, if tablets matter — consider pulling this much earlier |
+| 10 | M3, T1, H4, T4 | The big physical projects |
+
+---
+
+## Known issues to fix before any of this ships
+
+These are real and they sit in the path of the work above.
+
+- **`api/server.py:723` hardcodes `"role": "admin"` for every registered user**,
+  despite a comment claiming otherwise. Open registration currently mints
+  administrators. Fix before anything is exposed beyond the LAN.
+- **API keys bypass permissions entirely** — `has_permission` returns `True`
+  unconditionally for API-key callers (`api/server.py:605`). The Socket.IO stream
+  accepts only API keys, so any key living in a kid's browser is a full admin
+  credential. Tighten before handing out keys.
+- **`scripts/player-stats-tracker.sh` inflates counts** on every run by
+  re-parsing the whole log and adding to existing totals. Superseded by F2.
+- **`apply_server_preset()` (`api/server.py:2673`) has no route decorator** and is
+  unreachable. Either wire it up or delete it.
+- **1.20.4 to 1.20.5 changes item NBT to components.** Every `/give` with NBT in
+  this document breaks on that upgrade. Isolate item construction in one module.
+- **SD-card write endurance.** The event bus appends JSONL continuously. Put
+  `data/events/` on an SSD if there is one, batch the flushes, and rotate hard.
diff --git a/docs/INDEX.md b/docs/INDEX.md
index 5e09746..7a1feee 100644
--- a/docs/INDEX.md
+++ b/docs/INDEX.md
@@ -56,6 +56,7 @@ New here? Read `INSTALL.md`, then keep `QUICK_REFERENCE.md` open.
 | [PLUGIN_MANAGEMENT.md](PLUGIN_MANAGEMENT.md) | Installing and configuring Bukkit/Spigot/Paper plugins |
 | [MOD_SUPPORT.md](MOD_SUPPORT.md) | Mod loader detection and mod pack installation |
 | [RCON.md](RCON.md) | Remote console setup and usage |
+| [EVENT_BUS.md](EVENT_BUS.md) | Typed game events parsed from the server log |
 | [ANALYTICS.md](ANALYTICS.md) | Player and server analytics collection and reports |
 
 ---
@@ -98,6 +99,7 @@ New here? Read `INSTALL.md`, then keep `QUICK_REFERENCE.md` open.
 | [CURSOR_CONFIGURATION.md](CURSOR_CONFIGURATION.md) | Cursor IDE setup |
 | [ROADMAP.md](ROADMAP.md) | Planned work, by phase |
 | [TASKS.md](TASKS.md) | Detailed task backlog |
+| [FAMILY_SERVER_ROADMAP.md](FAMILY_SERVER_ROADMAP.md) | Wow-factor feature roadmap for the family server |
 
 AI assistants read [`../AGENTS.md`](../AGENTS.md) — the single source of truth for
 conventions. See also [`../CONTRIBUTING.md`](../CONTRIBUTING.md).
diff --git a/tests/api/test_events.py b/tests/api/test_events.py
new file mode 100644
index 0000000..81536b0
--- /dev/null
+++ b/tests/api/test_events.py
@@ -0,0 +1,530 @@
+"""Tests for the game event bus (api/events.py)."""
+
+import json
+import sys
+import threading
+import time
+from pathlib import Path
+
+import pytest
+
+PROJECT_ROOT = Path(__file__).parent.parent.parent
+sys.path.insert(0, str(PROJECT_ROOT))
+
+from api.events import (  # noqa: E402
+    EVENT_ADVANCEMENT,
+    EVENT_CHAT,
+    EVENT_COMMAND,
+    EVENT_CONNECT,
+    EVENT_DEATH,
+    EVENT_JOIN,
+    EVENT_LEAVE,
+    EVENT_SERVER_READY,
+    EVENT_SERVER_STOPPING,
+    EventBus,
+    GameEvent,
+    get_bus,
+    parse_line,
+    reset_bus,
+)
+
+
+def line(message, thread="Server thread/INFO", time="16:04:23"):
+    """Build a realistic server log line around a message."""
+    return f"[{time}] [{thread}]: {message}"
+
+
+@pytest.fixture
+def bus(tmp_path):
+    """A bus writing to a temp directory, flushing on every event."""
+    return EventBus(events_dir=tmp_path / "events", flush_every_events=1)
+
+
+@pytest.fixture(autouse=True)
+def clean_shared_bus():
+    reset_bus()
+    yield
+    reset_bus()
+
+
+@pytest.mark.unit
+class TestParsingPlayerEvents:
+    def test_chat_message(self):
+        event = parse_line(line("<Jonah> hello silas"))
+        assert event.type == EVENT_CHAT
+        assert event.player == "Jonah"
+        assert event.data["message"] == "hello silas"
+
+    def test_chat_preserves_inner_punctuation(self):
+        event = parse_line(line("<Silas> wait... what?! >:)"))
+        assert event.data["message"] == "wait... what?! >:)"
+
+    def test_empty_chat_message_still_parses(self):
+        event = parse_line(line("<Jonah> "))
+        assert event.type == EVENT_CHAT
+        assert event.data["message"] == ""
+
+    def test_join(self):
+        event = parse_line(line("Jonah joined the game"))
+        assert (event.type, event.player) == (EVENT_JOIN, "Jonah")
+
+    def test_leave(self):
+        event = parse_line(line("Silas left the game"))
+        assert (event.type, event.player) == (EVENT_LEAVE, "Silas")
+
+    def test_login_line_is_a_connect_not_a_join(self):
+        """One session logs two lines; counting joins must not double-count."""
+        event = parse_line(line("Jonah[/192.168.1.40:54321] logged in with entity id 214 at (1.5, 64.0, -9.5)"))
+        assert event.type == EVENT_CONNECT
+        assert event.player == "Jonah"
+        assert event.data["address"] == "192.168.1.40:54321"
+
+    def test_a_full_session_produces_exactly_one_join(self):
+        raws = [
+            line("Jonah[/192.168.1.40:54321] logged in with entity id 214 at (1.5, 64.0, -9.5)"),
+            line("Jonah joined the game"),
+        ]
+        types = [parse_line(raw).type for raw in raws]
+        assert types.count(EVENT_JOIN) == 1
+        assert types.count(EVENT_CONNECT) == 1
+
+    def test_issued_command(self):
+        event = parse_line(line("Jonah issued server command: /gamemode creative"))
+        assert event.type == EVENT_COMMAND
+        assert event.data["command"] == "/gamemode creative"
+
+    def test_advancement(self):
+        event = parse_line(line("Silas has made the advancement [Stone Age]"))
+        assert event.type == EVENT_ADVANCEMENT
+        assert event.data["advancement"] == "Stone Age"
+
+    def test_challenge_and_goal_are_advancements_too(self):
+        challenge = parse_line(line("Jonah has completed the challenge [Adventuring Time]"))
+        goal = parse_line(line("Silas has reached the goal [Sky's the Limit]"))
+        assert challenge.type == EVENT_ADVANCEMENT
+        assert goal.data["advancement"] == "Sky's the Limit"
+
+
+@pytest.mark.unit
+class TestParsingDeaths:
+    @pytest.mark.parametrize(
+        "message,player",
+        [
+            ("Jonah was slain by Zombie", "Jonah"),
+            ("Silas fell from a high place", "Silas"),
+            ("Jonah drowned", "Jonah"),
+            ("Silas blew up", "Silas"),
+            ("Jonah was blown up by Creeper", "Jonah"),
+            ("Silas tried to swim in lava", "Silas"),
+            ("Jonah was shot by Skeleton", "Jonah"),
+            ("Silas burned to death", "Silas"),
+            ("Jonah starved to death", "Jonah"),
+            ("Silas suffocated in a wall", "Silas"),
+            ("Jonah withered away", "Jonah"),
+            ("Silas froze to death", "Silas"),
+            ("Jonah hit the ground too hard", "Jonah"),
+            ("Silas fell out of the world", "Silas"),
+            ("Jonah was killed by magic", "Jonah"),
+            ("Silas was pricked to death", "Silas"),
+            ("Jonah was squashed by a falling anvil", "Jonah"),
+            ("Silas went up in flames", "Silas"),
+            ("Jonah was struck by lightning", "Jonah"),
+            ("Silas was impaled on a stalagmite", "Silas"),
+            ("Jonah was poked to death by a sweet berry bush", "Jonah"),
+            ("Silas discovered the floor was lava", "Silas"),
+            ("Jonah was obliterated by a sonically-charged shriek", "Jonah"),
+            ("Silas fell off scaffolding", "Silas"),
+        ],
+    )
+    def test_vanilla_death_messages(self, message, player):
+        event = parse_line(line(message))
+        assert event is not None, f"not recognised as a death: {message}"
+        assert event.type == EVENT_DEATH
+        assert event.player == player
+        assert event.data["message"] == message
+
+    def test_death_cause_is_captured(self):
+        event = parse_line(line("Jonah was slain by Zombie using Iron Sword"))
+        assert event.data["cause"] == "was slain by Zombie using Iron Sword"
+
+    def test_chat_that_looks_like_a_death_is_still_chat(self):
+        """A kid typing a fake death message must not fake a death event."""
+        event = parse_line(line("<Silas> Jonah was slain by Zombie"))
+        assert event.type == EVENT_CHAT
+        assert event.player == "Silas"
+
+    def test_ordinary_sentence_is_not_a_death(self):
+        assert parse_line(line("Jonah is building something")) is None
+
+
+@pytest.mark.unit
+class TestParsingServerEvents:
+    def test_server_ready(self):
+        event = parse_line(line('Done (21.174s)! For help, type "help"'))
+        assert event.type == EVENT_SERVER_READY
+        assert event.data["startup_time"] == "21.174s"
+        assert event.player is None
+
+    def test_server_stopping(self):
+        assert parse_line(line("Stopping the server")).type == EVENT_SERVER_STOPPING
+
+
+@pytest.mark.unit
+class TestParsingNonEvents:
+    @pytest.mark.parametrize(
+        "raw",
+        [
+            "",
+            "   ",
+            "not a log line at all",
+            "[16:04:23] [Server thread/INFO]: ",
+            "[16:04:23] [Server thread/INFO]: Preparing spawn area: 82%",
+            "[16:04:23] [Server thread/WARN]: Can't keep up! Is the server overloaded?",
+            "[16:04:23] [Worker-Main-3/INFO]: Preparing spawn area: 4%",
+        ],
+    )
+    def test_uninteresting_lines_produce_nothing(self, raw):
+        assert parse_line(raw) is None
+
+    def test_malformed_line_does_not_raise(self):
+        assert parse_line("[[[]]]") is None
+
+    def test_other_thread_names_still_parse(self):
+        """Paper logs chat from an async thread, not the main server thread."""
+        event = parse_line(line("<Jonah> hi", thread="Async Chat Thread-#0/INFO"))
+        assert event.type == EVENT_CHAT
+
+    def test_log_time_is_kept_separate_from_ingest_time(self):
+        event = parse_line(line("Jonah joined the game", time="23:59:58"))
+        assert event.data["log_time"] == "23:59:58"
+        assert event.timestamp.endswith("+00:00")
+
+
+@pytest.mark.unit
+class TestDispatch:
+    def test_handler_receives_event(self, bus):
+        seen = []
+        bus.subscribe(seen.append)
+        bus.handle_line(line("Jonah joined the game"))
+        assert [event.player for event in seen] == ["Jonah"]
+
+    def test_non_event_lines_are_not_dispatched(self, bus):
+        seen = []
+        bus.subscribe(seen.append)
+        bus.handle_line(line("Preparing spawn area: 82%"))
+        assert seen == []
+
+    def test_multiple_handlers_all_fire(self, bus):
+        first, second = [], []
+        bus.subscribe(first.append)
+        bus.subscribe(second.append)
+        bus.handle_line(line("Silas left the game"))
+        assert len(first) == len(second) == 1
+
+    def test_subscribing_twice_does_not_double_deliver(self, bus):
+        seen = []
+        bus.subscribe(seen.append)
+        bus.subscribe(seen.append)
+        bus.handle_line(line("Jonah joined the game"))
+        assert len(seen) == 1
+
+    def test_unsubscribe_stops_delivery(self, bus):
+        seen = []
+        bus.subscribe(seen.append)
+        bus.unsubscribe(seen.append)
+        bus.handle_line(line("Jonah joined the game"))
+        assert seen == []
+
+    def test_failing_handler_does_not_stop_the_others(self, bus):
+        """One broken feature must not silence the whole bus."""
+        errors = []
+        survived = []
+
+        def broken(_event):
+            raise RuntimeError("handler exploded")
+
+        bus.set_error_logger(errors.append)
+        bus.subscribe(broken)
+        bus.subscribe(survived.append)
+
+        bus.handle_line(line("Jonah joined the game"))
+
+        assert len(survived) == 1
+        assert "handler exploded" in errors[0]
+
+    def test_failing_handler_still_persists_the_event(self, bus):
+        def broken(_event):
+            raise RuntimeError("nope")
+
+        bus.set_error_logger(lambda _message: None)
+        bus.subscribe(broken)
+        bus.handle_line(line("Jonah joined the game"))
+
+        assert len(bus.read()) == 1
+
+
+@pytest.mark.unit
+class TestPersistence:
+    def test_event_is_written_as_jsonl(self, bus, tmp_path):
+        bus.handle_line(line("Jonah joined the game"))
+        files = list((tmp_path / "events").glob("*.jsonl"))
+        assert len(files) == 1
+
+        record = json.loads(files[0].read_text().strip())
+        assert record["type"] == EVENT_JOIN
+        assert record["player"] == "Jonah"
+        assert record["raw"].endswith("Jonah joined the game")
+
+    def test_writes_are_buffered_until_the_threshold(self, tmp_path):
+        """Batching exists to spare the Pi's SD card."""
+        buffered = EventBus(events_dir=tmp_path / "events", flush_every_events=5, flush_every_seconds=3600)
+        for _ in range(4):
+            buffered.handle_line(line("Jonah joined the game"))
+        assert not list((tmp_path / "events").glob("*.jsonl"))
+
+        buffered.handle_line(line("Jonah joined the game"))
+        assert len(list((tmp_path / "events").glob("*.jsonl"))) == 1
+
+    def test_flush_returns_the_count_written(self, tmp_path):
+        buffered = EventBus(events_dir=tmp_path / "events", flush_every_events=100, flush_every_seconds=3600)
+        for _ in range(3):
+            buffered.handle_line(line("Jonah joined the game"))
+        assert buffered.flush() == 3
+        assert buffered.flush() == 0
+
+    def test_read_returns_newest_first(self, bus):
+        bus.handle_line(line("Jonah joined the game"))
+        bus.handle_line(line("Silas joined the game"))
+        assert [record["player"] for record in bus.read()] == ["Silas", "Jonah"]
+
+    def test_read_filters_by_type(self, bus):
+        bus.handle_line(line("Jonah joined the game"))
+        bus.handle_line(line("Jonah was slain by Zombie"))
+        deaths = bus.read(event_type=EVENT_DEATH)
+        assert len(deaths) == 1
+        assert deaths[0]["data"]["cause"].startswith("was slain")
+
+    def test_read_filters_by_player(self, bus):
+        bus.handle_line(line("Jonah joined the game"))
+        bus.handle_line(line("Silas joined the game"))
+        assert len(bus.read(player="Silas")) == 1
+
+    def test_read_respects_the_limit(self, bus):
+        for _ in range(10):
+            bus.handle_line(line("Jonah joined the game"))
+        assert len(bus.read(limit=4)) == 4
+
+    def test_read_flushes_pending_events_first(self, tmp_path):
+        buffered = EventBus(events_dir=tmp_path / "events", flush_every_events=100, flush_every_seconds=3600)
+        buffered.handle_line(line("Jonah joined the game"))
+        assert len(buffered.read()) == 1
+
+    def test_read_is_empty_when_nothing_was_recorded(self, bus):
+        assert bus.read() == []
+
+    def test_torn_final_line_is_skipped(self, bus, tmp_path):
+        bus.handle_line(line("Jonah joined the game"))
+        target = next((tmp_path / "events").glob("*.jsonl"))
+        target.write_text(target.read_text() + '{"type": "death", "incomplete"\n')
+        assert len(bus.read()) == 1
+
+    def test_unwritable_directory_does_not_raise(self, tmp_path):
+        """A full or read-only disk must not take the server down."""
+        blocker = tmp_path / "events"
+        blocker.write_text("this is a file, not a directory")
+
+        errors = []
+        broken = EventBus(events_dir=blocker, flush_every_events=1)
+        broken.set_error_logger(errors.append)
+        broken.handle_line(line("Jonah joined the game"))
+
+        assert errors and "Could not persist" in errors[0]
+
+
+@pytest.mark.unit
+class TestRetention:
+    def test_old_files_are_pruned(self, tmp_path):
+        events_dir = tmp_path / "events"
+        events_dir.mkdir()
+        (events_dir / "2020-01-01.jsonl").write_text("{}\n")
+        (events_dir / "2999-01-01.jsonl").write_text("{}\n")
+
+        assert EventBus(events_dir=events_dir, retention_days=30).prune() == 1
+        assert not (events_dir / "2020-01-01.jsonl").exists()
+        assert (events_dir / "2999-01-01.jsonl").exists()
+
+    def test_unrecognised_filenames_are_left_alone(self, tmp_path):
+        events_dir = tmp_path / "events"
+        events_dir.mkdir()
+        (events_dir / "notes.jsonl").write_text("{}\n")
+        EventBus(events_dir=events_dir, retention_days=1).prune()
+        assert (events_dir / "notes.jsonl").exists()
+
+    def test_retention_can_be_disabled(self, tmp_path):
+        events_dir = tmp_path / "events"
+        events_dir.mkdir()
+        (events_dir / "2020-01-01.jsonl").write_text("{}\n")
+        assert EventBus(events_dir=events_dir, retention_days=0).prune() == 0
+
+
+@pytest.mark.unit
+class TestSharedBus:
+    def test_get_bus_returns_a_singleton(self):
+        assert get_bus() is get_bus()
+
+    def test_reset_forces_a_rebuild(self):
+        first = get_bus()
+        reset_bus()
+        assert get_bus() is not first
+
+    def test_event_serialises_cleanly(self):
+        event = GameEvent(type=EVENT_CHAT, timestamp="2026-01-01T00:00:00+00:00", player="Jonah", data={"a": 1})
+        assert json.loads(event.to_json())["player"] == "Jonah"
+
+
+@pytest.mark.api
+class TestEventsEndpoints:
+    """Tests for GET /api/events and GET /api/events/types."""
+
+    @pytest.fixture(autouse=True)
+    def bus_with_events(self, tmp_path, monkeypatch):
+        """Point the shared bus at a temp directory holding known events."""
+        temp_bus = EventBus(events_dir=tmp_path / "events", flush_every_events=1)
+        temp_bus.handle_line(line("Jonah joined the game"))
+        temp_bus.handle_line(line("Silas was slain by Zombie"))
+        temp_bus.handle_line(line("<Jonah> gg"))
+
+        monkeypatch.setattr("api.events.get_bus", lambda: temp_bus)
+        return temp_bus
+
+    def test_requires_authentication(self, client):
+        assert client.get("/api/events").status_code == 401
+
+    def test_returns_events_newest_first(self, client, mock_api_keys):
+        response = client.get("/api/events", headers={"X-API-Key": mock_api_keys})
+        assert response.status_code == 200
+
+        payload = response.get_json()
+        assert payload["count"] == 3
+        assert [event["type"] for event in payload["events"]] == [EVENT_CHAT, EVENT_DEATH, EVENT_JOIN]
+
+    def test_filters_by_type(self, client, mock_api_keys):
+        response = client.get("/api/events?type=death", headers={"X-API-Key": mock_api_keys})
+        payload = response.get_json()
+        assert payload["count"] == 1
+        assert payload["events"][0]["player"] == "Silas"
+
+    def test_filters_by_player(self, client, mock_api_keys):
+        response = client.get("/api/events?player=Jonah", headers={"X-API-Key": mock_api_keys})
+        assert response.get_json()["count"] == 2
+
+    def test_respects_limit(self, client, mock_api_keys):
+        response = client.get("/api/events?limit=1", headers={"X-API-Key": mock_api_keys})
+        assert response.get_json()["count"] == 1
+
+    def test_rejects_unknown_type(self, client, mock_api_keys):
+        response = client.get("/api/events?type=nonsense", headers={"X-API-Key": mock_api_keys})
+        assert response.status_code == 400
+        assert "valid_types" in response.get_json()
+
+    def test_rejects_non_numeric_limit(self, client, mock_api_keys):
+        response = client.get("/api/events?limit=lots", headers={"X-API-Key": mock_api_keys})
+        assert response.status_code == 400
+
+    def test_limit_is_capped_at_500(self, client, mock_api_keys, bus_with_events):
+        """An unbounded limit would let one request read every stored event."""
+        for _ in range(600):
+            bus_with_events.handle_line(line("Jonah joined the game"))
+
+        response = client.get("/api/events?limit=99999", headers={"X-API-Key": mock_api_keys})
+        assert response.status_code == 200
+        assert response.get_json()["count"] == 500
+
+    def test_types_endpoint_lists_every_type(self, client, mock_api_keys):
+        response = client.get("/api/events/types", headers={"X-API-Key": mock_api_keys})
+        assert response.status_code == 200
+        assert set(response.get_json()["types"]) == {
+            EVENT_CHAT,
+            EVENT_CONNECT,
+            EVENT_JOIN,
+            EVENT_LEAVE,
+            EVENT_DEATH,
+            EVENT_ADVANCEMENT,
+            EVENT_COMMAND,
+            EVENT_SERVER_READY,
+            EVENT_SERVER_STOPPING,
+        }
+
+
+@pytest.mark.unit
+class TestFlushSafety:
+    """Concurrency and quiet-period behaviour of the buffered writer."""
+
+    def test_concurrent_flushes_do_not_interleave(self, tmp_path):
+        """Two flushes racing must not write half a line inside another."""
+        bus = EventBus(events_dir=tmp_path / "events", flush_every_events=10000, flush_every_seconds=3600)
+        for index in range(200):
+            bus.publish(GameEvent(type=EVENT_JOIN, timestamp="t", player=f"p{index}"))
+
+        barrier = threading.Barrier(4)
+
+        def flush_now():
+            barrier.wait()
+            bus.flush()
+
+        threads = [threading.Thread(target=flush_now) for _ in range(4)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+
+        written = (tmp_path / "events").glob("*.jsonl")
+        lines = [ln for path in written for ln in path.read_text().splitlines() if ln.strip()]
+
+        assert len(lines) == 200
+        # Every line must be a complete, parseable record.
+        players = {json.loads(ln)["player"] for ln in lines}
+        assert players == {f"p{index}" for index in range(200)}
+
+    def test_periodic_flush_writes_without_further_events(self, tmp_path):
+        """A quiet server previously left the last events buffered forever."""
+        bus = EventBus(
+            events_dir=tmp_path / "events",
+            flush_every_events=10000,
+            flush_every_seconds=0.1,
+        )
+        bus.handle_line(line("Jonah joined the game"))
+        assert not list((tmp_path / "events").glob("*.jsonl"))
+
+        bus.start_periodic_flush()
+        try:
+            deadline = time.monotonic() + 5
+            while time.monotonic() < deadline:
+                if list((tmp_path / "events").glob("*.jsonl")):
+                    break
+                time.sleep(0.05)
+            assert list((tmp_path / "events").glob("*.jsonl"))
+        finally:
+            bus.stop_periodic_flush()
+
+    def test_starting_the_timer_twice_arms_only_one(self, tmp_path):
+        bus = EventBus(events_dir=tmp_path / "events", flush_every_seconds=3600)
+        bus.start_periodic_flush()
+        first = bus._flush_timer
+        bus.start_periodic_flush()
+        try:
+            assert bus._flush_timer is first
+        finally:
+            bus.stop_periodic_flush()
+
+    def test_stopping_the_timer_flushes_what_is_left(self, tmp_path):
+        bus = EventBus(events_dir=tmp_path / "events", flush_every_events=10000, flush_every_seconds=3600)
+        bus.start_periodic_flush()
+        bus.handle_line(line("Jonah joined the game"))
+        bus.stop_periodic_flush()
+
+        assert len(bus.read()) == 1
+        assert bus._flush_timer is None
+
+    def test_stopping_without_starting_is_safe(self, tmp_path):
+        EventBus(events_dir=tmp_path / "events").stop_periodic_flush()
diff --git a/tests/api/test_log_streaming.py b/tests/api/test_log_streaming.py
index 1f5ee46..2713d9e 100644
--- a/tests/api/test_log_streaming.py
+++ b/tests/api/test_log_streaming.py
@@ -2,10 +2,12 @@
 """Tests for the WebSocket log stream.
 
 The stream was rewritten from a per-client `docker logs --tail` poll into a
-single `docker logs -f` follower that fans lines out to every subscriber. None
-of it had test coverage, so the behaviour that matters (one follower regardless
-of client count, lines delivered in order, the follower stopping when the last
-client leaves) was unverified.
+single `docker logs -f` follower that fans lines out to every subscriber, and
+then into an always-on follower that also drives the game event bus.
+
+The behaviour that matters: one follower regardless of client count, lines
+delivered in order, capture continuing with no clients attached, and the
+follower stopping only when it is explicitly told to.
 """
 
 import subprocess
@@ -31,9 +33,16 @@ def clean_stream_state():
     """Each test starts with no subscribers and no running follower."""
     api_module.active_log_streams.clear()
     api_module._log_reader_state["running"] = False
+    api_module._log_reader_state["stop"] = False
     yield
     api_module.active_log_streams.clear()
     api_module._log_reader_state["running"] = False
+    api_module._log_reader_state["stop"] = False
+
+
+def stop_after_first_attach(*_args, **_kwargs):
+    """Side effect for socketio.sleep that ends the follower's retry loop."""
+    api_module._log_reader_state["stop"] = True
 
 
 class TestGetLogTail:
@@ -90,8 +99,29 @@ def test_repeated_calls_start_only_one(self):
 
 
 class TestLogReader:
-    def test_exits_immediately_when_nobody_is_subscribed(self):
+    def test_follows_the_log_even_with_nobody_subscribed(self):
+        """Capture must not depend on a browser being open.
+
+        Events that happen while the dashboard is closed are exactly the ones
+        worth recording, so an empty subscriber set no longer stops the reader.
+        """
+        api_module._log_reader_state["running"] = True
+
+        proc = MagicMock()
+        proc.stdout = iter(["[16:04:23] [Server thread/INFO]: Jonah joined the game\n"])
+
+        with patch("api.server.subprocess.Popen", return_value=proc) as popen, patch.object(
+            api_module, "_publish_log_line"
+        ) as publish, patch.object(api_module.socketio, "sleep", side_effect=stop_after_first_attach):
+            api_module._log_reader()
+
+        popen.assert_called_once()
+        publish.assert_called_once()
+        assert api_module._log_reader_state["running"] is False
+
+    def test_stops_when_asked_to(self):
         api_module._log_reader_state["running"] = True
+        api_module._log_reader_state["stop"] = True
 
         with patch("api.server.subprocess.Popen") as popen:
             api_module._log_reader()
@@ -99,19 +129,27 @@ def test_exits_immediately_when_nobody_is_subscribed(self):
         popen.assert_not_called()
         assert api_module._log_reader_state["running"] is False
 
+    def test_stop_log_reader_sets_the_flag(self):
+        api_module.stop_log_reader()
+        assert api_module._log_reader_state["stop"] is True
+
+    def test_starting_clears_a_previous_stop(self):
+        api_module._log_reader_state["stop"] = True
+
+        with patch.object(api_module.socketio, "start_background_task"):
+            api_module._ensure_log_reader()
+
+        assert api_module._log_reader_state["stop"] is False
+
     def test_fans_each_line_out_to_every_subscriber(self):
         api_module.active_log_streams.update({"sid-a", "sid-b"})
 
         proc = MagicMock()
         proc.stdout = iter(["hello\n", "world\n"])
 
-        def drop_subscribers(*_args, **_kwargs):
-            # Ends the outer retry loop after the lines are consumed
-            api_module.active_log_streams.clear()
-
         with patch("api.server.subprocess.Popen", return_value=proc), patch.object(
             api_module.socketio, "emit"
-        ) as emit, patch.object(api_module.socketio, "sleep", side_effect=drop_subscribers):
+        ) as emit, patch.object(api_module.socketio, "sleep", side_effect=stop_after_first_attach):
             api_module._log_reader()
 
         rooms = [c.kwargs["room"] for c in emit.call_args_list]
@@ -128,9 +166,7 @@ def test_blank_lines_are_not_forwarded(self):
 
         with patch("api.server.subprocess.Popen", return_value=proc), patch.object(
             api_module.socketio, "emit"
-        ) as emit, patch.object(
-            api_module.socketio, "sleep", side_effect=lambda *_: api_module.active_log_streams.clear()
-        ):
+        ) as emit, patch.object(api_module.socketio, "sleep", side_effect=stop_after_first_attach):
             api_module._log_reader()
 
         forwarded = [c.args[1]["logs"][0] for c in emit.call_args_list if c.args[0] == "logs"]
@@ -156,9 +192,7 @@ def test_kills_the_follower_process_on_the_way_out(self):
 
         with patch("api.server.subprocess.Popen", return_value=proc), patch.object(
             api_module.socketio, "emit"
-        ), patch.object(
-            api_module.socketio, "sleep", side_effect=lambda *_: api_module.active_log_streams.clear()
-        ):
+        ), patch.object(api_module.socketio, "sleep", side_effect=stop_after_first_attach):
             api_module._log_reader()
 
         proc.kill.assert_called()
@@ -186,3 +220,153 @@ def test_removes_the_disconnecting_session(self):
             api_module.handle_disconnect()
 
         assert api_module.active_log_streams == {"sid-b"}
+
+
+class TestBacklogIsNotReplayed:
+    """The follower persists what it reads, so a replayed tail duplicates events.
+
+    Attaching with `--tail 200` meant every API restart and every re-attach
+    recorded the same chat, join and death lines again, and the counts climbed
+    with each one. New clients still get scrollback from `get_log_tail()` at
+    connect time, which does not touch the bus.
+    """
+
+    def test_follower_attaches_with_no_backlog(self):
+        api_module._log_reader_state["running"] = True
+
+        proc = MagicMock()
+        proc.stdout = iter([])
+
+        with patch("api.server.subprocess.Popen", return_value=proc) as popen, patch.object(
+            api_module.socketio, "emit"
+        ), patch.object(api_module.socketio, "sleep", side_effect=stop_after_first_attach):
+            api_module._log_reader()
+
+        command = popen.call_args.args[0]
+        assert "--tail" in command
+        assert command[command.index("--tail") + 1] == "0"
+
+    def test_reattaching_does_not_republish_old_lines(self):
+        """Two attaches in a row must publish each line exactly once."""
+        api_module._log_reader_state["running"] = True
+
+        attaches = []
+
+        def make_proc(*_args, **_kwargs):
+            proc = MagicMock()
+            # A replaying tail would hand back the same line on every attach.
+            proc.stdout = iter(["[16:04:23] [Server thread/INFO]: Jonah joined the game\n"])
+            attaches.append(proc)
+            return proc
+
+        def stop_on_second_attach(*_args, **_kwargs):
+            if len(attaches) >= 2:
+                api_module._log_reader_state["stop"] = True
+
+        with patch("api.server.subprocess.Popen", side_effect=make_proc), patch.object(
+            api_module, "_publish_log_line"
+        ) as publish, patch.object(api_module.socketio, "emit"), patch.object(
+            api_module.socketio, "sleep", side_effect=stop_on_second_attach
+        ):
+            api_module._log_reader()
+
+        # Two attaches, one line each, because neither replays a backlog.
+        assert publish.call_count == len(attaches)
+
+    def test_connecting_client_still_receives_scrollback(self):
+        """Dropping the follower's tail must not cost the UI its history."""
+        api_module.API_KEYS["scrollback-key"] = {"enabled": True}
+        request = MagicMock()
+        request.sid = "sid-scrollback"
+
+        try:
+            with patch("api.server.request", request), patch.object(
+                api_module, "get_log_tail", return_value=["old line"]
+            ) as tail, patch.object(api_module.socketio, "emit") as emit, patch.object(
+                api_module, "_ensure_log_reader"
+            ):
+                api_module.handle_connect({"api_key": "scrollback-key"})
+
+            tail.assert_called_once()
+            initial = [c.args[1] for c in emit.call_args_list if c.args[0] == "logs"]
+            assert initial and initial[0]["type"] == "initial"
+            assert initial[0]["logs"] == ["old line"]
+        finally:
+            api_module.API_KEYS.pop("scrollback-key", None)
+            api_module.active_log_streams.discard("sid-scrollback")
+
+
+class TestStoppingTheReader:
+    def test_stop_kills_the_running_process(self):
+        """A quiet server blocks in the read, so the flag alone cannot stop it."""
+        proc = MagicMock()
+        api_module._log_reader_state["proc"] = proc
+
+        try:
+            api_module.stop_log_reader()
+            assert api_module._log_reader_state["stop"] is True
+            proc.kill.assert_called_once()
+        finally:
+            api_module._log_reader_state["proc"] = None
+
+    def test_stop_is_safe_when_no_process_is_running(self):
+        api_module._log_reader_state["proc"] = None
+        api_module.stop_log_reader()
+        assert api_module._log_reader_state["stop"] is True
+
+    def test_stop_survives_a_process_that_already_exited(self):
+        proc = MagicMock()
+        proc.kill.side_effect = ProcessLookupError
+        api_module._log_reader_state["proc"] = proc
+
+        try:
+            api_module.stop_log_reader()
+            assert api_module._log_reader_state["stop"] is True
+        finally:
+            api_module._log_reader_state["proc"] = None
+
+    def test_process_reference_is_cleared_when_the_reader_exits(self):
+        api_module._log_reader_state["running"] = True
+
+        proc = MagicMock()
+        proc.stdout = iter([])
+
+        with patch("api.server.subprocess.Popen", return_value=proc), patch.object(
+            api_module.socketio, "emit"
+        ), patch.object(api_module.socketio, "sleep", side_effect=stop_after_first_attach):
+            api_module._log_reader()
+
+        assert api_module._log_reader_state["proc"] is None
+
+
+class TestCommandInputValidation:
+    """A truthy non-string command raised inside the sanitizer, which sits
+    outside the handler's try block, so the client got no error at all."""
+
+    def _emit_for(self, payload):
+        request = MagicMock()
+        request.sid = "sid-a"
+        with patch("api.server.request", request), patch.object(api_module.socketio, "emit") as emit, patch.object(
+            api_module, "run_rcon_command"
+        ) as run:
+            api_module.handle_execute_command(payload)
+        return emit, run
+
+    def test_numeric_command_returns_an_error(self):
+        emit, run = self._emit_for({"command": 1})
+
+        errors = [c.args[1]["message"] for c in emit.call_args_list if c.args[0] == "command_error"]
+        assert errors and "string" in errors[0].lower()
+        run.assert_not_called()
+
+    def test_list_command_returns_an_error(self):
+        emit, run = self._emit_for({"command": ["say", "hi"]})
+
+        assert any(c.args[0] == "command_error" for c in emit.call_args_list)
+        run.assert_not_called()
+
+    def test_missing_command_still_returns_an_error(self):
+        emit, run = self._emit_for({})
+
+        assert any(c.args[0] == "command_error" for c in emit.call_args_list)
+        run.assert_not_called()
diff --git a/tests/api/test_rcon.py b/tests/api/test_rcon.py
new file mode 100644
index 0000000..24f90f3
--- /dev/null
+++ b/tests/api/test_rcon.py
@@ -0,0 +1,555 @@
+"""Tests for the in-process RCON client (api/rcon.py).
+
+These run against a real socket server that speaks the RCON protocol, so the
+framing, authentication and multi-packet reassembly are exercised end to end
+rather than mocked.
+"""
+
+import socket
+import struct
+import sys
+import threading
+from pathlib import Path
+
+import pytest
+
+PROJECT_ROOT = Path(__file__).parent.parent.parent
+sys.path.insert(0, str(PROJECT_ROOT))
+
+from api.rcon import (  # noqa: E402
+    MAX_COMMAND_LENGTH,
+    PACKET_TYPE_AUTH_RESPONSE,
+    PACKET_TYPE_COMMAND,
+    PACKET_TYPE_LOGIN,
+    PACKET_TYPE_RESPONSE,
+    RconAuthError,
+    RconClient,
+    RconConnectionError,
+    RconError,
+    RconNotConfiguredError,
+    RconNotSentError,
+    RconUnknownOutcomeError,
+    _encode_packet,
+    execute,
+    get_client,
+    load_rcon_config,
+    reset_client,
+)
+
+# Fixture value only. The fake server below compares against this exact
+# string; it is not a credential for anything.
+TEST_PASSWORD = "fake-rcon-password-for-tests"
+
+
+def _encode(request_id, packet_type, body):
+    return _encode_packet(request_id, packet_type, body)
+
+
+def _read_packet(sock):
+    """Read one framed packet from a socket, for the fake server's use."""
+    header = b""
+    while len(header) < 4:
+        chunk = sock.recv(4 - len(header))
+        if not chunk:
+            return None
+        header += chunk
+    (length,) = struct.unpack("<i", header)
+
+    payload = b""
+    while len(payload) < length:
+        chunk = sock.recv(length - len(payload))
+        if not chunk:
+            return None
+        payload += chunk
+
+    request_id, packet_type = struct.unpack("<ii", payload[:8])
+    return request_id, packet_type, payload[8:-2].decode("utf-8", errors="replace")
+
+
+class FakeRconServer:
+    """A minimal RCON server that mirrors Minecraft's behaviour.
+
+    Responses longer than 4096 bytes are split into several packets carrying the
+    original request id, exactly as Minecraft's RconClient does, and unknown
+    packet types get an "Unknown request" reply so the sentinel works.
+    """
+
+    def __init__(self, password=TEST_PASSWORD, responses=None, answer_sentinel=True, drop_after_command=False):
+        self.password = password
+        self.responses = responses or {}
+        self.answer_sentinel = answer_sentinel
+        # Simulates a server that receives a command, may well act on it, then
+        # dies before the response is read.
+        self.drop_after_command = drop_after_command
+        self.received_commands = []
+        self.connection_count = 0
+        self._server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self._server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self._server.bind(("127.0.0.1", 0))
+        self._server.listen(5)
+        self.host, self.port = self._server.getsockname()
+        self._stop = threading.Event()
+        self._thread = threading.Thread(target=self._serve, daemon=True)
+
+    def start(self):
+        self._thread.start()
+        return self
+
+    def stop(self):
+        self._stop.set()
+        try:
+            self._server.close()
+        except OSError:
+            # Already closed, or the accept loop closed it first. Either way
+            # the listener is gone, which is what stop() is for.
+            pass
+
+    def _serve(self):
+        while not self._stop.is_set():
+            try:
+                conn, _ = self._server.accept()
+            except OSError:
+                return
+            self.connection_count += 1
+            threading.Thread(target=self._handle, args=(conn,), daemon=True).start()
+
+    def _handle(self, conn):
+        conn.settimeout(5.0)
+        authenticated = False
+        try:
+            while not self._stop.is_set():
+                packet = _read_packet(conn)
+                if packet is None:
+                    return
+                request_id, packet_type, body = packet
+
+                if packet_type == PACKET_TYPE_LOGIN:
+                    if body == self.password:
+                        authenticated = True
+                        conn.sendall(_encode(request_id, PACKET_TYPE_AUTH_RESPONSE, ""))
+                    else:
+                        # The protocol signals a bad password with id -1.
+                        conn.sendall(_encode(-1, PACKET_TYPE_AUTH_RESPONSE, ""))
+                elif packet_type == PACKET_TYPE_COMMAND and authenticated:
+                    self.received_commands.append(body)
+                    if self.drop_after_command:
+                        return
+                    self._send_response(conn, request_id, self.responses.get(body, f"ran: {body}"))
+                elif self.answer_sentinel:
+                    conn.sendall(_encode(request_id, PACKET_TYPE_RESPONSE, "Unknown request 0"))
+        except (OSError, socket.timeout):
+            return
+        finally:
+            try:
+                conn.close()
+            except OSError:
+                # The peer has usually closed first; nothing left to clean up.
+                pass
+
+    def _send_response(self, conn, request_id, message):
+        """Split into 4096-byte packets the way Minecraft does."""
+        if not message:
+            conn.sendall(_encode(request_id, PACKET_TYPE_RESPONSE, ""))
+            return
+        while message:
+            chunk, message = message[:4096], message[4096:]
+            conn.sendall(_encode(request_id, PACKET_TYPE_RESPONSE, chunk))
+
+
+@pytest.fixture
+def rcon_server():
+    server = FakeRconServer().start()
+    yield server
+    server.stop()
+
+
+@pytest.fixture(autouse=True)
+def clean_shared_client():
+    """Keep the module singleton from leaking between tests."""
+    reset_client()
+    yield
+    reset_client()
+
+
+@pytest.mark.unit
+class TestPacketEncoding:
+    """The length prefix and null terminators must match the protocol exactly."""
+
+    def test_length_field_covers_everything_after_itself(self):
+        packet = _encode_packet(7, PACKET_TYPE_COMMAND, "list")
+        (length,) = struct.unpack("<i", packet[:4])
+        assert length == len(packet) - 4
+        assert length == 4 + 4 + len("list") + 2
+
+    def test_fields_round_trip(self):
+        packet = _encode_packet(42, PACKET_TYPE_COMMAND, "say hello")
+        request_id, packet_type = struct.unpack("<ii", packet[4:12])
+        assert request_id == 42
+        assert packet_type == PACKET_TYPE_COMMAND
+        assert packet.endswith(b"\x00\x00")
+
+    def test_body_is_utf8_encoded(self):
+        packet = _encode_packet(1, PACKET_TYPE_COMMAND, "say héllo")
+        assert "héllo".encode("utf-8") in packet
+
+
+@pytest.mark.unit
+class TestConfigLoading:
+    def test_missing_file_returns_defaults(self, tmp_path):
+        settings = load_rcon_config(tmp_path / "absent.conf")
+        assert settings == {"host": "localhost", "port": 25575, "password": ""}
+
+    def test_parses_shell_assignments(self, tmp_path):
+        config = tmp_path / "rcon.conf"
+        config.write_text("# RCON settings\nRCON_HOST=10.0.0.5\nRCON_PORT=25580\nRCON_PASSWORD=example-value\n")
+        settings = load_rcon_config(config)
+        assert settings == {"host": "10.0.0.5", "port": 25580, "password": "example-value"}
+
+    def test_strips_quotes_and_ignores_comments(self, tmp_path):
+        config = tmp_path / "rcon.conf"
+        config.write_text('#RCON_PASSWORD=ignored\nRCON_PASSWORD="quoted"\n')
+        assert load_rcon_config(config)["password"] == "quoted"
+
+    def test_invalid_port_falls_back_to_default(self, tmp_path):
+        config = tmp_path / "rcon.conf"
+        config.write_text("RCON_PORT=not-a-number\n")
+        assert load_rcon_config(config)["port"] == 25575
+
+
+@pytest.mark.integration
+class TestCommandExecution:
+    def test_simple_command_round_trip(self, rcon_server):
+        with RconClient(rcon_server.host, rcon_server.port, TEST_PASSWORD) as client:
+            assert client.command("list") == "ran: list"
+        assert rcon_server.received_commands == ["list"]
+
+    def test_long_response_is_reassembled(self, rcon_server):
+        """The old shell fallback truncated at 4096 bytes; this must not."""
+        long_body = "x" * 10000
+        rcon_server.responses["dump"] = long_body
+        with RconClient(rcon_server.host, rcon_server.port, TEST_PASSWORD) as client:
+            assert client.command("dump") == long_body
+
+    def test_response_exactly_at_chunk_boundary(self, rcon_server):
+        body = "y" * 4096
+        rcon_server.responses["edge"] = body
+        with RconClient(rcon_server.host, rcon_server.port, TEST_PASSWORD) as client:
+            assert client.command("edge") == body
+
+    def test_empty_response_is_handled(self, rcon_server):
+        rcon_server.responses["quiet"] = ""
+        with RconClient(rcon_server.host, rcon_server.port, TEST_PASSWORD) as client:
+            assert client.command("quiet") == ""
+
+    def test_connection_is_reused_across_commands(self, rcon_server):
+        """The whole point of the module: one connection, many commands."""
+        with RconClient(rcon_server.host, rcon_server.port, TEST_PASSWORD) as client:
+            for _ in range(5):
+                client.command("list")
+        assert rcon_server.connection_count == 1
+
+    def test_batch_runs_in_order(self, rcon_server):
+        with RconClient(rcon_server.host, rcon_server.port, TEST_PASSWORD) as client:
+            results = client.commands(["one", "two", "three"])
+        assert results == ["ran: one", "ran: two", "ran: three"]
+        assert rcon_server.received_commands == ["one", "two", "three"]
+        assert rcon_server.connection_count == 1
+
+    def test_utf8_response_survives(self, rcon_server):
+        rcon_server.responses["motd"] = "Bienvenüe ✦"
+        with RconClient(rcon_server.host, rcon_server.port, TEST_PASSWORD) as client:
+            assert client.command("motd") == "Bienvenüe ✦"
+
+
+@pytest.mark.integration
+class TestAuthentication:
+    def test_wrong_password_raises_auth_error(self, rcon_server):
+        client = RconClient(rcon_server.host, rcon_server.port, "wrong-password")
+        with pytest.raises(RconAuthError):
+            client.connect()
+        assert not client.is_connected
+
+    def test_missing_password_raises_not_configured(self, rcon_server):
+        client = RconClient(rcon_server.host, rcon_server.port, "")
+        with pytest.raises(RconNotConfiguredError):
+            client.connect()
+
+    def test_failed_auth_does_not_leak_the_socket(self, rcon_server):
+        client = RconClient(rcon_server.host, rcon_server.port, "wrong-password")
+        with pytest.raises(RconAuthError):
+            client.connect()
+        assert client._sock is None
+
+
+@pytest.mark.integration
+class TestConnectionHandling:
+    def test_unreachable_server_raises_connection_error(self):
+        # Port 1 on localhost refuses connections.
+        client = RconClient("127.0.0.1", 1, TEST_PASSWORD, timeout=1.0)
+        with pytest.raises(RconConnectionError):
+            client.connect()
+
+    def test_connect_is_idempotent(self, rcon_server):
+        client = RconClient(rcon_server.host, rcon_server.port, TEST_PASSWORD)
+        client.connect()
+        client.connect()
+        assert rcon_server.connection_count == 1
+        client.close()
+
+    def test_close_is_safe_when_never_connected(self):
+        RconClient("127.0.0.1", 1, TEST_PASSWORD).close()
+
+    def test_reconnects_after_server_restart(self, rcon_server):
+        """A dropped connection must not turn into a failed command."""
+        client = RconClient(rcon_server.host, rcon_server.port, TEST_PASSWORD)
+        assert client.command("list") == "ran: list"
+
+        # Simulate the container restarting under us.
+        client._sock.close()
+
+        assert client.command("list") == "ran: list"
+        assert rcon_server.connection_count == 2
+        client.close()
+
+    def test_sentinel_timeout_still_returns_the_response(self):
+        """Some servers ignore the sentinel; the real response must survive."""
+        server = FakeRconServer(answer_sentinel=False).start()
+        try:
+            client = RconClient(server.host, server.port, TEST_PASSWORD, timeout=1.0)
+            assert client.command("list") == "ran: list"
+            client.close()
+        finally:
+            server.stop()
+
+
+@pytest.mark.unit
+class TestCommandValidation:
+    def test_empty_command_rejected(self, rcon_server):
+        client = RconClient(rcon_server.host, rcon_server.port, TEST_PASSWORD)
+        with pytest.raises(RconError):
+            client.command("   ")
+
+    def test_overlong_command_rejected_before_sending(self, rcon_server):
+        """The server would silently truncate these, so fail loudly instead."""
+        client = RconClient(rcon_server.host, rcon_server.port, TEST_PASSWORD)
+        with pytest.raises(RconError, match="truncated"):
+            client.command("a" * (MAX_COMMAND_LENGTH + 1))
+        assert rcon_server.received_commands == []
+
+    def test_request_ids_never_reach_the_reserved_value(self, rcon_server):
+        client = RconClient(rcon_server.host, rcon_server.port, TEST_PASSWORD)
+        client._request_id = 0x3FFFFFFF
+        assert client._next_request_id() == 1
+
+
+@pytest.mark.unit
+class TestSharedClient:
+    def test_get_client_returns_a_singleton(self):
+        assert get_client() is get_client()
+
+    def test_reset_client_forces_a_rebuild(self):
+        first = get_client()
+        reset_client()
+        assert get_client() is not first
+
+    def test_execute_reports_missing_config_as_503(self, monkeypatch):
+        """503 is the caller's signal to fall back to the shell script."""
+        monkeypatch.setattr("api.rcon.load_rcon_config", lambda *_: {"host": "h", "port": 1, "password": ""})
+        stdout, stderr, code = execute("list")
+        assert (stdout, code) == (None, 503)
+        assert "password" in stderr.lower()
+
+    def test_execute_reports_connection_failure_as_502(self, monkeypatch):
+        monkeypatch.setattr(
+            "api.rcon.load_rcon_config",
+            lambda *_: {"host": "127.0.0.1", "port": 1, "password": TEST_PASSWORD},
+        )
+        stdout, _, code = execute("list")
+        assert (stdout, code) == (None, 502)
+
+    def test_execute_returns_zero_on_success(self, monkeypatch, rcon_server):
+        monkeypatch.setattr(
+            "api.rcon.load_rcon_config",
+            lambda *_: {"host": rcon_server.host, "port": rcon_server.port, "password": TEST_PASSWORD},
+        )
+        stdout, stderr, code = execute("list")
+        assert (stdout, stderr, code) == ("ran: list", None, 0)
+
+
+@pytest.mark.integration
+class TestRetrySafety:
+    """Minecraft commands are not idempotent, so a retry must be provably safe.
+
+    Re-sending a `give`, `summon` or `kill` after a lost response would apply
+    the effect twice, so only failures that happened before the command reached
+    the server may be retried.
+    """
+
+    def test_stale_connection_is_retried(self):
+        """Nothing was sent, so the same command on a fresh socket is correct."""
+        server = FakeRconServer().start()
+        try:
+            client = RconClient(server.host, server.port, TEST_PASSWORD)
+            client.command("say first")
+            client._sock.close()
+
+            assert client.command("give Jonah diamond 1") == "ran: give Jonah diamond 1"
+            assert server.received_commands == ["say first", "give Jonah diamond 1"]
+            client.close()
+        finally:
+            server.stop()
+
+    def test_lost_response_is_not_retried(self):
+        """The server may already have run it, so it must not run again."""
+        server = FakeRconServer(drop_after_command=True).start()
+        try:
+            client = RconClient(server.host, server.port, TEST_PASSWORD, timeout=1.0)
+            with pytest.raises(RconUnknownOutcomeError):
+                client.command("give Jonah diamond 64")
+
+            assert server.received_commands == ["give Jonah diamond 64"]
+            client.close()
+        finally:
+            server.stop()
+
+    def test_not_sent_error_is_a_connection_error(self):
+        """Existing callers that catch RconConnectionError keep working."""
+        assert issubclass(RconNotSentError, RconConnectionError)
+
+    def test_unknown_outcome_is_not_a_connection_error(self):
+        """It must not be swept up by handlers that retry connection errors."""
+        assert not issubclass(RconUnknownOutcomeError, RconConnectionError)
+
+    def test_execute_maps_lost_response_to_500(self, monkeypatch):
+        """500 tells run_rcon_command not to re-run it through the shell."""
+        server = FakeRconServer(drop_after_command=True).start()
+        try:
+            monkeypatch.setattr(
+                "api.rcon.load_rcon_config",
+                lambda *_: {"host": server.host, "port": server.port, "password": TEST_PASSWORD},
+            )
+            stdout, stderr, code = execute("kill @e")
+            assert (stdout, code) == (None, 500)
+            assert "sent" in stderr.lower()
+        finally:
+            server.stop()
+
+
+@pytest.mark.unit
+class TestCredentialReload:
+    def test_changed_config_rebuilds_the_client(self, tmp_path, monkeypatch):
+        """Rotating the RCON password should not require an API restart."""
+        config = tmp_path / "rcon.conf"
+        config.write_text("RCON_HOST=localhost\nRCON_PORT=25575\nRCON_PASSWORD=first\n")
+        monkeypatch.setattr("api.rcon.RCON_CONFIG_FILE", config)
+
+        first = get_client()
+        assert first.password == "first"
+
+        # Timestamps have coarse resolution on some filesystems, so move the
+        # mtime explicitly rather than relying on the write being "later".
+        config.write_text("RCON_HOST=localhost\nRCON_PORT=25575\nRCON_PASSWORD=second\n")
+        import os
+
+        stat = config.stat()
+        os.utime(config, (stat.st_atime + 10, stat.st_mtime + 10))
+
+        second = get_client()
+        assert second is not first
+        assert second.password == "second"
+
+    def test_unchanged_config_keeps_the_same_client(self, tmp_path, monkeypatch):
+        config = tmp_path / "rcon.conf"
+        config.write_text("RCON_PASSWORD=stable\n")
+        monkeypatch.setattr("api.rcon.RCON_CONFIG_FILE", config)
+
+        assert get_client() is get_client()
+
+
+@pytest.mark.unit
+class TestCommandTypeValidation:
+    def test_non_string_command_is_rejected(self, rcon_server):
+        """The length check calls str methods, so this must fail cleanly."""
+        client = RconClient(rcon_server.host, rcon_server.port, TEST_PASSWORD)
+        with pytest.raises(RconError, match="string"):
+            client.command(42)
+        assert rcon_server.received_commands == []
+
+    def test_none_command_is_rejected(self, rcon_server):
+        client = RconClient(rcon_server.host, rcon_server.port, TEST_PASSWORD)
+        with pytest.raises(RconError):
+            client.command(None)
+
+
+@pytest.mark.api
+class TestRunRconCommandFallback:
+    """api/server.py falls back to the shell script only when that is safe.
+
+    Minecraft commands are not idempotent, so the fallback must run only when
+    the pooled client proved the command never reached the server.
+    """
+
+    @pytest.fixture
+    def server_module(self):
+        import api.server as module
+
+        return module
+
+    def _execute_returning(self, server_module, monkeypatch, code):
+        calls = {"script": 0}
+
+        def fake_execute(_command):
+            return (None, "failed", code) if code else ("ok", None, 0)
+
+        def fake_run_script(*_args, **_kwargs):
+            calls["script"] += 1
+            return "from script", None, 0
+
+        monkeypatch.setattr(server_module.rcon, "execute", fake_execute)
+        monkeypatch.setattr(server_module, "run_script", fake_run_script)
+        result = server_module.run_rcon_command("give Jonah diamond 1")
+        return result, calls["script"]
+
+    @pytest.mark.parametrize("code", [502, 503])
+    def test_falls_back_when_nothing_was_sent(self, server_module, monkeypatch, code):
+        result, script_calls = self._execute_returning(server_module, monkeypatch, code)
+        assert script_calls == 1
+        assert result[0] == "from script"
+
+    def test_does_not_fall_back_when_the_outcome_is_unknown(self, server_module, monkeypatch):
+        """The server may already have run it; running it again would duplicate."""
+        result, script_calls = self._execute_returning(server_module, monkeypatch, 500)
+        assert script_calls == 0
+        assert result[2] == 500
+
+    @pytest.mark.parametrize("code", [400, 401])
+    def test_does_not_fall_back_on_rejection(self, server_module, monkeypatch, code):
+        result, script_calls = self._execute_returning(server_module, monkeypatch, code)
+        assert script_calls == 0
+        assert result[2] == code
+
+    def test_success_never_touches_the_script(self, server_module, monkeypatch):
+        result, script_calls = self._execute_returning(server_module, monkeypatch, 0)
+        assert script_calls == 0
+        assert result == ("ok", None, 0)
+
+
+@pytest.mark.api
+class TestCommandEndpointTypeValidation:
+    def test_numeric_command_is_a_bad_request_not_a_crash(self, client, mock_api_keys):
+        """The sanitizer calls string methods, which raised a 500 before."""
+        response = client.post(
+            "/api/server/command",
+            headers={"X-API-Key": mock_api_keys},
+            json={"command": 1},
+        )
+        assert response.status_code == 400
+        assert "string" in response.get_json()["error"].lower()
+
+    def test_object_command_is_a_bad_request(self, client, mock_api_keys):
+        response = client.post(
+            "/api/server/command",
+            headers={"X-API-Key": mock_api_keys},
+            json={"command": {"say": "hi"}},
+        )
+        assert response.status_code == 400
diff --git a/web/src/services/__tests__/api.test.js b/web/src/services/__tests__/api.test.js
index 5eeb47d..59287b8 100644
--- a/web/src/services/__tests__/api.test.js
+++ b/web/src/services/__tests__/api.test.js
@@ -57,6 +57,65 @@ describe('API Service', () => {
     });
   });
 
+  describe('getEvents', () => {
+    it('sends the default limit and no filters', async () => {
+      mockAxiosInstance.get.mockResolvedValue({ data: { events: [], count: 0 } });
+
+      await api.getEvents();
+
+      expect(mockAxiosInstance.get).toHaveBeenCalledWith('/events', {
+        params: { limit: 100 },
+      });
+    });
+
+    it('passes type and player filters through', async () => {
+      mockAxiosInstance.get.mockResolvedValue({ data: { events: [], count: 0 } });
+
+      await api.getEvents({ limit: 25, type: 'death', player: 'Silas' });
+
+      expect(mockAxiosInstance.get).toHaveBeenCalledWith('/events', {
+        params: { limit: 25, type: 'death', player: 'Silas' },
+      });
+    });
+
+    it('omits filters that are not set', async () => {
+      mockAxiosInstance.get.mockResolvedValue({ data: { events: [], count: 0 } });
+
+      await api.getEvents({ type: 'chat' });
+
+      expect(mockAxiosInstance.get).toHaveBeenCalledWith('/events', {
+        params: { limit: 100, type: 'chat' },
+      });
+    });
+
+    it('returns the response payload', async () => {
+      const payload = { events: [{ type: 'death', player: 'Jonah' }], count: 1 };
+      mockAxiosInstance.get.mockResolvedValue({ data: payload });
+
+      await expect(api.getEvents()).resolves.toEqual(payload);
+    });
+  });
+
+  describe('getEventTypes', () => {
+    it('calls the types endpoint', async () => {
+      mockAxiosInstance.get.mockResolvedValue({ data: { types: ['chat', 'death'] } });
+
+      const result = await api.getEventTypes();
+
+      expect(result.types).toEqual(['chat', 'death']);
+      expect(mockAxiosInstance.get).toHaveBeenCalledWith('/events/types', { params: {} });
+    });
+
+    it('serves a repeat call from cache', async () => {
+      mockAxiosInstance.get.mockResolvedValue({ data: { types: ['chat'] } });
+
+      await api.getEventTypes();
+      await api.getEventTypes();
+
+      expect(mockAxiosInstance.get).toHaveBeenCalledTimes(1);
+    });
+  });
+
   describe('server control', () => {
     it('startServer calls correct endpoint', async () => {
       mockAxiosInstance.post.mockResolvedValue({
diff --git a/web/src/services/api.js b/web/src/services/api.js
index ff356e6..ece64b8 100644
--- a/web/src/services/api.js
+++ b/web/src/services/api.js
@@ -147,6 +147,19 @@ export const api = {
     return response.data;
   },
 
+  // Game events (chat, joins, deaths, advancements) parsed from the server log
+  async getEvents({ limit = 100, type = null, player = null } = {}) {
+    const params = { limit };
+    if (type) params.type = type;
+    if (player) params.player = player;
+    const response = await apiClient.get('/events', { params });
+    return response.data;
+  },
+
+  async getEventTypes() {
+    return cachedGet('/events/types', {}, 300000);
+  },
+
   // Players (cached for 3 seconds)
   async getPlayers() {
     return cachedGet('/players', {}, 3000);