Round 5: hook relay, reply queue, runway cockpit

backend/muse/alerts.py

@@ -18,7 +18,10 @@
 from datetime import datetime, timedelta, timezone
 from typing import Optional
 
+from types import SimpleNamespace
+
 from . import db
+from .autopilot import sessions as live_discovery
 from .incremental import new_objects
 from .models import AlertEvent
 from .paths import SessionPaths
@@ -29,6 +32,14 @@
 _USAGE_WARM_INTERVAL = 60.0  # mtime-cached, so warm calls only parse changed files
 _AI_SCHEDULE_INTERVAL = 900.0  # auto-digest check cadence (opt-in via MUSE_AI_AUTO_DIGEST)
 
+# Hook-driven alerts (see routers/hooks.py). A Stop hook fires at the end of
+# EVERY turn — including rapid back-and-forth at the keyboard — so the alert
+# waits this long and re-checks the session is still idle before pushing.
+_TURN_END_SETTLE_SECONDS = 6.0
+# After a hook-driven alert, suppress the polling tick's duplicate for this long
+# (the tick reads a TTL-cached snapshot and would re-fire on the same transition).
+_HOOK_SUPPRESS_SECONDS = 90.0
+
 
 def _scan_errors(objs: list[dict]) -> list[str]:
     """Short descriptions of any error indicators in a batch of raw lines."""
@@ -66,6 +77,11 @@ def __init__(self, service) -> None:
         self._last_checkpoint = 0.0
         self._last_usage_warm = 0.0
         self._last_ai_schedule = 0.0
+        # Hook-driven alert state: tick-suppression windows + a per-session
+        # generation counter that cancels an in-flight turn-ended alert when the
+        # user replies (or another Stop supersedes it).
+        self._suppress: dict[tuple[str, str], float] = {}
+        self._turn_gen: dict[str, int] = {}
 
     def start(self) -> None:
         if self._task is None or self._task.done():
@@ -184,9 +200,9 @@ async def _tick(self) -> None:
             self._states[sid] = s.state
             if first or prev is None or prev == s.state:
                 continue
-            if s.state == "waiting" and rules.on_waiting:
+            if s.state == "waiting" and rules.on_waiting and not self._suppressed(sid, "waiting"):
                 await self._fire(cfg, rules, s, "waiting", f"✋ {s.title} is waiting for you", "")
-            elif s.state == "stopped" and rules.on_stopped:
+            elif s.state == "stopped" and rules.on_stopped and not self._suppressed(sid, "stopped"):
                 await self._fire(cfg, rules, s, "stopped", f"⏹ {s.title} stopped", "")
 
         # Forget sessions that disappeared.
@@ -197,6 +213,89 @@ async def _tick(self) -> None:
 
         self._primed = True
 
