"""Self-surgery control channel.

Claude (the conversation, or any background subagent) drives the running
app by writing files into control/ — natural for a model with file tools:

  control/config.json   hot-reload settings, applied within ~0.5s, no
                        restart: {"voice": "am_adam", "speed": 1.1}
  control/say.txt       speak this text aloud right now (lets background
                        workers announce results the moment they finish)
  control/restart       ask the app to restart itself. The new code is
                        py_compile-validated FIRST — a broken edit refuses
                        to restart instead of bricking the assistant. The
                        Claude session resumes after restart, so the
                        conversation continues with memory intact.

The watcher applies commands only when the app isn't mid-response, so a
restart requested during a reply lands after the sentence finishes.
"""

from __future__ import annotations

import asyncio
import json
import os
import subprocess
import sys
from pathlib import Path

from voice.events import emit

PROJECT = Path(__file__).resolve().parent.parent
CONTROL = PROJECT / "control"
SESSION_FILE = CONTROL / "session_id"


def _validate_code() -> str | None:
    """Compile-check the app's own source. None = OK, else error text."""
    files = [str(PROJECT / "main.py")] + [
        str(p) for p in (PROJECT / "voice").glob("*.py")
    ]
    proc = subprocess.run(
        [sys.executable, "-m", "py_compile", *files],
        capture_output=True, text=True, timeout=30,
    )
    return None if proc.returncode == 0 else proc.stderr.strip()


