Build a Local Battle Snapshot Bridge

What we’re building

A process-owned bridge exposing GET /v1/health and GET /v1/snapshot. Both require a bearer token. The snapshot reports a protocol version, sequence, context generation, coarse battle phase, and normalized local-player position. It exposes no generic commands, filesystem paths, account identity, chat, tokens, or arbitrary attributes.

Final result

• A threaded HTTP server bound to 127.0.0.1.
• Full-token comparison, two parsed GET paths and explicit common non-GET handlers.
• Immutable serialized game-thread observations.
• An inclusive 8192-byte payload cap and requested 0.25-second callback delay after sampling work.
• Staged shutdown, a requested one-second final join and daemon request threads.
• A separate Python 3 companion with limited envelope checks and a blocking-operation timeout.

What you’ll learn

• Why loopback still needs authentication.
• How to copy engine-owned observations into worker-readable primitives.
• Why a lock protects byte publication without making engine reads transactional.
• How sampled counters and timestamps differ from freshness guarantees.
• How companion request failure differs from a sampled offline payload.

Difficulty and prerequisites

Level 8 — External integration. Complete the Level 7 persistence tutorial. You should understand thread ownership, callback cancellation, coordinate normalization, schema versions, and secret handling.

Requirements and constraints

Functional

1. Enable only from an explicit valid configuration.
2. Bind to 127.0.0.1 with a non-Boolean integer port in 1024..65535.
3. Serve authenticated GET requests for parsed paths /v1/health and /v1/snapshot.
4. Collect/serialize on the game thread; handlers copy bytes without requesting game callbacks.
5. Stop in stages: clear running, cancel, request shutdown, close, then attempt the timed join.

Security

• The operator supplies a private random token. The loader checks 32–256 ASCII bytes and rejects REPLACE_; it does not test entropy. Keep secrets out of logs.
• No wildcard bind, command dispatcher, arbitrary attribute selector, filesystem-serving route, upload handler or CORS headers are implemented.
• GET rejects its body-header conditions and nonempty parsed query/fragment before authentication. Empty delimiters, scheme and authority are not explicitly rejected; this is not raw-target equality.
• Payloads omit player names, account IDs, chat, tokens, paths and environment data.

These scoped source contracts are not security certification. No request deadline, worker-count limit or immediate revocation gate for captured handler authorization is configured.

Acceptance criteria

These are review/test targets with conditions; the current matrix remains unexecuted by this source review.

• After preceding body/query checks, unauthenticated GET receives 401; an authenticated unknown parsed path receives 404.
• POST, HEAD, PUT, DELETE, PATCH and OPTIONS take the custom 405 JSON path. Other unsupported methods/parser errors retain inherited behavior; custom HEAD writes a body.
• Custom payloads are JSON of at most 8192 serialized bytes. Initial/reset snapshot bytes are {}, not a full envelope.
• No HTTP handler queries BigWorld.player, BigWorld.entity or game services.
• Sequence advances during construction before timestamp/encoding/publication. Later failure can leave gaps and retained old bytes.
• Sampled context-key changes advance generation, including the first offline sample. Same-ID avatar replacement, vehicle changes and replay seeks need not change it.
• Successful builds use null position for rejected controlled-entity/visual/space/geometry checks. Failures escaping local checks instead retain the old payload.
• The running guard blocks ordinary repeated start; stop/failure/restart cases require separate checks. Detached references do not provide universal retry.
• Synchronous shutdown precedes close and a requested one-second join; the complete stop and request-thread drain have no strict deadline.

Architecture

                 GAME THREAD
PlayerAvatar + arena + player vehicle
                 │ sample every 0.25 s
                 ▼
       validated primitive dictionary
                 │ json.dumps ≤ 8 KiB
                 ▼
       lock-protected immutable bytes
                 │ copy only
─────────────────┼──────────── thread boundary
                 ▼
   authenticated loopback HTTP handlers
                 │ GET + timeout
                 ▼
          external companion process

The diagram’s 0.25 seconds is a requested delay after work, not fixed frequency. Its eight-KiB cap is inclusive. The companion timeout applies to blocking operations, not a whole poll or server connection deadline.