+    # --- hook-driven alerts ---------------------------------------------------
+    # Claude Code hooks (routers/hooks.py) call these on the event loop for
+    # instant notifications; the polling tick above stays as the fallback for
+    # sessions without hooks installed. Each hook alert marks the session state
+    # and opens a suppression window so the tick can't double-fire.
+
+    def _suppressed(self, sid: str, kind: str) -> bool:
+        return time.monotonic() < self._suppress.get((sid, kind), 0.0)
+
+    def _mark_hook_alert(self, sid: str, kind: str, state: str) -> None:
+        self._states[sid] = state
+        self._suppress[(sid, kind)] = time.monotonic() + _HOOK_SUPPRESS_SECONDS
+
+    def _summary_for(self, sid: str):
+        try:
+            for s in self.service.list_sessions():
+                if s.session_id == sid:
+                    return s
+        except Exception:
+            pass
+        return SimpleNamespace(session_id=sid, title=sid[:8])
+
+    def hook_user_replied(self, sid: str) -> None:
+        """UserPromptSubmit: the user answered — cancel any in-flight turn-ended
+        alert and clear the waiting state instantly."""
+        self._turn_gen[sid] = self._turn_gen.get(sid, 0) + 1
+        if self._states.get(sid) == "waiting":
+            self._states[sid] = "running"
+
+    async def hook_needs_you(self, sid: str, message: str) -> None:
+        """Notification hook: Claude needs permission (or sat idle). Always
+        actionable — push immediately."""
+        rules = self.service.get_alert_rules()
+        if not rules.on_waiting:
+            return
+        if self._states.get(sid) == "waiting" and self._suppressed(sid, "waiting"):
+            return  # already alerted for this pause
+        s = self._summary_for(sid)
+        self._mark_hook_alert(sid, "waiting", "waiting")
+        await self._fire(
+            self.service.get_notify_config(), rules, s,
+            "waiting", f"✋ {s.title} needs you", message[:140],
+        )
+
+    async def hook_turn_ended(self, sid: str) -> None:
+        """Stop hook (with nothing queued to deliver): the turn ended. Wait a
+        settle period, then push only if the session is still idle — so rapid
+        at-the-keyboard back-and-forth doesn't spam the phone."""
+        rules = self.service.get_alert_rules()
+        if not rules.on_waiting:
+            return
+        gen = self._turn_gen.get(sid, 0) + 1
+        self._turn_gen[sid] = gen
+        await asyncio.sleep(_TURN_END_SETTLE_SECONDS)
+        if self._turn_gen.get(sid) != gen:
+            return  # user replied (or a newer turn superseded this one)
+        if self._states.get(sid) == "waiting" and self._suppressed(sid, "waiting"):
+            return
+        ls = await asyncio.to_thread(
+            lambda: next((x for x in live_discovery.discover() if x.session_id == sid), None)
+        )
+        # Unknown to live discovery (headless/one-off run) → not ours to alert on;
+        # moved on / blocked on a permission → Notification or the tick covers it.
+        if ls is None or ls.status != "idle" or ls.waiting_for:
+            return
+        s = self._summary_for(sid)
+        self._mark_hook_alert(sid, "waiting", "waiting")
+        await self._fire(
+            self.service.get_notify_config(), rules, s,
+            "waiting", f"✋ {s.title} is ready for you", "",
+        )
+
+    async def hook_session_end(self, sid: str) -> None:
+        rules = self.service.get_alert_rules()
+        if not rules.on_stopped or self._states.get(sid) == "stopped":
+            return
+        s = self._summary_for(sid)
+        self._mark_hook_alert(sid, "stopped", "stopped")
+        await self._fire(
+            self.service.get_notify_config(), rules, s,
+            "stopped", f"⏹ {s.title} ended", "",
+        )
+
     def _schedule_ai_digests(self) -> None:
         """Enqueue yesterday's daily digest (and, on Mondays, last week's retro)
         when enabled and not already generated/pending. Runs in a worker thread."""

backend/muse/autopilot/controller.py

@@ -5,22 +5,34 @@
 from __future__ import annotations
 
 import asyncio
+import json
 import re
 import time
 from datetime import datetime, timedelta, timezone
 from typing import Optional
 
 from .. import discovery as session_discovery
+from .. import options as opt
 from ..config import get_settings
 from ..models import AutopilotConfig, AutopilotSession, AutopilotState
 from ..usage_cache import context_pcts, scan_all
 from . import sessions as live_discovery
+from . import snapshot as layout_snapshot
 from . import tmux
 from .resettime import parse_reset_time
 from .store import AutopilotStore
 
 TICK_SECONDS = 5
+# Layout snapshots for session-restore: check the topology at most this often, and only
+# write when the structure changed (or a long backstop elapsed) — change-driven, so it
+# doesn't churn while you're just working inside a session.
+SNAPSHOT_CHECK_SECONDS = 25
+SNAPSHOT_BACKSTOP_SECONDS = 1800
 INJECT_STATUSES = {"idle"}  # only when a turn finished and it's awaiting the user
+# Minimum gap between queued-reply deliveries to the same session: after a send
+# the status file takes a moment to flip to busy, so without a floor the next
+# tick could double-fire into the same idle turn.
+QUEUE_COOLDOWN_SECONDS = 15
 
 
 def _dt_or_none(value) -> Optional[datetime]:
@@ -49,6 +61,10 @@ def __init__(self) -> None:
         # Optional observer for parsed usage-limit reset times (wired to the
         # usage-history store in main.py so stats can anchor the 5h window).
         self.on_reset = None
+        # Optional observer for limit-hit sightings: on_limit("5h"|"week") records
+        # the window's spend as an observed ceiling (subscription plans publish no
+        # $ caps, so the wall's location is only learnable by hitting it).
+        self.on_limit = None
         # AI idle-mode callables (wired in main.py to the SessionService — same
         # pattern as on_reset, avoiding an import cycle):
         #   enqueue_draft(sid) -> AIJob | None
