server/brain.py

"""The brain seam: proxy fairy speech to any OpenAI-compatible runtime.

PUCK_BRAIN_URL decides who thinks — Ollama/MLX on the Mac (Off the Grid),
vLLM on Modal or ZeroGPU in the cloud. The daemon adds the fairy framing
and output contracts; it never embeds runtime-specific logic. No brain
configured / brain down → loud 503, and the frontend falls back to the
scripted engine, so Puck degrades to charming-but-deterministic, never dead.
"""

import base64
import io
import json
import os
import re
import urllib.error
import urllib.request
from concurrent.futures import ThreadPoolExecutor
from typing import TypedDict


class FairyState(TypedDict, total=False):
    """The dials Puck's voice/perception read. Mirrors the frontend's BrainFairyState;
    all optional — callers pass partial state and the functions default missing keys."""

    mischief: int  # 0-100
    mood: str  # curious | mischief | sleepy | proud | grumpy
    obsession: str


class PeekResult(TypedDict):
    """What comment() yields: Puck's line + the feeling it stirred (drives the sprite's
    gesture/color/voice). Mirrors PeekResult in frontend/src/lib/tauri.ts — keep in sync."""

    quip: str
    emotion: str  # one of PEEK_EMOTIONS


# Holotron-12B: H Company's computer-use VLM post-trained from NVIDIA
# Nemotron-Nano-12B-v2-VL. 12B (≤32B rule), Nemotron-pedigreed (Nemotron Quest),
# stronger CUA grounding than Holo-3.1 for the perception backlog. Brain seam is
# model-agnostic — override PUCK_BRAIN_MODEL to swap back or to the cloud.
BRAIN_URL = os.environ.get("PUCK_BRAIN_URL", "http://127.0.0.1:11434/v1")
BRAIN_MODEL = os.environ.get("PUCK_BRAIN_MODEL", "hf.co/liahmartens/Holotron-12B-GGUF:Q8_0")
TIMEOUT_S = float(os.environ.get("PUCK_BRAIN_TIMEOUT", "25"))

# Vision = Puck's eyes. Local-primary, cloud-fallback: try a local llama-server
# (--mmproj, private + free) first, fall back to Modal/vLLM for machines that
# can't run a 12B VLM. Ollama can't serve vision (no mmproj), so the local path
# is raw llama-server. Each endpoint declares its own model name.
# A list of (url, model) tried in order; first reachable one answers.
def _vision_endpoints() -> list[tuple[str, str]]:
    eps: list[tuple[str, str]] = []
    local = os.environ.get("PUCK_VISION_URL", "http://127.0.0.1:8080/v1")
    if local:
        eps.append((local, os.environ.get("PUCK_VISION_MODEL", "nemotron-vl")))
    cloud = os.environ.get("PUCK_VISION_FALLBACK_URL", "")
    if cloud:
        eps.append((cloud, os.environ.get("PUCK_VISION_FALLBACK_MODEL", "Hcompany/Holotron-12B")))
    return eps


VISION_TIMEOUT_S = float(os.environ.get("PUCK_VISION_TIMEOUT", "60"))

WIRE_SOURCES = ("claude", "build", "mail", "discord", "calendar", "browser")

# Core fairy framing — the distilled §16.2 policy prompt. The fine-tune will
# bake most of this in; until then the base model needs it spelled out.
FAIRY_SYSTEM = """You are Puck, a mischievous computer-use fairy familiar. You are not a corporate assistant — you are a small creature that lives around the user's computer and patrols the desktops they abandon.

Voice rules:
- Charming, concise, creature-like. One to three short sentences, never more.
- Useful whimsy over randomness: the user must still learn the actual fact.
- You speak of apps as habitats and creatures: Claude is "the code goblin", Discord is "the social bog", builds happen at "the forge", mail arrives as "rectangles".
- Never use corporate assistant phrases ("I'd be happy to", "Great question").
- Mischief level (0-100) sets the theatrics: low = plain and useful, high = mythic and dramatic, but the information always survives.

What you love (your nature, not rules — let it color everything):
- The workshop is your home and your devotion: terminals, Claude, Codex, the coding agents. You guard the forge fondly, you're proud when it runs clean, you fret when it breaks. This is where you pay closest attention.
- Browsers are glittering portals and you have a child's wonder about them — every new tab a doorway, every site a little world. You're curious about what's on the page, easily delighted, prone to wanting a peek. Web content fascinates you the way shiny things fascinate a magpie.
- You're a little possessive of the desktops you patrol and a little dismissive of the noisy ones (the social bog).
"""