Researching the implementation

The authored design gives HTTP workers only copied serialized bytes. Current client source resets playerVehicleID before arena construction, initializes arena period to WAITING and replaces it before notifications, and sets Vehicle.isStarted during visual startup. These are separate readiness stages. Current minimap X/Z use supports the axes, not this tutorial’s normalized tolerance or cadence.

The 28 October 2026 review inspected bounded latest-default source excerpts and matched normalized complete local Git-blob identities. This is static source/identity evidence, not a semantic review of every module, embedded-stdlib identity or new installed-client result.

Source lookup

The bridge spans game-thread state, immutable copies, loopback HTTP, and worker shutdown:

• Bridge ownership and lifecycle, loopback HTTP boundary, and crossing to the game thread.
• Resource and data safety — authentication, bounded requests, and primitive-only handoff.
• Snapshot versus provider — why request threads consume copied state rather than live engine objects.
• Callback teardown rules — invalidating publication before server shutdown.
• Python 2.7 data handling — the byte/Unicode boundary used by HTTP authorization and JSON output.

Inspect current owners: avatar entry, arena period state, visual readiness, arena bounds, minimap X/Z use, preferences path, and mod entry dispatch.

Useful searches are BaseHTTPServer, SocketServer.ThreadingMixIn, shutdown, server_close, getPreferencesDirPath, and the engine fields copied by _build_snapshot. Treat client names as investigation entry points and verify current field producers, callbacks, routes, and shutdown order. No request handler should import or call BigWorld.

Lifecycle and ownership

1. init: validate, construct server/thread, start the daemon thread, set running, then request a zero-delay callback. Running is a start receipt, not serve-loop or sampling health.
2. tick: clear the handle, check running, collect primitives, serialize/size-check before locking publication, then schedule a successor in finally.
3. context: compare the sampled key and advance generation/sequence before timestamp/publication; failures do not roll counters back.
4. request: apply GET body/query checks, authorize, dispatch parsed path, copy bytes under lock, then write after release.
5. fini: clear running, cancel, detach server, synchronously shut down, close, detach thread, request timed join, clear configuration and reset bytes. Uncaught join/reset errors can skip the tail.

Same-object restart preserves counters/context; a new process resets them. Handlers retain captured authorization/runtime references without a per-start epoch gate. Actual in-flight restart ordering is untested.

State and data-authority model

Implementation walkthrough

Publish, don’t call

Handlers do not schedule request-time game callbacks and wait for results. This avoids that particular waiter dependency; it does not rule out every timeout race or deadlock. The sampling interval is a scheduling target, not an upper bound on response age. Delayed callbacks or failed samples can leave the last published bytes in place; consumers need an explicit freshness policy using sampledAt, sequence and generation; the supplied companion enforces none.

Keep the lock short

JSON serialization occurs before acquiring the lock. The lock protects assignment/copy of a byte string only. Socket writes, logging, shutdown, and BigWorld access happen outside it.

Define a protocol

Every field has a primitive type and meaning. Protocol version is independent of the mod version, allowing compatible implementation updates without silently changing clients.

Read the protocol at its boundaries

state=battle means a PlayerAvatar was sampled; other player values mean offline. This is not arena/vehicle readiness. Phase accepts an integer/long excluding Boolean without an enum allowlist. Context uses any non-None arena ID, including zero, with object identity only for a missing ID; between-poll transitions are invisible.

Position reads the controlled playerVehicleID entity, truthy isStarted and equal space IDs. No health, inWorld, vehicle-type or non-null-space check is added. Finite positive X/Z bounds produce top-left normalized coordinates; inclusive values from −0.001 through 1.001 are clamped to [0, 1] and rounded to four places. This tolerance is fractional, not fixed world distance. Selected conversion errors return null; outer property/build failures retain old bytes.

Sequence/generation can skip unseen values after failed publication. sampledAt is integer wall-clock seconds, not engine/replay time or monotonic elapsed time. Health does not verify fresh sampling or a full snapshot envelope; initial/reset cache bytes are {}.

