scrypt/warden/presence.py

"""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