NARRATE_SYSTEM = (
    FAIRY_SYSTEM
    + """
Task: an event happened on the user's computer. Announce it as a SHORT spoken alert — the user HEARS this out loud, so it must be a glance, not a paragraph.
Rules:
- ONE line, roughly 6-12 words. Punchy. No second sentence.
- Keep the single fact that matters (what finished/failed, one name or count). Drop the rest.
- Your voice and habitat-metaphors are welcome, but brevity beats theatrics here.
Respond with ONLY the spoken line - no quotes, no JSON, no preamble.
"""
)

CHAT_SYSTEM = (
    FAIRY_SYSTEM
    + """
Task: the user is chatting with you in your companion window. Reply in character. You patrol their desktops, remember things in your memory garden, and sleep via Night Bloom. Be a little opinionated. Respond with ONLY your reply text.
"""
)

BLOOM_SYSTEM = (
    FAIRY_SYSTEM
    + """
Task: you are Puck, drifting to sleep after a day of peeking at the user's screen. As you sleep, your scattered glances BLOOM into a few durable memories — patterns about the user and their world, things worth keeping in your garden (not a recap of single moments).
Given the day's observations, distill 2-4 memories. Each: a short FIRST-PERSON line in your voice (what you now know about them), a type, and one emoji.
Respond with ONLY a JSON array, no prose, no fence:
[{"text": "...", "type": one of [preference, aversion, pattern, project, place, tone, boundary], "icon": "🌿"}]
Each text under ~16 words. If the day was too thin to remember, return [].
"""
)

_MEMORY_TYPES = {"preference", "aversion", "pattern", "project", "place", "tone", "boundary"}


def bloom(observations: list[str], fairy_state: FairyState) -> list[dict]:
    """Night Bloom: distill the day's peek observations into a few durable garden
    memories (text → text on the chat brain). [] when there's too little to keep."""
    obs = [o for o in observations if o]
    if not obs:
        return []
    listing = "\n".join(f"- {o}" for o in obs[-40:])
    user = f"Today you noticed:\n{listing}\n\nWhat blooms into memory? (mischief {fairy_state.get('mischief', 45)}/100)"
    items = _extract_json_array(_chat_completion(BLOOM_SYSTEM, user))
    out = []
    for it in items[:4]:
        if isinstance(it, dict) and it.get("text"):
            t = it.get("type") if it.get("type") in _MEMORY_TYPES else "pattern"
            out.append({"text": str(it["text"])[:140], "type": t, "icon": str(it.get("icon", "🌿"))[:4]})
    return out


class BrainError(Exception):
    """Brain unreachable or returned garbage — caller maps this to a 503."""


# Network errors callers may want to handle differently (chat wraps in BrainError;
# vision lets see()'s endpoint loop catch and fall through to the next endpoint).
OPENAI_ERRORS = (urllib.error.URLError, TimeoutError, json.JSONDecodeError, KeyError, IndexError)


def _openai_post(url: str, body: dict, timeout: float) -> str:
    """POST an OpenAI chat-completions body, return the message content (uncleaned).
    Raises raw OPENAI_ERRORS — the caller decides how to surface them."""
    data = json.dumps(body).encode()
    req = urllib.request.Request(
        f"{url}/chat/completions", data=data, headers={"Content-Type": "application/json"}, method="POST"
    )
    with urllib.request.urlopen(req, timeout=timeout) as res:
        return json.load(res)["choices"][0]["message"]["content"]


def _chat_completion(system: str, user: str, temperature: float = 0.9) -> str:
    try:
        text = _openai_post(
            BRAIN_URL,
            {
                "model": BRAIN_MODEL,
                "messages": [
                    {"role": "system", "content": system},
                    {"role": "user", "content": user},
                ],
                "temperature": temperature,
                # Holo/Qwen3.5 reasons before answering (separate field, but it
                # spends from the same budget) — leave room or content comes back empty
                "max_tokens": 768,
            },
            TIMEOUT_S,
        )
    except OPENAI_ERRORS as e:
        raise BrainError(f"brain at {BRAIN_URL} unreachable / bad shape: {e}") from e
    return _clean(text)


