Playa hardening: per-séance printing, 80mm receipts, self-healing LLM probe, Android voice

.gitignore

@@ -1,5 +1,7 @@
-# Full-resolution & print-bleed art — kept local (too big for git; add via Git LFS later)
-cards/art/
+# PNG masters and print-bleed fronts stay local — the committed archive of every
+# card is cards/art/*.jpg (q95, 4:4:4), which the tools fall back to when the
+# .png master is absent. A fresh clone can rebuild print/fronts from the JPEGs.
+cards/art/*.png
 print/fronts/
 # Ephemera
 server.log
@@ -9,3 +11,6 @@ __pycache__/
 .DS_Store
 app/models/
 app/state/
+
+# Local style tests / generation scratch
+.scratch/

app/oracle/deck.py

@@ -1,12 +1,36 @@
 """Load the deck and model the Tree spread."""
 import json
 import os
+import random
 import time
 
 REPO = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
 CARDS_PATH = os.path.join(REPO, "data", "cards.json")
 
 SPREAD_REALMS = ("roots", "trunk", "branches")
+
+# The Tree spread pulls Root / Trunk / Branch. The twelve Shell cards are the axis itself,
+# and cards.json has always promised they "can substitute into any position" — but nothing
+# ever implemented it, so a quarter of the printed deck (including The World Turtle, the
+# camp's own mascot) could never appear in a reading.
+#
+# One slot per spread may become a Shell card, at most, so it stays an event rather than a
+# flavour. At 1-in-10 a busy night sees a handful — often enough that the turtles witness
+# it, rare enough that it means something when it lands.
+SHELL_CHANCE = float(os.environ.get("ORACLE_SHELL_CHANCE", "0.10"))
+
+
+def draw_spread(by_realm, rng=random):
+    """Pure chance, one card per slot. Returns (picks, axis_slot|None).
+
+    axis_slot names the slot the Turtle's own axis spoke into, if any.
+    """
+    picks = {slot: rng.choice(by_realm[slot]) for slot in SPREAD_REALMS}
+    axis_slot = None
+    if by_realm.get("shell") and rng.random() < SHELL_CHANCE:
+        axis_slot = rng.choice(SPREAD_REALMS)
+        picks[axis_slot] = rng.choice(by_realm["shell"])
+    return picks, axis_slot
 SLOT_LABEL = {
     "roots": "what to face",
     "trunk": "where you stand",

app/oracle/llm.py

@@ -1,22 +1,40 @@
 """Local LLM adapter (Ollama over stdlib urllib). Any failure -> None, so callers fall back."""
 import json
 import os
+import time
 import urllib.request
 
+# Models that emit a <think> preamble unless told not to. The Turtle has no use for one.
+NO_THINK = ("qwen3", "deepseek", "gpt-oss", "magistral")
+
+# How long a probe result is trusted. Short enough that Ollama coming up late (a power blip
+# on playa reorders systemd units) heals on its own within a seeker or two; long enough that
+# we don't probe on every LLM touch.
+PROBE_TTL = float(os.environ.get("ORACLE_PROBE_TTL", "30"))
+
 
 class LLM:
     def __init__(self, model=None, host=None):
         self.model = model or os.environ.get("ORACLE_MODEL", "qwen2.5")
         self.host = (host or os.environ.get("OLLAMA_HOST", "http://localhost:11434")).rstrip("/")
         self._available = None
+        self._probed_at = 0.0
 
     def available(self):
-        if self._available is None:
+        """Probe Ollama, re-probing after PROBE_TTL.
+
+        Never cache the answer for the life of the process: if Ollama is down when the
+        oracle starts and comes up later, a permanently-cached False leaves the Turtle in
+        template mode until a human restarts it — and on playa nobody is at a keyboard.
+        """
+        now = time.monotonic()
+        if self._available is None or (now - self._probed_at) > PROBE_TTL:
             try:
                 with urllib.request.urlopen(self.host + "/api/tags", timeout=1.5) as r:
                     self._available = r.status == 200
             except Exception:
                 self._available = False
+            self._probed_at = now
         return self._available
 
     def generate(self, prompt, system=None, timeout=90, as_json=False):
@@ -27,7 +45,7 @@ def generate(self, prompt, system=None, timeout=90, as_json=False):
             "options": {"temperature": 0.75},
             "keep_alive": -1,  # stay resident: no 20s reload between seekers
         }
-        if self.model.startswith(("qwen3", "deepseek")):
+        if self.model.startswith(NO_THINK):
             body["think"] = False
         if system:
             body["system"] = system

app/oracle/printer.py

@@ -1,8 +1,14 @@
-"""Thermal receipt: format the reading + quest for a 58mm ESC/POS printer.
+"""Thermal receipt: format the reading + quest for an ESC/POS printer.
 
-Prints to a USB ESC/POS printer if python-escpos is installed and ESCPOS_VENDOR_ID /
-ESCPOS_PRODUCT_ID are set; otherwise saves a text preview to app/receipts/ so you can see
-exactly what would print. 58mm printers are ~32 characters wide.
+The camp runs Epson TM-m30III units: 80mm paper, 48 columns in font A. Set
+ORACLE_PRINT_WIDTH to 32 for a 58mm printer.
+
+Transport, in order of preference:
+  ESCPOS_HOST            -> network printer on port 9100 (the TM-m30III's ethernet port;
+                            one cable to the camp router, nothing to kick loose)
+  ESCPOS_VENDOR_ID/_PRODUCT_ID -> USB
+  neither                -> save a text preview to app/receipts/ so you can see exactly
+                            what would have printed
 """
 import os
 import textwrap
@@ -11,7 +17,9 @@
 from .deck import REPO
 from .geo import COMPASS_ROSE, directions_lines
 
-WIDTH = 32
+# 80mm / font A = 48 cols on the TM-m30III; 58mm printers are 32.
+WIDTH = int(os.environ.get("ORACLE_PRINT_WIDTH", "48"))
+PROFILE = os.environ.get("ESCPOS_PROFILE", "TM-m30")
 RECEIPTS = os.path.join(REPO, "app", "receipts")
 
 
@@ -27,6 +35,17 @@ def _wrap(s):
     return textwrap.wrap(s, WIDTH) or [""]
 
 
+def _center_block(block):
+    """Centre a fixed-width ASCII block (the compass rose) inside WIDTH.
+
+    The rose is hand-drawn at <=32 cols; on 80mm paper it would otherwise hug the
+    left edge while everything around it fills the width.
+    """
+    lines = block.split("\n")
+    pad = max(0, (WIDTH - max(len(l) for l in lines)) // 2)
+    return "\n".join(" " * pad + l for l in lines)
+
+
 def format_receipt(payload, picks, located, quest=None):
     L = []
     L.append(_center("* THE TERRIBLE TURTLE *"))
@@ -43,7 +62,10 @@ def format_receipt(payload, picks, located, quest=None):
     labels = {"roots": "FACE", "trunk": "STAND", "branches": "REACH"}
     for realm in ("roots", "trunk", "branches"):
         c = picks[realm]
-        L.extend(_wrap(f"[{labels[realm]}] {c['name']}"))
+        # a Shell card standing in a Tree slot is the axis; say so on the paper too, so
+        # the seeker still has the evidence of it days later
+        label = "* AXIS *" if c.get("realm") == "shell" else labels[realm]
+        L.extend(_wrap(f"[{label}] {c['name']}"))
     L.append(_rule())
     L.append("")
     L.append("THE READING")
@@ -78,7 +100,7 @@ def format_receipt(payload, picks, located, quest=None):
         L.append("")
     L.append(_rule())
     L.append("WHERE TO GO")
-    L.append(COMPASS_ROSE)
+    L.append(_center_block(COMPASS_ROSE))
     L.append("")
     for line in directions_lines(picks, located):
         for w in _wrap("> " + line):
@@ -91,14 +113,30 @@ def format_receipt(payload, picks, located, quest=None):
     return "\n".join(L)
 
 
-def print_or_preview(text):
-    """Try the USB printer; fall back to saving a preview file. Returns a status dict."""
+def print_or_preview(text, host=None):
+    """Print to the network or USB printer; fall back to a preview file. Returns a status dict.
+
+    host overrides ESCPOS_HOST so a station can be bound to its own printer.
+    """
+    host = host or os.environ.get("ESCPOS_HOST")
     vid = os.environ.get("ESCPOS_VENDOR_ID")
     pid = os.environ.get("ESCPOS_PRODUCT_ID")
+    if host:
+        try:
+            from escpos.printer import Network  # requires: pip install python-escpos
+            p = Network(host, port=int(os.environ.get("ESCPOS_PORT", "9100")),
+                        timeout=10, profile=PROFILE)
+            p.text(text + "\n")
+            p.cut()
+            p.close()
+            return {"status": "printed", "target": f"net {host}"}
+        except Exception as e:  # noqa: BLE001
+            preview = _save_preview(text)
+            return {"status": "preview", "path": preview, "error": f"printer error: {e}"}
     if vid and pid:
         try:
             from escpos.printer import Usb  # requires: pip install python-escpos pyusb
-            p = Usb(int(vid, 16), int(pid, 16), profile="TM-T88III")
+            p = Usb(int(vid, 16), int(pid, 16), profile=PROFILE)
             p.text(text + "\n")
             p.cut()
             return {"status": "printed", "target": f"usb {vid}:{pid}"}
@@ -107,7 +145,8 @@ def print_or_preview(text):
             return {"status": "preview", "path": preview, "error": f"printer error: {e}"}
     preview = _save_preview(text)
     return {"status": "preview", "path": preview,
-            "note": "No ESCPOS_VENDOR_ID/PRODUCT_ID set — saved a preview instead of printing."}
+            "note": "No ESCPOS_HOST or ESCPOS_VENDOR_ID/PRODUCT_ID set — saved a preview "
+                    "instead of printing."}
 
 
 def _save_preview(text):

app/oracle/server.py

@@ -27,7 +27,30 @@
 LLM_SINGLETON = LLM()
 ART_RE = re.compile(r"^(shell|roots|trunk|branches)-\d{2}\.png$")
 WEBIMG_RE = re.compile(r"^((shell|roots|trunk|branches)-\d{2}|back)\.jpg$")
-LAST = {}  # last built reading (kiosk = one seeker at a time), for /api/print
+
+# Legacy single-question flow (/api/reading, app/web/index.html) only. The séance
+# (/api/session/*) never touches this — its state is per-session, because two tablets
+# plus phones on the camp network mean several seekers are mid-reading at once and a
+# shared "last reading" prints one person's quest for another.
+LAST = {}
+LAST_LOCK = threading.Lock()
+
+# Per-station printer binding: ESCPOS_HOST_1 / _2 pick a printer for kiosk=1 / kiosk=2.
+def _printer_host(kiosk):
+    return os.environ.get(f"ESCPOS_HOST_{kiosk}") if kiosk else None
+
+
+# The fallback weave is good enough that a dead model reads as a working oracle — the
+# séance completes, the reading is real, nobody notices. On playa that could go a whole
+# night. Count which tier actually served each reading so a camp turtle can check.
+TIERS = {"llm": 0, "fallback": 0}
+TIERS_LOCK = threading.Lock()
+
+
+def note_tier(mode):
+    with TIERS_LOCK:
+        if mode in TIERS:
+            TIERS[mode] += 1
 
 
 def build_reading(question):
@@ -44,11 +67,27 @@ def build_reading(question):
         "directions": directions_lines(picks, located),
         "modes": {"select": sel_mode, "weave": weave_mode},
     }
-    LAST.clear()
-    LAST.update({"payload": payload, "picks": picks, "located": located})
+    with LAST_LOCK:
+        LAST.clear()
+        LAST.update({"payload": payload, "picks": picks, "located": located})
     return payload
 
 
+def receipt_for_session(sid):
+    """Format the receipt for one séance, from that séance's own state."""
+    sess = session.snapshot(sid)
+    if not sess:
+        return None
+    payload = {
+        "question": " / ".join(sess["shares"]),
+        "reading": sess["reading"],
+        "adventure": sess["adventure"],
+        "name": sess.get("name"),
+    }
+    return printer.format_receipt(payload, sess["picks"], sess["located"],
+                                  quest=sess.get("quest"))
+
+
 REALM_ORDER = {"shell": 0, "roots": 1, "trunk": 2, "branches": 3}
 
 
@@ -122,6 +161,23 @@ def do_GET(self):
             return self._send(200, all_cards_payload())
         if path == "/api/lore":
             return self._send(200, lore.counts())
+        if path == "/api/health":
+            with TIERS_LOCK:
+                tiers = dict(TIERS)
+            total = tiers["llm"] + tiers["fallback"]
+            return self._send(200, {
+                "llm_reachable": LLM_SINGLETON.available(),
+                "model": LLM_SINGLETON.model,
+                "ears": ears.available(),
+                "readings": tiers,
+                # the number to look at: if this is climbing, the Turtle has gone dumb
+                # and is hiding it behind a very convincing template
+                "fallback_pct": round(100.0 * tiers["fallback"] / total, 1) if total else None,
+                "live_seances": len(session.SESSIONS),
+                "printer": ("network" if os.environ.get("ESCPOS_HOST") or
+                            os.environ.get("ESCPOS_HOST_1")
+                            else "usb" if os.environ.get("ESCPOS_VENDOR_ID") else "preview-only"),
+            })
         if path.startswith("/thumb/") or path.startswith("/med/"):
             sub = "thumb" if path.startswith("/thumb/") else "med"
             name = os.path.basename(path)
@@ -162,11 +218,24 @@ def do_POST(self):
             except Exception as e:
                 return self._send(500, {"error": str(e)})
         if path == "/api/print":
-            if not LAST:
-                return self._send(400, {"error": "no reading to print yet"})
-            text = printer.format_receipt(LAST["payload"], LAST["picks"], LAST["located"],
-                                          quest=LAST.get("quest"))
-            result = printer.print_or_preview(text)
+            try:
+                body = json.loads(raw or b"{}")
+            except Exception:
+                body = {}
+            sid = (body.get("session") or "").strip()
+            if sid:
+                text = receipt_for_session(sid)
+                if text is None:
+                    return self._send(400, {"error": "no such séance to print"})
+            else:
+                # legacy /api/reading flow
+                with LAST_LOCK:
+                    if not LAST:
+                        return self._send(400, {"error": "no reading to print yet"})
+                    snap = dict(LAST)
+                text = printer.format_receipt(snap["payload"], snap["picks"], snap["located"],
+                                              quest=snap.get("quest"))
+            result = printer.print_or_preview(text, host=_printer_host(body.get("kiosk")))
             result["receipt"] = text
             return self._send(200, result)
         if path == "/api/transcribe":
@@ -191,22 +260,13 @@ def do_POST(self):
                     return self._send(200, session.start(mode))
                 sid = (body.get("session") or "").strip()
                 if action == "say":
-                    return self._send(200, session.hear(sid, body, LLM_SINGLETON))
-                if action == "accept":
-                    event = session.accept(sid, LLM_SINGLETON)
-                    sess = session.snapshot(sid)
-                    if sess and sess.get("quest"):
-                        # stage the sealed quest for /api/print
-                        told = " / ".join(sess["shares"])
-                        LAST.clear()
-                        LAST.update({
-                            "payload": {"question": told, "reading": sess["reading"],
-                                        "adventure": sess["adventure"],
-                                        "name": sess.get("name")},
-                            "picks": sess["picks"], "located": sess["located"],
-                            "quest": sess["quest"],
-                        })
+                    event = session.hear(sid, body, LLM_SINGLETON)
+                    note_tier((event.get("modes") or {}).get("weave"))
                     return self._send(200, event)
+                if action == "accept":
+                    # The sealed quest stays on the session; /api/print reads it back by
+                    # session id. Nothing is staged in shared state.
+                    return self._send(200, session.accept(sid, LLM_SINGLETON))
             except Exception as e:
                 return self._send(500, {"error": str(e)})
             return self._send(404, {"error": "unknown séance action"})