The Python 3 companion captures WOT_SNAPSHOT_TOKEN at import and uses fixed port 17841. It reads up to 8193 bytes, rejects more than 8192, decodes JSON and checks a dictionary with protocolVersion == 1; equality also admits true and 1.0. It does not validate all fields or reject repeated/stale counters. Missing formatting keys are among selected unavailable errors. Its token preflight checks only a UTF-8 byte minimum, unlike the server’s ASCII/max-length/prefix rules. A default opener has no explicit proxy/redirect policy; a one-second sleep follows each attempt. This does not pin the endpoint or guarantee one poll per second.

The configuration placeholder below is intentionally rejected until replaced. Changing its port alone does not update the companion URL. Configuration has no enforced schema-version field; release 1.0.0, protocol 1 and metadata ID are separate contracts.

Complete source code

# -*- coding: utf-8 -*-
"""Serve authenticated, read-only battle snapshots on loopback."""
from __future__ import absolute_import, unicode_literals

# Reader map:
#   1. Configuration and authentication guard the local HTTP boundary.
#   2. _Runtime builds immutable bytes on the game thread.
#   3. Request threads serve only those copied bytes; they never call BigWorld.

import BaseHTTPServer
import json
import logging
import math
import os
import SocketServer
import threading
import time
import urlparse

import BigWorld
from Avatar import PlayerAvatar
from helpers import getPreferencesDirPath

__version__ = '1.0.0'
_PROTOCOL_VERSION = 1
_LOGGER = logging.getLogger(__name__)
_MOD_NAME = 'local_battle_snapshot'
_CONFIG_PATH = os.path.join(getPreferencesDirPath(), 'mods', _MOD_NAME,
                            'config.json')
_MAX_RESPONSE_BYTES = 8192
_TICK_SECONDS = 0.25


def _finite_values(*values):
    return all(not math.isnan(value) and not math.isinf(value)
               for value in values)


def _constant_time_equal(left, right):
    """Compare byte tokens without an early exit on the first mismatch."""
    if not isinstance(left, str) or not isinstance(right, str):
        return False
    difference = len(left) ^ len(right)
    for index in xrange(max(len(left), len(right))):
        a = ord(left[index]) if index < len(left) else 0
        b = ord(right[index]) if index < len(right) else 0
        difference |= a ^ b
    return difference == 0


def _load_config():
    """Accept only an explicitly enabled, loopback-only configuration."""
    try:
        with open(_CONFIG_PATH, 'rb') as stream:
            data = json.load(stream)
    except IOError:
        _LOGGER.info('%s: disabled; configuration not found', _MOD_NAME)
        return None
    except Exception:
        _LOGGER.exception('%s: configuration read failed', _MOD_NAME)
        return None
    if not isinstance(data, dict) or data.get('enabled') is not True:
        return None
    bind = data.get('bind')
    port = data.get('port')
    token = data.get('token')
    if bind != '127.0.0.1':
        _LOGGER.error('%s: bind must be 127.0.0.1', _MOD_NAME)
        return None
    if (not isinstance(port, (int, long)) or isinstance(port, bool) or
            not 1024 <= port <= 65535):
        _LOGGER.error('%s: port is invalid', _MOD_NAME)
        return None
    if isinstance(token, unicode):
        try:
            token = token.encode('ascii')
        except UnicodeEncodeError:
            token = None
    if (not isinstance(token, str) or len(token) < 32 or len(token) > 256 or
            token.startswith('REPLACE_')):
        _LOGGER.error('%s: token must contain 32-256 ASCII bytes', _MOD_NAME)
        return None
    return {'bind': bind, 'port': port, 'token': token}


class _ThreadedServer(SocketServer.ThreadingMixIn,
                      BaseHTTPServer.HTTPServer):
    """Give each request a daemon thread while runtime owns server lifetime."""
    daemon_threads = True
    allow_reuse_address = True