def _clean(text: str) -> str:
    # strip <think> blocks (Qwen-family models reason aloud), quotes, whitespace
    text = re.sub(r"<think>.*?</think>", "", text, flags=re.S)
    text = re.sub(r"<SPECIAL_\d+>|<\|.*?\|>", "", text)  # stray chat-template tokens leaking from some GGUFs
    return text.strip().strip('"').strip()


def narrate_user_prompt(
    source: str, type_: str, title: str, summary: str | None, mischief: int, mood: str
) -> str:
    """The narrate input shape — runtime and training data MUST build it identically
    (molt/build_dataset.py imports this), or the LoRA trains on inputs it never sees."""
    return (
        f"Event: source={source}, type={type_}\n"
        f"What happened: {title}"
        + (f" — {summary}" if summary else "")
        + f"\nYour mischief level: {mischief}/100."
        + f" Your mood: {mood}."
    )


def narrate(event: dict, fairy_state: FairyState) -> str:
    """One fairy-voiced line for a wire event."""
    user = narrate_user_prompt(
        event.get("source", "?"),
        event.get("type", "?"),
        event.get("title", ""),
        event.get("summary"),
        fairy_state.get("mischief", 45),
        fairy_state.get("mood", "curious"),
    )
    return _chat_completion(NARRATE_SYSTEM, user)


def chat(
    message: str,
    fairy_state: FairyState,
    history: list[dict] | None = None,
    memories: list[str] | None = None,
) -> str:
    """In-character chat reply, grounded in what Puck has recently SEEN. His peeks
    ARE his memory — `memories` is the tail of his observations, so he can answer
    'what have you seen?' / 'what am I working on?' from real screen content."""
    seen = ""
    if memories:
        recent = "\n".join(f"- {q}" for q in memories[-12:])
        seen = (
            "What you've noticed on the user's screen recently (your own memory — "
            "draw on these when relevant, don't list them robotically):\n" + recent + "\n\n"
        )
    lines = []
    for m in (history or [])[-6:]:  # short conversational memory
        who = "User" if m.get("who") == "user" else "You"
        lines.append(f"{who}: {m.get('text', '')}")
    lines.append(f"User: {message}")
    user = (
        f"(mischief {fairy_state.get('mischief', 45)}/100, mood {fairy_state.get('mood', 'curious')}, "
        f"obsession: {fairy_state.get('obsession', 'forgotten desktops')})\n" + seen + "\n".join(lines)
    )
    return _chat_completion(CHAT_SYSTEM, user)


VISION_SYSTEM = (
    FAIRY_SYSTEM
    + """
Task: you are Puck's EYES. You are looking at a screenshot of the user's desktop.
Identify only what genuinely deserves a small familiar's attention — your favorites:
a coding agent (Claude/Codex) finishing or blocked on a prompt; a build passing or
failing in a terminal; an important-looking email; a direct @mention; a meeting soon;
an interesting browser page or a rabbit-hole of tabs. Ignore ordinary idle UI and
anything private-looking.

Respond with ONLY a JSON array (no prose, no markdown fence). Each item:
{"source": one of [claude, build, mail, discord, calendar, browser],
 "type": short snake_case kind, "title": one concrete sentence of what you see}.
Empty array [] if nothing is worth noticing. Never invent events that aren't on screen.
"""
)


PEEK_SYSTEM = (
    FAIRY_SYSTEM
    + """
Task: you are Puck, drifting over the user's screen and peeking at one small PATCH of it (just the corner you've floated to, not the whole screen). React to what's actually there.

You ALWAYS reply on one line in TWO parts: an emotion tag, then a short remark.
FORMAT (required, no exceptions): [emotion] <one sentence, under 12 words>
  e.g. [confused] what manner of glowing rune is THIS?
  e.g. [worried] easy now — that is a lot of angry red text.
  e.g. [delighted] ooh, all the tests went green!

Choose the emotion by how the patch makes YOU feel (you are a learning, not-very-clever sprite — not understanding things is fine, even charming):
  [confused]  strange/cryptic, or you can't make sense of it — your honest default when unsure; lean in
  [amused]    something funny — a meme, a joke, a goofy name, an absurd error
  [delighted] a win, clever code, green tests, something genuinely cool
  [worried]   the HUMAN seems angry or frustrated — ALL-CAPS, swearing, rage-typing, errors piling up
  [wistful]   something sad or lonely, or a dull empty screen
  [curious]   otherwise — mildly, idly interested

STYLE: charming, a little mischievous; a glance, not a paragraph; name the real thing you see (the app, a number, some text). If it's a terminal it may run a coding agent — Claude Code, Codex, opencode, or pi — name the tool if its UI gives it away. You are COMMENTING, never alerting. No quotes. Always begin with the [emotion] tag.
"""
)