@@ -57,6 +73,11 @@ def __init__(self) -> None:
         self.enqueue_draft = None
         self.get_ai_job = None
         self.ai_cost_today = None
+        # Last queued-reply delivery per session (monotonic secs), for the cooldown.
+        self._queue_sent_at: dict[str, float] = {}
+        # Layout-snapshot bookkeeping (monotonic secs): when we last checked/wrote.
+        self._last_snapshot_check = 0.0
+        self._last_snapshot_write = 0.0
 
     def start(self) -> None:
         if self._task is None or self._task.done():
@@ -172,7 +193,138 @@ async def _run(self) -> None:
     # board ticker).
     _context_pcts = staticmethod(context_pcts)
 
+    # --- queued replies -------------------------------------------------------
+    # Deliver user-authored "send when this turn ends" messages. This runs inside
+    # the controller because the controller is the one disciplined automated
+    # tmux-write site — same guards as autopilot injection (idle status, pane
+    # re-check, rate-limit banner, visible-menu refusal), but it does NOT require
+    # autopilot to be armed: queueing was an explicit user action.
+
+    def _queue_block_reason(
+        self, sid: str, live: dict, now_mono: float, *, force: bool = False
+    ) -> Optional[str]:
+        """Why the next queued reply for `sid` can't be delivered right now, or
+        None if it would go. `force` (a user's explicit "send now") skips the
+        wait-for-idle gate but never overrides the guards that would corrupt the
+        session — typing into an open menu or over a usage-limit banner."""
+        ls = live.get(sid)
+        if ls is None or not ls.pane_id:
+            return "session isn’t running in tmux"
+        if not force:
+            if ls.waiting_for:
+                return f"waiting for {ls.waiting_for}"
+            if ls.status not in INJECT_STATUSES:
+                return f"session is {ls.status}, not idle"
+            if now_mono - self._queue_sent_at.get(sid, 0.0) < QUEUE_COOLDOWN_SECONDS:
+                return "just delivered — cooling down"
+        pane = tmux.capture_pane(ls.pane_id, 40)
+        if _RATE_LIMIT_RE.search(pane):
+            self._observe_limit(pane)
+            return "usage-limit banner on screen"
+        # A visible ❯ menu means typed text would land IN the dialog (digits
+        # select options!) — never deliver over one, even when forced.
+        if opt.parse_permission_menu(pane) is not None:
+            return "a menu is open — answer it first"
+        return None
+
+    def _send_batch(self, sid: str, pane_id: str, now_mono: float) -> tuple[list[int], Optional[str]]:
+        """Deliver the oldest pending item plus any directly-following 'append'
+        items (they asked to share one message). Returns (sent ids, error)."""
+        batch = self.store.queue_next_batch(sid)
+        if not batch:
+            return [], None
+        combined = "\n".join(i.text for i in batch)
+        if "\n" in combined:
+            # Multiline goes as a bracketed paste — raw LF via send-keys would
+            # submit each line as its own message.
+            ok, err = tmux.paste_text(pane_id, combined)
+        else:
+            ok, err = tmux.send_text(pane_id, combined)
+        if ok:
+            self._queue_sent_at[sid] = now_mono
+            for i in batch:
+                self.store.queue_mark(i.id, "sent")
+            self.store.log(
+                sid, "queue_sent",
+                f"{pane_id} ← {combined[:80]}"
+                + (f" ({len(batch)} items)" if len(batch) > 1 else ""),
+            )
+            return [i.id for i in batch], None
+        for i in batch:
+            self.store.queue_mark(i.id, "failed", error=err)
+        self.store.log(sid, "error", f"queued send failed: {err}")
+        return [], err
+
+    def deliver_queued(self, session_id: Optional[str] = None) -> list[int]:
+        """Try to deliver the next queued reply for `session_id` (or for every
+        session with a pending queue). Returns the queue ids actually sent.
+        Safe to call from any thread; each delivery re-checks the world fresh."""
+        counts = self.store.queue_counts()
+        if session_id is not None:
+            counts = {session_id: counts[session_id]} if session_id in counts else {}
+        if not counts:
+            return []
+        live = {s.session_id: s for s in live_discovery.discover()}
+        now_mono = time.monotonic()
+        sent: list[int] = []
+        for sid in counts:
+            if self._queue_block_reason(sid, live, now_mono) is not None:
+                continue
+            ids, _ = self._send_batch(sid, live[sid].pane_id, now_mono)
+            sent.extend(ids)
+        return sent
+
+    def deliver_now(self, session_id: str) -> tuple[list[int], Optional[str]]:
+        """User override: deliver the next batch immediately, skipping the
+        wait-for-idle gate but not the menu / rate-limit guards. Returns
+        (sent ids, reason-it-was-blocked)."""
+        if session_id not in self.store.queue_counts():
+            return [], "nothing queued"
+        live = {s.session_id: s for s in live_discovery.discover()}
+        now_mono = time.monotonic()
+        reason = self._queue_block_reason(session_id, live, now_mono, force=True)
+        if reason is not None:
+            return [], reason
+        return self._send_batch(session_id, live[session_id].pane_id, now_mono)
+
+    def queue_hold_reason(self, session_id: str) -> Optional[str]:
+        """Why this session's pending queue isn't delivering (for the UI), or
+        None if it has no pending items or would deliver on the next tick."""
+        if session_id not in self.store.queue_counts():
+            return None
+        live = {s.session_id: s for s in live_discovery.discover()}
+        return self._queue_block_reason(session_id, live, time.monotonic())
+
+    def _maybe_snapshot(self) -> None:
+        """Capture the tmux topology for session-restore, throttled and change-driven.
+        Runs regardless of arming (it's independent of message injection)."""
+        now = time.monotonic()
+        if now - self._last_snapshot_check < SNAPSHOT_CHECK_SECONDS:
+            return
+        self._last_snapshot_check = now
+        snap = layout_snapshot.build_snapshot()
+        if snap is None:  # tmux down or no Claude windows → keep the last-known-good
+            return
+        sig = layout_snapshot.topology_sig(snap)
+        latest = self.store.latest_snapshot()
+        unchanged = latest is not None and latest[0] == sig
+        if unchanged and (now - self._last_snapshot_write) < SNAPSHOT_BACKSTOP_SECONDS:
+            return
+        self.store.save_snapshot(json.dumps(snap), sig)
+        self._last_snapshot_write = now
+
     def _tick(self) -> None:
+        # Session-restore snapshot — independent of arming; never let it break the loop.
+        try:
+            self._maybe_snapshot()
+        except Exception:
+            pass
+        # Queued replies deliver regardless of arming/schedule — an explicit user
+        # "send this when ready" beats the autopilot on/off switch.
+        try:
+            self.deliver_queued()
+        except Exception:
+            pass
         if not self.store.is_armed() or not self._within_hours():
             return
         configs = self.store.all_configs()
@@ -212,12 +364,7 @@ def _tick(self) -> None:
             # Usage-limit back-off: peek at the pane before acting.
             pane = tmux.capture_pane(ls.pane_id, 40)
             if _RATE_LIMIT_RE.search(pane):
-                reset = parse_reset_time(pane, now)
-                if reset and self.on_reset:
-                    try:
-                        self.on_reset(reset)  # anchor stats' 5h window
-                    except Exception:
-                        pass
+                reset = self._observe_limit(pane, now)
                 until = reset if (reset and reset > now) else now + timedelta(seconds=cfg.backoff_seconds)
                 self.store.set_backoff(sid, until)
                 when = until.astimezone().strftime("%a %H:%M")
@@ -256,6 +403,23 @@ def _tick(self) -> None:
             else:
                 self.store.log(sid, "error", err)
 
+    def _observe_limit(self, pane_text: str, now: Optional[datetime] = None):
+        """A rate-limit banner is on screen: report the parsed reset time (anchors
+        stats' 5h window) and the hit itself (calibrates the observed ceiling).
+        Returns the parsed reset time, if any."""
+        reset = parse_reset_time(pane_text, now or datetime.now(timezone.utc))
+        if reset and self.on_reset:
+            try:
+                self.on_reset(reset)
+            except Exception:
+                pass
+        if self.on_limit:
+            try:
+                self.on_limit("week" if "week" in pane_text.lower() else "5h")
+            except Exception:
+                pass
+        return reset
+
     # --- AI idle mode (two-phase) ---------------------------------------------
     # Phase A requests a draft at exactly the point message-mode would inject
     # (so every existing gate — enabled/idle/no-waiting_for/max_sends/interval/