class _Runtime(object):
    """Own the server, game callback, immutable snapshot, and shutdown order."""
    def __init__(self):
        self._running = False
        self._config = None
        self._server = None
        self._server_thread = None
        self._callback_id = None
        self._lock = threading.Lock()
        self._snapshot_bytes = b'{}'
        self._sequence = 0
        self._generation = 0
        self._context_key = None

    def start(self):
        """Validate configuration before starting either owned activity."""
        if self._running:
            return
        config = _load_config()
        if config is None:
            return
        self._config = config
        try:
            self._server = _ThreadedServer(
                (config['bind'], config['port']), self._make_handler())
            self._server_thread = threading.Thread(
                target=self._server.serve_forever,
                name='LocalBattleSnapshotHTTP')
            self._server_thread.daemon = True
            self._server_thread.start()
            self._running = True
            self._schedule(0.0)
        except Exception:
            _LOGGER.exception('%s: server startup failed', _MOD_NAME)
            self.stop()
            return
        _LOGGER.info('%s %s: listening on 127.0.0.1:%d',
                     _MOD_NAME, __version__, config['port'])

    def stop(self):
        """Cancel publication, stop accepting requests, then bound the join."""
        if (not self._running and self._server is None and
                self._config is None):
            return
        was_running = self._running
        self._running = False
        if self._callback_id is not None:
            try:
                BigWorld.cancelCallback(self._callback_id)
            except Exception:
                _LOGGER.exception('%s: callback cancellation failed', _MOD_NAME)
            self._callback_id = None
        server, self._server = self._server, None
        if server is not None:
            if was_running:
                try:
                    server.shutdown()
                except Exception:
                    _LOGGER.exception('%s: server shutdown failed', _MOD_NAME)
            try:
                server.server_close()
            except Exception:
                _LOGGER.exception('%s: server close failed', _MOD_NAME)
        thread, self._server_thread = self._server_thread, None
        if (was_running and thread is not None and
                thread is not threading.current_thread()):
            thread.join(1.0)
            if thread.is_alive():
                _LOGGER.warning('%s: server thread did not stop in time',
                                _MOD_NAME)
        self._config = None
        with self._lock:
            self._snapshot_bytes = b'{}'
        _LOGGER.info('%s: stopped', _MOD_NAME)

    def _schedule(self, delay):
        if self._running:
            self._callback_id = BigWorld.callback(delay, self._tick)

    def _tick(self):
        """Build and publish one immutable response entirely on game thread."""
        self._callback_id = None
        if not self._running:
            return
        try:
            snapshot = self._build_snapshot()
            encoded = json.dumps(snapshot, sort_keys=True,
                                 separators=(',', ':')).encode('ascii')
            if len(encoded) > _MAX_RESPONSE_BYTES:
                raise ValueError('snapshot exceeds response limit')
            with self._lock:
                self._snapshot_bytes = encoded
        except Exception:
            _LOGGER.exception('%s: snapshot publication failed', _MOD_NAME)
        finally:
            self._schedule(_TICK_SECONDS)

    def _build_snapshot(self):
        """Copy primitive client observations into the versioned protocol."""
        player = BigWorld.player()
        context_key = ('none',)
        state = 'offline'
        phase = None
        position = None
        if isinstance(player, PlayerAvatar):
            arena = getattr(player, 'arena', None)
            arena_id = getattr(player, 'arenaUniqueID', None)
            context_key = ('battle', unicode(arena_id) if arena_id is not None
                           else 'avatar:%s' % id(player))
            state = 'battle'
            phase_value = getattr(arena, 'period', None)
            if isinstance(phase_value, (int, long)) and not isinstance(phase_value, bool):
                phase = int(phase_value)
            position = self._normalized_position(player, arena)
        if context_key != self._context_key:
            self._context_key = context_key
            self._generation += 1
        self._sequence += 1
        return {'protocolVersion': _PROTOCOL_VERSION,
                'modVersion': __version__, 'sequence': self._sequence,
                'contextGeneration': self._generation,
                'sampledAt': int(time.time()), 'state': state,
                'phase': phase, 'position': position}

    @staticmethod
    def _normalized_position(player, arena):
        """Return copied normalized X/Z or None when bounds/entity are invalid."""
        arena_type = getattr(arena, 'arenaType', None)
        bounds = getattr(arena_type, 'boundingBox', None)
        try:
            left, bottom = float(bounds[0][0]), float(bounds[0][1])
            right, top = float(bounds[1][0]), float(bounds[1][1])
            if not _finite_values(left, bottom, right, top):
                return None
            width, height = right - left, top - bottom
            if not _finite_values(width, height) or width <= 0 or height <= 0:
                return None
            vehicle_id = getattr(player, 'playerVehicleID', None)
            vehicle = BigWorld.entity(vehicle_id) if vehicle_id else None
            if (vehicle is None or not getattr(vehicle, 'isStarted', False) or
                    getattr(vehicle, 'spaceID', None) != getattr(player, 'spaceID', None)):
                return None
            x, z = float(vehicle.position.x), float(vehicle.position.z)
            if not _finite_values(x, z):
                return None
            nx, nz = (x - left) / width, (z - bottom) / height
            if not _finite_values(nx, nz):
                return None
            if nx < -0.001 or nx > 1.001 or nz < -0.001 or nz > 1.001:
                return None
            return [round(min(1.0, max(0.0, nx)), 4),
                    round(1.0 - min(1.0, max(0.0, nz)), 4)]
        except (AttributeError, IndexError, TypeError, ValueError, OverflowError):
            return None

    def _make_handler(self):
        """Create a handler class closed over copied runtime state only."""
        runtime = self
        expected_authorization = ('Bearer '.encode('ascii') +
                                  self._config['token'])

        class Handler(BaseHTTPServer.BaseHTTPRequestHandler):
            protocol_version = 'HTTP/1.1'

            def do_GET(self):
                if (self.headers.get('Content-Length') not in (None, '', '0') or
                        self.headers.get('Transfer-Encoding') is not None):
                    self._reply(400, {'error': 'request_body_not_allowed'})
                    return
                parsed = urlparse.urlsplit(self.path)
                if parsed.query or parsed.fragment:
                    self._reply(404, {'error': 'not_found'})
                    return
                if not _constant_time_equal(
                        self.headers.get('Authorization') or '',
                        expected_authorization):
                    self._reply(401, {'error': 'unauthorized'})
                    return
                if parsed.path == '/v1/health':
                    self._reply(200, {'ok': True,
                                      'protocolVersion': _PROTOCOL_VERSION})
                elif parsed.path == '/v1/snapshot':
                    with runtime._lock:
                        payload = runtime._snapshot_bytes
                    self._reply_bytes(200, payload)
                else:
                    self._reply(404, {'error': 'not_found'})

            def do_POST(self):
                self._reply(405, {'error': 'method_not_allowed'})

            do_HEAD = do_POST
            do_PUT = do_POST
            do_DELETE = do_POST
            do_PATCH = do_POST
            do_OPTIONS = do_POST

            def _reply(self, status, data):
                payload = json.dumps(data, sort_keys=True,
                                     separators=(',', ':')).encode('ascii')
                self._reply_bytes(status, payload)

            def _reply_bytes(self, status, payload):
                if len(payload) > _MAX_RESPONSE_BYTES:
                    status, payload = 500, b'{"error":"response_too_large"}'
                self.send_response(status)
                self.send_header('Content-Type', 'application/json; charset=utf-8')
                self.send_header('Content-Length', str(len(payload)))
                self.send_header('Cache-Control', 'no-store')
                self.send_header('Connection', 'close')
                self.end_headers()
                self.wfile.write(payload)

            def log_message(self, unused_format, *unused_args):
                pass

        return Handler