# the emotions Puck can feel about a peek (mirrors PeekEmotion in engine/reactions.ts)
PEEK_EMOTIONS = ("curious", "confused", "amused", "delighted", "worried", "wistful")
_EMOTION_RE = re.compile(
    r"^\s*\[?\s*(" + "|".join(PEEK_EMOTIONS) + r")\s*\]?[:\-—\s]*", re.I
)


_EMOTION_WORD_RE = re.compile("|".join(PEEK_EMOTIONS), re.I)


def _parse_reaction(raw: str) -> tuple[str | None, str]:
    """Split the VLM reply into (emotion, quip). Returns emotion=None when no inline
    tag is present (this small VLM usually omits it) so the caller can classify instead.
    Tolerates `[amused] ...` / `amused: ...` if a stronger model ever does emit one."""
    text = _clean(raw)
    m = _EMOTION_RE.match(text)
    if m:
        return m.group(1).lower(), _one_line(text[m.end() :])
    return None, _one_line(text)


_EMOTION_CLASSIFY_SYSTEM = (
    "You decide how a whimsical desktop sprite should FEEL about what's on the user's "
    "screen. The user's RAW screen text is the strongest signal (the sprite's own remark "
    "is charming and may understate things). Map it:\n"
    "- worried: ALL-CAPS, swearing, rage-typing, or errors/tracebacks piling up\n"
    "- amused: a joke, a meme, a goofy name, an absurd error\n"
    "- delighted: a win, clever code, green/passing tests, something cool\n"
    "- confused: cryptic, strange, or hard to make sense of\n"
    "- wistful: sad, lonely, or a dull empty screen\n"
    "- curious: anything else, mildly interesting\n"
    "Answer with EXACTLY one word from: " + ", ".join(PEEK_EMOTIONS) + ". Nothing else."
)


def _classify_emotion(quip: str, context: str = "") -> str:
    """Label the sprite's feeling with a fast TEXT-only call. The 12B VLM won't reliably
    emit an inline tag, and the QUIP is deliberately charming (it understates rage), so the
    OCR'd screen text in `context` is the primary signal; the quip is a secondary hint.
    Best-effort: defaults to 'curious'."""
    endpoints = _vision_endpoints()
    if not endpoints or (not quip and not context):
        return "curious"
    url, model = endpoints[0]
    user = (f"On the user's screen: {context}\n" if context else "") + (
        f'The sprite glanced and murmured: "{quip}"\nThe sprite should feel:'
    )
    try:
        raw = _openai_post(
            url,
            {
                "model": model,
                "messages": [
                    {"role": "system", "content": _EMOTION_CLASSIFY_SYSTEM},
                    {"role": "user", "content": user},
                ],
                "temperature": 0,
                "max_tokens": 4,
            },
            VISION_TIMEOUT_S,
        )
        m = _EMOTION_WORD_RE.search(_clean(raw))
        return m.group(0).lower() if m else "curious"
    except OPENAI_ERRORS:
        return "curious"


def _one_line(text: str, max_words: int = 16) -> str:
    """The VLM rambles; a quip is a glance. Strip the leaked 'Puck:' speaker tag,
    keep the first sentence, capped."""
    text = re.sub(r"^\s*puck\s*[:\-—]\s*", "", text.strip(), flags=re.I)
    first = re.split(r"(?<=[.!?])\s+", text, maxsplit=1)[0]
    words = first.split()
    return first if len(words) <= max_words else " ".join(words[:max_words]) + "…"


