"""WardenPresence: one brain, one queue, every screen. The game has exactly one LLM and a CPU latency budget of ~2s per line, so every screen talks to the Warden through this service instead of calling the voice directly. The rules: - One request in flight, highest priority first. - Everything carries a scripted fallback; the game never goes mute or stalls because the model is slow, absent, or rude. - Lines go stale: if the player has moved on, the Warden does not say it. - Ambient/quip chatter is rate-limited. Scarcity is the aesthetic — a Warden that comments on everything is a chatbot; one that goes quiet and then proves it was watching is an antagonist. - prefetch() is speculative pregeneration: generate likely lines during the player's thinking time, deliver instantly on a key match. Delivery is a single sink (the active screen). Screens attach on mount; a line whose moment has passed dies quietly with its TTL. """ from __future__ import annotations import asyncio import heapq import itertools import time from dataclasses import dataclass, field from typing import Callable # Priorities. ANSWER never goes stale — the player spoke and is waiting. AMBIENT, QUIP, REACTION, ANSWER = 0, 1, 2, 3 TTL = {AMBIENT: 30.0, QUIP: 20.0, REACTION: 12.0, ANSWER: float("inf")} # Minimum seconds between delivered lines of low-priority chatter. MIN_GAP = {AMBIENT: 45.0, QUIP: 15.0} _seq = itertools.count() @dataclass(order=True) class _Request: sort_key: tuple[int, int] moment: str = field(compare=False) fallback: str = field(compare=False) priority: int = field(compare=False) digest: str = field(compare=False) tags: set = field(compare=False) key: str | None = field(compare=False) born: float = field(compare=False) taboo: str = field(compare=False, default="") class WardenPresence: def __init__(self, voice=None, *, clock: Callable[[], float] = time.monotonic): self.voice = voice # WardenVoice | None (None = scripted fallbacks only) self.clock = clock self._heap: list[_Request] = [] self._wake = asyncio.Event() self._task: asyncio.Task | None = None self._sink: Callable[[str, int], None] | None = None self._cache: dict[str, str] = {} self._prefetching: set[str] = set() self._last_spoke: dict[int, float] = {} # ------------------------------------------------------------ wiring def attach(self, sink: Callable[[str, int], None]) -> None: """The active screen claims the Warden's mouth.""" self._sink = sink def detach(self, sink: Callable[[str, int], None]) -> None: if self._sink is sink: self._sink = None def _ensure_worker(self) -> None: if self._task is None or self._task.done(): self._task = asyncio.get_event_loop().create_task(self._run()) def stop(self) -> None: if self._task is not None: self._task.cancel() self._task = None # ----------------------------------------------------------- submit def submit( self, moment: str, *, fallback: str = "", priority: int = QUIP, digest: str = "", tags: set | None = None, key: str | None = None, taboo: str = "", ) -> None: """Queue a thing worth reacting to. Returns immediately. taboo: raw player text the line must never parrot back.""" now = self.clock() gap = MIN_GAP.get(priority) if gap is not None and now - self._last_spoke.get(priority, -1e9) < gap: return # too chatty; let the silence work heapq.heappush( self._heap, _Request( sort_key=(-priority, next(_seq)), moment=moment, fallback=fallback, priority=priority, digest=digest, tags=tags or set(), key=key, born=now, taboo=taboo, ), ) self._wake.set() self._ensure_worker() def prefetch(self, key: str, moment: str, *, digest: str = "", tags: set | None = None) -> None: """Speculatively generate a line now; a later submit() with the same key delivers it instantly. Misses cost idle CPU nobody was using.""" if self.voice is None or key in self._cache or key in self._prefetching: return self._prefetching.add(key) asyncio.get_event_loop().create_task(self._prefetch(key, moment, digest, tags or set())) async def _prefetch(self, key: str, moment: str, digest: str, tags: set) -> None: try: async for line in self.voice.react(moment, digest=digest, tags=tags): self._cache[key] = line finally: self._prefetching.discard(key) def clear_cache(self) -> None: """New fight, new context: speculative lines from the old one die.""" self._cache.clear() # ------------------------------------------------------------ worker async def _run(self) -> None: while True: while not self._heap: self._wake.clear() await self._wake.wait() req = heapq.heappop(self._heap) if self.clock() - req.born > TTL[req.priority]: continue # the moment passed; saying it now would be lame try: line = await self._line_for(req) except Exception: line = req.fallback # one bad call never kills the queue if line and self._sink is not None: self._last_spoke[req.priority] = self.clock() self._sink(line, req.priority) async def _line_for(self, req: _Request) -> str: if req.key is not None and req.key in self._cache: return self._cache.pop(req.key) if self.voice is None: return req.fallback got = "" async for line in self.voice.react( req.moment, digest=req.digest, tags=req.tags, taboo=req.taboo ): got = line return got or req.fallback