Game of Life source

"""Conway engine for the ``/apps/life`` demo.

One shared board. Commands queue between ticks; ``step()`` advances one

generation, applies the queue on top of it, and reports which cells

changed. Every painted cell stamps a plus onto a dead anchor — strokes

bloom into evolving trails and taps leave ornaments that settle into

pulsing oscillators. A live anchor falls through to the erase toggle.

Cells carry no owner. Each live cell stores its age in generations and

the view maps that to an ember ramp — newborns glow, elders go ashen.

"""

import textwrap

from bisect import bisect_right

from collections.abc import Iterable

from dataclasses import dataclass

COLS = 256

ROWS = 256

CELLS = COLS * ROWS

TICK_HZ = 5

TICK_SECONDS = 1 / TICK_HZ

# The board is served in 16x16 chunks; clients render a viewport of them.

CHUNK = 16

CHUNK_COLS = COLS // CHUNK

CHUNK_ROWS = ROWS // CHUNK

CHUNKS = CHUNK_COLS * CHUNK_ROWS

# A viewport may span at most this many chunks in each direction — bounds

# the payload a single stream can ask for.

MAX_VIEW_COLS = 15

MAX_VIEW_ROWS = 10

MAX_PENDING = 4096  # commands buffered per tick, all players together

MAX_PER_PLAYER = 96  # commands one player may queue per tick

# Live-cell count where step() switches neighbor counting from a dict to a

# flat CELLS-long list — measured crossover near 2k cells.

_DENSE_CUTOFF = 2048

# Seed cells start mid-ember so the opening board looks established.

SEED_AGE = 4

def _stamp(art: str) -> tuple[tuple[int, int], ...]:

    """ASCII art -> centered ``(dx, dy)`` offsets. ``X`` is a live cell."""

    rows = textwrap.dedent(art).strip("\n").split("\n")

    width = max(len(row) for row in rows)

    cy, cx = len(rows) // 2, width // 2

    return tuple((x - cx, y - cy) for y, row in enumerate(rows) for x, ch in enumerate(row) if ch == "X")

# Seed-only patterns. The demo has no stamp picker — clicks toggle.

STAMPS: dict[str, tuple[tuple[int, int], ...]] = {

    "blinker": _stamp("XXX"),

    "beacon": _stamp(

        """

        XX..

        XX..

        ..XX

        ..XX

        """

    ),

    "glider": _stamp(

        """

        .X.

        ..X

        XXX

        """

    ),

    "rpentomino": _stamp(

        """

        .XX

        XX.

        .X.

        """

    ),

    "pulsar": _stamp(

        """

        ..XXX...XXX..

        .............

        X....X.X....X

        X....X.X....X

        X....X.X....X

        ..XXX...XXX..

        .............

        ..XXX...XXX..

        X....X.X....X

        X....X.X....X

        X....X.X....X

        .............

        ..XXX...XXX..

        """

    ),

}

# The tap ornament: a five-cell plus. Alone on the board it churns about

# eight generations and settles into a twelve-cell period-2 oscillator —

# a permanent twinkle where someone poked the board.

PLUS = ((0, 0), (1, 0), (-1, 0), (0, -1), (0, 1))

STAMPABLE = {"plus": PLUS}

# Age (generations alive) -> ember band a0..a7. ~9s from spark to ash.

# The band is also the repaint unit: a chunk re-renders when any cell in it

# dies, is born, or crosses a band boundary.

_AGE_STEPS = (1, 3, 6, 11, 18, 29, 46)

_STEP_SET = frozenset(_AGE_STEPS)

def band(age: int) -> int:

    return min(bisect_right(_AGE_STEPS, age), 7)

_GOSPER_GUN = _stamp(

    """

    ........................X

    ......................X.X

    ............XX......XX............XX

    ...........X...X....XX............XX

    XX........X.....X...XX

    XX........X...X.XX....X.X

    ..........X.....X.......X

    ...........X...X

    ............XX

    """

)

# Origins are stamp anchors: stamps are centered, so the gun needs x >= 18.

_SEEDS: tuple[tuple[tuple[tuple[int, int], ...], int, int], ...] = (

    (_GOSPER_GUN, 30, 30),

    (STAMPS["glider"], 120, 25),

    (STAMPS["rpentomino"], 128, 150),

    (STAMPS["pulsar"], 190, 100),

    (STAMPS["blinker"], 70, 200),

    (STAMPS["beacon"], 220, 210),

)

# Index offsets of the 8 neighbors — valid for non-edge cells.

_NEIGHBOR_OFFS = (-COLS - 1, -COLS, -COLS + 1, -1, 1, COLS - 1, COLS, COLS + 1)

def _idx(x: int, y: int) -> int | None:

    """Board index for ``(x, y)``; ``None`` off the board. Edges are dead."""

    if 0 <= x < COLS and 0 <= y < ROWS:

        return y * COLS + x

    return None