class ControlWatcher:
    """Polls control/ and applies commands against the running app."""

    def __init__(self, app) -> None:
        self.app = app
        self._gestures = None      # hands.py subprocess (Popen)
        self._gestures_on = False
        self._watcher = None       # watcher.py subprocess (Popen)
        CONTROL.mkdir(exist_ok=True)
        # apply any settings persisted in config.json BEFORE wiping it, so
        # values like wake_word survive a restart
        cfg = CONTROL / "config.json"
        if cfg.is_file():
            try:
                data = json.loads(cfg.read_text() or "{}")
                if data:
                    self._apply_config(data)
            except Exception as e:
                print(f"       [startup config error: {e}]")
        # clear stale commands from a previous run
        for name in ("restart", "say.txt", "config.json", "report.txt",
                     "mute", "show.json", "text_input.json", "window.json",
                     "screenshot.json", "screenshot.done", "gestures.json",
                     "watch.json", "card_action.json", "email_action.json",
                     "loc_bind.json", "agent_action.json", "entity_action.json",
                     "memory_query.json", "memory_action.json",
                     "graph_query.json", "graph_action.json",
                     "storage_query.json", "storage_action.json"):
            (CONTROL / name).unlink(missing_ok=True)

    async def run(self) -> None:
        try:
            while True:
                await asyncio.sleep(0.5)
                try:
                    await self._tick()
                except Exception as e:
                    print(f"       [control error: {e}]")
        finally:
            self._stop_gestures()  # never leave the camera process behind

    async def _tick(self) -> None:
        cfg = CONTROL / "config.json"
        if cfg.is_file():
            data = json.loads(cfg.read_text() or "{}")
            cfg.unlink()
            self._apply_config(data)

        say = CONTROL / "say.txt"
        if say.is_file():
            text = say.read_text().strip()
            say.unlink()
            if text and self.app.tts:
                pcm = await asyncio.to_thread(self.app.tts.synth, text[:600])
                self.app.speaker.play(pcm)
                print(f"NOTIFY {text}")
                emit("notify", text=text)

        # typed message from the HUD chat box -> text turn (no voice unless
        # want_voice). Run it as a task so this watcher stays responsive
        # (mute/show still work during a long reply); the brain lock keeps
        # text and voice turns from overlapping.
        ti = CONTROL / "text_input.json"
        if ti.is_file():
            payload = ti.read_text()
            ti.unlink()
            try:
                data = json.loads(payload)
            except Exception:
                data = None
            if data and (data.get("text") or data.get("attachments")):
                self.app.speaker.cut()
                asyncio.create_task(self.app.handle_text(
                    data.get("text", ""),
                    data.get("attachments") or [],
                    bool(data.get("want_voice", False)),
                ))

        # workers append completion reports here; they go to the MASTER
        # agent (via app.inbox), who relays them to Ahmed in his own words.
        # (With the agent pool on, the pool delivers results to app.inbox
        # itself — this path stays for watcher.py and AGENT_POOL=0.)
        report = CONTROL / "report.txt"
        if report.is_file():
            lines = report.read_text().strip().splitlines()
            report.unlink()
            for line in lines:
                if line.strip():
                    self.app.inbox.put_nowait(line.strip())

        # HUD ✕ on an agent chip -> kill that worker NOW. The pool disconnects
        # its CLI session and emits task_done, so the chip clears immediately —
        # no more "killed it but the UI is still stuck there".
        act = CONTROL / "agent_action.json"
        if act.is_file():
            payload = act.read_text()
            act.unlink()
            try:
                data = json.loads(payload)
            except Exception:  # noqa: BLE001
                data = {}
            if data.get("action") == "kill":
                from voice import agent_pool
                target = str(data.get("id") or "latest")
                out = await agent_pool.pool.kill(target)
                print(f"       [agent kill: {out}]")
                self.app.inbox.put_nowait(
                    f"(Ahmed killed the background agent '{target}' from the "
                    f"HUD himself — {out} Acknowledge it in one short line; "
                    f"don't restart it unless he asks.)")

        # HUD (or anything) can toggle the mic by touching control/mute
        mute = CONTROL / "mute"
        if mute.is_file():
            mute.unlink()
            from voice.audio_io import _toggle_mute, is_muted
            _toggle_mute()
            emit("muted", on=is_muted())

        # window-management command from Jarvis -> forwarded to the HUD, which
        # owns the Accessibility API tiler. Pass the whole command through.
        win = CONTROL / "window.json"
        if win.is_file():
            payload = win.read_text()
            win.unlink()
            try:
                cmd = json.loads(payload)
                emit("window", command=cmd)
                print(f"       [window: {cmd.get('action')}]")
            except Exception:
                pass

        # the master displays things on the HUD via control/show.json
        show = CONTROL / "show.json"
        if show.is_file():
            payload = show.read_text()
            show.unlink()
            try:
                data = json.loads(payload)
                emit("show", path=str(data.get("path", "")),
                     title=str(data.get("title", "")))
                print(f"       [show: {data.get('path')}]")
            except Exception:
                pass

        # compose/confirm card — Ahmed acted on an email/event draft in the HUD
        # (edited a field, or clicked Send/Cancel/✕). Apply it and, for the
        # terminal actions, drop a note in the master's inbox so Jarvis follows
        # up in his own words. Edits are stored silently (no chatter while he
        # types) — a later spoken "send it" uses the edited values.
        card = CONTROL / "card_action.json"
        if card.is_file():
            payload = card.read_text()
            card.unlink()
            try:
                data = json.loads(payload or "{}")
            except Exception:
                data = {}
            await self._card_action(data)

        # email reader card — Ahmed acted on the thread card in the HUD: opened
        # an attachment, hit "Load images", or typed a reply and clicked Send.
        # Only a sent reply speaks (a note in the master's inbox); the rest are
        # silent — he's reading, not talking.
        mail = CONTROL / "email_action.json"
        if mail.is_file():
            payload = mail.read_text()
            mail.unlink()
            try:
                data = json.loads(payload or "{}")
            except Exception:
                data = {}
            await self._email_action(data)

        # ENTITY DOSSIER card — Ahmed acted on a profile card in the HUD
        # (drilled into a related entity, added a note, or edited the summary).
        # Same atomic-file mechanism as the email reader card above, so both
        # cards flow through here identically.
        ent = CONTROL / "entity_action.json"
        if ent.is_file():
            payload = ent.read_text()
            ent.unlink()
            try:
                data = json.loads(payload or "{}")
            except Exception:
                data = {}
            await self._entity_action(data)

        # MEMORY VIEWER (load/search) — the HUD asks to see the shared brain;
        # the engine holds the API key, so it queries Railway and emits the
        # results back as a `memories` event. Empty q = newest memories.
        mq = CONTROL / "memory_query.json"
        if mq.is_file():
            payload = mq.read_text()
            mq.unlink()
            try:
                data = json.loads(payload or "{}")
            except Exception:
                data = {}
            from voice import memory_control
            await asyncio.to_thread(memory_control.refresh_panel,
                                    str(data.get("q", "")).strip())

        # MEMORY EDIT/DELETE — Ahmed corrected or removed a fact in the viewer.
        ma = CONTROL / "memory_action.json"
        if ma.is_file():
            payload = ma.read_text()
            ma.unlink()
            try:
                data = json.loads(payload or "{}")
            except Exception:
                data = {}
            await self._memory_action(data)

        # MEMORY GRAPH (load) — the HUD switched to graph mode (or hit reload);
        # the engine fetches the node/edge graph from Railway and emits it back
        # as a `graph` event. Body is ignored — any file here means "refresh".
        gq = CONTROL / "graph_query.json"
        if gq.is_file():
            gq.unlink()
            from voice import memory_control
            await asyncio.to_thread(memory_control.refresh_graph)

        # MEMORY GRAPH EDIT — Ahmed connected or cut an edge in the graph view.
        ga = CONTROL / "graph_action.json"
        if ga.is_file():
            payload = ga.read_text()
            ga.unlink()
            try:
                data = json.loads(payload or "{}")
            except Exception:
                data = {}
            await self._graph_action(data)

        # STORAGE (load/refresh) — the HUD opened the file panel (or an action
        # asked for a refresh). The engine holds the Supabase key, so it lists
        # the bucket (METADATA ONLY) and emits a `storage` event back. Body
        # ignored — any file here means "refresh".
        sq = CONTROL / "storage_query.json"
        if sq.is_file():
            sq.unlink()
            from voice import storage_control
            await asyncio.to_thread(storage_control.show)

        # STORAGE ACTION — Ahmed clicked a file (open), a trash icon (delete), or
        # dropped files on the panel (upload).
        sa = CONTROL / "storage_action.json"
        if sa.is_file():
            payload = sa.read_text()
            sa.unlink()
            try:
                data = json.loads(payload or "{}")
            except Exception:
                data = {}
            await self._storage_action(data)

        # LOCATION BIND — Jarvis labels the current (unknown) network as a place
        # after Ahmed says where he is ("this is the office").
        lb = CONTROL / "loc_bind.json"
        if lb.is_file():
            payload = lb.read_text()
            lb.unlink()
            try:
                data = json.loads(payload or "{}")
                from voice import location
                fp = await asyncio.to_thread(
                    location.bind, str(data.get("place", "")).strip(),
                    str(data.get("label", "")).strip(),
                    str(data.get("ssid", "")).strip())
                if fp:
                    self.app._write_status()
                    print(f"       [location: bound {fp} -> {data.get('place')}]")
            except Exception as e:  # noqa: BLE001
                print(f"       [loc_bind failed: {e}]")

        # SEE THE SCREEN — Jarvis requests a capture; the ENGINE (this process,
        # a child of Jarvis.app which holds Screen Recording) does it, so the
        # grant actually applies. Result path/error land in screenshot.done.
        shot = CONTROL / "screenshot.json"
        if shot.is_file():
            payload = shot.read_text()
            shot.unlink()
            try:
                data = json.loads(payload or "{}")
            except Exception:
                data = {}
            path = str(data.get("path") or "/tmp/jarvis_screen.png")
            display = data.get("display")
            from voice import screen
            ok, err = await asyncio.to_thread(screen.capture, path, display)
            (CONTROL / "screenshot.done").write_text(json.dumps(
                {"ok": ok, "path": path if ok else None, "error": err}))
            emit("screenshot", ok=ok, path=path if ok else "")
            print(f"       [screenshot: ok={ok} {err or path}]")

        # screen watcher on/off — Jarvis writes control/watch.json
        # {"on": true, "task": "...", "minutes": 5, "interval": 4};
        # watcher.py (local VLM via Ollama) reports via report.txt.
        watch = CONTROL / "watch.json"
        if watch.is_file():
            payload = watch.read_text()
            watch.unlink()
            try:
                data = json.loads(payload or "{}")
            except Exception:
                data = {}
            self._set_watcher(data)

        # hand-gesture tracker on/off — HUD (or Jarvis) writes
        # control/gestures.json {"on": true}; hands.py streams UDP :47831
        gest = CONTROL / "gestures.json"
        if gest.is_file():
            payload = gest.read_text()
            gest.unlink()
            try:
                data = json.loads(payload or "{}")
            except Exception:
                data = {}
            self._set_gestures(bool(data.get("on")))
        elif self._gestures_on and (
                self._gestures is None or self._gestures.poll() is not None):
            self._gestures_on = False  # tracker died on its own (reaped)
            print("       [gestures: tracker exited]")
            emit("gestures", on=False)

        restart = CONTROL / "restart"
        if restart.is_file():
            restart.unlink()
            await self._restart()

    async def _card_action(self, data: dict) -> None:
        """Apply a card action from the HUD (email/event draft, or a task tick)."""
        action = str(data.get("action", "")).strip()
        kind = str(data.get("kind", "draft")).strip() or "draft"

        # Ahmed ticked a task/reminder off on the HUD → complete it on Railway
        # (syncs everywhere) and tell Jarvis so he can acknowledge it.
        if kind == "task":
            tid = str(data.get("id", "")).strip()
            text = str(data.get("text", "")).strip()
            if action in ("done", "complete") and tid:
                try:
                    import asyncio as _a
                    from voice import tasks_control
                    await _a.to_thread(tasks_control.mark_done, tid)
                    self.app.inbox.put_nowait(
                        f"[task] Ahmed ticked off a task on the HUD"
                        + (f': "{text}"' if text else "") + " — it's done.")
                except Exception as e:  # noqa: BLE001
                    print(f"       [task done failed: {e}]")
            return

        from voice import compose
        did = str(data.get("draft_id", "")).strip()
        fields = data.get("fields") or {}
        if not did or not compose.get_draft(did):
            return
        # Fold in whatever Ahmed had typed in the card first. `attachments`
        # (files he added via Finder/drag-drop, or removed as chips) rides in
        # `fields` exactly like to/cc/subject/body — normalize it to a clean
        # list of paths ([] is a real value: he removed them all).
        if isinstance(fields, dict) and "attachments" in fields:
            fields = dict(fields)
            fields["attachments"] = compose._parse_attachments(
                fields.get("attachments"))
        if isinstance(fields, dict) and fields:
            compose.update_draft(did, fields, reemit=False)

        if action == "edit":
            return  # stored silently; no follow-up speech
        if action == "send":
            ok, out = await compose.send_draft(did)
            note = (f"[compose] Ahmed clicked Send on the {kind} — {out}" if ok
                    else f"[compose] Ahmed clicked Send on the {kind} but it "
                         f"FAILED: {out}")
            self.app.inbox.put_nowait(note)
        elif action == "cancel":
            compose.cancel_draft(did)
            self.app.inbox.put_nowait(
                f"[compose] Ahmed cancelled the {kind} draft — do not send it.")
        elif action == "close":
            compose.cancel_draft(did)
            self.app.inbox.put_nowait(
                f"[compose] Ahmed dismissed the {kind} card without sending.")

    async def _email_action(self, data: dict) -> None:
        """Apply an email-card action from the HUD (control/email_action.json):
          attachment      → download it, then emit email_attachment (chip lights up)
          open_attachment → open the downloaded file in Preview/Finder
          reply/reply_all → send it on the thread, then tell Jarvis to say so
          images          → re-emit the thread with that message's remote images
          close           → drop the engine-side thread state (the HUD already
                            dismissed the card itself, so no event goes back)
        Blocking backends (Gmail API / IMAP+SMTP) run off-thread so the voice
        loop keeps listening while a big attachment downloads."""
        from voice import mail_view
        action = str(data.get("action", "")).strip()
        tid = str(data.get("thread_id", "")).strip()
        acct = str(data.get("account", "")).strip()
        mid = str(data.get("msg_id", "")).strip()
        try:
            if action == "attachment":
                await asyncio.to_thread(
                    mail_view.download, tid, mid,
                    str(data.get("att_id", "")).strip(),
                    str(data.get("filename", "")).strip(), acct)
            elif action == "open_attachment":
                path = str(data.get("path", "")).strip()
                if path:
                    await asyncio.to_thread(mail_view.open_path, path)
            elif action in ("reply", "reply_all"):
                ok, out = await asyncio.to_thread(
                    mail_view.reply, tid, mid, str(data.get("body", "")),
                    action == "reply_all", acct)
                self.app.inbox.put_nowait(
                    f"[mail] Ahmed sent a reply from the email card — {out}"
                    if ok else
                    f"[mail] Ahmed's reply from the email card FAILED: {out}")
            elif action == "images":
                mail_view.load_images(tid, mid)
            elif action == "close":
                mail_view.drop(tid)
        except Exception as e:  # noqa: BLE001 — a HUD action must never crash the app
            print(f"       [email action failed: {e}]")

    async def _entity_action(self, data: dict) -> None:
        """Apply an Entity Dossier card action from the HUD
        (control/entity_action.json). Mirrors _email_action — same atomic-file
        mechanism, blocking backends run off-thread so the voice loop keeps
        listening:
          show_entity    {name}       → drill-down: pop that entity's card
          entity_note    {key, text}  → save a note attached to the entity (memory)
          entity_summary {key, text}  → upsert the entity's current-state summary
        Best-effort; a bad payload never crashes the watcher."""
        from voice import entity_show, memory_control
        cmd = str(data.get("cmd", "")).strip()
        try:
            if cmd == "show_entity":
                name = str(data.get("name", "")).strip()
                if name:
                    await asyncio.to_thread(entity_show.show_entity, name)
            elif cmd == "entity_note":
                key = str(data.get("key", "")).strip()
                text = str(data.get("text", "")).strip()
                if key and text:
                    # The card's note field is a CORRECTION/ADDITION order,
                    # not a sticky note: a pooled Claude worker (Max sub, $0)
                    # reads the entity's memories, fixes what the note says is
                    # wrong / saves what's new, then refreshes the card.
                    asyncio.ensure_future(self._entity_note_agent(key, text))
            elif cmd == "entity_summary":
                key = str(data.get("key", "")).strip()
                text = str(data.get("text", "")).strip()
                if key and text:
                    await asyncio.to_thread(
                        memory_control._req, "POST", "/summary",  # noqa: SLF001
                        {"name": key, "text": text,
                         "group": memory_control._GROUP})  # noqa: SLF001
            elif cmd == "entity_unrelate":
                # Ahmed right-clicked "Not related" on a relation row — the
                # HUMAN confirmation that severs a wrong edge (never the LLM).
                key = str(data.get("key", "")).strip()
                other = str(data.get("other", "")).strip()
                if key and other:
                    await asyncio.to_thread(
                        memory_control._req, "POST", "/entity/unrelate",  # noqa: SLF001
                        {"a": key, "b": other,
                         "group": memory_control._GROUP})  # noqa: SLF001
        except Exception as e:  # noqa: BLE001 — a HUD action must never crash the app
            print(f"       [entity action failed: {str(e)[:120]}]")

    async def _entity_note_agent(self, key: str, text: str) -> None:
        """Run a dossier-card note through a pooled Claude worker as a memory
        CURATION task: interpret Ahmed's note as a correction or an addition,
        apply it to the graph (memory_update supersede / memory_remember), and
        re-show the entity so the corrected card replaces the stale one. Falls
        back to a plain saved note if the pool is unavailable — the note is
        never lost either way."""
        from voice import agent_pool, memory_control
        task = (
            f'Ahmed typed this on the dossier card of "{key}" in his memory '
            f'HUD: "{text}"\n\n'
            "This is a memory-CURATION order about that entity. Do exactly "
            "this:\n"
            "1. mcp__memory__memory_search for that entity's facts — search "
            "the entity name AND the topic of his note; read every result "
            "(ids included).\n"
            "2. If the note CORRECTS something: for EVERY fact that states "
            "the wrong version, call mcp__memory__memory_update (old_id = "
            "that fact's id, text = the corrected fact, written cleanly in "
            "third person). Fix all of them, not just the first.\n"
            "3. If the note ADDS something new: save it with "
            f'mcp__memory__memory_remember (entities="{key}").\n'
            "4. When the graph is fixed, call "
            f'mcp__entity__show_entity("{key}") so his card refreshes with '
            "the corrected dossier.\n"
            "Be surgical: change only what the note actually says. Reply with "
            "one line summarizing what you fixed/added."
        )
        try:
            out = await agent_pool.pool.dispatch(
                task, title=f"fix memory: {key}")
            if isinstance(out, str) and out.startswith("REFUSED"):
                raise RuntimeError(out)
        except Exception as e:  # noqa: BLE001 — pool down ⇒ never lose the note
            print(f"       [entity note agent failed, saving plain note: "
                  f"{str(e)[:100]}]")
            memory_control.fire_save(f"Note about {key}: {text}", kind="note")

    async def _memory_action(self, data: dict) -> None:
        """Apply a memory action from the HUD viewer: forget (permanent delete)
        or edit (supersede with corrected text). Then refresh the panel and drop
        a note so Jarvis knows the brain changed."""
        from voice import memory_control
        action = str(data.get("action", "")).strip()
        mid = str(data.get("id", "")).strip()
        if not mid:
            return
        q = str(data.get("q", "")).strip()   # keep the current view/search
        text = str(data.get("text", "")).strip()
        try:
            if action == "forget":
                await asyncio.to_thread(memory_control.forget, mid)
                emit("memory_forgotten", id=mid)
                self.app.inbox.put_nowait(
                    "[memory] Ahmed deleted a memory in the viewer"
                    + (f': "{text[:80]}"' if text else "")
                    + " — forget it, it was wrong; don't bring it up again.")
            elif action == "edit" and text:
                await asyncio.to_thread(memory_control.edit, mid, text)
                self.app.inbox.put_nowait(
                    f'[memory] Ahmed corrected a memory in the viewer to: '
                    f'"{text[:100]}".')
            else:
                return
            await asyncio.to_thread(memory_control.refresh_panel, q)
        except Exception as e:  # noqa: BLE001
            print(f"       [memory action failed: {e}]")

    async def _graph_action(self, data: dict) -> None:
        """Apply a graph edit from the HUD viewer: link (connect two memories)
        or unlink (cut the edge). Then refresh the graph and drop a note so
        Jarvis knows the brain's connections changed."""
        from voice import memory_control
        action = str(data.get("action", "")).strip()
        a = str(data.get("a", "")).strip()
        b = str(data.get("b", "")).strip()
        if not a or not b:
            return
        try:
            if action == "link":
                typ = str(data.get("type", "related")).strip() or "related"
                await asyncio.to_thread(memory_control.link, a, b, typ)
                self.app.inbox.put_nowait(
                    "[memory] Ahmed connected two memories in the graph view.")
            elif action == "unlink":
                await asyncio.to_thread(memory_control.unlink, a, b)
                self.app.inbox.put_nowait(
                    "[memory] Ahmed disconnected two memories in the graph view.")
            else:
                return
            await asyncio.to_thread(memory_control.refresh_graph)
        except Exception as e:  # noqa: BLE001
            print(f"       [graph action failed: {e}]")

    async def _storage_action(self, data: dict) -> None:
        """Apply a storage action from the HUD file panel:
          open   → download the file (cached) then open it in its default app
          delete → remove it, then re-emit the list so the panel refreshes
          upload → upload the dropped file(s), then refresh the panel
        Only `open` pulls bytes, and only for the one file clicked."""
        from voice import storage_control
        action = str(data.get("action", "")).strip()
        try:
            if action == "open":
                name = str(data.get("name", "")).strip()
                if not name:
                    return
                await asyncio.to_thread(storage_control.open_file, name)
            elif action == "delete":
                name = str(data.get("name", "")).strip()
                if not name:
                    return
                await asyncio.to_thread(storage_control.delete, name)
                await asyncio.to_thread(storage_control.show)   # refresh panel
                self.app.inbox.put_nowait(
                    f'[storage] Ahmed deleted "{name}" from his files on the HUD.')
            elif action == "upload":
                # accept a single `path` or a list of `paths`
                paths = data.get("paths")
                if not paths:
                    one = str(data.get("path", "")).strip()
                    paths = [one] if one else []
                done = []
                for p in paths:
                    p = str(p).strip()
                    if not p:
                        continue
                    try:
                        done.append(await asyncio.to_thread(
                            storage_control.upload, p))
                    except Exception as e:  # noqa: BLE001 — one bad file mustn't stop the rest
                        print(f"       [storage upload failed for {p}: {e}]")
                if done:
                    await asyncio.to_thread(storage_control.show)  # refresh panel
                    self.app.inbox.put_nowait(
                        "[storage] Ahmed uploaded "
                        + ", ".join(f'"{n}"' for n in done)
                        + " to his files on the HUD.")
            else:
                return
        except Exception as e:  # noqa: BLE001 — a HUD action must never crash the app
            print(f"       [storage action failed: {e}]")

    def _apply_config(self, data: dict) -> None:
        applied = []
        if "voice" in data and self.app.tts:
            self.app.tts._voice = str(data["voice"])
            applied.append(f"voice={data['voice']}")
        if "speed" in data and self.app.tts:
            self.app.tts._speed = float(data["speed"])
            applied.append(f"speed={data['speed']}")
        if "wake_word" in data:
            self.app.wake_word = bool(data["wake_word"])
            applied.append(f"wake_word={self.app.wake_word}")
            emit("wake_word", on=self.app.wake_word)
        if "text_voice" in data:
            self.app.text_voice = bool(data["text_voice"])
            applied.append(f"text_voice={self.app.text_voice}")
            emit("text_voice", on=self.app.text_voice)
        if "collapsed" in data:  # minimize/expand the HUD bubble
            emit("collapsed", on=bool(data["collapsed"]))
            applied.append(f"collapsed={bool(data['collapsed'])}")
        if "dock" in data:  # dock Jarvis to an edge / undock
            payload = {"on": bool(data["dock"])}
            if "dock_width" in data:
                payload["width"] = float(data["dock_width"])
            if "dock_side" in data:
                payload["side"] = str(data["dock_side"])
            emit("dock", **payload)
            applied.append(f"dock={payload}")
        if applied:
            print(f"       [hot-reload: {', '.join(applied)}]")

    def _terminate_gestures(self) -> None:
        """SIGTERM the tracker and WAIT for it to actually exit (so the camera
        is released) — poll up to ~2s, then hard-kill. Keeps the camera-busy
        window near zero across an off→on restart, so the fresh tracker doesn't
        sit in its 20x1s open-retry loop while an old one still holds camera 0."""
        proc = self._gestures
        if proc is not None and proc.poll() is None:
            proc.terminate()  # SIGTERM -> hands.py frees the camera cleanly
            try:
                proc.wait(timeout=2)
            except subprocess.TimeoutExpired:
                proc.kill()
                try:
                    proc.wait(timeout=1)
                except subprocess.TimeoutExpired:
                    pass
        self._gestures = None

    def _set_gestures(self, on: bool) -> None:
        running = self._gestures is not None and self._gestures.poll() is None
        if on:
            # if an old tracker is still alive (e.g. a fast off→on toggle),
            # tear it down and WAIT for the camera to free BEFORE spawning the
            # new one — otherwise the new process burns ~20s in its retry loop.
            if running:
                self._terminate_gestures()
            # hands.py is standalone: prefer the dedicated venv if one exists
            # (fallback for a mediapipe/numpy conflict in the main venv)
            if os.name == "nt":
                venv = PROJECT / ".venv-hands" / "Scripts" / "python.exe"
            else:
                venv = PROJECT / ".venv-hands" / "bin" / "python"
            py = str(venv) if venv.is_file() else sys.executable
            self._gestures = subprocess.Popen(
                [py, str(PROJECT / "voice" / "hands.py")])
            print("       [gestures: tracker started]")
        else:
            if running:
                self._terminate_gestures()
                print("       [gestures: tracker stopped]")
        if on != self._gestures_on:
            self._gestures_on = on
            emit("gestures", on=on)

    def _stop_gestures(self) -> None:
        self._terminate_gestures()  # never leave the camera process behind
        if self._gestures_on:
            self._gestures_on = False
            emit("gestures", on=False)

    def _set_watcher(self, data: dict) -> None:
        """Start/stop the local-VLM screen watcher (voice/watcher.py)."""
        running = self._watcher is not None and self._watcher.poll() is None
        if not data.get("on"):
            if running:
                self._watcher.terminate()
                print("       [watcher: stopped]")
            self._watcher = None
            return
        if running:  # restart with the new task
            self._watcher.terminate()
            try:
                self._watcher.wait(timeout=2)
            except subprocess.TimeoutExpired:
                self._watcher.kill()
        env = dict(os.environ,
                   WATCH_TASK=str(data.get("task", "the screen")),
                   WATCH_MINUTES=str(data.get("minutes", 5)),
                   WATCH_INTERVAL=str(data.get("interval", 4)),
                   WATCH_LIVE="1" if data.get("live") else "0")
        self._watcher = subprocess.Popen(
            [sys.executable, str(PROJECT / "voice" / "watcher.py")], env=env)
        print(f"       [watcher: started — {data.get('task', 'the screen')}]")

    async def _restart(self) -> None:
        err = _validate_code()
        if err:
            print(f"       [restart REFUSED — new code is broken]\n{err}")
            if self.app.tts:
                pcm = await asyncio.to_thread(
                    self.app.tts.synth,
                    "I can't restart — the new code has an error. "
                    "Check the terminal.")
                self.app.speaker.play(pcm)
            return
        # wait for the current exchange to finish
        while self.app.responding or self.app.speaker.speaking:
            await asyncio.sleep(0.3)
        print("       [self-restart: code validated, handing off to HUD]")
        emit("restarting")
        if self.app.tts:
            pcm = await asyncio.to_thread(
                self.app.tts.synth, "Restarting myself. Back in a moment.")
            self.app.speaker.play(pcm)
            await self.app.speaker.wait_done()
        await self.app.stop()
        # Signal the HUD to kill and relaunch the engine subprocess.
        # If the HUD is running, it picks this up and does a clean restart
        # with a fresh stdout pipe attached.  If we're running without the
        # HUD (plain terminal via run.sh), fall back to the classic execv.
        hud_restart = CONTROL / "hud_restart"
        hud_restart.touch()
        # Give the HUD a moment to detect the file; if it hasn't picked it up
        # within 2 s we are running without the HUD — do the classic execv.
        for _ in range(20):
            await asyncio.sleep(0.1)
            if not hud_restart.exists():
                # HUD consumed the file — it will relaunch the engine
                return
        # No HUD — fall back to in-process restart
        hud_restart.unlink(missing_ok=True)
        self._stop_gestures()  # execv skips run()'s finally
        os.chdir(PROJECT)
        os.execv(sys.executable, [sys.executable, str(PROJECT / "main.py")])


def load_saved_session() -> str | None:
    if SESSION_FILE.is_file():
        sid = SESSION_FILE.read_text().strip()
        return sid or None
    return None


def save_session(session_id: str) -> None:
    CONTROL.mkdir(exist_ok=True)
    SESSION_FILE.write_text(session_id)