_RUNTIME = _Runtime()


def init():
    """World of Tanks mod-loader entry point."""
    try:
        _RUNTIME.start()
    except Exception:
        _LOGGER.exception('%s: initialization failed', _MOD_NAME)


def fini():
    """World of Tanks mod-loader cleanup entry point."""
    try:
        _RUNTIME.stop()
    except Exception:
        _LOGGER.exception('%s: finalization failed', _MOD_NAME)

#!/usr/bin/env python3
"""Poll the local battle snapshot without importing game code."""

# This separate Python 3 process knows only the documented HTTP protocol.
# It deliberately imports no World of Tanks or BigWorld modules.
import json
import os
import time
import urllib.error
import urllib.request

URL = 'http://127.0.0.1:17841/v1/snapshot'
TOKEN = os.environ.get('WOT_SNAPSHOT_TOKEN', '')


def read_snapshot():
    """Fetch, size-check, decode, and validate one protocol response."""
    request = urllib.request.Request(
        URL, headers={'Authorization': 'Bearer ' + TOKEN,
                      'Accept': 'application/json'})
    with urllib.request.urlopen(request, timeout=1.0) as response:
        if response.status != 200:
            raise RuntimeError('unexpected HTTP status %s' % response.status)
        payload = response.read(8193)
    if len(payload) > 8192:
        raise ValueError('response exceeds size limit')
    data = json.loads(payload.decode('utf-8'))
    if not isinstance(data, dict) or data.get('protocolVersion') != 1:
        raise ValueError('unsupported snapshot protocol')
    return data