def _chunk_of(i: int) -> int:

    """Chunk id for a board cell index."""

    return (i // COLS // CHUNK) * CHUNK_COLS + (i % COLS) // CHUNK

@dataclass(frozen=True, slots=True)

class Frame:

    """One computed tick: what the broadcast has to repaint."""

    generation: int

    changed: frozenset[int]

    changed_chunks: frozenset[int]

    applied: int

class Game:

    """The one shared board. Mutated only by ``join``, ``leave``, ``queue``,

    and ``step`` — all on the app loop.

    ``cells`` maps board index -> age in generations. ``players`` is the

    set of visitor ids holding an open stream; it only feeds the count.

    ``viewports``/``rendered`` track which 16x16 chunks each stream sees.

    """

    def __init__(self, *, seed: bool = True) -> None:

        self.cells: dict[int, int] = {}

        self.players: set[str] = set()

        self.viewports: dict[str, tuple[int, int, int, int]] = {}

        self.rendered: dict[str, frozenset[int]] = {}

        self.catchup: dict[str, set[int]] = {}

        # chunk_rev bumps per changed chunk each step; chunk_cache/chunk_seen

        # make chunk_html content-addressed — it renders only stale chunks.

        self.chunk_rev = [0] * CHUNKS

        self.chunk_seen = [-1] * CHUNKS

        self.chunk_cache: list[str | None] = [None] * CHUNKS

        self.pending: list[tuple[str, int, int, str | None]] = []

        self.generation = 0

        self.last_applied = 0

        self._quota: dict[str, int] = {}

        if seed:

            self._seed()

    def _seed(self) -> None:

        for pattern, ox, oy in _SEEDS:

            for dx, dy in pattern:

                i = _idx(ox + dx, oy + dy)

                if i is not None:

                    self.cells[i] = SEED_AGE

    # -- players ------------------------------------------------------

    def join(self, visitor: str) -> None:

        self.players.add(visitor)

    def leave(self, visitor: str) -> None:

        self.players.discard(visitor)

        self.viewports.pop(visitor, None)

        self.rendered.pop(visitor, None)

        self.catchup.pop(visitor, None)

    def view(self, visitor: str, cx0: int, cy0: int, cx1: int, cy1: int) -> frozenset[int]:

        """Record a visitor's chunk viewport; return chunks needing a paint.

        Chunks that leave the viewport drop out of ``rendered`` — they go

        stale off-screen and repaint in full if the visitor scrolls back.

        """

        cx0 = max(0, min(cx0, CHUNK_COLS - 1))

        cy0 = max(0, min(cy0, CHUNK_ROWS - 1))

        cx1 = max(cx0, min(cx1, min(CHUNK_COLS - 1, cx0 + MAX_VIEW_COLS - 1)))

        cy1 = max(cy0, min(cy1, min(CHUNK_ROWS - 1, cy0 + MAX_VIEW_ROWS - 1)))

        viewport = frozenset(cy * CHUNK_COLS + cx for cy in range(cy0, cy1 + 1) for cx in range(cx0, cx1 + 1))

        needed = viewport - self.rendered.get(visitor, frozenset())

        self.viewports[visitor] = (cx0, cy0, cx1, cy1)

        self.rendered[visitor] = viewport

        # Keep needed buffered as well: a /view publish can land before the

        # visitor's stream subscribes, so the join (or next tick) drains it.

        if needed:

            self.catchup.setdefault(visitor, set()).update(needed)

        return needed

    # -- commands -----------------------------------------------------

    def queue(self, visitor: str, x: int, y: int, stamp: str | None = None) -> bool:

        """Buffer a command for the next tick. ``False`` = dropped.

        ``stamp`` names a shape planted on a dead cell (a tap upgrade);

        no stamp toggles the cell the clicker saw. Player membership is

        not required — a click that lands before the stream connects is

        still a real intent, and the caps below bound abuse either way.

        """

        if _idx(x, y) is None:

            return False

        if stamp is not None and stamp not in STAMPABLE:

            return False

        if len(self.pending) >= MAX_PENDING:

            return False

        used = self._quota.get(visitor, 0)

        if used >= MAX_PER_PLAYER:

            return False

        self._quota[visitor] = used + 1

        self.pending.append((visitor, x, y, stamp))

        return True

    # -- tick ---------------------------------------------------------

    def step(self) -> Frame:

        """Apply queued commands, run one Conway generation, report the diff."""

        prev = self.cells

        self.last_applied = len(self.pending)

        pending, self.pending = self.pending, []

        self._quota.clear()

        # Neighbor counting — two containers, picked by density. A flat

        # list wins on dense boards (indexed adds beat dict get/set); a

        # dict wins on sparse boards (the survival pass skips dead cells).

        # Interior cells take the offset fast path either way. The flat

        # pass must not rescan the whole board for nonzero cells — a full

        # enumerate costs more than the counting itself at 5 Hz, so

        # ``touched`` records each index the first time it gets a count.

        counted: Iterable[tuple[int, int]]

        if len(prev) > _DENSE_CUTOFF:

            flat = [0] * CELLS

            touched: list[int] = []

            tpush = touched.append

            for i in prev:

                x, y = i % COLS, i // COLS

                if 0 < x < COLS - 1 and 0 < y < ROWS - 1:

                    for o in _NEIGHBOR_OFFS:

                        j = i + o

                        v = flat[j] + 1

                        flat[j] = v

                        if v == 1:

                            tpush(j)

                else:

                    for ny in (y - 1, y, y + 1):

                        for nx in (x - 1, x, x + 1):

                            if (nx != x or ny != y) and 0 <= nx < COLS and 0 <= ny < ROWS:

                                j = ny * COLS + nx

                                v = flat[j] + 1

                                flat[j] = v

                                if v == 1:

                                    tpush(j)

            counted = ((i, flat[i]) for i in touched)

        else:

            counts: dict[int, int] = {}

            cget = counts.get

            for i in prev:

                x, y = i % COLS, i // COLS

                if 0 < x < COLS - 1 and 0 < y < ROWS - 1:

                    for o in _NEIGHBOR_OFFS:

                        j = i + o

                        counts[j] = cget(j, 0) + 1

                else:

                    for ny in (y - 1, y, y + 1):

                        for nx in (x - 1, x, x + 1):

                            if (nx != x or ny != y) and 0 <= nx < COLS and 0 <= ny < ROWS:

                                j = ny * COLS + nx

                                counts[j] = cget(j, 0) + 1

            counted = counts.items()

        nxt: dict[int, int] = {}

        changed: set[int] = set()

        pget = prev.get

        for i, n in counted:

            a = pget(i)

            if a is None:

                if n == 3:

                    nxt[i] = 0

                    changed.add(i)

            elif n == 2 or n == 3:

                nxt[i] = a + 1

                # Repaint a survivor only when it crosses a band boundary.

                if a + 1 in _STEP_SET:

                    changed.add(i)

        # Commands land on the finished frame, toggled by what the clicker

        # saw (``prev``): a live cell is erased, a dead cell is born. This

        # keeps a lone placement visible for the frame it lands on — run

        # before Conway, it would die of loneliness before ever rendering.

        # A stamp plants its shape only on a dead cell, so a tap on a live

        # cell can never resurrect it. Erasures fall out of the death diff.

        for _visitor, x, y, stamp in pending:

            i = y * COLS + x

            if stamp and i not in prev:

                for dx, dy in STAMPABLE[stamp]:

                    j = _idx(x + dx, y + dy)

                    if j is not None:

                        nxt[j] = 0

                        changed.add(j)

            elif i in prev:

                nxt.pop(i, None)

            else:

                nxt[i] = 0

                changed.add(i)

        if not nxt:

            # Never leave a dead void on screen — reseed the demo.

            self._seed()

            nxt = dict(self.cells)

        # Deaths and erasures fall out of the diff; reseeded cells count too.

        changed |= (prev.keys() - nxt.keys()) | (nxt.keys() - prev.keys())

        self.cells = nxt

        self.generation += 1

        changed_chunks = frozenset(_chunk_of(i) for i in changed)

        for c in changed_chunks:

            self.chunk_rev[c] += 1

        return Frame(self.generation, frozenset(changed), changed_chunks, self.last_applied)

"""Conway's Game of Life demo mounted at ``/apps/life`` on the docs site.

One shared 256x256 board, one 5 Hz tick. The board renders as 16x16

chunks. Each tick publishes the changed chunks once; every stream patches

only the chunks inside its own viewport, so a client paints a screenful

of the world instead of the whole board. Scrolling moves the viewport —

the ``/view`` command sends catch-up patches for newly visible chunks.

"""

import asyncio

import contextlib

import time

import uuid

from pathlib import Path

from stario import App, Assets, Context, Relay, Route, Span, Writer, h

from stario.datastar import SSE, at, data, read_signals

from stario.exceptions import StarioError

from stario.markup import HtmlElement, SafeString, baked, render

from app.assets import BUNNY_FONTS_HREF, FAVICON_32_URL, STARIO_LOGO_URL

from .game import (

    CELLS,

    CHUNK,

    CHUNK_COLS,

    CHUNK_ROWS,

    CHUNKS,

    COLS,

    TICK_HZ,

    TICK_SECONDS,

    Game,

    _chunk_of,

    band,

)

MOUNT = "/apps/life"

ASSETS = Assets(Path(__file__).parent / "static", url_prefix=MOUNT + "/static")

STYLE_CSS = ASSETS.href("css/style.css")

DATASTAR_JS = ASSETS.href("js/datastar.js")

LIFE_JS = ASSETS.href("js/life.js")

HOME = Route("GET", MOUNT)

LIVE = Route("GET", MOUNT + "/live")

DROP = Route("POST", MOUNT + "/drop")

VIEW = Route("POST", MOUNT + "/view")

# Every stream subscribes to its own subject: the tick loop publishes one

# combined patch (stats + catch-up + changed visible chunks) per visitor.

V_PREFIX = "life.v."

CELL_PX = 11

# 16 cells of 11 px plus the 1 px gridlines around them -> 192 px pitch.

CHUNK_PX = CHUNK * (CELL_PX + 1)

_CELL_SPAN = CELL_PX + 1

# Last valid board-content pixel: 16 chunks of 191 px plus 15 gaps.

_BOARD_LAST = CHUNK_COLS * CHUNK_PX - 2

TIP_TEXT = (

    "One shared Conway board. Click or drag stamps a plus — dead cells "

    "bloom into evolving ornaments, live cells are erased. On touch "

    "screens one finger paints, two fingers pan the board. Moves land "

    "on the next tick: one frame every 200 ms, computed once, broadcast to "

    "every viewer. Cells glow hot when born and cool to ash with age."

)

# One delegated listener set on the board. A pointer position maps to a

# cell by arithmetic, not by hit-testing: cells run 11 px with 1 px

# hairlines inside a 191 px chunk, chunks repeat every 192 px inside the

# board's 1 px border. Every board pixel — border, gridline, or a chunk

# that has not painted yet — resolves to its nearest cell, so hover and

# clicks never land on a dead spot. `el` is reserved inside Datastar

# expressions (the element owning the attribute). The ghost classes mark

# the pending command — the next tick's morph replaces them with truth.

# Every painted cell stamps a plus: strokes bloom into evolving trails and

# taps leave ornaments. Erasing still works — the server only stamps when

# the anchor was dead, so a live anchor falls through to the erase toggle.

# el._pos tracks active pointers: one finger paints, two fingers pan, and

# el._pan keeps a two-finger gesture panning until every finger lifts.

# Board gestures live in static/js/life.js (app.life). The handlers return

# the queued cell ids to post, or "" when the 90 ms batch window has not

# elapsed — the expression posts only on a non-empty string.

_DROP_POST = f"@post(`{DROP.href()}?cells=${{c}}&stamp=plus`)"

_DOWN_EXPR = f"const c = app.life.down(el, evt); c && {_DROP_POST}"

_DRAG_EXPR = f"const c = app.life.drag(el, evt); c && {_DROP_POST}"

_RESET_EXPR = f"const c = app.life.reset(el, evt); c && {_DROP_POST}"

_SCROLL_EXPR = f"[$vx0,$vy0,$vx1,$vy1] = app.life.view(el), @post(`{VIEW.href()}`)"

_RESIZE_EXPR = f"[$vx1,$vy1] = app.life.fit($vx0, $vy0), @post(`{VIEW.href()}`)"

# Board markup is assembled as strings — cells are too numerous for element

# trees. Per-index prefixes are computed once at import. #board itself is

# rendered once per page and never morphed, so its listeners bind once.

_BOARD_ATTRS = (

    data.on("pointerdown", _DOWN_EXPR).rendered

    + data.on("pointermove", _DRAG_EXPR, throttle="15ms").rendered

    + data.on("pointerup", _RESET_EXPR, target="window").rendered

    + data.on("pointerleave", _RESET_EXPR).rendered

    + data.on("pointercancel", _RESET_EXPR).rendered

    + data.on("contextmenu", "evt.preventDefault()").rendered

)

# Board geometry for app.life — cells run CELL_PX with 1 px hairlines inside

# a CHUNK_PX chunk; every board pixel resolves to its nearest cell.

_BOARD_OPEN = (

    f'<div id="board" class="board"{_BOARD_ATTRS}'

    f' data-cells="{CELLS}" data-cols="{COLS}" data-chunk="{CHUNK}"'

    f' data-chunk-cols="{CHUNK_COLS}" data-chunk-rows="{CHUNK_ROWS}"'

    f' data-chunk-px="{CHUNK_PX}" data-cell-span="{_CELL_SPAN}"'

    f' data-board-last="{_BOARD_LAST}" data-ghost-ms="{int(TICK_SECONDS * 3000)}">'

)

_CHUNK_PRE = tuple(f'<div id="chunk-{c}" class="chunk">' for c in range(CHUNKS))

_CELL_PRE = tuple(f'<div id="cell-{i}" class="cell' for i in range(CELLS))

_CELL_DEAD = '"></div>'

_CELL_ALIVE = tuple(f' alive a{k}"></div>' for k in range(8))

_DIV_CLOSE = "</div>"

_BOARD_SHELL = _BOARD_OPEN + "".join(p + _DIV_CLOSE for p in _CHUNK_PRE) + _DIV_CLOSE

def chunk_html(game: Game, c: int) -> str:

    """One 16x16 chunk as an outer-morph payload for ``#chunk-{c}``.

    Cached by content revision: ``game.chunk_rev[c]`` bumps only when the

    chunk's cells change, so catch-up paints over static territory reuse

    the memoized HTML instead of rendering 256 cells again.

    """

    if game.chunk_seen[c] == game.chunk_rev[c]:

        html = game.chunk_cache[c]

        if html is not None:

            return html

    cx0, cy0 = (c % CHUNK_COLS) * CHUNK, (c // CHUNK_COLS) * CHUNK

    get = game.cells.get

    parts = [_CHUNK_PRE[c]]

    append = parts.append

    for y in range(cy0, cy0 + CHUNK):

        base = y * COLS

        for x in range(cx0, cx0 + CHUNK):

            age = get(base + x)

            i = base + x

            append(_CELL_PRE[i])

            append(_CELL_DEAD if age is None else _CELL_ALIVE[band(age)])

    parts.append(_DIV_CLOSE)

    html = "".join(parts)

    game.chunk_cache[c] = html

    game.chunk_seen[c] = game.chunk_rev[c]

    return html

@baked

def page(body: HtmlElement) -> HtmlElement:

    return h.HtmlDocument(

        {"lang": "en"},

        h.Head(

            h.Meta({"charset": "UTF-8"}),

            h.Meta({"name": "viewport", "content": "width=device-width, initial-scale=1"}),

            h.Meta(

                {

                    "name": "description",

                    "content": (

                        "One shared Game of Life board, computed once per tick and broadcast to everyone watching."

                    ),

                }

            ),

            h.Title("Life — Stario"),

            h.Link({"rel": "icon", "type": "image/png", "href": FAVICON_32_URL}),

            h.Link({"rel": "stylesheet", "href": BUNNY_FONTS_HREF}),

            h.Link({"rel": "stylesheet", "href": STYLE_CSS}),

            h.Script({"src": LIFE_JS, "defer": True}),

            h.Script({"type": "module", "src": DATASTAR_JS}),

        ),

        h.Body(body),

    )

def stats_view(game: Game) -> HtmlElement:

    """The ``#life-stats`` element — re-rendered and morphed on every tick.

    The connection dot binds to ``$live`` — set by ``data-indicator`` while

    the stream request is in flight — so the shared payload still tracks

    each client's own connection.

    """

    return h.Div(

        {"id": "life-stats", "class": "stats"},

        h.Span(

            {"class": "conn", "title": "stream connection"},

            data.class_("on", "$live"),

        ),

        h.Span(

            {"class": "stat"},

            h.B(str(len(game.players))),

            " playing",

        ),

        h.Span({"class": "sep", "aria-hidden": "true"}, "·"),

        h.Span({"class": "stat"}, h.B(str(len(game.cells))), " alive"),

        h.Span({"class": "sep", "aria-hidden": "true"}, "·"),

        h.Span({"class": "stat muted"}, f"{TICK_HZ} Hz"),

        h.Span({"class": "sep", "aria-hidden": "true"}, "·"),

        h.Span({"class": "tip", "title": TIP_TEXT}, "?"),

    )

def topbar_view(game: Game) -> HtmlElement:

    return h.Header(

        {"class": "topbar"},

        h.A(

            {"href": "/", "class": "brand", "aria-label": "Stario home"},

            h.Img(

                {

                    "src": STARIO_LOGO_URL,

                    "alt": "",

                    "width": "20",

                    "height": "20",

                    "decoding": "async",

                }

            ),

            h.Span({"class": "brand-name"}, "Stario"),

        ),

        h.Span({"class": "crumb", "aria-hidden": "true"}, "/"),

        h.Span({"class": "app-name"}, "Life"),

        stats_view(game),

    )

def home_view(visitor: str, game: Game) -> HtmlElement:

    return page(

        h.Div(

            {"id": "life"},

            data.signals(

                {

                    "life_id": visitor,

                    "vx0": 0,

                    "vy0": 0,

                    "vx1": 7,

                    "vy1": 5,

                },

                if_missing=True,

            ),

            data.indicator("live"),

            data.init("[$vx1,$vy1] = app.life.size(), " + at.get(LIVE.href(), retry="always")),

            topbar_view(game),

            h.Main(

                {"class": "stage"},

                data.on("scroll", _SCROLL_EXPR, throttle="120ms"),

                data.on("resize", _RESIZE_EXPR, target="window", throttle="200ms"),

                SafeString(_BOARD_SHELL),

            ),

        )

    )

def home(game: Game):

    async def handler(c: Context, w: Writer) -> None:

        w.html(home_view(str(uuid.uuid4())[:8], game))

    return handler

# ``w.write()`` never blocks — a client that vanishes without closing TCP

# just keeps buffering every tick's patch in the transport, and the process

# swells until the health probe times out (the 36 GB incident). ``drain()``

# waits until the client took the bytes; a stream that cannot drain in this

# window is dead for a 5 Hz board, so aborting is correct — the finally in

# live() then drops its subscription, player, viewport, and catch-up.

_DRAIN_TIMEOUT_S = 10.0

async def _flush(sse: SSE, w: Writer, html: str) -> bool:

    """Patch the client, bounded. ``False`` = stream is dead; caller returns."""

    sse.patch_elements(html)

    try:

        await asyncio.wait_for(w.drain(), _DRAIN_TIMEOUT_S)

    except TimeoutError, ConnectionError:

        w.abort()

        return False

    return not w.closing

def _visitor_id(signals: dict) -> str | None:

    """The per-tab visitor id, or ``None`` when the signal is forged.

    Only the shape matters: the id becomes one Relay subject segment in

    ``life.v.<id>``, so dots and wildcards are rejected — a bad publish

    would kill the tick loop for every viewer.

    """

    v = signals.get("life_id")

    if not isinstance(v, str) or not 1 <= len(v) <= 64:

        return None

    return v if "." not in v and "*" not in v else None

def _viewport(signals: dict) -> tuple[int, int, int, int] | None:

    """Chunk range ``(cx0, cy0, cx1, cy1)`` from client signals, unclamped."""

    try:

        return (

            int(signals.get("vx0", 0)),

            int(signals.get("vy0", 0)),

            int(signals.get("vx1", 7)),

            int(signals.get("vy1", 5)),

        )

    except TypeError, ValueError:

        return None

def live(game: Game, relay: Relay[str]):

    async def handler(c: Context, w: Writer) -> None:

        signals = await read_signals(c.req)

        visitor = _visitor_id(signals)

        vp = _viewport(signals)

        if visitor is None or vp is None:

            # A datastar GET: navigate reloads the page and mints a fresh id.

            SSE(w).navigate(HOME.href())

            return

        # leave() must run even when a write fails — otherwise the player,

        # viewport, rendered, and catch-up entries of a dead stream linger.

        try:

            async with relay.subscribe(V_PREFIX + visitor) as sub:

                sse = SSE(w)

                game.join(visitor)

                # Catch-up: repaint stats and every chunk in the viewport now so

                # the window between the GET and this subscription is not stale.

                # Chunks a raced /view buffered for this visitor paint too.

                needed = game.view(visitor, *vp) | game.catchup.pop(visitor, set())

                catch_up = render(stats_view(game)) + "".join(chunk_html(game, ch) for ch in sorted(needed))

                if not await _flush(sse, w, catch_up):

                    return

                async for msg in c.alive(sub):

                    if not await _flush(sse, w, msg[1]):

                        return

        finally:

            game.leave(visitor)

    return handler

def drop(game: Game):

    async def handler(c: Context, w: Writer) -> None:

        signals = await read_signals(c.req)

        visitor = _visitor_id(signals)

        if visitor is None:

            # Forged signal — an HTTP redirect is a dead end for a datastar

            # action, so navigate the client to mint a fresh id.

            SSE(w).navigate(HOME.href())

            return

        # Batched paints arrive as one comma list per ~90ms window; a lone

        # click still uses ``cell`` — either way each id is validated.

        raw = c.req.query.get("cells") or c.req.query.get("cell") or ""

        try:

            cells = [int(n) for n in raw.split(",")][:256]

        except ValueError:

            cells = []

        if not cells or any(not 0 <= cell < CELLS for cell in cells):

            w.text("unknown cell", status=400)

            return

        stamp = c.req.query.get("stamp")

        if stamp is not None and stamp != "plus":

            w.text("unknown stamp", status=400)

            return

        w.empty(204)

        for cell in cells:

            game.queue(visitor, cell % COLS, cell // COLS, stamp)

    return handler

def view(game: Game, relay: Relay[str]):

    """Viewport update: always 204. Newly visible chunks are published to

    the visitor's own subject and arrive on their stream.

    """

    async def handler(c: Context, w: Writer) -> None:

        signals = await read_signals(c.req)

        visitor = _visitor_id(signals)

        if visitor is None:

            SSE(w).navigate(HOME.href())

            return

        vp = _viewport(signals)

        if vp is None:

            w.text("bad viewport", status=400)

            return

        if visitor not in game.players:

            # Stream not subscribed yet (or mid-reconnect): join() paints

            # the viewport from the live GET's own signals anyway, so a

            # reload here would only wipe the client's state.

            w.empty()

            return

        needed = game.view(visitor, *vp)

        if needed:

            relay.publish(

                V_PREFIX + visitor,

                "".join(chunk_html(game, ch) for ch in sorted(needed)),

            )

            # This visitor is a player, so its stream is subscribed and the

            # publish covered these chunks. game.view also buffered them for

            # the subscribe race — drop that or the next tick resends them.

            buffered = game.catchup.get(visitor)

            if buffered is not None:

                buffered -= needed

        w.empty()

    return handler

def broadcast(game: Game, relay: Relay[str], state: dict) -> None:

    """One tick of fanout, shared by the loop and the benchmark.

    Stats re-render only when the counts change. Full chunks render only

    for catch-up (slots that were never painted); steady-state updates are

    cell-level morphs — ~80 bytes per changed cell vs ~14KB per chunk.

    Viewers whose patch would be identical share one built payload.

    """

    frame = game.step()

    stats_key = (len(game.players), len(game.cells))

    fresh_stats = state.get("stats_key") != stats_key

    if fresh_stats:

        state["stats_key"] = stats_key

        state["stats"] = render(stats_view(game))

    stats: str = state["stats"]

    rendered = game.rendered

    # Changed cells grouped by chunk, so a viewer's intersection is cheap.

    by_chunk: dict[int, list[int]] = {}

    for i in frame.changed:

        by_chunk.setdefault(_chunk_of(i), []).append(i)

    # One blob per changed chunk — its cells' morphs joined once, then

    # concatenated into every viewport that sees that chunk.

    cell_blob: dict[int, str] = {}

    catchup_html: dict[int, str] = {}

    get_age = game.cells.get

    def blob(c: int) -> str:

        html = cell_blob.get(c)

        if html is None:

            html = cell_blob[c] = "".join(

                _CELL_PRE[i] + (_CELL_DEAD if (a := get_age(i)) is None else _CELL_ALIVE[band(a)]) + _DIV_CLOSE

                for i in by_chunk[c]

            )

        return html

    # Viewers sharing a viewport get the same cells — group by rendered set

    # so the cell payload builds once per distinct viewport, not per user.

    groups: dict[frozenset[int], list[str]] = {}

    for vid, visible in list(rendered.items()):

        groups.setdefault(visible, []).append(vid)

    for visible, vids in groups.items():

        base = "".join(blob(c) for c in visible if c in by_chunk)

        for vid in vids:

            catchup = game.catchup.pop(vid, set())

            parts = [stats] if fresh_stats else []

            if catchup:

                for c in sorted(catchup):

                    html = catchup_html.get(c)

                    if html is None:

                        html = catchup_html[c] = chunk_html(game, c)

                    parts.append(html)

            if base:

                parts.append(base)

            if parts:

                try:

                    relay.publish(V_PREFIX + vid, "".join(parts))

                except StarioError:

                    # A bad subject must not kill the loop for everyone —

                    # drop the viewer; their next action reloads the page.

                    game.leave(vid)

async def _tick_loop(game: Game, relay: Relay[str]) -> None:

    """The metronome: step once, fan out one patch per viewer."""

    state: dict = {}

    while True:

        started = time.monotonic()

        broadcast(game, relay, state)

        await asyncio.sleep(max(0.01, TICK_SECONDS - (time.monotonic() - started)))

async def register(app: App, span: Span):

    game = Game()

    relay: Relay[str] = Relay()

    await ASSETS.attach(app)

    app.add(HOME, home(game))

    app.add(LIVE, live(game, relay))

    app.add(DROP, drop(game))

    app.add(VIEW, view(game, relay))

    # A daemon loop, not request work: keep it off `app.create_task` so

    # `drain_tasks` (tests, shutdown) never waits on it. We cancel it below.

    ticker = asyncio.create_task(_tick_loop(game, relay))

    yield

    ticker.cancel()

    with contextlib.suppress(asyncio.CancelledError):

        await ticker

window.app ??= {};

app.life = {

    g() {

        if (!this._g) {

            const d = document.getElementById("board").dataset;

            this._g = {

                cells: +d.cells,

                cols: +d.cols,

                chunk: +d.chunk,

                chunkCols: +d.chunkCols,

                chunkRows: +d.chunkRows,

                chunkPx: +d.chunkPx,

                cellSpan: +d.cellSpan,

                boardLast: +d.boardLast,

                ghostMs: +d.ghostMs,

            };

        }

        return this._g;

    },

    hit(el, cx, cy) {

        const g = this.g();

        const r = el.getBoundingClientRect();

        const bx = Math.max(0, Math.min(g.boardLast, cx - r.left - 1));

        const by = Math.max(0, Math.min(g.boardLast, cy - r.top - 1));

        return (

            (Math.min(g.chunkRows - 1, Math.floor(by / g.chunkPx)) * g.chunk +

                Math.min(g.chunk - 1, Math.floor((by % g.chunkPx) / g.cellSpan))) *

                g.cols +

            Math.min(g.chunkCols - 1, Math.floor(bx / g.chunkPx)) * g.chunk +

            Math.min(g.chunk - 1, Math.floor((bx % g.chunkPx) / g.cellSpan))

        );

    },

    mark(el, id) {

        if (el._hov === id) return;

        document.getElementById("cell-" + el._hov)?.classList.remove("sel");

        document.getElementById("cell-" + id)?.classList.add("sel");

        el._hov = id;

    },

    paint(el, id) {

        const g = this.g();

        const cell = document.getElementById("cell-" + id);

        const col = id % g.cols;

        const targets = [id];

        if (col > 0) targets.push(id - 1);

        if (col < g.cols - 1) targets.push(id + 1);

        if (id >= g.cols) targets.push(id - g.cols);

        if (id < g.cells - g.cols) targets.push(id + g.cols);

        if (cell && cell.classList.contains("alive")) {

            cell.classList.add("ghost-erase");

            setTimeout(() => cell.classList.remove("ghost-erase"), g.ghostMs);

        } else {

            for (const n of targets) {

                const t = document.getElementById("cell-" + n);

                if (t && !t.classList.contains("alive")) {

                    t.classList.add("ghost");

                    setTimeout(() => t.classList.remove("ghost"), g.ghostMs);

                }

            }

        }

        (el._q ??= {})[id] = 1;

    },

    due(el, ts) {

        if (el._qt && ts - el._qt <= 90) return "";

        el._qt = ts;

        const c = Object.keys(el._q).join(",");

        el._q = {};

        return c;

    },

    down(el, evt) {

        if (evt.button !== 0) return "";

        const pos = (el._pos ??= {});

        pos[evt.pointerId] = [evt.clientX, evt.clientY];

        if (Object.keys(pos).length >= 2) {

            el._pan = 1;

            return "";

        }

        const id = this.hit(el, evt.clientX, evt.clientY);

        if (id === el._last) return "";

        el._last = id;

        this.paint(el, id);

        return this.due(el, evt.timeStamp);

    },

    drag(el, evt) {

        let id = this.hit(el, evt.clientX, evt.clientY);

        this.mark(el, id);

        if (!(evt.buttons & 1)) return "";

        const pos = el._pos;

        if (!pos) return "";

        const ids = Object.keys(pos);

        const prev = pos[evt.pointerId];

        pos[evt.pointerId] = [evt.clientX, evt.clientY];

        if (ids.length >= 2) {

            el._pan = 1;

            if (!prev) return "";

            let ox = 0, oy = 0, nx = 0, ny = 0;

            for (const k of ids) {

                const q = k === String(evt.pointerId) ? prev : pos[k];

                ox += q[0];

                oy += q[1];

                nx += pos[k][0];

                ny += pos[k][1];

            }

            const st = el.closest(".stage");

            st.scrollLeft -= (nx - ox) / ids.length;

            st.scrollTop -= (ny - oy) / ids.length;

            return "";

        }

        if (el._pan) return "";

        let pts = evt.getCoalescedEvents ? evt.getCoalescedEvents() : [];

        if (!pts.length) pts = [evt];

        for (const p of pts) {

            id = this.hit(el, p.clientX, p.clientY);

            if (id === el._last) continue;

            el._last = id;

            this.paint(el, id);

        }

        return this.due(el, evt.timeStamp);

    },

    reset(el, evt) {

        const pos = el._pos;

        if (pos) {

            delete pos[evt.pointerId];

            if (!Object.keys(pos).length) el._pan = 0;

        }

        el._last = -1;

        if (el._hov >= 0 && (evt.pointerType === "touch" || evt.type !== "pointerup")) {

            document.getElementById("cell-" + el._hov)?.classList.remove("sel");

            el._hov = -1;

        }

        if (el._q && Object.keys(el._q).length) {

            const c = Object.keys(el._q).join(",");

            el._q = {};

            el._qt = 0;

            return c;

        }

        return "";

    },

    view(el) {

        const g = this.g();

        return [

            Math.max(0, Math.floor(el.scrollLeft / g.chunkPx)),

            Math.max(0, Math.floor(el.scrollTop / g.chunkPx)),

            Math.min(g.chunkCols - 1, Math.floor((el.scrollLeft + el.clientWidth) / g.chunkPx) + 1),

            Math.min(g.chunkRows - 1, Math.floor((el.scrollTop + el.clientHeight) / g.chunkPx) + 1),

        ];

    },

    size() {

        const g = this.g();

        return [

            Math.ceil(innerWidth / g.chunkPx) + 1,

            Math.ceil((innerHeight - 48) / g.chunkPx) + 1,

        ];

    },

    fit(x0, y0) {

        const g = this.g();

        return [

            Math.min(g.chunkCols - 1, x0 + Math.ceil(innerWidth / g.chunkPx) + 1),

            Math.min(g.chunkRows - 1, y0 + Math.ceil((innerHeight - 48) / g.chunkPx) + 1),

        ];

    },

};