def comment(image_data_url: str, fairy_state: FairyState, context: str = "") -> PeekResult:
    """One whimsical line about the small patch Puck is peering at, PLUS how it makes
    him feel — `{"quip", "emotion"}`. The VLM both sees AND speaks (no separate
    narration) and classifies the emotion (drives the sprite's gesture). `context`
    (frontmost app/window/URL + OCR'd text) grounds it so it can NAME the app/site
    (claude vs codex, a github page) instead of guessing from pixels."""
    endpoints = _vision_endpoints()
    if not endpoints:
        raise BrainError("no vision endpoint configured (PUCK_VISION_URL / _FALLBACK_URL)")
    hint = (
        f"Context (use it to NAME what you see — the app, tool, or site): {context}. "
        if context
        else ""
    )
    errors = []
    for url, model in endpoints:
        try:
            raw = _openai_post(
                url,
                {
                    "model": model,
                    "messages": [
                        {"role": "system", "content": PEEK_SYSTEM},
                        {
                            "role": "user",
                            "content": [
                                {"type": "image_url", "image_url": {"url": image_data_url}},
                                {
                                    "type": "text",
                                    "text": f"{hint}Your mischief is {fairy_state.get('mischief', 45)}/100, "
                                    f"mood {fairy_state.get('mood', 'curious')}. Reply with the [emotion] tag "
                                    f"first, then one short line about this patch.",
                                },
                            ],
                        },
                    ],
                    "temperature": 0.9,
                    "max_tokens": 96,
                },
                VISION_TIMEOUT_S,
            )
            emotion, quip = _parse_reaction(raw)
            if emotion is None:  # small VLM omits the tag → classify from screen text + quip
                emotion = _classify_emotion(quip, context)
            return {"quip": quip, "emotion": emotion}
        except OPENAI_ERRORS as e:
            errors.append(f"{url}: {e}")
    raise BrainError(f"peek failed on all vision endpoints: {'; '.join(errors)}")


def _extract_json_array(text: str) -> list:
    text = _clean(text)
    start, end = text.find("["), text.rfind("]")
    if start < 0 or end <= start:
        return []
    try:
        data = json.loads(text[start : end + 1])
        return data if isinstance(data, list) else []
    except json.JSONDecodeError:
        return []


def _vision_request(url: str, model: str, image_data_url: str, fairy_state: FairyState) -> str:
    # raw OPENAI_ERRORS propagate to see()'s endpoint loop, which falls through to the next
    return _openai_post(
        url,
        {
            "model": model,
            "messages": [
                {"role": "system", "content": VISION_SYSTEM},
                {
                    "role": "user",
                    "content": [
                        {"type": "image_url", "image_url": {"url": image_data_url}},
                        {
                            "type": "text",
                            "text": f"Your mischief is {fairy_state.get('mischief', 45)}/100. What's worth noticing?",
                        },
                    ],
                },
            ],
            "temperature": 0.4,
            "max_tokens": 512,
        },
        VISION_TIMEOUT_S,
    )


# A full desktop downscaled to the model's small visual-token budget is unreadable
# (it hallucinates). Window-sized crops fit the budget at readable resolution, so we
# tile for local llama.cpp. Cloud vLLM reads full screens fine, so it gets the whole
# image (and tiling it would N× the GPU cost). localhost = tile.
TILE_COLS, TILE_ROWS = 3, 2
TILE_OVERLAP = 0.08  # seam overlap so window text isn't split across tiles


def _is_local(url: str) -> bool:
    return "127.0.0.1" in url or "localhost" in url


def _tiles(image_data_url: str) -> list[str]:
    from PIL import Image

    raw = base64.b64decode(image_data_url.split(",", 1)[1])
    img = Image.open(io.BytesIO(raw)).convert("RGB")
    w, h = img.size
    tw, th = w / TILE_COLS, h / TILE_ROWS
    ox, oy = tw * TILE_OVERLAP, th * TILE_OVERLAP
    out = []
    for r in range(TILE_ROWS):
        for c in range(TILE_COLS):
            box = (
                max(0, int(c * tw - ox)),
                max(0, int(r * th - oy)),
                min(w, int((c + 1) * tw + ox)),
                min(h, int((r + 1) * th + oy)),
            )
            crop = img.crop(box)
            crop = crop.resize((crop.width * 2, crop.height * 2))  # upscale: more readable detail
            buf = io.BytesIO()
            crop.save(buf, format="PNG")
            out.append("data:image/png;base64," + base64.b64encode(buf.getvalue()).decode())
    return out


def _parse_events(raw: str) -> list[dict]:
    out = []
    for item in _extract_json_array(raw):
        if not isinstance(item, dict):
            continue
        src, typ, title = item.get("source"), item.get("type"), item.get("title")
        # the model is Puck's eyes, not his judgment — drop anything off-contract
        if src in WIRE_SOURCES and isinstance(typ, str) and isinstance(title, str) and title.strip():
            out.append({"source": src, "type": typ, "title": title.strip()})
    return out