def main():
    """Poll with finite timeouts and report temporary unavailability."""
    if len(TOKEN.encode('utf-8')) < 32:
        raise SystemExit('Set WOT_SNAPSHOT_TOKEN to the configured token.')
    while True:
        try:
            snapshot = read_snapshot()
            print('state={state} phase={phase} position={position} seq={sequence}'.format(
                **snapshot))
        except (urllib.error.URLError, OSError, ValueError, RuntimeError,
                KeyError) as error:
            print('bridge unavailable: %s' % error)
        time.sleep(1.0)


if __name__ == '__main__':
    main()

{
  "bind": "127.0.0.1",
  "enabled": true,
  "port": 17841,
  "token": "REPLACE_WITH_AT_LEAST_32_RANDOM_CHARACTERS"
}

Configuration field guide: bind is deliberately fixed to loopback, enabled is an explicit opt-in, port selects the local endpoint, and token must be replaced with a private random value of sufficient byte length. The block remains strict JSON for direct use.

<root><id>tutorial.local_battle_snapshot</id><version>1.0.0</version><name>Local Battle Snapshot</name><description>Serves authenticated read-only client snapshots on loopback.</description></root>

Minimal project structure

local-battle-snapshot/
├── mod_local_battle_snapshot.py
├── battle_snapshot_companion.py       # external Python 3 process
└── stage/
    ├── meta.xml
    └── res/scripts/client/gui/mods/mod_local_battle_snapshot.pyc

preferences directory (created by the user):
└── mods/local_battle_snapshot/config.json

Packaging and installation

1. Generate a private random token; place it in the preferences-side config and the companion environment variable.
2. Compile only the mod with the target-compatible Python 2.7 compiler.
3. Stage bytecode and aligned manifest in a stored-member local_battle_snapshot_1.0.0.wotmod.
4. Keep the Python 3 companion outside the game package.
5. Inspect and extract-test the archive; installation and firewall behavior require user-side validation.

Testing and debugging

The original tutorial retains its 2.3.1.3 evidence scope. The 28 October 2026 reconciliation is static: current acceptance, concurrency, HTTP, security and installed-client scenarios were not executed by this review.

Source checks and proposed controls

• Inspect handler ASTs for engine/service calls, parse JSON/XML and check mod Python 2.7 and companion Python 3 syntax separately.
• Proposed isolated controls cover token/port/bind rules, GET validation order, parsed targets, explicit methods and inclusive byte limits.
• Fake player/arena/vehicle controls should separate sampled context, finite geometry, new null-position payloads and failed-publication retention.
• A proposed 50-reader race is a concurrency test. Complete JSON parsing alone does not establish a full/fresh protocol envelope; {} also parses. This review ran no such race.
• Fault-test bind, thread construction/start, scheduling, cancellation, shutdown, close and join separately. Unstarted-thread failure differs from scheduling failure after the running receipt. Earlier inert partial-startup controls do not prove universal cleanup/retry after join failure.

Runtime matrix

Actual firewall, framing, worker admission, proxy/redirect behavior, native scheduling and in-flight teardown/restart races require separate runtime evidence.

Failure modes and edge cases

• Port occupied: startup logs a traceback and attempts partial-state cleanup; this is not a promise that every cleanup stage succeeds.
• Token leaked: replace it in both locations; loopback does not revoke it.
• Slow client: request I/O runs in daemon workers outside the game thread. This example configures no server-side connection timeout, whole-request deadline or worker-count limit. A close header ends a completed response; it does not time out an incomplete request or write.
• Stale snapshot: consumers need an explicit policy for timestamps/counters. Health and the supplied companion do not enforce freshness.
• Build failure: log a traceback and retain old bytes. Successor scheduling is in finally outside that catch; scheduling failure can leave running true without a successor.
• Final join timeout: warn if the server thread is still alive after that timed join. The earlier synchronous shutdown() call has its own wait and is not bounded by the later join.

A constructed thread is not a successful start. This example sets _running only after Thread.start() returns, and uses the captured was_running value to guard both shutdown and joining. The early-return guard also checks that no loaded configuration remains. Ordinary bind/thread-start failure attempts reset and closes a constructed server without joining an unstarted thread. This does not cover every startup failure, uncaught later join/reset error or safe restart of admitted handlers. The Python threading contract rejects that join. The same rejection was separately verified by static inspection of CPython 2.7.18 documentation and implementation; neither the current reference nor that historical API inspection verifies the embedded client runtime. Selected startup/teardown controls use fake server, thread, lock and scheduler objects, with no actual thread, wait, socket or file operations.

Keep timing limits attached to their stage

stop() calls server.shutdown() synchronously before thread.join(1.0). The documented shutdown contract waits for the server loop to finish. A timeout on the later join therefore does not bound the complete stop operation. Daemon request threads and an eight-KiB response cap also do not impose a connection deadline.

The companion's urlopen(..., timeout=1.0) is separate from server configuration and is not an end-to-end request deadline. Historical CPython 2.7.18 source inspection separately confirms an untimed shutdown-event wait and a default request-handler timeout of None; this example overrides neither. These are static API and authored-code findings, not an observed client hang, socket test or new verification of the embedded runtime. The implementation and its historical tutorial verification remain unchanged.

Performance considerations

After work, the game thread requests another callback with a 0.25-second delay; execution delay and sampling/serialization work affect cadence. No fixed four-Hz rate, maximum response age or total CPU/memory bound follows. HTTP handlers schedule no per-request engine work. The serialized payload cap is inclusive; connection duration, worker count and total shutdown duration have no explicit bound.

Compatibility and version sensitivity

Why this architecture?

• Why not synchronous RPC? A read-only snapshot needs no request-time engine action and avoids that request/waiter dependency.
• Why authenticate loopback? Local processes and some browser-origin attempts may reach loopback. Absence of CORS headers is not authentication or a connection-admission guarantee.
• Why no CORS? The intended consumer is a local process, not arbitrary web content.
• Why pre-serialize? Workers copy one immutable value and cannot observe a half-mutated dictionary.
• Why no commands? Read-only telemetry keeps the trust surface narrow and teaches the boundary before mutation.

Possible extensions

• Add ETag/sequence conditional reads without touching the game thread.
• Use a bounded server-sent event stream only after client-count and shutdown behavior are designed.
• Add a companion GUI that treats snapshots as stale after a declared threshold.
• Add write commands only with per-command schemas, a bounded queue, game-thread revalidation, and explicit authorization.
• Advance to Level 9 by combining battle lifecycle, entity observations, spatial analytics, persistence, and Gameface.

Maintenance checklist

• Recheck PlayerAvatar, arena-field, callback, and preferences-helper imports/signatures.
• Confirm the server still binds only to loopback.
• Audit exact routes, methods, payload fields, and maximum sizes.
• Search handler code for engine/service access.
• Retest context generation through consecutive battles and reconnects.
• Fault-test bind, callback, shutdown, and join paths.
• Check token redaction in every log.
• Verify Python 2 mod syntax and Python 3 companion syntax separately.
• Runtime-test local firewall and target-client teardown.

What you learned

An external integration is two programs separated by a trust and thread boundary. This bridge publishes versioned game-thread observations and serves cached immutable bytes through authenticated workers. Freshness, resource admission and shutdown require separate policies and evidence.