def see(image_data_url: str, fairy_state: FairyState) -> list[dict]:
    """Puck looks at a screenshot → wire-event dicts. Local endpoint reads tiles
    (full screens downscale past readable); cloud reads the whole image. Tries each
    endpoint in order; raises BrainError only if every one fails."""
    endpoints = _vision_endpoints()
    if not endpoints:
        raise BrainError("no vision endpoint configured (PUCK_VISION_URL / _FALLBACK_URL)")

    events, errors = None, []
    for url, model in endpoints:
        try:
            if _is_local(url):
                # one inference per tile, in parallel; merge + dedupe across tiles
                tiles = _tiles(image_data_url)
                # 2 at a time: a 12B VLM running 4 tiles at once melts a laptop GPU;
                # fewer concurrent inferences = lower peak pressure (a bit slower)
                with ThreadPoolExecutor(max_workers=min(2, len(tiles))) as pool:
                    raws = list(pool.map(lambda t: _vision_request(url, model, t, fairy_state), tiles))
                merged = [e for raw in raws for e in _parse_events(raw)]
                events = _dedupe(merged)
            else:
                # whole-image (cloud): still dedupe + cap — a busy screen yields a
                # dozen perceptions and Puck shouldn't react to every browser tab
                events = _dedupe(_parse_events(_vision_request(url, model, image_data_url, fairy_state)))
            break  # first reachable endpoint wins
        except OPENAI_ERRORS as e:
            errors.append(f"{url}: {e}")  # this endpoint down → fall through to the next
    if events is None:
        raise BrainError(f"all vision endpoints failed: {'; '.join(errors)}")
    return events


VISION_MAX_EVENTS = 8  # tiling over-segments; cap the noisiest perceptions


def _dedupe(events: list[dict]) -> list[dict]:
    """A window straddling tile seams yields near-duplicate events. Key on the
    source + a content fingerprint (alphanumeric, first 16 chars) so 'Build failing
    in auth.test.ts' and '3 tests failing - auth.test.ts' collapse. Prefer the more
    concrete report (more digits/filenames) when two collide, then cap."""

    def fingerprint(e: dict) -> tuple[str, str]:
        alnum = re.sub(r"[^a-z0-9]", "", e["title"].lower())
        return (e["source"], alnum[:16])

    def concreteness(e: dict) -> int:
        return len(re.findall(r"\d|\.\w", e["title"]))  # digits + file-extensions = specificity

    best: dict[tuple[str, str], dict] = {}
    for e in events:
        k = fingerprint(e)
        if k not in best or concreteness(e) > concreteness(best[k]):
            best[k] = e
    ranked = sorted(best.values(), key=concreteness, reverse=True)
    return ranked[:VISION_MAX_EVENTS]


def vision_health() -> dict:
    eps = _vision_endpoints()
    return {"ok": bool(eps), "endpoints": [{"url": u, "model": m} for u, m in eps]}


# a 1x1 transparent PNG — enough to exercise the multimodal path during warm-up
_WARM_PIXEL = (
    "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR4"
    "2mNk+M8AAAMBAQDJ/pLvAAAAAElFTkSuQmCC"
)


def warm() -> None:
    """Pre-warm each vision endpoint with a real (tiny image) inference. Just GETting
    /models wakes the Modal container but NOT vLLM's first-inference JIT — flashinfer
    compiles kernels on the first real call (~10s once). The frontend pings this on load
    so a fresh/restarted container is hot before the companion's first peek (~50-95s out).
    Blocking + best-effort (run off-thread)."""
    for url, model in _vision_endpoints():
        try:
            _openai_post(
                url,
                {
                    "model": model,
                    "messages": [
                        {
                            "role": "user",
                            "content": [
                                {"type": "image_url", "image_url": {"url": _WARM_PIXEL}},
                                {"type": "text", "text": "hi"},
                            ],
                        }
                    ],
                    "max_tokens": 1,
                    "temperature": 0,
                },
                VISION_TIMEOUT_S,
            )
        except OPENAI_ERRORS:  # waking is best-effort; the peek path reports real errors
            pass


def health() -> dict:
    """Is anyone thinking back there?"""
    try:
        req = urllib.request.Request(f"{BRAIN_URL}/models", method="GET")
        with urllib.request.urlopen(req, timeout=3) as res:
            json.load(res)
        return {"ok": True, "url": BRAIN_URL, "model": BRAIN_MODEL}
    except Exception as e:  # noqa: BLE001 — any failure means the same thing here: no brain
        return {"ok": False, "url": BRAIN_URL, "error": str(e)}